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