Extract most of PacedSender into PacedSendingController.

The Pacer now just handles interaction with Module/ProcessThread and
forwarding packets to PacketRouter.
All other logic is moved to PacedSendingController, including tests.
PacedSender unittest are now just some basic sanity tests.

Bug: webrtc:10809
Change-Id: I69223cd9d8300997375b03706d2e99c88e46241c
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/149041
Reviewed-by: Sebastian Jansson <srte@webrtc.org>
Commit-Queue: Erik Språng <sprang@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#28886}

modules/pacing/BUILD.gn

@@ -19,6 +19,8 @@ rtc_static_library("pacing") {
     "bitrate_prober.h",
     "paced_sender.cc",
     "paced_sender.h",
+    "pacing_controller.cc",
+    "pacing_controller.h",
     "packet_router.cc",
     "packet_router.h",
     "round_robin_packet_queue.cc",
@@ -75,11 +77,13 @@ if (rtc_include_tests) {
       "bitrate_prober_unittest.cc",
       "interval_budget_unittest.cc",
       "paced_sender_unittest.cc",
+      "pacing_controller_unittest.cc",
       "packet_router_unittest.cc",
     ]
     deps = [
       ":interval_budget",
       ":pacing",
+      "../../api/units:data_rate",
       "../../api/units:time_delta",
       "../../rtc_base:checks",
       "../../rtc_base:rtc_base_approved",

modules/pacing/paced_sender.cc

@@ -16,8 +16,6 @@
 
 #include "absl/memory/memory.h"
 #include "api/rtc_event_log/rtc_event_log.h"
-#include "modules/pacing/bitrate_prober.h"
-#include "modules/pacing/interval_budget.h"
 #include "modules/utility/include/process_thread.h"
 #include "rtc_base/checks.h"
 #include "rtc_base/logging.h"
@@ -25,50 +23,6 @@
 #include "system_wrappers/include/clock.h"
 
 namespace webrtc {
-namespace {
-// Time limit in milliseconds between packet bursts.
-constexpr TimeDelta kDefaultMinPacketLimit = TimeDelta::Millis<5>();
-constexpr TimeDelta kCongestedPacketInterval = TimeDelta::Millis<500>();
-constexpr TimeDelta kPausedProcessInterval = kCongestedPacketInterval;
-constexpr TimeDelta kMaxElapsedTime = TimeDelta::Seconds<2>();
-
-// Upper cap on process interval, in case process has not been called in a long
-// time.
-constexpr TimeDelta kMaxProcessingInterval = TimeDelta::Millis<30>();
-
-bool IsDisabled(const WebRtcKeyValueConfig& field_trials,
-                absl::string_view key) {
-  return field_trials.Lookup(key).find("Disabled") == 0;
-}
-
-bool IsEnabled(const WebRtcKeyValueConfig& field_trials,
-               absl::string_view key) {
-  return field_trials.Lookup(key).find("Enabled") == 0;
-}
-
-int GetPriorityForType(RtpPacketToSend::Type type) {
-  switch (type) {
-    case RtpPacketToSend::Type::kAudio:
-      // Audio is always prioritized over other packet types.
-      return 0;
-    case RtpPacketToSend::Type::kRetransmission:
-      // Send retransmissions before new media.
-      return 1;
-    case RtpPacketToSend::Type::kVideo:
-      // Video has "normal" priority, in the old speak.
-      return 2;
-    case RtpPacketToSend::Type::kForwardErrorCorrection:
-      // Send redundancy concurrently to video. If it is delayed it might have a
-      // lower chance of being useful.
-      return 2;
-    case RtpPacketToSend::Type::kPadding:
-      // Packets that are in themselves likely useless, only sent to keep the
-      // BWE high.
-      return 3;
-  }
-}
-
-}  // namespace
 const int64_t PacedSender::kMaxQueueLengthMs = 2000;
 const float PacedSender::kDefaultPaceMultiplier = 2.5f;
 
@@ -76,60 +30,24 @@ PacedSender::PacedSender(Clock* clock,
                          PacketRouter* packet_router,
                          RtcEventLog* event_log,
                          const WebRtcKeyValueConfig* field_trials)
-    : clock_(clock),
+    : pacing_controller_(clock,
+                         static_cast<PacingController::PacketSender*>(this),
+                         event_log,
+                         field_trials),
       packet_router_(packet_router),
-      fallback_field_trials_(
+      process_thread_(nullptr) {}
-          !field_trials ? absl::make_unique<FieldTrialBasedConfig>() : nullptr),
-      field_trials_(field_trials ? field_trials : fallback_field_trials_.get()),
-      drain_large_queues_(
-          !IsDisabled(*field_trials_, "WebRTC-Pacer-DrainQueue")),
-      send_padding_if_silent_(
-          IsEnabled(*field_trials_, "WebRTC-Pacer-PadInSilence")),
-      pace_audio_(!IsDisabled(*field_trials_, "WebRTC-Pacer-BlockAudio")),
-      min_packet_limit_(kDefaultMinPacketLimit),
-      last_timestamp_(clock_->CurrentTime()),
-      paused_(false),
-      media_budget_(0),
-      padding_budget_(0),
-      prober_(*field_trials_),
-      probing_send_failure_(false),
-      pacing_bitrate_(DataRate::Zero()),
-      time_last_process_(clock->CurrentTime()),
-      last_send_time_(time_last_process_),
-      packets_(time_last_process_, field_trials),
-      packet_counter_(0),
-      congestion_window_size_(DataSize::PlusInfinity()),
-      outstanding_data_(DataSize::Zero()),
-      process_thread_(nullptr),
-      queue_time_limit(TimeDelta::ms(kMaxQueueLengthMs)),
-      account_for_audio_(false),
-      legacy_packet_referencing_(
-          IsEnabled(*field_trials_, "WebRTC-Pacer-LegacyPacketReferencing")) {
-  if (!drain_large_queues_) {
-    RTC_LOG(LS_WARNING) << "Pacer queues will not be drained,"
-                           "pushback experiment must be enabled.";
-  }
-  FieldTrialParameter<int> min_packet_limit_ms("", min_packet_limit_.ms());
-  ParseFieldTrial({&min_packet_limit_ms},
-                  field_trials_->Lookup("WebRTC-Pacer-MinPacketLimitMs"));
-  min_packet_limit_ = TimeDelta::ms(min_packet_limit_ms.Get());
-  UpdateBudgetWithElapsedTime(min_packet_limit_);
-}
 
-PacedSender::~PacedSender() {}
+PacedSender::~PacedSender() = default;
 
 void PacedSender::CreateProbeCluster(DataRate bitrate, int cluster_id) {
   rtc::CritScope cs(&critsect_);
-  prober_.CreateProbeCluster(bitrate.bps(), CurrentTime().ms(), cluster_id);
+  return pacing_controller_.CreateProbeCluster(bitrate, cluster_id);
 }
 
 void PacedSender::Pause() {
   {
     rtc::CritScope cs(&critsect_);
-    if (!paused_)
+    pacing_controller_.Pause();
-      RTC_LOG(LS_INFO) << "PacedSender paused.";
-    paused_ = true;
-    packets_.SetPauseState(true, CurrentTime());
   }
   rtc::CritScope cs(&process_thread_lock_);
   // Tell the process thread to call our TimeUntilNextProcess() method to get
@@ -141,10 +59,7 @@ void PacedSender::Pause() {
 void PacedSender::Resume() {
   {
     rtc::CritScope cs(&critsect_);
-    if (paused_)
+    pacing_controller_.Resume();
-      RTC_LOG(LS_INFO) << "PacedSender resumed.";
-    paused_ = false;
-    packets_.SetPauseState(false, CurrentTime());
   }
   rtc::CritScope cs(&process_thread_lock_);
   // Tell the process thread to call our TimeUntilNextProcess() method to
@@ -155,49 +70,22 @@ void PacedSender::Resume() {
 
 void PacedSender::SetCongestionWindow(DataSize congestion_window_size) {
   rtc::CritScope cs(&critsect_);
-  congestion_window_size_ = congestion_window_size;
+  pacing_controller_.SetCongestionWindow(congestion_window_size);
 }
 
 void PacedSender::UpdateOutstandingData(DataSize outstanding_data) {
   rtc::CritScope cs(&critsect_);
-  outstanding_data_ = outstanding_data;
+  pacing_controller_.UpdateOutstandingData(outstanding_data);
-}
-
-bool PacedSender::Congested() const {
-  if (congestion_window_size_.IsFinite()) {
-    return outstanding_data_ >= congestion_window_size_;
-  }
-  return false;
-}
-
-Timestamp PacedSender::CurrentTime() const {
-  Timestamp time = clock_->CurrentTime();
-  if (time < last_timestamp_) {
-    RTC_LOG(LS_WARNING)
-        << "Non-monotonic clock behavior observed. Previous timestamp: "
-        << last_timestamp_.ms() << ", new timestamp: " << time.ms();
-    RTC_DCHECK_GE(time, last_timestamp_);
-    time = last_timestamp_;
-  }
-  last_timestamp_ = time;
-  return time;
 }
 
 void PacedSender::SetProbingEnabled(bool enabled) {
   rtc::CritScope cs(&critsect_);
-  RTC_CHECK_EQ(0, packet_counter_);
+  pacing_controller_.SetProbingEnabled(enabled);
-  prober_.SetEnabled(enabled);
 }
 
 void PacedSender::SetPacingRates(DataRate pacing_rate, DataRate padding_rate) {
   rtc::CritScope cs(&critsect_);
-  RTC_DCHECK_GT(pacing_rate, DataRate::Zero());
+  pacing_controller_.SetPacingRates(pacing_rate, padding_rate);
-  pacing_bitrate_ = pacing_rate;
-  padding_budget_.set_target_rate_kbps(padding_rate.kbps());
-
-  RTC_LOG(LS_VERBOSE) << "bwe:pacer_updated pacing_kbps="
-                      << pacing_bitrate_.kbps()
-                      << " padding_budget_kbps=" << padding_rate.kbps();
 }
 
 void PacedSender::InsertPacket(RtpPacketSender::Priority priority,
@@ -207,288 +95,69 @@ void PacedSender::InsertPacket(RtpPacketSender::Priority priority,
                                size_t bytes,
                                bool retransmission) {
   rtc::CritScope cs(&critsect_);
-  RTC_DCHECK(pacing_bitrate_ > DataRate::Zero())
+  pacing_controller_.InsertPacket(priority, ssrc, sequence_number,
-      << "SetPacingRate must be called before InsertPacket.";
+                                  capture_time_ms, bytes, retransmission);
-
-  Timestamp now = CurrentTime();
-  prober_.OnIncomingPacket(bytes);
-
-  if (capture_time_ms < 0)
-    capture_time_ms = now.ms();
-
-  RtpPacketToSend::Type type;
-  switch (priority) {
-    case RtpPacketSender::kHighPriority:
-      type = RtpPacketToSend::Type::kAudio;
-      break;
-    case RtpPacketSender::kNormalPriority:
-      type = RtpPacketToSend::Type::kRetransmission;
-      break;
-    default:
-      type = RtpPacketToSend::Type::kVideo;
-  }
-  packets_.Push(GetPriorityForType(type), type, ssrc, sequence_number,
-                capture_time_ms, now, DataSize::bytes(bytes), retransmission,
-                packet_counter_++);
 }
 
 void PacedSender::EnqueuePacket(std::unique_ptr<RtpPacketToSend> packet) {
   rtc::CritScope cs(&critsect_);
-  RTC_DCHECK(pacing_bitrate_ > DataRate::Zero())
+  pacing_controller_.EnqueuePacket(std::move(packet));
-      << "SetPacingRate must be called before InsertPacket.";
-
-  Timestamp now = CurrentTime();
-  prober_.OnIncomingPacket(packet->payload_size());
-
-  if (packet->capture_time_ms() < 0) {
-    packet->set_capture_time_ms(now.ms());
-  }
-
-  RTC_CHECK(packet->packet_type());
-  int priority = GetPriorityForType(*packet->packet_type());
-  packets_.Push(priority, now, packet_counter_++, std::move(packet));
 }
 
 void PacedSender::SetAccountForAudioPackets(bool account_for_audio) {
   rtc::CritScope cs(&critsect_);
-  account_for_audio_ = account_for_audio;
+  pacing_controller_.SetAccountForAudioPackets(account_for_audio);
 }
 
 TimeDelta PacedSender::ExpectedQueueTime() const {
   rtc::CritScope cs(&critsect_);
-  RTC_DCHECK_GT(pacing_bitrate_, DataRate::Zero());
+  return pacing_controller_.ExpectedQueueTime();
-  return TimeDelta::ms(
-      (QueueSizeData().bytes() * 8 * rtc::kNumMillisecsPerSec) /
-      pacing_bitrate_.bps());
 }
 
 size_t PacedSender::QueueSizePackets() const {
   rtc::CritScope cs(&critsect_);
-  return packets_.SizeInPackets();
+  return pacing_controller_.QueueSizePackets();
 }
 
 DataSize PacedSender::QueueSizeData() const {
   rtc::CritScope cs(&critsect_);
-  return packets_.Size();
+  return pacing_controller_.QueueSizeData();
 }
 
 absl::optional<Timestamp> PacedSender::FirstSentPacketTime() const {
   rtc::CritScope cs(&critsect_);
-  return first_sent_packet_time_;
+  return pacing_controller_.FirstSentPacketTime();
 }
 
 TimeDelta PacedSender::OldestPacketWaitTime() const {
   rtc::CritScope cs(&critsect_);
-  Timestamp oldest_packet = packets_.OldestEnqueueTime();
+  return pacing_controller_.OldestPacketWaitTime();
-  if (oldest_packet.IsInfinite()) {
-    return TimeDelta::Zero();
-  }
-
-  return CurrentTime() - oldest_packet;
 }
 
 int64_t PacedSender::TimeUntilNextProcess() {
   rtc::CritScope cs(&critsect_);
-  TimeDelta elapsed_time = CurrentTime() - time_last_process_;
+
   // When paused we wake up every 500 ms to send a padding packet to ensure
   // we won't get stuck in the paused state due to no feedback being received.
-  if (paused_) {
+  TimeDelta elapsed_time = pacing_controller_.TimeElapsedSinceLastProcess();
-    return std::max(kPausedProcessInterval - elapsed_time, TimeDelta::Zero())
+  if (pacing_controller_.IsPaused()) {
+    return std::max(PacingController::kPausedProcessInterval - elapsed_time,
+                    TimeDelta::Zero())
         .ms();
   }
 
-  if (prober_.IsProbing()) {
+  auto next_probe = pacing_controller_.TimeUntilNextProbe();
-    int64_t ret = prober_.TimeUntilNextProbe(CurrentTime().ms());
+  if (next_probe) {
-    if (ret > 0 || (ret == 0 && !probing_send_failure_))
+    return next_probe->ms();
-      return ret;
-  }
-  return std::max(min_packet_limit_ - elapsed_time, TimeDelta::Zero()).ms();
   }
 
-TimeDelta PacedSender::UpdateTimeAndGetElapsed(Timestamp now) {
+  const TimeDelta min_packet_limit = TimeDelta::ms(5);
-  TimeDelta elapsed_time = now - time_last_process_;
+  return std::max(min_packet_limit - elapsed_time, TimeDelta::Zero()).ms();
-  time_last_process_ = now;
-  if (elapsed_time > kMaxElapsedTime) {
-    RTC_LOG(LS_WARNING) << "Elapsed time (" << elapsed_time.ms()
-                        << " ms) longer than expected, limiting to "
-                        << kMaxElapsedTime.ms();
-    elapsed_time = kMaxElapsedTime;
-  }
-  return elapsed_time;
-}
-
-bool PacedSender::ShouldSendKeepalive(Timestamp now) const {
-  if (send_padding_if_silent_ || paused_ || Congested()) {
-    // We send a padding packet every 500 ms to ensure we won't get stuck in
-    // congested state due to no feedback being received.
-    TimeDelta elapsed_since_last_send = now - last_send_time_;
-    if (elapsed_since_last_send >= kCongestedPacketInterval) {
-      // We can not send padding unless a normal packet has first been sent. If
-      // we do, timestamps get messed up.
-      if (packet_counter_ > 0) {
-        return true;
-      }
-    }
-  }
-  return false;
 }
 
 void PacedSender::Process() {
   rtc::CritScope cs(&critsect_);
-  Timestamp now = CurrentTime();
+  pacing_controller_.ProcessPackets();
-  TimeDelta elapsed_time = UpdateTimeAndGetElapsed(now);
-  if (ShouldSendKeepalive(now)) {
-    if (legacy_packet_referencing_) {
-      critsect_.Leave();
-      size_t bytes_sent =
-          packet_router_->TimeToSendPadding(1, PacedPacketInfo());
-      critsect_.Enter();
-      OnPaddingSent(DataSize::bytes(bytes_sent));
-    } else {
-      DataSize keepalive_data_sent = DataSize::Zero();
-      critsect_.Leave();
-      std::vector<std::unique_ptr<RtpPacketToSend>> keepalive_packets =
-          packet_router_->GeneratePadding(1);
-      for (auto& packet : keepalive_packets) {
-        keepalive_data_sent +=
-            DataSize::bytes(packet->payload_size() + packet->padding_size());
-        packet_router_->SendPacket(std::move(packet), PacedPacketInfo());
-      }
-      critsect_.Enter();
-      OnPaddingSent(keepalive_data_sent);
-    }
-  }
-
-  if (paused_)
-    return;
-
-  if (elapsed_time > TimeDelta::Zero()) {
-    DataRate target_rate = pacing_bitrate_;
-    DataSize queue_size_data = packets_.Size();
-    if (queue_size_data > DataSize::Zero()) {
-      // Assuming equal size packets and input/output rate, the average packet
-      // has avg_time_left_ms left to get queue_size_bytes out of the queue, if
-      // time constraint shall be met. Determine bitrate needed for that.
-      packets_.UpdateQueueTime(CurrentTime());
-      if (drain_large_queues_) {
-        TimeDelta avg_time_left = std::max(
-            TimeDelta::ms(1), queue_time_limit - packets_.AverageQueueTime());
-        DataRate min_rate_needed = queue_size_data / avg_time_left;
-        if (min_rate_needed > target_rate) {
-          target_rate = min_rate_needed;
-          RTC_LOG(LS_VERBOSE) << "bwe:large_pacing_queue pacing_rate_kbps="
-                              << target_rate.kbps();
-        }
-      }
-    }
-
-    media_budget_.set_target_rate_kbps(target_rate.kbps());
-    UpdateBudgetWithElapsedTime(elapsed_time);
-  }
-
-  bool is_probing = prober_.IsProbing();
-  PacedPacketInfo pacing_info;
-  absl::optional<DataSize> recommended_probe_size;
-  if (is_probing) {
-    pacing_info = prober_.CurrentCluster();
-    recommended_probe_size = DataSize::bytes(prober_.RecommendedMinProbeSize());
-  }
-
-  DataSize data_sent = DataSize::Zero();
-  // The paused state is checked in the loop since it leaves the critical
-  // section allowing the paused state to be changed from other code.
-  while (!paused_) {
-    auto* packet = GetPendingPacket(pacing_info);
-    if (packet == nullptr) {
-      // No packet available to send, check if we should send padding.
-      if (!legacy_packet_referencing_) {
-        DataSize padding_to_add =
-            PaddingToAdd(recommended_probe_size, data_sent);
-        if (padding_to_add > DataSize::Zero()) {
-          critsect_.Leave();
-          std::vector<std::unique_ptr<RtpPacketToSend>> padding_packets =
-              packet_router_->GeneratePadding(padding_to_add.bytes());
-          critsect_.Enter();
-          if (padding_packets.empty()) {
-            // No padding packets were generated, quite send loop.
-            break;
-          }
-          for (auto& packet : padding_packets) {
-            EnqueuePacket(std::move(packet));
-          }
-          // Continue loop to send the padding that was just added.
-          continue;
-        }
-      }
-
-      // Can't fetch new packet and no padding to send, exit send loop.
-      break;
-    }
-
-    std::unique_ptr<RtpPacketToSend> rtp_packet = packet->ReleasePacket();
-    const bool owned_rtp_packet = rtp_packet != nullptr;
-    RtpPacketSendResult success;
-
-    if (rtp_packet != nullptr) {
-      critsect_.Leave();
-      packet_router_->SendPacket(std::move(rtp_packet), pacing_info);
-      critsect_.Enter();
-      success = RtpPacketSendResult::kSuccess;
-    } else {
-      critsect_.Leave();
-      success = packet_router_->TimeToSendPacket(
-          packet->ssrc(), packet->sequence_number(), packet->capture_time_ms(),
-          packet->is_retransmission(), pacing_info);
-      critsect_.Enter();
-    }
-
-    if (success == RtpPacketSendResult::kSuccess ||
-        success == RtpPacketSendResult::kPacketNotFound) {
-      // Packet sent or invalid packet, remove it from queue.
-      // TODO(webrtc:8052): Don't consume media budget on kInvalid.
-      data_sent += packet->size();
-      // Send succeeded, remove it from the queue.
-      OnPacketSent(packet);
-      if (recommended_probe_size && data_sent > *recommended_probe_size)
-        break;
-    } else if (owned_rtp_packet) {
-      // Send failed, but we can't put it back in the queue, remove it without
-      // consuming budget.
-      packets_.FinalizePop();
-      break;
-    } else {
-      // Send failed, put it back into the queue.
-      packets_.CancelPop();
-      break;
-    }
-  }
-
-  if (legacy_packet_referencing_ && packets_.Empty() && !Congested()) {
-    // We can not send padding unless a normal packet has first been sent. If we
-    // do, timestamps get messed up.
-    if (packet_counter_ > 0) {
-      DataSize padding_needed =
-          (recommended_probe_size && *recommended_probe_size > data_sent)
-              ? (*recommended_probe_size - data_sent)
-              : DataSize::bytes(padding_budget_.bytes_remaining());
-      if (padding_needed > DataSize::Zero()) {
-        DataSize padding_sent = DataSize::Zero();
-        critsect_.Leave();
-        padding_sent = DataSize::bytes(packet_router_->TimeToSendPadding(
-            padding_needed.bytes(), pacing_info));
-        critsect_.Enter();
-        data_sent += padding_sent;
-        OnPaddingSent(padding_sent);
-      }
-    }
-  }
-
-  if (is_probing) {
-    probing_send_failure_ = data_sent == DataSize::Zero();
-    if (!probing_send_failure_) {
-      prober_.ProbeSent(CurrentTime().ms(), data_sent.bytes());
-    }
-  }
 }
 
 void PacedSender::ProcessThreadAttached(ProcessThread* process_thread) {
@@ -497,93 +166,49 @@ void PacedSender::ProcessThreadAttached(ProcessThread* process_thread) {
   process_thread_ = process_thread;
 }
 
-DataSize PacedSender::PaddingToAdd(
-    absl::optional<DataSize> recommended_probe_size,
-    DataSize data_sent) {
-  if (!packets_.Empty()) {
-    // Actual payload available, no need to add padding.
-    return DataSize::Zero();
-  }
-
-  if (Congested()) {
-    // Don't add padding if congested, even if requested for probing.
-    return DataSize::Zero();
-  }
-
-  if (packet_counter_ == 0) {
-    // We can not send padding unless a normal packet has first been sent. If we
-    // do, timestamps get messed up.
-    return DataSize::Zero();
-  }
-
-  if (recommended_probe_size) {
-    if (*recommended_probe_size > data_sent) {
-      return *recommended_probe_size - data_sent;
-    }
-    return DataSize::Zero();
-  }
-
-  return DataSize::bytes(padding_budget_.bytes_remaining());
-}
-
-RoundRobinPacketQueue::QueuedPacket* PacedSender::GetPendingPacket(
-    const PacedPacketInfo& pacing_info) {
-  if (packets_.Empty()) {
-    return nullptr;
-  }
-
-  // Since we need to release the lock in order to send, we first pop the
-  // element from the priority queue but keep it in storage, so that we can
-  // reinsert it if send fails.
-  RoundRobinPacketQueue::QueuedPacket* packet = packets_.BeginPop();
-  bool audio_packet = packet->type() == RtpPacketToSend::Type::kAudio;
-  bool apply_pacing = !audio_packet || pace_audio_;
-  if (apply_pacing && (Congested() || (media_budget_.bytes_remaining() == 0 &&
-                                       pacing_info.probe_cluster_id ==
-                                           PacedPacketInfo::kNotAProbe))) {
-    packets_.CancelPop();
-    return nullptr;
-  }
-  return packet;
-}
-
-void PacedSender::OnPacketSent(RoundRobinPacketQueue::QueuedPacket* packet) {
-  Timestamp now = CurrentTime();
-  if (!first_sent_packet_time_) {
-    first_sent_packet_time_ = now;
-  }
-  bool audio_packet = packet->type() == RtpPacketToSend::Type::kAudio;
-  if (!audio_packet || account_for_audio_) {
-    // Update media bytes sent.
-    UpdateBudgetWithSentData(packet->size());
-    last_send_time_ = now;
-  }
-  // Send succeeded, remove it from the queue.
-  packets_.FinalizePop();
-}
-
-void PacedSender::OnPaddingSent(DataSize data_sent) {
-  if (data_sent > DataSize::Zero()) {
-    UpdateBudgetWithSentData(data_sent);
-  }
-  last_send_time_ = CurrentTime();
-}
-
-void PacedSender::UpdateBudgetWithElapsedTime(TimeDelta delta) {
-  delta = std::min(kMaxProcessingInterval, delta);
-  media_budget_.IncreaseBudget(delta.ms());
-  padding_budget_.IncreaseBudget(delta.ms());
-}
-
-void PacedSender::UpdateBudgetWithSentData(DataSize size) {
-  outstanding_data_ += size;
-  media_budget_.UseBudget(size.bytes());
-  padding_budget_.UseBudget(size.bytes());
-}
-
 void PacedSender::SetQueueTimeLimit(TimeDelta limit) {
   rtc::CritScope cs(&critsect_);
-  queue_time_limit = limit;
+  pacing_controller_.SetQueueTimeLimit(limit);
+}
+
+void PacedSender::SendRtpPacket(std::unique_ptr<RtpPacketToSend> packet,
+                                const PacedPacketInfo& cluster_info) {
+  critsect_.Leave();
+  packet_router_->SendPacket(std::move(packet), cluster_info);
+  critsect_.Enter();
+}
+
+std::vector<std::unique_ptr<RtpPacketToSend>> PacedSender::GeneratePadding(
+    DataSize size) {
+  std::vector<std::unique_ptr<RtpPacketToSend>> padding_packets;
+  critsect_.Leave();
+  padding_packets = packet_router_->GeneratePadding(size.bytes());
+  critsect_.Enter();
+  return padding_packets;
+}
+
+RtpPacketSendResult PacedSender::TimeToSendPacket(
+    uint32_t ssrc,
+    uint16_t sequence_number,
+    int64_t capture_timestamp,
+    bool retransmission,
+    const PacedPacketInfo& packet_info) {
+  RtpPacketSendResult result;
+  critsect_.Leave();
+  result = packet_router_->TimeToSendPacket(
+      ssrc, sequence_number, capture_timestamp, retransmission, packet_info);
+  critsect_.Enter();
+  return result;
+}
+
+DataSize PacedSender::TimeToSendPadding(DataSize size,
+                                        const PacedPacketInfo& pacing_info) {
+  size_t padding_bytes_sent;
+  critsect_.Leave();
+  padding_bytes_sent =
+      packet_router_->TimeToSendPadding(size.bytes(), pacing_info);
+  critsect_.Enter();
+  return DataSize::bytes(padding_bytes_sent);
 }
 
 }  // namespace webrtc

modules/pacing/paced_sender.h

@@ -16,6 +16,7 @@
 
 #include <atomic>
 #include <memory>
+#include <vector>
 
 #include "absl/types/optional.h"
 #include "api/function_view.h"
@@ -25,14 +26,13 @@
 #include "modules/include/module.h"
 #include "modules/pacing/bitrate_prober.h"
 #include "modules/pacing/interval_budget.h"
+#include "modules/pacing/pacing_controller.h"
 #include "modules/pacing/packet_router.h"
-#include "modules/pacing/round_robin_packet_queue.h"
 #include "modules/pacing/rtp_packet_pacer.h"
 #include "modules/rtp_rtcp/include/rtp_packet_sender.h"
 #include "modules/rtp_rtcp/source/rtp_packet_to_send.h"
 #include "modules/utility/include/process_thread.h"
 #include "rtc_base/critical_section.h"
-#include "rtc_base/experiments/field_trial_parser.h"
 #include "rtc_base/thread_annotations.h"
 
 namespace webrtc {
@@ -41,7 +41,8 @@ class RtcEventLog;
 
 class PacedSender : public Module,
                     public RtpPacketPacer,
-                    public RtpPacketSender {
+                    public RtpPacketSender,
+                    private PacingController::PacketSender {
  public:
   // Expected max pacer delay in ms. If ExpectedQueueTime() is higher than
   // this value, the packet producers should wait (eg drop frames rather than
@@ -116,6 +117,7 @@ class PacedSender : public Module,
   // Below are methods specific to this implementation, such as things related
   // to module processing thread specifics or methods exposed for test.
 
+  // TODO(bugs.webrtc.org/10809): Remove when cleanup up unit tests.
   // Enable bitrate probing. Enabled by default, mostly here to simplify
   // testing. Must be called before any packets are being sent to have an
   // effect.
@@ -134,69 +136,30 @@ class PacedSender : public Module,
   void ProcessThreadAttached(ProcessThread* process_thread) override;
 
  private:
-  TimeDelta UpdateTimeAndGetElapsed(Timestamp now)
+  // Methods implementing PacedSenderController:PacketSender.
-      RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
+
-  bool ShouldSendKeepalive(Timestamp now) const
+  void SendRtpPacket(std::unique_ptr<RtpPacketToSend> packet,
+                     const PacedPacketInfo& cluster_info) override
       RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
 
-  // Updates the number of bytes that can be sent for the next time interval.
+  std::vector<std::unique_ptr<RtpPacketToSend>> GeneratePadding(
-  void UpdateBudgetWithElapsedTime(TimeDelta delta)
+      DataSize size) override RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-      RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-  void UpdateBudgetWithSentData(DataSize size)
-      RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
 
-  DataSize PaddingToAdd(absl::optional<DataSize> recommended_probe_size,
+  // TODO(bugs.webrtc.org/10633): Remove these when old code path is gone.
-                        DataSize data_sent)
+  RtpPacketSendResult TimeToSendPacket(uint32_t ssrc,
+                                       uint16_t sequence_number,
+                                       int64_t capture_timestamp,
+                                       bool retransmission,
+                                       const PacedPacketInfo& packet_info)
+      override RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
+  DataSize TimeToSendPadding(DataSize size,
+                             const PacedPacketInfo& pacing_info) override
       RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
 
-  RoundRobinPacketQueue::QueuedPacket* GetPendingPacket(
-      const PacedPacketInfo& pacing_info)
-      RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-  void OnPacketSent(RoundRobinPacketQueue::QueuedPacket* packet)
-      RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-  void OnPaddingSent(DataSize padding_sent)
-      RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-
-  bool Congested() const RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-  Timestamp CurrentTime() const RTC_EXCLUSIVE_LOCKS_REQUIRED(critsect_);
-
-  Clock* const clock_;
-  PacketRouter* const packet_router_;
-  const std::unique_ptr<FieldTrialBasedConfig> fallback_field_trials_;
-  const WebRtcKeyValueConfig* field_trials_;
-
-  const bool drain_large_queues_;
-  const bool send_padding_if_silent_;
-  const bool pace_audio_;
-  TimeDelta min_packet_limit_;
-
   rtc::CriticalSection critsect_;
-  // TODO(webrtc:9716): Remove this when we are certain clocks are monotonic.
+  PacingController pacing_controller_ RTC_GUARDED_BY(critsect_);
-  // The last millisecond timestamp returned by |clock_|.
-  mutable Timestamp last_timestamp_ RTC_GUARDED_BY(critsect_);
-  bool paused_ RTC_GUARDED_BY(critsect_);
-  // This is the media budget, keeping track of how many bits of media
-  // we can pace out during the current interval.
-  IntervalBudget media_budget_ RTC_GUARDED_BY(critsect_);
-  // This is the padding budget, keeping track of how many bits of padding we're
-  // allowed to send out during the current interval. This budget will be
-  // utilized when there's no media to send.
-  IntervalBudget padding_budget_ RTC_GUARDED_BY(critsect_);
 
-  BitrateProber prober_ RTC_GUARDED_BY(critsect_);
+  PacketRouter* const packet_router_;
-  bool probing_send_failure_ RTC_GUARDED_BY(critsect_);
-
-  DataRate pacing_bitrate_ RTC_GUARDED_BY(critsect_);
-
-  Timestamp time_last_process_ RTC_GUARDED_BY(critsect_);
-  Timestamp last_send_time_ RTC_GUARDED_BY(critsect_);
-  absl::optional<Timestamp> first_sent_packet_time_ RTC_GUARDED_BY(critsect_);
-
-  RoundRobinPacketQueue packets_ RTC_GUARDED_BY(critsect_);
-  uint64_t packet_counter_ RTC_GUARDED_BY(critsect_);
-
-  DataSize congestion_window_size_ RTC_GUARDED_BY(critsect_);
-  DataSize outstanding_data_ RTC_GUARDED_BY(critsect_);
 
   // Lock to avoid race when attaching process thread. This can happen due to
   // the Call class setting network state on RtpTransportControllerSend, which
@@ -205,14 +168,6 @@ class PacedSender : public Module,
   // queue separate from the thread used by Call, this causes a race.
   rtc::CriticalSection process_thread_lock_;
   ProcessThread* process_thread_ RTC_GUARDED_BY(process_thread_lock_);
-
-  TimeDelta queue_time_limit RTC_GUARDED_BY(critsect_);
-  bool account_for_audio_ RTC_GUARDED_BY(critsect_);
-
-  // If true, PacedSender should only reference packets as in legacy mode.
-  // If false, PacedSender may have direct ownership of RtpPacketToSend objects.
-  // Defaults to true, will be changed to default false soon.
-  const bool legacy_packet_referencing_;
 };
 }  // namespace webrtc
 #endif  // MODULES_PACING_PACED_SENDER_H_

modules/pacing/pacing_controller.cc

@@ -0,0 +1,552 @@
+/*
+ *  Copyright (c) 2019 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 "modules/pacing/pacing_controller.h"
+
+#include <algorithm>
+#include <utility>
+#include <vector>
+
+#include "absl/memory/memory.h"
+#include "modules/pacing/bitrate_prober.h"
+#include "modules/pacing/interval_budget.h"
+#include "modules/utility/include/process_thread.h"
+#include "rtc_base/checks.h"
+#include "rtc_base/logging.h"
+#include "rtc_base/time_utils.h"
+#include "system_wrappers/include/clock.h"
+
+namespace webrtc {
+namespace {
+// Time limit in milliseconds between packet bursts.
+constexpr TimeDelta kDefaultMinPacketLimit = TimeDelta::Millis<5>();
+constexpr TimeDelta kCongestedPacketInterval = TimeDelta::Millis<500>();
+constexpr TimeDelta kMaxElapsedTime = TimeDelta::Seconds<2>();
+
+// Upper cap on process interval, in case process has not been called in a long
+// time.
+constexpr TimeDelta kMaxProcessingInterval = TimeDelta::Millis<30>();
+
+bool IsDisabled(const WebRtcKeyValueConfig& field_trials,
+                absl::string_view key) {
+  return field_trials.Lookup(key).find("Disabled") == 0;
+}
+
+bool IsEnabled(const WebRtcKeyValueConfig& field_trials,
+               absl::string_view key) {
+  return field_trials.Lookup(key).find("Enabled") == 0;
+}
+
+int GetPriorityForType(RtpPacketToSend::Type type) {
+  switch (type) {
+    case RtpPacketToSend::Type::kAudio:
+      // Audio is always prioritized over other packet types.
+      return 0;
+    case RtpPacketToSend::Type::kRetransmission:
+      // Send retransmissions before new media.
+      return 1;
+    case RtpPacketToSend::Type::kVideo:
+      // Video has "normal" priority, in the old speak.
+      return 2;
+    case RtpPacketToSend::Type::kForwardErrorCorrection:
+      // Send redundancy concurrently to video. If it is delayed it might have a
+      // lower chance of being useful.
+      return 2;
+    case RtpPacketToSend::Type::kPadding:
+      // Packets that are in themselves likely useless, only sent to keep the
+      // BWE high.
+      return 3;
+  }
+}
+
+}  // namespace
+
+const TimeDelta PacingController::kMaxExpectedQueueLength =
+    TimeDelta::Millis<2000>();
+const float PacingController::kDefaultPaceMultiplier = 2.5f;
+const TimeDelta PacingController::kPausedProcessInterval =
+    kCongestedPacketInterval;
+
+PacingController::PacingController(Clock* clock,
+                                   PacketSender* packet_sender,
+                                   RtcEventLog* event_log,
+                                   const WebRtcKeyValueConfig* field_trials)
+    : clock_(clock),
+      packet_sender_(packet_sender),
+      fallback_field_trials_(
+          !field_trials ? absl::make_unique<FieldTrialBasedConfig>() : nullptr),
+      field_trials_(field_trials ? field_trials : fallback_field_trials_.get()),
+      drain_large_queues_(
+          !IsDisabled(*field_trials_, "WebRTC-Pacer-DrainQueue")),
+      send_padding_if_silent_(
+          IsEnabled(*field_trials_, "WebRTC-Pacer-PadInSilence")),
+      pace_audio_(!IsDisabled(*field_trials_, "WebRTC-Pacer-BlockAudio")),
+      min_packet_limit_(kDefaultMinPacketLimit),
+      last_timestamp_(clock_->CurrentTime()),
+      paused_(false),
+      media_budget_(0),
+      padding_budget_(0),
+      prober_(*field_trials_),
+      probing_send_failure_(false),
+      padding_failure_state_(false),
+      pacing_bitrate_(DataRate::Zero()),
+      time_last_process_(clock->CurrentTime()),
+      last_send_time_(time_last_process_),
+      packet_queue_(time_last_process_, field_trials),
+      packet_counter_(0),
+      congestion_window_size_(DataSize::PlusInfinity()),
+      outstanding_data_(DataSize::Zero()),
+      queue_time_limit(kMaxExpectedQueueLength),
+      account_for_audio_(false),
+      legacy_packet_referencing_(
+          IsEnabled(*field_trials_, "WebRTC-Pacer-LegacyPacketReferencing")) {
+  if (!drain_large_queues_) {
+    RTC_LOG(LS_WARNING) << "Pacer queues will not be drained,"
+                           "pushback experiment must be enabled.";
+  }
+  FieldTrialParameter<int> min_packet_limit_ms("", min_packet_limit_.ms());
+  ParseFieldTrial({&min_packet_limit_ms},
+                  field_trials_->Lookup("WebRTC-Pacer-MinPacketLimitMs"));
+  min_packet_limit_ = TimeDelta::ms(min_packet_limit_ms.Get());
+  UpdateBudgetWithElapsedTime(min_packet_limit_);
+}
+
+PacingController::~PacingController() = default;
+
+void PacingController::CreateProbeCluster(DataRate bitrate, int cluster_id) {
+  prober_.CreateProbeCluster(bitrate.bps(), CurrentTime().ms(), cluster_id);
+}
+
+void PacingController::Pause() {
+  if (!paused_)
+    RTC_LOG(LS_INFO) << "PacedSender paused.";
+  paused_ = true;
+  packet_queue_.SetPauseState(true, CurrentTime());
+}
+
+void PacingController::Resume() {
+  if (paused_)
+    RTC_LOG(LS_INFO) << "PacedSender resumed.";
+  paused_ = false;
+  packet_queue_.SetPauseState(false, CurrentTime());
+}
+
+bool PacingController::IsPaused() const {
+  return paused_;
+}
+
+void PacingController::SetCongestionWindow(DataSize congestion_window_size) {
+  congestion_window_size_ = congestion_window_size;
+}
+
+void PacingController::UpdateOutstandingData(DataSize outstanding_data) {
+  outstanding_data_ = outstanding_data;
+}
+
+bool PacingController::Congested() const {
+  if (congestion_window_size_.IsFinite()) {
+    return outstanding_data_ >= congestion_window_size_;
+  }
+  return false;
+}
+
+Timestamp PacingController::CurrentTime() const {
+  Timestamp time = clock_->CurrentTime();
+  if (time < last_timestamp_) {
+    RTC_LOG(LS_WARNING)
+        << "Non-monotonic clock behavior observed. Previous timestamp: "
+        << last_timestamp_.ms() << ", new timestamp: " << time.ms();
+    RTC_DCHECK_GE(time, last_timestamp_);
+    time = last_timestamp_;
+  }
+  last_timestamp_ = time;
+  return time;
+}
+
+void PacingController::SetProbingEnabled(bool enabled) {
+  RTC_CHECK_EQ(0, packet_counter_);
+  prober_.SetEnabled(enabled);
+}
+
+void PacingController::SetPacingRates(DataRate pacing_rate,
+                                      DataRate padding_rate) {
+  RTC_DCHECK_GT(pacing_rate, DataRate::Zero());
+  pacing_bitrate_ = pacing_rate;
+  padding_budget_.set_target_rate_kbps(padding_rate.kbps());
+
+  RTC_LOG(LS_VERBOSE) << "bwe:pacer_updated pacing_kbps="
+                      << pacing_bitrate_.kbps()
+                      << " padding_budget_kbps=" << padding_rate.kbps();
+}
+
+void PacingController::InsertPacket(RtpPacketSender::Priority priority,
+                                    uint32_t ssrc,
+                                    uint16_t sequence_number,
+                                    int64_t capture_time_ms,
+                                    size_t bytes,
+                                    bool retransmission) {
+  RTC_DCHECK(pacing_bitrate_ > DataRate::Zero())
+      << "SetPacingRate must be called before InsertPacket.";
+
+  Timestamp now = CurrentTime();
+  prober_.OnIncomingPacket(bytes);
+
+  if (capture_time_ms < 0)
+    capture_time_ms = now.ms();
+
+  RtpPacketToSend::Type type;
+  switch (priority) {
+    case RtpPacketSender::kHighPriority:
+      type = RtpPacketToSend::Type::kAudio;
+      break;
+    case RtpPacketSender::kNormalPriority:
+      type = RtpPacketToSend::Type::kRetransmission;
+      break;
+    default:
+      type = RtpPacketToSend::Type::kVideo;
+  }
+  packet_queue_.Push(GetPriorityForType(type), type, ssrc, sequence_number,
+                     capture_time_ms, now, DataSize::bytes(bytes),
+                     retransmission, packet_counter_++);
+}
+
+void PacingController::EnqueuePacket(std::unique_ptr<RtpPacketToSend> packet) {
+  RTC_DCHECK(pacing_bitrate_ > DataRate::Zero())
+      << "SetPacingRate must be called before InsertPacket.";
+
+  Timestamp now = CurrentTime();
+  prober_.OnIncomingPacket(packet->payload_size());
+
+  if (packet->capture_time_ms() < 0) {
+    packet->set_capture_time_ms(now.ms());
+  }
+
+  RTC_CHECK(packet->packet_type());
+  int priority = GetPriorityForType(*packet->packet_type());
+  packet_queue_.Push(priority, now, packet_counter_++, std::move(packet));
+}
+
+void PacingController::SetAccountForAudioPackets(bool account_for_audio) {
+  account_for_audio_ = account_for_audio;
+}
+
+TimeDelta PacingController::ExpectedQueueTime() const {
+  RTC_DCHECK_GT(pacing_bitrate_, DataRate::Zero());
+  return TimeDelta::ms(
+      (QueueSizeData().bytes() * 8 * rtc::kNumMillisecsPerSec) /
+      pacing_bitrate_.bps());
+}
+
+size_t PacingController::QueueSizePackets() const {
+  return packet_queue_.SizeInPackets();
+}
+
+DataSize PacingController::QueueSizeData() const {
+  return packet_queue_.Size();
+}
+
+absl::optional<Timestamp> PacingController::FirstSentPacketTime() const {
+  return first_sent_packet_time_;
+}
+
+TimeDelta PacingController::OldestPacketWaitTime() const {
+  Timestamp oldest_packet = packet_queue_.OldestEnqueueTime();
+  if (oldest_packet.IsInfinite()) {
+    return TimeDelta::Zero();
+  }
+
+  return CurrentTime() - oldest_packet;
+}
+
+TimeDelta PacingController::UpdateTimeAndGetElapsed(Timestamp now) {
+  TimeDelta elapsed_time = now - time_last_process_;
+  time_last_process_ = now;
+  if (elapsed_time > kMaxElapsedTime) {
+    RTC_LOG(LS_WARNING) << "Elapsed time (" << elapsed_time.ms()
+                        << " ms) longer than expected, limiting to "
+                        << kMaxElapsedTime.ms();
+    elapsed_time = kMaxElapsedTime;
+  }
+  return elapsed_time;
+}
+
+bool PacingController::ShouldSendKeepalive(Timestamp now) const {
+  if (send_padding_if_silent_ || paused_ || Congested()) {
+    // We send a padding packet every 500 ms to ensure we won't get stuck in
+    // congested state due to no feedback being received.
+    TimeDelta elapsed_since_last_send = now - last_send_time_;
+    if (elapsed_since_last_send >= kCongestedPacketInterval) {
+      // We can not send padding unless a normal packet has first been sent. If
+      // we do, timestamps get messed up.
+      if (packet_counter_ > 0) {
+        return true;
+      }
+    }
+  }
+  return false;
+}
+
+absl::optional<TimeDelta> PacingController::TimeUntilNextProbe() {
+  if (!prober_.IsProbing()) {
+    return absl::nullopt;
+  }
+
+  TimeDelta time_delta =
+      TimeDelta::ms(prober_.TimeUntilNextProbe(CurrentTime().ms()));
+  if (time_delta > TimeDelta::Zero() ||
+      (time_delta == TimeDelta::Zero() && !probing_send_failure_)) {
+    return time_delta;
+  }
+
+  return absl::nullopt;
+}
+
+TimeDelta PacingController::TimeElapsedSinceLastProcess() const {
+  return CurrentTime() - time_last_process_;
+}
+
+void PacingController::ProcessPackets() {
+  Timestamp now = CurrentTime();
+  TimeDelta elapsed_time = UpdateTimeAndGetElapsed(now);
+  if (ShouldSendKeepalive(now)) {
+    if (legacy_packet_referencing_) {
+      OnPaddingSent(packet_sender_->TimeToSendPadding(DataSize::bytes(1),
+                                                      PacedPacketInfo()));
+    } else {
+      DataSize keepalive_data_sent = DataSize::Zero();
+      std::vector<std::unique_ptr<RtpPacketToSend>> keepalive_packets =
+          packet_sender_->GeneratePadding(DataSize::bytes(1));
+      for (auto& packet : keepalive_packets) {
+        keepalive_data_sent +=
+            DataSize::bytes(packet->payload_size() + packet->padding_size());
+        packet_sender_->SendRtpPacket(std::move(packet), PacedPacketInfo());
+      }
+      OnPaddingSent(keepalive_data_sent);
+    }
+  }
+
+  if (paused_)
+    return;
+
+  if (elapsed_time > TimeDelta::Zero()) {
+    DataRate target_rate = pacing_bitrate_;
+    DataSize queue_size_data = packet_queue_.Size();
+    if (queue_size_data > DataSize::Zero()) {
+      // Assuming equal size packets and input/output rate, the average packet
+      // has avg_time_left_ms left to get queue_size_bytes out of the queue, if
+      // time constraint shall be met. Determine bitrate needed for that.
+      packet_queue_.UpdateQueueTime(CurrentTime());
+      if (drain_large_queues_) {
+        TimeDelta avg_time_left =
+            std::max(TimeDelta::ms(1),
+                     queue_time_limit - packet_queue_.AverageQueueTime());
+        DataRate min_rate_needed = queue_size_data / avg_time_left;
+        if (min_rate_needed > target_rate) {
+          target_rate = min_rate_needed;
+          RTC_LOG(LS_VERBOSE) << "bwe:large_pacing_queue pacing_rate_kbps="
+                              << target_rate.kbps();
+        }
+      }
+    }
+
+    media_budget_.set_target_rate_kbps(target_rate.kbps());
+    UpdateBudgetWithElapsedTime(elapsed_time);
+  }
+
+  bool is_probing = prober_.IsProbing();
+  PacedPacketInfo pacing_info;
+  absl::optional<DataSize> recommended_probe_size;
+  if (is_probing) {
+    pacing_info = prober_.CurrentCluster();
+    recommended_probe_size = DataSize::bytes(prober_.RecommendedMinProbeSize());
+  }
+
+  DataSize data_sent = DataSize::Zero();
+  // The paused state is checked in the loop since it leaves the critical
+  // section allowing the paused state to be changed from other code.
+  while (!paused_) {
+    auto* packet = GetPendingPacket(pacing_info);
+    if (packet == nullptr) {
+      // No packet available to send, check if we should send padding.
+      if (!legacy_packet_referencing_) {
+        DataSize padding_to_add =
+            PaddingToAdd(recommended_probe_size, data_sent);
+        if (padding_to_add > DataSize::Zero()) {
+          std::vector<std::unique_ptr<RtpPacketToSend>> padding_packets =
+              packet_sender_->GeneratePadding(padding_to_add);
+          if (padding_packets.empty()) {
+            // No padding packets were generated, quite send loop.
+            break;
+          }
+          for (auto& packet : padding_packets) {
+            EnqueuePacket(std::move(packet));
+          }
+          // Continue loop to send the padding that was just added.
+          continue;
+        }
+      }
+
+      // Can't fetch new packet and no padding to send, exit send loop.
+      break;
+    }
+
+    std::unique_ptr<RtpPacketToSend> rtp_packet = packet->ReleasePacket();
+    const bool owned_rtp_packet = rtp_packet != nullptr;
+    RtpPacketSendResult success;
+
+    if (rtp_packet != nullptr) {
+      packet_sender_->SendRtpPacket(std::move(rtp_packet), pacing_info);
+      success = RtpPacketSendResult::kSuccess;
+    } else {
+      success = packet_sender_->TimeToSendPacket(
+          packet->ssrc(), packet->sequence_number(), packet->capture_time_ms(),
+          packet->is_retransmission(), pacing_info);
+    }
+
+    if (success == RtpPacketSendResult::kSuccess ||
+        success == RtpPacketSendResult::kPacketNotFound) {
+      // Packet sent or invalid packet, remove it from queue.
+      // TODO(webrtc:8052): Don't consume media budget on kInvalid.
+      data_sent += packet->size();
+      // Send succeeded, remove it from the queue.
+      OnPacketSent(packet);
+      if (recommended_probe_size && data_sent > *recommended_probe_size)
+        break;
+    } else if (owned_rtp_packet) {
+      // Send failed, but we can't put it back in the queue, remove it without
+      // consuming budget.
+      packet_queue_.FinalizePop();
+      break;
+    } else {
+      // Send failed, put it back into the queue.
+      packet_queue_.CancelPop();
+      break;
+    }
+  }
+
+  if (legacy_packet_referencing_ && packet_queue_.Empty() && !Congested()) {
+    // We can not send padding unless a normal packet has first been sent. If we
+    // do, timestamps get messed up.
+    if (packet_counter_ > 0) {
+      DataSize padding_needed =
+          (recommended_probe_size && *recommended_probe_size > data_sent)
+              ? (*recommended_probe_size - data_sent)
+              : DataSize::bytes(padding_budget_.bytes_remaining());
+      if (padding_needed > DataSize::Zero()) {
+        DataSize padding_sent = DataSize::Zero();
+        padding_sent =
+            packet_sender_->TimeToSendPadding(padding_needed, pacing_info);
+        data_sent += padding_sent;
+        OnPaddingSent(padding_sent);
+      }
+    }
+  }
+
+  if (is_probing) {
+    probing_send_failure_ = data_sent == DataSize::Zero();
+    if (!probing_send_failure_) {
+      prober_.ProbeSent(CurrentTime().ms(), data_sent.bytes());
+    }
+  }
+}
+
+DataSize PacingController::PaddingToAdd(
+    absl::optional<DataSize> recommended_probe_size,
+    DataSize data_sent) {
+  if (!packet_queue_.Empty()) {
+    // Actual payload available, no need to add padding.
+    return DataSize::Zero();
+  }
+
+  if (Congested()) {
+    // Don't add padding if congested, even if requested for probing.
+    return DataSize::Zero();
+  }
+
+  if (packet_counter_ == 0) {
+    // We can not send padding unless a normal packet has first been sent. If we
+    // do, timestamps get messed up.
+    return DataSize::Zero();
+  }
+
+  if (recommended_probe_size) {
+    if (*recommended_probe_size > data_sent) {
+      return *recommended_probe_size - data_sent;
+    }
+    return DataSize::Zero();
+  }
+
+  return DataSize::bytes(padding_budget_.bytes_remaining());
+}
+
+RoundRobinPacketQueue::QueuedPacket* PacingController::GetPendingPacket(
+    const PacedPacketInfo& pacing_info) {
+  if (packet_queue_.Empty()) {
+    return nullptr;
+  }
+
+  // Since we need to release the lock in order to send, we first pop the
+  // element from the priority queue but keep it in storage, so that we can
+  // reinsert it if send fails.
+  RoundRobinPacketQueue::QueuedPacket* packet = packet_queue_.BeginPop();
+  bool audio_packet = packet->type() == RtpPacketToSend::Type::kAudio;
+  bool apply_pacing = !audio_packet || pace_audio_;
+  if (apply_pacing && (Congested() || (media_budget_.bytes_remaining() == 0 &&
+                                       pacing_info.probe_cluster_id ==
+                                           PacedPacketInfo::kNotAProbe))) {
+    packet_queue_.CancelPop();
+    return nullptr;
+  }
+  return packet;
+}
+
+void PacingController::OnPacketSent(
+    RoundRobinPacketQueue::QueuedPacket* packet) {
+  Timestamp now = CurrentTime();
+  if (!first_sent_packet_time_) {
+    first_sent_packet_time_ = now;
+  }
+  bool audio_packet = packet->type() == RtpPacketToSend::Type::kAudio;
+  if (!audio_packet || account_for_audio_) {
+    // Update media bytes sent.
+    UpdateBudgetWithSentData(packet->size());
+    last_send_time_ = now;
+  }
+  // Send succeeded, remove it from the queue.
+  packet_queue_.FinalizePop();
+  padding_failure_state_ = false;
+}
+
+void PacingController::OnPaddingSent(DataSize data_sent) {
+  if (data_sent > DataSize::Zero()) {
+    UpdateBudgetWithSentData(data_sent);
+  } else {
+    padding_failure_state_ = true;
+  }
+  last_send_time_ = CurrentTime();
+}
+
+void PacingController::UpdateBudgetWithElapsedTime(TimeDelta delta) {
+  delta = std::min(kMaxProcessingInterval, delta);
+  media_budget_.IncreaseBudget(delta.ms());
+  padding_budget_.IncreaseBudget(delta.ms());
+}
+
+void PacingController::UpdateBudgetWithSentData(DataSize size) {
+  outstanding_data_ += size;
+  media_budget_.UseBudget(size.bytes());
+  padding_budget_.UseBudget(size.bytes());
+}
+
+void PacingController::SetQueueTimeLimit(TimeDelta limit) {
+  queue_time_limit = limit;
+}
+
+}  // namespace webrtc

modules/pacing/pacing_controller.h

@@ -0,0 +1,221 @@
+/*
+ *  Copyright (c) 2019 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.
+ */
+
+#ifndef MODULES_PACING_PACING_CONTROLLER_H_
+#define MODULES_PACING_PACING_CONTROLLER_H_
+
+#include <stddef.h>
+#include <stdint.h>
+
+#include <atomic>
+#include <memory>
+#include <vector>
+
+#include "absl/types/optional.h"
+#include "api/function_view.h"
+#include "api/rtc_event_log/rtc_event_log.h"
+#include "api/transport/field_trial_based_config.h"
+#include "api/transport/network_types.h"
+#include "api/transport/webrtc_key_value_config.h"
+#include "modules/pacing/bitrate_prober.h"
+#include "modules/pacing/interval_budget.h"
+#include "modules/pacing/round_robin_packet_queue.h"
+#include "modules/pacing/rtp_packet_pacer.h"
+#include "modules/rtp_rtcp/include/rtp_packet_sender.h"
+#include "modules/rtp_rtcp/source/rtp_packet_to_send.h"
+#include "rtc_base/critical_section.h"
+#include "rtc_base/experiments/field_trial_parser.h"
+#include "rtc_base/thread_annotations.h"
+
+namespace webrtc {
+
+// This class implements a leaky-buck packet pacing algorithm. It handles the
+// logic of determining which packets to send when, but the actual timing of
+// the processing is done externally (e.g. PacedSender). Furthermore, the
+// forwarding of packets when they are ready to be sent is also handled
+// externally, via the PacedSendingController::PacketSender interface.
+//
+class PacingController {
+ public:
+  class PacketSender {
+   public:
+    virtual ~PacketSender() = default;
+    virtual void SendRtpPacket(std::unique_ptr<RtpPacketToSend> packet,
+                               const PacedPacketInfo& cluster_info) = 0;
+    virtual std::vector<std::unique_ptr<RtpPacketToSend>> GeneratePadding(
+        DataSize size) = 0;
+
+    // TODO(bugs.webrtc.org/10633): Remove these when old code path is gone.
+    virtual RtpPacketSendResult TimeToSendPacket(
+        uint32_t ssrc,
+        uint16_t sequence_number,
+        int64_t capture_timestamp,
+        bool retransmission,
+        const PacedPacketInfo& packet_info) = 0;
+    virtual DataSize TimeToSendPadding(DataSize size,
+                                       const PacedPacketInfo& pacing_info) = 0;
+  };
+
+  // Expected max pacer delay. If ExpectedQueueTime() is higher than
+  // this value, the packet producers should wait (eg drop frames rather than
+  // encoding them). Bitrate sent may temporarily exceed target set by
+  // UpdateBitrate() so that this limit will be upheld.
+  static const TimeDelta kMaxExpectedQueueLength;
+  // Pacing-rate relative to our target send rate.
+  // Multiplicative factor that is applied to the target bitrate to calculate
+  // the number of bytes that can be transmitted per interval.
+  // Increasing this factor will result in lower delays in cases of bitrate
+  // overshoots from the encoder.
+  static const float kDefaultPaceMultiplier;
+  // If no media or paused, wake up at least every |kPausedProcessIntervalMs| in
+  // order to send a keep-alive packet so we don't get stuck in a bad state due
+  // to lack of feedback.
+  static const TimeDelta kPausedProcessInterval;
+
+  PacingController(Clock* clock,
+                   PacketSender* packet_sender,
+                   RtcEventLog* event_log,
+                   const WebRtcKeyValueConfig* field_trials);
+
+  ~PacingController();
+
+  // Adds the packet information to the queue and calls TimeToSendPacket
+  // when it's time to send.
+  void InsertPacket(RtpPacketSender::Priority priority,
+                    uint32_t ssrc,
+                    uint16_t sequence_number,
+                    int64_t capture_time_ms,
+                    size_t bytes,
+                    bool retransmission);
+  // Adds the packet to the queue and calls PacketRouter::SendPacket() when
+  // it's time to send.
+  void EnqueuePacket(std::unique_ptr<RtpPacketToSend> packet);
+
+  void CreateProbeCluster(DataRate bitrate, int cluster_id);
+
+  void Pause();   // Temporarily pause all sending.
+  void Resume();  // Resume sending packets.
+  bool IsPaused() const;
+
+  void SetCongestionWindow(DataSize congestion_window_size);
+  void UpdateOutstandingData(DataSize outstanding_data);
+
+  // Sets the pacing rates. Must be called once before packets can be sent.
+  void SetPacingRates(DataRate pacing_rate, DataRate padding_rate);
+
+  // Currently audio traffic is not accounted by pacer and passed through.
+  // With the introduction of audio BWE audio traffic will be accounted for
+  // the pacer budget calculation. The audio traffic still will be injected
+  // at high priority.
+  void SetAccountForAudioPackets(bool account_for_audio);
+
+  // Returns the time since the oldest queued packet was enqueued.
+  TimeDelta OldestPacketWaitTime() const;
+
+  size_t QueueSizePackets() const;
+  DataSize QueueSizeData() const;
+
+  // Returns the time when the first packet was sent;
+  absl::optional<Timestamp> FirstSentPacketTime() const;
+
+  // Returns the number of milliseconds it will take to send the current
+  // packets in the queue, given the current size and bitrate, ignoring prio.
+  TimeDelta ExpectedQueueTime() const;
+
+  void SetQueueTimeLimit(TimeDelta limit);
+
+  // Enable bitrate probing. Enabled by default, mostly here to simplify
+  // testing. Must be called before any packets are being sent to have an
+  // effect.
+  void SetProbingEnabled(bool enabled);
+
+  // Time until next probe should be sent. If this value is set, it should be
+  // respected - i.e. don't call ProcessPackets() before this specified time as
+  // that can have unintended side effects.
+  absl::optional<TimeDelta> TimeUntilNextProbe();
+
+  // Time since ProcessPackets() was last executed.
+  TimeDelta TimeElapsedSinceLastProcess() const;
+
+  TimeDelta TimeUntilAvailableBudget() const;
+
+  // Check queue of pending packets and send them or padding packets, if budget
+  // is available.
+  void ProcessPackets();
+
+  bool Congested() const;
+
+ private:
+  TimeDelta UpdateTimeAndGetElapsed(Timestamp now);
+  bool ShouldSendKeepalive(Timestamp now) const;
+
+  // Updates the number of bytes that can be sent for the next time interval.
+  void UpdateBudgetWithElapsedTime(TimeDelta delta);
+  void UpdateBudgetWithSentData(DataSize size);
+
+  DataSize PaddingToAdd(absl::optional<DataSize> recommended_probe_size,
+                        DataSize data_sent);
+
+  RoundRobinPacketQueue::QueuedPacket* GetPendingPacket(
+      const PacedPacketInfo& pacing_info);
+  void OnPacketSent(RoundRobinPacketQueue::QueuedPacket* packet);
+  void OnPaddingSent(DataSize padding_sent);
+
+  Timestamp CurrentTime() const;
+
+  Clock* const clock_;
+  PacketSender* const packet_sender_;
+  const std::unique_ptr<FieldTrialBasedConfig> fallback_field_trials_;
+  const WebRtcKeyValueConfig* field_trials_;
+
+  const bool drain_large_queues_;
+  const bool send_padding_if_silent_;
+  const bool pace_audio_;
+  TimeDelta min_packet_limit_;
+
+  // TODO(webrtc:9716): Remove this when we are certain clocks are monotonic.
+  // The last millisecond timestamp returned by |clock_|.
+  mutable Timestamp last_timestamp_;
+  bool paused_;
+  // This is the media budget, keeping track of how many bits of media
+  // we can pace out during the current interval.
+  IntervalBudget media_budget_;
+  // This is the padding budget, keeping track of how many bits of padding we're
+  // allowed to send out during the current interval. This budget will be
+  // utilized when there's no media to send.
+  IntervalBudget padding_budget_;
+
+  BitrateProber prober_;
+  bool probing_send_failure_;
+  bool padding_failure_state_;
+
+  DataRate pacing_bitrate_;
+
+  Timestamp time_last_process_;
+  Timestamp last_send_time_;
+  absl::optional<Timestamp> first_sent_packet_time_;
+
+  RoundRobinPacketQueue packet_queue_;
+  uint64_t packet_counter_;
+
+  DataSize congestion_window_size_;
+  DataSize outstanding_data_;
+
+  TimeDelta queue_time_limit;
+  bool account_for_audio_;
+
+  // If true, PacedSender should only reference packets as in legacy mode.
+  // If false, PacedSender may have direct ownership of RtpPacketToSend objects.
+  // Defaults to true, will be changed to default false soon.
+  const bool legacy_packet_referencing_;
+};
+}  // namespace webrtc
+
+#endif  // MODULES_PACING_PACING_CONTROLLER_H_