blob: dd14a7d8e052cc33bab249969e361199db052e84 [file] [log] [blame]
tommic06b1332016-05-14 18:31:401/*
2 * Copyright 2016 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 Chapovalov826f2e72019-02-20 17:13:0911#include "rtc_base/task_queue_win.h"
tommic06b1332016-05-14 18:31:4012
Yves Gerey665174f2018-06-19 13:03:0513// clang-format off
14// clang formating would change include order.
15
Danil Chapovalov02fddf62018-02-12 11:41:1616// Include winsock2.h before including <windows.h> to maintain consistency with
Niels Möllerb06b0a62018-05-25 08:05:3417// win32.h. To include win32.h directly, it must be broken out into its own
18// build target.
Danil Chapovalov02fddf62018-02-12 11:41:1619#include <winsock2.h>
20#include <windows.h>
Yves Gerey665174f2018-06-19 13:03:0521#include <sal.h> // Must come after windows headers.
Danil Chapovalov02fddf62018-02-12 11:41:1622#include <mmsystem.h> // Must come after windows headers.
Yves Gerey665174f2018-06-19 13:03:0523// clang-format on
tommic06b1332016-05-14 18:31:4024#include <string.h>
tommic06b1332016-05-14 18:31:4025
tommif9d91542017-02-17 10:47:1126#include <algorithm>
Mirko Bonadei317a1f02019-09-17 15:06:1827#include <memory>
tommi0b942152017-03-10 17:33:5328#include <queue>
Danil Chapovalov6f09ae22017-10-12 12:39:2529#include <utility>
tommif9d91542017-02-17 10:47:1130
Danil Chapovalov826f2e72019-02-20 17:13:0931#include "absl/strings/string_view.h"
Markus Handellad5037b2021-05-07 13:02:3632#include "absl/types/optional.h"
Danil Chapovalov826f2e72019-02-20 17:13:0933#include "api/task_queue/queued_task.h"
34#include "api/task_queue/task_queue_base.h"
Mirko Bonadei92ea95e2017-09-15 04:47:3135#include "rtc_base/arraysize.h"
36#include "rtc_base/checks.h"
37#include "rtc_base/event.h"
38#include "rtc_base/logging.h"
Karl Wiberge40468b2017-11-22 09:42:2639#include "rtc_base/numerics/safe_conversions.h"
Mirko Bonadei92ea95e2017-09-15 04:47:3140#include "rtc_base/platform_thread.h"
Markus Handell18523c32020-07-08 15:55:5841#include "rtc_base/synchronization/mutex.h"
Markus Handellad5037b2021-05-07 13:02:3642#include "rtc_base/time_utils.h"
tommic06b1332016-05-14 18:31:4043
Danil Chapovalov826f2e72019-02-20 17:13:0944namespace webrtc {
tommic06b1332016-05-14 18:31:4045namespace {
46#define WM_RUN_TASK WM_USER + 1
47#define WM_QUEUE_DELAYED_TASK WM_USER + 2
48
tommic06b1332016-05-14 18:31:4049void CALLBACK InitializeQueueThread(ULONG_PTR param) {
50 MSG msg;
tommif9d91542017-02-17 10:47:1151 ::PeekMessage(&msg, nullptr, WM_USER, WM_USER, PM_NOREMOVE);
Danil Chapovalov826f2e72019-02-20 17:13:0952 rtc::Event* data = reinterpret_cast<rtc::Event*>(param);
53 data->Set();
tommic06b1332016-05-14 18:31:4054}
tommic9bb7912017-02-24 18:42:1455
Danil Chapovalov826f2e72019-02-20 17:13:0956rtc::ThreadPriority TaskQueuePriorityToThreadPriority(
57 TaskQueueFactory::Priority priority) {
tommic9bb7912017-02-24 18:42:1458 switch (priority) {
Danil Chapovalov826f2e72019-02-20 17:13:0959 case TaskQueueFactory::Priority::HIGH:
Markus Handellad5037b2021-05-07 13:02:3660 return rtc::ThreadPriority::kRealtime;
Danil Chapovalov826f2e72019-02-20 17:13:0961 case TaskQueueFactory::Priority::LOW:
Markus Handellad5037b2021-05-07 13:02:3662 return rtc::ThreadPriority::kLow;
Danil Chapovalov826f2e72019-02-20 17:13:0963 case TaskQueueFactory::Priority::NORMAL:
Markus Handellad5037b2021-05-07 13:02:3664 return rtc::ThreadPriority::kNormal;
tommic9bb7912017-02-24 18:42:1465 }
tommic9bb7912017-02-24 18:42:1466}
tommi5bdee472017-03-03 13:20:1267
tommi0b942152017-03-10 17:33:5368int64_t GetTick() {
tommi5bdee472017-03-03 13:20:1269 static const UINT kPeriod = 1;
70 bool high_res = (timeBeginPeriod(kPeriod) == TIMERR_NOERROR);
Danil Chapovalov826f2e72019-02-20 17:13:0971 int64_t ret = rtc::TimeMillis();
tommi5bdee472017-03-03 13:20:1272 if (high_res)
73 timeEndPeriod(kPeriod);
74 return ret;
75}
tommic06b1332016-05-14 18:31:4076
tommi0b942152017-03-10 17:33:5377class DelayedTaskInfo {
tommif9d91542017-02-17 10:47:1178 public:
tommi0b942152017-03-10 17:33:5379 // Default ctor needed to support priority_queue::pop().
80 DelayedTaskInfo() {}
81 DelayedTaskInfo(uint32_t milliseconds, std::unique_ptr<QueuedTask> task)
82 : due_time_(GetTick() + milliseconds), task_(std::move(task)) {}
83 DelayedTaskInfo(DelayedTaskInfo&&) = default;
tommif9d91542017-02-17 10:47:1184
tommi0b942152017-03-10 17:33:5385 // Implement for priority_queue.
86 bool operator>(const DelayedTaskInfo& other) const {
87 return due_time_ > other.due_time_;
88 }
tommif9d91542017-02-17 10:47:1189
tommi0b942152017-03-10 17:33:5390 // Required by priority_queue::pop().
91 DelayedTaskInfo& operator=(DelayedTaskInfo&& other) = default;
92
93 // See below for why this method is const.
94 void Run() const {
95 RTC_DCHECK(due_time_);
96 task_->Run() ? task_.reset() : static_cast<void>(task_.release());
97 }
98
99 int64_t due_time() const { return due_time_; }
100
101 private:
102 int64_t due_time_ = 0; // Absolute timestamp in milliseconds.
103
Artem Titov96e3b992021-07-26 14:03:14104 // `task` needs to be mutable because std::priority_queue::top() returns
tommi0b942152017-03-10 17:33:53105 // a const reference and a key in an ordered queue must not be changed.
106 // There are two basic workarounds, one using const_cast, which would also
Artem Titov96e3b992021-07-26 14:03:14107 // make the key (`due_time`), non-const and the other is to make the non-key
108 // (`task`), mutable.
109 // Because of this, the `task` variable is made private and can only be
110 // mutated by calling the `Run()` method.
tommi0b942152017-03-10 17:33:53111 mutable std::unique_ptr<QueuedTask> task_;
112};
113
114class MultimediaTimer {
115 public:
tommi83722262017-03-15 11:36:29116 // Note: We create an event that requires manual reset.
117 MultimediaTimer() : event_(::CreateEvent(nullptr, true, false, nullptr)) {}
tommif9d91542017-02-17 10:47:11118
tommi0b942152017-03-10 17:33:53119 ~MultimediaTimer() {
120 Cancel();
121 ::CloseHandle(event_);
tommif9d91542017-02-17 10:47:11122 }
123
Artem Titov6cae2d52022-01-26 15:01:10124 MultimediaTimer(const MultimediaTimer&) = delete;
125 MultimediaTimer& operator=(const MultimediaTimer&) = delete;
126
tommi0b942152017-03-10 17:33:53127 bool StartOneShotTimer(UINT delay_ms) {
tommif9d91542017-02-17 10:47:11128 RTC_DCHECK_EQ(0, timer_id_);
129 RTC_DCHECK(event_ != nullptr);
tommif9d91542017-02-17 10:47:11130 timer_id_ =
131 ::timeSetEvent(delay_ms, 0, reinterpret_cast<LPTIMECALLBACK>(event_), 0,
132 TIME_ONESHOT | TIME_CALLBACK_EVENT_SET);
133 return timer_id_ != 0;
134 }
135
tommi0b942152017-03-10 17:33:53136 void Cancel() {
tommif9d91542017-02-17 10:47:11137 if (timer_id_) {
138 ::timeKillEvent(timer_id_);
139 timer_id_ = 0;
140 }
Danil Chapovalovfa733932020-01-13 11:56:13141 // Now that timer is killed and not able to set the event, reset the event.
142 // Doing it in opposite order is racy because event may be set between
143 // event was reset and timer is killed leaving MultimediaTimer in surprising
144 // state where both event is set and timer is canceled.
145 ::ResetEvent(event_);
tommif9d91542017-02-17 10:47:11146 }
147
tommi0b942152017-03-10 17:33:53148 HANDLE* event_for_wait() { return &event_; }
tommif9d91542017-02-17 10:47:11149
150 private:
tommif9d91542017-02-17 10:47:11151 HANDLE event_ = nullptr;
152 MMRESULT timer_id_ = 0;
tommif9d91542017-02-17 10:47:11153};
154
Danil Chapovalov826f2e72019-02-20 17:13:09155class TaskQueueWin : public TaskQueueBase {
tommi0b942152017-03-10 17:33:53156 public:
Danil Chapovalov826f2e72019-02-20 17:13:09157 TaskQueueWin(absl::string_view queue_name, rtc::ThreadPriority priority);
158 ~TaskQueueWin() override = default;
tommi0b942152017-03-10 17:33:53159
Danil Chapovalov826f2e72019-02-20 17:13:09160 void Delete() override;
161 void PostTask(std::unique_ptr<QueuedTask> task) override;
162 void PostDelayedTask(std::unique_ptr<QueuedTask> task,
163 uint32_t milliseconds) override;
nisse341c8e402017-09-06 11:38:22164
165 void RunPendingTasks();
tommi0b942152017-03-10 17:33:53166
167 private:
Danil Chapovalov826f2e72019-02-20 17:13:09168 void RunThreadMain();
169 bool ProcessQueuedMessages();
170 void RunDueTasks();
171 void ScheduleNextTimer();
172 void CancelTimers();
nisse341c8e402017-09-06 11:38:22173
Danil Chapovalov826f2e72019-02-20 17:13:09174 // Since priority_queue<> by defult orders items in terms of
175 // largest->smallest, using std::less<>, and we want smallest->largest,
176 // we would like to use std::greater<> here. Alas it's only available in
177 // C++14 and later, so we roll our own compare template that that relies on
178 // operator<().
179 template <typename T>
180 struct greater {
181 bool operator()(const T& l, const T& r) { return l > r; }
nisse341c8e402017-09-06 11:38:22182 };
183
Danil Chapovalov826f2e72019-02-20 17:13:09184 MultimediaTimer timer_;
185 std::priority_queue<DelayedTaskInfo,
186 std::vector<DelayedTaskInfo>,
187 greater<DelayedTaskInfo>>
188 timer_tasks_;
189 UINT_PTR timer_id_ = 0;
Markus Handellad5037b2021-05-07 13:02:36190 rtc::PlatformThread thread_;
Markus Handell18523c32020-07-08 15:55:58191 Mutex pending_lock_;
danilchapa37de392017-09-09 11:17:22192 std::queue<std::unique_ptr<QueuedTask>> pending_
193 RTC_GUARDED_BY(pending_lock_);
tommi83722262017-03-15 11:36:29194 HANDLE in_queue_;
tommi0b942152017-03-10 17:33:53195};
196
Danil Chapovalov826f2e72019-02-20 17:13:09197TaskQueueWin::TaskQueueWin(absl::string_view queue_name,
198 rtc::ThreadPriority priority)
Markus Handellad5037b2021-05-07 13:02:36199 : in_queue_(::CreateEvent(nullptr, true, false, nullptr)) {
tommi83722262017-03-15 11:36:29200 RTC_DCHECK(in_queue_);
Markus Handellad5037b2021-05-07 13:02:36201 thread_ = rtc::PlatformThread::SpawnJoinable(
202 [this] { RunThreadMain(); }, queue_name,
203 rtc::ThreadAttributes().SetPriority(priority));
204
Danil Chapovalov826f2e72019-02-20 17:13:09205 rtc::Event event(false, false);
tommic06b1332016-05-14 18:31:40206 RTC_CHECK(thread_.QueueAPC(&InitializeQueueThread,
Danil Chapovalov826f2e72019-02-20 17:13:09207 reinterpret_cast<ULONG_PTR>(&event)));
208 event.Wait(rtc::Event::kForever);
tommic06b1332016-05-14 18:31:40209}
210
Danil Chapovalov826f2e72019-02-20 17:13:09211void TaskQueueWin::Delete() {
tommic06b1332016-05-14 18:31:40212 RTC_DCHECK(!IsCurrent());
Markus Handellad5037b2021-05-07 13:02:36213 RTC_CHECK(thread_.GetHandle() != absl::nullopt);
214 while (
215 !::PostThreadMessage(GetThreadId(*thread_.GetHandle()), WM_QUIT, 0, 0)) {
kwiberg352444f2016-11-28 23:58:53216 RTC_CHECK_EQ(ERROR_NOT_ENOUGH_QUOTA, ::GetLastError());
tommic06b1332016-05-14 18:31:40217 Sleep(1);
218 }
Markus Handellad5037b2021-05-07 13:02:36219 thread_.Finalize();
tommi83722262017-03-15 11:36:29220 ::CloseHandle(in_queue_);
Danil Chapovalov826f2e72019-02-20 17:13:09221 delete this;
tommic06b1332016-05-14 18:31:40222}
223
Danil Chapovalov826f2e72019-02-20 17:13:09224void TaskQueueWin::PostTask(std::unique_ptr<QueuedTask> task) {
Markus Handell18523c32020-07-08 15:55:58225 MutexLock lock(&pending_lock_);
tommi83722262017-03-15 11:36:29226 pending_.push(std::move(task));
227 ::SetEvent(in_queue_);
tommic06b1332016-05-14 18:31:40228}
229
Danil Chapovalov826f2e72019-02-20 17:13:09230void TaskQueueWin::PostDelayedTask(std::unique_ptr<QueuedTask> task,
231 uint32_t milliseconds) {
tommi0b942152017-03-10 17:33:53232 if (!milliseconds) {
233 PostTask(std::move(task));
234 return;
235 }
236
237 // TODO(tommi): Avoid this allocation. It is currently here since
238 // the timestamp stored in the task info object, is a 64bit timestamp
239 // and WPARAM is 32bits in 32bit builds. Otherwise, we could pass the
240 // task pointer and timestamp as LPARAM and WPARAM.
241 auto* task_info = new DelayedTaskInfo(milliseconds, std::move(task));
Markus Handellad5037b2021-05-07 13:02:36242 RTC_CHECK(thread_.GetHandle() != absl::nullopt);
243 if (!::PostThreadMessage(GetThreadId(*thread_.GetHandle()),
244 WM_QUEUE_DELAYED_TASK, 0,
tommi0b942152017-03-10 17:33:53245 reinterpret_cast<LPARAM>(task_info))) {
246 delete task_info;
tommic06b1332016-05-14 18:31:40247 }
248}
249
Danil Chapovalov826f2e72019-02-20 17:13:09250void TaskQueueWin::RunPendingTasks() {
tommi83722262017-03-15 11:36:29251 while (true) {
252 std::unique_ptr<QueuedTask> task;
253 {
Markus Handell18523c32020-07-08 15:55:58254 MutexLock lock(&pending_lock_);
tommi83722262017-03-15 11:36:29255 if (pending_.empty())
256 break;
257 task = std::move(pending_.front());
258 pending_.pop();
259 }
260
261 if (!task->Run())
262 task.release();
263 }
264}
265
Danil Chapovalov826f2e72019-02-20 17:13:09266void TaskQueueWin::RunThreadMain() {
267 CurrentTaskQueueSetter set_current(this);
Yves Gerey665174f2018-06-19 13:03:05268 HANDLE handles[2] = {*timer_.event_for_wait(), in_queue_};
tommib89257a2016-07-12 08:24:36269 while (true) {
tommif9d91542017-02-17 10:47:11270 // Make sure we do an alertable wait as that's required to allow APCs to run
271 // (e.g. required for InitializeQueueThread and stopping the thread in
272 // PlatformThread).
tommi0b942152017-03-10 17:33:53273 DWORD result = ::MsgWaitForMultipleObjectsEx(
tommi83722262017-03-15 11:36:29274 arraysize(handles), handles, INFINITE, QS_ALLEVENTS, MWMO_ALERTABLE);
tommib89257a2016-07-12 08:24:36275 RTC_CHECK_NE(WAIT_FAILED, result);
tommi83722262017-03-15 11:36:29276 if (result == (WAIT_OBJECT_0 + 2)) {
tommi0b942152017-03-10 17:33:53277 // There are messages in the message queue that need to be handled.
278 if (!ProcessQueuedMessages())
tommib89257a2016-07-12 08:24:36279 break;
tommi83722262017-03-15 11:36:29280 }
281
Yves Gerey665174f2018-06-19 13:03:05282 if (result == WAIT_OBJECT_0 ||
283 (!timer_tasks_.empty() &&
284 ::WaitForSingleObject(*timer_.event_for_wait(), 0) == WAIT_OBJECT_0)) {
tommi0b942152017-03-10 17:33:53285 // The multimedia timer was signaled.
286 timer_.Cancel();
tommi0b942152017-03-10 17:33:53287 RunDueTasks();
288 ScheduleNextTimer();
tommi83722262017-03-15 11:36:29289 }
290
291 if (result == (WAIT_OBJECT_0 + 1)) {
292 ::ResetEvent(in_queue_);
Danil Chapovalov826f2e72019-02-20 17:13:09293 RunPendingTasks();
tommib89257a2016-07-12 08:24:36294 }
295 }
tommib89257a2016-07-12 08:24:36296}
tommic06b1332016-05-14 18:31:40297
Danil Chapovalov826f2e72019-02-20 17:13:09298bool TaskQueueWin::ProcessQueuedMessages() {
tommib89257a2016-07-12 08:24:36299 MSG msg = {};
tommi83722262017-03-15 11:36:29300 // To protect against overly busy message queues, we limit the time
301 // we process tasks to a few milliseconds. If we don't do that, there's
302 // a chance that timer tasks won't ever run.
303 static const int kMaxTaskProcessingTimeMs = 500;
304 auto start = GetTick();
tommif9d91542017-02-17 10:47:11305 while (::PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) &&
tommib89257a2016-07-12 08:24:36306 msg.message != WM_QUIT) {
tommic06b1332016-05-14 18:31:40307 if (!msg.hwnd) {
308 switch (msg.message) {
tommi83722262017-03-15 11:36:29309 // TODO(tommi): Stop using this way of queueing tasks.
tommic06b1332016-05-14 18:31:40310 case WM_RUN_TASK: {
311 QueuedTask* task = reinterpret_cast<QueuedTask*>(msg.lParam);
312 if (task->Run())
313 delete task;
314 break;
315 }
316 case WM_QUEUE_DELAYED_TASK: {
tommi0b942152017-03-10 17:33:53317 std::unique_ptr<DelayedTaskInfo> info(
318 reinterpret_cast<DelayedTaskInfo*>(msg.lParam));
319 bool need_to_schedule_timers =
320 timer_tasks_.empty() ||
321 timer_tasks_.top().due_time() > info->due_time();
322 timer_tasks_.emplace(std::move(*info.get()));
323 if (need_to_schedule_timers) {
324 CancelTimers();
325 ScheduleNextTimer();
tommif9d91542017-02-17 10:47:11326 }
tommic06b1332016-05-14 18:31:40327 break;
328 }
329 case WM_TIMER: {
tommi0b942152017-03-10 17:33:53330 RTC_DCHECK_EQ(timer_id_, msg.wParam);
tommif9d91542017-02-17 10:47:11331 ::KillTimer(nullptr, msg.wParam);
tommi0b942152017-03-10 17:33:53332 timer_id_ = 0;
333 RunDueTasks();
334 ScheduleNextTimer();
tommic06b1332016-05-14 18:31:40335 break;
336 }
337 default:
Artem Titovd3251962021-11-15 15:57:07338 RTC_DCHECK_NOTREACHED();
tommic06b1332016-05-14 18:31:40339 break;
340 }
341 } else {
tommif9d91542017-02-17 10:47:11342 ::TranslateMessage(&msg);
343 ::DispatchMessage(&msg);
tommic06b1332016-05-14 18:31:40344 }
tommi83722262017-03-15 11:36:29345
346 if (GetTick() > start + kMaxTaskProcessingTimeMs)
347 break;
tommic06b1332016-05-14 18:31:40348 }
tommib89257a2016-07-12 08:24:36349 return msg.message != WM_QUIT;
tommic06b1332016-05-14 18:31:40350}
tommib89257a2016-07-12 08:24:36351
Danil Chapovalov826f2e72019-02-20 17:13:09352void TaskQueueWin::RunDueTasks() {
tommi0b942152017-03-10 17:33:53353 RTC_DCHECK(!timer_tasks_.empty());
354 auto now = GetTick();
355 do {
356 const auto& top = timer_tasks_.top();
357 if (top.due_time() > now)
358 break;
359 top.Run();
360 timer_tasks_.pop();
361 } while (!timer_tasks_.empty());
362}
363
Danil Chapovalov826f2e72019-02-20 17:13:09364void TaskQueueWin::ScheduleNextTimer() {
tommi0b942152017-03-10 17:33:53365 RTC_DCHECK_EQ(timer_id_, 0);
366 if (timer_tasks_.empty())
367 return;
368
369 const auto& next_task = timer_tasks_.top();
370 int64_t delay_ms = std::max(0ll, next_task.due_time() - GetTick());
371 uint32_t milliseconds = rtc::dchecked_cast<uint32_t>(delay_ms);
372 if (!timer_.StartOneShotTimer(milliseconds))
373 timer_id_ = ::SetTimer(nullptr, 0, milliseconds, nullptr);
374}
375
Danil Chapovalov826f2e72019-02-20 17:13:09376void TaskQueueWin::CancelTimers() {
tommi0b942152017-03-10 17:33:53377 timer_.Cancel();
378 if (timer_id_) {
379 ::KillTimer(nullptr, timer_id_);
380 timer_id_ = 0;
381 }
382}
383
Danil Chapovalov826f2e72019-02-20 17:13:09384class TaskQueueWinFactory : public TaskQueueFactory {
385 public:
386 std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue(
387 absl::string_view name,
388 Priority priority) const override {
389 return std::unique_ptr<TaskQueueBase, TaskQueueDeleter>(
390 new TaskQueueWin(name, TaskQueuePriorityToThreadPriority(priority)));
391 }
392};
393
394} // namespace
395
396std::unique_ptr<TaskQueueFactory> CreateTaskQueueWinFactory() {
Mirko Bonadei317a1f02019-09-17 15:06:18397 return std::make_unique<TaskQueueWinFactory>();
nisse341c8e402017-09-06 11:38:22398}
399
Danil Chapovalov826f2e72019-02-20 17:13:09400} // namespace webrtc