Refactor Windows TaskQueue code to only need a single high res timer.

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 <set>
 
 #include "webrtc/base/arraysize.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
+#include "webrtc/base/timeutils.h"
 
 namespace rtc {
 namespace {
 #define WM_RUN_TASK WM_USER + 1
 #define WM_QUEUE_DELAYED_TASK WM_USER + 2
 
 using Priority = TaskQueue::Priority;
 
 DWORD g_queue_ptr_tls = 0;
 
@@ skipping 29 matching lines @@
       return kLowPriority;
     case Priority::NORMAL:
       return kNormalPriority;
     default:
       RTC_NOTREACHED();
       break;
   }
   return kNormalPriority;
 }
 
-#if defined(_WIN64)
-DWORD GetTick() {
+int64_t GetTick() {
   static const UINT kPeriod = 1;
   bool high_res = (timeBeginPeriod(kPeriod) == TIMERR_NOERROR);
-  DWORD ret = timeGetTime();
+  int64_t ret = TimeMillis();
   if (high_res)
     timeEndPeriod(kPeriod);
   return ret;
 }
-#endif
-}  // 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;
+struct DelayedTaskInfo {
+  DelayedTaskInfo(uint32_t milliseconds, std::unique_ptr<QueuedTask> task)
+      : due_time(GetTick() + milliseconds), task(std::move(task)) {}
+  DelayedTaskInfo(const DelayedTaskInfo&) = delete;
+  DelayedTaskInfo(DelayedTaskInfo&&) = default;

[the sun, 2017/03/07 11:24:24]

Maybe delete operator= too?

[tommi, 2017/03/09 20:27:42]

Done. (added RTC_DISALLOW_COPY_AND_ASSIGN)

[tommi, 2017/03/09 20:29:08]

oops, discard that. I had to support the operator and default ctor after
switching to priority_queue (due to pop()).

 
-  // 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;
+  // Implement for <set>. to maintain an order of increasing |due_time|.

[the sun, 2017/03/07 11:24:24]

nit: , after <set>

[tommi, 2017/03/09 20:27:42]

Done.

+  bool operator<(const DelayedTaskInfo& other) const {
+    return due_time < other.due_time;
   }
 
-  ~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;
+  // See below for why this method is const.
+  void Run() const {
+    task->Run() ? task.reset() : static_cast<void>(task.release());
   }
 
-  bool StartOneShotTimer(std::unique_ptr<QueuedTask> task, UINT delay_ms) {
+  const int64_t due_time;  // Absolute timestamp in milliseconds.
+
+ private:
+  // |task| needs to be mutable because of an issue with std::multiset whereby
+  // set::begin() returns a const_iterator by default.
+  // There are two basic workarounds, one using const_cast, which would also
+  // make the key (|due_time|), non-const and the other is to make the non-key
+  // (|task|), mutable.
+  // Because of this, the |task| variable is made private and can only be
+  // mutated by calling the |Run()| method.
+  mutable std::unique_ptr<QueuedTask> task;
+};
+
+class MultimediaTimer {
+ public:
+  MultimediaTimer() : event_(::CreateEvent(nullptr, false, false, nullptr)) {}
+  MultimediaTimer(MultimediaTimer&&) = delete;

[the sun, 2017/03/07 11:24:24]

Strictly speaking I don't think this is needed:
http://en.cppreference.com/w/cpp/language/move_constructor

...but I don't mind being overly clear. However, then it seems operator=(&&)
should be explicitly deleted as well.

[tommi, 2017/03/09 20:27:42]

Removed. I only needed it as I was changing the previous code and wanted to make
sure we didn't require this somewhere.

+
+  ~MultimediaTimer() {
+    Cancel();
+    ::CloseHandle(event_);
+  }
+
+  bool StartOneShotTimer(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() {
+  void 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; }
+  HANDLE* event_for_wait() { return &event_; }
 
  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);
 };
 
+}  // namespace
+
+class TaskQueue::ThreadState {
+ public:
+  ThreadState() {}
+  ~ThreadState() {}
+
+  void RunThreadMain();
+
+ private:
+  bool ProcessQueuedMessages();
+  void RunDueTasks();
+  void ScheduleNextTimer();
+  void CancelTimers();
+
+  MultimediaTimer timer_;
+  std::multiset<DelayedTaskInfo> timer_tasks_;

[the sun, 2017/03/07 11:24:24]

It's not clear to me that multiset is the best structure here. For a small set
of pending tasks, wouldn't it be outperformed by a priority_queue/heap (which
would over time avoid heap allocs)? The devil is in the profile data...

[tommi, 2017/03/09 20:27:42]

Done.

+  UINT_PTR timer_id_ = 0;
+};
+
 TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
     : thread_(&TaskQueue::ThreadMain,
               this,
               queue_name,
               TaskQueuePriorityToThreadPriority(priority)) {
   RTC_DCHECK(queue_name);
   thread_.Start();
   Event event(false, false);
   ThreadStartupData startup = {&event, this};
   RTC_CHECK(thread_.QueueAPC(&InitializeQueueThread,
@@ skipping 27 matching lines @@
 
 void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
   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>(GetTick()) << 32) | milliseconds;
-#else
-  wparam = milliseconds;
-#endif
-  if (::PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam,
-                          reinterpret_cast<LPARAM>(task.get()))) {
-    task.release();
+  if (!milliseconds) {
+    PostTask(std::move(task));
+    return;
+  }
+
+  auto* task_info = new DelayedTaskInfo(milliseconds, std::move(task));

[the sun, 2017/03/07 11:24:24]

Add a TODO that we should remove this heap alloc.

[tommi, 2017/03/09 20:27:42]

Done.

+  if (!::PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, 0,
+                           reinterpret_cast<LPARAM>(task_info))) {
+    delete task_info;
   }
 }
 
 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))) {
       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
 void 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;
+  ThreadState state;
+  state.RunThreadMain();
+}
 
+void TaskQueue::ThreadState::RunThreadMain() {
   while (true) {
-    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);
+    DWORD result = ::MsgWaitForMultipleObjectsEx(
+        1, timer_.event_for_wait(), INFINITE, QS_ALLEVENTS, MWMO_ALERTABLE);
     RTC_CHECK_NE(WAIT_FAILED, result);
-    // 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))
+    if (result == (WAIT_OBJECT_0 + 1)) {
+      // There are messages in the message queue that need to be handled.
+      if (!ProcessQueuedMessages())
         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 if (result == WAIT_OBJECT_0) {
+      // The multimedia timer was signaled.
+      timer_.Cancel();
+      RTC_DCHECK(!timer_tasks_.empty());
+      RunDueTasks();
+      ScheduleNextTimer();
     } else {
       RTC_DCHECK_EQ(WAIT_IO_COMPLETION, result);
     }
   }
 }
 
-// static
-bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks,
-                                      std::vector<MultimediaTimer>* timers) {
+bool TaskQueue::ThreadState::ProcessQueuedMessages() {
   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: {
-          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 = GetTick();
-          DWORD post_time = now - (msg.wParam >> 32);
-          milliseconds =
-              post_time > milliseconds ? 0 : milliseconds - post_time;
-#endif
-          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()));
+          std::unique_ptr<DelayedTaskInfo> info(
+              reinterpret_cast<DelayedTaskInfo*>(msg.lParam));
+          auto inserted = timer_tasks_.insert(std::move(*info.get()));
+          // If the new task was not added to the front, we're done since a
+          // timer has already been scheduled.
+          if (inserted == timer_tasks_.begin()) {
+            CancelTimers();
+            ScheduleNextTimer();
           }
           break;
         }
         case WM_TIMER: {
+          RTC_DCHECK_EQ(timer_id_, 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);
+          timer_id_ = 0;
+          RunDueTasks();
+          ScheduleNextTimer();
           break;
         }
         default:
           RTC_NOTREACHED();
           break;
       }
     } else {
       ::TranslateMessage(&msg);
       ::DispatchMessage(&msg);
     }
   }
   return msg.message != WM_QUIT;
 }
 
+void TaskQueue::ThreadState::RunDueTasks() {
+  std::multiset<DelayedTaskInfo>::iterator it = timer_tasks_.begin();
+  do {
+    if ((*it).due_time > GetTick())
+      break;
+    (*it).Run();
+    ++it;
+  } while (it != timer_tasks_.end());
+
+  timer_tasks_.erase(timer_tasks_.begin(), it);

[the sun, 2017/03/07 11:24:24]

I assume nothing will be erased from the set if both arguments are at .begin(),
but docs I found are ambiguous.

[tommi, 2017/03/09 20:27:42]

Yes - however, the implementation has changed slightly after switching over to
priority_queue. I did add an RTC_DCHECK() at the top to show that the queue
should never be empty when the function is called.

+}
+
+void TaskQueue::ThreadState::ScheduleNextTimer() {
+  RTC_DCHECK_EQ(timer_id_, 0);
+  if (timer_tasks_.empty())
+    return;
+
+  uint32_t milliseconds =
+      static_cast<uint32_t>(timer_tasks_.begin()->due_time - GetTick());

[the sun, 2017/03/07 11:24:24]

It seems to me you need to manage the case when we get a negative difference
here. It's not the same GetTick() call, and result, that you got when purging
due tasks.

[tommi, 2017/03/09 20:27:42]

Good catch. Done.

+  if (!timer_.StartOneShotTimer(milliseconds))
+    timer_id_ = ::SetTimer(nullptr, 0, milliseconds, nullptr);
+}
+
+void TaskQueue::ThreadState::CancelTimers() {
+  timer_.Cancel();
+  if (timer_id_) {
+    ::KillTimer(nullptr, timer_id_);
+    timer_id_ = 0;
+  }
+}
+
 }  // namespace rtc