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