Add a new Task Queue for WinUWP.

WinUWP cannot use the win task queue as post/peek message event loop
is not available. A replacement version written using stdlib compatible
with WinUWP is added as an alternative.

Change-Id: Ie9d6e6f11f395d1815d8f04633772a0c597ed30a
Bug: webrtc:10046
Reviewed-on: https://webrtc-review.googlesource.com/c/108520
Commit-Queue: Patrik Höglund <phoglund@webrtc.org>
Reviewed-by: Niels Moller <nisse@webrtc.org>
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#25781}

rtc_base/BUILD.gn

@@ -243,6 +243,7 @@ rtc_source_set("platform_thread") {
     ":rtc_base_approved",
     ":rtc_task_queue_libevent",
     ":rtc_task_queue_win",
+    ":rtc_task_queue_stdlib",
     ":sequenced_task_checker",
   ]
   sources = [
@@ -604,6 +605,26 @@ if (is_win) {
   }
 }
 
+rtc_source_set("rtc_task_queue_stdlib") {
+  visibility = [ ":rtc_task_queue_impl" ]
+  sources = [
+    "task_queue_stdlib.cc",
+  ]
+  deps = [
+    ":checks",
+    ":criticalsection",
+    ":logging",
+    ":macromagic",
+    ":platform_thread",
+    ":ptr_util",
+    ":refcount",
+    ":rtc_event",
+    ":rtc_task_queue_api",
+    ":safe_conversions",
+    ":timeutils",
+  ]
+}
+
 rtc_source_set("rtc_task_queue_impl") {
   visibility = [ "*" ]
   if (rtc_enable_libevent) {
@@ -617,12 +638,18 @@ rtc_source_set("rtc_task_queue_impl") {
       ]
     }
     if (is_win) {
+      if (current_os == "winuwp") {
+        deps = [
+          ":rtc_task_queue_stdlib",
+        ]
+      } else {
         deps = [
           ":rtc_task_queue_win",
         ]
       }
     }
   }
+}
 
 rtc_source_set("sequenced_task_checker") {
   sources = [

rtc_base/task_queue_stdlib.cc

@@ -0,0 +1,399 @@
+/*
+ *  Copyright 2018 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 "rtc_base/task_queue.h"
+
+#include <string.h>
+#include <algorithm>
+#include <atomic>
+#include <condition_variable>
+#include <map>
+#include <queue>
+#include <utility>
+
+#include "rtc_base/checks.h"
+#include "rtc_base/criticalsection.h"
+#include "rtc_base/event.h"
+#include "rtc_base/logging.h"
+#include "rtc_base/platform_thread.h"
+#include "rtc_base/refcount.h"
+#include "rtc_base/refcountedobject.h"
+#include "rtc_base/thread_annotations.h"
+#include "rtc_base/timeutils.h"
+
+namespace rtc {
+namespace {
+
+using Priority = TaskQueue::Priority;
+
+ThreadPriority TaskQueuePriorityToThreadPriority(Priority priority) {
+  switch (priority) {
+    case Priority::HIGH:
+      return kRealtimePriority;
+    case Priority::LOW:
+      return kLowPriority;
+    case Priority::NORMAL:
+      return kNormalPriority;
+    default:
+      RTC_NOTREACHED();
+      return kNormalPriority;
+  }
+  return kNormalPriority;
+}
+
+}  // namespace
+
+class TaskQueue::Impl : public RefCountInterface {
+ public:
+  Impl(const char* queue_name, TaskQueue* queue, Priority priority);
+  ~Impl() override;
+
+  static TaskQueue::Impl* Current();
+  static TaskQueue* CurrentQueue();
+
+  // Used for DCHECKing the current queue.
+  bool IsCurrent() const;
+
+  template <class Closure,
+            typename std::enable_if<!std::is_convertible<
+                Closure,
+                std::unique_ptr<QueuedTask>>::value>::type* = nullptr>
+  void PostTask(Closure&& closure) {
+    PostTask(NewClosure(std::forward<Closure>(closure)));
+  }
+
+  void PostTask(std::unique_ptr<QueuedTask> task);
+  void PostTaskAndReply(std::unique_ptr<QueuedTask> task,
+                        std::unique_ptr<QueuedTask> reply,
+                        TaskQueue::Impl* reply_queue);
+
+  void PostDelayedTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds);
+
+  class WorkerThread : public PlatformThread {
+   public:
+    WorkerThread(ThreadRunFunction func,
+                 void* obj,
+                 const char* thread_name,
+                 ThreadPriority priority)
+        : PlatformThread(func, obj, thread_name, priority) {}
+  };
+
+  using OrderId = uint64_t;
+
+  struct DelayedEntryTimeout {
+    int64_t next_fire_at_ms_{};
+    OrderId order_{};
+
+    bool operator<(const DelayedEntryTimeout& o) const {
+      return std::tie(next_fire_at_ms_, order_) <
+             std::tie(o.next_fire_at_ms_, o.order_);
+    }
+  };
+
+  struct NextTask {
+    bool final_task_{false};
+    std::unique_ptr<QueuedTask> run_task_;
+    int64_t sleep_time_ms_{};
+  };
+
+ protected:
+  NextTask GetNextTask();
+
+ private:
+  // The ThreadQueue::Current() method requires that the current thread
+  // returns the task queue if the current thread is the active task
+  // queue and this variable holds the value needed in thread_local to
+  // on the initialized worker thread holding the queue.
+  static thread_local TaskQueue::Impl* thread_context_;
+
+  static void ThreadMain(void* context);
+
+  void ProcessTasks();
+
+  void NotifyWake();
+
+  // The back pointer from the owner task queue object
+  // from this implementation detail.
+  TaskQueue* const queue_;
+
+  // Indicates if the thread has started.
+  Event started_;
+
+  // Indicates if the thread has stopped.
+  Event stopped_;
+
+  // Signaled whenever a new task is pending.
+  Event flag_notify_;
+
+  // Contains the active worker thread assigned to processing
+  // tasks (including delayed tasks).
+  WorkerThread thread_;
+
+  rtc::CriticalSection pending_lock_;
+
+  // Indicates if the worker thread needs to shutdown now.
+  bool thread_should_quit_ RTC_GUARDED_BY(pending_lock_){false};
+
+  // Holds the next order to use for the next task to be
+  // put into one of the pending queues.
+  OrderId thread_posting_order_ RTC_GUARDED_BY(pending_lock_){};
+
+  // The list of all pending tasks that need to be processed in the
+  // FIFO queue ordering on the worker thread.
+  std::queue<std::pair<OrderId, std::unique_ptr<QueuedTask>>> pending_queue_
+      RTC_GUARDED_BY(pending_lock_);
+
+  // The list of all pending tasks that need to be processed at a future
+  // time based upon a delay. On the off change the delayed task should
+  // happen at exactly the same time interval as another task then the
+  // task is processed based on FIFO ordering. std::priority_queue was
+  // considered but rejected due to its inability to extract the
+  // std::unique_ptr out of the queue without the presence of a hack.
+  std::map<DelayedEntryTimeout, std::unique_ptr<QueuedTask>> delayed_queue_
+      RTC_GUARDED_BY(pending_lock_);
+};
+
+// static
+thread_local TaskQueue::Impl* TaskQueue::Impl::thread_context_ = nullptr;
+
+TaskQueue::Impl::Impl(const char* queue_name,
+                      TaskQueue* queue,
+                      Priority priority)
+    : queue_(queue),
+      started_(/*manual_reset=*/false, /*initially_signaled=*/false),
+      stopped_(/*manual_reset=*/false, /*initially_signaled=*/false),
+      flag_notify_(/*manual_reset=*/false, /*initially_signaled=*/false),
+      thread_(&TaskQueue::Impl::ThreadMain,
+              this,
+              queue_name,
+              TaskQueuePriorityToThreadPriority(priority)) {
+  RTC_DCHECK(queue_name);
+  thread_.Start();
+  started_.Wait(Event::kForever);
+}
+
+TaskQueue::Impl::~Impl() {
+  RTC_DCHECK(!IsCurrent());
+
+  {
+    CritScope lock(&pending_lock_);
+    thread_should_quit_ = true;
+  }
+
+  NotifyWake();
+
+  stopped_.Wait(Event::kForever);
+  thread_.Stop();
+}
+
+// static
+TaskQueue::Impl* TaskQueue::Impl::Current() {
+  return thread_context_;
+}
+
+// static
+TaskQueue* TaskQueue::Impl::CurrentQueue() {
+  TaskQueue::Impl* current = Current();
+  return current ? current->queue_ : nullptr;
+}
+
+bool TaskQueue::Impl::IsCurrent() const {
+  return IsThreadRefEqual(thread_.GetThreadRef(), CurrentThreadRef());
+}
+
+void TaskQueue::Impl::PostTask(std::unique_ptr<QueuedTask> task) {
+  {
+    CritScope lock(&pending_lock_);
+    OrderId order = thread_posting_order_++;
+
+    pending_queue_.push(std::pair<OrderId, std::unique_ptr<QueuedTask>>(
+        order, std::move(task)));
+  }
+
+  NotifyWake();
+}
+
+void TaskQueue::Impl::PostDelayedTask(std::unique_ptr<QueuedTask> task,
+                                      uint32_t milliseconds) {
+  auto fire_at = rtc::TimeMillis() + milliseconds;
+
+  DelayedEntryTimeout delay;
+  delay.next_fire_at_ms_ = fire_at;
+
+  {
+    CritScope lock(&pending_lock_);
+    delay.order_ = ++thread_posting_order_;
+    delayed_queue_[delay] = std::move(task);
+  }
+
+  NotifyWake();
+}
+
+void TaskQueue::Impl::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
+                                       std::unique_ptr<QueuedTask> reply,
+                                       TaskQueue::Impl* reply_queue) {
+  QueuedTask* task_ptr = task.release();
+  QueuedTask* reply_task_ptr = reply.release();
+  PostTask([task_ptr, reply_task_ptr, reply_queue]() {
+    if (task_ptr->Run())
+      delete task_ptr;
+
+    reply_queue->PostTask(std::unique_ptr<QueuedTask>(reply_task_ptr));
+  });
+}
+
+TaskQueue::Impl::NextTask TaskQueue::Impl::GetNextTask() {
+  NextTask result{};
+
+  auto tick = rtc::TimeMillis();
+
+  CritScope lock(&pending_lock_);
+
+  if (thread_should_quit_) {
+    result.final_task_ = true;
+    return result;
+  }
+
+  if (delayed_queue_.size() > 0) {
+    auto delayed_entry = delayed_queue_.begin();
+    const auto& delay_info = delayed_entry->first;
+    auto& delay_run = delayed_entry->second;
+    if (tick >= delay_info.next_fire_at_ms_) {
+      if (pending_queue_.size() > 0) {
+        auto& entry = pending_queue_.front();
+        auto& entry_order = entry.first;
+        auto& entry_run = entry.second;
+        if (entry_order < delay_info.order_) {
+          result.run_task_ = std::move(entry_run);
+          pending_queue_.pop();
+          return result;
+        }
+      }
+
+      result.run_task_ = std::move(delay_run);
+      delayed_queue_.erase(delayed_entry);
+      return result;
+    }
+
+    result.sleep_time_ms_ = delay_info.next_fire_at_ms_ - tick;
+  }
+
+  if (pending_queue_.size() > 0) {
+    auto& entry = pending_queue_.front();
+    result.run_task_ = std::move(entry.second);
+    pending_queue_.pop();
+  }
+
+  return result;
+}
+
+// static
+void TaskQueue::Impl::ThreadMain(void* context) {
+  TaskQueue::Impl* me = static_cast<TaskQueue::Impl*>(context);
+  me->ProcessTasks();
+}
+
+void TaskQueue::Impl::ProcessTasks() {
+  thread_context_ = this;
+  started_.Set();
+
+  while (true) {
+    auto task = GetNextTask();
+
+    if (task.final_task_)
+      break;
+
+    if (task.run_task_) {
+      // process entry immediately then try again
+      QueuedTask* release_ptr = task.run_task_.release();
+      if (release_ptr->Run())
+        delete release_ptr;
+
+      // attempt to sleep again
+      continue;
+    }
+
+    if (0 == task.sleep_time_ms_)
+      flag_notify_.Wait(Event::kForever);
+    else
+      flag_notify_.Wait(task.sleep_time_ms_);
+  }
+
+  stopped_.Set();
+}
+
+void TaskQueue::Impl::NotifyWake() {
+  // The queue holds pending tasks to complete. Either tasks are to be
+  // executed immediately or tasks are to be run at some future delayed time.
+  // For immediate tasks the task queue's thread is busy running the task and
+  // the thread will not be waiting on the flag_notify_ event. If no immediate
+  // tasks are available but a delayed task is pending then the thread will be
+  // waiting on flag_notify_ with a delayed time-out of the nearest timed task
+  // to run. If no immediate or pending tasks are available, the thread will
+  // wait on flag_notify_ until signaled that a task has been added (or the
+  // thread to be told to shutdown).
+
+  // In all cases, when a new immediate task, delayed task, or request to
+  // shutdown the thread is added the flag_notify_ is signaled after. If the
+  // thread was waiting then the thread will wake up immediately and re-assess
+  // what task needs to be run next (i.e. run a task now, wait for the nearest
+  // timed delayed task, or shutdown the thread). If the thread was not waiting
+  // then the thread will remained signaled to wake up the next time any
+  // attempt to wait on the flag_notify_ event occurs.
+
+  // Any immediate or delayed pending task (or request to shutdown the thread)
+  // must always be added to the queue prior to signaling flag_notify_ to wake
+  // up the possibly sleeping thread. This prevents a race condition where the
+  // thread is notified to wake up but the task queue's thread finds nothing to
+  // do so it waits once again to be signaled where such a signal may never
+  // happen.
+  flag_notify_.Set();
+}
+
+// Boilerplate for the PIMPL pattern.
+TaskQueue::TaskQueue(const char* queue_name, Priority priority)
+    : impl_(new RefCountedObject<TaskQueue::Impl>(queue_name, this, priority)) {
+}
+
+TaskQueue::~TaskQueue() {}
+
+// static
+TaskQueue* TaskQueue::Current() {
+  return TaskQueue::Impl::CurrentQueue();
+}
+
+// Used for DCHECKing the current queue.
+bool TaskQueue::IsCurrent() const {
+  return impl_->IsCurrent();
+}
+
+void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
+  return TaskQueue::impl_->PostTask(std::move(task));
+}
+
+void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
+                                 std::unique_ptr<QueuedTask> reply,
+                                 TaskQueue* reply_queue) {
+  return TaskQueue::impl_->PostTaskAndReply(std::move(task), std::move(reply),
+                                            reply_queue->impl_.get());
+}
+
+void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
+                                 std::unique_ptr<QueuedTask> reply) {
+  return TaskQueue::impl_->PostTaskAndReply(std::move(task), std::move(reply),
+                                            impl_.get());
+}
+
+void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
+                                uint32_t milliseconds) {
+  return TaskQueue::impl_->PostDelayedTask(std::move(task), milliseconds);
+}
+
+}  // namespace rtc