1 /*
2 * Copyright 2018 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 #include "rtc_base/task_queue_stdlib.h"
12
13 #include <string.h>
14
15 #include <algorithm>
16 #include <map>
17 #include <memory>
18 #include <queue>
19 #include <utility>
20
21 #include "absl/strings/string_view.h"
22 #include "api/task_queue/queued_task.h"
23 #include "api/task_queue/task_queue_base.h"
24 #include "rtc_base/checks.h"
25 #include "rtc_base/event.h"
26 #include "rtc_base/logging.h"
27 #include "rtc_base/platform_thread.h"
28 #include "rtc_base/synchronization/mutex.h"
29 #include "rtc_base/thread_annotations.h"
30 #include "rtc_base/time_utils.h"
31
32 namespace webrtc {
33 namespace {
34
TaskQueuePriorityToThreadPriority(TaskQueueFactory::Priority priority)35 rtc::ThreadPriority TaskQueuePriorityToThreadPriority(
36 TaskQueueFactory::Priority priority) {
37 switch (priority) {
38 case TaskQueueFactory::Priority::HIGH:
39 return rtc::kRealtimePriority;
40 case TaskQueueFactory::Priority::LOW:
41 return rtc::kLowPriority;
42 case TaskQueueFactory::Priority::NORMAL:
43 return rtc::kNormalPriority;
44 default:
45 RTC_NOTREACHED();
46 return rtc::kNormalPriority;
47 }
48 }
49
50 class TaskQueueStdlib final : public TaskQueueBase {
51 public:
52 TaskQueueStdlib(absl::string_view queue_name, rtc::ThreadPriority priority);
53 ~TaskQueueStdlib() override = default;
54
55 void Delete() override;
56 void PostTask(std::unique_ptr<QueuedTask> task) override;
57 void PostDelayedTask(std::unique_ptr<QueuedTask> task,
58 uint32_t milliseconds) override;
59
60 private:
61 using OrderId = uint64_t;
62
63 struct DelayedEntryTimeout {
64 int64_t next_fire_at_ms_{};
65 OrderId order_{};
66
operator <webrtc::__anon09e449650111::TaskQueueStdlib::DelayedEntryTimeout67 bool operator<(const DelayedEntryTimeout& o) const {
68 return std::tie(next_fire_at_ms_, order_) <
69 std::tie(o.next_fire_at_ms_, o.order_);
70 }
71 };
72
73 struct NextTask {
74 bool final_task_{false};
75 std::unique_ptr<QueuedTask> run_task_;
76 int64_t sleep_time_ms_{};
77 };
78
79 NextTask GetNextTask();
80
81 static void ThreadMain(void* context);
82
83 void ProcessTasks();
84
85 void NotifyWake();
86
87 // Indicates if the thread has started.
88 rtc::Event started_;
89
90 // Indicates if the thread has stopped.
91 rtc::Event stopped_;
92
93 // Signaled whenever a new task is pending.
94 rtc::Event flag_notify_;
95
96 // Contains the active worker thread assigned to processing
97 // tasks (including delayed tasks).
98 rtc::PlatformThread thread_;
99
100 Mutex pending_lock_;
101
102 // Indicates if the worker thread needs to shutdown now.
RTC_GUARDED_BY(pending_lock_)103 bool thread_should_quit_ RTC_GUARDED_BY(pending_lock_){false};
104
105 // Holds the next order to use for the next task to be
106 // put into one of the pending queues.
RTC_GUARDED_BY(pending_lock_)107 OrderId thread_posting_order_ RTC_GUARDED_BY(pending_lock_){};
108
109 // The list of all pending tasks that need to be processed in the
110 // FIFO queue ordering on the worker thread.
111 std::queue<std::pair<OrderId, std::unique_ptr<QueuedTask>>> pending_queue_
112 RTC_GUARDED_BY(pending_lock_);
113
114 // The list of all pending tasks that need to be processed at a future
115 // time based upon a delay. On the off change the delayed task should
116 // happen at exactly the same time interval as another task then the
117 // task is processed based on FIFO ordering. std::priority_queue was
118 // considered but rejected due to its inability to extract the
119 // std::unique_ptr out of the queue without the presence of a hack.
120 std::map<DelayedEntryTimeout, std::unique_ptr<QueuedTask>> delayed_queue_
121 RTC_GUARDED_BY(pending_lock_);
122 };
123
TaskQueueStdlib(absl::string_view queue_name,rtc::ThreadPriority priority)124 TaskQueueStdlib::TaskQueueStdlib(absl::string_view queue_name,
125 rtc::ThreadPriority priority)
126 : started_(/*manual_reset=*/false, /*initially_signaled=*/false),
127 stopped_(/*manual_reset=*/false, /*initially_signaled=*/false),
128 flag_notify_(/*manual_reset=*/false, /*initially_signaled=*/false),
129 thread_(&TaskQueueStdlib::ThreadMain, this, queue_name, priority) {
130 thread_.Start();
131 started_.Wait(rtc::Event::kForever);
132 }
133
Delete()134 void TaskQueueStdlib::Delete() {
135 RTC_DCHECK(!IsCurrent());
136
137 {
138 MutexLock lock(&pending_lock_);
139 thread_should_quit_ = true;
140 }
141
142 NotifyWake();
143
144 stopped_.Wait(rtc::Event::kForever);
145 thread_.Stop();
146 delete this;
147 }
148
PostTask(std::unique_ptr<QueuedTask> task)149 void TaskQueueStdlib::PostTask(std::unique_ptr<QueuedTask> task) {
150 {
151 MutexLock lock(&pending_lock_);
152 OrderId order = thread_posting_order_++;
153
154 pending_queue_.push(std::pair<OrderId, std::unique_ptr<QueuedTask>>(
155 order, std::move(task)));
156 }
157
158 NotifyWake();
159 }
160
PostDelayedTask(std::unique_ptr<QueuedTask> task,uint32_t milliseconds)161 void TaskQueueStdlib::PostDelayedTask(std::unique_ptr<QueuedTask> task,
162 uint32_t milliseconds) {
163 auto fire_at = rtc::TimeMillis() + milliseconds;
164
165 DelayedEntryTimeout delay;
166 delay.next_fire_at_ms_ = fire_at;
167
168 {
169 MutexLock lock(&pending_lock_);
170 delay.order_ = ++thread_posting_order_;
171 delayed_queue_[delay] = std::move(task);
172 }
173
174 NotifyWake();
175 }
176
GetNextTask()177 TaskQueueStdlib::NextTask TaskQueueStdlib::GetNextTask() {
178 NextTask result{};
179
180 auto tick = rtc::TimeMillis();
181
182 MutexLock lock(&pending_lock_);
183
184 if (thread_should_quit_) {
185 result.final_task_ = true;
186 return result;
187 }
188
189 if (delayed_queue_.size() > 0) {
190 auto delayed_entry = delayed_queue_.begin();
191 const auto& delay_info = delayed_entry->first;
192 auto& delay_run = delayed_entry->second;
193 if (tick >= delay_info.next_fire_at_ms_) {
194 if (pending_queue_.size() > 0) {
195 auto& entry = pending_queue_.front();
196 auto& entry_order = entry.first;
197 auto& entry_run = entry.second;
198 if (entry_order < delay_info.order_) {
199 result.run_task_ = std::move(entry_run);
200 pending_queue_.pop();
201 return result;
202 }
203 }
204
205 result.run_task_ = std::move(delay_run);
206 delayed_queue_.erase(delayed_entry);
207 return result;
208 }
209
210 result.sleep_time_ms_ = delay_info.next_fire_at_ms_ - tick;
211 }
212
213 if (pending_queue_.size() > 0) {
214 auto& entry = pending_queue_.front();
215 result.run_task_ = std::move(entry.second);
216 pending_queue_.pop();
217 }
218
219 return result;
220 }
221
222 // static
ThreadMain(void * context)223 void TaskQueueStdlib::ThreadMain(void* context) {
224 TaskQueueStdlib* me = static_cast<TaskQueueStdlib*>(context);
225 CurrentTaskQueueSetter set_current(me);
226 me->ProcessTasks();
227 }
228
ProcessTasks()229 void TaskQueueStdlib::ProcessTasks() {
230 started_.Set();
231
232 while (true) {
233 auto task = GetNextTask();
234
235 if (task.final_task_)
236 break;
237
238 if (task.run_task_) {
239 // process entry immediately then try again
240 QueuedTask* release_ptr = task.run_task_.release();
241 if (release_ptr->Run())
242 delete release_ptr;
243
244 // attempt to sleep again
245 continue;
246 }
247
248 if (0 == task.sleep_time_ms_)
249 flag_notify_.Wait(rtc::Event::kForever);
250 else
251 flag_notify_.Wait(task.sleep_time_ms_);
252 }
253
254 stopped_.Set();
255 }
256
NotifyWake()257 void TaskQueueStdlib::NotifyWake() {
258 // The queue holds pending tasks to complete. Either tasks are to be
259 // executed immediately or tasks are to be run at some future delayed time.
260 // For immediate tasks the task queue's thread is busy running the task and
261 // the thread will not be waiting on the flag_notify_ event. If no immediate
262 // tasks are available but a delayed task is pending then the thread will be
263 // waiting on flag_notify_ with a delayed time-out of the nearest timed task
264 // to run. If no immediate or pending tasks are available, the thread will
265 // wait on flag_notify_ until signaled that a task has been added (or the
266 // thread to be told to shutdown).
267
268 // In all cases, when a new immediate task, delayed task, or request to
269 // shutdown the thread is added the flag_notify_ is signaled after. If the
270 // thread was waiting then the thread will wake up immediately and re-assess
271 // what task needs to be run next (i.e. run a task now, wait for the nearest
272 // timed delayed task, or shutdown the thread). If the thread was not waiting
273 // then the thread will remained signaled to wake up the next time any
274 // attempt to wait on the flag_notify_ event occurs.
275
276 // Any immediate or delayed pending task (or request to shutdown the thread)
277 // must always be added to the queue prior to signaling flag_notify_ to wake
278 // up the possibly sleeping thread. This prevents a race condition where the
279 // thread is notified to wake up but the task queue's thread finds nothing to
280 // do so it waits once again to be signaled where such a signal may never
281 // happen.
282 flag_notify_.Set();
283 }
284
285 class TaskQueueStdlibFactory final : public TaskQueueFactory {
286 public:
CreateTaskQueue(absl::string_view name,Priority priority) const287 std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue(
288 absl::string_view name,
289 Priority priority) const override {
290 return std::unique_ptr<TaskQueueBase, TaskQueueDeleter>(
291 new TaskQueueStdlib(name, TaskQueuePriorityToThreadPriority(priority)));
292 }
293 };
294
295 } // namespace
296
CreateTaskQueueStdlibFactory()297 std::unique_ptr<TaskQueueFactory> CreateTaskQueueStdlibFactory() {
298 return std::make_unique<TaskQueueStdlibFactory>();
299 }
300
301 } // namespace webrtc
302