Fully inplement TaskRunner for UWP (flutter/flutter#70890)

ci/licenses_golden/licenses_flutter

@@ -1753,7 +1753,9 @@ FILE: ../../../flutter/shell/platform/windows/system_utils.h
 FILE: ../../../flutter/shell/platform/windows/system_utils_unittests.cc
 FILE: ../../../flutter/shell/platform/windows/system_utils_win32.cc
 FILE: ../../../flutter/shell/platform/windows/system_utils_winuwp.cc
+FILE: ../../../flutter/shell/platform/windows/task_runner.cc
 FILE: ../../../flutter/shell/platform/windows/task_runner.h
+FILE: ../../../flutter/shell/platform/windows/task_runner_unittests.cc
 FILE: ../../../flutter/shell/platform/windows/task_runner_win32.cc
 FILE: ../../../flutter/shell/platform/windows/task_runner_win32.h
 FILE: ../../../flutter/shell/platform/windows/task_runner_win32_window.cc

shell/platform/windows/BUILD.gn

@@ -77,6 +77,7 @@ source_set("flutter_windows_source") {
     "sequential_id_generator.cc",
     "sequential_id_generator.h",
     "system_utils.h",
+    "task_runner.cc",
     "task_runner.h",
     "text_input_plugin.cc",
     "text_input_plugin.h",
@@ -226,6 +227,7 @@ executable("flutter_windows_unittests") {
     "sequential_id_generator_unittests.cc",
     "string_conversion_unittests.cc",
     "system_utils_unittests.cc",
+    "task_runner_unittests.cc",
     "testing/engine_modifier.h",
     "testing/mock_gl_functions.h",
   ]

shell/platform/windows/flutter_windows_engine.cc

@@ -143,8 +143,7 @@ FlutterWindowsEngine::FlutterWindowsEngine(const FlutterProjectBundle& project)
   FlutterEngineGetProcAddresses(&embedder_api_);
 
   task_runner_ = TaskRunner::Create(
-      GetCurrentThreadId(), embedder_api_.GetCurrentTime,
-      [this](const auto* task) {
+      embedder_api_.GetCurrentTime, [this](const auto* task) {
         if (!engine_) {
           std::cerr << "Cannot post an engine task when engine is not running."
                     << std::endl;

shell/platform/windows/task_runner.cc

@@ -0,0 +1,104 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "flutter/shell/platform/windows/task_runner.h"
+
+#include <atomic>
+#include <utility>
+
+namespace flutter {
+
+TaskRunner::TaskRunner(CurrentTimeProc get_current_time,
+                       const TaskExpiredCallback& on_task_expired)
+    : get_current_time_(get_current_time),
+      on_task_expired_(std::move(on_task_expired)) {}
+
+std::chrono::nanoseconds TaskRunner::ProcessTasks() {
+  const TaskTimePoint now = TaskTimePoint::clock::now();
+
+  std::vector<Task> expired_tasks;
+
+  // Process expired tasks.
+  {
+    std::lock_guard<std::mutex> lock(task_queue_mutex_);
+    while (!task_queue_.empty()) {
+      const auto& top = task_queue_.top();
+      // If this task (and all tasks after this) has not yet expired, there is
+      // nothing more to do. Quit iterating.
+      if (top.fire_time > now) {
+        break;
+      }
+
+      // Make a record of the expired task. Do NOT service the task here
+      // because we are still holding onto the task queue mutex. We don't want
+      // other threads to block on posting tasks onto this thread till we are
+      // done processing expired tasks.
+      expired_tasks.push_back(task_queue_.top());
+
+      // Remove the tasks from the delayed tasks queue.
+      task_queue_.pop();
+    }
+  }
+
+  // Fire expired tasks.
+  {
+    // Flushing tasks here without holing onto the task queue mutex.
+    for (const auto& task : expired_tasks) {
+      if (auto flutter_task = std::get_if<FlutterTask>(&task.variant)) {
+        on_task_expired_(flutter_task);
+      } else if (auto closure = std::get_if<TaskClosure>(&task.variant))
+        (*closure)();
+    }
+  }
+
+  // Calculate duration to sleep for on next iteration.
+  {
+    std::lock_guard<std::mutex> lock(task_queue_mutex_);
+    const auto next_wake = task_queue_.empty() ? TaskTimePoint::max()
+                                               : task_queue_.top().fire_time;
+
+    return std::min(next_wake - now, std::chrono::nanoseconds::max());
+  }
+}
+
+TaskRunner::TaskTimePoint TaskRunner::TimePointFromFlutterTime(
+    uint64_t flutter_target_time_nanos) const {
+  const auto now = TaskTimePoint::clock::now();
+  const auto flutter_duration = flutter_target_time_nanos - get_current_time_();
+  return now + std::chrono::nanoseconds(flutter_duration);
+}
+
+void TaskRunner::PostFlutterTask(FlutterTask flutter_task,
+                                 uint64_t flutter_target_time_nanos) {
+  Task task;
+  task.fire_time = TimePointFromFlutterTime(flutter_target_time_nanos);
+  task.variant = flutter_task;
+  EnqueueTask(std::move(task));
+}
+
+void TaskRunner::PostTask(TaskClosure closure) {
+  Task task;
+  task.fire_time = TaskTimePoint::clock::now();
+  task.variant = std::move(closure);
+  EnqueueTask(std::move(task));
+}
+
+void TaskRunner::EnqueueTask(Task task) {
+  static std::atomic_uint64_t sGlobalTaskOrder(0);
+
+  task.order = ++sGlobalTaskOrder;
+  {
+    std::lock_guard<std::mutex> lock(task_queue_mutex_);
+    task_queue_.push(task);
+
+    // Make sure the queue mutex is unlocked before waking up the loop. In case
+    // the wake causes this thread to be descheduled for the primary thread to
+    // process tasks, the acquisition of the lock on that thread while holding
+    // the lock here momentarily till the end of the scope is a pessimization.
+  }
+
+  WakeUp();
+}
+
+}  // namespace flutter

shell/platform/windows/task_runner.h

@@ -5,18 +5,20 @@
 #ifndef FLUTTER_SHELL_PLATFORM_WINDOWS_TASK_RUNNER_H_
 #define FLUTTER_SHELL_PLATFORM_WINDOWS_TASK_RUNNER_H_
 
-#include <windows.h>
-
 #include <chrono>
+#include <deque>
+#include <functional>
 #include <memory>
+#include <mutex>
+#include <queue>
+#include <variant>
 
 #include "flutter/shell/platform/embedder/embedder.h"
 
 namespace flutter {
 
 typedef uint64_t (*CurrentTimeProc)();
 
-// Abstract custom task runner for scheduling custom tasks.
 class TaskRunner {
  public:
   using TaskTimePoint = std::chrono::steady_clock::time_point;
@@ -25,18 +27,18 @@ class TaskRunner {
 
   virtual ~TaskRunner() = default;
 
-  // Returns if the current thread is the UI thread.
+  // Returns `true` if the current thread is this runner's thread.
   virtual bool RunsTasksOnCurrentThread() const = 0;
 
   // Post a Flutter engine task to the event loop for delayed execution.
-  virtual void PostFlutterTask(FlutterTask flutter_task,
-                               uint64_t flutter_target_time_nanos) = 0;
+  void PostFlutterTask(FlutterTask flutter_task,
+                       uint64_t flutter_target_time_nanos);
 
   // Post a task to the event loop.
-  virtual void PostTask(TaskClosure task) = 0;
+  void PostTask(TaskClosure task);
 
   // Post a task to the event loop or run it immediately if this is being called
-  // from the main thread.
+  // from the runner's thread.
   void RunNowOrPostTask(TaskClosure task) {
     if (RunsTasksOnCurrentThread()) {
       task();
@@ -45,12 +47,57 @@ class TaskRunner {
     }
   }
 
-  // Creates a new task runner with the given main thread ID, current time
+  // Creates a new task runner with the current thread, current time
   // provider, and callback for tasks that are ready to be run.
   static std::unique_ptr<TaskRunner> Create(
-      DWORD main_thread_id,
       CurrentTimeProc get_current_time,
       const TaskExpiredCallback& on_task_expired);
+
+ protected:
+  TaskRunner(CurrentTimeProc get_current_time,
+             const TaskExpiredCallback& on_task_expired);
+
+  // Schedules timers to call `ProcessTasks()` at the runner's thread.
+  virtual void WakeUp() = 0;
+
+  // Executes expired task, and returns the duration until the next task
+  // deadline if exists, otherwise returns `std::chrono::nanoseconds::max()`.
+  //
+  // Each platform implementations must call this to schedule the tasks.
+  std::chrono::nanoseconds ProcessTasks();
+
+ private:
+  typedef std::variant<FlutterTask, TaskClosure> TaskVariant;
+
+  struct Task {
+    uint64_t order;
+    TaskTimePoint fire_time;
+    TaskVariant variant;
+
+    struct Comparer {
+      bool operator()(const Task& a, const Task& b) {
+        if (a.fire_time == b.fire_time) {
+          return a.order > b.order;
+        }
+        return a.fire_time > b.fire_time;
+      }
+    };
+  };
+
+  // Enqueues the given task.
+  void EnqueueTask(Task task);
+
+  // Returns a TaskTimePoint computed from the given target time from Flutter.
+  TaskTimePoint TimePointFromFlutterTime(
+      uint64_t flutter_target_time_nanos) const;
+
+  CurrentTimeProc get_current_time_;
+  TaskExpiredCallback on_task_expired_;
+  std::mutex task_queue_mutex_;
+  std::priority_queue<Task, std::deque<Task>, Task::Comparer> task_queue_;
+
+  TaskRunner(const TaskRunner&) = delete;
+  TaskRunner& operator=(const TaskRunner&) = delete;
 };
 
 }  // namespace flutter

shell/platform/windows/task_runner_unittests.cc

@@ -0,0 +1,89 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <chrono>
+
+#include "flutter/fml/time/time_point.h"
+
+#include "flutter/shell/platform/windows/task_runner.h"
+
+#include "gtest/gtest.h"
+
+namespace flutter {
+namespace testing {
+
+namespace {
+class MockTaskRunner : public TaskRunner {
+ public:
+  MockTaskRunner(CurrentTimeProc get_current_time,
+                 const TaskExpiredCallback& on_task_expired)
+      : TaskRunner(get_current_time, on_task_expired) {}
+
+  virtual bool RunsTasksOnCurrentThread() const override { return true; }
+
+  void SimulateTimerAwake() { ProcessTasks(); }
+
+ protected:
+  virtual void WakeUp() override {
+    // Do nothing to avoid processing tasks immediately after the tasks is
+    // posted.
+  }
+};
+
+uint64_t MockGetCurrentTime() {
+  return static_cast<uint64_t>(
+      fml::TimePoint::Now().ToEpochDelta().ToNanoseconds());
+}
+}  // namespace
+
+TEST(TaskRunnerTest, MaybeExecuteTaskWithExactOrder) {
+  std::vector<uint64_t> executed_task_order;
+  auto runner =
+      MockTaskRunner(MockGetCurrentTime,
+                     [&executed_task_order](const FlutterTask* expired_task) {
+                       executed_task_order.push_back(expired_task->task);
+                     });
+
+  uint64_t time_now = MockGetCurrentTime();
+
+  runner.PostFlutterTask(FlutterTask{nullptr, 1}, time_now);
+  runner.PostFlutterTask(FlutterTask{nullptr, 2}, time_now);
+  runner.PostTask(
+      [&executed_task_order]() { executed_task_order.push_back(3); });
+  runner.PostTask(
+      [&executed_task_order]() { executed_task_order.push_back(4); });
+
+  runner.SimulateTimerAwake();
+
+  std::vector<uint64_t> posted_task_order{1, 2, 3, 4};
+  EXPECT_EQ(executed_task_order, posted_task_order);
+}
+
+TEST(TaskRunnerTest, MaybeExecuteTaskOnlyExpired) {
+  std::set<uint64_t> executed_task;
+  auto runner = MockTaskRunner(
+      MockGetCurrentTime, [&executed_task](const FlutterTask* expired_task) {
+        executed_task.insert(expired_task->task);
+      });
+
+  uint64_t time_now = MockGetCurrentTime();
+
+  uint64_t task_expired_before_now = 1;
+  uint64_t time_before_now = time_now - 10000;
+  runner.PostFlutterTask(FlutterTask{nullptr, task_expired_before_now},
+                         time_before_now);
+
+  uint64_t task_expired_after_now = 2;
+  uint64_t time_after_now = time_now + 10000;
+  runner.PostFlutterTask(FlutterTask{nullptr, task_expired_after_now},
+                         time_after_now);
+
+  runner.SimulateTimerAwake();
+
+  std::set<uint64_t> only_task_expired_before_now{task_expired_before_now};
+  EXPECT_EQ(executed_task, only_task_expired_before_now);
+}
+
+}  // namespace testing
+}  // namespace flutter