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 #ifndef MODULES_PACING_TASK_QUEUE_PACED_SENDER_H_ 12 #define MODULES_PACING_TASK_QUEUE_PACED_SENDER_H_ 13 14 #include <stddef.h> 15 #include <stdint.h> 16 17 #include <functional> 18 #include <memory> 19 #include <queue> 20 #include <vector> 21 22 #include "absl/types/optional.h" 23 #include "api/task_queue/task_queue_factory.h" 24 #include "api/units/data_size.h" 25 #include "api/units/time_delta.h" 26 #include "api/units/timestamp.h" 27 #include "modules/include/module.h" 28 #include "modules/pacing/pacing_controller.h" 29 #include "modules/pacing/packet_router.h" 30 #include "modules/pacing/rtp_packet_pacer.h" 31 #include "modules/rtp_rtcp/source/rtp_packet_to_send.h" 32 #include "rtc_base/synchronization/mutex.h" 33 #include "rtc_base/synchronization/sequence_checker.h" 34 #include "rtc_base/task_queue.h" 35 #include "rtc_base/thread_annotations.h" 36 37 namespace webrtc { 38 class Clock; 39 class RtcEventLog; 40 41 class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender { 42 public: 43 // The |hold_back_window| parameter sets a lower bound on time to sleep if 44 // there is currently a pacer queue and packets can't immediately be 45 // processed. Increasing this reduces thread wakeups at the expense of higher 46 // latency. 47 // TODO(bugs.webrtc.org/10809): Remove default value for hold_back_window. 48 TaskQueuePacedSender( 49 Clock* clock, 50 PacketRouter* packet_router, 51 RtcEventLog* event_log, 52 const WebRtcKeyValueConfig* field_trials, 53 TaskQueueFactory* task_queue_factory, 54 TimeDelta hold_back_window = PacingController::kMinSleepTime); 55 56 ~TaskQueuePacedSender() override; 57 58 // Methods implementing RtpPacketSender. 59 60 // Adds the packet to the queue and calls PacketRouter::SendPacket() when 61 // it's time to send. 62 void EnqueuePackets( 63 std::vector<std::unique_ptr<RtpPacketToSend>> packets) override; 64 65 // Methods implementing RtpPacketPacer: 66 67 void CreateProbeCluster(DataRate bitrate, int cluster_id) override; 68 69 // Temporarily pause all sending. 70 void Pause() override; 71 72 // Resume sending packets. 73 void Resume() override; 74 75 void SetCongestionWindow(DataSize congestion_window_size) override; 76 void UpdateOutstandingData(DataSize outstanding_data) override; 77 78 // Sets the pacing rates. Must be called once before packets can be sent. 79 void SetPacingRates(DataRate pacing_rate, DataRate padding_rate) override; 80 81 // Currently audio traffic is not accounted for by pacer and passed through. 82 // With the introduction of audio BWE, audio traffic will be accounted for 83 // in the pacer budget calculation. The audio traffic will still be injected 84 // at high priority. 85 void SetAccountForAudioPackets(bool account_for_audio) override; 86 87 void SetIncludeOverhead() override; 88 void SetTransportOverhead(DataSize overhead_per_packet) override; 89 90 // Returns the time since the oldest queued packet was enqueued. 91 TimeDelta OldestPacketWaitTime() const override; 92 93 // Returns total size of all packets in the pacer queue. 94 DataSize QueueSizeData() const override; 95 96 // Returns the time when the first packet was sent; 97 absl::optional<Timestamp> FirstSentPacketTime() const override; 98 99 // Returns the number of milliseconds it will take to send the current 100 // packets in the queue, given the current size and bitrate, ignoring prio. 101 TimeDelta ExpectedQueueTime() const override; 102 103 // Set the max desired queuing delay, pacer will override the pacing rate 104 // specified by SetPacingRates() if needed to achieve this goal. 105 void SetQueueTimeLimit(TimeDelta limit) override; 106 107 protected: 108 // Exposed as protected for test. 109 struct Stats { StatsStats110 Stats() 111 : oldest_packet_wait_time(TimeDelta::Zero()), 112 queue_size(DataSize::Zero()), 113 expected_queue_time(TimeDelta::Zero()) {} 114 TimeDelta oldest_packet_wait_time; 115 DataSize queue_size; 116 TimeDelta expected_queue_time; 117 absl::optional<Timestamp> first_sent_packet_time; 118 }; 119 virtual void OnStatsUpdated(const Stats& stats); 120 121 private: 122 // Check if it is time to send packets, or schedule a delayed task if not. 123 // Use Timestamp::MinusInfinity() to indicate that this call has _not_ 124 // been scheduled by the pacing controller. If this is the case, check if 125 // can execute immediately otherwise schedule a delay task that calls this 126 // method again with desired (finite) scheduled process time. 127 void MaybeProcessPackets(Timestamp scheduled_process_time); 128 129 void MaybeUpdateStats(bool is_scheduled_call) RTC_RUN_ON(task_queue_); 130 Stats GetStats() const; 131 132 Clock* const clock_; 133 const TimeDelta hold_back_window_; 134 PacingController pacing_controller_ RTC_GUARDED_BY(task_queue_); 135 136 // We want only one (valid) delayed process task in flight at a time. 137 // If the value of |next_process_time_| is finite, it is an id for a 138 // delayed task that will call MaybeProcessPackets() with that time 139 // as parameter. 140 // Timestamp::MinusInfinity() indicates no valid pending task. 141 Timestamp next_process_time_ RTC_GUARDED_BY(task_queue_); 142 143 // Since we don't want to support synchronous calls that wait for a 144 // task execution, we poll the stats at some interval and update 145 // |current_stats_|, which can in turn be polled at any time. 146 147 // True iff there is delayed task in flight that that will call 148 // UdpateStats(). 149 bool stats_update_scheduled_ RTC_GUARDED_BY(task_queue_); 150 // Last time stats were updated. 151 Timestamp last_stats_time_ RTC_GUARDED_BY(task_queue_); 152 153 // Indicates if this task queue is shutting down. If so, don't allow 154 // posting any more delayed tasks as that can cause the task queue to 155 // never drain. 156 bool is_shutdown_ RTC_GUARDED_BY(task_queue_); 157 158 mutable Mutex stats_mutex_; 159 Stats current_stats_ RTC_GUARDED_BY(stats_mutex_); 160 161 rtc::TaskQueue task_queue_; 162 }; 163 } // namespace webrtc 164 #endif // MODULES_PACING_TASK_QUEUE_PACED_SENDER_H_ 165