FrameCadenceAdapter: now sets queue_overload based on encoder load

Measures the time consumed by OnFrame (e.g. the encoding time) and
sets an overload flag during N subsequent frames if the time is
longer than the current frame time. N is set to the number of
received frames on the network thread while being blocked by
encoding.

The queue overload mechanism for zero hertz can be disabled using the
WebRTC-ZeroHertzQueueOverload kill switch.

Also adds a UMA called WebRTC.Screenshare.ZeroHz.QueueOverload.

Bug: webrtc:15539
Change-Id: If81481c265d3e845485f79a2a1ac03dcbcc3ffc3
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/332381
Commit-Queue: Henrik Andreassson <henrika@webrtc.org>
Reviewed-by: Markus Handell <handellm@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#41489}

experiments/field_trials.py

@@ -125,6 +125,9 @@ ACTIVE_FIELD_TRIALS: FrozenSet[FieldTrial] = frozenset([
     FieldTrial('WebRTC-VideoEncoderSettings',
                'chromium:1406331',
                date(2024, 4, 1)),
+    FieldTrial('WebRTC-ZeroHertzQueueOverload',
+               'webrtc:332381',
+               date(2024, 7, 1)),
     # keep-sorted end
 ])  # yapf: disable
 

test/time_controller/simulated_time_controller.cc

@@ -218,9 +218,6 @@ void GlobalSimulatedTimeController::SkipForwardBy(TimeDelta duration) {
   impl_.AdvanceTime(target_time);
   sim_clock_.AdvanceTimeMicroseconds(duration.us());
   global_clock_.AdvanceTime(duration);
-
-  // Run tasks that were pending during the skip.
-  impl_.RunReadyRunners();
 }
 
 void GlobalSimulatedTimeController::Register(

test/time_controller/simulated_time_controller.h

@@ -139,7 +139,6 @@ class GlobalSimulatedTimeController : public TimeController {
   void AdvanceTime(TimeDelta duration) override;
 
   // Advances time by `duration`and do not run delayed tasks in the meantime.
-  // Runs any pending tasks at the end.
   // Useful for simulating contention on destination queues.
   void SkipForwardBy(TimeDelta duration);
 

test/time_controller/simulated_time_controller_unittest.cc

@@ -159,6 +159,8 @@ TEST(SimulatedTimeControllerTest, SkipsDelayedTaskForward) {
   }));
   main_thread->PostDelayedTask(fun.AsStdFunction(), shorter_duration);
   sim.SkipForwardBy(duration_during_which_nothing_runs);
+  // Run tasks that were pending during the skip.
+  sim.AdvanceTime(TimeDelta::Zero());
 }
 
 }  // namespace webrtc

video/frame_cadence_adapter.cc

@@ -103,7 +103,9 @@ class ZeroHertzAdapterMode : public AdapterMode {
   ZeroHertzAdapterMode(TaskQueueBase* queue,
                        Clock* clock,
                        FrameCadenceAdapterInterface::Callback* callback,
-                       double max_fps);
+                       double max_fps,
+                       std::atomic<int>& frames_scheduled_for_processing,
+                       bool zero_hertz_queue_overload);
   ~ZeroHertzAdapterMode() { refresh_frame_requester_.Stop(); }
 
   // Reconfigures according to parameters.
@@ -190,12 +192,20 @@ class ZeroHertzAdapterMode : public AdapterMode {
   // have arrived.
   void ProcessRepeatedFrameOnDelayedCadence(int frame_id)
       RTC_RUN_ON(sequence_checker_);
-  // Sends a frame, updating the timestamp to the current time.
-  void SendFrameNow(Timestamp post_time, const VideoFrame& frame) const
-      RTC_RUN_ON(sequence_checker_);
+  // Sends a frame, updating the timestamp to the current time. Also updates
+  // `queue_overload_count_` based on the time it takes to encode a frame and
+  // the amount of received frames while encoding. The `queue_overload`
+  // parameter in the OnFrame callback will be true while
+  // `queue_overload_count_` is larger than zero to allow the client to drop
+  // frames and thereby mitigate delay buildups.
+  // Repeated frames are sent with `post_time` set to absl::nullopt.
+  void SendFrameNow(absl::optional<Timestamp> post_time,
+                    const VideoFrame& frame) RTC_RUN_ON(sequence_checker_);
   // Returns the repeat duration depending on if it's an idle repeat or not.
   TimeDelta RepeatDuration(bool idle_repeat) const
       RTC_RUN_ON(sequence_checker_);
+  // Returns the frame duration taking potential restrictions into account.
+  TimeDelta FrameDuration() const RTC_RUN_ON(sequence_checker_);
   // Unless timer already running, starts repeatedly requesting refresh frames
   // after a grace_period. If a frame appears before the grace_period has
   // passed, the request is cancelled.
@@ -208,6 +218,14 @@ class ZeroHertzAdapterMode : public AdapterMode {
   // The configured max_fps.
   // TODO(crbug.com/1255737): support max_fps updates.
   const double max_fps_;
+
+  // Number of frames that are currently scheduled for processing on the
+  // `queue_`.
+  const std::atomic<int>& frames_scheduled_for_processing_;
+
+  // Can be used as kill-switch for the queue overload mechanism.
+  const bool zero_hertz_queue_overload_enabled_;
+
   // How much the incoming frame sequence is delayed by.
   const TimeDelta frame_delay_ = TimeDelta::Seconds(1) / max_fps_;
 
@@ -231,6 +249,9 @@ class ZeroHertzAdapterMode : public AdapterMode {
   // the max frame rate.
   absl::optional<TimeDelta> restricted_frame_delay_
       RTC_GUARDED_BY(sequence_checker_);
+  // Set in OnSendFrame to reflect how many future frames will be forwarded with
+  // the `queue_overload` flag set to true.
+  int queue_overload_count_ RTC_GUARDED_BY(sequence_checker_) = 0;
 
   ScopedTaskSafety safety_;
 };
@@ -359,6 +380,9 @@ class FrameCadenceAdapterImpl : public FrameCadenceAdapterInterface {
   // 0 Hz.
   const bool zero_hertz_screenshare_enabled_;
 
+  // Kill-switch for the queue overload mechanism in zero-hertz mode.
+  const bool frame_cadence_adapter_zero_hertz_queue_overload_enabled_;
+
   // The three possible modes we're under.
   absl::optional<PassthroughAdapterMode> passthrough_adapter_;
   absl::optional<ZeroHertzAdapterMode> zero_hertz_adapter_;
@@ -405,8 +429,15 @@ ZeroHertzAdapterMode::ZeroHertzAdapterMode(
     TaskQueueBase* queue,
     Clock* clock,
     FrameCadenceAdapterInterface::Callback* callback,
-    double max_fps)
-    : queue_(queue), clock_(clock), callback_(callback), max_fps_(max_fps) {
+    double max_fps,
+    std::atomic<int>& frames_scheduled_for_processing,
+    bool zero_hertz_queue_overload_enabled)
+    : queue_(queue),
+      clock_(clock),
+      callback_(callback),
+      max_fps_(max_fps),
+      frames_scheduled_for_processing_(frames_scheduled_for_processing),
+      zero_hertz_queue_overload_enabled_(zero_hertz_queue_overload_enabled) {
   sequence_checker_.Detach();
   MaybeStartRefreshFrameRequester();
 }
@@ -655,36 +686,70 @@ void ZeroHertzAdapterMode::ProcessRepeatedFrameOnDelayedCadence(int frame_id) {
 
   // Schedule another repeat before sending the frame off which could take time.
   ScheduleRepeat(frame_id, HasQualityConverged());
-  // Mark `post_time` with 0 to signal that this is a repeated frame.
-  SendFrameNow(Timestamp::Zero(), frame);
+  SendFrameNow(absl::nullopt, frame);
 }
 
-void ZeroHertzAdapterMode::SendFrameNow(Timestamp post_time,
-                                        const VideoFrame& frame) const {
+void ZeroHertzAdapterMode::SendFrameNow(absl::optional<Timestamp> post_time,
+                                        const VideoFrame& frame) {
   RTC_DCHECK_RUN_ON(&sequence_checker_);
   TRACE_EVENT0("webrtc", __func__);
-  Timestamp now = clock_->CurrentTime();
-  // Exclude repeated frames which are marked with zero as post time.
-  if (post_time != Timestamp::Zero()) {
-    TimeDelta delay = (now - post_time);
+
+  Timestamp encode_start_time = clock_->CurrentTime();
+  if (post_time.has_value()) {
+    TimeDelta delay = (encode_start_time - *post_time);
     RTC_HISTOGRAM_COUNTS_10000("WebRTC.Screenshare.ZeroHz.DelayMs", delay.ms());
   }
-  // TODO(crbug.com/1255737): ensure queue_overload is computed from current
-  // conditions on the encoder queue.
-  callback_->OnFrame(/*post_time=*/now,
-                     /*queue_overload=*/false, frame);
+
+  // Forward the frame and set `queue_overload` if is has been detected that it
+  // is not possible to deliver frames at the expected rate due to slow
+  // encoding.
+  callback_->OnFrame(/*post_time=*/encode_start_time, queue_overload_count_ > 0,
+                     frame);
+
+  // WebRTC-ZeroHertzQueueOverload kill-switch.
+  if (!zero_hertz_queue_overload_enabled_)
+    return;
+
+  // `queue_overload_count_` determines for how many future frames the
+  // `queue_overload` flag will be set and it is only increased if:
+  // o We are not already in an overload state.
+  // o New frames have been scheduled for processing on the queue while encoding
+  //   took place in OnFrame.
+  // o The duration of OnFrame is longer than the current frame duration.
+  // If all these conditions are fulfilled, `queue_overload_count_` is set to
+  // `frames_scheduled_for_processing_` and any pending repeat is canceled since
+  // new frames are available and the repeat is not needed.
+  // If the adapter is already in an overload state, simply decrease
+  // `queue_overload_count_` by one.
+  if (queue_overload_count_ == 0) {
+    const int frames_scheduled_for_processing =
+        frames_scheduled_for_processing_.load(std::memory_order_relaxed);
+    if (frames_scheduled_for_processing > 0) {
+      TimeDelta encode_time = clock_->CurrentTime() - encode_start_time;
+      if (encode_time > FrameDuration()) {
+        queue_overload_count_ = frames_scheduled_for_processing;
+        // Invalidates any outstanding repeat to avoid sending pending repeat
+        // directly after too long encode.
+        current_frame_id_++;
+      }
+    }
+  } else {
+    queue_overload_count_--;
+  }
+  RTC_HISTOGRAM_BOOLEAN("WebRTC.Screenshare.ZeroHz.QueueOverload",
+                        queue_overload_count_ > 0);
+}
+
+TimeDelta ZeroHertzAdapterMode::FrameDuration() const {
+  RTC_DCHECK_RUN_ON(&sequence_checker_);
+  return std::max(frame_delay_, restricted_frame_delay_.value_or(frame_delay_));
 }
 
 TimeDelta ZeroHertzAdapterMode::RepeatDuration(bool idle_repeat) const {
   RTC_DCHECK_RUN_ON(&sequence_checker_);
-  // By default use `frame_delay_` in non-idle repeat mode but use the
-  // restricted frame delay instead if it is set in
-  // UpdateVideoSourceRestrictions.
-  TimeDelta frame_delay =
-      std::max(frame_delay_, restricted_frame_delay_.value_or(frame_delay_));
   return idle_repeat
              ? FrameCadenceAdapterInterface::kZeroHertzIdleRepeatRatePeriod
-             : frame_delay;
+             : FrameDuration();
 }
 
 void ZeroHertzAdapterMode::MaybeStartRefreshFrameRequester() {
@@ -770,6 +835,8 @@ FrameCadenceAdapterImpl::FrameCadenceAdapterImpl(
       queue_(queue),
       zero_hertz_screenshare_enabled_(
           !field_trials.IsDisabled("WebRTC-ZeroHertzScreenshare")),
+      frame_cadence_adapter_zero_hertz_queue_overload_enabled_(
+          !field_trials.IsDisabled("WebRTC-ZeroHertzQueueOverload")),
       metronome_(metronome),
       worker_queue_(worker_queue) {}
 
@@ -943,8 +1010,10 @@ void FrameCadenceAdapterImpl::MaybeReconfigureAdapters(
     if (!was_zero_hertz_enabled || max_fps_has_changed) {
       RTC_LOG(LS_INFO) << "Zero hertz mode enabled (max_fps="
                        << source_constraints_->max_fps.value() << ")";
-      zero_hertz_adapter_.emplace(queue_, clock_, callback_,
-                                  source_constraints_->max_fps.value());
+      zero_hertz_adapter_.emplace(
+          queue_, clock_, callback_, source_constraints_->max_fps.value(),
+          frames_scheduled_for_processing_,
+          frame_cadence_adapter_zero_hertz_queue_overload_enabled_);
       zero_hertz_adapter_->UpdateVideoSourceRestrictions(
           restricted_max_frame_rate_);
       zero_hertz_adapter_created_timestamp_ = clock_->CurrentTime();

video/frame_cadence_adapter_unittest.cc

@@ -39,9 +39,11 @@ namespace {
 
 using ::testing::_;
 using ::testing::ElementsAre;
+using ::testing::InSequence;
 using ::testing::Invoke;
 using ::testing::InvokeWithoutArgs;
 using ::testing::Mock;
+using ::testing::NiceMock;
 using ::testing::Pair;
 using ::testing::Values;
 
@@ -310,6 +312,7 @@ TEST(FrameCadenceAdapterTest, DelayedProcessingUnderHeavyContention) {
   }));
   adapter->OnFrame(CreateFrame());
   time_controller.SkipForwardBy(time_skipped);
+  time_controller.AdvanceTime(TimeDelta::Zero());
 }
 
 TEST(FrameCadenceAdapterTest, RepeatsFramesDelayed) {
@@ -1155,5 +1158,166 @@ TEST(FrameCadenceAdapterRealTimeTest,
   finalized.Wait(rtc::Event::kForever);
 }
 
+class ZeroHertzQueueOverloadTest : public ::testing::Test {
+ public:
+  static constexpr int kMaxFps = 10;
+
+  ZeroHertzQueueOverloadTest() {
+    Initialize();
+    metrics::Reset();
+  }
+
+  void Initialize() {
+    adapter_->Initialize(&callback_);
+    adapter_->SetZeroHertzModeEnabled(
+        FrameCadenceAdapterInterface::ZeroHertzModeParams{
+            /*num_simulcast_layers=*/1});
+    adapter_->OnConstraintsChanged(
+        VideoTrackSourceConstraints{/*min_fps=*/0, kMaxFps});
+    time_controller_.AdvanceTime(TimeDelta::Zero());
+  }
+
+  void ScheduleDelayed(TimeDelta delay, absl::AnyInvocable<void() &&> task) {
+    TaskQueueBase::Current()->PostDelayedTask(std::move(task), delay);
+  }
+
+  void PassFrame() { adapter_->OnFrame(CreateFrame()); }
+
+  void AdvanceTime(TimeDelta duration) {
+    time_controller_.AdvanceTime(duration);
+  }
+
+  void SkipForwardBy(TimeDelta duration) {
+    time_controller_.SkipForwardBy(duration);
+  }
+
+  Timestamp CurrentTime() { return time_controller_.GetClock()->CurrentTime(); }
+
+ protected:
+  test::ScopedKeyValueConfig field_trials_;
+  NiceMock<MockCallback> callback_;
+  GlobalSimulatedTimeController time_controller_{Timestamp::Zero()};
+  std::unique_ptr<FrameCadenceAdapterInterface> adapter_{
+      CreateAdapter(field_trials_, time_controller_.GetClock())};
+};
+
+TEST_F(ZeroHertzQueueOverloadTest,
+       ForwardedFramesDuringTooLongEncodeTimeAreFlaggedWithQueueOverload) {
+  InSequence s;
+  PassFrame();
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).WillOnce(InvokeWithoutArgs([&] {
+    PassFrame();
+    PassFrame();
+    PassFrame();
+    SkipForwardBy(TimeDelta::Millis(301));
+  }));
+  EXPECT_CALL(callback_, OnFrame(_, true, _)).Times(3);
+  AdvanceTime(TimeDelta::Millis(100));
+  EXPECT_THAT(metrics::Samples("WebRTC.Screenshare.ZeroHz.QueueOverload"),
+              ElementsAre(Pair(false, 1), Pair(true, 3)));
+}
+
+TEST_F(ZeroHertzQueueOverloadTest,
+       ForwardedFramesAfterOverloadBurstAreNotFlaggedWithQueueOverload) {
+  InSequence s;
+  PassFrame();
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).WillOnce(InvokeWithoutArgs([&] {
+    PassFrame();
+    PassFrame();
+    PassFrame();
+    SkipForwardBy(TimeDelta::Millis(301));
+  }));
+  EXPECT_CALL(callback_, OnFrame(_, true, _)).Times(3);
+  AdvanceTime(TimeDelta::Millis(100));
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).Times(2);
+  PassFrame();
+  PassFrame();
+  AdvanceTime(TimeDelta::Millis(100));
+  EXPECT_THAT(metrics::Samples("WebRTC.Screenshare.ZeroHz.QueueOverload"),
+              ElementsAre(Pair(false, 3), Pair(true, 3)));
+}
+
+TEST_F(ZeroHertzQueueOverloadTest,
+       ForwardedFramesDuringNormalEncodeTimeAreNotFlaggedWithQueueOverload) {
+  InSequence s;
+  PassFrame();
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).WillOnce(InvokeWithoutArgs([&] {
+    PassFrame();
+    PassFrame();
+    PassFrame();
+    // Long but not too long encode time.
+    SkipForwardBy(TimeDelta::Millis(99));
+  }));
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).Times(3);
+  AdvanceTime(TimeDelta::Millis(199));
+  EXPECT_THAT(metrics::Samples("WebRTC.Screenshare.ZeroHz.QueueOverload"),
+              ElementsAre(Pair(false, 4)));
+}
+
+TEST_F(
+    ZeroHertzQueueOverloadTest,
+    AvoidSettingQueueOverloadAndSendRepeatWhenNoNewPacketsWhileTooLongEncode) {
+  // Receive one frame only and let OnFrame take such a long time that an
+  // overload normally is warranted. But the fact that no new frames arrive
+  // while being blocked should trigger a non-idle repeat to ensure that the
+  // video stream does not freeze and queue overload should be false.
+  PassFrame();
+  EXPECT_CALL(callback_, OnFrame(_, false, _))
+      .WillOnce(
+          InvokeWithoutArgs([&] { SkipForwardBy(TimeDelta::Millis(101)); }))
+      .WillOnce(InvokeWithoutArgs([&] {
+        // Non-idle repeat.
+        EXPECT_EQ(CurrentTime(), Timestamp::Zero() + TimeDelta::Millis(201));
+      }));
+  AdvanceTime(TimeDelta::Millis(100));
+  EXPECT_THAT(metrics::Samples("WebRTC.Screenshare.ZeroHz.QueueOverload"),
+              ElementsAre(Pair(false, 2)));
+}
+
+TEST_F(ZeroHertzQueueOverloadTest,
+       EnterFastRepeatAfterQueueOverloadWhenReceivedOnlyOneFrameDuringEncode) {
+  InSequence s;
+  // - Forward one frame frame during high load which triggers queue overload.
+  // - Receive only one new frame while being blocked and verify that the
+  //   cancelled repeat was for the first frame and not the second.
+  // - Fast repeat mode should happen after second frame.
+  PassFrame();
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).WillOnce(InvokeWithoutArgs([&] {
+    PassFrame();
+    SkipForwardBy(TimeDelta::Millis(101));
+  }));
+  EXPECT_CALL(callback_, OnFrame(_, true, _));
+  AdvanceTime(TimeDelta::Millis(100));
+
+  // Fast repeats should take place from here on.
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).Times(5);
+  AdvanceTime(TimeDelta::Millis(500));
+  EXPECT_THAT(metrics::Samples("WebRTC.Screenshare.ZeroHz.QueueOverload"),
+              ElementsAre(Pair(false, 6), Pair(true, 1)));
+}
+
+TEST_F(ZeroHertzQueueOverloadTest,
+       QueueOverloadIsDisabledForZeroHerzWhenKillSwitchIsEnabled) {
+  webrtc::test::ScopedKeyValueConfig field_trials(
+      field_trials_, "WebRTC-ZeroHertzQueueOverload/Disabled/");
+  adapter_.reset();
+  adapter_ = CreateAdapter(field_trials, time_controller_.GetClock());
+  Initialize();
+
+  // Same as ForwardedFramesDuringTooLongEncodeTimeAreFlaggedWithQueueOverload
+  // but this time the queue overload mechanism is disabled.
+  InSequence s;
+  PassFrame();
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).WillOnce(InvokeWithoutArgs([&] {
+    PassFrame();
+    PassFrame();
+    PassFrame();
+    SkipForwardBy(TimeDelta::Millis(301));
+  }));
+  EXPECT_CALL(callback_, OnFrame(_, false, _)).Times(3);
+  AdvanceTime(TimeDelta::Millis(100));
+  EXPECT_EQ(metrics::NumSamples("WebRTC.Screenshare.ZeroHz.QueueOverload"), 0);
+}
+
 }  // namespace
 }  // namespace webrtc