From f9d91548084c39fae6c355fd03202fb8f66eb5a1 Mon Sep 17 00:00:00 2001 From: tommi Date: Fri, 17 Feb 2017 02:47:11 -0800 Subject: [PATCH] Add support for multimedia timers to TaskQueue on Windows. Multimedia timers are higher precision than WM_TIMER, but they're also a limited resource and more costly. So this implementation is a best effort implementation that falls back on WM_TIMER when multimedia timers aren't available. A possible future change could be to make high precision timers in a TaskQueue, optional. The reason for doing so would be for TaskQueues that don't need high precision timers, won't eat up timers from TQ instances that really need it. BUG=webrtc:7151 Review-Url: https://codereview.webrtc.org/2691973002 Cr-Commit-Position: refs/heads/master@{#16661} --- webrtc/base/task_queue.h | 10 +- webrtc/base/task_queue_unittest.cc | 4 +- webrtc/base/task_queue_win.cc | 213 ++++++++++++++++++++++++++--- 3 files changed, 202 insertions(+), 25 deletions(-) diff --git a/webrtc/base/task_queue.h b/webrtc/base/task_queue.h index eeabe05fab..a6a70364ac 100644 --- a/webrtc/base/task_queue.h +++ b/webrtc/base/task_queue.h @@ -178,6 +178,11 @@ class LOCKABLE TaskQueue { void PostTaskAndReply(std::unique_ptr task, std::unique_ptr reply); + // Schedules a task to execute a specified number of milliseconds from when + // the call is made. The precision should be considered as "best effort" + // and in some cases, such as on Windows when all high precision timers have + // been used up, can be off by as much as 15 millseconds (although 8 would be + // more likely). This can be mitigated by limiting the use of delayed tasks. void PostDelayedTask(std::unique_ptr task, uint32_t milliseconds); template @@ -185,6 +190,7 @@ class LOCKABLE TaskQueue { PostTask(std::unique_ptr(new ClosureTask(closure))); } + // See documentation above for performance expectations. template void PostDelayedTask(const Closure& closure, uint32_t milliseconds) { PostDelayedTask( @@ -254,10 +260,12 @@ class LOCKABLE TaskQueue { dispatch_queue_t queue_; QueueContext* const context_; #elif defined(WEBRTC_WIN) + class MultimediaTimer; typedef std::unordered_map> DelayedTasks; static bool ThreadMain(void* context); - static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks); + static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks, + std::vector* timers); class WorkerThread : public PlatformThread { public: diff --git a/webrtc/base/task_queue_unittest.cc b/webrtc/base/task_queue_unittest.cc index 08123d073c..74433b9144 100644 --- a/webrtc/base/task_queue_unittest.cc +++ b/webrtc/base/task_queue_unittest.cc @@ -112,14 +112,14 @@ TEST(TaskQueueTest, PostMultipleDelayed) { TaskQueue queue(kQueueName); std::vector> events; - for (int i = 0; i < 10; ++i) { + for (int i = 0; i < 100; ++i) { events.push_back(std::unique_ptr(new Event(false, false))); queue.PostDelayedTask( Bind(&CheckCurrent, kQueueName, events.back().get(), &queue), 10); } for (const auto& e : events) - EXPECT_TRUE(e->Wait(100)); + EXPECT_TRUE(e->Wait(1000)); } TEST(TaskQueueTest, PostDelayedAfterDestruct) { diff --git a/webrtc/base/task_queue_win.cc b/webrtc/base/task_queue_win.cc index 81b1cd1b9f..11aa81de03 100644 --- a/webrtc/base/task_queue_win.cc +++ b/webrtc/base/task_queue_win.cc @@ -10,8 +10,12 @@ #include "webrtc/base/task_queue.h" +#include #include +#include + +#include "webrtc/base/arraysize.h" #include "webrtc/base/checks.h" #include "webrtc/base/logging.h" @@ -29,7 +33,7 @@ BOOL CALLBACK InitializeTls(PINIT_ONCE init_once, void* param, void** context) { DWORD GetQueuePtrTls() { static INIT_ONCE init_once = INIT_ONCE_STATIC_INIT; - InitOnceExecuteOnce(&init_once, InitializeTls, nullptr, nullptr); + ::InitOnceExecuteOnce(&init_once, InitializeTls, nullptr, nullptr); return g_queue_ptr_tls; } @@ -40,13 +44,107 @@ struct ThreadStartupData { void CALLBACK InitializeQueueThread(ULONG_PTR param) { MSG msg; - PeekMessage(&msg, NULL, WM_USER, WM_USER, PM_NOREMOVE); + ::PeekMessage(&msg, nullptr, WM_USER, WM_USER, PM_NOREMOVE); ThreadStartupData* data = reinterpret_cast(param); - TlsSetValue(GetQueuePtrTls(), data->thread_context); + ::TlsSetValue(GetQueuePtrTls(), data->thread_context); data->started->Set(); } } // namespace +class TaskQueue::MultimediaTimer { + public: + // kMaxTimers defines the limit of how many MultimediaTimer instances should + // be created. + // Background: The maximum number of supported handles for Wait functions, is + // MAXIMUM_WAIT_OBJECTS - 1 (63). + // There are some ways to work around the limitation but as it turns out, the + // limit of concurrently active multimedia timers per process, is much lower, + // or 16. So there isn't much value in going to the lenghts required to + // overcome the Wait limitations. + // kMaxTimers is larger than 16 though since it is possible that 'complete' or + // signaled timers that haven't been handled, are counted as part of + // kMaxTimers and thus a multimedia timer can actually be queued even though + // as far as we're concerned, there are more than 16 that are pending. + static const int kMaxTimers = MAXIMUM_WAIT_OBJECTS - 1; + + // Controls how many MultimediaTimer instances a queue can hold before + // attempting to garbage collect (GC) timers that aren't in use. + static const int kInstanceThresholdGC = 8; + + MultimediaTimer() : event_(::CreateEvent(nullptr, false, false, nullptr)) {} + + MultimediaTimer(MultimediaTimer&& timer) + : event_(timer.event_), + timer_id_(timer.timer_id_), + task_(std::move(timer.task_)) { + RTC_DCHECK(event_); + timer.event_ = nullptr; + timer.timer_id_ = 0; + } + + ~MultimediaTimer() { Close(); } + + // Implementing this operator is required because of the way + // some stl algorithms work, such as std::rotate(). + MultimediaTimer& operator=(MultimediaTimer&& timer) { + if (this != &timer) { + Close(); + event_ = timer.event_; + timer.event_ = nullptr; + task_ = std::move(timer.task_); + timer_id_ = timer.timer_id_; + timer.timer_id_ = 0; + } + return *this; + } + + bool StartOneShotTimer(std::unique_ptr task, UINT delay_ms) { + RTC_DCHECK_EQ(0, timer_id_); + RTC_DCHECK(event_ != nullptr); + RTC_DCHECK(!task_.get()); + RTC_DCHECK(task.get()); + task_ = std::move(task); + timer_id_ = + ::timeSetEvent(delay_ms, 0, reinterpret_cast(event_), 0, + TIME_ONESHOT | TIME_CALLBACK_EVENT_SET); + return timer_id_ != 0; + } + + std::unique_ptr Cancel() { + if (timer_id_) { + ::timeKillEvent(timer_id_); + timer_id_ = 0; + } + return std::move(task_); + } + + void OnEventSignaled() { + RTC_DCHECK_NE(0, timer_id_); + timer_id_ = 0; + task_->Run() ? task_.reset() : static_cast(task_.release()); + } + + HANDLE event() const { return event_; } + + bool is_active() const { return timer_id_ != 0; } + + private: + void Close() { + Cancel(); + + if (event_) { + ::CloseHandle(event_); + event_ = nullptr; + } + } + + HANDLE event_ = nullptr; + MMRESULT timer_id_ = 0; + std::unique_ptr task_; + + RTC_DISALLOW_COPY_AND_ASSIGN(MultimediaTimer); +}; + TaskQueue::TaskQueue(const char* queue_name) : thread_(&TaskQueue::ThreadMain, this, queue_name) { RTC_DCHECK(queue_name); @@ -60,7 +158,7 @@ TaskQueue::TaskQueue(const char* queue_name) TaskQueue::~TaskQueue() { RTC_DCHECK(!IsCurrent()); - while (!PostThreadMessage(thread_.GetThreadRef(), WM_QUIT, 0, 0)) { + while (!::PostThreadMessage(thread_.GetThreadRef(), WM_QUIT, 0, 0)) { RTC_CHECK_EQ(ERROR_NOT_ENOUGH_QUOTA, ::GetLastError()); Sleep(1); } @@ -69,7 +167,7 @@ TaskQueue::~TaskQueue() { // static TaskQueue* TaskQueue::Current() { - return static_cast(TlsGetValue(GetQueuePtrTls())); + return static_cast(::TlsGetValue(GetQueuePtrTls())); } // static @@ -83,8 +181,8 @@ bool TaskQueue::IsCurrent() const { } void TaskQueue::PostTask(std::unique_ptr task) { - if (PostThreadMessage(thread_.GetThreadRef(), WM_RUN_TASK, 0, - reinterpret_cast(task.get()))) { + if (::PostThreadMessage(thread_.GetThreadRef(), WM_RUN_TASK, 0, + reinterpret_cast(task.get()))) { task.release(); } } @@ -100,8 +198,8 @@ void TaskQueue::PostDelayedTask(std::unique_ptr task, #else wparam = milliseconds; #endif - if (PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam, - reinterpret_cast(task.get()))) { + if (::PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam, + reinterpret_cast(task.get()))) { task.release(); } } @@ -117,8 +215,8 @@ void TaskQueue::PostTaskAndReply(std::unique_ptr task, delete task_ptr; // If the thread's message queue is full, we can't queue the task and will // have to drop it (i.e. delete). - if (!PostThreadMessage(reply_thread_id, WM_RUN_TASK, 0, - reinterpret_cast(reply_task_ptr))) { + if (!::PostThreadMessage(reply_thread_id, WM_RUN_TASK, 0, + reinterpret_cast(reply_task_ptr))) { delete reply_task_ptr; } }); @@ -131,25 +229,69 @@ void TaskQueue::PostTaskAndReply(std::unique_ptr task, // static bool TaskQueue::ThreadMain(void* context) { + HANDLE timer_handles[MultimediaTimer::kMaxTimers]; + // Active multimedia timers. + std::vector mm_timers; + // Tasks that have been queued by using SetTimer/WM_TIMER. DelayedTasks delayed_tasks; + while (true) { - DWORD result = ::MsgWaitForMultipleObjectsEx(0, nullptr, INFINITE, + RTC_DCHECK(mm_timers.size() <= arraysize(timer_handles)); + DWORD count = 0; + for (const auto& t : mm_timers) { + if (!t.is_active()) + break; + timer_handles[count++] = t.event(); + } + // Make sure we do an alertable wait as that's required to allow APCs to run + // (e.g. required for InitializeQueueThread and stopping the thread in + // PlatformThread). + DWORD result = ::MsgWaitForMultipleObjectsEx(count, timer_handles, INFINITE, QS_ALLEVENTS, MWMO_ALERTABLE); RTC_CHECK_NE(WAIT_FAILED, result); - if (result == WAIT_OBJECT_0) { - if (!ProcessQueuedMessages(&delayed_tasks)) + // If we're not waiting for any timers, then count will be equal to + // WAIT_OBJECT_0. If we're waiting for timers, then |count| represents + // "One more than the number of timers", which means that there's a + // message in the queue that needs to be handled. + // If |result| is less than |count|, then its value will be the index of the + // timer that has been signaled. + if (result == (WAIT_OBJECT_0 + count)) { + if (!ProcessQueuedMessages(&delayed_tasks, &mm_timers)) break; + } else if (result < (WAIT_OBJECT_0 + count)) { + mm_timers[result].OnEventSignaled(); + RTC_DCHECK(!mm_timers[result].is_active()); + // Reuse timer events by moving inactive timers to the back of the vector. + // When new delayed tasks are queued, they'll get reused. + if (mm_timers.size() > 1) { + auto it = mm_timers.begin() + result; + std::rotate(it, it + 1, mm_timers.end()); + } + + // Collect some garbage. + if (mm_timers.size() > MultimediaTimer::kInstanceThresholdGC) { + const auto inactive = std::find_if( + mm_timers.begin(), mm_timers.end(), + [](const MultimediaTimer& t) { return !t.is_active(); }); + if (inactive != mm_timers.end()) { + // Since inactive timers are always moved to the back, we can + // safely delete all timers following the first inactive one. + mm_timers.erase(inactive, mm_timers.end()); + } + } } else { RTC_DCHECK_EQ(WAIT_IO_COMPLETION, result); } } + return false; } // static -bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) { +bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks, + std::vector* timers) { MSG msg = {}; - while (PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) && + while (::PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) && msg.message != WM_QUIT) { if (!msg.hwnd) { switch (msg.message) { @@ -160,7 +302,8 @@ bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) { break; } case WM_QUEUE_DELAYED_TASK: { - QueuedTask* task = reinterpret_cast(msg.lParam); + std::unique_ptr task( + reinterpret_cast(msg.lParam)); uint32_t milliseconds = msg.wParam & 0xFFFFFFFF; #if defined(_WIN64) // Subtract the time it took to queue the timer. @@ -169,12 +312,38 @@ bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) { milliseconds = post_time > milliseconds ? 0 : milliseconds - post_time; #endif - UINT_PTR timer_id = SetTimer(nullptr, 0, milliseconds, nullptr); - delayed_tasks->insert(std::make_pair(timer_id, task)); + bool timer_queued = false; + if (timers->size() < MultimediaTimer::kMaxTimers) { + MultimediaTimer* timer = nullptr; + auto available = std::find_if( + timers->begin(), timers->end(), + [](const MultimediaTimer& t) { return !t.is_active(); }); + if (available != timers->end()) { + timer = &(*available); + } else { + timers->emplace_back(); + timer = &timers->back(); + } + + timer_queued = + timer->StartOneShotTimer(std::move(task), milliseconds); + if (!timer_queued) { + // No more multimedia timers can be queued. + // Detach the task and fall back on SetTimer. + task = timer->Cancel(); + } + } + + // When we fail to use multimedia timers, we fall back on the more + // coarse SetTimer/WM_TIMER approach. + if (!timer_queued) { + UINT_PTR timer_id = ::SetTimer(nullptr, 0, milliseconds, nullptr); + delayed_tasks->insert(std::make_pair(timer_id, task.release())); + } break; } case WM_TIMER: { - KillTimer(nullptr, msg.wParam); + ::KillTimer(nullptr, msg.wParam); auto found = delayed_tasks->find(msg.wParam); RTC_DCHECK(found != delayed_tasks->end()); if (!found->second->Run()) @@ -187,8 +356,8 @@ bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) { break; } } else { - TranslateMessage(&msg); - DispatchMessage(&msg); + ::TranslateMessage(&msg); + ::DispatchMessage(&msg); } } return msg.message != WM_QUIT;