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