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