Fix bug in calculation of averge queue time in paced sender.

webrtc/modules/pacing/paced_sender.cc

 /*
  *  Copyright (c) 2012 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.
  */
 
@@ skipping 14 matching lines @@
 namespace {
 // Time limit in milliseconds between packet bursts.
 const int64_t kMinPacketLimitMs = 5;
 
 // Upper cap on process interval, in case process has not been called in a long
 // time.
 const int64_t kMaxIntervalTimeMs = 30;
 
 }  // 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 {
   Packet(RtpPacketSender::Priority priority,
          uint32_t ssrc,
          uint16_t seq_number,
          int64_t capture_time_ms,
          int64_t enqueue_time_ms,
          size_t length_in_bytes,
          bool retransmission,
@@ skipping 41 matching lines @@
 class PacketQueue {
  public:
   explicit PacketQueue(Clock* clock)
       : bytes_(0),
         clock_(clock),
         queue_time_sum_(0),
         time_last_updated_(clock_->TimeInMilliseconds()) {}
   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.
     packet_list_.push_front(packet);
     std::list<Packet>::iterator it = packet_list_.begin();
     it->this_it = it;          // Handle for direct removal from list.
     prio_queue_.push(&(*it));  // Pointer into list.
     bytes_ += packet.bytes;
   }
 
   const Packet& BeginPop() {
     const Packet& packet = *prio_queue_.top();
     prio_queue_.pop();
     return packet;
   }
 
   void CancelPop(const Packet& packet) { prio_queue_.push(&(*packet.this_it)); }
 
   void FinalizePop(const Packet& packet) {
     RemoveFromDupeSet(packet);
     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(); }
 
   size_t SizeInPackets() const { return prio_queue_.size(); }
 
   uint64_t SizeInBytes() const { return bytes_; }
 
   int64_t OldestEnqueueTimeMs() const {
     auto it = packet_list_.rbegin();
     if (it == packet_list_.rend())
       return 0;
     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:
   // Try to add a packet to the set of ssrc/seqno identifiers currently in the
   // queue. Return true if inserted, false if this is a duplicate.
   bool AddToDupeSet(const Packet& packet) {
     SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc);
     if (it == dupe_map_.end()) {
       // First for this ssrc, just insert.
       dupe_map_[packet.ssrc].insert(packet.sequence_number);
@@ skipping 129 matching lines @@
                                uint16_t sequence_number,
                                int64_t capture_time_ms,
                                size_t bytes,
                                bool retransmission) {
   CriticalSectionScoped cs(critsect_.get());
 
   if (probing_enabled_ && !prober_->IsProbing())
     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 {
   CriticalSectionScoped cs(critsect_.get());
   RTC_DCHECK_GT(max_bitrate_kbps_, 0);
   return static_cast<int64_t>(packets_->SizeInBytes() * 8 / max_bitrate_kbps_);
 }
 
 size_t PacedSender::QueueSizePackets() const {
   CriticalSectionScoped cs(critsect_.get());
   return packets_->SizeInPackets();
 }
 
 int64_t PacedSender::QueueInMs() const {
   CriticalSectionScoped cs(critsect_.get());
 
   int64_t oldest_packet = packets_->OldestEnqueueTimeMs();
   if (oldest_packet == 0)
     return 0;
 
   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()) {
     int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds());
     if (ret >= 0)
       return ret;
   }
   int64_t elapsed_time_us = clock_->TimeInMicroseconds() - time_last_update_us_;
   int64_t elapsed_time_ms = (elapsed_time_us + 500) / 1000;
   return std::max<int64_t>(kMinPacketLimitMs - elapsed_time_ms, 0);
 }
 
 int32_t PacedSender::Process() {
   int64_t now_us = clock_->TimeInMicroseconds();
   CriticalSectionScoped cs(critsect_.get());
   int64_t elapsed_time_ms = (now_us - time_last_update_us_ + 500) / 1000;
   time_last_update_us_ = now_us;
   if (paused_)
     return 0;
   int target_bitrate_kbps = max_bitrate_kbps_;
   if (elapsed_time_ms > 0) {
     size_t queue_size_bytes = packets_->SizeInBytes();
     if (queue_size_bytes > 0) {
       // 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 =
           static_cast<int>(queue_size_bytes * 8 / avg_time_left_ms);
       if (min_bitrate_needed_kbps > target_bitrate_kbps)
         target_bitrate_kbps = min_bitrate_needed_kbps;
     }
 
     media_budget_->set_target_rate_kbps(target_bitrate_kbps);
 
@@ skipping 63 matching lines @@
     media_budget_->UseBudget(bytes_sent);
     padding_budget_->UseBudget(bytes_sent);
   }
 }
 
 void PacedSender::UpdateBytesPerInterval(int64_t delta_time_ms) {
   media_budget_->IncreaseBudget(delta_time_ms);
   padding_budget_->IncreaseBudget(delta_time_ms);
 }
 }  // namespace webrtc