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}
diff --git a/webrtc/base/task_queue.h b/webrtc/base/task_queue.h
index eeabe05..a6a7036 100644
--- a/webrtc/base/task_queue.h
+++ b/webrtc/base/task_queue.h
@@ -178,6 +178,11 @@
void PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> 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<QueuedTask> task, uint32_t milliseconds);
template <class Closure>
@@ -185,6 +190,7 @@
PostTask(std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure)));
}
+ // See documentation above for performance expectations.
template <class Closure>
void PostDelayedTask(const Closure& closure, uint32_t milliseconds) {
PostDelayedTask(
@@ -254,10 +260,12 @@
dispatch_queue_t queue_;
QueueContext* const context_;
#elif defined(WEBRTC_WIN)
+ class MultimediaTimer;
typedef std::unordered_map<UINT_PTR, std::unique_ptr<QueuedTask>>
DelayedTasks;
static bool ThreadMain(void* context);
- static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks);
+ static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks,
+ std::vector<MultimediaTimer>* timers);
class WorkerThread : public PlatformThread {
public:
diff --git a/webrtc/base/task_queue_unittest.cc b/webrtc/base/task_queue_unittest.cc
index 08123d0..74433b9 100644
--- a/webrtc/base/task_queue_unittest.cc
+++ b/webrtc/base/task_queue_unittest.cc
@@ -112,14 +112,14 @@
TaskQueue queue(kQueueName);
std::vector<std::unique_ptr<Event>> events;
- for (int i = 0; i < 10; ++i) {
+ for (int i = 0; i < 100; ++i) {
events.push_back(std::unique_ptr<Event>(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 81b1cd1..11aa81d 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 <mmsystem.h>
#include <string.h>
+#include <algorithm>
+
+#include "webrtc/base/arraysize.h"
#include "webrtc/base/checks.h"
#include "webrtc/base/logging.h"
@@ -29,7 +33,7 @@
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 @@
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<ThreadStartupData*>(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<QueuedTask> 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<LPTIMECALLBACK>(event_), 0,
+ TIME_ONESHOT | TIME_CALLBACK_EVENT_SET);
+ return timer_id_ != 0;
+ }
+
+ std::unique_ptr<QueuedTask> 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<void>(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<QueuedTask> 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() {
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 @@
// static
TaskQueue* TaskQueue::Current() {
- return static_cast<TaskQueue*>(TlsGetValue(GetQueuePtrTls()));
+ return static_cast<TaskQueue*>(::TlsGetValue(GetQueuePtrTls()));
}
// static
@@ -83,8 +181,8 @@
}
void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
- if (PostThreadMessage(thread_.GetThreadRef(), WM_RUN_TASK, 0,
- reinterpret_cast<LPARAM>(task.get()))) {
+ if (::PostThreadMessage(thread_.GetThreadRef(), WM_RUN_TASK, 0,
+ reinterpret_cast<LPARAM>(task.get()))) {
task.release();
}
}
@@ -100,8 +198,8 @@
#else
wparam = milliseconds;
#endif
- if (PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam,
- reinterpret_cast<LPARAM>(task.get()))) {
+ if (::PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam,
+ reinterpret_cast<LPARAM>(task.get()))) {
task.release();
}
}
@@ -117,8 +215,8 @@
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<LPARAM>(reply_task_ptr))) {
+ if (!::PostThreadMessage(reply_thread_id, WM_RUN_TASK, 0,
+ reinterpret_cast<LPARAM>(reply_task_ptr))) {
delete reply_task_ptr;
}
});
@@ -131,25 +229,69 @@
// static
bool TaskQueue::ThreadMain(void* context) {
+ HANDLE timer_handles[MultimediaTimer::kMaxTimers];
+ // Active multimedia timers.
+ std::vector<MultimediaTimer> 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<MultimediaTimer>* 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 @@
break;
}
case WM_QUEUE_DELAYED_TASK: {
- QueuedTask* task = reinterpret_cast<QueuedTask*>(msg.lParam);
+ std::unique_ptr<QueuedTask> task(
+ reinterpret_cast<QueuedTask*>(msg.lParam));
uint32_t milliseconds = msg.wParam & 0xFFFFFFFF;
#if defined(_WIN64)
// Subtract the time it took to queue the timer.
@@ -169,12 +312,38 @@
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 @@
break;
}
} else {
- TranslateMessage(&msg);
- DispatchMessage(&msg);
+ ::TranslateMessage(&msg);
+ ::DispatchMessage(&msg);
}
}
return msg.message != WM_QUIT;
