Add support for multimedia timers to TaskQueue on Windows.

webrtc/base/task_queue_win.cc

 /*
  *  Copyright 2016 The WebRTC Project Authors. All rights reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #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"
 
 namespace rtc {
 namespace {
 #define WM_RUN_TASK WM_USER + 1
 #define WM_QUEUE_DELAYED_TASK WM_USER + 2
 
 DWORD g_queue_ptr_tls = 0;
 
 BOOL CALLBACK InitializeTls(PINIT_ONCE init_once, void* param, void** context) {
   g_queue_ptr_tls = TlsAlloc();
   return TRUE;
 }
 
 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;
 }
 
 struct ThreadStartupData {
   Event* started;
   void* thread_context;
 };
 
 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);
   thread_.Start();
   Event event(false, false);
   ThreadStartupData startup = {&event, this};
   RTC_CHECK(thread_.QueueAPC(&InitializeQueueThread,
                              reinterpret_cast<ULONG_PTR>(&startup)));
   event.Wait(Event::kForever);
 }
 
 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);
   }
   thread_.Stop();
 }
 
 // static
 TaskQueue* TaskQueue::Current() {
-  return static_cast<TaskQueue*>(TlsGetValue(GetQueuePtrTls()));
+  return static_cast<TaskQueue*>(::TlsGetValue(GetQueuePtrTls()));
 }
 
 // static
 bool TaskQueue::IsCurrent(const char* queue_name) {
   TaskQueue* current = Current();
   return current && current->thread_.name().compare(queue_name) == 0;
 }
 
 bool TaskQueue::IsCurrent() const {
   return IsThreadRefEqual(thread_.GetThreadRef(), CurrentThreadRef());
 }
 
 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();
   }
 }
 
 void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
                                 uint32_t milliseconds) {
   WPARAM wparam;
 #if defined(_WIN64)
   // GetTickCount() returns a fairly coarse tick count (resolution or about 8ms)
   // so this compensation isn't that accurate, but since we have unused 32 bits
   // on Win64, we might as well use them.
   wparam = (static_cast<WPARAM>(::GetTickCount()) << 32) | milliseconds;
 #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();
   }
 }
 
 void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
                                  std::unique_ptr<QueuedTask> reply,
                                  TaskQueue* reply_queue) {
   QueuedTask* task_ptr = task.release();
   QueuedTask* reply_task_ptr = reply.release();
   DWORD reply_thread_id = reply_queue->thread_.GetThreadRef();
   PostTask([task_ptr, reply_task_ptr, reply_thread_id]() {
     if (task_ptr->Run())
       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;
     }
   });
 }
 
 void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
                                  std::unique_ptr<QueuedTask> reply) {
   return PostTaskAndReply(std::move(task), std::move(reply), Current());
 }
 
 // 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) {
         case WM_RUN_TASK: {
           QueuedTask* task = reinterpret_cast<QueuedTask*>(msg.lParam);
           if (task->Run())
             delete task;
           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.
           const DWORD now = GetTickCount();
           DWORD post_time = now - (msg.wParam >> 32);
           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())
             found->second.release();
           delayed_tasks->erase(found);
           break;
         }
         default:
           RTC_NOTREACHED();
           break;
       }
     } else {
-      TranslateMessage(&msg);
-      DispatchMessage(&msg);
+      ::TranslateMessage(&msg);
+      ::DispatchMessage(&msg);
     }
   }
   return msg.message != WM_QUIT;
 }
 
 }  // namespace rtc