TaskQueue[Win] DOS handling

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 <queue>
 
+#include "webrtc/base/arraysize.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/safe_conversions.h"
 #include "webrtc/base/timeutils.h"
 
 namespace rtc {
 namespace {
 #define WM_RUN_TASK WM_USER + 1
 #define WM_QUEUE_DELAYED_TASK WM_USER + 2
 
@@ skipping 81 matching lines @@
   // 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)) {}
+  // Note: We create an event that requires manual reset.
+  MultimediaTimer() : event_(::CreateEvent(nullptr, true, false, nullptr)) {}
 
   ~MultimediaTimer() {
     Cancel();
     ::CloseHandle(event_);
   }
 
   bool StartOneShotTimer(UINT delay_ms) {
     RTC_DCHECK_EQ(0, timer_id_);
     RTC_DCHECK(event_ != nullptr);
     timer_id_ =
         ::timeSetEvent(delay_ms, 0, reinterpret_cast<LPTIMECALLBACK>(event_), 0,
                        TIME_ONESHOT | TIME_CALLBACK_EVENT_SET);
     return timer_id_ != 0;
   }
 
   void Cancel() {
+    ::ResetEvent(event_);
     if (timer_id_) {
       ::timeKillEvent(timer_id_);
       timer_id_ = 0;
     }
   }
 
   HANDLE* event_for_wait() { return &event_; }
 
  private:
   HANDLE event_ = nullptr;
   MMRESULT timer_id_ = 0;
 
   RTC_DISALLOW_COPY_AND_ASSIGN(MultimediaTimer);
 };
 
 }  // namespace
 
 class TaskQueue::ThreadState {
  public:
-  ThreadState() {}
+  explicit ThreadState(HANDLE in_queue) : in_queue_(in_queue) {}
   ~ThreadState() {}
 
   void RunThreadMain();
 
  private:
   bool ProcessQueuedMessages();
   void RunDueTasks();
   void ScheduleNextTimer();
   void CancelTimers();
 
   // Since priority_queue<> by defult orders items in terms of
   // largest->smallest, using std::less<>, and we want smallest->largest,
   // we would like to use std::greater<> here. Alas it's only available in
   // C++14 and later, so we roll our own compare template that that relies on
   // operator<().
   template <typename T>
   struct greater {
     bool operator()(const T& l, const T& r) { return l > r; }
   };
 
   MultimediaTimer timer_;
   std::priority_queue<DelayedTaskInfo,
                       std::vector<DelayedTaskInfo>,
                       greater<DelayedTaskInfo>>
       timer_tasks_;
   UINT_PTR timer_id_ = 0;
+  HANDLE in_queue_;
 };
 
 TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
     : thread_(&TaskQueue::ThreadMain,
               this,
               queue_name,
-              TaskQueuePriorityToThreadPriority(priority)) {
+              TaskQueuePriorityToThreadPriority(priority)),
+       in_queue_(::CreateEvent(nullptr, true, false, nullptr)) {
   RTC_DCHECK(queue_name);
+  RTC_DCHECK(in_queue_);
   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_EQ(ERROR_NOT_ENOUGH_QUOTA, ::GetLastError());
     Sleep(1);
   }
   thread_.Stop();
+  ::CloseHandle(in_queue_);
 }
 
 // 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();
-  }
+  rtc::CritScope lock(&pending_lock_);
+  pending_.push(std::move(task));
+  ::SetEvent(in_queue_);
 }
 
 void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
                                 uint32_t milliseconds) {
   if (!milliseconds) {
     PostTask(std::move(task));
     return;
   }
 
   // TODO(tommi): Avoid this allocation.  It is currently here since
@@ skipping 23 matching lines @@
       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());
 }
 
+void TaskQueue::RunPendingTasks() {
+  while (true) {
+    std::unique_ptr<QueuedTask> task;
+    {
+      rtc::CritScope lock(&pending_lock_);
+      if (pending_.empty())
+        break;
+      task = std::move(pending_.front());
+      pending_.pop();
+    }
+
+    if (!task->Run())
+      task.release();
+  }
+}
+
 // static
 void TaskQueue::ThreadMain(void* context) {
-  ThreadState state;
+  ThreadState state(static_cast<TaskQueue*>(context)->in_queue_);
   state.RunThreadMain();
 }
 
 void TaskQueue::ThreadState::RunThreadMain() {
+  HANDLE handles[2] = { *timer_.event_for_wait(), in_queue_ };
   while (true) {
     // 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(
-        1, timer_.event_for_wait(), INFINITE, QS_ALLEVENTS, MWMO_ALERTABLE);
+        arraysize(handles), handles, INFINITE, QS_ALLEVENTS, MWMO_ALERTABLE);
     RTC_CHECK_NE(WAIT_FAILED, result);
-    if (result == (WAIT_OBJECT_0 + 1)) {
+    if (result == (WAIT_OBJECT_0 + 2)) {
       // There are messages in the message queue that need to be handled.
       if (!ProcessQueuedMessages())
         break;
-    } else if (result == WAIT_OBJECT_0) {
+    }
+
+    if (result == WAIT_OBJECT_0 || (!timer_tasks_.empty() &&
+        ::WaitForSingleObject(*timer_.event_for_wait(), 0) == 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);
+    }
+
+    if (result == (WAIT_OBJECT_0 + 1)) {
+      ::ResetEvent(in_queue_);
+      TaskQueue::Current()->RunPendingTasks();
     }
   }
 }
 
 bool TaskQueue::ThreadState::ProcessQueuedMessages() {
   MSG msg = {};
+  // To protect against overly busy message queues, we limit the time
+  // we process tasks to a few milliseconds. If we don't do that, there's
+  // a chance that timer tasks won't ever run.
+  static const int kMaxTaskProcessingTimeMs = 500;
+  auto start = GetTick();
   while (::PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) &&
          msg.message != WM_QUIT) {
     if (!msg.hwnd) {
       switch (msg.message) {
+        // TODO(tommi): Stop using this way of queueing tasks.
         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<DelayedTaskInfo> info(
               reinterpret_cast<DelayedTaskInfo*>(msg.lParam));
           bool need_to_schedule_timers =
@@ skipping 15 matching lines @@
           break;
         }
         default:
           RTC_NOTREACHED();
           break;
       }
     } else {
       ::TranslateMessage(&msg);
       ::DispatchMessage(&msg);
     }
+
+    if (GetTick() > start + kMaxTaskProcessingTimeMs)
+      break;
   }
   return msg.message != WM_QUIT;
 }
 
 void TaskQueue::ThreadState::RunDueTasks() {
   RTC_DCHECK(!timer_tasks_.empty());
   auto now = GetTick();
   do {
     const auto& top = timer_tasks_.top();
     if (top.due_time() > now)
@@ skipping 17 matching lines @@
 
 void TaskQueue::ThreadState::CancelTimers() {
   timer_.Cancel();
   if (timer_id_) {
     ::KillTimer(nullptr, timer_id_);
     timer_id_ = 0;
   }
 }
 
 }  // namespace rtc