Add functionality which limits the number of bytes on the network.

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 10 matching lines @@
 #include "webrtc/modules/pacing/bitrate_prober.h"
 #include "webrtc/modules/utility/include/process_thread.h"
 #include "webrtc/rtc_base/checks.h"
 #include "webrtc/rtc_base/logging.h"
 #include "webrtc/system_wrappers/include/clock.h"
 #include "webrtc/system_wrappers/include/field_trial.h"
 
 namespace {
 // Time limit in milliseconds between packet bursts.
 const int64_t kMinPacketLimitMs = 5;
+const int64_t kPausedPacketIntervalMs = 500;
 
 // 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.
 
@@ skipping 230 matching lines @@
 }
 
 PacedSender::~PacedSender() {}
 
 void PacedSender::CreateProbeCluster(int bitrate_bps) {
   rtc::CritScope cs(&critsect_);
   prober_->CreateProbeCluster(bitrate_bps, clock_->TimeInMilliseconds());
 }
 
 void PacedSender::Pause() {
-  LOG(LS_INFO) << "PacedSender paused.";
   {
     rtc::CritScope cs(&critsect_);
+    if (!paused_)
+      LOG(LS_INFO) << "PacedSender paused.";
     paused_ = true;
   }
   // Tell the process thread to call our TimeUntilNextProcess() method to get
   // a new (longer) estimate for when to call Process().
   if (process_thread_)
     process_thread_->WakeUp(this);
 }
 
 void PacedSender::Resume() {
-  LOG(LS_INFO) << "PacedSender resumed.";
   {
     rtc::CritScope cs(&critsect_);
+    if (paused_)
+      LOG(LS_INFO) << "PacedSender resumed.";
     paused_ = false;
   }
   // Tell the process thread to call our TimeUntilNextProcess() method to
   // refresh the estimate for when to call Process().
   if (process_thread_)
     process_thread_->WakeUp(this);
 }
 
 void PacedSender::SetProbingEnabled(bool enabled) {
   RTC_CHECK_EQ(0, packet_counter_);
@@ skipping 81 matching lines @@
 }
 
 int64_t PacedSender::AverageQueueTimeMs() {
   rtc::CritScope cs(&critsect_);
   packets_->UpdateQueueTime(clock_->TimeInMilliseconds());
   return packets_->AverageQueueTimeMs();
 }
 
 int64_t PacedSender::TimeUntilNextProcess() {
   rtc::CritScope cs(&critsect_);
+  int64_t elapsed_time_us = clock_->TimeInMicroseconds() - time_last_update_us_;
+  int64_t elapsed_time_ms = (elapsed_time_us + 500) / 1000;
+  // When paused we wake up every 500 ms to send a padding packet to ensure
+  // we won't get stuck in the paused state due to no feedback being received.
   if (paused_)
-    return 1000 * 60 * 60;
+    return std::max<int64_t>(kPausedPacketIntervalMs - elapsed_time_ms, 0);
 
   if (prober_->IsProbing()) {
     int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds());
     if (ret > 0 || (ret == 0 && !probing_send_failure_))
       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);
 }
 
 void PacedSender::Process() {
   int64_t now_us = clock_->TimeInMicroseconds();
   rtc::CritScope cs(&critsect_);
   int64_t elapsed_time_ms = (now_us - time_last_update_us_ + 500) / 1000;
-  time_last_update_us_ = now_us;
   int target_bitrate_kbps = pacing_bitrate_kbps_;
-  if (!paused_ && elapsed_time_ms > 0) {
+
+  if (paused_) {
+    PacedPacketInfo pacing_info;
+    size_t bytes_sent = SendPadding(1, pacing_info);

[philipel, 2017/08/02 15:35:16]

Is there a way to completely stop the pacer, not just pause it and still have it
send padding.

[stefan-webrtc, 2017/08/04 09:50:00]

We could, but the problem is that if we lose some feedback packets, or if the
packets in flight are lost, we may never resume. We need to send something to
ensure that we sooner or later get feedback and can resume...

+    alr_detector_->OnBytesSent(bytes_sent, now_us / 1000);
+    time_last_update_us_ = now_us;
+    return;
+  }
+
+  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, queue_time_limit - 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);
 
     elapsed_time_ms = std::min(kMaxIntervalTimeMs, elapsed_time_ms);
     UpdateBudgetWithElapsedTime(elapsed_time_ms);
   }
 
+  time_last_update_us_ = now_us;
+
   bool is_probing = prober_->IsProbing();
   PacedPacketInfo pacing_info;
   size_t bytes_sent = 0;
   size_t recommended_probe_size = 0;
   if (is_probing) {
     pacing_info = prober_->CurrentCluster();
     recommended_probe_size = prober_->RecommendedMinProbeSize();
   }
   while (!packets_->Empty()) {
     // 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, pacing_info)) {
       // Send succeeded, remove it from the queue.
       if (first_sent_packet_ms_ == -1)
         first_sent_packet_ms_ = clock_->TimeInMilliseconds();
       bytes_sent += packet.bytes;
       packets_->FinalizePop(packet);
       if (is_probing && bytes_sent > recommended_probe_size)
         break;
     } else {
       // Send failed, put it back into the queue.
       packets_->CancelPop(packet);
       break;
     }
   }
 
-  if (packets_->Empty() && !paused_) {
-    // We can not send padding unless a normal packet has first been sent. If we
-    // do, timestamps get messed up.
-    if (packet_counter_ > 0) {
-      int padding_needed =
-          static_cast<int>(is_probing ? (recommended_probe_size - bytes_sent)
-                                      : padding_budget_->bytes_remaining());
-
-      if (padding_needed > 0)
-        bytes_sent += SendPadding(padding_needed, pacing_info);
-    }
+  if (packets_->Empty()) {

[philipel, 2017/08/02 15:35:16]

Is this correct?

I assume you want to be able to send padding while paused, even if there are
packets in the queue?

[stefan-webrtc, 2017/08/04 09:50:00]

Why would we do that if there are packets in the queue? I don't think this is a
change in behavior?

[philipel, 2017/08/04 14:13:04]

Ah, you're right.

+    int padding_needed =
+        static_cast<int>(is_probing ? (recommended_probe_size - bytes_sent)
+                                    : padding_budget_->bytes_remaining());
+    if (padding_needed > 0)
+      bytes_sent += SendPadding(padding_needed, pacing_info);
   }
   if (is_probing) {
     probing_send_failure_ = bytes_sent == 0;
     if (!probing_send_failure_)
       prober_->ProbeSent(clock_->TimeInMilliseconds(), bytes_sent);
   }
   alr_detector_->OnBytesSent(bytes_sent, now_us / 1000);
 }
 
 void PacedSender::ProcessThreadAttached(ProcessThread* process_thread) {
   LOG(LS_INFO) << "ProcessThreadAttached 0x" << std::hex << process_thread;
   process_thread_ = process_thread;
 }
 
 bool PacedSender::SendPacket(const paced_sender::Packet& packet,
                              const PacedPacketInfo& pacing_info) {
-  if (paused_)
-    return false;

[philipel, 2017/08/02 15:35:16]

Make it into an RTC_DCHECK

[stefan-webrtc, 2017/08/04 09:50:00]

Done.

   if (media_budget_->bytes_remaining() == 0 &&
       pacing_info.probe_cluster_id == PacedPacketInfo::kNotAProbe) {
     return false;
   }
 
   critsect_.Leave();
   const bool success = packet_sender_->TimeToSendPacket(
       packet.ssrc, packet.sequence_number, packet.capture_time_ms,
       packet.retransmission, pacing_info);
   critsect_.Enter();
 
   if (success) {
     // TODO(holmer): High priority packets should only be accounted for if we
     // are allocating bandwidth for audio.
     if (packet.priority != kHighPriority) {
       // Update media bytes sent.
       UpdateBudgetWithBytesSent(packet.bytes);
     }
   }
 
   return success;
 }
 
 size_t PacedSender::SendPadding(size_t padding_needed,
                                 const PacedPacketInfo& pacing_info) {
+  // We can not send padding unless a normal packet has first been sent. If we
+  // do, timestamps get messed up.
+  if (packet_counter_ == 0)

[philipel, 2017/08/02 15:35:16]

I think it's somewhat clearer to keep this check where it was, moving it here
just makes this function behave in an unexpected way.

Maybe even add a DCHECK_GT for the |packet_counter_|.

[stefan-webrtc, 2017/08/04 09:50:00]

Then the code will have to be duplicated since we're calling SendPadding from
two places now. Is that preferable?

[philipel, 2017/08/04 14:13:04]

I still think that would be preferable.

+    return 0;
   critsect_.Leave();
   size_t bytes_sent =
       packet_sender_->TimeToSendPadding(padding_needed, pacing_info);
   critsect_.Enter();
 
   if (bytes_sent > 0) {
     UpdateBudgetWithBytesSent(bytes_sent);
   }
   return bytes_sent;
 }
@@ skipping 12 matching lines @@
   rtc::CritScope cs(&critsect_);
   pacing_factor_ = pacing_factor;
 }
 
 void PacedSender::SetQueueTimeLimit(int limit_ms) {
   rtc::CritScope cs(&critsect_);
   queue_time_limit = limit_ms;
 }
 
 }  // namespace webrtc