[Embedder] Implement merged platform and UI thread (flutter#162944)

knopp · Jonah Williams · loic-sharma · web-flow · commit 94cd4b14c979 · 2025-02-11T12:24:13.000Z
Fixes flutter#152337 Introduces `ui_task_runner` field on `FlutterCustomTaskRunners`. This lets the embedder to specify task runner for UI isolate tasks, allowing for merging platform and UI threads. With custom UI task runner there is no `MessageLoop` anymore for the UI thread, so the message loop task observer can no longer be used to flush microtask queue. Instead the microtask queue is flushed at the end of each `FlutterEngineRunTask` invocation if needed. This is handled internally by the embedder. ## Pre-launch Checklist - [x] I read the [Contributor Guide] and followed the process outlined there for submitting PRs. - [x] I read the [Tree Hygiene] wiki page, which explains my responsibilities. - [x] I read and followed the [Flutter Style Guide], including [Features we expect every widget to implement]. - [x] I signed the [CLA]. - [x] I listed at least one issue that this PR fixes in the description above. - [x] I updated/added relevant documentation (doc comments with `///`). - [x] I added new tests to check the change I am making, or this PR is [test-exempt]. - [x] I followed the [breaking change policy] and added [Data Driven Fixes] where supported. - [x] All existing and new tests are passing. If you need help, consider asking for advice on the #hackers-new channel on [Discord].  [Contributor Guide]: https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md#overview [Tree Hygiene]: https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md [test-exempt]: https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md#tests [Flutter Style Guide]: https://github.com/flutter/flutter/blob/main/docs/contributing/Style-guide-for-Flutter-repo.md [Features we expect every widget to implement]: https://github.com/flutter/flutter/blob/main/docs/contributing/Style-guide-for-Flutter-repo.md#features-we-expect-every-widget-to-implement [CLA]: https://cla.developers.google.com/ [flutter/tests]: https://github.com/flutter/tests [breaking change policy]: https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md#handling-breaking-changes [Discord]: https://github.com/flutter/flutter/blob/main/docs/contributing/Chat.md [Data Driven Fixes]: https://github.com/flutter/flutter/blob/main/docs/contributing/Data-driven-Fixes.md --------- Co-authored-by: Jonah Williams <jonahwilliams@google.com> Co-authored-by: Loïc Sharma <737941+loic-sharma@users.noreply.github.com>
diff --git a/engine/src/flutter/fml/message_loop_task_queues.cc b/engine/src/flutter/fml/message_loop_task_queues.cc
@@ -17,6 +17,7 @@
 namespace fml {
 
 const size_t TaskQueueId::kUnmerged = ULONG_MAX;
+const size_t TaskQueueId::kInvalid = ULONG_MAX - 1;
 
 namespace {
 
diff --git a/engine/src/flutter/fml/task_queue_id.h b/engine/src/flutter/fml/task_queue_id.h
@@ -18,13 +18,19 @@ class TaskQueueId {
   /// runner.
   static const size_t kUnmerged;
 
+  /// This constant indicates an invalid task queue. Used in embedder
+  /// supplied task runners not associated with a task queue.
+  static const size_t kInvalid;
+
   /// Intializes a task queue with the given value as it's ID.
   explicit TaskQueueId(size_t value) : value_(value) {}
 
   operator size_t() const {  // NOLINT(google-explicit-constructor)
     return value_;
   }
 
+  bool is_valid() const { return value_ != kInvalid; }
+
  private:
   size_t value_ = kUnmerged;
 };
diff --git a/engine/src/flutter/fml/task_runner.h b/engine/src/flutter/fml/task_runner.h
@@ -55,6 +55,9 @@ class TaskRunner : public fml::RefCountedThreadSafe<TaskRunner>,
 
   /// Returns the unique identifier associated with the TaskRunner.
   /// \see fml::MessageLoopTaskQueues
+  ///
+  /// Will be TaskQueueId::kInvalid for embedder supplied task runners
+  /// that are not associated with a task queue.
   virtual TaskQueueId GetTaskQueueId();
 
   /// Executes the \p task directly if the TaskRunner \p runner is the
diff --git a/engine/src/flutter/runtime/dart_isolate.cc b/engine/src/flutter/runtime/dart_isolate.cc
@@ -510,7 +510,13 @@ bool DartIsolate::Initialize(Dart_Isolate dart_isolate) {
     SetMessageHandlingTaskRunner(GetTaskRunners().GetPlatformTaskRunner(),
                                  true);
   } else {
-    SetMessageHandlingTaskRunner(GetTaskRunners().GetUITaskRunner(), false);
+    // When running with custom UI task runner post directly to runner (there is
+    // no task queue).
+    bool post_directly_to_runner =
+        GetTaskRunners().GetUITaskRunner() &&
+        !GetTaskRunners().GetUITaskRunner()->GetTaskQueueId().is_valid();
+    SetMessageHandlingTaskRunner(GetTaskRunners().GetUITaskRunner(),
+                                 post_directly_to_runner);
   }
 
   if (tonic::CheckAndHandleError(
diff --git a/engine/src/flutter/runtime/runtime_controller.h b/engine/src/flutter/runtime/runtime_controller.h
@@ -655,6 +655,12 @@ class RuntimeController : public PlatformConfigurationClient,
     return root_isolate_;
   }
 
+  void FlushMicrotaskQueue() {
+    if (auto isolate = root_isolate_.lock()) {
+      isolate->FlushMicrotasksNow();
+    }
+  }
+
   std::shared_ptr<PlatformIsolateManager> GetPlatformIsolateManager() override {
     return platform_isolate_manager_;
   }
diff --git a/engine/src/flutter/shell/common/engine.cc b/engine/src/flutter/shell/common/engine.cc
@@ -627,4 +627,8 @@ void Engine::ShutdownPlatformIsolates() {
   runtime_controller_->ShutdownPlatformIsolates();
 }
 
+void Engine::FlushMicrotaskQueue() {
+  runtime_controller_->FlushMicrotaskQueue();
+}
+
 }  // namespace flutter
diff --git a/engine/src/flutter/shell/common/engine.h b/engine/src/flutter/shell/common/engine.h
@@ -978,6 +978,11 @@ class Engine final : public RuntimeDelegate, PointerDataDispatcher::Delegate {
   ///
   void ShutdownPlatformIsolates();
 
+  //--------------------------------------------------------------------------
+  /// @brief      Flushes the microtask queue of the root isolate.
+  ///
+  void FlushMicrotaskQueue();
+
  private:
   // |RuntimeDelegate|
   std::string DefaultRouteName() override;
diff --git a/engine/src/flutter/shell/common/shell.cc b/engine/src/flutter/shell/common/shell.cc
@@ -634,6 +634,12 @@ void Shell::NotifyLowMemoryWarning() const {
   // to purge them.
 }
 
+void Shell::FlushMicrotaskQueue() const {
+  if (engine_) {
+    engine_->FlushMicrotaskQueue();
+  }
+}
+
 void Shell::RunEngine(RunConfiguration run_configuration) {
   RunEngine(std::move(run_configuration), nullptr);
 }
diff --git a/engine/src/flutter/shell/common/shell.h b/engine/src/flutter/shell/common/shell.h
@@ -289,6 +289,13 @@ class Shell final : public PlatformView::Delegate,
   ///             the rasterizer cache is purged.
   void NotifyLowMemoryWarning() const;
 
+  //----------------------------------------------------------------------------
+  /// @brief      Used by embedders to flush the microtask queue. Required
+  ///             when running with merged platform and UI threads, in which
+  ///             case the embedder is responsible for flushing the microtask
+  ///             queue.
+  void FlushMicrotaskQueue() const;
+
   //----------------------------------------------------------------------------
   /// @brief      Used by embedders to check if all shell subcomponents are
   ///             initialized. It is the embedder's responsibility to make this
diff --git a/engine/src/flutter/shell/common/vsync_waiter.cc b/engine/src/flutter/shell/common/vsync_waiter.cc
@@ -154,12 +154,16 @@ void VsyncWaiter::FireCallback(fml::TimePoint frame_start_time,
 void VsyncWaiter::PauseDartEventLoopTasks() {
   auto ui_task_queue_id = task_runners_.GetUITaskRunner()->GetTaskQueueId();
   auto task_queues = fml::MessageLoopTaskQueues::GetInstance();
-  task_queues->PauseSecondarySource(ui_task_queue_id);
+  if (ui_task_queue_id.is_valid()) {
+    task_queues->PauseSecondarySource(ui_task_queue_id);
+  }
 }
 
 void VsyncWaiter::ResumeDartEventLoopTasks(fml::TaskQueueId ui_task_queue_id) {
   auto task_queues = fml::MessageLoopTaskQueues::GetInstance();
-  task_queues->ResumeSecondarySource(ui_task_queue_id);
+  if (ui_task_queue_id.is_valid()) {
+    task_queues->ResumeSecondarySource(ui_task_queue_id);
+  }
 }
 
 }  // namespace flutter
diff --git a/engine/src/flutter/shell/platform/embedder/embedder.cc b/engine/src/flutter/shell/platform/embedder/embedder.cc
@@ -2087,12 +2087,6 @@ FlutterEngineResult FlutterEngineInitialize(size_t version,
     settings.application_kernel_asset = kApplicationKernelSnapshotFileName;
   }
 
-  settings.task_observer_add = [](intptr_t key, const fml::closure& callback) {
-    fml::MessageLoop::GetCurrent().AddTaskObserver(key, callback);
-  };
-  settings.task_observer_remove = [](intptr_t key) {
-    fml::MessageLoop::GetCurrent().RemoveTaskObserver(key);
-  };
   if (SAFE_ACCESS(args, root_isolate_create_callback, nullptr) != nullptr) {
     VoidCallback callback =
         SAFE_ACCESS(args, root_isolate_create_callback, nullptr);
@@ -2355,6 +2349,24 @@ FlutterEngineResult FlutterEngineInitialize(size_t version,
                               "Task runner configuration was invalid.");
   }
 
+  // Embedder supplied UI task runner runner does not have a message loop.
+  bool has_ui_thread_message_loop =
+      task_runners.GetUITaskRunner()->GetTaskQueueId().is_valid();
+  // Message loop observers are used to flush the microtask queue.
+  // If there is no message loop the queue is flushed from
+  // EmbedderEngine::RunTask.
+  settings.task_observer_add = [has_ui_thread_message_loop](
+                                   intptr_t key, const fml::closure& callback) {
+    if (has_ui_thread_message_loop) {
+      fml::MessageLoop::GetCurrent().AddTaskObserver(key, callback);
+    }
+  };
+  settings.task_observer_remove = [has_ui_thread_message_loop](intptr_t key) {
+    if (has_ui_thread_message_loop) {
+      fml::MessageLoop::GetCurrent().RemoveTaskObserver(key);
+    }
+  };
+
   auto run_configuration =
       flutter::RunConfiguration::InferFromSettings(settings);
 
diff --git a/engine/src/flutter/shell/platform/embedder/embedder.h b/engine/src/flutter/shell/platform/embedder/embedder.h
@@ -1703,6 +1703,10 @@ typedef struct {
   /// Specify a callback that is used to set the thread priority for embedder
   /// task runners.
   void (*thread_priority_setter)(FlutterThreadPriority);
+  /// Specify the task runner for the thread on which the UI tasks will be run.
+  /// This may be same as platform_task_runner, in which case the Flutter engine
+  /// will run the UI isolate on platform thread.
+  const FlutterTaskRunnerDescription* ui_task_runner;
 } FlutterCustomTaskRunners;
 
 typedef struct {
diff --git a/engine/src/flutter/shell/platform/embedder/embedder_engine.cc b/engine/src/flutter/shell/platform/embedder/embedder_engine.cc
@@ -243,8 +243,20 @@ bool EmbedderEngine::RunTask(const FlutterTask* task) {
   if (task == nullptr) {
     return false;
   }
-  return thread_host_->PostTask(reinterpret_cast<int64_t>(task->runner),
-                                task->task);
+  auto result = thread_host_->PostTask(reinterpret_cast<int64_t>(task->runner),
+                                       task->task);
+  // If the UI and platform threads are separate, the microtask queue is
+  // flushed through MessageLoopTaskQueues observer.
+  // If the UI and platform threads are merged, the UI task runner has no
+  // associated task queue, and microtasks need to be flushed manually
+  // after running the task.
+  if (result && shell_ && task_runners_.GetUITaskRunner() &&
+      task_runners_.GetUITaskRunner()->RunsTasksOnCurrentThread() &&
+      !task_runners_.GetUITaskRunner()->GetTaskQueueId().is_valid()) {
+    shell_->FlushMicrotaskQueue();
+  }
+
+  return result;
 }
 
 bool EmbedderEngine::PostTaskOnEngineManagedNativeThreads(
diff --git a/engine/src/flutter/shell/platform/embedder/embedder_task_runner.cc b/engine/src/flutter/shell/platform/embedder/embedder_task_runner.cc
@@ -14,8 +14,7 @@ EmbedderTaskRunner::EmbedderTaskRunner(DispatchTable table,
     : TaskRunner(nullptr /* loop implemenation*/),
       embedder_identifier_(embedder_identifier),
       dispatch_table_(std::move(table)),
-      placeholder_id_(
-          fml::MessageLoopTaskQueues::GetInstance()->CreateTaskQueue()) {
+      placeholder_id_(fml::TaskQueueId(fml::TaskQueueId::kInvalid)) {
   FML_DCHECK(dispatch_table_.post_task_callback);
   FML_DCHECK(dispatch_table_.runs_task_on_current_thread_callback);
   FML_DCHECK(dispatch_table_.destruction_callback);
diff --git a/engine/src/flutter/shell/platform/embedder/embedder_thread_host.cc b/engine/src/flutter/shell/platform/embedder/embedder_thread_host.cc
@@ -142,15 +142,15 @@ EmbedderThreadHost::CreateEmbedderManagedThreadHost(
 
   auto thread_host_config = ThreadHost::ThreadHostConfig(config_setter);
 
-  // The UI and IO threads are always created by the engine and the embedder has
+  // The IO threads are always created by the engine and the embedder has
   // no opportunity to specify task runners for the same.
   //
   // If/when more task runners are exposed, this mask will need to be updated.
-  thread_host_config.SetUIConfig(MakeThreadConfig(
-      ThreadHost::Type::kUi, fml::Thread::ThreadPriority::kDisplay));
   thread_host_config.SetIOConfig(MakeThreadConfig(
       ThreadHost::Type::kIo, fml::Thread::ThreadPriority::kBackground));
 
+  auto ui_task_runner_pair = CreateEmbedderTaskRunner(
+      SAFE_ACCESS(custom_task_runners, ui_task_runner, nullptr));
   auto platform_task_runner_pair = CreateEmbedderTaskRunner(
       SAFE_ACCESS(custom_task_runners, platform_task_runner, nullptr));
   auto render_task_runner_pair = CreateEmbedderTaskRunner(
@@ -163,6 +163,12 @@ EmbedderThreadHost::CreateEmbedderManagedThreadHost(
     return nullptr;
   }
 
+  // If the embedder has not supplied a UI task runner, one needs to be created.
+  if (!ui_task_runner_pair.second) {
+    thread_host_config.SetUIConfig(MakeThreadConfig(
+        ThreadHost::Type::kUi, fml::Thread::ThreadPriority::kDisplay));
+  }
+
   // If the embedder has not supplied a raster task runner, one needs to be
   // created.
   if (!render_task_runner_pair.second) {
@@ -179,6 +185,15 @@ EmbedderThreadHost::CreateEmbedderManagedThreadHost(
     }
   }
 
+  // If both platform task runner and UI task runner are specified and have
+  // the same identifier, store only one.
+  if (platform_task_runner_pair.second && ui_task_runner_pair.second) {
+    if (platform_task_runner_pair.second->GetEmbedderIdentifier() ==
+        ui_task_runner_pair.second->GetEmbedderIdentifier()) {
+      ui_task_runner_pair.second = platform_task_runner_pair.second;
+    }
+  }
+
   // Create a thread host with just the threads that need to be managed by the
   // engine. The embedder has provided the rest.
   ThreadHost thread_host(thread_host_config);
@@ -197,12 +212,17 @@ EmbedderThreadHost::CreateEmbedderManagedThreadHost(
                                       render_task_runner_pair.second)
                                 : thread_host.raster_thread->GetTaskRunner();
 
+  auto ui_task_runner = ui_task_runner_pair.second
+                            ? static_cast<fml::RefPtr<fml::TaskRunner>>(
+                                  ui_task_runner_pair.second)
+                            : thread_host.ui_thread->GetTaskRunner();
+
   flutter::TaskRunners task_runners(
       kFlutterThreadName,
-      platform_task_runner,                    // platform
-      render_task_runner,                      // raster
-      thread_host.ui_thread->GetTaskRunner(),  // ui (always engine managed)
-      thread_host.io_thread->GetTaskRunner()   // io (always engine managed)
+      platform_task_runner,                   // platform
+      render_task_runner,                     // raster
+      ui_task_runner,                         // ui
+      thread_host.io_thread->GetTaskRunner()  // io (always engine managed)
   );
 
   if (!task_runners.IsValid()) {
@@ -219,6 +239,10 @@ EmbedderThreadHost::CreateEmbedderManagedThreadHost(
     embedder_task_runners.insert(render_task_runner_pair.second);
   }
 
+  if (ui_task_runner_pair.second) {
+    embedder_task_runners.insert(ui_task_runner_pair.second);
+  }
+
   auto embedder_host = std::make_unique<EmbedderThreadHost>(
       std::move(thread_host), std::move(task_runners),
       std::move(embedder_task_runners));
diff --git a/engine/src/flutter/shell/platform/embedder/fixtures/main.dart b/engine/src/flutter/shell/platform/embedder/fixtures/main.dart
@@ -62,6 +62,28 @@ void invokePlatformTaskRunner() {
   PlatformDispatcher.instance.sendPlatformMessage('OhHi', null, null);
 }
 
+@pragma('vm:entry-point')
+void canSpecifyCustomUITaskRunner() {
+  signalNativeTest();
+  PlatformDispatcher.instance.sendPlatformMessage('OhHi', null, null);
+}
+
+@pragma('vm:entry-point')
+void mergedPlatformUIThread() {
+  signalNativeTest();
+  PlatformDispatcher.instance.sendPlatformMessage('OhHi', null, null);
+}
+
+@pragma('vm:entry-point')
+void uiTaskRunnerFlushesMicrotasks() {
+  // Microtasks are always flushed at the beginning of the frame, hence the delay.
+  Future.delayed(const Duration(milliseconds: 50), () {
+    Future.microtask(() {
+      signalNativeTest();
+    });
+  });
+}
+
 @pragma('vm:entry-point')
 void invokePlatformThreadIsolate() {
   signalNativeTest();
diff --git a/engine/src/flutter/shell/platform/embedder/tests/embedder_config_builder.cc b/engine/src/flutter/shell/platform/embedder/tests/embedder_config_builder.cc
@@ -163,6 +163,15 @@ void EmbedderConfigBuilder::SetPlatformTaskRunner(
   project_args_.custom_task_runners = &custom_task_runners_;
 }
 
+void EmbedderConfigBuilder::SetUITaskRunner(
+    const FlutterTaskRunnerDescription* runner) {
+  if (runner == nullptr) {
+    return;
+  }
+  custom_task_runners_.ui_task_runner = runner;
+  project_args_.custom_task_runners = &custom_task_runners_;
+}
+
 void EmbedderConfigBuilder::SetupVsyncCallback() {
   project_args_.vsync_callback = [](void* user_data, intptr_t baton) {
     auto context = reinterpret_cast<EmbedderTestContext*>(user_data);
diff --git a/engine/src/flutter/shell/platform/embedder/tests/embedder_config_builder.h b/engine/src/flutter/shell/platform/embedder/tests/embedder_config_builder.h
@@ -74,6 +74,8 @@ class EmbedderConfigBuilder {
 
   void SetPlatformTaskRunner(const FlutterTaskRunnerDescription* runner);
 
+  void SetUITaskRunner(const FlutterTaskRunnerDescription* runner);
+
   void SetRenderTaskRunner(const FlutterTaskRunnerDescription* runner);
 
   void SetPlatformMessageCallback(
diff --git a/engine/src/flutter/shell/platform/embedder/tests/embedder_unittests.cc b/engine/src/flutter/shell/platform/embedder/tests/embedder_unittests.cc

Original file line number	Diff line number	Diff line change
`@@ -655,6 +655,12 @@ class RuntimeController : public PlatformConfigurationClient,`
`655`	`655`	`return root_isolate_;`
`656`	`656`	`}`
`657`	`657`
	`658`	`+ void FlushMicrotaskQueue() {`
	`659`	`+ if (auto isolate = root_isolate_.lock()) {`
	`660`	`+ isolate->FlushMicrotasksNow();`
	`661`	`+ }`
	`662`	`+ }`
	`663`	`+`
`658`	`664`	`std::shared_ptr<PlatformIsolateManager> GetPlatformIsolateManager() override {`
`659`	`665`	`return platform_isolate_manager_;`
`660`	`666`	`}`
Original file line number	Diff line number	Diff line change
`@@ -627,4 +627,8 @@ void Engine::ShutdownPlatformIsolates() {`
`627`	`627`	`runtime_controller_->ShutdownPlatformIsolates();`
`628`	`628`	`}`
`629`	`629`
	`630`	`+void Engine::FlushMicrotaskQueue() {`
	`631`	`+ runtime_controller_->FlushMicrotaskQueue();`
	`632`	`+}`
	`633`	`+`
`630`	`634`	`} // namespace flutter`
Original file line number	Diff line number	Diff line change
`@@ -634,6 +634,12 @@ void Shell::NotifyLowMemoryWarning() const {`
`634`	`634`	`// to purge them.`
`635`	`635`	`}`
`636`	`636`
	`637`	`+void Shell::FlushMicrotaskQueue() const {`
	`638`	`+ if (engine_) {`
	`639`	`+ engine_->FlushMicrotaskQueue();`
	`640`	`+ }`
	`641`	`+}`
	`642`	`+`
`637`	`643`	`void Shell::RunEngine(RunConfiguration run_configuration) {`
`638`	`644`	`RunEngine(std::move(run_configuration), nullptr);`
`639`	`645`	`}`