DualPi2: Implement WRR credit scheduler Implements the RFC 9332 / Linux sch_dualpi2 byte-weighted round-robin (WRR) credit scheduler (Config::c_protection, defaulting to 10% weight for the classic queue and 90% weight for the L4S queue): 1. Prevents classic queue starvation: Previously, l4s_queue_ had strict priority over classic_queue_, allowing sustained L4S traffic to starve non-ECT packets (such as ICE STUN pings) in classic_queue_ and inflate RTT measurements. 2. Coupled queue delay sampling: Updates UpdateBaseMarkingProbability() to sample max(l4s_queue_delay, classic_queue_delay) per RFC 9332 so alternating dequeues between classic and L4S queues do not cause negative integral updates while both queues are backlogged. Bug: webrtc:447037083, webrtc:42225697 Change-Id: Ie0cc4fb39ad9e5c9f412a2046f648381f94d5f13 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/508180 Reviewed-by: Rasmus Brandt <brandtr@webrtc.org> Commit-Queue: Per Kjellander <perkj@webrtc.org> Cr-Commit-Position: refs/heads/main@{#48842}
diff --git a/api/test/network_emulation/BUILD.gn b/api/test/network_emulation/BUILD.gn index 3cdee50..7b0b3a7 100644 --- a/api/test/network_emulation/BUILD.gn +++ b/api/test/network_emulation/BUILD.gn
@@ -95,6 +95,7 @@ "../../../rtc_base:logging", "../../../rtc_base:random", "../../transport:ecn_marking", + "../../units:data_size", "../../units:time_delta", "../../units:timestamp", ]
diff --git a/api/test/network_emulation/dual_pi2_network_queue.cc b/api/test/network_emulation/dual_pi2_network_queue.cc index 098fa4d..acf33a3 100644 --- a/api/test/network_emulation/dual_pi2_network_queue.cc +++ b/api/test/network_emulation/dual_pi2_network_queue.cc
@@ -9,13 +9,16 @@ */ #include "api/test/network_emulation/dual_pi2_network_queue.h" +#include <algorithm> #include <cstddef> +#include <cstdint> #include <optional> #include <queue> #include "api/sequence_checker.h" #include "api/test/simulated_network.h" #include "api/transport/ecn_marking.h" +#include "api/units/data_size.h" #include "api/units/time_delta.h" #include "api/units/timestamp.h" #include "rtc_base/checks.h" @@ -25,6 +28,9 @@ DualPi2NetworkQueue::DualPi2NetworkQueue(const Config& config) : config_(config), random_(config.seed) { + RTC_DCHECK_GE(config_.c_protection, 0); + RTC_DCHECK_LE(config_.c_protection, 100); + ResetCProtectionCredit(); sequence_checker_.Detach(); } @@ -69,27 +75,39 @@ std::optional<PacketInFlightInfo> DualPi2NetworkQueue::PeekNextPacket() const { RTC_DCHECK_RUN_ON(&sequence_checker_); - if (!l4s_queue_.empty()) { - return l4s_queue_.front(); + if (empty()) { + return std::nullopt; } - if (!classic_queue_.empty()) { - return classic_queue_.front(); - } - return std::nullopt; + return ShouldDequeueFromClassicQueue() ? classic_queue_.front() + : l4s_queue_.front(); } std::optional<PacketInFlightInfo> DualPi2NetworkQueue::DequeuePacket( Timestamp time_now) { RTC_DCHECK_RUN_ON(&sequence_checker_); - std::queue<PacketInFlightInfo>& queue = - l4s_queue_.empty() ? classic_queue_ : l4s_queue_; - if (queue.empty()) { + if (empty()) { return std::nullopt; } + const bool dequeue_classic = ShouldDequeueFromClassicQueue(); + std::queue<PacketInFlightInfo>& queue = + dequeue_classic ? classic_queue_ : l4s_queue_; + PacketInFlightInfo packet_info = queue.front(); + if (!l4s_queue_.empty() && !classic_queue_.empty()) { + if (dequeue_classic) { + c_protection_credit_ -= + packet_info.packet_size().bytes() * (100 - config_.c_protection); + } else { + c_protection_credit_ += + packet_info.packet_size().bytes() * config_.c_protection; + } + } else { + ResetCProtectionCredit(); + } + + UpdateBaseMarkingProbability(time_now); TimeDelta sojourn_time = time_now - packet_info.send_time(); - UpdateBaseMarkingProbability(time_now, sojourn_time); if (packet_info.ecn == EcnMarking::kEct1 && HasStandingQueueBacklog()) { if (sojourn_time > config_.step_threshold || random_.Rand<double>() < l4s_marking_probability()) { @@ -105,13 +123,14 @@ return classic_queue_.empty() && l4s_queue_.empty(); } -void DualPi2NetworkQueue::UpdateBaseMarkingProbability(Timestamp time_now, - TimeDelta sojourn_time) { +void DualPi2NetworkQueue::UpdateBaseMarkingProbability(Timestamp time_now) { if (time_now - config_.probability_update_interval < last_probability_update_time_) { return; } last_probability_update_time_ = time_now; + TimeDelta sojourn_time = + std::max(l4s_queue_delay(time_now), classic_queue_delay(time_now)); TimeDelta proportional_update = config_.alpha * (sojourn_time - config_.target_delay); TimeDelta integral_update = @@ -130,9 +149,34 @@ << base_marking_probability_; } +TimeDelta DualPi2NetworkQueue::l4s_queue_delay(Timestamp time_now) const { + return l4s_queue_.empty() ? TimeDelta::Zero() + : time_now - l4s_queue_.front().send_time(); +} + +TimeDelta DualPi2NetworkQueue::classic_queue_delay(Timestamp time_now) const { + return classic_queue_.empty() ? TimeDelta::Zero() + : time_now - classic_queue_.front().send_time(); +} + bool DualPi2NetworkQueue::HasStandingQueueBacklog() const { return l4s_queue_.size() + classic_queue_.size() >= config_.min_backlog_packets; } +bool DualPi2NetworkQueue::ShouldDequeueFromClassicQueue() const { + if (l4s_queue_.empty()) { + return !classic_queue_.empty(); + } + if (classic_queue_.empty()) { + return false; + } + return c_protection_credit_ >= 0; +} + +void DualPi2NetworkQueue::ResetCProtectionCredit() { + constexpr int64_t kAssumedMtuBytes = 1500; + c_protection_credit_ = -kAssumedMtuBytes * (100 - config_.c_protection); +} + } // namespace webrtc
diff --git a/api/test/network_emulation/dual_pi2_network_queue.h b/api/test/network_emulation/dual_pi2_network_queue.h index ce64ac5..851fb47 100644 --- a/api/test/network_emulation/dual_pi2_network_queue.h +++ b/api/test/network_emulation/dual_pi2_network_queue.h
@@ -11,6 +11,7 @@ #define API_TEST_NETWORK_EMULATION_DUAL_PI2_NETWORK_QUEUE_H_ #include <cstddef> +#include <cstdint> #include <memory> #include <optional> #include <queue> @@ -29,12 +30,6 @@ // https://github.com/L4STeam/linux/. Concepts are described in // https://datatracker.ietf.org/doc/html/rfc9332. // Developed for testing purposes. -// Note that this implementation does not support the credit-based system -// (c_protection) from the real implementation and thus a L4S stream can -// completely starve a classic stream. -// -// TODO: bugs.webrtc.org/42225697 - Implement c_protection to better -// support testing of cross traffic with classic TCP. class DualPi2NetworkQueue : public NetworkQueue { public: struct Config { @@ -55,6 +50,11 @@ // Coupling factor. int k = 2; + // Classic queue protection weight percentage (0-100) for the WRR credit + // scheduler (c_protection in Linux sch_dualpi2). Default is 10%, meaning + // 10% weight for the classic queue and 90% weight for the L4S queue. + int c_protection = 10; + // How often the base marking probability is updated. TimeDelta probability_update_interval = TimeDelta::Millis(16); int seed = 1; @@ -86,8 +86,12 @@ } private: - void UpdateBaseMarkingProbability(Timestamp time_now, TimeDelta sojourn_time); + void UpdateBaseMarkingProbability(Timestamp time_now); + TimeDelta l4s_queue_delay(Timestamp time_now) const; + TimeDelta classic_queue_delay(Timestamp time_now) const; bool HasStandingQueueBacklog() const; + bool ShouldDequeueFromClassicQueue() const; + void ResetCProtectionCredit(); SequenceChecker sequence_checker_; @@ -99,6 +103,7 @@ Random random_; std::optional<size_t> max_packet_capacity_; + int64_t c_protection_credit_ = 0; double base_marking_probability_ = 0; Timestamp last_probability_update_time_ = Timestamp::MinusInfinity(); // The delay of the queue after the last probability update.
diff --git a/api/test/network_emulation/dual_pi2_network_queue_unittest.cc b/api/test/network_emulation/dual_pi2_network_queue_unittest.cc index c169ff5..a6cb7b6 100644 --- a/api/test/network_emulation/dual_pi2_network_queue_unittest.cc +++ b/api/test/network_emulation/dual_pi2_network_queue_unittest.cc
@@ -221,9 +221,8 @@ std::optional<PacketInFlightInfo> dequeued_packet = queue.DequeuePacket(now); ASSERT_TRUE(dequeued_packet.has_value()); - // Dequeued packets are always L4S. EXPECT_THAT(dequeued_packet->ecn, - AnyOf(EcnMarking::kEct1, EcnMarking::kCe)); + AnyOf(EcnMarking::kEct0, EcnMarking::kEct1, EcnMarking::kCe)); } EXPECT_TRUE(has_dropped_classic_packet); } @@ -352,5 +351,42 @@ Optional(Field(&PacketInFlightInfo::ecn, EcnMarking::kEct1))); } +TEST(DualPi2NetworkQueueTest, WrrPreventsClassicQueueStarvation) { + // Default c_protection is 10% (90% L4S, 10% Classic). + DualPi2NetworkQueue queue; + Timestamp now = Timestamp::Seconds(123); + + // Enqueue 100 L4S packets and 20 Classic packets of equal size (1500B). + constexpr DataSize kMtuSize = DataSize::Bytes(1500); + int packet_id = 0; + for (int i = 0; i < 100; ++i) { + ASSERT_TRUE(queue.EnqueuePacket(PacketInFlightInfo( + kMtuSize, now, /*packet_id=*/packet_id++, EcnMarking::kEct1))); + } + for (int i = 0; i < 20; ++i) { + ASSERT_TRUE(queue.EnqueuePacket(PacketInFlightInfo( + kMtuSize, now, /*packet_id=*/packet_id++, EcnMarking::kNotEct))); + } + + int l4s_dequeued = 0; + int classic_dequeued = 0; + for (int i = 0; i < 100; ++i) { + std::optional<PacketInFlightInfo> peeked = queue.PeekNextPacket(); + std::optional<PacketInFlightInfo> dequeued = queue.DequeuePacket(now); + ASSERT_TRUE(peeked.has_value()); + ASSERT_TRUE(dequeued.has_value()); + EXPECT_EQ(peeked->packet_id, dequeued->packet_id); + if (dequeued->ecn == EcnMarking::kNotEct) { + ++classic_dequeued; + } else { + ++l4s_dequeued; + } + } + // Out of 100 dequeued packets while both queues are backlogged, 10% (10) + // should be from the classic queue and 90% (90) from the L4S queue. + EXPECT_EQ(classic_dequeued, 10); + EXPECT_EQ(l4s_dequeued, 90); +} + } // namespace } // namespace webrtc