Allow pacer to boost bitrate in order to meet time constraints.

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.
  */
 
 #include "webrtc/modules/pacing/include/paced_sender.h"
 
-#include <assert.h>
-
 #include <map>
 #include <queue>
 #include <set>
 
+#include "webrtc/base/checks.h"
 #include "webrtc/modules/interface/module_common_types.h"
 #include "webrtc/modules/pacing/bitrate_prober.h"
 #include "webrtc/system_wrappers/interface/clock.h"
 #include "webrtc/system_wrappers/interface/critical_section_wrapper.h"
 #include "webrtc/system_wrappers/interface/field_trial.h"
 #include "webrtc/system_wrappers/interface/logging.h"
 
 namespace {
 // Time limit in milliseconds between packet bursts.
 const int64_t kMinPacketLimitMs = 5;
@@ skipping 111 matching lines @@
       dupe_map_[packet.ssrc].insert(packet.sequence_number);
       return true;
     }
 
     // Insert returns a pair, where second is a bool set to true if new element.
     return it->second.insert(packet.sequence_number).second;
   }
 
   void RemoveFromDupeSet(const Packet& packet) {
     SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc);
-    assert(it != dupe_map_.end());
+    RTC_DCHECK(it != dupe_map_.end());
     it->second.erase(packet.sequence_number);
     if (it->second.empty()) {
       dupe_map_.erase(it);
     }
   }
 
   // List of packets, in the order the were enqueued. Since dequeueing may
   // occur out of order, use list instead of vector.
   std::list<Packet> packet_list_;
   // Priority queue of the packets, sorted according to Comparator.
@@ skipping 41 matching lines @@
   int target_rate_kbps() const { return target_rate_kbps_; }
 
  private:
   static const int kWindowMs = 500;
 
   int target_rate_kbps_;
   int bytes_remaining_;
 };
 }  // namespace paced_sender
 
+const int64_t PacedSender::kMaxQueueLengthMs = 2000;
 const float PacedSender::kDefaultPaceMultiplier = 2.5f;
 
 PacedSender::PacedSender(Clock* clock,
                          Callback* callback,
                          int bitrate_kbps,
                          int max_bitrate_kbps,
                          int min_bitrate_kbps)
     : clock_(clock),
       callback_(callback),
       critsect_(CriticalSectionWrapper::CreateCriticalSection()),
       paused_(false),
       probing_enabled_(true),
       media_budget_(new paced_sender::IntervalBudget(max_bitrate_kbps)),
       padding_budget_(new paced_sender::IntervalBudget(min_bitrate_kbps)),
       prober_(new BitrateProber()),
       bitrate_bps_(1000 * bitrate_kbps),
+      max_bitrate_kbps_(max_bitrate_kbps),
+      bitrate_boost_start_us_(-1),
       time_last_update_us_(clock->TimeInMicroseconds()),
       packets_(new paced_sender::PacketQueue()),
       packet_counter_(0) {
   UpdateBytesPerInterval(kMinPacketLimitMs);
 }
 
 PacedSender::~PacedSender() {}
 
 void PacedSender::Pause() {
   CriticalSectionScoped cs(critsect_.get());
   paused_ = true;
 }
 
 void PacedSender::Resume() {
   CriticalSectionScoped cs(critsect_.get());
   paused_ = false;
 }
 
 void PacedSender::SetProbingEnabled(bool enabled) {
-  assert(packet_counter_ == 0);
+  RTC_CHECK_EQ(0u, packet_counter_);
   probing_enabled_ = enabled;
 }
 
 void PacedSender::UpdateBitrate(int bitrate_kbps,
                                 int max_bitrate_kbps,
                                 int min_bitrate_kbps) {
   CriticalSectionScoped cs(critsect_.get());
-  media_budget_->set_target_rate_kbps(max_bitrate_kbps);
+  if (bitrate_boost_start_us_ == -1 ||
+      max_bitrate_kbps > media_budget_->target_rate_kbps()) {
+    media_budget_->set_target_rate_kbps(max_bitrate_kbps);
+    bitrate_boost_start_us_ = -1;
+  }
   padding_budget_->set_target_rate_kbps(min_bitrate_kbps);
   bitrate_bps_ = 1000 * bitrate_kbps;
+  max_bitrate_kbps_ = max_bitrate_kbps;
 }
 
 void PacedSender::InsertPacket(RtpPacketSender::Priority priority,
                                uint32_t ssrc,
                                uint16_t sequence_number,
                                int64_t capture_time_ms,
                                size_t bytes,
                                bool retransmission) {
   CriticalSectionScoped cs(critsect_.get());
 
-  if (probing_enabled_ && !prober_->IsProbing()) {
+  if (probing_enabled_ && !prober_->IsProbing())
     prober_->SetEnabled(true);
-  }
   prober_->MaybeInitializeProbe(bitrate_bps_);
 
-  if (capture_time_ms < 0) {
+  if (capture_time_ms < 0)
     capture_time_ms = clock_->TimeInMilliseconds();
-  }
 
   packets_->Push(paced_sender::Packet(
       priority, ssrc, sequence_number, capture_time_ms,
       clock_->TimeInMilliseconds(), bytes, retransmission, packet_counter_++));
 }
 
 int64_t PacedSender::ExpectedQueueTimeMs() const {
   CriticalSectionScoped cs(critsect_.get());
-  int target_rate = media_budget_->target_rate_kbps();
-  assert(target_rate > 0);
-  return static_cast<int64_t>(packets_->SizeInBytes() * 8 / target_rate);
+  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_->OldestEnqueueTime();
   if (oldest_packet == 0)
     return 0;
 
   return clock_->TimeInMilliseconds() - oldest_packet;
 }
 
 int64_t PacedSender::TimeUntilNextProcess() {
   CriticalSectionScoped cs(critsect_.get());
   if (prober_->IsProbing()) {
     int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds());
-    if (ret >= 0) {
+    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;
   if (elapsed_time_ms > 0) {
+    if (bitrate_boost_start_us_ != -1 &&

[stefan-webrtc, 2015/10/28 15:45:23]

Is the bitrate_boost_start_us_ really needed? Can't we compute a new
required_bitrate_kbps every time we call Process() as long as the
expected_queue_length_ms is greater than the max queue length? Maybe even go as
far as always doing:

media_budget_->set_target_rate_kbps(std::max(required_bitrate_kbps,
max_bitrate_kbps_));

If the expected_queue_length_ms goes below kMaxQueueLengthMs,
required_bitrate_kbps will automatically be less than max_bitrate_kbps_, and we
will automatically be back to normal.

[sprang_webrtc, 2015/11/02 12:17:56]

This was how I first implemented it, and it turns out it doesn't work (if we
actually want a packet that enters the queue to be out two seconds later). The
reason is that for each Process() call we calculate how fast we need to go to
get everything currently in the queue out before two seconds from _now_, not
from when the packets entered the queue. 

A simplified example:

We have 1000 bytes in the queue. Transmit rate is 250 bytes per second. In order
to get everything out in two seconds we need 500 bytes per second transmit rate,
so use that. 

One second later we call Process() again, we have transmitted half the queue so
have 500 bytes left. Since the transmit rate is 250 bytes per second is
sufficient to send 500 bytes in one seconds we use that for the remainder of the
queue.

The result is that the data in the queue took three seconds to transmit. Calling
Process() more frequently can make this behavior worse...

[stefan-webrtc, 2015/11/10 14:57:02]

This isn't correct I think. Those 500 bytes have been in the queue for one
second already, so therefore they must be sent within the next second.
I do understand the issue related to this based on the offline discussion.

There are two events affecting the boosted pacing rate, packets being added or
removed from the queue. If we know we always have a pacing rate which is at
least high enough for the queue to empty within 2 seconds, shouldn't we then be
able to modify the rate every time a packet is added or removed?

Say if we have 1000 bytes packets in there, as in your example and the boosted
rate is 500 bytes/second (while the actual rate is 250 bytes/sec). Then we add
one packet 100 ms after the last packet in the queue was added, shouldn't we
then be able to update the boosted rate with the new packet given that we know
how long the estimated queue currently is? For instance:

t=0: 500 bytes added, rate=500/2 = 250 bytes/s, q=2 sec
t=0.1: 500 bytes added, rate=(rate*q + 500) / (q+0.1) = 476 bytes/sec, q=2.1 sec
and so on...

Does this make sense? I will still blame my jetlag if it doesn't. :P

[stefan-webrtc, 2015/11/10 14:59:44]

And something similar could be done when a packet is removed from the queue...
and we have to make sure the rate never goes below max_bitrate_kbps_.

[sprang_webrtc, 2015/11/19 10:21:48]

This method won't work quite right either, since we can't weight bitrate needed.
Also, the method we discussed offline also won't work, since it again will
require going through the list of packets and update min send bitrate if the
actual send bitrate exceeds the min needed (or we risk significantly
overshooting the rate needed if bitrate drops).

In lieu of this I've implemented a simpler approach which also isn't perfectly
accurate but it is quite cheap and should be good enough. Basically we track the
average queue time of the packets in the pacer. 

Assuming (among other things) equal input and output rates, we have 2s -
avg_queue_time to send the full size of the queue. If we disregard input rate,
queue size / 2 would be more accurate, but I'm not sure about that assumption...


WDYT?

+        (now_us - bitrate_boost_start_us_) > kMaxQueueLengthMs * 1000) {
+      // Boost period is up, reset to nominal bitrate.
+      media_budget_->set_target_rate_kbps(max_bitrate_kbps_);
+      bitrate_boost_start_us_ = -1;
+    }
+
+    int64_t expected_queue_length_ms =
+        static_cast<int64_t>(packets_->SizeInBytes() * 8 / max_bitrate_kbps_);

[stefan-webrtc, 2015/10/28 15:45:23]

Call ExpectedQueueTimeMs() instead.

[sprang_webrtc, 2015/11/02 12:17:56]

Sure. Or do we care about recursively taking locks?

[stefan-webrtc, 2015/11/10 14:57:02]

I would prefer to not take them recursively. Just create a private version of
the method which requires the lock to be acquired. Maybe name it Internal

+    if (expected_queue_length_ms > kMaxQueueLengthMs) {
+      // Too much data has been put in the pacer queue, or the target bitrate
+      // has suddenly decreased, making the expected max pacer delay too long.
+      // Boost the media target bitrate so that all data should have left the
+      // queue within kDefaultMaxQueueLengthMs.
+      int required_bitrate_kbps =
+          expected_queue_length_ms * max_bitrate_kbps_ / kMaxQueueLengthMs;
+      if (required_bitrate_kbps > media_budget_->target_rate_kbps()) {
+        media_budget_->set_target_rate_kbps(required_bitrate_kbps);
+        bitrate_boost_start_us_ = now_us;
+      }
+    }
+
     int64_t delta_time_ms = std::min(kMaxIntervalTimeMs, elapsed_time_ms);
     UpdateBytesPerInterval(delta_time_ms);
   }
   while (!packets_->Empty()) {
-    if (media_budget_->bytes_remaining() == 0 && !prober_->IsProbing()) {
+    if (media_budget_->bytes_remaining() == 0 && !prober_->IsProbing())
       return 0;
-    }
 
     // Since we need to release the lock in order to send, we first pop the
     // element from the priority queue but keep it in storage, so that we can
     // reinsert it if send fails.
     const paced_sender::Packet& packet = packets_->BeginPop();
     if (SendPacket(packet)) {
       // Send succeeded, remove it from the queue.
       packets_->FinalizePop(packet);
-      if (prober_->IsProbing()) {
+      if (prober_->IsProbing())
         return 0;
-      }
     } else {
       // Send failed, put it back into the queue.
       packets_->CancelPop(packet);
       return 0;
     }
   }
 
   if (!packets_->Empty())
     return 0;
 
   size_t padding_needed;
-  if (prober_->IsProbing())
+  if (prober_->IsProbing()) {
     padding_needed = prober_->RecommendedPacketSize();
-  else
+  } else {
     padding_needed = padding_budget_->bytes_remaining();
+  }
 
   if (padding_needed > 0)
     SendPadding(static_cast<size_t>(padding_needed));
   return 0;
 }
 
 bool PacedSender::SendPacket(const paced_sender::Packet& packet) {
   critsect_->Leave();
   const bool success = callback_->TimeToSendPacket(packet.ssrc,
                                                    packet.sequence_number,
@@ skipping 21 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