src/third_party/webrtc/modules/pacing/paced_sender.cc

   1 /*
   2  *  Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
   3  *
   4  *  Use of this source code is governed by a BSD-style license
   5  *  that can be found in the LICENSE file in the root of the source
   6  *  tree. An additional intellectual property rights grant can be found
   7  *  in the file PATENTS.  All contributing project authors may
   8  *  be found in the AUTHORS file in the root of the source tree.
   9  */
  10
  11 #include "webrtc/modules/pacing/include/paced_sender.h"
  12
  13 #include <assert.h>
  14
  15 #include "webrtc/modules/interface/module_common_types.h"
  16 #include "webrtc/system_wrappers/interface/critical_section_wrapper.h"
  17 #include "webrtc/system_wrappers/interface/trace_event.h"
  18
  19 namespace {
  20 // Time limit in milliseconds between packet bursts.
  21 const int kMinPacketLimitMs = 5;
  22
  23 // Upper cap on process interval, in case process has not been called in a long
  24 // time.
  25 const int kMaxIntervalTimeMs = 30;
  26
  27 // Max time that the first packet in the queue can sit in the queue if no
  28 // packets are sent, regardless of buffer state. In practice only in effect at
  29 // low bitrates (less than 320 kbits/s).
  30 const int kMaxQueueTimeWithoutSendingMs = 30;
  31
  32 }  // namespace
  33
  34 namespace webrtc {
  35
  36 namespace paced_sender {
  37 struct Packet {
  38   Packet(uint32_t ssrc, uint16_t seq_number, int64_t capture_time_ms,
  39          int length_in_bytes)
  40       : ssrc_(ssrc),
  41         sequence_number_(seq_number),
  42         capture_time_ms_(capture_time_ms),
  43         bytes_(length_in_bytes) {
  44   }
  45   uint32_t ssrc_;
  46   uint16_t sequence_number_;
  47   int64_t capture_time_ms_;
  48   int bytes_;
  49 };
  50
  51 // STL list style class which prevents duplicates in the list.
  52 class PacketList {
  53  public:
  54   PacketList() {};
  55
  56   bool empty() const {
  57     return packet_list_.empty();
  58   }
  59
  60   Packet front() const {
  61     return packet_list_.front();
  62   }
  63
  64   void pop_front() {
  65     Packet& packet = packet_list_.front();
  66     uint16_t sequence_number = packet.sequence_number_;
  67     packet_list_.pop_front();
  68     sequence_number_set_.erase(sequence_number);
  69   }
  70
  71   void push_back(const Packet& packet) {
  72     if (sequence_number_set_.find(packet.sequence_number_) ==
  73         sequence_number_set_.end()) {
  74       // Don't insert duplicates.
  75       packet_list_.push_back(packet);
  76       sequence_number_set_.insert(packet.sequence_number_);
  77     }
  78   }
  79
  80  private:
  81   std::list<Packet> packet_list_;
  82   std::set<uint16_t> sequence_number_set_;
  83 };
  84
  85 class IntervalBudget {
  86  public:
  87   explicit IntervalBudget(int initial_target_rate_kbps)
  88       : target_rate_kbps_(initial_target_rate_kbps),
  89         bytes_remaining_(0) {}
  90
  91   void set_target_rate_kbps(int target_rate_kbps) {
  92     target_rate_kbps_ = target_rate_kbps;
  93   }
  94
  95   void IncreaseBudget(int delta_time_ms) {
  96     int bytes = target_rate_kbps_ * delta_time_ms / 8;
  97     if (bytes_remaining_ < 0) {
  98       // We overused last interval, compensate this interval.
  99       bytes_remaining_ = bytes_remaining_ + bytes;
 100     } else {
 101       // If we underused last interval we can't use it this interval.
 102       bytes_remaining_ = bytes;
 103     }
 104   }
 105
 106   void UseBudget(int bytes) {
 107     bytes_remaining_ = std::max(bytes_remaining_ - bytes,
 108                                 -100 * target_rate_kbps_ / 8);
 109   }
 110
 111   int bytes_remaining() const { return bytes_remaining_; }
 112
 113  private:
 114   int target_rate_kbps_;
 115   int bytes_remaining_;
 116 };
 117 }  // namespace paced_sender
 118
 119 PacedSender::PacedSender(Callback* callback, int target_bitrate_kbps,
 120                          float pace_multiplier)
 121     : callback_(callback),
 122       pace_multiplier_(pace_multiplier),
 123       enabled_(false),
 124       paused_(false),
 125       critsect_(CriticalSectionWrapper::CreateCriticalSection()),
 126       media_budget_(new paced_sender::IntervalBudget(
 127           pace_multiplier_ * target_bitrate_kbps)),
 128       padding_budget_(new paced_sender::IntervalBudget(0)),
 129       // No padding until UpdateBitrate is called.
 130       pad_up_to_bitrate_budget_(new paced_sender::IntervalBudget(0)),
 131       time_last_update_(TickTime::Now()),
 132       capture_time_ms_last_queued_(0),
 133       capture_time_ms_last_sent_(0),
 134       high_priority_packets_(new paced_sender::PacketList),
 135       normal_priority_packets_(new paced_sender::PacketList),
 136       low_priority_packets_(new paced_sender::PacketList) {
 137   UpdateBytesPerInterval(kMinPacketLimitMs);
 138 }
 139
 140 PacedSender::~PacedSender() {
 141 }
 142
 143 void PacedSender::Pause() {
 144   CriticalSectionScoped cs(critsect_.get());
 145   paused_ = true;
 146 }
 147
 148 void PacedSender::Resume() {
 149   CriticalSectionScoped cs(critsect_.get());
 150   paused_ = false;
 151 }
 152
 153 void PacedSender::SetStatus(bool enable) {
 154   CriticalSectionScoped cs(critsect_.get());
 155   enabled_ = enable;
 156 }
 157
 158 bool PacedSender::Enabled() const {
 159   CriticalSectionScoped cs(critsect_.get());
 160   return enabled_;
 161 }
 162
 163 void PacedSender::UpdateBitrate(int target_bitrate_kbps,
 164                                 int max_padding_bitrate_kbps,
 165                                 int pad_up_to_bitrate_kbps) {
 166   CriticalSectionScoped cs(critsect_.get());
 167   media_budget_->set_target_rate_kbps(pace_multiplier_ * target_bitrate_kbps);
 168   padding_budget_->set_target_rate_kbps(max_padding_bitrate_kbps);
 169   pad_up_to_bitrate_budget_->set_target_rate_kbps(pad_up_to_bitrate_kbps);
 170 }
 171
 172 bool PacedSender::SendPacket(Priority priority, uint32_t ssrc,
 173     uint16_t sequence_number, int64_t capture_time_ms, int bytes) {
 174   CriticalSectionScoped cs(critsect_.get());
 175
 176   if (!enabled_) {
 177     return true;  // We can send now.
 178   }
 179   if (capture_time_ms < 0) {
 180     capture_time_ms = TickTime::MillisecondTimestamp();
 181   }
 182   if (priority != kHighPriority &&
 183       capture_time_ms > capture_time_ms_last_queued_) {
 184     capture_time_ms_last_queued_ = capture_time_ms;
 185     TRACE_EVENT_ASYNC_BEGIN1("webrtc_rtp", "PacedSend", capture_time_ms,
 186                              "capture_time_ms", capture_time_ms);
 187   }
 188   paced_sender::PacketList* packet_list = NULL;
 189   switch (priority) {
 190     case kHighPriority:
 191       packet_list = high_priority_packets_.get();
 192       break;
 193     case kNormalPriority:
 194       packet_list = normal_priority_packets_.get();
 195       break;
 196     case kLowPriority:
 197       packet_list = low_priority_packets_.get();
 198       break;
 199   }
 200   packet_list->push_back(paced_sender::Packet(ssrc, sequence_number,
 201                                               capture_time_ms, bytes));
 202   return false;
 203 }
 204
 205 int PacedSender::QueueInMs() const {
 206   CriticalSectionScoped cs(critsect_.get());
 207   int64_t now_ms = TickTime::MillisecondTimestamp();
 208   int64_t oldest_packet_capture_time = now_ms;
 209   if (!high_priority_packets_->empty()) {
 210     oldest_packet_capture_time = std::min(
 211         oldest_packet_capture_time,
 212         high_priority_packets_->front().capture_time_ms_);
 213   }
 214   if (!normal_priority_packets_->empty()) {
 215     oldest_packet_capture_time = std::min(
 216         oldest_packet_capture_time,
 217         normal_priority_packets_->front().capture_time_ms_);
 218   }
 219   if (!low_priority_packets_->empty()) {
 220     oldest_packet_capture_time = std::min(
 221         oldest_packet_capture_time,
 222         low_priority_packets_->front().capture_time_ms_);
 223   }
 224   return now_ms - oldest_packet_capture_time;
 225 }
 226
 227 int32_t PacedSender::TimeUntilNextProcess() {
 228   CriticalSectionScoped cs(critsect_.get());
 229   int64_t elapsed_time_ms =
 230       (TickTime::Now() - time_last_update_).Milliseconds();
 231   if (elapsed_time_ms <= 0) {
 232     return kMinPacketLimitMs;
 233   }
 234   if (elapsed_time_ms >= kMinPacketLimitMs) {
 235     return 0;
 236   }
 237   return kMinPacketLimitMs - elapsed_time_ms;
 238 }
 239
 240 int32_t PacedSender::Process() {
 241   TickTime now = TickTime::Now();
 242   CriticalSectionScoped cs(critsect_.get());
 243   int elapsed_time_ms = (now - time_last_update_).Milliseconds();
 244   time_last_update_ = now;
 245   if (!enabled_) {
 246     return 0;
 247   }
 248   if (!paused_) {
 249     if (elapsed_time_ms > 0) {
 250       uint32_t delta_time_ms = std::min(kMaxIntervalTimeMs, elapsed_time_ms);
 251       UpdateBytesPerInterval(delta_time_ms);
 252     }
 253     uint32_t ssrc;
 254     uint16_t sequence_number;
 255     int64_t capture_time_ms;
 256     paced_sender::PacketList* packet_list;
 257     while (ShouldSendNextPacket(&packet_list)) {
 258       GetNextPacketFromList(packet_list, &ssrc, &sequence_number,
 259                             &capture_time_ms);
 260       critsect_->Leave();
 261
 262       const bool success = callback_->TimeToSendPacket(ssrc, sequence_number,
 263                                                        capture_time_ms);
 264       critsect_->Enter();
 265       // If packet cannot be sent then keep it in packet list and exit early.
 266       // There's no need to send more packets.
 267       if (!success) {
 268         return 0;
 269       }
 270       packet_list->pop_front();
 271       const bool last_packet = packet_list->empty() ||
 272           packet_list->front().capture_time_ms_ > capture_time_ms;
 273       if (packet_list != high_priority_packets_.get()) {
 274         if (capture_time_ms > capture_time_ms_last_sent_) {
 275           capture_time_ms_last_sent_ = capture_time_ms;
 276         } else if (capture_time_ms == capture_time_ms_last_sent_ &&
 277                    last_packet) {
 278           TRACE_EVENT_ASYNC_END0("webrtc_rtp", "PacedSend", capture_time_ms);
 279         }
 280       }
 281     }
 282     if (high_priority_packets_->empty() &&
 283         normal_priority_packets_->empty() &&
 284         low_priority_packets_->empty() &&
 285         padding_budget_->bytes_remaining() > 0 &&
 286         pad_up_to_bitrate_budget_->bytes_remaining() > 0) {
 287       int padding_needed = std::min(
 288           padding_budget_->bytes_remaining(),
 289           pad_up_to_bitrate_budget_->bytes_remaining());
 290       critsect_->Leave();
 291       int bytes_sent = callback_->TimeToSendPadding(padding_needed);
 292       critsect_->Enter();
 293       media_budget_->UseBudget(bytes_sent);
 294       padding_budget_->UseBudget(bytes_sent);
 295       pad_up_to_bitrate_budget_->UseBudget(bytes_sent);
 296     }
 297   }
 298   return 0;
 299 }
 300
 301 // MUST have critsect_ when calling.
 302 void PacedSender::UpdateBytesPerInterval(uint32_t delta_time_ms) {
 303   media_budget_->IncreaseBudget(delta_time_ms);
 304   padding_budget_->IncreaseBudget(delta_time_ms);
 305   pad_up_to_bitrate_budget_->IncreaseBudget(delta_time_ms);
 306 }
 307
 308 // MUST have critsect_ when calling.
 309 bool PacedSender::ShouldSendNextPacket(paced_sender::PacketList** packet_list) {
 310   if (media_budget_->bytes_remaining() <= 0) {
 311     // All bytes consumed for this interval.
 312     // Check if we have not sent in a too long time.
 313     if ((TickTime::Now() - time_last_send_).Milliseconds() >
 314         kMaxQueueTimeWithoutSendingMs) {
 315       if (!high_priority_packets_->empty()) {
 316         *packet_list = high_priority_packets_.get();
 317         return true;
 318       }
 319       if (!normal_priority_packets_->empty()) {
 320         *packet_list = normal_priority_packets_.get();
 321         return true;
 322       }
 323     }
 324     return false;
 325   }
 326   if (!high_priority_packets_->empty()) {
 327     *packet_list = high_priority_packets_.get();
 328     return true;
 329   }
 330   if (!normal_priority_packets_->empty()) {
 331     *packet_list = normal_priority_packets_.get();
 332     return true;
 333   }
 334   if (!low_priority_packets_->empty()) {
 335     *packet_list = low_priority_packets_.get();
 336     return true;
 337   }
 338   return false;
 339 }
 340
 341 void PacedSender::GetNextPacketFromList(paced_sender::PacketList* packets,
 342     uint32_t* ssrc, uint16_t* sequence_number, int64_t* capture_time_ms) {
 343   paced_sender::Packet packet = packets->front();
 344   UpdateMediaBytesSent(packet.bytes_);
 345   *sequence_number = packet.sequence_number_;
 346   *ssrc = packet.ssrc_;
 347   *capture_time_ms = packet.capture_time_ms_;
 348 }
 349
 350 // MUST have critsect_ when calling.
 351 void PacedSender::UpdateMediaBytesSent(int num_bytes) {
 352   time_last_send_ = TickTime::Now();
 353   media_budget_->UseBudget(num_bytes);
 354   pad_up_to_bitrate_budget_->UseBudget(num_bytes);
 355 }
 356
 357 }  // namespace webrtc