Allows bitrate prober to discard delayed probes, unit type refactorings

This CL adds a parameter to the BirateProber field trial config, which
allows the prober to actually discard probe cluster is pacer scheduling
is too delayed. Today it just keeps going at a too low rate.
Some refactoring was needed anyway, so also switch to using unit types
in more places.

Initially keeps legacy behavior default, to verify no perf regressions.

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

7 files changed, 289 insertions, 136 deletions

diff --git a/modules/pacing/BUILD.gn b/modules/pacing/BUILD.gn
index b19c304e1f..1a4e9a5512 100644
--- a/modules/pacing/BUILD.gn
+++ b/modules/pacing/BUILD.gn
@@ -102,6 +102,7 @@ if (rtc_include_tests) {
       "../../rtc_base/experiments:alr_experiment",
       "../../system_wrappers",
       "../../system_wrappers:field_trial",
+      "../../test:explicit_key_value_config",
       "../../test:field_trial",
       "../../test:test_support",
       "../../test/time_controller:time_controller",
diff --git a/modules/pacing/bitrate_prober.cc b/modules/pacing/bitrate_prober.cc
index e7ce01d95c..1949570d86 100644
--- a/modules/pacing/bitrate_prober.cc
+++ b/modules/pacing/bitrate_prober.cc
@@ -26,7 +26,7 @@ namespace {
 // The min probe packet size is scaled with the bitrate we're probing at.
 // This defines the max min probe packet size, meaning that on high bitrates
 // we have a min probe packet size of 200 bytes.
-constexpr size_t kMinProbePacketSize = 200;
+constexpr DataSize kMinProbePacketSize = DataSize::Bytes(200);
 
 constexpr TimeDelta kProbeClusterTimeout = TimeDelta::Seconds(5);
 
@@ -37,13 +37,17 @@ BitrateProberConfig::BitrateProberConfig(
     : min_probe_packets_sent("min_probe_packets_sent", 5),
       min_probe_delta("min_probe_delta", TimeDelta::Millis(1)),
       min_probe_duration("min_probe_duration", TimeDelta::Millis(15)),
-      max_probe_delay("max_probe_delay", TimeDelta::Millis(3)) {
-  ParseFieldTrial({&min_probe_packets_sent, &min_probe_delta,
-                   &min_probe_duration, &max_probe_delay},
-                  key_value_config->Lookup("WebRTC-Bwe-ProbingConfiguration"));
-  ParseFieldTrial({&min_probe_packets_sent, &min_probe_delta,
-                   &min_probe_duration, &max_probe_delay},
-                  key_value_config->Lookup("WebRTC-Bwe-ProbingBehavior"));
+      max_probe_delay("max_probe_delay", TimeDelta::Millis(3)),
+      // TODO(bugs.webrtc.org/11780): Change to default true.
+      abort_delayed_probes("abort_delayed_probes", false) {
+  ParseFieldTrial(
+      {&min_probe_packets_sent, &min_probe_delta, &min_probe_duration,
+       &max_probe_delay, &abort_delayed_probes},
+      key_value_config->Lookup("WebRTC-Bwe-ProbingConfiguration"));
+  ParseFieldTrial(
+      {&min_probe_packets_sent, &min_probe_delta, &min_probe_duration,
+       &max_probe_delay, &abort_delayed_probes},
+      key_value_config->Lookup("WebRTC-Bwe-ProbingBehavior"));
 }
 
 BitrateProber::~BitrateProber() {
@@ -74,12 +78,11 @@ void BitrateProber::SetEnabled(bool enable) {
   }
 }
 
-void BitrateProber::OnIncomingPacket(size_t packet_size) {
+void BitrateProber::OnIncomingPacket(DataSize packet_size) {
   // Don't initialize probing unless we have something large enough to start
   // probing.
   if (probing_state_ == ProbingState::kInactive && !clusters_.empty() &&
-      packet_size >=
-          std::min<size_t>(RecommendedMinProbeSize(), kMinProbePacketSize)) {
+      packet_size >= std::min(RecommendedMinProbeSize(), kMinProbePacketSize)) {
     // Send next probe right away.
     next_probe_time_ = Timestamp::MinusInfinity();
     probing_state_ = ProbingState::kActive;
@@ -125,7 +128,8 @@ Timestamp BitrateProber::NextProbeTime(Timestamp now) const {
     return Timestamp::PlusInfinity();
   }
 
-  if (next_probe_time_.IsFinite() &&
+  // Legacy behavior, just warn about late probe and return as if not probing.
+  if (!config_.abort_delayed_probes && next_probe_time_.IsFinite() &&
       now - next_probe_time_ > config_.max_probe_delay.Get()) {
     RTC_DLOG(LS_WARNING) << "Probe delay too high"
                             " (next_ms:"
@@ -137,9 +141,24 @@ Timestamp BitrateProber::NextProbeTime(Timestamp now) const {
   return next_probe_time_;
 }
 
-PacedPacketInfo BitrateProber::CurrentCluster() const {
-  RTC_DCHECK(!clusters_.empty());
-  RTC_DCHECK(probing_state_ == ProbingState::kActive);
+absl::optional<PacedPacketInfo> BitrateProber::CurrentCluster(Timestamp now) {
+  if (clusters_.empty() || probing_state_ != ProbingState::kActive) {
+    return absl::nullopt;
+  }
+
+  if (config_.abort_delayed_probes && next_probe_time_.IsFinite() &&
+      now - next_probe_time_ > config_.max_probe_delay.Get()) {
+    RTC_DLOG(LS_WARNING) << "Probe delay too high"
+                            " (next_ms:"
+                         << next_probe_time_.ms() << ", now_ms: " << now.ms()
+                         << "), discarding probe cluster.";
+    clusters_.pop();
+    if (clusters_.empty()) {
+      probing_state_ = ProbingState::kSuspended;
+      return absl::nullopt;
+    }
+  }
+
   PacedPacketInfo info = clusters_.front().pace_info;
   info.probe_cluster_bytes_sent = clusters_.front().sent_bytes;
   return info;
@@ -148,15 +167,18 @@ PacedPacketInfo BitrateProber::CurrentCluster() const {
 // Probe size is recommended based on the probe bitrate required. We choose
 // a minimum of twice |kMinProbeDeltaMs| interval to allow scheduling to be
 // feasible.
-size_t BitrateProber::RecommendedMinProbeSize() const {
-  RTC_DCHECK(!clusters_.empty());
-  return clusters_.front().pace_info.send_bitrate_bps * 2 *
-         config_.min_probe_delta->ms() / (8 * 1000);
+DataSize BitrateProber::RecommendedMinProbeSize() const {
+  if (clusters_.empty()) {
+    return DataSize::Zero();
+  }
+  DataRate send_rate =
+      DataRate::BitsPerSec(clusters_.front().pace_info.send_bitrate_bps);
+  return 2 * send_rate * config_.min_probe_delta;
 }
 
-void BitrateProber::ProbeSent(Timestamp now, size_t bytes) {
+void BitrateProber::ProbeSent(Timestamp now, DataSize size) {
   RTC_DCHECK(probing_state_ == ProbingState::kActive);
-  RTC_DCHECK_GT(bytes, 0);
+  RTC_DCHECK(!size.IsZero());
 
   if (!clusters_.empty()) {
     ProbeCluster* cluster = &clusters_.front();
@@ -164,7 +186,7 @@ void BitrateProber::ProbeSent(Timestamp now, size_t bytes) {
       RTC_DCHECK(cluster->started_at.IsInfinite());
       cluster->started_at = now;
     }
-    cluster->sent_bytes += static_cast<int>(bytes);
+    cluster->sent_bytes += size.bytes<int>();
     cluster->sent_probes += 1;
     next_probe_time_ = CalculateNextProbeTime(*cluster);
     if (cluster->sent_bytes >= cluster->pace_info.probe_cluster_min_bytes &&
@@ -178,8 +200,9 @@ void BitrateProber::ProbeSent(Timestamp now, size_t bytes) {
 
       clusters_.pop();
     }
-    if (clusters_.empty())
+    if (clusters_.empty()) {
       probing_state_ = ProbingState::kSuspended;
+    }
   }
 }
 
diff --git a/modules/pacing/bitrate_prober.h b/modules/pacing/bitrate_prober.h
index 3ebe26ac1f..5a89aac435 100644
--- a/modules/pacing/bitrate_prober.h
+++ b/modules/pacing/bitrate_prober.h
@@ -35,9 +35,11 @@ struct BitrateProberConfig {
   FieldTrialParameter<TimeDelta> min_probe_delta;
   // The minimum probing duration.
   FieldTrialParameter<TimeDelta> min_probe_duration;
-  // Maximum amount of time each probe can be delayed. Probe cluster is reset
-  // and retried from the start when this limit is reached.
+  // Maximum amount of time each probe can be delayed.
   FieldTrialParameter<TimeDelta> max_probe_delay;
+  // If NextProbeTime() is called with a delay higher than specified by
+  // |max_probe_delay|, abort it.
+  FieldTrialParameter<bool> abort_delayed_probes;
 };
 
 // Note that this class isn't thread-safe by itself and therefore relies
@@ -57,29 +59,29 @@ class BitrateProber {
   // Initializes a new probing session if the prober is allowed to probe. Does
   // not initialize the prober unless the packet size is large enough to probe
   // with.
-  void OnIncomingPacket(size_t packet_size);
+  void OnIncomingPacket(DataSize packet_size);
 
   // Create a cluster used to probe for |bitrate_bps| with |num_probes| number
   // of probes.
   void CreateProbeCluster(DataRate bitrate, Timestamp now, int cluster_id);
 
-  // Returns the at which the next probe should be sent to get accurate probing.
-  // If probing is not desired at this time, Timestamp::PlusInfinity() will be
-  // returned.
+  // Returns the time at which the next probe should be sent to get accurate
+  // probing. If probing is not desired at this time, Timestamp::PlusInfinity()
+  // will be returned.
+  // TODO(bugs.webrtc.org/11780): Remove |now| argument when old mode is gone.
   Timestamp NextProbeTime(Timestamp now) const;
 
   // Information about the current probing cluster.
-  PacedPacketInfo CurrentCluster() const;
+  absl::optional<PacedPacketInfo> CurrentCluster(Timestamp now);
 
   // Returns the minimum number of bytes that the prober recommends for
-  // the next probe.
-  size_t RecommendedMinProbeSize() const;
+  // the next probe, or zero if not probing.
+  DataSize RecommendedMinProbeSize() const;
 
   // Called to report to the prober that a probe has been sent. In case of
   // multiple packets per probe, this call would be made at the end of sending
-  // the last packet in probe. |probe_size| is the total size of all packets
-  // in probe.
-  void ProbeSent(Timestamp now, size_t probe_size);
+  // the last packet in probe. |size| is the total size of all packets in probe.
+  void ProbeSent(Timestamp now, DataSize size);
 
  private:
   enum class ProbingState {
diff --git a/modules/pacing/bitrate_prober_unittest.cc b/modules/pacing/bitrate_prober_unittest.cc
index 62277a0d2f..5627db0519 100644
--- a/modules/pacing/bitrate_prober_unittest.cc
+++ b/modules/pacing/bitrate_prober_unittest.cc
@@ -12,6 +12,7 @@
 
 #include <algorithm>
 
+#include "test/explicit_key_value_config.h"
 #include "test/gtest.h"
 
 namespace webrtc {
@@ -28,7 +29,7 @@ TEST(BitrateProberTest, VerifyStatesAndTimeBetweenProbes) {
   const DataRate kTestBitrate1 = DataRate::KilobitsPerSec(900);
   const DataRate kTestBitrate2 = DataRate::KilobitsPerSec(1800);
   const int kClusterSize = 5;
-  const int kProbeSize = 1000;
+  const DataSize kProbeSize = DataSize::Bytes(1000);
   const TimeDelta kMinProbeDuration = TimeDelta::Millis(15);
 
   prober.CreateProbeCluster(kTestBitrate1, now, 0);
@@ -37,7 +38,7 @@ TEST(BitrateProberTest, VerifyStatesAndTimeBetweenProbes) {
 
   prober.OnIncomingPacket(kProbeSize);
   EXPECT_TRUE(prober.is_probing());
-  EXPECT_EQ(0, prober.CurrentCluster().probe_cluster_id);
+  EXPECT_EQ(0, prober.CurrentCluster(now)->probe_cluster_id);
 
   // First packet should probe as soon as possible.
   EXPECT_EQ(Timestamp::MinusInfinity(), prober.NextProbeTime(now));
@@ -45,14 +46,13 @@ TEST(BitrateProberTest, VerifyStatesAndTimeBetweenProbes) {
   for (int i = 0; i < kClusterSize; ++i) {
     now = std::max(now, prober.NextProbeTime(now));
     EXPECT_EQ(now, std::max(now, prober.NextProbeTime(now)));
-    EXPECT_EQ(0, prober.CurrentCluster().probe_cluster_id);
+    EXPECT_EQ(0, prober.CurrentCluster(now)->probe_cluster_id);
     prober.ProbeSent(now, kProbeSize);
   }
 
   EXPECT_GE(now - start_time, kMinProbeDuration);
   // Verify that the actual bitrate is withing 10% of the target.
-  DataRate bitrate =
-      DataSize::Bytes(kProbeSize * (kClusterSize - 1)) / (now - start_time);
+  DataRate bitrate = kProbeSize * (kClusterSize - 1) / (now - start_time);
   EXPECT_GT(bitrate, kTestBitrate1 * 0.9);
   EXPECT_LT(bitrate, kTestBitrate1 * 1.1);
 
@@ -62,14 +62,14 @@ TEST(BitrateProberTest, VerifyStatesAndTimeBetweenProbes) {
   for (int i = 0; i < kClusterSize; ++i) {
     now = std::max(now, prober.NextProbeTime(now));
     EXPECT_EQ(now, std::max(now, prober.NextProbeTime(now)));
-    EXPECT_EQ(1, prober.CurrentCluster().probe_cluster_id);
+    EXPECT_EQ(1, prober.CurrentCluster(now)->probe_cluster_id);
     prober.ProbeSent(now, kProbeSize);
   }
 
   // Verify that the actual bitrate is withing 10% of the target.
   TimeDelta duration = now - probe2_started;
   EXPECT_GE(duration, kMinProbeDuration);
-  bitrate = DataSize::Bytes(kProbeSize * (kClusterSize - 1)) / duration;
+  bitrate = (kProbeSize * (kClusterSize - 1)) / duration;
   EXPECT_GT(bitrate, kTestBitrate2 * 0.9);
   EXPECT_LT(bitrate, kTestBitrate2 * 1.1);
 
@@ -80,6 +80,7 @@ TEST(BitrateProberTest, VerifyStatesAndTimeBetweenProbes) {
 TEST(BitrateProberTest, DoesntProbeWithoutRecentPackets) {
   const FieldTrialBasedConfig config;
   BitrateProber prober(config);
+  const DataSize kProbeSize = DataSize::Bytes(1000);
 
   Timestamp now = Timestamp::Zero();
   EXPECT_EQ(prober.NextProbeTime(now), Timestamp::PlusInfinity());
@@ -87,19 +88,74 @@ TEST(BitrateProberTest, DoesntProbeWithoutRecentPackets) {
   prober.CreateProbeCluster(DataRate::KilobitsPerSec(900), now, 0);
   EXPECT_FALSE(prober.is_probing());
 
-  prober.OnIncomingPacket(1000);
+  prober.OnIncomingPacket(kProbeSize);
   EXPECT_TRUE(prober.is_probing());
   EXPECT_EQ(now, std::max(now, prober.NextProbeTime(now)));
-  prober.ProbeSent(now, 1000);
-  // Let time pass, no large enough packets put into prober.
-  now += TimeDelta::Seconds(6);
+  prober.ProbeSent(now, kProbeSize);
+}
+
+TEST(BitrateProberTest, DoesntDiscardDelayedProbesInLegacyMode) {
+  const TimeDelta kMaxProbeDelay = TimeDelta::Millis(3);
+  const test::ExplicitKeyValueConfig trials(
+      "WebRTC-Bwe-ProbingBehavior/"
+      "abort_delayed_probes:0,"
+      "max_probe_delay:3ms/");
+  BitrateProber prober(trials);
+  const DataSize kProbeSize = DataSize::Bytes(1000);
+
+  Timestamp now = Timestamp::Zero();
+  prober.CreateProbeCluster(DataRate::KilobitsPerSec(900), now, 0);
+  prober.OnIncomingPacket(kProbeSize);
+  EXPECT_TRUE(prober.is_probing());
+  EXPECT_EQ(prober.CurrentCluster(now)->probe_cluster_id, 0);
+  // Advance to first probe time and indicate sent probe.
+  now = std::max(now, prober.NextProbeTime(now));
+  prober.ProbeSent(now, kProbeSize);
+
+  // Advance time 1ms past timeout for the next probe.
+  Timestamp next_probe_time = prober.NextProbeTime(now);
+  EXPECT_GT(next_probe_time, now);
+  now += next_probe_time - now + kMaxProbeDelay + TimeDelta::Millis(1);
+
   EXPECT_EQ(prober.NextProbeTime(now), Timestamp::PlusInfinity());
   // Check that legacy behaviour where prober is reset in TimeUntilNextProbe is
   // no longer there. Probes are no longer retried if they are timed out.
-  prober.OnIncomingPacket(1000);
+  prober.OnIncomingPacket(kProbeSize);
   EXPECT_EQ(prober.NextProbeTime(now), Timestamp::PlusInfinity());
 }
 
+TEST(BitrateProberTest, DiscardsDelayedProbesWhenNotInLegacyMode) {
+  const TimeDelta kMaxProbeDelay = TimeDelta::Millis(3);
+  const test::ExplicitKeyValueConfig trials(
+      "WebRTC-Bwe-ProbingBehavior/"
+      "abort_delayed_probes:1,"
+      "max_probe_delay:3ms/");
+  BitrateProber prober(trials);
+  const DataSize kProbeSize = DataSize::Bytes(1000);
+
+  Timestamp now = Timestamp::Zero();
+
+  // Add two probe clusters.
+  prober.CreateProbeCluster(DataRate::KilobitsPerSec(900), now, /*id=*/0);
+
+  prober.OnIncomingPacket(kProbeSize);
+  EXPECT_TRUE(prober.is_probing());
+  EXPECT_EQ(prober.CurrentCluster(now)->probe_cluster_id, 0);
+  // Advance to first probe time and indicate sent probe.
+  now = std::max(now, prober.NextProbeTime(now));
+  prober.ProbeSent(now, kProbeSize);
+
+  // Advance time 1ms past timeout for the next probe.
+  Timestamp next_probe_time = prober.NextProbeTime(now);
+  EXPECT_GT(next_probe_time, now);
+  now += next_probe_time - now + kMaxProbeDelay + TimeDelta::Millis(1);
+
+  // Still indicates the time we wanted to probe at.
+  EXPECT_EQ(prober.NextProbeTime(now), next_probe_time);
+  // First and only cluster removed due to timeout.
+  EXPECT_FALSE(prober.CurrentCluster(now).has_value());
+}
+
 TEST(BitrateProberTest, DoesntInitializeProbingForSmallPackets) {
   const FieldTrialBasedConfig config;
   BitrateProber prober(config);
@@ -107,7 +163,7 @@ TEST(BitrateProberTest, DoesntInitializeProbingForSmallPackets) {
   prober.SetEnabled(true);
   EXPECT_FALSE(prober.is_probing());
 
-  prober.OnIncomingPacket(100);
+  prober.OnIncomingPacket(DataSize::Bytes(100));
   EXPECT_FALSE(prober.is_probing());
 }
 
@@ -121,7 +177,7 @@ TEST(BitrateProberTest, VerifyProbeSizeOnHighBitrate) {
                             /*cluster_id=*/0);
   // Probe size should ensure a minimum of 1 ms interval.
   EXPECT_GT(prober.RecommendedMinProbeSize(),
-            (kHighBitrate * TimeDelta::Millis(1)).bytes<size_t>());
+            kHighBitrate * TimeDelta::Millis(1));
 }
 
 TEST(BitrateProberTest, MinumumNumberOfProbingPackets) {
@@ -130,14 +186,14 @@ TEST(BitrateProberTest, MinumumNumberOfProbingPackets) {
   // Even when probing at a low bitrate we expect a minimum number
   // of packets to be sent.
   const DataRate kBitrate = DataRate::KilobitsPerSec(100);
-  const int kPacketSizeBytes = 1000;
+  const DataSize kPacketSize = DataSize::Bytes(1000);
 
   Timestamp now = Timestamp::Millis(0);
   prober.CreateProbeCluster(kBitrate, now, 0);
-  prober.OnIncomingPacket(kPacketSizeBytes);
+  prober.OnIncomingPacket(kPacketSize);
   for (int i = 0; i < 5; ++i) {
     EXPECT_TRUE(prober.is_probing());
-    prober.ProbeSent(now, kPacketSizeBytes);
+    prober.ProbeSent(now, kPacketSize);
   }
 
   EXPECT_FALSE(prober.is_probing());
@@ -147,17 +203,17 @@ TEST(BitrateProberTest, ScaleBytesUsedForProbing) {
   const FieldTrialBasedConfig config;
   BitrateProber prober(config);
   const DataRate kBitrate = DataRate::KilobitsPerSec(10000);  // 10 Mbps.
-  const int kPacketSizeBytes = 1000;
-  const int kExpectedBytesSent = (kBitrate * TimeDelta::Millis(15)).bytes();
+  const DataSize kPacketSize = DataSize::Bytes(1000);
+  const DataSize kExpectedDataSent = kBitrate * TimeDelta::Millis(15);
 
   Timestamp now = Timestamp::Millis(0);
   prober.CreateProbeCluster(kBitrate, now, /*cluster_id=*/0);
-  prober.OnIncomingPacket(kPacketSizeBytes);
-  int bytes_sent = 0;
-  while (bytes_sent < kExpectedBytesSent) {
+  prober.OnIncomingPacket(kPacketSize);
+  DataSize data_sent = DataSize::Zero();
+  while (data_sent < kExpectedDataSent) {
     ASSERT_TRUE(prober.is_probing());
-    prober.ProbeSent(now, kPacketSizeBytes);
-    bytes_sent += kPacketSizeBytes;
+    prober.ProbeSent(now, kPacketSize);
+    data_sent += kPacketSize;
   }
 
   EXPECT_FALSE(prober.is_probing());
@@ -167,17 +223,17 @@ TEST(BitrateProberTest, HighBitrateProbing) {
   const FieldTrialBasedConfig config;
   BitrateProber prober(config);
   const DataRate kBitrate = DataRate::KilobitsPerSec(1000000);  // 1 Gbps.
-  const int kPacketSizeBytes = 1000;
-  const int kExpectedBytesSent = (kBitrate * TimeDelta::Millis(15)).bytes();
+  const DataSize kPacketSize = DataSize::Bytes(1000);
+  const DataSize kExpectedDataSent = kBitrate * TimeDelta::Millis(15);
 
   Timestamp now = Timestamp::Millis(0);
   prober.CreateProbeCluster(kBitrate, now, 0);
-  prober.OnIncomingPacket(kPacketSizeBytes);
-  int bytes_sent = 0;
-  while (bytes_sent < kExpectedBytesSent) {
+  prober.OnIncomingPacket(kPacketSize);
+  DataSize data_sent = DataSize::Zero();
+  while (data_sent < kExpectedDataSent) {
     ASSERT_TRUE(prober.is_probing());
-    prober.ProbeSent(now, kPacketSizeBytes);
-    bytes_sent += kPacketSizeBytes;
+    prober.ProbeSent(now, kPacketSize);
+    data_sent += kPacketSize;
   }
 
   EXPECT_FALSE(prober.is_probing());
@@ -187,9 +243,9 @@ TEST(BitrateProberTest, ProbeClusterTimeout) {
   const FieldTrialBasedConfig config;
   BitrateProber prober(config);
   const DataRate kBitrate = DataRate::KilobitsPerSec(300);
-  const int kSmallPacketSize = 20;
+  const DataSize kSmallPacketSize = DataSize::Bytes(20);
   // Expecting two probe clusters of 5 packets each.
-  const int kExpectedBytesSent = 20 * 2 * 5;
+  const DataSize kExpectedDataSent = kSmallPacketSize * 2 * 5;
   const TimeDelta kTimeout = TimeDelta::Millis(5000);
 
   Timestamp now = Timestamp::Millis(0);
@@ -204,11 +260,11 @@ TEST(BitrateProberTest, ProbeClusterTimeout) {
   prober.CreateProbeCluster(kBitrate / 10, now, /*cluster_id=*/2);
   prober.OnIncomingPacket(kSmallPacketSize);
   EXPECT_TRUE(prober.is_probing());
-  int bytes_sent = 0;
-  while (bytes_sent < kExpectedBytesSent) {
+  DataSize data_sent = DataSize::Zero();
+  while (data_sent < kExpectedDataSent) {
     ASSERT_TRUE(prober.is_probing());
     prober.ProbeSent(now, kSmallPacketSize);
-    bytes_sent += kSmallPacketSize;
+    data_sent += kSmallPacketSize;
   }
 
   EXPECT_FALSE(prober.is_probing());
diff --git a/modules/pacing/pacing_controller.cc b/modules/pacing/pacing_controller.cc
index 7e7a01b628..33780e001c 100644
--- a/modules/pacing/pacing_controller.cc
+++ b/modules/pacing/pacing_controller.cc
@@ -289,7 +289,7 @@ TimeDelta PacingController::OldestPacketWaitTime() const {
 void PacingController::EnqueuePacketInternal(
     std::unique_ptr<RtpPacketToSend> packet,
     int priority) {
-  prober_.OnIncomingPacket(packet->payload_size());
+  prober_.OnIncomingPacket(DataSize::Bytes(packet->payload_size()));
 
   // TODO(sprang): Make sure tests respect this, replace with DCHECK.
   Timestamp now = CurrentTime();
@@ -335,7 +335,7 @@ bool PacingController::ShouldSendKeepalive(Timestamp now) const {
 }
 
 Timestamp PacingController::NextSendTime() const {
-  Timestamp now = CurrentTime();
+  const Timestamp now = CurrentTime();
 
   if (paused_) {
     return last_send_time_ + kPausedProcessInterval;
@@ -486,13 +486,21 @@ void PacingController::ProcessPackets() {
   }
 
   bool first_packet_in_probe = false;
-  bool is_probing = prober_.is_probing();
   PacedPacketInfo pacing_info;
-  absl::optional<DataSize> recommended_probe_size;
+  DataSize recommended_probe_size = DataSize::Zero();
+  bool is_probing = prober_.is_probing();
   if (is_probing) {
-    pacing_info = prober_.CurrentCluster();
-    first_packet_in_probe = pacing_info.probe_cluster_bytes_sent == 0;
-    recommended_probe_size = DataSize::Bytes(prober_.RecommendedMinProbeSize());
+    // Probe timing is sensitive, and handled explicitly by BitrateProber, so
+    // use actual send time rather than target.
+    pacing_info = prober_.CurrentCluster(now).value_or(PacedPacketInfo());
+    if (pacing_info.probe_cluster_id != PacedPacketInfo::kNotAProbe) {
+      first_packet_in_probe = pacing_info.probe_cluster_bytes_sent == 0;
+      recommended_probe_size = prober_.RecommendedMinProbeSize();
+      RTC_DCHECK_GT(recommended_probe_size, DataSize::Zero());
+    } else {
+      // No valid probe cluster returned, probe might have timed out.
+      is_probing = false;
+    }
   }
 
   DataSize data_sent = DataSize::Zero();
@@ -571,8 +579,12 @@ void PacingController::ProcessPackets() {
 
     // Send done, update send/process time to the target send time.
     OnPacketSent(packet_type, packet_size, target_send_time);
-    if (recommended_probe_size && data_sent > *recommended_probe_size)
+
+    // If we are currently probing, we need to stop the send loop when we have
+    // reached the send target.
+    if (is_probing && data_sent > recommended_probe_size) {
       break;
+    }
 
     if (mode_ == ProcessMode::kDynamic) {
       // Update target send time in case that are more packets that we are late
@@ -591,14 +603,13 @@ void PacingController::ProcessPackets() {
   if (is_probing) {
     probing_send_failure_ = data_sent == DataSize::Zero();
     if (!probing_send_failure_) {
-      prober_.ProbeSent(CurrentTime(), data_sent.bytes());
+      prober_.ProbeSent(CurrentTime(), data_sent);
     }
   }
 }
 
-DataSize PacingController::PaddingToAdd(
-    absl::optional<DataSize> recommended_probe_size,
-    DataSize data_sent) const {
+DataSize PacingController::PaddingToAdd(DataSize recommended_probe_size,
+                                        DataSize data_sent) const {
   if (!packet_queue_.Empty()) {
     // Actual payload available, no need to add padding.
     return DataSize::Zero();
@@ -615,9 +626,9 @@ DataSize PacingController::PaddingToAdd(
     return DataSize::Zero();
   }
 
-  if (recommended_probe_size) {
-    if (*recommended_probe_size > data_sent) {
-      return *recommended_probe_size - data_sent;
+  if (!recommended_probe_size.IsZero()) {
+    if (recommended_probe_size > data_sent) {
+      return recommended_probe_size - data_sent;
     }
     return DataSize::Zero();
   }
diff --git a/modules/pacing/pacing_controller.h b/modules/pacing/pacing_controller.h
index 971d84c8f0..6e0f9bd5b2 100644
--- a/modules/pacing/pacing_controller.h
+++ b/modules/pacing/pacing_controller.h
@@ -159,7 +159,7 @@ class PacingController {
   void UpdateBudgetWithElapsedTime(TimeDelta delta);
   void UpdateBudgetWithSentData(DataSize size);
 
-  DataSize PaddingToAdd(absl::optional<DataSize> recommended_probe_size,
+  DataSize PaddingToAdd(DataSize recommended_probe_size,
                         DataSize data_sent) const;
 
   std::unique_ptr<RtpPacketToSend> GetPendingPacket(
diff --git a/modules/pacing/pacing_controller_unittest.cc b/modules/pacing/pacing_controller_unittest.cc
index 9194d079a9..8aaa67ce51 100644
--- a/modules/pacing/pacing_controller_unittest.cc
+++ b/modules/pacing/pacing_controller_unittest.cc
@@ -20,6 +20,7 @@
 #include "api/units/data_rate.h"
 #include "modules/pacing/packet_router.h"
 #include "system_wrappers/include/clock.h"
+#include "test/explicit_key_value_config.h"
 #include "test/field_trial.h"
 #include "test/gmock.h"
 #include "test/gtest.h"
@@ -169,6 +170,7 @@ class PacingControllerProbing : public PacingController::PacketSender {
     if (packet->packet_type() != RtpPacketMediaType::kPadding) {
       ++packets_sent_;
     }
+    last_pacing_info_ = pacing_info;
   }
 
   std::vector<std::unique_ptr<RtpPacketToSend>> FetchFec() override {
@@ -194,12 +196,14 @@ class PacingControllerProbing : public PacingController::PacketSender {
   }
 
   int packets_sent() const { return packets_sent_; }
-
   int padding_sent() const { return padding_sent_; }
+  int total_packets_sent() const { return packets_sent_ + padding_sent_; }
+  PacedPacketInfo last_pacing_info() const { return last_pacing_info_; }
 
  private:
   int packets_sent_;
   int padding_sent_;
+  PacedPacketInfo last_pacing_info_;
 };
 
 class PacingControllerTest
@@ -1443,63 +1447,119 @@ TEST_P(PacingControllerTest, ProbingWithInsertedPackets) {
 TEST_P(PacingControllerTest, SkipsProbesWhenProcessIntervalTooLarge) {
   const size_t kPacketSize = 1200;
   const int kInitialBitrateBps = 300000;
-  uint32_t ssrc = 12346;
-  uint16_t sequence_number = 1234;
+  const uint32_t ssrc = 12346;
+  const int kProbeClusterId = 3;
 
-  PacingControllerProbing packet_sender;
-  pacer_ = std::make_unique<PacingController>(&clock_, &packet_sender, nullptr,
-                                              nullptr, GetParam());
-  pacer_->SetPacingRates(
-      DataRate::BitsPerSec(kInitialBitrateBps * kPaceMultiplier),
-      DataRate::Zero());
+  // Test with both legacy and new probe discard modes.
+  // TODO(bugs.webrtc.org/11780): Clean up when legacy is gone.
+  for (bool abort_delayed_probes : {false, true}) {
+    uint16_t sequence_number = 1234;
 
-  for (int i = 0; i < 10; ++i) {
+    PacingControllerProbing packet_sender;
+
+    const test::ExplicitKeyValueConfig trials(
+        abort_delayed_probes ? "WebRTC-Bwe-ProbingBehavior/"
+                               "abort_delayed_probes:1,max_probe_delay:2ms/"
+                             : "WebRTC-Bwe-ProbingBehavior/"
+                               "abort_delayed_probes:0,max_probe_delay:2ms/");
+    pacer_ = std::make_unique<PacingController>(&clock_, &packet_sender,
+                                                nullptr, &trials, GetParam());
+    pacer_->SetPacingRates(
+        DataRate::BitsPerSec(kInitialBitrateBps * kPaceMultiplier),
+        DataRate::BitsPerSec(kInitialBitrateBps));
+
+    for (int i = 0; i < 10; ++i) {
+      Send(RtpPacketMediaType::kVideo, ssrc, sequence_number++,
+           clock_.TimeInMilliseconds(), kPacketSize);
+    }
+    while (pacer_->QueueSizePackets() > 0) {
+      clock_.AdvanceTime(TimeUntilNextProcess());
+      pacer_->ProcessPackets();
+    }
+
+    // Probe at a very high rate.
+    pacer_->CreateProbeCluster(DataRate::KilobitsPerSec(10000),  // 10 Mbps.
+                               /*cluster_id=*/kProbeClusterId);
+    // We need one packet to start the probe.
     Send(RtpPacketMediaType::kVideo, ssrc, sequence_number++,
          clock_.TimeInMilliseconds(), kPacketSize);
-  }
-  while (pacer_->QueueSizePackets() > 0) {
+    const int packets_sent_before_probe = packet_sender.packets_sent();
     clock_.AdvanceTime(TimeUntilNextProcess());
     pacer_->ProcessPackets();
-  }
+    EXPECT_EQ(packet_sender.packets_sent(), packets_sent_before_probe + 1);
 
-  // Probe at a very high rate.
-  pacer_->CreateProbeCluster(DataRate::KilobitsPerSec(10000),  // 10 Mbps.
-                             /*cluster_id=*/3);
-  // We need one packet to start the probe.
-  Send(RtpPacketMediaType::kVideo, ssrc, sequence_number++,
-       clock_.TimeInMilliseconds(), kPacketSize);
-  const int packets_sent_before_probe = packet_sender.packets_sent();
-  clock_.AdvanceTime(TimeUntilNextProcess());
-  pacer_->ProcessPackets();
-  EXPECT_EQ(packet_sender.packets_sent(), packets_sent_before_probe + 1);
+    // Figure out how long between probe packets.
+    Timestamp start_time = clock_.CurrentTime();
+    clock_.AdvanceTime(TimeUntilNextProcess());
+    TimeDelta time_between_probes = clock_.CurrentTime() - start_time;
+    // Advance that distance again + 1ms.
+    clock_.AdvanceTime(time_between_probes);
 
-  // Figure out how long between probe packets.
-  Timestamp start_time = clock_.CurrentTime();
-  clock_.AdvanceTime(TimeUntilNextProcess());
-  TimeDelta time_between_probes = clock_.CurrentTime() - start_time;
-  // Advance that distance again + 1ms.
-  clock_.AdvanceTime(time_between_probes);
+    // Send second probe packet.
+    Send(RtpPacketMediaType::kVideo, ssrc, sequence_number++,
+         clock_.TimeInMilliseconds(), kPacketSize);
+    pacer_->ProcessPackets();
+    EXPECT_EQ(packet_sender.packets_sent(), packets_sent_before_probe + 2);
+    PacedPacketInfo last_pacing_info = packet_sender.last_pacing_info();
+    EXPECT_EQ(last_pacing_info.probe_cluster_id, kProbeClusterId);
+
+    // We're exactly where we should be for the next probe.
+    const Timestamp probe_time = clock_.CurrentTime();
+    EXPECT_EQ(pacer_->NextSendTime(), clock_.CurrentTime());
+
+    BitrateProberConfig probing_config(&trials);
+    EXPECT_GT(probing_config.max_probe_delay.Get(), TimeDelta::Zero());
+    // Advance to within max probe delay, should still return same target.
+    clock_.AdvanceTime(probing_config.max_probe_delay.Get());
+    EXPECT_EQ(pacer_->NextSendTime(), probe_time);
+
+    // Too high probe delay, drop it!
+    clock_.AdvanceTime(TimeDelta::Micros(1));
+
+    int packets_sent_before_timeout = packet_sender.total_packets_sent();
+    if (abort_delayed_probes) {
+      // Expected next process time is unchanged, but calling should not
+      // generate new packets.
+      EXPECT_EQ(pacer_->NextSendTime(), probe_time);
+      pacer_->ProcessPackets();
+      EXPECT_EQ(packet_sender.total_packets_sent(),
+                packets_sent_before_timeout);
 
-  // Send second probe packet.
-  Send(RtpPacketMediaType::kVideo, ssrc, sequence_number++,
-       clock_.TimeInMilliseconds(), kPacketSize);
-  pacer_->ProcessPackets();
-  EXPECT_EQ(packet_sender.packets_sent(), packets_sent_before_probe + 2);
-
-  // We're exactly where we should be for the next probe.
-  const Timestamp probe_time = clock_.CurrentTime();
-  EXPECT_EQ(pacer_->NextSendTime(), clock_.CurrentTime());
-
-  FieldTrialBasedConfig field_trial_config;
-  BitrateProberConfig probing_config(&field_trial_config);
-  EXPECT_GT(probing_config.max_probe_delay.Get(), TimeDelta::Zero());
-  // Advance to within max probe delay, should still return same target.
-  clock_.AdvanceTime(probing_config.max_probe_delay.Get());
-  EXPECT_EQ(pacer_->NextSendTime(), probe_time);
-
-  // Too high probe delay, drop it!
-  clock_.AdvanceTime(TimeDelta::Micros(1));
-  EXPECT_GT(pacer_->NextSendTime(), probe_time);
+      // Next packet sent is not part of probe.
+      if (PeriodicProcess()) {
+        do {
+          AdvanceTimeAndProcess();
+        } while (packet_sender.total_packets_sent() ==
+                 packets_sent_before_timeout);
+      } else {
+        AdvanceTimeAndProcess();
+      }
+      const int expected_probe_id = PacedPacketInfo::kNotAProbe;
+      EXPECT_EQ(packet_sender.last_pacing_info().probe_cluster_id,
+                expected_probe_id);
+    } else {
+      // Legacy behaviour, probe "aborted" so send time moved back. Next call to
+      // ProcessPackets() still results in packets being marked as part of probe
+      // cluster.
+      EXPECT_GT(pacer_->NextSendTime(), probe_time);
+      AdvanceTimeAndProcess();
+      EXPECT_GT(packet_sender.total_packets_sent(),
+                packets_sent_before_timeout);
+      const int expected_probe_id = last_pacing_info.probe_cluster_id;
+      EXPECT_EQ(packet_sender.last_pacing_info().probe_cluster_id,
+                expected_probe_id);
+
+      // Time between sent packets keeps being too large, but we still mark the
+      // packets as being part of the cluster.
+      Timestamp a = clock_.CurrentTime();
+      AdvanceTimeAndProcess();
+      EXPECT_GT(packet_sender.total_packets_sent(),
+                packets_sent_before_timeout);
+      EXPECT_EQ(packet_sender.last_pacing_info().probe_cluster_id,
+                expected_probe_id);
+      EXPECT_GT(clock_.CurrentTime() - a, time_between_probes);
+    }
+  }
 }
 
 TEST_P(PacingControllerTest, ProbingWithPaddingSupport) {