From b89257a51b5363628e9d694584dc5a64da772809 Mon Sep 17 00:00:00 2001
From: tommi <tommi@webrtc.org>
Date: Tue, 12 Jul 2016 01:24:36 -0700
Subject: [PATCH] Handle initialization race in TaskQueue on Windows,
 discovered by drmemory. The issue was that if the main entry routine of the
 thread would get called before the first APC, the thread would hang on a
 GetMessage() call and the APC sent to initialize the thread, would never run.

BUG=

Review-Url: https://codereview.webrtc.org/2139723003
Cr-Commit-Position: refs/heads/master@{#13438}
---
 webrtc/base/task_queue.h      |  4 ++++
 webrtc/base/task_queue_win.cc | 37 +++++++++++++++++++++++------------
 2 files changed, 29 insertions(+), 12 deletions(-)

diff --git a/webrtc/base/task_queue.h b/webrtc/base/task_queue.h
index f57c6b4348..eeabe05fab 100644
--- a/webrtc/base/task_queue.h
+++ b/webrtc/base/task_queue.h
@@ -13,6 +13,7 @@
 
 #include <list>
 #include <memory>
+#include <unordered_map>
 
 #if defined(WEBRTC_MAC) && !defined(WEBRTC_BUILD_LIBEVENT)
 #include <dispatch/dispatch.h>
@@ -253,7 +254,10 @@ class LOCKABLE TaskQueue {
   dispatch_queue_t queue_;
   QueueContext* const context_;
 #elif defined(WEBRTC_WIN)
+  typedef std::unordered_map<UINT_PTR, std::unique_ptr<QueuedTask>>
+      DelayedTasks;
   static bool ThreadMain(void* context);
+  static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks);
 
   class WorkerThread : public PlatformThread {
    public:
diff --git a/webrtc/base/task_queue_win.cc b/webrtc/base/task_queue_win.cc
index e18c1f7a42..6d7ca4e6df 100644
--- a/webrtc/base/task_queue_win.cc
+++ b/webrtc/base/task_queue_win.cc
@@ -11,7 +11,6 @@
 #include "webrtc/base/task_queue.h"
 
 #include <string.h>
-#include <unordered_map>
 
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
@@ -132,12 +131,26 @@ void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
 
 // static
 bool TaskQueue::ThreadMain(void* context) {
-  std::unordered_map<UINT_PTR, std::unique_ptr<QueuedTask>> delayed_tasks;
+  DelayedTasks delayed_tasks;
+  while (true) {
+    DWORD result = ::MsgWaitForMultipleObjectsEx(0, nullptr, INFINITE,
+                                                 QS_ALLEVENTS, MWMO_ALERTABLE);
+    RTC_CHECK_NE(WAIT_FAILED, result);
+    if (result == WAIT_OBJECT_0) {
+      if (!ProcessQueuedMessages(&delayed_tasks))
+        break;
+    } else {
+      RTC_DCHECK_EQ(WAIT_IO_COMPLETION, result);
+    }
+  }
+  return false;
+}
 
-  BOOL ret;
-  MSG msg;
-
-  while ((ret = GetMessage(&msg, nullptr, 0, 0)) != 0 && ret != -1) {
+// static
+bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) {
+  MSG msg = {};
+  while (PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) &&
+         msg.message != WM_QUIT) {
     if (!msg.hwnd) {
       switch (msg.message) {
         case WM_RUN_TASK: {
@@ -157,16 +170,16 @@ bool TaskQueue::ThreadMain(void* context) {
               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));
+          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());
+          auto found = delayed_tasks->find(msg.wParam);
+          RTC_DCHECK(found != delayed_tasks->end());
           if (!found->second->Run())
             found->second.release();
-          delayed_tasks.erase(found);
+          delayed_tasks->erase(found);
           break;
         }
         default:
@@ -178,7 +191,7 @@ bool TaskQueue::ThreadMain(void* context) {
       DispatchMessage(&msg);
     }
   }
-
-  return false;
+  return msg.message != WM_QUIT;
 }
+
 }  // namespace rtc