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 11 matching lines @@
 #include "webrtc/modules/pacing/interval_budget.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 out to separate
 // files, so that we can more easily test them.
 
@@ skipping 190 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;
+    time_last_update_us_ = now_us;
+    // We can not send padding unless a normal packet has first been sent. If we
+    // do, timestamps get messed up.
+    if (packet_counter_ == 0)
+      return;
+    size_t bytes_sent = SendPadding(1, pacing_info);
+    alr_detector_->OnBytesSent(bytes_sent, now_us / 1000);
+    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_) {
+  if (packets_->Empty()) {
     // 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 (is_probing) {
     probing_send_failure_ = bytes_sent == 0;
     if (!probing_send_failure_)
       prober_->ProbeSent(clock_->TimeInMilliseconds(), bytes_sent);
   }
   alr_detector_->OnBytesSent(bytes_sent, elapsed_time_ms);
 }
 
 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;
+  RTC_DCHECK(!paused_);
   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.
       // TODO(eladalon): TimeToSendPacket() can also return |true| in some
       // situations where nothing actually ended up being sent to the network,
       // and we probably don't want to update the budget in such cases.
       // https://bugs.chromium.org/p/webrtc/issues/detail?id=8052
       UpdateBudgetWithBytesSent(packet.bytes);
     }
   }
 
   return success;
 }
 
 size_t PacedSender::SendPadding(size_t padding_needed,
                                 const PacedPacketInfo& pacing_info) {
+  RTC_DCHECK_GT(packet_counter_, 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