/* * Copyright (c) 2022 The WebRTC project authors. All Rights Reserved. * * Use of this source code is governed by a BSD-style license * that can be found in the LICENSE file in the root of the source * tree. An additional intellectual property rights grant can be found * in the file PATENTS. All contributing project authors may * be found in the AUTHORS file in the root of the source tree. */ #include "video/frame_buffer_proxy.h" #include #include #include #include "absl/base/attributes.h" #include "absl/functional/bind_front.h" #include "api/sequence_checker.h" #include "api/units/data_size.h" #include "api/video/encoded_frame.h" #include "api/video/frame_buffer.h" #include "api/video/video_content_type.h" #include "modules/video_coding/frame_buffer2.h" #include "modules/video_coding/frame_helpers.h" #include "modules/video_coding/timing/inter_frame_delay.h" #include "modules/video_coding/timing/jitter_estimator.h" #include "rtc_base/checks.h" #include "rtc_base/logging.h" #include "rtc_base/thread_annotations.h" #include "video/frame_decode_timing.h" #include "video/task_queue_frame_decode_scheduler.h" #include "video/video_receive_stream_timeout_tracker.h" namespace webrtc { namespace { class FrameBuffer2Proxy : public FrameBufferProxy { public: FrameBuffer2Proxy(Clock* clock, VCMTiming* timing, VCMReceiveStatisticsCallback* stats_proxy, rtc::TaskQueue* decode_queue, FrameSchedulingReceiver* receiver, TimeDelta max_wait_for_keyframe, TimeDelta max_wait_for_frame, const FieldTrialsView& field_trials) : max_wait_for_keyframe_(max_wait_for_keyframe), max_wait_for_frame_(max_wait_for_frame), frame_buffer_(clock, timing, stats_proxy, field_trials), decode_queue_(decode_queue), stats_proxy_(stats_proxy), receiver_(receiver) { RTC_DCHECK(decode_queue_); RTC_DCHECK(stats_proxy_); RTC_DCHECK(receiver_); } void StopOnWorker() override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); decode_queue_->PostTask([this] { frame_buffer_.Stop(); decode_safety_->SetNotAlive(); }); } void SetProtectionMode(VCMVideoProtection protection_mode) override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); frame_buffer_.SetProtectionMode(kProtectionNackFEC); } void Clear() override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); frame_buffer_.Clear(); } absl::optional InsertFrame( std::unique_ptr frame) override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); int64_t last_continuous_pid = frame_buffer_.InsertFrame(std::move(frame)); if (last_continuous_pid != -1) return last_continuous_pid; return absl::nullopt; } void UpdateRtt(int64_t max_rtt_ms) override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); frame_buffer_.UpdateRtt(max_rtt_ms); } void StartNextDecode(bool keyframe_required) override { if (!decode_queue_->IsCurrent()) { decode_queue_->PostTask(SafeTask( decode_safety_, [this, keyframe_required] { StartNextDecode(keyframe_required); })); return; } RTC_DCHECK_RUN_ON(decode_queue_); frame_buffer_.NextFrame( MaxWait(keyframe_required).ms(), keyframe_required, decode_queue_, /* encoded frame handler */ [this, keyframe_required](std::unique_ptr frame) { RTC_DCHECK_RUN_ON(decode_queue_); if (!decode_safety_->alive()) return; if (frame) { receiver_->OnEncodedFrame(std::move(frame)); } else { receiver_->OnDecodableFrameTimeout(MaxWait(keyframe_required)); } }); } int Size() override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); return frame_buffer_.Size(); } private: TimeDelta MaxWait(bool keyframe_required) const { return keyframe_required ? max_wait_for_keyframe_ : max_wait_for_frame_; } RTC_NO_UNIQUE_ADDRESS SequenceChecker worker_sequence_checker_; const TimeDelta max_wait_for_keyframe_; const TimeDelta max_wait_for_frame_; video_coding::FrameBuffer frame_buffer_; rtc::TaskQueue* const decode_queue_; VCMReceiveStatisticsCallback* const stats_proxy_; FrameSchedulingReceiver* const receiver_; rtc::scoped_refptr decode_safety_ = PendingTaskSafetyFlag::CreateDetached(); }; // Max number of frames the buffer will hold. static constexpr size_t kMaxFramesBuffered = 800; // Max number of decoded frame info that will be saved. static constexpr int kMaxFramesHistory = 1 << 13; // Default value for the maximum decode queue size that is used when the // low-latency renderer is used. static constexpr size_t kZeroPlayoutDelayDefaultMaxDecodeQueueSize = 8; struct FrameMetadata { explicit FrameMetadata(const EncodedFrame& frame) : is_last_spatial_layer(frame.is_last_spatial_layer), is_keyframe(frame.is_keyframe()), size(frame.size()), contentType(frame.contentType()), delayed_by_retransmission(frame.delayed_by_retransmission()), rtp_timestamp(frame.Timestamp()), receive_time(frame.ReceivedTimestamp()) {} const bool is_last_spatial_layer; const bool is_keyframe; const size_t size; const VideoContentType contentType; const bool delayed_by_retransmission; const uint32_t rtp_timestamp; const absl::optional receive_time; }; Timestamp ReceiveTime(const EncodedFrame& frame) { absl::optional ts = frame.ReceivedTimestamp(); RTC_DCHECK(ts.has_value()) << "Received frame must have a timestamp set!"; return *ts; } // Encapsulates use of the new frame buffer for use in // VideoReceiveStreamInterface. This behaves the same as the FrameBuffer2Proxy // but uses frame_buffer instead. Responsibilities from frame_buffer2, like // stats, jitter and frame timing accounting are moved into this pro class FrameBuffer3Proxy : public FrameBufferProxy { public: FrameBuffer3Proxy( Clock* clock, TaskQueueBase* worker_queue, VCMTiming* timing, VCMReceiveStatisticsCallback* stats_proxy, rtc::TaskQueue* decode_queue, FrameSchedulingReceiver* receiver, TimeDelta max_wait_for_keyframe, TimeDelta max_wait_for_frame, std::unique_ptr frame_decode_scheduler, const FieldTrialsView& field_trials) : field_trials_(field_trials), clock_(clock), worker_queue_(worker_queue), decode_queue_(decode_queue), stats_proxy_(stats_proxy), receiver_(receiver), timing_(timing), frame_decode_scheduler_(std::move(frame_decode_scheduler)), jitter_estimator_(clock_, field_trials), buffer_(std::make_unique(kMaxFramesBuffered, kMaxFramesHistory, field_trials)), decode_timing_(clock_, timing_), timeout_tracker_(clock_, worker_queue_, VideoReceiveStreamTimeoutTracker::Timeouts{ .max_wait_for_keyframe = max_wait_for_keyframe, .max_wait_for_frame = max_wait_for_frame}, absl::bind_front(&FrameBuffer3Proxy::OnTimeout, this)), zero_playout_delay_max_decode_queue_size_( "max_decode_queue_size", kZeroPlayoutDelayDefaultMaxDecodeQueueSize) { RTC_DCHECK(decode_queue_); RTC_DCHECK(stats_proxy_); RTC_DCHECK(receiver_); RTC_DCHECK(timing_); RTC_DCHECK(worker_queue_); RTC_DCHECK(clock_); RTC_DCHECK(frame_decode_scheduler_); RTC_LOG(LS_WARNING) << "Using FrameBuffer3"; ParseFieldTrial({&zero_playout_delay_max_decode_queue_size_}, field_trials.Lookup("WebRTC-ZeroPlayoutDelay")); } // FrameBufferProxy implementation. void StopOnWorker() override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); frame_decode_scheduler_->Stop(); timeout_tracker_.Stop(); decoder_ready_for_new_frame_ = false; decode_queue_->PostTask([this] { RTC_DCHECK_RUN_ON(decode_queue_); decode_safety_->SetNotAlive(); }); } void SetProtectionMode(VCMVideoProtection protection_mode) override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); protection_mode_ = kProtectionNackFEC; } void Clear() override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); stats_proxy_->OnDroppedFrames(buffer_->CurrentSize()); buffer_ = std::make_unique(kMaxFramesBuffered, kMaxFramesHistory, field_trials_); frame_decode_scheduler_->CancelOutstanding(); } absl::optional InsertFrame( std::unique_ptr frame) override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); FrameMetadata metadata(*frame); int complete_units = buffer_->GetTotalNumberOfContinuousTemporalUnits(); if (buffer_->InsertFrame(std::move(frame))) { RTC_DCHECK(metadata.receive_time) << "Frame receive time must be set!"; if (!metadata.delayed_by_retransmission && metadata.receive_time) timing_->IncomingTimestamp(metadata.rtp_timestamp, *metadata.receive_time); if (complete_units < buffer_->GetTotalNumberOfContinuousTemporalUnits()) { stats_proxy_->OnCompleteFrame(metadata.is_keyframe, metadata.size, metadata.contentType); MaybeScheduleFrameForRelease(); } } return buffer_->LastContinuousFrameId(); } void UpdateRtt(int64_t max_rtt_ms) override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); jitter_estimator_.UpdateRtt(TimeDelta::Millis(max_rtt_ms)); } void StartNextDecode(bool keyframe_required) override { if (!worker_queue_->IsCurrent()) { worker_queue_->PostTask(SafeTask( worker_safety_.flag(), [this, keyframe_required] { StartNextDecode(keyframe_required); })); return; } RTC_DCHECK_RUN_ON(&worker_sequence_checker_); if (!timeout_tracker_.Running()) timeout_tracker_.Start(keyframe_required); keyframe_required_ = keyframe_required; if (keyframe_required_) { timeout_tracker_.SetWaitingForKeyframe(); } decoder_ready_for_new_frame_ = true; MaybeScheduleFrameForRelease(); } int Size() override { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); return buffer_->CurrentSize(); } void OnFrameReady( absl::InlinedVector, 4> frames, Timestamp render_time) { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); RTC_DCHECK(!frames.empty()); timeout_tracker_.OnEncodedFrameReleased(); Timestamp now = clock_->CurrentTime(); bool superframe_delayed_by_retransmission = false; DataSize superframe_size = DataSize::Zero(); const EncodedFrame& first_frame = *frames.front(); Timestamp receive_time = ReceiveTime(first_frame); if (first_frame.is_keyframe()) keyframe_required_ = false; // Gracefully handle bad RTP timestamps and render time issues. if (FrameHasBadRenderTiming(render_time, now, timing_->TargetVideoDelay())) { jitter_estimator_.Reset(); timing_->Reset(); render_time = timing_->RenderTime(first_frame.Timestamp(), now); } for (std::unique_ptr& frame : frames) { frame->SetRenderTime(render_time.ms()); superframe_delayed_by_retransmission |= frame->delayed_by_retransmission(); receive_time = std::max(receive_time, ReceiveTime(*frame)); superframe_size += DataSize::Bytes(frame->size()); } if (!superframe_delayed_by_retransmission) { auto frame_delay = inter_frame_delay_.CalculateDelay( first_frame.Timestamp(), receive_time); if (frame_delay) { jitter_estimator_.UpdateEstimate(*frame_delay, superframe_size); } float rtt_mult = protection_mode_ == kProtectionNackFEC ? 0.0 : 1.0; absl::optional rtt_mult_add_cap_ms = absl::nullopt; if (rtt_mult_settings_.has_value()) { rtt_mult = rtt_mult_settings_->rtt_mult_setting; rtt_mult_add_cap_ms = TimeDelta::Millis(rtt_mult_settings_->rtt_mult_add_cap_ms); } timing_->SetJitterDelay( jitter_estimator_.GetJitterEstimate(rtt_mult, rtt_mult_add_cap_ms)); timing_->UpdateCurrentDelay(render_time, now); } else if (RttMultExperiment::RttMultEnabled()) { jitter_estimator_.FrameNacked(); } // Update stats. UpdateDroppedFrames(); UpdateJitterDelay(); UpdateTimingFrameInfo(); std::unique_ptr frame = CombineAndDeleteFrames(std::move(frames)); timing_->SetLastDecodeScheduledTimestamp(now); decoder_ready_for_new_frame_ = false; // VideoReceiveStream2 wants frames on the decoder thread. decode_queue_->PostTask( SafeTask(decode_safety_, [this, frame = std::move(frame)]() mutable { receiver_->OnEncodedFrame(std::move(frame)); })); } void OnTimeout(TimeDelta delay) { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); // If the stream is paused then ignore the timeout. if (!decoder_ready_for_new_frame_) { timeout_tracker_.Stop(); return; } receiver_->OnDecodableFrameTimeout(delay); // Stop sending timeouts until receive starts waiting for a new frame. timeout_tracker_.Stop(); decoder_ready_for_new_frame_ = false; } private: void FrameReadyForDecode(uint32_t rtp_timestamp, Timestamp render_time) { RTC_DCHECK_RUN_ON(&worker_sequence_checker_); auto frames = buffer_->ExtractNextDecodableTemporalUnit(); RTC_DCHECK(frames[0]->Timestamp() == rtp_timestamp) << "Frame buffer's next decodable frame was not the one sent for " "extraction rtp=" << rtp_timestamp << " extracted rtp=" << frames[0]->Timestamp(); OnFrameReady(std::move(frames), render_time); } void UpdateDroppedFrames() RTC_RUN_ON(&worker_sequence_checker_) { const int dropped_frames = buffer_->GetTotalNumberOfDroppedFrames() - frames_dropped_before_last_new_frame_; if (dropped_frames > 0) stats_proxy_->OnDroppedFrames(dropped_frames); frames_dropped_before_last_new_frame_ = buffer_->GetTotalNumberOfDroppedFrames(); } void UpdateJitterDelay() { auto timings = timing_->GetTimings(); if (timings.num_decoded_frames) { stats_proxy_->OnFrameBufferTimingsUpdated( timings.max_decode_duration.ms(), timings.current_delay.ms(), timings.target_delay.ms(), timings.jitter_buffer_delay.ms(), timings.min_playout_delay.ms(), timings.render_delay.ms()); } } void UpdateTimingFrameInfo() { absl::optional info = timing_->GetTimingFrameInfo(); if (info) stats_proxy_->OnTimingFrameInfoUpdated(*info); } bool IsTooManyFramesQueued() const RTC_RUN_ON(&worker_sequence_checker_) { return buffer_->CurrentSize() > zero_playout_delay_max_decode_queue_size_; } void ForceKeyFrameReleaseImmediately() RTC_RUN_ON(&worker_sequence_checker_) { RTC_DCHECK(keyframe_required_); // Iterate through the frame buffer until there is a complete keyframe and // release this right away. while (buffer_->DecodableTemporalUnitsInfo()) { auto next_frame = buffer_->ExtractNextDecodableTemporalUnit(); if (next_frame.empty()) { RTC_DCHECK_NOTREACHED() << "Frame buffer should always return at least 1 frame."; continue; } // Found keyframe - decode right away. if (next_frame.front()->is_keyframe()) { auto render_time = timing_->RenderTime(next_frame.front()->Timestamp(), clock_->CurrentTime()); OnFrameReady(std::move(next_frame), render_time); return; } } } void MaybeScheduleFrameForRelease() RTC_RUN_ON(&worker_sequence_checker_) { auto decodable_tu_info = buffer_->DecodableTemporalUnitsInfo(); if (!decoder_ready_for_new_frame_ || !decodable_tu_info) { return; } if (keyframe_required_) { return ForceKeyFrameReleaseImmediately(); } // If already scheduled then abort. if (frame_decode_scheduler_->ScheduledRtpTimestamp() == decodable_tu_info->next_rtp_timestamp) { return; } TimeDelta max_wait = timeout_tracker_.TimeUntilTimeout(); // Ensures the frame is scheduled for decode before the stream times out. // This is otherwise a race condition. max_wait = std::max(max_wait - TimeDelta::Millis(1), TimeDelta::Zero()); absl::optional schedule; while (decodable_tu_info) { schedule = decode_timing_.OnFrameBufferUpdated( decodable_tu_info->next_rtp_timestamp, decodable_tu_info->last_rtp_timestamp, max_wait, IsTooManyFramesQueued()); if (schedule) { // Don't schedule if already waiting for the same frame. if (frame_decode_scheduler_->ScheduledRtpTimestamp() != decodable_tu_info->next_rtp_timestamp) { frame_decode_scheduler_->CancelOutstanding(); frame_decode_scheduler_->ScheduleFrame( decodable_tu_info->next_rtp_timestamp, *schedule, absl::bind_front(&FrameBuffer3Proxy::FrameReadyForDecode, this)); } return; } // If no schedule for current rtp, drop and try again. buffer_->DropNextDecodableTemporalUnit(); decodable_tu_info = buffer_->DecodableTemporalUnitsInfo(); } } RTC_NO_UNIQUE_ADDRESS SequenceChecker worker_sequence_checker_; const FieldTrialsView& field_trials_; const absl::optional rtt_mult_settings_ = RttMultExperiment::GetRttMultValue(); Clock* const clock_; TaskQueueBase* const worker_queue_; rtc::TaskQueue* const decode_queue_; VCMReceiveStatisticsCallback* const stats_proxy_; FrameSchedulingReceiver* const receiver_; VCMTiming* const timing_; const std::unique_ptr frame_decode_scheduler_ RTC_GUARDED_BY(&worker_sequence_checker_); JitterEstimator jitter_estimator_ RTC_GUARDED_BY(&worker_sequence_checker_); InterFrameDelay inter_frame_delay_ RTC_GUARDED_BY(&worker_sequence_checker_); bool keyframe_required_ RTC_GUARDED_BY(&worker_sequence_checker_) = false; std::unique_ptr buffer_ RTC_GUARDED_BY(&worker_sequence_checker_); FrameDecodeTiming decode_timing_ RTC_GUARDED_BY(&worker_sequence_checker_); VideoReceiveStreamTimeoutTracker timeout_tracker_ RTC_GUARDED_BY(&worker_sequence_checker_); int frames_dropped_before_last_new_frame_ RTC_GUARDED_BY(&worker_sequence_checker_) = 0; VCMVideoProtection protection_mode_ RTC_GUARDED_BY(&worker_sequence_checker_) = kProtectionNack; // This flag guards frames from queuing in front of the decoder. Without this // guard, encoded frames will not wait for the decoder to finish decoding a // frame and just queue up, meaning frames will not be dropped or // fast-forwarded when the decoder is slow or hangs. bool decoder_ready_for_new_frame_ RTC_GUARDED_BY(&worker_sequence_checker_) = false; // Maximum number of frames in the decode queue to allow pacing. If the // queue grows beyond the max limit, pacing will be disabled and frames will // be pushed to the decoder as soon as possible. This only has an effect // when the low-latency rendering path is active, which is indicated by // the frame's render time == 0. FieldTrialParameter zero_playout_delay_max_decode_queue_size_; rtc::scoped_refptr decode_safety_ = PendingTaskSafetyFlag::CreateDetached(); ScopedTaskSafety worker_safety_; }; enum class FrameBufferArm { kFrameBuffer2, kFrameBuffer3, kSyncDecode, }; constexpr const char* kFrameBufferFieldTrial = "WebRTC-FrameBuffer3"; FrameBufferArm ParseFrameBufferFieldTrial(const FieldTrialsView& field_trials) { webrtc::FieldTrialEnum arm( "arm", FrameBufferArm::kFrameBuffer2, { {"FrameBuffer2", FrameBufferArm::kFrameBuffer2}, {"FrameBuffer3", FrameBufferArm::kFrameBuffer3}, {"SyncDecoding", FrameBufferArm::kSyncDecode}, }); ParseFieldTrial({&arm}, field_trials.Lookup(kFrameBufferFieldTrial)); return arm.Get(); } } // namespace std::unique_ptr FrameBufferProxy::CreateFromFieldTrial( Clock* clock, TaskQueueBase* worker_queue, VCMTiming* timing, VCMReceiveStatisticsCallback* stats_proxy, rtc::TaskQueue* decode_queue, FrameSchedulingReceiver* receiver, TimeDelta max_wait_for_keyframe, TimeDelta max_wait_for_frame, DecodeSynchronizer* decode_sync, const FieldTrialsView& field_trials) { switch (ParseFrameBufferFieldTrial(field_trials)) { case FrameBufferArm::kFrameBuffer2: return std::make_unique( clock, timing, stats_proxy, decode_queue, receiver, max_wait_for_keyframe, max_wait_for_frame, field_trials); case FrameBufferArm::kSyncDecode: { std::unique_ptr scheduler; if (decode_sync) { scheduler = decode_sync->CreateSynchronizedFrameScheduler(); } else { RTC_LOG(LS_ERROR) << "In FrameBuffer with sync decode trial, but " "no DecodeSynchronizer was present!"; // Crash in debug, but in production use the task queue scheduler. RTC_DCHECK_NOTREACHED(); scheduler = std::make_unique( clock, worker_queue); } return std::make_unique( clock, worker_queue, timing, stats_proxy, decode_queue, receiver, max_wait_for_keyframe, max_wait_for_frame, std::move(scheduler), field_trials); } case FrameBufferArm::kFrameBuffer3: ABSL_FALLTHROUGH_INTENDED; default: { auto scheduler = std::make_unique(clock, worker_queue); return std::make_unique( clock, worker_queue, timing, stats_proxy, decode_queue, receiver, max_wait_for_keyframe, max_wait_for_frame, std::move(scheduler), field_trials); } } } } // namespace webrtc