New task queueing primitive for async tasks: TaskQueue.

webrtc/base/task_queue_win.cc

Index: webrtc/base/task_queue_win.cc
diff --git a/webrtc/base/task_queue_win.cc b/webrtc/base/task_queue_win.cc
new file mode 100644
index 0000000000000000000000000000000000000000..35c8daadabb788041e46cc1619102e1a49bfc7a2
--- /dev/null
+++ b/webrtc/base/task_queue_win.cc
@@ -0,0 +1,182 @@
+/*
+ *  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 <string.h>
+#include <unordered_map>
+
+#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);

[perkj_webrtc, 2016/04/28 09:58:32]

Why PeekMessage here? 

[tommi, 2016/04/28 12:04:02]

Because no message has actually been posted.  GetMessage would hang
indefinitely.  The purpose of this call is just to create the queue on the
thread and then let the starting thread know.

+  ThreadStartupData* data = reinterpret_cast<ThreadStartupData*>(param);
+  TlsSetValue(GetQueuePtrTls(), data->thread_context);
+  data->started->Set();
+}
+}  // namespace
+
+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)) {
+    RTC_CHECK(ERROR_NOT_ENOUGH_QUOTA == ::GetLastError());
+    Sleep(1);
+  }
+  thread_.Stop();
+}
+
+// static
+TaskQueue* TaskQueue::Current() {
+  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()))) {
+    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()))) {
+    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 (!PostThreadMessage(reply_thread_id, WM_RUN_TASK, 0,

[perkj_webrtc, 2016/04/28 09:58:32]

Please comment !PostThreadMessage return true.

[tommi, 2016/04/28 12:04:02]

Done.

+                           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) {
+  std::unordered_map<UINT_PTR, std::unique_ptr<QueuedTask>> delayed_tasks;
+
+  BOOL ret;
+  MSG msg;
+
+  while ((ret = GetMessage(&msg, nullptr, 0, 0)) != 0 && ret != -1) {
+    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);
+          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));
+          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 false;
+}
+}  // namespace rtc