[Embedder] Detect and ignore stale task runner tasks (#163129)

Fixes https://github.com/flutter/flutter/issues/163104

The core issue is that `EmbedderTaskRunner` can schedule things in
advance, but there is no checking if the task is stale when executing
it. It is possible that `FlutterEngineRunTask` will attempt to run the
task on engine that is non null but already being deallocated.

This in not a problem for raster and platform task runners, because
raster task runner shouldn't have any scheduled tasks after shutdown and
platform task runner executes the shutdown process, so the pointer is
always either valid or null, but it is an issue for custom
`ui_task_runner`, because it may be calling `FlutterEngineRunTask` with
engine pointer that is not yet null but already shutting down.

The proposed solution is to assign a unique identifier for each
`EmbedderTaskRunner`, use this identifier as the `runner` field inside
`FlutterTask` instead of casting the address of the runner, and verify
that this identifier is valid inside `FlutterEngineRunTask` before
calling anything on the engine.

Special care needs to be done to not invalidate the unique identifier
while `ui_task_runner` is running a task as it may lead to raciness. See
`EmbedderEngine::CollectThreadHost()`.

## 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].

<!-- Links -->
[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

engine/src/flutter/shell/platform/embedder/embedder.cc

@@ -2566,6 +2566,7 @@ FlutterEngineResult FlutterEngineDeinitialize(FLUTTER_API_SYMBOL(FlutterEngine)
   auto embedder_engine = reinterpret_cast<flutter::EmbedderEngine*>(engine);
   embedder_engine->NotifyDestroyed();
   embedder_engine->CollectShell();
+  embedder_engine->CollectThreadHost();
   return kSuccess;
 }
 
@@ -3226,6 +3227,13 @@ FlutterEngineResult FlutterEngineRunTask(FLUTTER_API_SYMBOL(FlutterEngine)
     return LOG_EMBEDDER_ERROR(kInvalidArguments, "Invalid engine handle.");
   }
 
+  if (!flutter::EmbedderThreadHost::RunnerIsValid(
+          reinterpret_cast<intptr_t>(task->runner))) {
+    // This task came too late, the embedder has already been destroyed.
+    // This is not an error, just ignore the task.
+    return kSuccess;
+  }
+
   return reinterpret_cast<flutter::EmbedderEngine*>(engine)->RunTask(task)
              ? kSuccess
              : LOG_EMBEDDER_ERROR(kInvalidArguments,

engine/src/flutter/shell/platform/embedder/embedder_engine.cc

@@ -65,6 +65,46 @@ bool EmbedderEngine::CollectShell() {
   return IsValid();
 }
 
+void EmbedderEngine::CollectThreadHost() {
+  if (!thread_host_) {
+    return;
+  }
+
+  // Once the collected, EmbedderThreadHost::RunnerIsValid will return false for
+  // all runners belonging to this thread host. This must be done with UI task
+  // runner blocked to prevent possible raciness that could happen when
+  // destroying the thread host in the middle of UI task runner execution. This
+  // is not an issue for other runners, because raster task runner should not
+  // have anything scheduled after engine shutdown and platform task runner is
+  // where this method is called from.
+  if (thread_host_->GetTaskRunners().GetUITaskRunner() &&
+      !thread_host_->GetTaskRunners()
+           .GetUITaskRunner()
+           ->RunsTasksOnCurrentThread()) {
+    fml::AutoResetWaitableEvent ui_thread_running;
+    fml::AutoResetWaitableEvent ui_thread_block;
+    fml::AutoResetWaitableEvent ui_thread_finished;
+
+    thread_host_->GetTaskRunners().GetUITaskRunner()->PostTask([&] {
+      ui_thread_running.Signal();
+      ui_thread_block.Wait();
+      ui_thread_finished.Signal();
+    });
+
+    // Wait until the task is running on the UI thread.
+    ui_thread_running.Wait();
+    thread_host_->InvalidateActiveRunners();
+    ui_thread_block.Signal();
+
+    // Needed to keep ui_thread_block in scope until the UI thread execution
+    // finishes.
+    ui_thread_finished.Wait();
+  } else {
+    thread_host_->InvalidateActiveRunners();
+  }
+  thread_host_.reset();
+}
+
 bool EmbedderEngine::RunRootIsolate() {
   if (!IsValid() || !run_configuration_.IsValid()) {
     return false;
@@ -96,6 +136,7 @@ bool EmbedderEngine::NotifyDestroyed() {
   }
 
   shell_->GetPlatformView()->NotifyDestroyed();
+
   return true;
 }
 
@@ -243,7 +284,7 @@ bool EmbedderEngine::RunTask(const FlutterTask* task) {
   if (task == nullptr) {
     return false;
   }
-  auto result = thread_host_->PostTask(reinterpret_cast<int64_t>(task->runner),
+  auto result = thread_host_->PostTask(reinterpret_cast<intptr_t>(task->runner),
                                        task->task);
   // If the UI and platform threads are separate, the microtask queue is
   // flushed through MessageLoopTaskQueues observer.

engine/src/flutter/shell/platform/embedder/embedder_engine.h

@@ -38,6 +38,8 @@ class EmbedderEngine {
 
   bool CollectShell();
 
+  void CollectThreadHost();
+
   const TaskRunners& GetTaskRunners() const;
 
   bool NotifyCreated();
@@ -88,7 +90,7 @@ class EmbedderEngine {
   Shell& GetShell();
 
  private:
-  const std::unique_ptr<EmbedderThreadHost> thread_host_;
+  std::unique_ptr<EmbedderThreadHost> thread_host_;
   TaskRunners task_runners_;
   RunConfiguration run_configuration_;
   std::unique_ptr<ShellArgs> shell_args_;

engine/src/flutter/shell/platform/embedder/embedder_task_runner.cc

@@ -9,12 +9,15 @@
 
 namespace flutter {
 
+std::atomic_intptr_t EmbedderTaskRunner::next_unique_id_ = 0;
+
 EmbedderTaskRunner::EmbedderTaskRunner(DispatchTable table,
                                        size_t embedder_identifier)
     : TaskRunner(nullptr /* loop implemenation*/),
       embedder_identifier_(embedder_identifier),
       dispatch_table_(std::move(table)),
-      placeholder_id_(fml::TaskQueueId(fml::TaskQueueId::kInvalid)) {
+      placeholder_id_(fml::TaskQueueId(fml::TaskQueueId::kInvalid)),
+      unique_id_(next_unique_id_++) {
   FML_DCHECK(dispatch_table_.post_task_callback);
   FML_DCHECK(dispatch_table_.runs_task_on_current_thread_callback);
   FML_DCHECK(dispatch_table_.destruction_callback);

engine/src/flutter/shell/platform/embedder/embedder_task_runner.h

@@ -75,6 +75,8 @@ class EmbedderTaskRunner final : public fml::TaskRunner {
 
   bool PostTask(uint64_t baton);
 
+  intptr_t unique_id() const { return unique_id_; }
+
  private:
   const size_t embedder_identifier_;
   DispatchTable dispatch_table_;
@@ -82,6 +84,9 @@ class EmbedderTaskRunner final : public fml::TaskRunner {
   uint64_t last_baton_ = 0;
   std::unordered_map<uint64_t, fml::closure> pending_tasks_;
   fml::TaskQueueId placeholder_id_;
+  intptr_t unique_id_;
+
+  static std::atomic_intptr_t next_unique_id_;
 
   // |fml::TaskRunner|
   void PostTask(const fml::closure& task) override;

engine/src/flutter/shell/platform/embedder/embedder_thread_host.cc

@@ -13,6 +13,9 @@
 
 namespace flutter {
 
+std::set<intptr_t> EmbedderThreadHost::active_runners_;
+std::mutex EmbedderThreadHost::active_runners_mutex_;
+
 //------------------------------------------------------------------------------
 /// @brief      Attempts to create a task runner from an embedder task runner
 ///             description. The first boolean in the pair indicate whether the
@@ -67,7 +70,7 @@ CreateEmbedderTaskRunner(const FlutterTaskRunnerDescription* description) {
                                 fml::TimePoint target_time) -> void {
         FlutterTask task = {
             // runner
-            reinterpret_cast<FlutterTaskRunner>(task_runner),
+            reinterpret_cast<FlutterTaskRunner>(task_runner->unique_id()),
             // task
             task_baton,
         };
@@ -306,13 +309,27 @@ EmbedderThreadHost::EmbedderThreadHost(
     const flutter::TaskRunners& runners,
     const std::set<fml::RefPtr<EmbedderTaskRunner>>& embedder_task_runners)
     : host_(std::move(host)), runners_(runners) {
+  std::lock_guard guard(active_runners_mutex_);
   for (const auto& runner : embedder_task_runners) {
-    runners_map_[reinterpret_cast<int64_t>(runner.get())] = runner;
+    runners_map_[runner->unique_id()] = runner;
+    active_runners_.insert(runner->unique_id());
   }
 }
 
 EmbedderThreadHost::~EmbedderThreadHost() = default;
 
+void EmbedderThreadHost::InvalidateActiveRunners() {
+  std::lock_guard guard(active_runners_mutex_);
+  for (const auto& runner : runners_map_) {
+    active_runners_.erase(runner.first);
+  }
+}
+
+bool EmbedderThreadHost::RunnerIsValid(intptr_t runner) {
+  std::lock_guard guard(active_runners_mutex_);
+  return active_runners_.find(runner) != active_runners_.end();
+}
+
 bool EmbedderThreadHost::IsValid() const {
   return runners_.IsValid();
 }
@@ -321,7 +338,7 @@ const flutter::TaskRunners& EmbedderThreadHost::GetTaskRunners() const {
   return runners_;
 }
 
-bool EmbedderThreadHost::PostTask(int64_t runner, uint64_t task) const {
+bool EmbedderThreadHost::PostTask(intptr_t runner, uint64_t task) const {
   auto found = runners_map_.find(runner);
   if (found == runners_map_.end()) {
     return false;

engine/src/flutter/shell/platform/embedder/embedder_thread_host.h

@@ -36,12 +36,19 @@ class EmbedderThreadHost {
 
   const flutter::TaskRunners& GetTaskRunners() const;
 
-  bool PostTask(int64_t runner, uint64_t task) const;
+  bool PostTask(intptr_t runner, uint64_t task) const;
+
+  static bool RunnerIsValid(intptr_t runner);
+
+  void InvalidateActiveRunners();
 
  private:
   ThreadHost host_;
   flutter::TaskRunners runners_;
-  std::map<int64_t, fml::RefPtr<EmbedderTaskRunner>> runners_map_;
+  std::map<intptr_t, fml::RefPtr<EmbedderTaskRunner>> runners_map_;
+
+  static std::set<intptr_t> active_runners_;
+  static std::mutex active_runners_mutex_;
 
   static std::unique_ptr<EmbedderThreadHost> CreateEmbedderManagedThreadHost(
       const FlutterCustomTaskRunners* custom_task_runners,

engine/src/flutter/shell/platform/embedder/tests/embedder_unittests.cc

@@ -269,6 +269,73 @@ TEST_F(EmbedderTest, CanSpecifyCustomUITaskRunner) {
   kill_latch.Wait();
 }
 
+TEST_F(EmbedderTest, IgnoresStaleTasks) {
+  auto& context = GetEmbedderContext<EmbedderTestContextSoftware>();
+  auto ui_task_runner = CreateNewThread("test_ui_thread");
+  auto platform_task_runner = CreateNewThread("test_platform_thread");
+  static std::mutex engine_mutex;
+  UniqueEngine engine;
+  FlutterEngine engine_ptr;
+
+  EmbedderTestTaskRunner test_ui_task_runner(
+      ui_task_runner, [&](FlutterTask task) {
+        // The check for engine.is_valid() is intentionally absent here.
+        // FlutterEngineRunTask must be able to detect and ignore stale tasks
+        // without crashing even if the engine pointer is not null.
+        // Because the engine is destroyed on platform thread,
+        // relying solely on engine.is_valid() in UI thread is not safe.
+        FlutterEngineRunTask(engine_ptr, &task);
+      });
+  EmbedderTestTaskRunner test_platform_task_runner(
+      platform_task_runner, [&](FlutterTask task) {
+        std::scoped_lock lock(engine_mutex);
+        if (!engine.is_valid()) {
+          return;
+        }
+        FlutterEngineRunTask(engine.get(), &task);
+      });
+
+  fml::AutoResetWaitableEvent init_latch;
+
+  platform_task_runner->PostTask([&]() {
+    EmbedderConfigBuilder builder(context);
+    const auto ui_task_runner_description =
+        test_ui_task_runner.GetFlutterTaskRunnerDescription();
+    const auto platform_task_runner_description =
+        test_platform_task_runner.GetFlutterTaskRunnerDescription();
+    builder.SetUITaskRunner(&ui_task_runner_description);
+    builder.SetPlatformTaskRunner(&platform_task_runner_description);
+    {
+      std::scoped_lock lock(engine_mutex);
+      engine = builder.InitializeEngine();
+    }
+    init_latch.Signal();
+  });
+
+  init_latch.Wait();
+  engine_ptr = engine.get();
+
+  auto flutter_engine = reinterpret_cast<EmbedderEngine*>(engine.get());
+
+  // Schedule task on UI thread that will likely run after the engine has shut
+  // down.
+  flutter_engine->GetTaskRunners().GetUITaskRunner()->PostDelayedTask(
+      []() {}, fml::TimeDelta::FromMilliseconds(50));
+
+  fml::AutoResetWaitableEvent kill_latch;
+  platform_task_runner->PostTask([&] {
+    engine.reset();
+    platform_task_runner->PostTask([&kill_latch] { kill_latch.Signal(); });
+  });
+  kill_latch.Wait();
+
+  // Ensure that the schedule task indeed runs.
+  kill_latch.Reset();
+  ui_task_runner->PostDelayedTask([&]() { kill_latch.Signal(); },
+                                  fml::TimeDelta::FromMilliseconds(50));
+  kill_latch.Wait();
+}
+
 TEST_F(EmbedderTest, MergedPlatformUIThread) {
   auto& context = GetEmbedderContext<EmbedderTestContextSoftware>();
   auto task_runner = CreateNewThread("test_thread");