Add ability to control TaskQueuePacedSender holdback window.

Holdback window can be specified as absolute time and in terms of packet
send times. Example:
WebRTC-TaskQueuePacer/Enabled,holdback_window:20ms,holdback_packet:3/

If current conditions have us running with 2000kbps pacing rate and
1250byte (10kbit) packets, each packet send time is 5ms.
The holdback window would then be min(20ms, 3*5ms) = 15ms.

The default is like before 1ms and packets no take into account when
TQ pacer is used, parameters have no effect with legacy process thread
pacer.

Bug: webrtc:10809
Change-Id: I800de05107e2d4df461eabaaf1ca04fb4c5de51e
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/233421
Commit-Queue: Erik Språng <sprang@webrtc.org>
Reviewed-by: Philip Eliasson <philipel@webrtc.org>
Reviewed-by: Henrik Boström <hbos@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#35266}

call/rtp_transport_controller_send.cc

@@ -75,6 +75,15 @@ bool IsRelayed(const rtc::NetworkRoute& route) {
 
 }  // namespace
 
+RtpTransportControllerSend::PacerSettings::PacerSettings(
+    const WebRtcKeyValueConfig* trials)
+    : tq_disabled("Disabled"),
+      holdback_window("holdback_window", PacingController::kMinSleepTime),
+      holdback_packets("holdback_packets", -1) {
+  ParseFieldTrial({&tq_disabled, &holdback_window, &holdback_packets},
+                  trials->Lookup("WebRTC-TaskQueuePacer"));
+}
+
 RtpTransportControllerSend::RtpTransportControllerSend(
     Clock* clock,
     webrtc::RtcEventLog* event_log,
@@ -89,8 +98,8 @@ RtpTransportControllerSend::RtpTransportControllerSend(
       bitrate_configurator_(bitrate_config),
       pacer_started_(false),
       process_thread_(std::move(process_thread)),
-      use_task_queue_pacer_(!IsDisabled(trials, "WebRTC-TaskQueuePacer")),
+      pacer_settings_(trials),
-      process_thread_pacer_(use_task_queue_pacer_
+      process_thread_pacer_(pacer_settings_.use_task_queue_pacer()
                                 ? nullptr
                                 : new PacedSender(clock,
                                                   &packet_router_,
@@ -98,14 +107,14 @@ RtpTransportControllerSend::RtpTransportControllerSend(
                                                   trials,
                                                   process_thread_.get())),
       task_queue_pacer_(
-          use_task_queue_pacer_
+          pacer_settings_.use_task_queue_pacer()
-              ? new TaskQueuePacedSender(
+              ? new TaskQueuePacedSender(clock,
-                    clock,
+                                         &packet_router_,
-                    &packet_router_,
+                                         event_log,
-                    event_log,
+                                         trials,
-                    trials,
+                                         task_queue_factory,
-                    task_queue_factory,
+                                         pacer_settings_.holdback_window.Get(),
-                    /*hold_back_window = */ PacingController::kMinSleepTime)
+                                         pacer_settings_.holdback_packets.Get())
               : nullptr),
       observer_(nullptr),
       controller_factory_override_(controller_factory),
@@ -194,14 +203,14 @@ void RtpTransportControllerSend::UpdateControlState() {
 }
 
 RtpPacketPacer* RtpTransportControllerSend::pacer() {
-  if (use_task_queue_pacer_) {
+  if (pacer_settings_.use_task_queue_pacer()) {
     return task_queue_pacer_.get();
   }
   return process_thread_pacer_.get();
 }
 
 const RtpPacketPacer* RtpTransportControllerSend::pacer() const {
-  if (use_task_queue_pacer_) {
+  if (pacer_settings_.use_task_queue_pacer()) {
     return task_queue_pacer_.get();
   }
   return process_thread_pacer_.get();
@@ -226,7 +235,7 @@ RtpTransportControllerSend::transport_feedback_observer() {
 }
 
 RtpPacketSender* RtpTransportControllerSend::packet_sender() {
-  if (use_task_queue_pacer_) {
+  if (pacer_settings_.use_task_queue_pacer()) {
     return task_queue_pacer_.get();
   }
   return process_thread_pacer_.get();
@@ -503,7 +512,7 @@ void RtpTransportControllerSend::IncludeOverheadInPacedSender() {
 void RtpTransportControllerSend::EnsureStarted() {
   if (!pacer_started_) {
     pacer_started_ = true;
-    if (use_task_queue_pacer_) {
+    if (pacer_settings_.use_task_queue_pacer()) {
       task_queue_pacer_->EnsureStarted();
     } else {
       process_thread_->Start();

call/rtp_transport_controller_send.h

@@ -128,6 +128,16 @@ class RtpTransportControllerSend final
   void OnRemoteNetworkEstimate(NetworkStateEstimate estimate) override;
 
  private:
+  struct PacerSettings {
+    explicit PacerSettings(const WebRtcKeyValueConfig* trials);
+
+    bool use_task_queue_pacer() const { return !tq_disabled.Get(); }
+
+    FieldTrialFlag tq_disabled;  // Kill-switch not normally used.
+    FieldTrialParameter<TimeDelta> holdback_window;
+    FieldTrialParameter<int> holdback_packets;
+  };
+
   void MaybeCreateControllers() RTC_RUN_ON(task_queue_);
   void UpdateInitialConstraints(TargetRateConstraints new_contraints)
       RTC_RUN_ON(task_queue_);
@@ -158,7 +168,7 @@ class RtpTransportControllerSend final
   std::map<std::string, rtc::NetworkRoute> network_routes_;
   bool pacer_started_;
   const std::unique_ptr<ProcessThread> process_thread_;
-  const bool use_task_queue_pacer_;
+  const PacerSettings pacer_settings_;
   std::unique_ptr<PacedSender> process_thread_pacer_;
   std::unique_ptr<TaskQueuePacedSender> task_queue_pacer_;
 

modules/pacing/BUILD.gn

@@ -48,6 +48,7 @@ rtc_library("pacing") {
     "../../logging:rtc_event_pacing",
     "../../rtc_base:checks",
     "../../rtc_base:rtc_base_approved",
+    "../../rtc_base:rtc_numerics",
     "../../rtc_base:rtc_task_queue",
     "../../rtc_base/experiments:field_trial_parser",
     "../../rtc_base/synchronization:mutex",

modules/pacing/pacing_controller.h

@@ -103,6 +103,7 @@ class PacingController {
 
   // Sets the pacing rates. Must be called once before packets can be sent.
   void SetPacingRates(DataRate pacing_rate, DataRate padding_rate);
+  DataRate pacing_rate() const { return pacing_bitrate_; }
 
   // Currently audio traffic is not accounted by pacer and passed through.
   // With the introduction of audio BWE audio traffic will be accounted for

modules/pacing/task_queue_paced_sender.cc

@@ -36,9 +36,11 @@ TaskQueuePacedSender::TaskQueuePacedSender(
     RtcEventLog* event_log,
     const WebRtcKeyValueConfig* field_trials,
     TaskQueueFactory* task_queue_factory,
-    TimeDelta hold_back_window)
+    TimeDelta max_hold_back_window,
+    int max_hold_back_window_in_packets)
     : clock_(clock),
-      hold_back_window_(hold_back_window),
+      max_hold_back_window_(max_hold_back_window),
+      max_hold_back_window_in_packets_(max_hold_back_window_in_packets),
       pacing_controller_(clock,
                          packet_sender,
                          event_log,
@@ -48,9 +50,12 @@ TaskQueuePacedSender::TaskQueuePacedSender(
       stats_update_scheduled_(false),
       last_stats_time_(Timestamp::MinusInfinity()),
       is_shutdown_(false),
+      packet_size_(/*alpha=*/0.95),
       task_queue_(task_queue_factory->CreateTaskQueue(
           "TaskQueuePacedSender",
-          TaskQueueFactory::Priority::NORMAL)) {}
+          TaskQueueFactory::Priority::NORMAL)) {
+  packet_size_.Apply(1, 0);
+}
 
 TaskQueuePacedSender::~TaskQueuePacedSender() {
   // Post an immediate task to mark the queue as shutting down.
@@ -144,6 +149,7 @@ void TaskQueuePacedSender::EnqueuePackets(
   task_queue_.PostTask([this, packets_ = std::move(packets)]() mutable {
     RTC_DCHECK_RUN_ON(&task_queue_);
     for (auto& packet : packets_) {
+      packet_size_.Apply(1, packet->size());
       RTC_DCHECK_GE(packet->capture_time_ms(), 0);
       pacing_controller_.EnqueuePacket(std::move(packet));
     }
@@ -227,6 +233,17 @@ void TaskQueuePacedSender::MaybeProcessPackets(
     next_process_time = pacing_controller_.NextSendTime();
   }
 
+  TimeDelta hold_back_window = max_hold_back_window_;
+  DataRate pacing_rate = pacing_controller_.pacing_rate();
+  DataSize avg_packet_size = DataSize::Bytes(packet_size_.filtered());
+  if (max_hold_back_window_in_packets_ > 0 && !pacing_rate.IsZero() &&
+      !avg_packet_size.IsZero()) {
+    TimeDelta avg_packet_send_time = avg_packet_size / pacing_rate;
+    hold_back_window =
+        std::min(hold_back_window,
+                 avg_packet_send_time * max_hold_back_window_in_packets_);
+  }
+
   absl::optional<TimeDelta> time_to_next_process;
   if (pacing_controller_.IsProbing() &&
       next_process_time != next_process_time_) {
@@ -241,11 +258,11 @@ void TaskQueuePacedSender::MaybeProcessPackets(
                    (next_process_time - now).RoundDownTo(TimeDelta::Millis(1)));
     }
   } else if (next_process_time_.IsMinusInfinity() ||
-             next_process_time <= next_process_time_ - hold_back_window_) {
+             next_process_time <= next_process_time_ - hold_back_window) {
     // Schedule a new task since there is none currently scheduled
     // (`next_process_time_` is infinite), or the new process time is at least
     // one holdback window earlier than whatever is currently scheduled.
-    time_to_next_process = std::max(next_process_time - now, hold_back_window_);
+    time_to_next_process = std::max(next_process_time - now, hold_back_window);
   }
 
   if (time_to_next_process) {

modules/pacing/task_queue_paced_sender.h

@@ -29,6 +29,7 @@
 #include "modules/pacing/pacing_controller.h"
 #include "modules/pacing/rtp_packet_pacer.h"
 #include "modules/rtp_rtcp/source/rtp_packet_to_send.h"
+#include "rtc_base/numerics/exp_filter.h"
 #include "rtc_base/synchronization/mutex.h"
 #include "rtc_base/task_queue.h"
 #include "rtc_base/thread_annotations.h"
@@ -43,14 +44,15 @@ class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender {
   // there is currently a pacer queue and packets can't immediately be
   // processed. Increasing this reduces thread wakeups at the expense of higher
   // latency.
-  // TODO(bugs.webrtc.org/10809): Remove default value for hold_back_window.
+  // TODO(bugs.webrtc.org/10809): Remove default values.
   TaskQueuePacedSender(
       Clock* clock,
       PacingController::PacketSender* packet_sender,
       RtcEventLog* event_log,
       const WebRtcKeyValueConfig* field_trials,
       TaskQueueFactory* task_queue_factory,
-      TimeDelta hold_back_window = PacingController::kMinSleepTime);
+      TimeDelta max_hold_back_window = PacingController::kMinSleepTime,
+      int max_hold_back_window_in_packets = -1);
 
   ~TaskQueuePacedSender() override;
 
@@ -132,7 +134,9 @@ class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender {
   Stats GetStats() const;
 
   Clock* const clock_;
-  const TimeDelta hold_back_window_;
+  const TimeDelta max_hold_back_window_;
+  const int max_hold_back_window_in_packets_;
+
   PacingController pacing_controller_ RTC_GUARDED_BY(task_queue_);
 
   // We want only one (valid) delayed process task in flight at a time.
@@ -161,6 +165,9 @@ class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender {
   // never drain.
   bool is_shutdown_ RTC_GUARDED_BY(task_queue_);
 
+  // Filtered size of enqueued packets, in bytes.
+  rtc::ExpFilter packet_size_ RTC_GUARDED_BY(task_queue_);
+
   mutable Mutex stats_mutex_;
   Stats current_stats_ RTC_GUARDED_BY(stats_mutex_);