1 /*
2  *  Copyright (c) 2019 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 "modules/pacing/pacing_controller.h"
12 
13 #include <algorithm>
14 #include <memory>
15 #include <utility>
16 #include <vector>
17 
18 #include "absl/strings/match.h"
19 #include "modules/pacing/bitrate_prober.h"
20 #include "modules/pacing/interval_budget.h"
21 #include "modules/utility/include/process_thread.h"
22 #include "rtc_base/checks.h"
23 #include "rtc_base/logging.h"
24 #include "rtc_base/time_utils.h"
25 #include "system_wrappers/include/clock.h"
26 
27 namespace webrtc {
28 namespace {
29 // Time limit in milliseconds between packet bursts.
30 constexpr TimeDelta kDefaultMinPacketLimit = TimeDelta::Millis(5);
31 constexpr TimeDelta kCongestedPacketInterval = TimeDelta::Millis(500);
32 // TODO(sprang): Consider dropping this limit.
33 // The maximum debt level, in terms of time, capped when sending packets.
34 constexpr TimeDelta kMaxDebtInTime = TimeDelta::Millis(500);
35 constexpr TimeDelta kMaxElapsedTime = TimeDelta::Seconds(2);
36 constexpr DataSize kDefaultPaddingTarget = DataSize::Bytes(50);
37 
38 // Upper cap on process interval, in case process has not been called in a long
39 // time.
40 constexpr TimeDelta kMaxProcessingInterval = TimeDelta::Millis(30);
41 
42 constexpr int kFirstPriority = 0;
43 
IsDisabled(const WebRtcKeyValueConfig & field_trials,absl::string_view key)44 bool IsDisabled(const WebRtcKeyValueConfig& field_trials,
45                 absl::string_view key) {
46   return absl::StartsWith(field_trials.Lookup(key), "Disabled");
47 }
48 
IsEnabled(const WebRtcKeyValueConfig & field_trials,absl::string_view key)49 bool IsEnabled(const WebRtcKeyValueConfig& field_trials,
50                absl::string_view key) {
51   return absl::StartsWith(field_trials.Lookup(key), "Enabled");
52 }
53 
GetPriorityForType(RtpPacketMediaType type)54 int GetPriorityForType(RtpPacketMediaType type) {
55   // Lower number takes priority over higher.
56   switch (type) {
57     case RtpPacketMediaType::kAudio:
58       // Audio is always prioritized over other packet types.
59       return kFirstPriority + 1;
60     case RtpPacketMediaType::kRetransmission:
61       // Send retransmissions before new media.
62       return kFirstPriority + 2;
63     case RtpPacketMediaType::kVideo:
64     case RtpPacketMediaType::kForwardErrorCorrection:
65       // Video has "normal" priority, in the old speak.
66       // Send redundancy concurrently to video. If it is delayed it might have a
67       // lower chance of being useful.
68       return kFirstPriority + 3;
69     case RtpPacketMediaType::kPadding:
70       // Packets that are in themselves likely useless, only sent to keep the
71       // BWE high.
72       return kFirstPriority + 4;
73   }
74 }
75 
76 }  // namespace
77 
78 const TimeDelta PacingController::kMaxExpectedQueueLength =
79     TimeDelta::Millis(2000);
80 const float PacingController::kDefaultPaceMultiplier = 2.5f;
81 const TimeDelta PacingController::kPausedProcessInterval =
82     kCongestedPacketInterval;
83 const TimeDelta PacingController::kMinSleepTime = TimeDelta::Millis(1);
84 
PacingController(Clock * clock,PacketSender * packet_sender,RtcEventLog * event_log,const WebRtcKeyValueConfig * field_trials,ProcessMode mode)85 PacingController::PacingController(Clock* clock,
86                                    PacketSender* packet_sender,
87                                    RtcEventLog* event_log,
88                                    const WebRtcKeyValueConfig* field_trials,
89                                    ProcessMode mode)
90     : mode_(mode),
91       clock_(clock),
92       packet_sender_(packet_sender),
93       fallback_field_trials_(
94           !field_trials ? std::make_unique<FieldTrialBasedConfig>() : nullptr),
95       field_trials_(field_trials ? field_trials : fallback_field_trials_.get()),
96       drain_large_queues_(
97           !IsDisabled(*field_trials_, "WebRTC-Pacer-DrainQueue")),
98       send_padding_if_silent_(
99           IsEnabled(*field_trials_, "WebRTC-Pacer-PadInSilence")),
100       pace_audio_(IsEnabled(*field_trials_, "WebRTC-Pacer-BlockAudio")),
101       small_first_probe_packet_(
102           IsEnabled(*field_trials_, "WebRTC-Pacer-SmallFirstProbePacket")),
103       ignore_transport_overhead_(
104           IsEnabled(*field_trials_, "WebRTC-Pacer-IgnoreTransportOverhead")),
105       min_packet_limit_(kDefaultMinPacketLimit),
106       transport_overhead_per_packet_(DataSize::Zero()),
107       last_timestamp_(clock_->CurrentTime()),
108       paused_(false),
109       media_budget_(0),
110       padding_budget_(0),
111       media_debt_(DataSize::Zero()),
112       padding_debt_(DataSize::Zero()),
113       media_rate_(DataRate::Zero()),
114       padding_rate_(DataRate::Zero()),
115       prober_(*field_trials_),
116       probing_send_failure_(false),
117       pacing_bitrate_(DataRate::Zero()),
118       last_process_time_(clock->CurrentTime()),
119       last_send_time_(last_process_time_),
120       packet_queue_(last_process_time_, field_trials_),
121       packet_counter_(0),
122       congestion_window_size_(DataSize::PlusInfinity()),
123       outstanding_data_(DataSize::Zero()),
124       queue_time_limit(kMaxExpectedQueueLength),
125       account_for_audio_(false),
126       include_overhead_(false) {
127   if (!drain_large_queues_) {
128     RTC_LOG(LS_WARNING) << "Pacer queues will not be drained,"
129                            "pushback experiment must be enabled.";
130   }
131   FieldTrialParameter<int> min_packet_limit_ms("", min_packet_limit_.ms());
132   ParseFieldTrial({&min_packet_limit_ms},
133                   field_trials_->Lookup("WebRTC-Pacer-MinPacketLimitMs"));
134   min_packet_limit_ = TimeDelta::Millis(min_packet_limit_ms.Get());
135   UpdateBudgetWithElapsedTime(min_packet_limit_);
136 }
137 
138 PacingController::~PacingController() = default;
139 
CreateProbeCluster(DataRate bitrate,int cluster_id)140 void PacingController::CreateProbeCluster(DataRate bitrate, int cluster_id) {
141   prober_.CreateProbeCluster(bitrate, CurrentTime(), cluster_id);
142 }
143 
Pause()144 void PacingController::Pause() {
145   if (!paused_)
146     RTC_LOG(LS_INFO) << "PacedSender paused.";
147   paused_ = true;
148   packet_queue_.SetPauseState(true, CurrentTime());
149 }
150 
Resume()151 void PacingController::Resume() {
152   if (paused_)
153     RTC_LOG(LS_INFO) << "PacedSender resumed.";
154   paused_ = false;
155   packet_queue_.SetPauseState(false, CurrentTime());
156 }
157 
IsPaused() const158 bool PacingController::IsPaused() const {
159   return paused_;
160 }
161 
SetCongestionWindow(DataSize congestion_window_size)162 void PacingController::SetCongestionWindow(DataSize congestion_window_size) {
163   const bool was_congested = Congested();
164   congestion_window_size_ = congestion_window_size;
165   if (was_congested && !Congested()) {
166     TimeDelta elapsed_time = UpdateTimeAndGetElapsed(CurrentTime());
167     UpdateBudgetWithElapsedTime(elapsed_time);
168   }
169 }
170 
UpdateOutstandingData(DataSize outstanding_data)171 void PacingController::UpdateOutstandingData(DataSize outstanding_data) {
172   const bool was_congested = Congested();
173   outstanding_data_ = outstanding_data;
174   if (was_congested && !Congested()) {
175     TimeDelta elapsed_time = UpdateTimeAndGetElapsed(CurrentTime());
176     UpdateBudgetWithElapsedTime(elapsed_time);
177   }
178 }
179 
Congested() const180 bool PacingController::Congested() const {
181   if (congestion_window_size_.IsFinite()) {
182     return outstanding_data_ >= congestion_window_size_;
183   }
184   return false;
185 }
186 
CurrentTime() const187 Timestamp PacingController::CurrentTime() const {
188   Timestamp time = clock_->CurrentTime();
189   if (time < last_timestamp_) {
190     RTC_LOG(LS_WARNING)
191         << "Non-monotonic clock behavior observed. Previous timestamp: "
192         << last_timestamp_.ms() << ", new timestamp: " << time.ms();
193     RTC_DCHECK_GE(time, last_timestamp_);
194     time = last_timestamp_;
195   }
196   last_timestamp_ = time;
197   return time;
198 }
199 
SetProbingEnabled(bool enabled)200 void PacingController::SetProbingEnabled(bool enabled) {
201   RTC_CHECK_EQ(0, packet_counter_);
202   prober_.SetEnabled(enabled);
203 }
204 
SetPacingRates(DataRate pacing_rate,DataRate padding_rate)205 void PacingController::SetPacingRates(DataRate pacing_rate,
206                                       DataRate padding_rate) {
207   RTC_DCHECK_GT(pacing_rate, DataRate::Zero());
208   media_rate_ = pacing_rate;
209   padding_rate_ = padding_rate;
210   pacing_bitrate_ = pacing_rate;
211   padding_budget_.set_target_rate_kbps(padding_rate.kbps());
212 
213   RTC_LOG(LS_VERBOSE) << "bwe:pacer_updated pacing_kbps="
214                       << pacing_bitrate_.kbps()
215                       << " padding_budget_kbps=" << padding_rate.kbps();
216 }
217 
EnqueuePacket(std::unique_ptr<RtpPacketToSend> packet)218 void PacingController::EnqueuePacket(std::unique_ptr<RtpPacketToSend> packet) {
219   RTC_DCHECK(pacing_bitrate_ > DataRate::Zero())
220       << "SetPacingRate must be called before InsertPacket.";
221   RTC_CHECK(packet->packet_type());
222   // Get priority first and store in temporary, to avoid chance of object being
223   // moved before GetPriorityForType() being called.
224   const int priority = GetPriorityForType(*packet->packet_type());
225   EnqueuePacketInternal(std::move(packet), priority);
226 }
227 
SetAccountForAudioPackets(bool account_for_audio)228 void PacingController::SetAccountForAudioPackets(bool account_for_audio) {
229   account_for_audio_ = account_for_audio;
230 }
231 
SetIncludeOverhead()232 void PacingController::SetIncludeOverhead() {
233   include_overhead_ = true;
234   packet_queue_.SetIncludeOverhead();
235 }
236 
SetTransportOverhead(DataSize overhead_per_packet)237 void PacingController::SetTransportOverhead(DataSize overhead_per_packet) {
238   if (ignore_transport_overhead_)
239     return;
240   transport_overhead_per_packet_ = overhead_per_packet;
241   packet_queue_.SetTransportOverhead(overhead_per_packet);
242 }
243 
ExpectedQueueTime() const244 TimeDelta PacingController::ExpectedQueueTime() const {
245   RTC_DCHECK_GT(pacing_bitrate_, DataRate::Zero());
246   return TimeDelta::Millis(
247       (QueueSizeData().bytes() * 8 * rtc::kNumMillisecsPerSec) /
248       pacing_bitrate_.bps());
249 }
250 
QueueSizePackets() const251 size_t PacingController::QueueSizePackets() const {
252   return packet_queue_.SizeInPackets();
253 }
254 
QueueSizeData() const255 DataSize PacingController::QueueSizeData() const {
256   return packet_queue_.Size();
257 }
258 
CurrentBufferLevel() const259 DataSize PacingController::CurrentBufferLevel() const {
260   return std::max(media_debt_, padding_debt_);
261 }
262 
FirstSentPacketTime() const263 absl::optional<Timestamp> PacingController::FirstSentPacketTime() const {
264   return first_sent_packet_time_;
265 }
266 
OldestPacketWaitTime() const267 TimeDelta PacingController::OldestPacketWaitTime() const {
268   Timestamp oldest_packet = packet_queue_.OldestEnqueueTime();
269   if (oldest_packet.IsInfinite()) {
270     return TimeDelta::Zero();
271   }
272 
273   return CurrentTime() - oldest_packet;
274 }
275 
EnqueuePacketInternal(std::unique_ptr<RtpPacketToSend> packet,int priority)276 void PacingController::EnqueuePacketInternal(
277     std::unique_ptr<RtpPacketToSend> packet,
278     int priority) {
279   prober_.OnIncomingPacket(packet->payload_size());
280 
281   // TODO(sprang): Make sure tests respect this, replace with DCHECK.
282   Timestamp now = CurrentTime();
283   if (packet->capture_time_ms() < 0) {
284     packet->set_capture_time_ms(now.ms());
285   }
286 
287   if (mode_ == ProcessMode::kDynamic && packet_queue_.Empty() &&
288       media_debt_.IsZero()) {
289     TimeDelta elapsed_time = UpdateTimeAndGetElapsed(now);
290     UpdateBudgetWithElapsedTime(elapsed_time);
291   }
292   packet_queue_.Push(priority, now, packet_counter_++, std::move(packet));
293 }
294 
UpdateTimeAndGetElapsed(Timestamp now)295 TimeDelta PacingController::UpdateTimeAndGetElapsed(Timestamp now) {
296   if (last_process_time_.IsMinusInfinity()) {
297     return TimeDelta::Zero();
298   }
299   RTC_DCHECK_GE(now, last_process_time_);
300   TimeDelta elapsed_time = now - last_process_time_;
301   last_process_time_ = now;
302   if (elapsed_time > kMaxElapsedTime) {
303     RTC_LOG(LS_WARNING) << "Elapsed time (" << elapsed_time.ms()
304                         << " ms) longer than expected, limiting to "
305                         << kMaxElapsedTime.ms();
306     elapsed_time = kMaxElapsedTime;
307   }
308   return elapsed_time;
309 }
310 
ShouldSendKeepalive(Timestamp now) const311 bool PacingController::ShouldSendKeepalive(Timestamp now) const {
312   if (send_padding_if_silent_ || paused_ || Congested() ||
313       packet_counter_ == 0) {
314     // We send a padding packet every 500 ms to ensure we won't get stuck in
315     // congested state due to no feedback being received.
316     TimeDelta elapsed_since_last_send = now - last_send_time_;
317     if (elapsed_since_last_send >= kCongestedPacketInterval) {
318       return true;
319     }
320   }
321   return false;
322 }
323 
NextSendTime() const324 Timestamp PacingController::NextSendTime() const {
325   Timestamp now = CurrentTime();
326 
327   if (paused_) {
328     return last_send_time_ + kPausedProcessInterval;
329   }
330 
331   // If probing is active, that always takes priority.
332   if (prober_.is_probing()) {
333     Timestamp probe_time = prober_.NextProbeTime(now);
334     // |probe_time| == PlusInfinity indicates no probe scheduled.
335     if (probe_time != Timestamp::PlusInfinity() && !probing_send_failure_) {
336       return probe_time;
337     }
338   }
339 
340   if (mode_ == ProcessMode::kPeriodic) {
341     // In periodic non-probing mode, we just have a fixed interval.
342     return last_process_time_ + min_packet_limit_;
343   }
344 
345   // In dynamic mode, figure out when the next packet should be sent,
346   // given the current conditions.
347 
348   if (!pace_audio_ && packet_queue_.NextPacketIsAudio()) {
349     return now;
350   }
351 
352   if (Congested() || packet_counter_ == 0) {
353     // We need to at least send keep-alive packets with some interval.
354     return last_send_time_ + kCongestedPacketInterval;
355   }
356 
357   // Check how long until media buffer has drained. We schedule a call
358   // for when the last packet in the queue drains as otherwise we may
359   // be late in starting padding.
360   if (media_rate_ > DataRate::Zero() &&
361       (!packet_queue_.Empty() || !media_debt_.IsZero())) {
362     return std::min(last_send_time_ + kPausedProcessInterval,
363                     last_process_time_ + media_debt_ / media_rate_);
364   }
365 
366   // If we _don't_ have pending packets, check how long until we have
367   // bandwidth for padding packets.
368   if (padding_rate_ > DataRate::Zero() && packet_queue_.Empty()) {
369     return std::min(last_send_time_ + kPausedProcessInterval,
370                     last_process_time_ + padding_debt_ / padding_rate_);
371   }
372 
373   if (send_padding_if_silent_) {
374     return last_send_time_ + kPausedProcessInterval;
375   }
376   return last_process_time_ + kPausedProcessInterval;
377 }
378 
ProcessPackets()379 void PacingController::ProcessPackets() {
380   Timestamp now = CurrentTime();
381   Timestamp target_send_time = now;
382   if (mode_ == ProcessMode::kDynamic) {
383     target_send_time = NextSendTime();
384     if (target_send_time.IsMinusInfinity()) {
385       target_send_time = now;
386     } else if (now < target_send_time) {
387       // We are too early, abort and regroup!
388       return;
389     }
390 
391     if (target_send_time < last_process_time_) {
392       // After the last process call, at time X, the target send time
393       // shifted to be earlier than X. This should normally not happen
394       // but we want to make sure rounding errors or erratic behavior
395       // of NextSendTime() does not cause issue. In particular, if the
396       // buffer reduction of
397       // rate * (target_send_time - previous_process_time)
398       // in the main loop doesn't clean up the existing debt we may not
399       // be able to send again. We don't want to check this reordering
400       // there as it is the normal exit condtion when the buffer is
401       // exhausted and there are packets in the queue.
402       UpdateBudgetWithElapsedTime(last_process_time_ - target_send_time);
403       target_send_time = last_process_time_;
404     }
405   }
406 
407   Timestamp previous_process_time = last_process_time_;
408   TimeDelta elapsed_time = UpdateTimeAndGetElapsed(now);
409 
410   if (ShouldSendKeepalive(now)) {
411     // We can not send padding unless a normal packet has first been sent. If
412     // we do, timestamps get messed up.
413     if (packet_counter_ == 0) {
414       last_send_time_ = now;
415     } else {
416       DataSize keepalive_data_sent = DataSize::Zero();
417       std::vector<std::unique_ptr<RtpPacketToSend>> keepalive_packets =
418           packet_sender_->GeneratePadding(DataSize::Bytes(1));
419       for (auto& packet : keepalive_packets) {
420         keepalive_data_sent +=
421             DataSize::Bytes(packet->payload_size() + packet->padding_size());
422         packet_sender_->SendRtpPacket(std::move(packet), PacedPacketInfo());
423       }
424       OnPaddingSent(keepalive_data_sent);
425     }
426   }
427 
428   if (paused_) {
429     return;
430   }
431 
432   if (elapsed_time > TimeDelta::Zero()) {
433     DataRate target_rate = pacing_bitrate_;
434     DataSize queue_size_data = packet_queue_.Size();
435     if (queue_size_data > DataSize::Zero()) {
436       // Assuming equal size packets and input/output rate, the average packet
437       // has avg_time_left_ms left to get queue_size_bytes out of the queue, if
438       // time constraint shall be met. Determine bitrate needed for that.
439       packet_queue_.UpdateQueueTime(now);
440       if (drain_large_queues_) {
441         TimeDelta avg_time_left =
442             std::max(TimeDelta::Millis(1),
443                      queue_time_limit - packet_queue_.AverageQueueTime());
444         DataRate min_rate_needed = queue_size_data / avg_time_left;
445         if (min_rate_needed > target_rate) {
446           target_rate = min_rate_needed;
447           RTC_LOG(LS_VERBOSE) << "bwe:large_pacing_queue pacing_rate_kbps="
448                               << target_rate.kbps();
449         }
450       }
451     }
452 
453     if (mode_ == ProcessMode::kPeriodic) {
454       // In periodic processing mode, the IntevalBudget allows positive budget
455       // up to (process interval duration) * (target rate), so we only need to
456       // update it once before the packet sending loop.
457       media_budget_.set_target_rate_kbps(target_rate.kbps());
458       UpdateBudgetWithElapsedTime(elapsed_time);
459     } else {
460       media_rate_ = target_rate;
461     }
462   }
463 
464   bool first_packet_in_probe = false;
465   bool is_probing = prober_.is_probing();
466   PacedPacketInfo pacing_info;
467   absl::optional<DataSize> recommended_probe_size;
468   if (is_probing) {
469     pacing_info = prober_.CurrentCluster();
470     first_packet_in_probe = pacing_info.probe_cluster_bytes_sent == 0;
471     recommended_probe_size = DataSize::Bytes(prober_.RecommendedMinProbeSize());
472   }
473 
474   DataSize data_sent = DataSize::Zero();
475 
476   // The paused state is checked in the loop since it leaves the critical
477   // section allowing the paused state to be changed from other code.
478   while (!paused_) {
479     if (small_first_probe_packet_ && first_packet_in_probe) {
480       // If first packet in probe, insert a small padding packet so we have a
481       // more reliable start window for the rate estimation.
482       auto padding = packet_sender_->GeneratePadding(DataSize::Bytes(1));
483       // If no RTP modules sending media are registered, we may not get a
484       // padding packet back.
485       if (!padding.empty()) {
486         // Insert with high priority so larger media packets don't preempt it.
487         EnqueuePacketInternal(std::move(padding[0]), kFirstPriority);
488         // We should never get more than one padding packets with a requested
489         // size of 1 byte.
490         RTC_DCHECK_EQ(padding.size(), 1u);
491       }
492       first_packet_in_probe = false;
493     }
494 
495     if (mode_ == ProcessMode::kDynamic &&
496         previous_process_time < target_send_time) {
497       // Reduce buffer levels with amount corresponding to time between last
498       // process and target send time for the next packet.
499       // If the process call is late, that may be the time between the optimal
500       // send times for two packets we should already have sent.
501       UpdateBudgetWithElapsedTime(target_send_time - previous_process_time);
502       previous_process_time = target_send_time;
503     }
504 
505     // Fetch the next packet, so long as queue is not empty or budget is not
506     // exhausted.
507     std::unique_ptr<RtpPacketToSend> rtp_packet =
508         GetPendingPacket(pacing_info, target_send_time, now);
509 
510     if (rtp_packet == nullptr) {
511       // No packet available to send, check if we should send padding.
512       DataSize padding_to_add = PaddingToAdd(recommended_probe_size, data_sent);
513       if (padding_to_add > DataSize::Zero()) {
514         std::vector<std::unique_ptr<RtpPacketToSend>> padding_packets =
515             packet_sender_->GeneratePadding(padding_to_add);
516         if (padding_packets.empty()) {
517           // No padding packets were generated, quite send loop.
518           break;
519         }
520         for (auto& packet : padding_packets) {
521           EnqueuePacket(std::move(packet));
522         }
523         // Continue loop to send the padding that was just added.
524         continue;
525       }
526 
527       // Can't fetch new packet and no padding to send, exit send loop.
528       break;
529     }
530 
531     RTC_DCHECK(rtp_packet);
532     RTC_DCHECK(rtp_packet->packet_type().has_value());
533     const RtpPacketMediaType packet_type = *rtp_packet->packet_type();
534     DataSize packet_size = DataSize::Bytes(rtp_packet->payload_size() +
535                                            rtp_packet->padding_size());
536 
537     if (include_overhead_) {
538       packet_size += DataSize::Bytes(rtp_packet->headers_size()) +
539                      transport_overhead_per_packet_;
540     }
541     packet_sender_->SendRtpPacket(std::move(rtp_packet), pacing_info);
542 
543     data_sent += packet_size;
544 
545     // Send done, update send/process time to the target send time.
546     OnPacketSent(packet_type, packet_size, target_send_time);
547     if (recommended_probe_size && data_sent > *recommended_probe_size)
548       break;
549 
550     if (mode_ == ProcessMode::kDynamic) {
551       // Update target send time in case that are more packets that we are late
552       // in processing.
553       Timestamp next_send_time = NextSendTime();
554       if (next_send_time.IsMinusInfinity()) {
555         target_send_time = now;
556       } else {
557         target_send_time = std::min(now, next_send_time);
558       }
559     }
560   }
561 
562   if (is_probing) {
563     probing_send_failure_ = data_sent == DataSize::Zero();
564     if (!probing_send_failure_) {
565       prober_.ProbeSent(CurrentTime(), data_sent.bytes());
566     }
567   }
568 }
569 
PaddingToAdd(absl::optional<DataSize> recommended_probe_size,DataSize data_sent) const570 DataSize PacingController::PaddingToAdd(
571     absl::optional<DataSize> recommended_probe_size,
572     DataSize data_sent) const {
573   if (!packet_queue_.Empty()) {
574     // Actual payload available, no need to add padding.
575     return DataSize::Zero();
576   }
577 
578   if (Congested()) {
579     // Don't add padding if congested, even if requested for probing.
580     return DataSize::Zero();
581   }
582 
583   if (packet_counter_ == 0) {
584     // We can not send padding unless a normal packet has first been sent. If we
585     // do, timestamps get messed up.
586     return DataSize::Zero();
587   }
588 
589   if (recommended_probe_size) {
590     if (*recommended_probe_size > data_sent) {
591       return *recommended_probe_size - data_sent;
592     }
593     return DataSize::Zero();
594   }
595 
596   if (mode_ == ProcessMode::kPeriodic) {
597     return DataSize::Bytes(padding_budget_.bytes_remaining());
598   } else if (padding_rate_ > DataRate::Zero() &&
599              padding_debt_ == DataSize::Zero()) {
600     return kDefaultPaddingTarget;
601   }
602   return DataSize::Zero();
603 }
604 
GetPendingPacket(const PacedPacketInfo & pacing_info,Timestamp target_send_time,Timestamp now)605 std::unique_ptr<RtpPacketToSend> PacingController::GetPendingPacket(
606     const PacedPacketInfo& pacing_info,
607     Timestamp target_send_time,
608     Timestamp now) {
609   if (packet_queue_.Empty()) {
610     return nullptr;
611   }
612 
613   // First, check if there is any reason _not_ to send the next queued packet.
614 
615   // Unpaced audio packets and probes are exempted from send checks.
616   bool unpaced_audio_packet = !pace_audio_ && packet_queue_.NextPacketIsAudio();
617   bool is_probe = pacing_info.probe_cluster_id != PacedPacketInfo::kNotAProbe;
618   if (!unpaced_audio_packet && !is_probe) {
619     if (Congested()) {
620       // Don't send anything if congested.
621       return nullptr;
622     }
623 
624     if (mode_ == ProcessMode::kPeriodic) {
625       if (media_budget_.bytes_remaining() <= 0) {
626         // Not enough budget.
627         return nullptr;
628       }
629     } else {
630       // Dynamic processing mode.
631       if (now <= target_send_time) {
632         // We allow sending slightly early if we think that we would actually
633         // had been able to, had we been right on time - i.e. the current debt
634         // is not more than would be reduced to zero at the target sent time.
635         TimeDelta flush_time = media_debt_ / media_rate_;
636         if (now + flush_time > target_send_time) {
637           return nullptr;
638         }
639       }
640     }
641   }
642 
643   return packet_queue_.Pop();
644 }
645 
OnPacketSent(RtpPacketMediaType packet_type,DataSize packet_size,Timestamp send_time)646 void PacingController::OnPacketSent(RtpPacketMediaType packet_type,
647                                     DataSize packet_size,
648                                     Timestamp send_time) {
649   if (!first_sent_packet_time_) {
650     first_sent_packet_time_ = send_time;
651   }
652   bool audio_packet = packet_type == RtpPacketMediaType::kAudio;
653   if (!audio_packet || account_for_audio_) {
654     // Update media bytes sent.
655     UpdateBudgetWithSentData(packet_size);
656   }
657   last_send_time_ = send_time;
658   last_process_time_ = send_time;
659 }
660 
OnPaddingSent(DataSize data_sent)661 void PacingController::OnPaddingSent(DataSize data_sent) {
662   if (data_sent > DataSize::Zero()) {
663     UpdateBudgetWithSentData(data_sent);
664   }
665   last_send_time_ = CurrentTime();
666   last_process_time_ = CurrentTime();
667 }
668 
UpdateBudgetWithElapsedTime(TimeDelta delta)669 void PacingController::UpdateBudgetWithElapsedTime(TimeDelta delta) {
670   if (mode_ == ProcessMode::kPeriodic) {
671     delta = std::min(kMaxProcessingInterval, delta);
672     media_budget_.IncreaseBudget(delta.ms());
673     padding_budget_.IncreaseBudget(delta.ms());
674   } else {
675     media_debt_ -= std::min(media_debt_, media_rate_ * delta);
676     padding_debt_ -= std::min(padding_debt_, padding_rate_ * delta);
677   }
678 }
679 
UpdateBudgetWithSentData(DataSize size)680 void PacingController::UpdateBudgetWithSentData(DataSize size) {
681   outstanding_data_ += size;
682   if (mode_ == ProcessMode::kPeriodic) {
683     media_budget_.UseBudget(size.bytes());
684     padding_budget_.UseBudget(size.bytes());
685   } else {
686     media_debt_ += size;
687     media_debt_ = std::min(media_debt_, media_rate_ * kMaxDebtInTime);
688     padding_debt_ += size;
689     padding_debt_ = std::min(padding_debt_, padding_rate_ * kMaxDebtInTime);
690   }
691 }
692 
SetQueueTimeLimit(TimeDelta limit)693 void PacingController::SetQueueTimeLimit(TimeDelta limit) {
694   queue_time_limit = limit;
695 }
696 
697 }  // namespace webrtc
698