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);
   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);
   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, is much lower, or 16.

[the sun, 2017/02/14 10:29:19]

Is the limit per process or across the system?

[tommi, 2017/02/14 10:43:07]

per process, comment updated.

+  // So there isn't much value in going to the lenghts required to overcome the
+  // Wait limitations.
+  // kMaxTimers 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;
+
+  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;
+  }
+
+  void Close() {
+    Cancel();
+
+    if (event_) {
+      CloseHandle(event_);
+      event_ = nullptr;
+    }
+  }
+
+  bool StartOneShotTimer(std::unique_ptr<QueuedTask> task, UINT delay_ms) {
+    RTC_DCHECK(timer_id_ == 0);
+    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,

[the sun, 2017/02/14 10:29:19]

MSDN says timeSetEvent() is obsolete (since XP/Server 2003) and that TimerQueues
should be used instead:
https://msdn.microsoft.com/en-us/library/dd757634(v=vs.85).aspx

and
https://msdn.microsoft.com/en-us/library/ms686796(v=vs.85).aspx

[tommi, 2017/02/14 10:43:07]

Yes. I looked into the various options and measurements.  As for example in the
source you linked to, the accuracy of the alternatives, is not as good as this
approach.  So I opted for picking the approach that we currently use in WebRTC. 
My thinking was to pick the approach that would be least likely to cause
regressions as we migrate over to using TaskQueue.

Moving forward, I think it's entirely reasonable that we rethink this and
evaluate the alternatives again.  There might be some ways to get the same
accuracy as timeSetEvent offers by using other APIs, since typically when
Microsoft deprecates an API, it builds the old one on top of the new one (e.g.
on newer versions of Windows that support CoreAudio, the Wave API was
re-implemented on top of CoreAudio for sake of compatibility).  Figuring that
out and digging deeper into performance and understanding what accuracy we
actually need, will require some time spent on investigating.

+                     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(timer_id_ != 0);
+    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:
+  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);
 }
@@ skipping 64 matching lines @@
   });
 }
 
 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();
+    }
+    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 (result == count) {
+      if (!ProcessQueuedMessages(&delayed_tasks, &mm_timers))
         break;
+    } else if (result < 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() > 8) {
+        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) &&
          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);
           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);
     }
   }
   return msg.message != WM_QUIT;
 }
 
 }  // namespace rtc