blob: 1ac01e1830ec711f23f538742ef5f81263ca9629 [file] [log] [blame]
Robin Raymond22027b92018-11-23 14:07:501/*
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
Danil Chapovalovfa52efa2019-02-21 10:13:5811#include "rtc_base/task_queue_stdlib.h"
Robin Raymond22027b92018-11-23 14:07:5012
13#include <string.h>
Jonas Olssona4d87372019-07-05 17:08:3314
Robin Raymond22027b92018-11-23 14:07:5015#include <algorithm>
Robin Raymond22027b92018-11-23 14:07:5016#include <map>
Mirko Bonadei317a1f02019-09-17 15:06:1817#include <memory>
Robin Raymond22027b92018-11-23 14:07:5018#include <queue>
19#include <utility>
20
Danil Chapovalovba570012022-07-19 16:36:3121#include "absl/functional/any_invocable.h"
Danil Chapovalovfa52efa2019-02-21 10:13:5822#include "absl/strings/string_view.h"
Danil Chapovalovfa52efa2019-02-21 10:13:5823#include "api/task_queue/task_queue_base.h"
Danil Chapovalovba570012022-07-19 16:36:3124#include "api/units/time_delta.h"
Robin Raymond22027b92018-11-23 14:07:5025#include "rtc_base/checks.h"
Robin Raymond22027b92018-11-23 14:07:5026#include "rtc_base/event.h"
27#include "rtc_base/logging.h"
Danil Chapovalovba570012022-07-19 16:36:3128#include "rtc_base/numerics/divide_round.h"
Robin Raymond22027b92018-11-23 14:07:5029#include "rtc_base/platform_thread.h"
Markus Handell18523c32020-07-08 15:55:5830#include "rtc_base/synchronization/mutex.h"
Robin Raymond22027b92018-11-23 14:07:5031#include "rtc_base/thread_annotations.h"
Steve Anton10542f22019-01-11 17:11:0032#include "rtc_base/time_utils.h"
Robin Raymond22027b92018-11-23 14:07:5033
Danil Chapovalovfa52efa2019-02-21 10:13:5834namespace webrtc {
Robin Raymond22027b92018-11-23 14:07:5035namespace {
36
Danil Chapovalovfa52efa2019-02-21 10:13:5837rtc::ThreadPriority TaskQueuePriorityToThreadPriority(
38 TaskQueueFactory::Priority priority) {
Robin Raymond22027b92018-11-23 14:07:5039 switch (priority) {
Danil Chapovalovfa52efa2019-02-21 10:13:5840 case TaskQueueFactory::Priority::HIGH:
Markus Handellad5037b2021-05-07 13:02:3641 return rtc::ThreadPriority::kRealtime;
Danil Chapovalovfa52efa2019-02-21 10:13:5842 case TaskQueueFactory::Priority::LOW:
Markus Handellad5037b2021-05-07 13:02:3643 return rtc::ThreadPriority::kLow;
Danil Chapovalovfa52efa2019-02-21 10:13:5844 case TaskQueueFactory::Priority::NORMAL:
Markus Handellad5037b2021-05-07 13:02:3645 return rtc::ThreadPriority::kNormal;
Robin Raymond22027b92018-11-23 14:07:5046 }
Robin Raymond22027b92018-11-23 14:07:5047}
48
Danil Chapovalovfa52efa2019-02-21 10:13:5849class TaskQueueStdlib final : public TaskQueueBase {
Robin Raymond22027b92018-11-23 14:07:5050 public:
Danil Chapovalovfa52efa2019-02-21 10:13:5851 TaskQueueStdlib(absl::string_view queue_name, rtc::ThreadPriority priority);
52 ~TaskQueueStdlib() override = default;
Robin Raymond22027b92018-11-23 14:07:5053
Danil Chapovalovfa52efa2019-02-21 10:13:5854 void Delete() override;
Markus Handell2a256c82023-02-27 11:41:3955
56 protected:
57 void PostTaskImpl(absl::AnyInvocable<void() &&> task,
58 const PostTaskTraits& traits,
59 const Location& location) override;
60 void PostDelayedTaskImpl(absl::AnyInvocable<void() &&> task,
61 TimeDelta delay,
62 const PostDelayedTaskTraits& traits,
63 const Location& location) override;
Robin Raymond22027b92018-11-23 14:07:5064
Danil Chapovalovfa52efa2019-02-21 10:13:5865 private:
Robin Raymond22027b92018-11-23 14:07:5066 using OrderId = uint64_t;
67
68 struct DelayedEntryTimeout {
Markus Handell2cfc1af2022-08-19 08:16:4869 // TODO(bugs.webrtc.org/13756): Migrate to Timestamp.
Danil Chapovalovba570012022-07-19 16:36:3170 int64_t next_fire_at_us{};
71 OrderId order{};
Robin Raymond22027b92018-11-23 14:07:5072
73 bool operator<(const DelayedEntryTimeout& o) const {
Danil Chapovalovba570012022-07-19 16:36:3174 return std::tie(next_fire_at_us, order) <
75 std::tie(o.next_fire_at_us, o.order);
Robin Raymond22027b92018-11-23 14:07:5076 }
77 };
78
79 struct NextTask {
Danil Chapovalovba570012022-07-19 16:36:3180 bool final_task = false;
81 absl::AnyInvocable<void() &&> run_task;
Markus Handell2cfc1af2022-08-19 08:16:4882 TimeDelta sleep_time = rtc::Event::kForever;
Robin Raymond22027b92018-11-23 14:07:5083 };
84
Tommi76d9c182022-04-22 13:48:3785 static rtc::PlatformThread InitializeThread(TaskQueueStdlib* me,
86 absl::string_view queue_name,
87 rtc::ThreadPriority priority);
88
Robin Raymond22027b92018-11-23 14:07:5089 NextTask GetNextTask();
90
Robin Raymond22027b92018-11-23 14:07:5091 void ProcessTasks();
92
93 void NotifyWake();
94
Robin Raymond22027b92018-11-23 14:07:5095 // Signaled whenever a new task is pending.
Danil Chapovalovfa52efa2019-02-21 10:13:5896 rtc::Event flag_notify_;
Robin Raymond22027b92018-11-23 14:07:5097
Markus Handell18523c32020-07-08 15:55:5898 Mutex pending_lock_;
Robin Raymond22027b92018-11-23 14:07:5099
100 // Indicates if the worker thread needs to shutdown now.
Tommi76d9c182022-04-22 13:48:37101 bool thread_should_quit_ RTC_GUARDED_BY(pending_lock_) = false;
Robin Raymond22027b92018-11-23 14:07:50102
103 // Holds the next order to use for the next task to be
104 // put into one of the pending queues.
Tommi76d9c182022-04-22 13:48:37105 OrderId thread_posting_order_ RTC_GUARDED_BY(pending_lock_) = 0;
Robin Raymond22027b92018-11-23 14:07:50106
107 // The list of all pending tasks that need to be processed in the
108 // FIFO queue ordering on the worker thread.
Danil Chapovalovba570012022-07-19 16:36:31109 std::queue<std::pair<OrderId, absl::AnyInvocable<void() &&>>> pending_queue_
Robin Raymond22027b92018-11-23 14:07:50110 RTC_GUARDED_BY(pending_lock_);
111
112 // The list of all pending tasks that need to be processed at a future
113 // time based upon a delay. On the off change the delayed task should
114 // happen at exactly the same time interval as another task then the
115 // task is processed based on FIFO ordering. std::priority_queue was
116 // considered but rejected due to its inability to extract the
Danil Chapovalovba570012022-07-19 16:36:31117 // move-only value out of the queue without the presence of a hack.
118 std::map<DelayedEntryTimeout, absl::AnyInvocable<void() &&>> delayed_queue_
Robin Raymond22027b92018-11-23 14:07:50119 RTC_GUARDED_BY(pending_lock_);
Markus Handell49cb4592021-06-22 08:02:26120
121 // Contains the active worker thread assigned to processing
122 // tasks (including delayed tasks).
123 // Placing this last ensures the thread doesn't touch uninitialized attributes
124 // throughout it's lifetime.
125 rtc::PlatformThread thread_;
Robin Raymond22027b92018-11-23 14:07:50126};
127
Danil Chapovalovfa52efa2019-02-21 10:13:58128TaskQueueStdlib::TaskQueueStdlib(absl::string_view queue_name,
129 rtc::ThreadPriority priority)
Tommi76d9c182022-04-22 13:48:37130 : flag_notify_(/*manual_reset=*/false, /*initially_signaled=*/false),
131 thread_(InitializeThread(this, queue_name, priority)) {}
132
133// static
134rtc::PlatformThread TaskQueueStdlib::InitializeThread(
135 TaskQueueStdlib* me,
136 absl::string_view queue_name,
137 rtc::ThreadPriority priority) {
138 rtc::Event started;
139 auto thread = rtc::PlatformThread::SpawnJoinable(
140 [&started, me] {
141 CurrentTaskQueueSetter set_current(me);
142 started.Set();
143 me->ProcessTasks();
144 },
145 queue_name, rtc::ThreadAttributes().SetPriority(priority));
146 started.Wait(rtc::Event::kForever);
147 return thread;
Robin Raymond22027b92018-11-23 14:07:50148}
149
Danil Chapovalovfa52efa2019-02-21 10:13:58150void TaskQueueStdlib::Delete() {
Robin Raymond22027b92018-11-23 14:07:50151 RTC_DCHECK(!IsCurrent());
152
153 {
Markus Handell18523c32020-07-08 15:55:58154 MutexLock lock(&pending_lock_);
Robin Raymond22027b92018-11-23 14:07:50155 thread_should_quit_ = true;
156 }
157
158 NotifyWake();
159
Danil Chapovalovfa52efa2019-02-21 10:13:58160 delete this;
Robin Raymond22027b92018-11-23 14:07:50161}
162
Markus Handell2a256c82023-02-27 11:41:39163void TaskQueueStdlib::PostTaskImpl(absl::AnyInvocable<void() &&> task,
164 const PostTaskTraits& traits,
165 const Location& location) {
Robin Raymond22027b92018-11-23 14:07:50166 {
Markus Handell18523c32020-07-08 15:55:58167 MutexLock lock(&pending_lock_);
Danil Chapovalovba570012022-07-19 16:36:31168 pending_queue_.push(
169 std::make_pair(++thread_posting_order_, std::move(task)));
Robin Raymond22027b92018-11-23 14:07:50170 }
171
172 NotifyWake();
173}
174
Markus Handell2a256c82023-02-27 11:41:39175void TaskQueueStdlib::PostDelayedTaskImpl(absl::AnyInvocable<void() &&> task,
176 TimeDelta delay,
177 const PostDelayedTaskTraits& traits,
178 const Location& location) {
Danil Chapovalovba570012022-07-19 16:36:31179 DelayedEntryTimeout delayed_entry;
180 delayed_entry.next_fire_at_us = rtc::TimeMicros() + delay.us();
Robin Raymond22027b92018-11-23 14:07:50181
182 {
Markus Handell18523c32020-07-08 15:55:58183 MutexLock lock(&pending_lock_);
Danil Chapovalovba570012022-07-19 16:36:31184 delayed_entry.order = ++thread_posting_order_;
185 delayed_queue_[delayed_entry] = std::move(task);
Robin Raymond22027b92018-11-23 14:07:50186 }
187
188 NotifyWake();
189}
190
Danil Chapovalovfa52efa2019-02-21 10:13:58191TaskQueueStdlib::NextTask TaskQueueStdlib::GetNextTask() {
Tommi76d9c182022-04-22 13:48:37192 NextTask result;
Robin Raymond22027b92018-11-23 14:07:50193
Danil Chapovalovba570012022-07-19 16:36:31194 const int64_t tick_us = rtc::TimeMicros();
Robin Raymond22027b92018-11-23 14:07:50195
Markus Handell18523c32020-07-08 15:55:58196 MutexLock lock(&pending_lock_);
Robin Raymond22027b92018-11-23 14:07:50197
198 if (thread_should_quit_) {
Danil Chapovalovba570012022-07-19 16:36:31199 result.final_task = true;
Robin Raymond22027b92018-11-23 14:07:50200 return result;
201 }
202
203 if (delayed_queue_.size() > 0) {
204 auto delayed_entry = delayed_queue_.begin();
205 const auto& delay_info = delayed_entry->first;
206 auto& delay_run = delayed_entry->second;
Danil Chapovalovba570012022-07-19 16:36:31207 if (tick_us >= delay_info.next_fire_at_us) {
Robin Raymond22027b92018-11-23 14:07:50208 if (pending_queue_.size() > 0) {
209 auto& entry = pending_queue_.front();
210 auto& entry_order = entry.first;
211 auto& entry_run = entry.second;
Danil Chapovalovba570012022-07-19 16:36:31212 if (entry_order < delay_info.order) {
213 result.run_task = std::move(entry_run);
Robin Raymond22027b92018-11-23 14:07:50214 pending_queue_.pop();
215 return result;
216 }
217 }
218
Danil Chapovalovba570012022-07-19 16:36:31219 result.run_task = std::move(delay_run);
Robin Raymond22027b92018-11-23 14:07:50220 delayed_queue_.erase(delayed_entry);
221 return result;
222 }
223
Markus Handell2cfc1af2022-08-19 08:16:48224 result.sleep_time = TimeDelta::Millis(
225 DivideRoundUp(delay_info.next_fire_at_us - tick_us, 1'000));
Robin Raymond22027b92018-11-23 14:07:50226 }
227
228 if (pending_queue_.size() > 0) {
229 auto& entry = pending_queue_.front();
Danil Chapovalovba570012022-07-19 16:36:31230 result.run_task = std::move(entry.second);
Robin Raymond22027b92018-11-23 14:07:50231 pending_queue_.pop();
232 }
233
234 return result;
235}
236
Danil Chapovalovfa52efa2019-02-21 10:13:58237void TaskQueueStdlib::ProcessTasks() {
Robin Raymond22027b92018-11-23 14:07:50238 while (true) {
239 auto task = GetNextTask();
240
Danil Chapovalovba570012022-07-19 16:36:31241 if (task.final_task)
Robin Raymond22027b92018-11-23 14:07:50242 break;
243
Danil Chapovalovba570012022-07-19 16:36:31244 if (task.run_task) {
Robin Raymond22027b92018-11-23 14:07:50245 // process entry immediately then try again
Danil Chapovalovba570012022-07-19 16:36:31246 std::move(task.run_task)();
Robin Raymond22027b92018-11-23 14:07:50247
Tommi76d9c182022-04-22 13:48:37248 // Attempt to run more tasks before going to sleep.
Robin Raymond22027b92018-11-23 14:07:50249 continue;
250 }
251
Markus Handell2cfc1af2022-08-19 08:16:48252 flag_notify_.Wait(task.sleep_time);
Robin Raymond22027b92018-11-23 14:07:50253 }
Markus Handell82da9322022-12-16 14:50:24254
255 // Ensure remaining deleted tasks are destroyed with Current() set up to this
256 // task queue.
257 std::queue<std::pair<OrderId, absl::AnyInvocable<void() &&>>> pending_queue;
258 {
259 MutexLock lock(&pending_lock_);
260 pending_queue_.swap(pending_queue);
261 }
262 pending_queue = {};
263#if RTC_DCHECK_IS_ON
264 MutexLock lock(&pending_lock_);
265 RTC_DCHECK(pending_queue_.empty());
266#endif
Robin Raymond22027b92018-11-23 14:07:50267}
268
Danil Chapovalovfa52efa2019-02-21 10:13:58269void TaskQueueStdlib::NotifyWake() {
Robin Raymond22027b92018-11-23 14:07:50270 // The queue holds pending tasks to complete. Either tasks are to be
271 // executed immediately or tasks are to be run at some future delayed time.
272 // For immediate tasks the task queue's thread is busy running the task and
273 // the thread will not be waiting on the flag_notify_ event. If no immediate
274 // tasks are available but a delayed task is pending then the thread will be
275 // waiting on flag_notify_ with a delayed time-out of the nearest timed task
276 // to run. If no immediate or pending tasks are available, the thread will
277 // wait on flag_notify_ until signaled that a task has been added (or the
278 // thread to be told to shutdown).
279
280 // In all cases, when a new immediate task, delayed task, or request to
281 // shutdown the thread is added the flag_notify_ is signaled after. If the
282 // thread was waiting then the thread will wake up immediately and re-assess
283 // what task needs to be run next (i.e. run a task now, wait for the nearest
284 // timed delayed task, or shutdown the thread). If the thread was not waiting
285 // then the thread will remained signaled to wake up the next time any
286 // attempt to wait on the flag_notify_ event occurs.
287
288 // Any immediate or delayed pending task (or request to shutdown the thread)
289 // must always be added to the queue prior to signaling flag_notify_ to wake
290 // up the possibly sleeping thread. This prevents a race condition where the
291 // thread is notified to wake up but the task queue's thread finds nothing to
292 // do so it waits once again to be signaled where such a signal may never
293 // happen.
294 flag_notify_.Set();
295}
296
Danil Chapovalovfa52efa2019-02-21 10:13:58297class TaskQueueStdlibFactory final : public TaskQueueFactory {
298 public:
299 std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue(
300 absl::string_view name,
301 Priority priority) const override {
302 return std::unique_ptr<TaskQueueBase, TaskQueueDeleter>(
303 new TaskQueueStdlib(name, TaskQueuePriorityToThreadPriority(priority)));
304 }
305};
306
307} // namespace
308
309std::unique_ptr<TaskQueueFactory> CreateTaskQueueStdlibFactory() {
Mirko Bonadei317a1f02019-09-17 15:06:18310 return std::make_unique<TaskQueueStdlibFactory>();
Robin Raymond22027b92018-11-23 14:07:50311}
312
Danil Chapovalovfa52efa2019-02-21 10:13:58313} // namespace webrtc