Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "client/videomanager.h"
19 : #include "video_rtp_session.h"
20 : #include "video_sender.h"
21 : #include "video_receive_thread.h"
22 : #include "video_mixer.h"
23 : #include "socket_pair.h"
24 : #include "manager.h"
25 : #ifdef ENABLE_PLUGIN
26 : #endif
27 : #include "logger.h"
28 : #include "string_utils.h"
29 : #include "call.h"
30 : #include "conference.h"
31 : #include "congestion_control.h"
32 :
33 : #include <dhtnet/ice_socket.h>
34 : #include <asio/post.hpp>
35 : #include <asio/io_context.hpp>
36 :
37 : #include <string>
38 : #include <chrono>
39 :
40 : namespace jami {
41 : namespace video {
42 :
43 : using std::string;
44 :
45 : static constexpr unsigned MAX_REMB_DEC {1};
46 :
47 : constexpr auto DELAY_AFTER_RESTART = std::chrono::milliseconds(1000);
48 : constexpr auto EXPIRY_TIME_RTCP = std::chrono::seconds(2);
49 : constexpr auto DELAY_AFTER_REMB_INC = std::chrono::seconds(1);
50 : constexpr auto DELAY_AFTER_REMB_DEC = std::chrono::milliseconds(500);
51 :
52 301 : VideoRtpSession::VideoRtpSession(const string& callId,
53 : const string& streamId,
54 : const DeviceParams& localVideoParams,
55 301 : const std::shared_ptr<MediaRecorder>& rec)
56 : : RtpSession(callId, streamId, MediaType::MEDIA_VIDEO)
57 301 : , localVideoParams_(localVideoParams)
58 301 : , videoBitrateInfo_ {}
59 924 : , rtcpCheckerThread_([] { return true; }, [this] { processRtcpChecker(); }, [] {})
60 602 : , cc(std::make_unique<CongestionControl>())
61 : {
62 301 : recorder_ = rec;
63 301 : setupVideoBitrateInfo(); // reset bitrate
64 1204 : JAMI_LOG("[{:p}] Video RTP session created for call {} (recorder {:p})",
65 : fmt::ptr(this),
66 : callId_,
67 : fmt::ptr(recorder_));
68 301 : }
69 :
70 301 : VideoRtpSession::~VideoRtpSession()
71 : {
72 301 : deinitRecorder();
73 301 : stop();
74 1204 : JAMI_LOG("[{:p}] Video RTP session destroyed", fmt::ptr(this));
75 301 : }
76 :
77 : const VideoBitrateInfo&
78 80 : VideoRtpSession::getVideoBitrateInfo()
79 : {
80 80 : return videoBitrateInfo_;
81 : }
82 :
83 : /// Setup internal VideoBitrateInfo structure from media descriptors.
84 : ///
85 : void
86 142 : VideoRtpSession::updateMedia(const MediaDescription& send, const MediaDescription& receive)
87 : {
88 142 : BaseType::updateMedia(send, receive);
89 : // adjust send->codec bitrate info for higher video resolutions
90 142 : auto codecVideo = std::static_pointer_cast<jami::SystemVideoCodecInfo>(send_.codec);
91 142 : if (codecVideo) {
92 142 : auto const pixels = localVideoParams_.height * localVideoParams_.width;
93 142 : codecVideo->bitrate = std::max((unsigned int) (pixels * 0.001), SystemCodecInfo::DEFAULT_VIDEO_BITRATE);
94 142 : codecVideo->maxBitrate = std::max((unsigned int) (pixels * 0.0015), SystemCodecInfo::DEFAULT_MAX_BITRATE);
95 : }
96 142 : setupVideoBitrateInfo();
97 142 : }
98 :
99 : void
100 142 : VideoRtpSession::setRequestKeyFrameCallback(std::function<void(void)> cb)
101 : {
102 142 : cbKeyFrameRequest_ = std::move(cb);
103 142 : }
104 :
105 : void
106 157 : VideoRtpSession::startSender()
107 : {
108 157 : std::lock_guard lock(mutex_);
109 :
110 677 : JAMI_LOG("[{}] Start video RTP sender: input [{}] - muted [{}]",
111 : fmt::ptr(this),
112 : conference_ ? "Video Mixer" : input_,
113 : send_.hold ? "YES" : "NO");
114 :
115 157 : if (not socketPair_) {
116 : // Ignore if the transport is not set yet
117 0 : JAMI_WARNING("[{}] Transport not set yet", fmt::ptr(this));
118 0 : return;
119 : }
120 :
121 157 : if (send_.enabled and not send_.hold) {
122 150 : if (sender_) {
123 15 : if (videoLocal_)
124 14 : videoLocal_->detach(sender_.get());
125 15 : if (videoMixer_)
126 15 : videoMixer_->detach(sender_.get());
127 60 : JAMI_WARNING("[{}] Restarting video sender", fmt::ptr(this));
128 : }
129 :
130 150 : if (not conference_) {
131 101 : videoLocal_ = getVideoInput(input_);
132 101 : if (videoLocal_) {
133 101 : videoLocal_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
134 0 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
135 0 : if (auto shared = w.lock())
136 0 : shared->attachLocalRecorder(ms);
137 0 : });
138 0 : });
139 101 : auto newParams = videoLocal_->getParams();
140 : try {
141 101 : if (newParams.valid() && newParams.wait_for(NEWPARAMS_TIMEOUT) == std::future_status::ready) {
142 91 : localVideoParams_ = newParams.get();
143 : } else {
144 40 : JAMI_ERROR("[{}] No valid new video parameters", fmt::ptr(this));
145 10 : return;
146 : }
147 0 : } catch (const std::exception& e) {
148 0 : JAMI_ERROR("Exception during retrieving video parameters: {}", e.what());
149 0 : return;
150 0 : }
151 101 : } else {
152 0 : JAMI_WARNING("Unable to lock video input");
153 0 : return;
154 : }
155 :
156 : #if (defined(__ANDROID__) || (defined(TARGET_OS_IOS) && TARGET_OS_IOS))
157 : videoLocal_->setupSink(localVideoParams_.width, localVideoParams_.height);
158 : #endif
159 : }
160 :
161 : // be sure to not send any packets before saving last RTP seq value
162 140 : socketPair_->stopSendOp();
163 :
164 140 : auto codecVideo = std::static_pointer_cast<jami::SystemVideoCodecInfo>(send_.codec);
165 140 : auto autoQuality = codecVideo->isAutoQualityEnabled;
166 :
167 140 : send_.linkableHW = conference_ == nullptr;
168 140 : send_.bitrate = videoBitrateInfo_.videoBitrateCurrent;
169 : // NOTE:
170 : // Current implementation does not handle resolution change
171 : // (needed by window sharing feature) with HW codecs, so HW
172 : // codecs will be disabled for now.
173 247 : bool allowHwAccel = (localVideoParams_.format != "x11grab" && localVideoParams_.format != "dxgigrab"
174 247 : && localVideoParams_.format != "lavfi");
175 :
176 140 : if (socketPair_)
177 140 : initSeqVal_ = socketPair_->lastSeqValOut();
178 :
179 : try {
180 140 : sender_.reset();
181 140 : socketPair_->stopSendOp(false);
182 140 : MediaStream ms = !videoMixer_
183 140 : ? MediaStream("video sender",
184 : AV_PIX_FMT_YUV420P,
185 91 : 1 / static_cast<rational<int>>(localVideoParams_.framerate),
186 91 : localVideoParams_.width == 0 ? 1080
187 0 : : static_cast<int>(localVideoParams_.width),
188 91 : localVideoParams_.height == 0
189 : ? 720
190 0 : : static_cast<int>(localVideoParams_.height),
191 91 : static_cast<int>(send_.bitrate),
192 : static_cast<rational<int>>(localVideoParams_.framerate))
193 420 : : videoMixer_->getStream("Video Sender");
194 140 : sender_.reset(
195 280 : new VideoSender(getRemoteRtpUri(), ms, send_, *socketPair_, initSeqVal_ + 1, mtu_, allowHwAccel));
196 140 : if (changeOrientationCallback_)
197 140 : sender_->setChangeOrientationCallback(changeOrientationCallback_);
198 140 : if (socketPair_)
199 0 : socketPair_->setPacketLossCallback([this]() { cbKeyFrameRequest_(); });
200 :
201 140 : } catch (const MediaEncoderException& e) {
202 0 : JAMI_ERROR("{}", e.what());
203 0 : send_.enabled = false;
204 0 : }
205 140 : lastMediaRestart_ = clock::now();
206 140 : last_REMB_inc_ = clock::now();
207 140 : last_REMB_dec_ = clock::now();
208 140 : if (autoQuality and not rtcpCheckerThread_.isRunning())
209 125 : rtcpCheckerThread_.start();
210 15 : else if (not autoQuality and rtcpCheckerThread_.isRunning())
211 0 : rtcpCheckerThread_.join();
212 : // Block reads to received feedback packets
213 140 : if (socketPair_)
214 140 : socketPair_->setReadBlockingMode(true);
215 140 : }
216 157 : }
217 :
218 : void
219 15 : VideoRtpSession::restartSender()
220 : {
221 15 : std::lock_guard lock(mutex_);
222 :
223 : // ensure that start has been called before restart
224 15 : if (not socketPair_)
225 0 : return;
226 :
227 15 : startSender();
228 :
229 15 : if (conference_)
230 15 : setupConferenceVideoPipeline(*conference_, Direction::SEND);
231 : else
232 0 : setupVideoPipeline();
233 15 : }
234 :
235 : void
236 659 : VideoRtpSession::stopSender(bool forceStopSocket)
237 : {
238 : // Concurrency protection must be done by caller.
239 :
240 2674 : JAMI_LOG("[{}] Stop video RTP sender: input [{}] - muted [{}]",
241 : fmt::ptr(this),
242 : conference_ ? "Video Mixer" : input_,
243 : send_.hold ? "YES" : "NO");
244 :
245 659 : if (sender_) {
246 125 : if (videoLocal_)
247 91 : videoLocal_->detach(sender_.get());
248 125 : if (videoMixer_)
249 3 : videoMixer_->detach(sender_.get());
250 125 : sender_.reset();
251 : }
252 :
253 659 : if (socketPair_) {
254 146 : bool const isReceivingVideo = receive_.enabled && !receive_.hold;
255 146 : if (forceStopSocket || !isReceivingVideo) {
256 144 : socketPair_->stopSendOp();
257 144 : socketPair_->setReadBlockingMode(false);
258 : }
259 : }
260 659 : }
261 :
262 : void
263 142 : VideoRtpSession::startReceiver()
264 : {
265 : // Concurrency protection must be done by caller.
266 :
267 568 : JAMI_LOG("[{}] Starting receiver", fmt::ptr(this));
268 :
269 142 : if (receive_.enabled and not receive_.hold) {
270 136 : if (receiveThread_)
271 56 : JAMI_WARNING("[{}] Already has a receiver, restarting", fmt::ptr(this));
272 136 : receiveThread_.reset(new VideoReceiveThread(callId_, !conference_, receive_.receiving_sdp, mtu_));
273 :
274 : // ensure that start has been called
275 136 : if (not socketPair_)
276 0 : return;
277 :
278 : // XXX keyframe requests can timeout if unanswered
279 136 : receiveThread_->addIOContext(*socketPair_);
280 136 : receiveThread_->setSuccessfulSetupCb(onSuccessfulSetup_);
281 136 : receiveThread_->startLoop();
282 205 : receiveThread_->setRequestKeyFrameCallback([this]() { cbKeyFrameRequest_(); });
283 272 : receiveThread_->setRotation(rotation_.load());
284 136 : if (videoMixer_ and conference_) {
285 : // Note, this should be managed differently, this is a bit hacky
286 34 : auto audioId = streamId_;
287 102 : string_replace(audioId, "video", "audio");
288 34 : auto activeStream = videoMixer_->verifyActive(audioId);
289 34 : videoMixer_->removeAudioOnlySource(callId_, audioId);
290 34 : if (activeStream)
291 0 : videoMixer_->setActiveStream(streamId_);
292 34 : }
293 136 : receiveThread_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
294 31 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
295 31 : if (auto shared = w.lock())
296 31 : shared->attachRemoteRecorder(ms);
297 31 : });
298 31 : });
299 136 : } else {
300 24 : JAMI_LOG("[{}] Video receiver disabled", fmt::ptr(this));
301 6 : if (videoMixer_ and conference_) {
302 : // Note, this should be managed differently, this is a bit hacky
303 0 : auto audioId_ = streamId_;
304 0 : string_replace(audioId_, "video", "audio");
305 0 : if (receiveThread_) {
306 0 : auto activeStream = videoMixer_->verifyActive(streamId_);
307 0 : videoMixer_->addAudioOnlySource(callId_, audioId_);
308 0 : receiveThread_->detach(videoMixer_.get());
309 0 : if (activeStream)
310 0 : videoMixer_->setActiveStream(audioId_);
311 : } else {
312 : // Add audio-only source when video is disabled or muted.
313 : // Called after ICE negotiation, when peers can properly create video sinks.
314 0 : if (not receive_.enabled or receive_.hold) {
315 0 : videoMixer_->addAudioOnlySource(callId_, audioId_);
316 : }
317 : }
318 0 : }
319 : }
320 142 : if (socketPair_)
321 142 : socketPair_->setReadBlockingMode(true);
322 : }
323 :
324 : void
325 649 : VideoRtpSession::stopReceiver(bool forceStopSocket)
326 : {
327 : // Concurrency protection must be done by caller.
328 :
329 2596 : JAMI_LOG("[{}] Stopping receiver", fmt::ptr(this));
330 :
331 649 : if (not receiveThread_)
332 373 : return;
333 :
334 276 : if (videoMixer_) {
335 3 : auto activeStream = videoMixer_->verifyActive(streamId_);
336 3 : auto audioId = streamId_;
337 9 : string_replace(audioId, "video", "audio");
338 3 : videoMixer_->addAudioOnlySource(callId_, audioId);
339 3 : receiveThread_->detach(videoMixer_.get());
340 3 : if (activeStream)
341 0 : videoMixer_->setActiveStream(audioId);
342 3 : }
343 :
344 : // We need to disable the read operation, otherwise the
345 : // receiver thread will block since the peer stopped sending
346 : // RTP packets.
347 276 : bool const isSendingVideo = send_.enabled && !send_.hold;
348 276 : if (socketPair_) {
349 139 : if (forceStopSocket || !isSendingVideo) {
350 139 : socketPair_->setReadBlockingMode(false);
351 139 : socketPair_->stopSendOp();
352 : }
353 : }
354 :
355 276 : auto ms = receiveThread_->getInfo();
356 276 : if (auto* ob = recorder_->getStream(ms.name)) {
357 31 : receiveThread_->detach(ob);
358 31 : recorder_->removeStream(ms);
359 : }
360 :
361 276 : if (forceStopSocket || !isSendingVideo)
362 276 : receiveThread_->stopLoop();
363 276 : receiveThread_->stopSink();
364 276 : }
365 :
366 : void
367 144 : VideoRtpSession::start(std::unique_ptr<dhtnet::IceSocket> rtp_sock, std::unique_ptr<dhtnet::IceSocket> rtcp_sock)
368 : {
369 144 : std::lock_guard lock(mutex_);
370 :
371 144 : if (not send_.enabled and not receive_.enabled) {
372 2 : stop();
373 2 : return;
374 : }
375 :
376 : try {
377 142 : if (rtp_sock and rtcp_sock) {
378 142 : if (send_.addr) {
379 142 : rtp_sock->setDefaultRemoteAddress(send_.addr);
380 : }
381 :
382 142 : auto& rtcpAddr = send_.rtcp_addr ? send_.rtcp_addr : send_.addr;
383 142 : if (rtcpAddr) {
384 142 : rtcp_sock->setDefaultRemoteAddress(rtcpAddr);
385 : }
386 142 : socketPair_.reset(new SocketPair(std::move(rtp_sock), std::move(rtcp_sock)));
387 : } else {
388 0 : socketPair_.reset(new SocketPair(getRemoteRtpUri().c_str(), receive_.addr.getPort()));
389 : }
390 :
391 142 : last_REMB_inc_ = clock::now();
392 142 : last_REMB_dec_ = clock::now();
393 :
394 5721 : socketPair_->setRtpDelayCallback([&](int gradient, int deltaT) { delayMonitor(gradient, deltaT); });
395 :
396 142 : if (send_.crypto and receive_.crypto) {
397 568 : socketPair_->createSRTP(receive_.crypto.getCryptoSuite().c_str(),
398 284 : receive_.crypto.getSrtpKeyInfo().c_str(),
399 284 : send_.crypto.getCryptoSuite().c_str(),
400 284 : send_.crypto.getSrtpKeyInfo().c_str());
401 : }
402 0 : } catch (const std::runtime_error& e) {
403 0 : JAMI_ERROR("[{}] Socket creation failed: {}", fmt::ptr(this), e.what());
404 0 : return;
405 0 : }
406 :
407 142 : startReceiver();
408 142 : startSender();
409 :
410 142 : if (conference_) {
411 34 : if (send_.enabled and not send_.hold) {
412 34 : setupConferenceVideoPipeline(*conference_, Direction::SEND);
413 : }
414 34 : if (receive_.enabled and not receive_.hold) {
415 34 : setupConferenceVideoPipeline(*conference_, Direction::RECV);
416 : }
417 : } else {
418 108 : setupVideoPipeline();
419 : }
420 144 : }
421 :
422 : void
423 649 : VideoRtpSession::stop()
424 : {
425 649 : std::lock_guard lock(mutex_);
426 :
427 649 : stopSender(true);
428 649 : stopReceiver(true);
429 :
430 649 : if (socketPair_)
431 142 : socketPair_->interrupt();
432 :
433 649 : rtcpCheckerThread_.join();
434 :
435 : // reset default video quality if exist
436 649 : if (videoBitrateInfo_.videoQualityCurrent != SystemCodecInfo::DEFAULT_NO_QUALITY)
437 0 : videoBitrateInfo_.videoQualityCurrent = SystemCodecInfo::DEFAULT_CODEC_QUALITY;
438 :
439 649 : videoBitrateInfo_.videoBitrateCurrent = SystemCodecInfo::DEFAULT_VIDEO_BITRATE;
440 649 : storeVideoBitrateInfo();
441 :
442 649 : socketPair_.reset();
443 649 : videoLocal_.reset();
444 649 : }
445 :
446 : void
447 292 : VideoRtpSession::setMuted(bool mute, Direction dir)
448 : {
449 292 : std::lock_guard lock(mutex_);
450 :
451 : // Sender
452 292 : if (dir == Direction::SEND) {
453 148 : if (send_.hold == mute) {
454 552 : JAMI_LOG("[{}] Local already {}", fmt::ptr(this), mute ? "muted" : "un-muted");
455 138 : return;
456 : }
457 :
458 10 : if ((send_.hold = mute)) {
459 10 : if (videoLocal_) {
460 3 : auto ms = videoLocal_->getInfo();
461 3 : if (auto* ob = recorder_->getStream(ms.name)) {
462 0 : videoLocal_->detach(ob);
463 0 : recorder_->removeStream(ms);
464 : }
465 3 : }
466 10 : stopSender();
467 : } else {
468 0 : restartSender();
469 : }
470 10 : return;
471 : }
472 :
473 : // Receiver
474 144 : if (receive_.hold == mute) {
475 576 : JAMI_LOG("[{}] Remote already {}", fmt::ptr(this), mute ? "muted" : "un-muted");
476 144 : return;
477 : }
478 :
479 0 : if ((receive_.hold = mute)) {
480 0 : if (receiveThread_) {
481 0 : auto ms = receiveThread_->getInfo();
482 0 : if (auto* ob = recorder_->getStream(ms.name)) {
483 0 : receiveThread_->detach(ob);
484 0 : recorder_->removeStream(ms);
485 : }
486 0 : }
487 0 : stopReceiver();
488 : } else {
489 0 : startReceiver();
490 0 : if (conference_ and not receive_.hold) {
491 0 : setupConferenceVideoPipeline(*conference_, Direction::RECV);
492 : }
493 : }
494 292 : }
495 :
496 : void
497 211 : VideoRtpSession::forceKeyFrame()
498 : {
499 211 : std::lock_guard lock(mutex_);
500 : #if __ANDROID__
501 : if (videoLocal_)
502 : emitSignal<libjami::VideoSignal::RequestKeyFrame>(videoLocal_->getName());
503 : #else
504 211 : if (sender_)
505 148 : sender_->forceKeyFrame();
506 : #endif
507 211 : }
508 :
509 : void
510 350 : VideoRtpSession::setRotation(int rotation)
511 : {
512 350 : rotation_.store(rotation);
513 350 : if (receiveThread_)
514 49 : receiveThread_->setRotation(rotation);
515 350 : }
516 :
517 : void
518 108 : VideoRtpSession::setupVideoPipeline()
519 : {
520 108 : if (sender_) {
521 91 : if (videoLocal_) {
522 364 : JAMI_LOG("[{}] Setup video pipeline on local capture device", fmt::ptr(this));
523 91 : videoLocal_->attach(sender_.get());
524 : }
525 : } else {
526 17 : videoLocal_.reset();
527 : }
528 108 : }
529 :
530 : void
531 98 : VideoRtpSession::setupConferenceVideoPipeline(Conference& conference, Direction dir)
532 : {
533 98 : if (dir == Direction::SEND) {
534 196 : JAMI_DEBUG("[conf:{}] Setup video sender pipeline for call {}", conference.getConfId(), callId_);
535 49 : videoMixer_ = conference.getVideoMixer();
536 49 : if (sender_) {
537 : // Swap sender from local video to conference video mixer
538 49 : if (videoLocal_)
539 14 : videoLocal_->detach(sender_.get());
540 49 : if (videoMixer_)
541 49 : videoMixer_->attach(sender_.get());
542 : } else {
543 0 : JAMI_WARNING("[{}] no sender", fmt::ptr(this));
544 : }
545 : } else {
546 196 : JAMI_DEBUG("[conf:{}] Setup video receiver pipeline for call {}", conference.getConfId(), callId_);
547 49 : if (receiveThread_) {
548 49 : receiveThread_->stopSink();
549 49 : if (videoMixer_)
550 49 : videoMixer_->attachVideo(receiveThread_.get(), callId_, streamId_);
551 : } else {
552 0 : JAMI_WARNING("[{}] no receiver", fmt::ptr(this));
553 : }
554 : }
555 98 : }
556 :
557 : void
558 59 : VideoRtpSession::enterConference(Conference& conference)
559 : {
560 59 : std::lock_guard lock(mutex_);
561 :
562 59 : exitConference();
563 :
564 59 : conference_ = &conference;
565 59 : videoMixer_ = conference.getVideoMixer();
566 236 : JAMI_DEBUG("[conf:{}] Entering conference", conference.getConfId());
567 :
568 59 : if (send_.enabled or receiveThread_) {
569 : // Restart encoder with conference parameter ON in order to unlink HW encoder
570 : // from HW decoder.
571 15 : restartSender();
572 15 : if (conference_) {
573 15 : setupConferenceVideoPipeline(conference, Direction::RECV);
574 : }
575 : }
576 59 : }
577 :
578 : void
579 117 : VideoRtpSession::exitConference()
580 : {
581 117 : std::lock_guard lock(mutex_);
582 :
583 117 : if (!conference_)
584 58 : return;
585 :
586 236 : JAMI_DEBUG("[conf:{}] Exiting conference", conference_->getConfId());
587 :
588 59 : if (videoMixer_) {
589 59 : if (sender_)
590 46 : videoMixer_->detach(sender_.get());
591 :
592 59 : if (receiveThread_) {
593 48 : auto activeStream = videoMixer_->verifyActive(streamId_);
594 48 : videoMixer_->detachVideo(receiveThread_.get());
595 48 : receiveThread_->startSink();
596 48 : if (activeStream)
597 0 : videoMixer_->setActiveStream(streamId_);
598 : }
599 :
600 59 : videoMixer_.reset();
601 : }
602 :
603 59 : conference_ = nullptr;
604 117 : }
605 :
606 : bool
607 373 : VideoRtpSession::check_RCTP_Info_RR(RTCPInfo& rtcpi)
608 : {
609 373 : auto rtcpInfoVect = socketPair_->getRtcpRR();
610 373 : unsigned totalLost = 0;
611 373 : unsigned totalJitter = 0;
612 373 : unsigned nbDropNotNull = 0;
613 373 : auto vectSize = rtcpInfoVect.size();
614 :
615 373 : if (vectSize != 0) {
616 20 : for (const auto& it : rtcpInfoVect) {
617 10 : if (it.fraction_lost != 0) // Exclude null drop
618 0 : nbDropNotNull++;
619 10 : totalLost += it.fraction_lost;
620 10 : totalJitter += ntohl(it.jitter);
621 : }
622 10 : rtcpi.packetLoss = nbDropNotNull ? static_cast<float>((100 * totalLost) / (256.0 * nbDropNotNull)) : 0;
623 : // Jitter is expressed in timestamp unit -> convert to milliseconds
624 : // https://stackoverflow.com/questions/51956520/convert-jitter-from-rtp-timestamp-unit-to-millisseconds
625 10 : rtcpi.jitter = static_cast<unsigned int>(
626 10 : (static_cast<float>(totalJitter) / static_cast<float>(vectSize) / 90000.0f) * 1000.0f);
627 10 : rtcpi.nb_sample = vectSize;
628 10 : rtcpi.latency = static_cast<float>(socketPair_->getLastLatency());
629 10 : return true;
630 : }
631 363 : return false;
632 373 : }
633 :
634 : bool
635 373 : VideoRtpSession::check_RCTP_Info_REMB(uint64_t* br)
636 : {
637 373 : auto rtcpInfoVect = socketPair_->getRtcpREMB();
638 :
639 373 : if (!rtcpInfoVect.empty()) {
640 176 : auto pkt = rtcpInfoVect.back();
641 176 : auto temp = cc->parseREMB(pkt);
642 176 : *br = (temp >> 10) | ((temp << 6) & 0xff00) | ((temp << 16) & 0x30000);
643 176 : return true;
644 : }
645 197 : return false;
646 373 : }
647 :
648 : void
649 373 : VideoRtpSession::adaptQualityAndBitrate()
650 : {
651 373 : setupVideoBitrateInfo();
652 :
653 : uint64_t br;
654 373 : if (check_RCTP_Info_REMB(&br)) {
655 176 : delayProcessing(static_cast<int>(br));
656 : }
657 :
658 373 : RTCPInfo rtcpi {};
659 373 : if (check_RCTP_Info_RR(rtcpi)) {
660 10 : dropProcessing(&rtcpi);
661 : }
662 373 : }
663 :
664 : void
665 10 : VideoRtpSession::dropProcessing(RTCPInfo* rtcpi)
666 : {
667 : // If bitrate has changed, let time to receive fresh RTCP packets
668 10 : auto now = clock::now();
669 10 : auto restartTimer = now - lastMediaRestart_;
670 10 : if (restartTimer < DELAY_AFTER_RESTART) {
671 0 : return;
672 : }
673 : #ifndef __ANDROID__
674 : // Do nothing if jitter is more than 1 second
675 10 : if (rtcpi->jitter > 1000) {
676 0 : return;
677 : }
678 : #endif
679 10 : auto pondLoss = getPonderateLoss(rtcpi->packetLoss);
680 10 : auto oldBitrate = videoBitrateInfo_.videoBitrateCurrent;
681 10 : int newBitrate = static_cast<int>(oldBitrate);
682 :
683 : // Fill histoLoss and histoJitter_ with samples
684 10 : if (restartTimer < DELAY_AFTER_RESTART + std::chrono::seconds(1)) {
685 0 : return;
686 : } else {
687 : // If ponderate drops are inferior to 10% that mean drop are not from congestion but from
688 : // network...
689 : // ... we can increase
690 10 : if (pondLoss >= 5.0f && rtcpi->packetLoss > 0.0f) {
691 0 : newBitrate = static_cast<int>(std::lround(newBitrate * (1.0f - rtcpi->packetLoss / 150.0f)));
692 0 : histoLoss_.clear();
693 0 : lastMediaRestart_ = now;
694 0 : JAMI_LOG("[BandwidthAdapt] Detected transmission bandwidth overuse, decrease bitrate from {} Kbps to {} "
695 : "Kbps, ratio {} (ponderate loss: {}%, packet loss rate: {}%)",
696 : oldBitrate,
697 : newBitrate,
698 : (float) newBitrate / oldBitrate,
699 : pondLoss,
700 : rtcpi->packetLoss);
701 : }
702 : }
703 :
704 10 : setNewBitrate(newBitrate);
705 : }
706 :
707 : void
708 176 : VideoRtpSession::delayProcessing(int br)
709 : {
710 176 : int newBitrate = static_cast<int>(videoBitrateInfo_.videoBitrateCurrent);
711 176 : if (br == 0x6803)
712 0 : newBitrate = static_cast<int>(std::lround(newBitrate * 0.85f));
713 176 : else if (br == 0x7378) {
714 176 : auto now = clock::now();
715 176 : auto msSinceLastDecrease = std::chrono::duration_cast<std::chrono::milliseconds>(now - lastBitrateDecrease);
716 176 : auto increaseCoefficient = std::min(static_cast<float>(msSinceLastDecrease.count()) / 600000.0f + 1.0f, 1.05f);
717 176 : newBitrate = static_cast<int>(std::lround(newBitrate * increaseCoefficient));
718 : } else
719 0 : return;
720 :
721 176 : setNewBitrate(newBitrate);
722 : }
723 :
724 : void
725 186 : VideoRtpSession::setNewBitrate(unsigned int newBR)
726 : {
727 186 : newBR = std::max(newBR, videoBitrateInfo_.videoBitrateMin);
728 186 : newBR = std::min(newBR, videoBitrateInfo_.videoBitrateMax);
729 :
730 186 : if (newBR < videoBitrateInfo_.videoBitrateCurrent)
731 0 : lastBitrateDecrease = clock::now();
732 :
733 186 : if (videoBitrateInfo_.videoBitrateCurrent != newBR) {
734 31 : videoBitrateInfo_.videoBitrateCurrent = newBR;
735 31 : storeVideoBitrateInfo();
736 :
737 : #if __ANDROID__
738 : if (auto input_device = std::dynamic_pointer_cast<VideoInput>(videoLocal_))
739 : emitSignal<libjami::VideoSignal::SetBitrate>(input_device->getConfig().name, (int) newBR);
740 : #endif
741 :
742 31 : if (sender_) {
743 31 : auto ret = sender_->setBitrate(newBR);
744 31 : if (ret == -1)
745 0 : JAMI_ERROR("Fail to access the encoder");
746 31 : else if (ret == 0)
747 0 : restartSender();
748 : } else {
749 0 : JAMI_ERROR("Fail to access the sender");
750 : }
751 : }
752 186 : }
753 :
754 : void
755 816 : VideoRtpSession::setupVideoBitrateInfo()
756 : {
757 816 : auto codecVideo = std::static_pointer_cast<jami::SystemVideoCodecInfo>(send_.codec);
758 816 : if (codecVideo) {
759 515 : videoBitrateInfo_ = {
760 515 : codecVideo->bitrate,
761 515 : codecVideo->minBitrate,
762 515 : codecVideo->maxBitrate,
763 515 : codecVideo->quality,
764 515 : codecVideo->minQuality,
765 515 : codecVideo->maxQuality,
766 515 : videoBitrateInfo_.cptBitrateChecking,
767 515 : videoBitrateInfo_.maxBitrateChecking,
768 515 : videoBitrateInfo_.packetLostThreshold,
769 : };
770 : } else {
771 301 : videoBitrateInfo_ = {0, 0, 0, 0, 0, 0, 0, MAX_ADAPTATIVE_BITRATE_ITERATION, PACKET_LOSS_THRESHOLD};
772 : }
773 816 : }
774 :
775 : void
776 680 : VideoRtpSession::storeVideoBitrateInfo()
777 : {
778 680 : if (auto codecVideo = std::static_pointer_cast<jami::SystemVideoCodecInfo>(send_.codec)) {
779 438 : codecVideo->bitrate = videoBitrateInfo_.videoBitrateCurrent;
780 438 : codecVideo->quality = videoBitrateInfo_.videoQualityCurrent;
781 680 : }
782 680 : }
783 :
784 : void
785 373 : VideoRtpSession::processRtcpChecker()
786 : {
787 373 : adaptQualityAndBitrate();
788 373 : socketPair_->waitForRTCP(std::chrono::seconds(rtcp_checking_interval));
789 373 : }
790 :
791 : void
792 32 : VideoRtpSession::attachRemoteRecorder(const MediaStream& ms)
793 : {
794 32 : std::lock_guard lock(mutex_);
795 32 : if (!recorder_ || !receiveThread_)
796 0 : return;
797 32 : if (auto* ob = recorder_->addStream(ms)) {
798 32 : receiveThread_->attach(ob);
799 : }
800 32 : }
801 :
802 : void
803 0 : VideoRtpSession::attachLocalRecorder(const MediaStream& ms)
804 : {
805 0 : std::lock_guard lock(mutex_);
806 0 : if (!recorder_ || !videoLocal_ || !Manager::instance().videoPreferences.getRecordPreview())
807 0 : return;
808 0 : if (auto* ob = recorder_->addStream(ms)) {
809 0 : videoLocal_->attach(ob);
810 : }
811 0 : }
812 :
813 : void
814 1 : VideoRtpSession::initRecorder()
815 : {
816 1 : if (!recorder_)
817 0 : return;
818 1 : if (receiveThread_) {
819 1 : receiveThread_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
820 1 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
821 1 : if (auto shared = w.lock())
822 1 : shared->attachRemoteRecorder(ms);
823 1 : });
824 1 : });
825 : }
826 1 : if (videoLocal_ && !send_.hold) {
827 1 : videoLocal_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
828 0 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
829 0 : if (auto shared = w.lock())
830 0 : shared->attachLocalRecorder(ms);
831 0 : });
832 0 : });
833 : }
834 : }
835 :
836 : void
837 302 : VideoRtpSession::deinitRecorder()
838 : {
839 302 : if (!recorder_)
840 0 : return;
841 302 : if (receiveThread_) {
842 122 : auto ms = receiveThread_->getInfo();
843 122 : if (auto* ob = recorder_->getStream(ms.name)) {
844 0 : receiveThread_->detach(ob);
845 0 : recorder_->removeStream(ms);
846 : }
847 122 : }
848 302 : if (videoLocal_) {
849 0 : auto ms = videoLocal_->getInfo();
850 0 : if (auto* ob = recorder_->getStream(ms.name)) {
851 0 : videoLocal_->detach(ob);
852 0 : recorder_->removeStream(ms);
853 : }
854 0 : }
855 : }
856 :
857 : void
858 142 : VideoRtpSession::setChangeOrientationCallback(std::function<void(int)> cb)
859 : {
860 142 : changeOrientationCallback_ = std::move(cb);
861 142 : if (sender_)
862 10 : sender_->setChangeOrientationCallback(changeOrientationCallback_);
863 142 : }
864 :
865 : float
866 10 : VideoRtpSession::getPonderateLoss(float lastLoss)
867 : {
868 10 : float pond = 0.0f, pondLoss = 0.0f, totalPond = 0.0f;
869 10 : constexpr float coefficient_a = -1 / 100.0f;
870 10 : constexpr float coefficient_b = 100.0f;
871 :
872 10 : auto now = clock::now();
873 :
874 10 : histoLoss_.emplace_back(now, lastLoss);
875 :
876 20 : for (auto it = histoLoss_.begin(); it != histoLoss_.end();) {
877 10 : auto delay = std::chrono::duration_cast<std::chrono::milliseconds>(now - it->first);
878 :
879 : // 1ms -> 100%
880 : // 2000ms -> 80%
881 10 : if (delay <= EXPIRY_TIME_RTCP) {
882 10 : if (it->second == 0.0f)
883 10 : pond = 20.0f; // Reduce weight of null drop
884 : else
885 0 : pond = std::min(static_cast<float>(delay.count()) * coefficient_a + coefficient_b, 100.0f);
886 10 : totalPond += pond;
887 10 : pondLoss += it->second * pond;
888 10 : ++it;
889 : } else
890 0 : it = histoLoss_.erase(it);
891 : }
892 10 : if (totalPond == 0)
893 0 : return 0.0f;
894 :
895 10 : return pondLoss / totalPond;
896 : }
897 :
898 : void
899 5579 : VideoRtpSession::delayMonitor(int gradient, int deltaT)
900 : {
901 5579 : float estimation = cc->kalmanFilter(gradient);
902 5579 : float thresh = cc->get_thresh();
903 :
904 5579 : cc->update_thresh(estimation, deltaT);
905 :
906 5579 : BandwidthUsage bwState = cc->get_bw_state(estimation, thresh);
907 5579 : auto now = clock::now();
908 :
909 5579 : if (bwState == BandwidthUsage::bwOverusing) {
910 0 : auto remb_timer_dec = now - last_REMB_dec_;
911 0 : if ((not remb_dec_cnt_) or (remb_timer_dec > DELAY_AFTER_REMB_DEC)) {
912 0 : last_REMB_dec_ = now;
913 0 : remb_dec_cnt_ = 0;
914 : }
915 :
916 : // Limit REMB decrease to MAX_REMB_DEC every DELAY_AFTER_REMB_DEC ms
917 0 : if (remb_dec_cnt_ < MAX_REMB_DEC && remb_timer_dec < DELAY_AFTER_REMB_DEC) {
918 0 : remb_dec_cnt_++;
919 0 : JAMI_WARNING("[BandwidthAdapt] Detected reception bandwidth overuse");
920 0 : uint8_t* buf = nullptr;
921 0 : uint64_t br = 0x6803; // Decrease 3
922 0 : auto v = cc->createREMB(br);
923 0 : buf = &v[0];
924 0 : socketPair_->writeData(buf, static_cast<int>(v.size()));
925 0 : last_REMB_inc_ = clock::now();
926 0 : }
927 5579 : } else if (bwState == BandwidthUsage::bwNormal) {
928 4315 : auto remb_timer_inc = now - last_REMB_inc_;
929 4315 : if (remb_timer_inc > DELAY_AFTER_REMB_INC) {
930 176 : uint8_t* buf = nullptr;
931 176 : uint64_t br = 0x7378; // INcrease
932 176 : auto v = cc->createREMB(br);
933 176 : buf = &v[0];
934 176 : socketPair_->writeData(buf, static_cast<int>(v.size()));
935 176 : last_REMB_inc_ = clock::now();
936 176 : }
937 : }
938 5579 : }
939 : } // namespace video
940 : } // namespace jami
|
