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);

[the sun, 2017/02/16 12:07:25]

nit: Would be nice to use global scope for all windows functions.

[tommi, 2017/02/16 17:38:47]

I was debating that since that's what I'm actually used to. I checked some of
the other code and based on that, removed the ::, however, since that would also
help you read the code, I changed the whole file.

   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 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 larger than 16 though since it is possible that 'complete' or

[the sun, 2017/02/16 12:07:25]

nit: Odd start of sentence

[tommi, 2017/02/16 17:38:47]

ah, there was a missing 'is'. fixed.

+  // 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() {

[the sun, 2017/02/16 12:07:25]

make private

[tommi, 2017/02/16 17:38:47]

Done.

+    Cancel();
+
+    if (event_) {
+      CloseHandle(event_);
+      event_ = nullptr;
+    }
+  }
+
+  bool StartOneShotTimer(std::unique_ptr<QueuedTask> task, UINT delay_ms) {
+    RTC_DCHECK(timer_id_ == 0);

[the sun, 2017/02/16 12:07:25]

nit: _EQ, _NE is what's preferred (though personally I think this is more
readable).

[tommi, 2017/02/16 17:38:47]

Done.  I couldn't do it in a couple of cases since the template code behind
RTC_DCHECK_NE goes berserk if trying to compare things like event_ to nullptr or
stl iterators.

+    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(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);

[the sun, 2017/02/16 12:07:25]

Ah, so we need MWMO_ALTERTABLE for the stop event in PlatformThread. That may be
worth a comment.

[tommi, 2017/02/16 17:38:47]

Done. It's also needed for InitializeQueueThread and might also come in handy if
we decide to go for a different timer approach in the future, such as waitable
timers.

     RTC_CHECK_NE(WAIT_FAILED, result);
-    if (result == WAIT_OBJECT_0) {
-      if (!ProcessQueuedMessages(&delayed_tasks))
+    if (result == count) {

[the sun, 2017/02/16 12:07:25]

I'd rather spell out WAIT_OBJECT_0 when checking the result.

[tommi, 2017/02/16 17:38:47]

|count| will only be equal to WAIT_OBJECT_0 if count is 0. However, if there are
timers that we wait on, |count| will be higher than WAIT_OBJECT_0.  I'll add a
comment.

[the sun, 2017/02/16 20:18:47]

I was thinking about being overly clear to the casual reader, with something
like:

if (result == (WAIT_OBJECT_0 + count)) {

and below

} else if (result < (WAIT_OBJECT_0 + count)) {

since not everyone reading the docs for MsgWaitForMultipleObjectsEx() will
immediately know that WAIT_OBJECT_0 == 0 (although they will for sure suspect
it).

(The comment is great, but just adding this extra icing makes the code
self-documented IMO. It's pretty obvious to the reader something special is
going on and so consulting API docs would be the next step...)

[tommi, 2017/02/17 09:43:57]

Ah, yes, that makes sense.

+      if (!ProcessQueuedMessages(&delayed_tasks, &mm_timers))
         break;
+    } else if (result < count) {

[the sun, 2017/02/16 12:07:25]

and here

[tommi, 2017/02/16 17:38:47]

Same here.  The purpose of this check is to make sure that we didn't get an
error. If |result| is between 0 and |count| (it can't be negative), we know a
timer has been fired.  If it's equal or higher than |count|, then an error has
occurred (error values are positive integers too, e.g. WAIT_IO_COMPLETION is
0xC0).  I added info to the above comment for this too.

+      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) {

[the sun, 2017/02/16 12:07:25]

Add a constant for the "8" inside MultimediaTimer::

[tommi, 2017/02/16 17:38:47]

Done.

+        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();

[the sun, 2017/02/16 12:07:25]

What's the overhead of recreating timers (CreateEvent/CloseHandle,
specifically). Is it easier to just have a pool of them than to figure out
potential lag in queueing timers?

[tommi, 2017/02/16 17:38:47]

CreateEvent creates a kernel object, so the cost can be noticeable (a few ms)
when the machine is busy.  They're also a limited resource, so while we do have
a pool here (unused ones are garbage collected when we have over 8 allocated),
it's not a big pool (for some definition of big).

[the sun, 2017/02/16 20:18:47]

Out of curiosity, what's the limit? The only limit I can quickly find, is 16M
(2^24) HANDLEs (I trust Russinovich on this stuff). Are there other limits per
kernel object? Not sure what other costs are associated with events.

[tommi, 2017/02/17 09:43:57]

That's probably correct, given the source :)  I don't know the exact cost
associated, but some things that come to mind that are different from a regular
in-process event-like object, would be bookkeeping overhead since events are
cross-process-capable synchronization objects (events can be named and shared
with other processes, duplicated etc), associated security descriptor, even a
stack trace and various other info as you probably saw when looking at Mark's
data (was it the 'testlimit article' you read?).

[the sun, 2017/02/17 10:50:14]

Yeah that was it.

+              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) {

[the sun, 2017/02/16 12:07:25]

Do we need logging when this happens? If I'm doing things that require high
precision, and I can't actually trust the task queue to provide it, I need to at
least know when it is taking me for a ride.

[tommi, 2017/02/16 17:38:47]

That sounds useful so I've added warning log.  However...

The LOG IO is so expensive that it caused the Wait(100) test in
PostMultipleDelayed, to start failing on my machine :-/ (powerful desktop)

The total time of running that test went from 11ms to 428ms!

I wonder if I should not add that log statement for now :-/

[the sun, 2017/02/16 20:18:47]

Sounds like the best option. :/ Can you add appropriate commentary in
task_queue.h on what to expect from timing?

[tommi, 2017/02/17 09:43:57]

Done.

+            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