blob: d797d478f4fc0a1f5a26e6ddb6a77ef775a143a8 [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"
Markus Handellad5037b2021-05-07 13:02:3637#include "rtc_base/constructor_magic.h"
Mirko Bonadei92ea95e2017-09-15 04:47:3138#include "rtc_base/event.h"
39#include "rtc_base/logging.h"
Karl Wiberge40468b2017-11-22 09:42:2640#include "rtc_base/numerics/safe_conversions.h"
Mirko Bonadei92ea95e2017-09-15 04:47:3141#include "rtc_base/platform_thread.h"
Markus Handell18523c32020-07-08 15:55:5842#include "rtc_base/synchronization/mutex.h"
Markus Handellad5037b2021-05-07 13:02:3643#include "rtc_base/time_utils.h"
tommic06b1332016-05-14 18:31:4044
Danil Chapovalov826f2e72019-02-20 17:13:0945namespace webrtc {
tommic06b1332016-05-14 18:31:4046namespace {
47#define WM_RUN_TASK WM_USER + 1
48#define WM_QUEUE_DELAYED_TASK WM_USER + 2
49
tommic06b1332016-05-14 18:31:4050void CALLBACK InitializeQueueThread(ULONG_PTR param) {
51 MSG msg;
tommif9d91542017-02-17 10:47:1152 ::PeekMessage(&msg, nullptr, WM_USER, WM_USER, PM_NOREMOVE);
Danil Chapovalov826f2e72019-02-20 17:13:0953 rtc::Event* data = reinterpret_cast<rtc::Event*>(param);
54 data->Set();
tommic06b1332016-05-14 18:31:4055}
tommic9bb7912017-02-24 18:42:1456
Danil Chapovalov826f2e72019-02-20 17:13:0957rtc::ThreadPriority TaskQueuePriorityToThreadPriority(
58 TaskQueueFactory::Priority priority) {
tommic9bb7912017-02-24 18:42:1459 switch (priority) {
Danil Chapovalov826f2e72019-02-20 17:13:0960 case TaskQueueFactory::Priority::HIGH:
Markus Handellad5037b2021-05-07 13:02:3661 return rtc::ThreadPriority::kRealtime;
Danil Chapovalov826f2e72019-02-20 17:13:0962 case TaskQueueFactory::Priority::LOW:
Markus Handellad5037b2021-05-07 13:02:3663 return rtc::ThreadPriority::kLow;
Danil Chapovalov826f2e72019-02-20 17:13:0964 case TaskQueueFactory::Priority::NORMAL:
Markus Handellad5037b2021-05-07 13:02:3665 return rtc::ThreadPriority::kNormal;
tommic9bb7912017-02-24 18:42:1466 }
tommic9bb7912017-02-24 18:42:1467}
tommi5bdee472017-03-03 13:20:1268
tommi0b942152017-03-10 17:33:5369int64_t GetTick() {
tommi5bdee472017-03-03 13:20:1270 static const UINT kPeriod = 1;
71 bool high_res = (timeBeginPeriod(kPeriod) == TIMERR_NOERROR);
Danil Chapovalov826f2e72019-02-20 17:13:0972 int64_t ret = rtc::TimeMillis();
tommi5bdee472017-03-03 13:20:1273 if (high_res)
74 timeEndPeriod(kPeriod);
75 return ret;
76}
tommic06b1332016-05-14 18:31:4077
tommi0b942152017-03-10 17:33:5378class DelayedTaskInfo {
tommif9d91542017-02-17 10:47:1179 public:
tommi0b942152017-03-10 17:33:5380 // Default ctor needed to support priority_queue::pop().
81 DelayedTaskInfo() {}
82 DelayedTaskInfo(uint32_t milliseconds, std::unique_ptr<QueuedTask> task)
83 : due_time_(GetTick() + milliseconds), task_(std::move(task)) {}
84 DelayedTaskInfo(DelayedTaskInfo&&) = default;
tommif9d91542017-02-17 10:47:1185
tommi0b942152017-03-10 17:33:5386 // Implement for priority_queue.
87 bool operator>(const DelayedTaskInfo& other) const {
88 return due_time_ > other.due_time_;
89 }
tommif9d91542017-02-17 10:47:1190
tommi0b942152017-03-10 17:33:5391 // Required by priority_queue::pop().
92 DelayedTaskInfo& operator=(DelayedTaskInfo&& other) = default;
93
94 // See below for why this method is const.
95 void Run() const {
96 RTC_DCHECK(due_time_);
97 task_->Run() ? task_.reset() : static_cast<void>(task_.release());
98 }
99
100 int64_t due_time() const { return due_time_; }
101
102 private:
103 int64_t due_time_ = 0; // Absolute timestamp in milliseconds.
104
105 // |task| needs to be mutable because std::priority_queue::top() returns
106 // a const reference and a key in an ordered queue must not be changed.
107 // There are two basic workarounds, one using const_cast, which would also
108 // make the key (|due_time|), non-const and the other is to make the non-key
109 // (|task|), mutable.
110 // Because of this, the |task| variable is made private and can only be
111 // mutated by calling the |Run()| method.
112 mutable std::unique_ptr<QueuedTask> task_;
113};
114
115class MultimediaTimer {
116 public:
tommi83722262017-03-15 11:36:29117 // Note: We create an event that requires manual reset.
118 MultimediaTimer() : event_(::CreateEvent(nullptr, true, false, nullptr)) {}
tommif9d91542017-02-17 10:47:11119
tommi0b942152017-03-10 17:33:53120 ~MultimediaTimer() {
121 Cancel();
122 ::CloseHandle(event_);
tommif9d91542017-02-17 10:47:11123 }
124
tommi0b942152017-03-10 17:33:53125 bool StartOneShotTimer(UINT delay_ms) {
tommif9d91542017-02-17 10:47:11126 RTC_DCHECK_EQ(0, timer_id_);
127 RTC_DCHECK(event_ != nullptr);
tommif9d91542017-02-17 10:47:11128 timer_id_ =
129 ::timeSetEvent(delay_ms, 0, reinterpret_cast<LPTIMECALLBACK>(event_), 0,
130 TIME_ONESHOT | TIME_CALLBACK_EVENT_SET);
131 return timer_id_ != 0;
132 }
133
tommi0b942152017-03-10 17:33:53134 void Cancel() {
tommif9d91542017-02-17 10:47:11135 if (timer_id_) {
136 ::timeKillEvent(timer_id_);
137 timer_id_ = 0;
138 }
Danil Chapovalovfa733932020-01-13 11:56:13139 // Now that timer is killed and not able to set the event, reset the event.
140 // Doing it in opposite order is racy because event may be set between
141 // event was reset and timer is killed leaving MultimediaTimer in surprising
142 // state where both event is set and timer is canceled.
143 ::ResetEvent(event_);
tommif9d91542017-02-17 10:47:11144 }
145
tommi0b942152017-03-10 17:33:53146 HANDLE* event_for_wait() { return &event_; }
tommif9d91542017-02-17 10:47:11147
148 private:
tommif9d91542017-02-17 10:47:11149 HANDLE event_ = nullptr;
150 MMRESULT timer_id_ = 0;
tommif9d91542017-02-17 10:47:11151
152 RTC_DISALLOW_COPY_AND_ASSIGN(MultimediaTimer);
153};
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:
338 RTC_NOTREACHED();
339 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