Fix bug in calculation of averge queue time in paced sender.
Also work around a flaw in fake encoder which caused bogus perf
regression in rampup tests.
BUG=560434
R=mflodman@webrtc.org, stefan@webrtc.org
Review URL: https://codereview.webrtc.org/1474533006 .
Cr-Commit-Position: refs/heads/master@{#10811}
diff --git a/webrtc/modules/pacing/paced_sender.cc b/webrtc/modules/pacing/paced_sender.cc
index e38405a..dcdf64e 100644
--- a/webrtc/modules/pacing/paced_sender.cc
+++ b/webrtc/modules/pacing/paced_sender.cc
@@ -32,6 +32,9 @@
} // namespace
+// TODO(sprang): Move at least PacketQueue and MediaBudget out to separate
+// files, so that we can more easily test them.
+
namespace webrtc {
namespace paced_sender {
struct Packet {
@@ -93,9 +96,11 @@
virtual ~PacketQueue() {}
void Push(const Packet& packet) {
- if (!AddToDupeSet(packet)) {
+ if (!AddToDupeSet(packet))
return;
- }
+
+ UpdateQueueTime(packet.enqueue_time_ms);
+
// Store packet in list, use pointers in priority queue for cheaper moves.
// Packets have a handle to its own iterator in the list, for easy removal
// when popping from queue.
@@ -119,6 +124,9 @@
bytes_ -= packet.bytes;
queue_time_sum_ -= (time_last_updated_ - packet.enqueue_time_ms);
packet_list_.erase(packet.this_it);
+ RTC_DCHECK_EQ(packet_list_.size(), prio_queue_.size());
+ if (packet_list_.empty())
+ RTC_DCHECK_EQ(0u, queue_time_sum_);
}
bool Empty() const { return prio_queue_.empty(); }
@@ -134,13 +142,20 @@
return it->enqueue_time_ms;
}
- int64_t AverageQueueTimeMs() {
- int64_t now = clock_->TimeInMilliseconds();
- RTC_DCHECK_GE(now, time_last_updated_);
- int64_t delta = now - time_last_updated_;
- queue_time_sum_ += delta * prio_queue_.size();
- time_last_updated_ = now;
- return queue_time_sum_ / prio_queue_.size();
+ void UpdateQueueTime(int64_t timestamp_ms) {
+ RTC_DCHECK_GE(timestamp_ms, time_last_updated_);
+ int64_t delta = timestamp_ms - time_last_updated_;
+ // Use packet packet_list_.size() not prio_queue_.size() here, as there
+ // might be an outstanding element popped from prio_queue_ currently in the
+ // SendPacket() call, while packet_list_ will always be correct.
+ queue_time_sum_ += delta * packet_list_.size();
+ time_last_updated_ = timestamp_ms;
+ }
+
+ int64_t AverageQueueTimeMs() const {
+ if (prio_queue_.empty())
+ return 0;
+ return queue_time_sum_ / packet_list_.size();
}
private:
@@ -290,12 +305,13 @@
prober_->SetEnabled(true);
prober_->MaybeInitializeProbe(bitrate_bps_);
+ int64_t now_ms = clock_->TimeInMilliseconds();
if (capture_time_ms < 0)
- capture_time_ms = clock_->TimeInMilliseconds();
+ capture_time_ms = now_ms;
- packets_->Push(paced_sender::Packet(
- priority, ssrc, sequence_number, capture_time_ms,
- clock_->TimeInMilliseconds(), bytes, retransmission, packet_counter_++));
+ packets_->Push(paced_sender::Packet(priority, ssrc, sequence_number,
+ capture_time_ms, now_ms, bytes,
+ retransmission, packet_counter_++));
}
int64_t PacedSender::ExpectedQueueTimeMs() const {
@@ -319,6 +335,12 @@
return clock_->TimeInMilliseconds() - oldest_packet;
}
+int64_t PacedSender::AverageQueueTimeMs() {
+ CriticalSectionScoped cs(critsect_.get());
+ packets_->UpdateQueueTime(clock_->TimeInMilliseconds());
+ return packets_->AverageQueueTimeMs();
+}
+
int64_t PacedSender::TimeUntilNextProcess() {
CriticalSectionScoped cs(critsect_.get());
if (prober_->IsProbing()) {
@@ -345,6 +367,7 @@
// 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(clock_->TimeInMilliseconds());
int64_t avg_time_left_ms = std::max<int64_t>(
1, kMaxQueueLengthMs - packets_->AverageQueueTimeMs());
int min_bitrate_needed_kbps =
diff --git a/webrtc/modules/pacing/paced_sender.h b/webrtc/modules/pacing/paced_sender.h
index d1e5ce3..62e794f 100644
--- a/webrtc/modules/pacing/paced_sender.h
+++ b/webrtc/modules/pacing/paced_sender.h
@@ -113,6 +113,10 @@
// packets in the queue, given the current size and bitrate, ignoring prio.
virtual int64_t ExpectedQueueTimeMs() const;
+ // Returns the average time since being enqueued, in milliseconds, for all
+ // packets currently in the pacer queue, or 0 if queue is empty.
+ virtual int64_t AverageQueueTimeMs();
+
// Returns the number of milliseconds until the module want a worker thread
// to call Process.
int64_t TimeUntilNextProcess() override;
diff --git a/webrtc/modules/pacing/paced_sender_unittest.cc b/webrtc/modules/pacing/paced_sender_unittest.cc
index 1a2936d..bf00a05 100644
--- a/webrtc/modules/pacing/paced_sender_unittest.cc
+++ b/webrtc/modules/pacing/paced_sender_unittest.cc
@@ -825,5 +825,50 @@
send_bucket_->Process();
}
+TEST_F(PacedSenderTest, AverageQueueTime) {
+ uint32_t ssrc = 12346;
+ uint16_t sequence_number = 1234;
+ const size_t kPacketSize = 1200;
+ const int kBitrateBps = 10 * kPacketSize * 8; // 10 packets per second.
+ const int kBitrateKbps = (kBitrateBps + 500) / 1000;
+
+ send_bucket_->UpdateBitrate(kBitrateKbps, kBitrateKbps, kBitrateKbps);
+
+ EXPECT_EQ(0, send_bucket_->AverageQueueTimeMs());
+
+ int64_t first_capture_time = clock_.TimeInMilliseconds();
+ send_bucket_->InsertPacket(PacedSender::kHighPriority, ssrc, sequence_number,
+ first_capture_time, kPacketSize, false);
+ clock_.AdvanceTimeMilliseconds(10);
+ send_bucket_->InsertPacket(PacedSender::kHighPriority, ssrc,
+ sequence_number + 1, clock_.TimeInMilliseconds(),
+ kPacketSize, false);
+ clock_.AdvanceTimeMilliseconds(10);
+
+ EXPECT_EQ((20 + 10) / 2, send_bucket_->AverageQueueTimeMs());
+
+ // Only first packet (queued for 20ms) should be removed, leave the second
+ // packet (queued for 10ms) alone in the queue.
+ EXPECT_CALL(callback_, TimeToSendPacket(ssrc, sequence_number,
+ first_capture_time, false))
+ .Times(1)
+ .WillRepeatedly(Return(true));
+ send_bucket_->Process();
+
+ EXPECT_EQ(10, send_bucket_->AverageQueueTimeMs());
+
+ clock_.AdvanceTimeMilliseconds(10);
+ EXPECT_CALL(callback_, TimeToSendPacket(ssrc, sequence_number + 1,
+ first_capture_time + 10, false))
+ .Times(1)
+ .WillRepeatedly(Return(true));
+ for (int i = 0; i < 3; ++i) {
+ clock_.AdvanceTimeMilliseconds(30); // Max delta.
+ send_bucket_->Process();
+ }
+
+ EXPECT_EQ(0, send_bucket_->AverageQueueTimeMs());
+}
+
} // namespace test
} // namespace webrtc
diff --git a/webrtc/test/fake_encoder.cc b/webrtc/test/fake_encoder.cc
index 44fb1c5..165fd3e 100644
--- a/webrtc/test/fake_encoder.cc
+++ b/webrtc/test/fake_encoder.cc
@@ -57,6 +57,11 @@
// at the display time of the previous frame.
time_since_last_encode_ms = time_now_ms - last_encode_time_ms_;
}
+ if (time_since_last_encode_ms > 3 * 1000 / config_.maxFramerate) {
+ // Rudimentary check to make sure we don't widely overshoot bitrate target
+ // when resuming encoding after a suspension.
+ time_since_last_encode_ms = 3 * 1000 / config_.maxFramerate;
+ }
size_t bits_available =
static_cast<size_t>(target_bitrate_kbps_ * time_since_last_encode_ms);
