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/sequence_checker.h" 24 #include "api/task_queue/task_queue_factory.h" 25 #include "api/units/data_size.h" 26 #include "api/units/time_delta.h" 27 #include "api/units/timestamp.h" 28 #include "modules/include/module.h" 29 #include "modules/pacing/pacing_controller.h" 30 #include "modules/pacing/packet_router.h" 31 #include "modules/pacing/rtp_packet_pacer.h" 32 #include "modules/rtp_rtcp/source/rtp_packet_to_send.h" 33 #include "rtc_base/synchronization/mutex.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 // Ensure that necessary delayed tasks are scheduled. 59 void EnsureStarted(); 60 61 // Methods implementing RtpPacketSender. 62 63 // Adds the packet to the queue and calls PacketRouter::SendPacket() when 64 // it's time to send. 65 void EnqueuePackets( 66 std::vector<std::unique_ptr<RtpPacketToSend>> packets) override; 67 68 // Methods implementing RtpPacketPacer: 69 70 void CreateProbeCluster(DataRate bitrate, int cluster_id) override; 71 72 // Temporarily pause all sending. 73 void Pause() override; 74 75 // Resume sending packets. 76 void Resume() override; 77 78 void SetCongestionWindow(DataSize congestion_window_size) override; 79 void UpdateOutstandingData(DataSize outstanding_data) override; 80 81 // Sets the pacing rates. Must be called once before packets can be sent. 82 void SetPacingRates(DataRate pacing_rate, DataRate padding_rate) override; 83 84 // Currently audio traffic is not accounted for by pacer and passed through. 85 // With the introduction of audio BWE, audio traffic will be accounted for 86 // in the pacer budget calculation. The audio traffic will still be injected 87 // at high priority. 88 void SetAccountForAudioPackets(bool account_for_audio) override; 89 90 void SetIncludeOverhead() override; 91 void SetTransportOverhead(DataSize overhead_per_packet) override; 92 93 // Returns the time since the oldest queued packet was enqueued. 94 TimeDelta OldestPacketWaitTime() const override; 95 96 // Returns total size of all packets in the pacer queue. 97 DataSize QueueSizeData() const override; 98 99 // Returns the time when the first packet was sent; 100 absl::optional<Timestamp> FirstSentPacketTime() const override; 101 102 // Returns the number of milliseconds it will take to send the current 103 // packets in the queue, given the current size and bitrate, ignoring prio. 104 TimeDelta ExpectedQueueTime() const override; 105 106 // Set the max desired queuing delay, pacer will override the pacing rate 107 // specified by SetPacingRates() if needed to achieve this goal. 108 void SetQueueTimeLimit(TimeDelta limit) override; 109 110 protected: 111 // Exposed as protected for test. 112 struct Stats { StatsStats113 Stats() 114 : oldest_packet_wait_time(TimeDelta::Zero()), 115 queue_size(DataSize::Zero()), 116 expected_queue_time(TimeDelta::Zero()) {} 117 TimeDelta oldest_packet_wait_time; 118 DataSize queue_size; 119 TimeDelta expected_queue_time; 120 absl::optional<Timestamp> first_sent_packet_time; 121 }; 122 virtual void OnStatsUpdated(const Stats& stats); 123 124 private: 125 // Check if it is time to send packets, or schedule a delayed task if not. 126 // Use Timestamp::MinusInfinity() to indicate that this call has _not_ 127 // been scheduled by the pacing controller. If this is the case, check if 128 // can execute immediately otherwise schedule a delay task that calls this 129 // method again with desired (finite) scheduled process time. 130 void MaybeProcessPackets(Timestamp scheduled_process_time); 131 132 void MaybeUpdateStats(bool is_scheduled_call) RTC_RUN_ON(task_queue_); 133 Stats GetStats() const; 134 135 Clock* const clock_; 136 const TimeDelta hold_back_window_; 137 PacingController pacing_controller_ RTC_GUARDED_BY(task_queue_); 138 139 // We want only one (valid) delayed process task in flight at a time. 140 // If the value of |next_process_time_| is finite, it is an id for a 141 // delayed task that will call MaybeProcessPackets() with that time 142 // as parameter. 143 // Timestamp::MinusInfinity() indicates no valid pending task. 144 Timestamp next_process_time_ RTC_GUARDED_BY(task_queue_); 145 146 // Since we don't want to support synchronous calls that wait for a 147 // task execution, we poll the stats at some interval and update 148 // |current_stats_|, which can in turn be polled at any time. 149 150 // True iff there is delayed task in flight that that will call 151 // UdpateStats(). 152 bool stats_update_scheduled_ RTC_GUARDED_BY(task_queue_); 153 // Last time stats were updated. 154 Timestamp last_stats_time_ RTC_GUARDED_BY(task_queue_); 155 156 // Indicates if this task queue is started. If not, don't allow 157 // posting delayed tasks yet. 158 // TODO(crbug.com/1152887): Initialize to false once all users call 159 // EnsureStarted(). 160 bool is_started_ RTC_GUARDED_BY(task_queue_) = true; 161 162 // Indicates if this task queue is shutting down. If so, don't allow 163 // posting any more delayed tasks as that can cause the task queue to 164 // never drain. 165 bool is_shutdown_ RTC_GUARDED_BY(task_queue_); 166 167 mutable Mutex stats_mutex_; 168 Stats current_stats_ RTC_GUARDED_BY(stats_mutex_); 169 170 rtc::TaskQueue task_queue_; 171 }; 172 } // namespace webrtc 173 #endif // MODULES_PACING_TASK_QUEUE_PACED_SENDER_H_ 174