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