[SlackedPacer] Don't slack while retransmissions or audio is in queue.
This CL introduces PacketQueue::SizeInPacketsPerRtpPacketMediaType
keeping track of the number of packets in the queue per
RtpPacketMediaType.
The TaskQueuePacedSender is updated not to apply slack if the queue
contains any kRetransmission or kAudio packets. The hope is that not
slacking retransmissions will make the NACK/retransmission regression
of the SlackedPacer experiment go away. Wanting to not slack audio
packets is unrelated to the regression but a sensible thing to due
since audio is highest priority.
This CL does not change anything when the SlackedPacer experiment is
not running, since if its not running then none of the packets are
slacked.
Bug: webrtc:14161
Change-Id: I1e588599b6b64ebfd7d890706b6afd0b84fd746d
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/265160
Reviewed-by: Erik Språng <sprang@webrtc.org>
Commit-Queue: Henrik Boström <hbos@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#37139}
diff --git a/modules/pacing/BUILD.gn b/modules/pacing/BUILD.gn
index e7c6a43..ed6d2bd 100644
--- a/modules/pacing/BUILD.gn
+++ b/modules/pacing/BUILD.gn
@@ -94,6 +94,7 @@
"pacing_controller_unittest.cc",
"packet_router_unittest.cc",
"prioritized_packet_queue_unittest.cc",
+ "round_robin_packet_queue_unittest.cc",
"task_queue_paced_sender_unittest.cc",
]
deps = [
@@ -103,6 +104,7 @@
"../../api/transport:network_control",
"../../api/units:data_rate",
"../../api/units:time_delta",
+ "../../api/units:timestamp",
"../../modules/utility:mock_process_thread",
"../../rtc_base:checks",
"../../rtc_base:rtc_base_tests_utils",
diff --git a/modules/pacing/pacing_controller.cc b/modules/pacing/pacing_controller.cc
index a0f4ecb..98123c8 100644
--- a/modules/pacing/pacing_controller.cc
+++ b/modules/pacing/pacing_controller.cc
@@ -259,6 +259,11 @@
return rtc::checked_cast<size_t>(packet_queue_->SizeInPackets());
}
+const std::array<int, kNumMediaTypes>&
+PacingController::SizeInPacketsPerRtpPacketMediaType() const {
+ return packet_queue_->SizeInPacketsPerRtpPacketMediaType();
+}
+
DataSize PacingController::QueueSizeData() const {
DataSize size = packet_queue_->SizeInPayloadBytes();
if (include_overhead_) {
diff --git a/modules/pacing/pacing_controller.h b/modules/pacing/pacing_controller.h
index f7c0939..b3949b6 100644
--- a/modules/pacing/pacing_controller.h
+++ b/modules/pacing/pacing_controller.h
@@ -14,6 +14,7 @@
#include <stddef.h>
#include <stdint.h>
+#include <array>
#include <atomic>
#include <memory>
#include <vector>
@@ -68,6 +69,11 @@
bool Empty() const { return SizeInPackets() == 0; }
virtual DataSize SizeInPayloadBytes() const = 0;
+ // Total packets in the queue per media type (RtpPacketMediaType values are
+ // used as lookup index).
+ virtual const std::array<int, kNumMediaTypes>&
+ SizeInPacketsPerRtpPacketMediaType() const = 0;
+
// If the next packet, that would be returned by Pop() if called
// now, is an audio packet this method returns the enqueue time
// of that packet. If queue is empty or top packet is not audio,
@@ -159,6 +165,10 @@
// Number of packets in the pacer queue.
size_t QueueSizePackets() const;
+ // Number of packets in the pacer queue per media type (RtpPacketMediaType
+ // values are used as lookup index).
+ const std::array<int, kNumMediaTypes>& SizeInPacketsPerRtpPacketMediaType()
+ const;
// Totals size of packets in the pacer queue.
DataSize QueueSizeData() const;
diff --git a/modules/pacing/prioritized_packet_queue.cc b/modules/pacing/prioritized_packet_queue.cc
index b5c0582..83ec77d 100644
--- a/modules/pacing/prioritized_packet_queue.cc
+++ b/modules/pacing/prioritized_packet_queue.cc
@@ -95,6 +95,7 @@
: queue_time_sum_(TimeDelta::Zero()),
pause_time_sum_(TimeDelta::Zero()),
size_packets_(0),
+ size_packets_per_media_type_({}),
size_payload_(DataSize::Zero()),
last_update_time_(creation_time),
paused_(false),
@@ -112,7 +113,9 @@
auto enqueue_time_iterator =
enqueue_times_.insert(enqueue_times_.end(), enqueue_time);
- int prio_level = GetPriorityForType(*packet->packet_type());
+ RTC_DCHECK(packet->packet_type().has_value());
+ RtpPacketMediaType packet_type = packet->packet_type().value();
+ int prio_level = GetPriorityForType(packet_type);
RTC_DCHECK_GE(prio_level, 0);
RTC_DCHECK_LT(prio_level, kNumPriorityLevels);
QueuedPacket queued_packed = {.packet = std::move(packet),
@@ -127,6 +130,7 @@
UpdateAverageQueueTime(enqueue_time);
queued_packed.enqueue_time -= pause_time_sum_;
++size_packets_;
+ ++size_packets_per_media_type_[static_cast<size_t>(packet_type)];
size_payload_ += queued_packed.PacketSize();
if (stream_queue->EnqueuePacket(std::move(queued_packed), prio_level)) {
@@ -160,6 +164,11 @@
StreamQueue& stream_queue = *streams_by_prio_[top_active_prio_level_].front();
QueuedPacket packet = stream_queue.DequePacket(top_active_prio_level_);
--size_packets_;
+ RTC_DCHECK(packet.packet->packet_type().has_value());
+ RtpPacketMediaType packet_type = packet.packet->packet_type().value();
+ --size_packets_per_media_type_[static_cast<size_t>(packet_type)];
+ RTC_DCHECK_GE(size_packets_per_media_type_[static_cast<size_t>(packet_type)],
+ 0);
size_payload_ -= packet.PacketSize();
// Calculate the total amount of time spent by this packet in the queue
@@ -207,6 +216,11 @@
return size_payload_;
}
+const std::array<int, kNumMediaTypes>&
+PrioritizedPacketQueue::SizeInPacketsPerRtpPacketMediaType() const {
+ return size_packets_per_media_type_;
+}
+
Timestamp PrioritizedPacketQueue::LeadingAudioPacketEnqueueTime() const {
if (streams_by_prio_[kAudioPrioLevel].empty()) {
return Timestamp::MinusInfinity();
diff --git a/modules/pacing/prioritized_packet_queue.h b/modules/pacing/prioritized_packet_queue.h
index 2b65365..c770435 100644
--- a/modules/pacing/prioritized_packet_queue.h
+++ b/modules/pacing/prioritized_packet_queue.h
@@ -37,6 +37,8 @@
std::unique_ptr<RtpPacketToSend> Pop() override;
int SizeInPackets() const override;
DataSize SizeInPayloadBytes() const override;
+ const std::array<int, kNumMediaTypes>& SizeInPacketsPerRtpPacketMediaType()
+ const override;
Timestamp LeadingAudioPacketEnqueueTime() const override;
Timestamp OldestEnqueueTime() const override;
TimeDelta AverageQueueTime() const override;
@@ -88,6 +90,8 @@
TimeDelta pause_time_sum_;
// Total number of packets stored in this queue.
int size_packets_;
+ // Total number of packets stored in this queue per RtpPacketMediaType.
+ std::array<int, kNumMediaTypes> size_packets_per_media_type_;
// Sum of payload sizes for all packts stored in this queue.
DataSize size_payload_;
// The last time queue/pause time sums were updated.
diff --git a/modules/pacing/prioritized_packet_queue_unittest.cc b/modules/pacing/prioritized_packet_queue_unittest.cc
index d8732e2..6e27ff0 100644
--- a/modules/pacing/prioritized_packet_queue_unittest.cc
+++ b/modules/pacing/prioritized_packet_queue_unittest.cc
@@ -13,6 +13,7 @@
#include <utility>
#include "api/units/time_delta.h"
+#include "modules/rtp_rtcp/include/rtp_rtcp_defines.h"
#include "modules/rtp_rtcp/source/rtp_packet_to_send.h"
#include "rtc_base/checks.h"
#include "test/gmock.h"
@@ -230,4 +231,61 @@
EXPECT_EQ(queue.LeadingAudioPacketEnqueueTime(), Timestamp::MinusInfinity());
}
+TEST(PrioritizedPacketQueue,
+ PushAndPopUpdatesSizeInPacketsPerRtpPacketMediaType) {
+ Timestamp now = Timestamp::Zero();
+ PrioritizedPacketQueue queue(now);
+
+ // Initially all sizes are zero.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[i], 0);
+ }
+
+ // Push packets.
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kAudio, 1));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kAudio)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kVideo, 2));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kVideo)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kRetransmission, 3));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kRetransmission)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kForwardErrorCorrection, 4));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kForwardErrorCorrection)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kPadding, 5));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kPadding)],
+ 1);
+
+ // Now all sizes are 1.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[i], 1);
+ }
+
+ // Popping happens in a priority order based on media type. This test does not
+ // assert what this order is, only that the counter for the popped packet's
+ // media type is decremented.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ auto popped_packet = queue.Pop();
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ popped_packet->packet_type().value())],
+ 0);
+ }
+
+ // We've popped all packets, so all sizes are zero.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[i], 0);
+ }
+}
+
} // namespace webrtc
diff --git a/modules/pacing/round_robin_packet_queue.cc b/modules/pacing/round_robin_packet_queue.cc
index 638ab07..d7525e9 100644
--- a/modules/pacing/round_robin_packet_queue.cc
+++ b/modules/pacing/round_robin_packet_queue.cc
@@ -137,6 +137,7 @@
enqueue_count_(0),
paused_(false),
size_packets_(0),
+ size_packets_per_media_type_({}),
size_(DataSize::Zero()),
max_size_(kMaxLeadingSize),
queue_time_sum_(TimeDelta::Zero()),
@@ -152,8 +153,9 @@
void RoundRobinPacketQueue::Push(Timestamp enqueue_time,
std::unique_ptr<RtpPacketToSend> packet) {
- RTC_DCHECK(packet->packet_type().has_value());
- int priority = GetPriorityForType(*packet->packet_type());
+ RTC_CHECK(packet->packet_type().has_value());
+ RtpPacketMediaType packet_type = packet->packet_type().value();
+ int priority = GetPriorityForType(packet_type);
if (size_packets_ == 0) {
// Single packet fast-path.
single_packet_queue_.emplace(
@@ -162,6 +164,7 @@
UpdateAverageQueueTime(enqueue_time);
single_packet_queue_->SubtractPauseTime(pause_time_sum_);
size_packets_ = 1;
+ ++size_packets_per_media_type_[static_cast<size_t>(packet_type)];
size_ += PacketSize(*single_packet_queue_);
} else {
MaybePromoteSinglePacketToNormalQueue();
@@ -178,6 +181,11 @@
single_packet_queue_.reset();
queue_time_sum_ = TimeDelta::Zero();
size_packets_ = 0;
+ RTC_CHECK(rtp_packet->packet_type().has_value());
+ RtpPacketMediaType packet_type = rtp_packet->packet_type().value();
+ size_packets_per_media_type_[static_cast<size_t>(packet_type)] -= 1;
+ RTC_CHECK_GE(size_packets_per_media_type_[static_cast<size_t>(packet_type)],
+ 0);
size_ = DataSize::Zero();
return rtp_packet;
}
@@ -213,7 +221,11 @@
size_ -= packet_size;
size_packets_ -= 1;
+ size_packets_per_media_type_[static_cast<size_t>(queued_packet.Type())] -= 1;
RTC_CHECK(size_packets_ > 0 || queue_time_sum_ == TimeDelta::Zero());
+ RTC_CHECK_GE(
+ size_packets_per_media_type_[static_cast<size_t>(queued_packet.Type())],
+ 0);
std::unique_ptr<RtpPacketToSend> rtp_packet(queued_packet.RtpPacket());
stream->packet_queue.pop();
@@ -239,6 +251,11 @@
return size_;
}
+const std::array<int, kNumMediaTypes>&
+RoundRobinPacketQueue::SizeInPacketsPerRtpPacketMediaType() const {
+ return size_packets_per_media_type_;
+}
+
Timestamp RoundRobinPacketQueue::LeadingAudioPacketEnqueueTime() const {
if (single_packet_queue_.has_value()) {
if (single_packet_queue_->Type() == RtpPacketMediaType::kAudio) {
@@ -339,6 +356,7 @@
packet.SubtractPauseTime(pause_time_sum_);
size_packets_ += 1;
+ size_packets_per_media_type_[static_cast<size_t>(packet.Type())] += 1;
size_ += PacketSize(packet);
}
diff --git a/modules/pacing/round_robin_packet_queue.h b/modules/pacing/round_robin_packet_queue.h
index 4d6f93b..052b98b 100644
--- a/modules/pacing/round_robin_packet_queue.h
+++ b/modules/pacing/round_robin_packet_queue.h
@@ -42,6 +42,8 @@
int SizeInPackets() const override;
DataSize SizeInPayloadBytes() const override;
+ const std::array<int, kNumMediaTypes>& SizeInPacketsPerRtpPacketMediaType()
+ const override;
Timestamp LeadingAudioPacketEnqueueTime() const override;
Timestamp OldestEnqueueTime() const override;
TimeDelta AverageQueueTime() const override;
@@ -142,6 +144,7 @@
bool paused_;
int size_packets_;
+ std::array<int, kNumMediaTypes> size_packets_per_media_type_;
DataSize size_;
DataSize max_size_;
TimeDelta queue_time_sum_;
diff --git a/modules/pacing/round_robin_packet_queue_unittest.cc b/modules/pacing/round_robin_packet_queue_unittest.cc
new file mode 100644
index 0000000..86f07be
--- /dev/null
+++ b/modules/pacing/round_robin_packet_queue_unittest.cc
@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2022 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/round_robin_packet_queue.h"
+
+#include <utility>
+
+#include "api/units/timestamp.h"
+#include "modules/rtp_rtcp/include/rtp_rtcp_defines.h"
+#include "modules/rtp_rtcp/source/rtp_packet_to_send.h"
+#include "rtc_base/checks.h"
+#include "test/gmock.h"
+#include "test/gtest.h"
+
+namespace webrtc {
+
+namespace {
+
+constexpr uint32_t kDefaultSsrc = 123;
+constexpr int kDefaultPayloadSize = 321;
+
+std::unique_ptr<RtpPacketToSend> CreatePacket(RtpPacketMediaType type,
+ uint16_t sequence_number) {
+ auto packet = std::make_unique<RtpPacketToSend>(/*extensions=*/nullptr);
+ packet->set_packet_type(type);
+ packet->SetSsrc(kDefaultSsrc);
+ packet->SetSequenceNumber(sequence_number);
+ packet->SetPayloadSize(kDefaultPayloadSize);
+ return packet;
+}
+
+} // namespace
+
+TEST(RoundRobinPacketQueueTest,
+ PushAndPopUpdatesSizeInPacketsPerRtpPacketMediaType) {
+ Timestamp now = Timestamp::Zero();
+ RoundRobinPacketQueue queue(now);
+
+ // Initially all sizes are zero.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[i], 0);
+ }
+
+ // Push packets.
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kAudio, 1));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kAudio)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kVideo, 2));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kVideo)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kRetransmission, 3));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kRetransmission)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kForwardErrorCorrection, 4));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kForwardErrorCorrection)],
+ 1);
+
+ queue.Push(now, CreatePacket(RtpPacketMediaType::kPadding, 5));
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ RtpPacketMediaType::kPadding)],
+ 1);
+
+ // Now all sizes are 1.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[i], 1);
+ }
+
+ // Popping happens in a priority order based on media type. This test does not
+ // assert what this order is, only that the counter for the popped packet's
+ // media type is decremented.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ auto popped_packet = queue.Pop();
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[static_cast<size_t>(
+ popped_packet->packet_type().value())],
+ 0);
+ }
+
+ // We've popped all packets, so all sizes are zero.
+ for (size_t i = 0; i < kNumMediaTypes; ++i) {
+ EXPECT_EQ(queue.SizeInPacketsPerRtpPacketMediaType()[i], 0);
+ }
+}
+
+} // namespace webrtc
diff --git a/modules/pacing/task_queue_paced_sender.cc b/modules/pacing/task_queue_paced_sender.cc
index 5f268a7..43c4c96 100644
--- a/modules/pacing/task_queue_paced_sender.cc
+++ b/modules/pacing/task_queue_paced_sender.cc
@@ -297,14 +297,23 @@
!pacing_controller_.IsProbing()
? TaskQueueBase::DelayPrecision::kLow
: TaskQueueBase::DelayPrecision::kHigh;
- // Optionally disable low precision if the expected queue time is greater
- // than `max_low_precision_expected_queue_time`.
- if (precision == TaskQueueBase::DelayPrecision::kLow &&
- slacked_pacer_flags_.max_low_precision_expected_queue_time &&
- pacing_controller_.ExpectedQueueTime() >=
- slacked_pacer_flags_.max_low_precision_expected_queue_time
- .Value()) {
- precision = TaskQueueBase::DelayPrecision::kHigh;
+ // Check for cases where we need high precision.
+ if (precision == TaskQueueBase::DelayPrecision::kLow) {
+ auto& packets_per_type =
+ pacing_controller_.SizeInPacketsPerRtpPacketMediaType();
+ bool audio_or_retransmission_packets_in_queue =
+ packets_per_type[static_cast<size_t>(RtpPacketMediaType::kAudio)] >
+ 0 ||
+ packets_per_type[static_cast<size_t>(
+ RtpPacketMediaType::kRetransmission)] > 0;
+ bool queue_time_too_large =
+ slacked_pacer_flags_.max_low_precision_expected_queue_time &&
+ pacing_controller_.ExpectedQueueTime() >=
+ slacked_pacer_flags_.max_low_precision_expected_queue_time
+ .Value();
+ if (audio_or_retransmission_packets_in_queue || queue_time_too_large) {
+ precision = TaskQueueBase::DelayPrecision::kHigh;
+ }
}
task_queue_.PostDelayedTaskWithPrecision(
