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/critical_section.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, 42 public RtpPacketSender, 43 private PacingController::PacketSender { 44 public: 45 TaskQueuePacedSender(Clock* clock, 46 PacketRouter* packet_router, 47 RtcEventLog* event_log, 48 const WebRtcKeyValueConfig* field_trials, 49 TaskQueueFactory* task_queue_factory); 50 51 ~TaskQueuePacedSender() override; 52 53 // Methods implementing RtpPacketSender. 54 55 // Adds the packet to the queue and calls PacketRouter::SendPacket() when 56 // it's time to send. 57 void EnqueuePackets( 58 std::vector<std::unique_ptr<RtpPacketToSend>> packets) override; 59 60 // Methods implementing RtpPacketPacer: 61 62 void CreateProbeCluster(DataRate bitrate, int cluster_id) override; 63 64 // Temporarily pause all sending. 65 void Pause() override; 66 67 // Resume sending packets. 68 void Resume() override; 69 70 void SetCongestionWindow(DataSize congestion_window_size) override; 71 void UpdateOutstandingData(DataSize outstanding_data) override; 72 73 // Sets the pacing rates. Must be called once before packets can be sent. 74 void SetPacingRates(DataRate pacing_rate, DataRate padding_rate) override; 75 76 // Currently audio traffic is not accounted for by pacer and passed through. 77 // With the introduction of audio BWE, audio traffic will be accounted for 78 // in the pacer budget calculation. The audio traffic will still be injected 79 // at high priority. 80 void SetAccountForAudioPackets(bool account_for_audio) override; 81 82 void SetIncludeOverhead() override; 83 void SetTransportOverhead(DataSize overhead_per_packet) override; 84 85 // Returns the time since the oldest queued packet was enqueued. 86 TimeDelta OldestPacketWaitTime() const override; 87 88 // Returns total size of all packets in the pacer queue. 89 DataSize QueueSizeData() const override; 90 91 // Returns the time when the first packet was sent; 92 absl::optional<Timestamp> FirstSentPacketTime() const override; 93 94 // Returns the number of milliseconds it will take to send the current 95 // packets in the queue, given the current size and bitrate, ignoring prio. 96 TimeDelta ExpectedQueueTime() const override; 97 98 // Set the max desired queuing delay, pacer will override the pacing rate 99 // specified by SetPacingRates() if needed to achieve this goal. 100 void SetQueueTimeLimit(TimeDelta limit) override; 101 102 private: 103 struct Stats { StatsStats104 Stats() 105 : oldest_packet_wait_time(TimeDelta::Zero()), 106 queue_size(DataSize::Zero()), 107 expected_queue_time(TimeDelta::Zero()) {} 108 TimeDelta oldest_packet_wait_time; 109 DataSize queue_size; 110 TimeDelta expected_queue_time; 111 absl::optional<Timestamp> first_sent_packet_time; 112 }; 113 114 // Check if it is time to send packets, or schedule a delayed task if not. 115 // Use Timestamp::MinusInfinity() to indicate that this call has _not_ 116 // been scheduled by the pacing controller. If this is the case, check if 117 // can execute immediately otherwise schedule a delay task that calls this 118 // method again with desired (finite) scheduled process time. 119 void MaybeProcessPackets(Timestamp scheduled_process_time); 120 121 // Methods implementing PacedSenderController:PacketSender. 122 123 void SendRtpPacket(std::unique_ptr<RtpPacketToSend> packet, 124 const PacedPacketInfo& cluster_info) override 125 RTC_RUN_ON(task_queue_); 126 127 std::vector<std::unique_ptr<RtpPacketToSend>> GeneratePadding( 128 DataSize size) override RTC_RUN_ON(task_queue_); 129 130 void MaybeUpdateStats(bool is_scheduled_call) RTC_RUN_ON(task_queue_); 131 Stats GetStats() const; 132 133 Clock* const clock_; 134 PacketRouter* const packet_router_ RTC_GUARDED_BY(task_queue_); 135 PacingController pacing_controller_ RTC_GUARDED_BY(task_queue_); 136 137 // We want only one (valid) delayed process task in flight at a time. 138 // If the value of |next_process_time_| is finite, it is an id for a 139 // delayed task that will call MaybeProcessPackets() with that time 140 // as parameter. 141 // Timestamp::MinusInfinity() indicates no valid pending task. 142 Timestamp next_process_time_ RTC_GUARDED_BY(task_queue_); 143 144 // Since we don't want to support synchronous calls that wait for a 145 // task execution, we poll the stats at some interval and update 146 // |current_stats_|, which can in turn be polled at any time. 147 148 // True iff there is delayed task in flight that that will call 149 // UdpateStats(). 150 bool stats_update_scheduled_ RTC_GUARDED_BY(task_queue_); 151 // Last time stats were updated. 152 Timestamp last_stats_time_ RTC_GUARDED_BY(task_queue_); 153 154 // Indicates if this task queue is shutting down. If so, don't allow 155 // posting any more delayed tasks as that can cause the task queue to 156 // never drain. 157 bool is_shutdown_ RTC_GUARDED_BY(task_queue_); 158 159 rtc::CriticalSection stats_crit_; 160 Stats current_stats_ RTC_GUARDED_BY(stats_crit_); 161 162 rtc::TaskQueue task_queue_; 163 }; 164 } // namespace webrtc 165 #endif // MODULES_PACING_TASK_QUEUE_PACED_SENDER_H_ 166