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