[engine] reland: more consistently flush message loops tasks (#56815)

Changes the following shell callbacks to flush the dart event loop:

* OnPlatformViewSetViewportMetrics
* OnPlatformViewDispatchPointerDataPacket
* OnPlatformViewDispatchPlatformMessage
* OnPlatformViewSetSemanticsEnabled
* OnPlatformViewSetAccessibilityFeatures
Using a new TaskRunner API RunNowAndFlushMessages. If the task runner can run tasks on the current thread, this will immediately invoke a callback and then post an empty task to the event loop to ensure dart listeners fire.

Unlike #56738, does not touch the vsync API - which looks like it depends on scheduling behavior today, at least for iOS.

Author: Jonah Williams

fml/task_runner.cc

@@ -52,6 +52,7 @@ bool TaskRunner::RunsTasksOnCurrentThread() {
                                                 loop_queue_id);
 }
 
+// static
 void TaskRunner::RunNowOrPostTask(const fml::RefPtr<fml::TaskRunner>& runner,
                                   const fml::closure& task) {
   FML_DCHECK(runner);
@@ -62,4 +63,20 @@ void TaskRunner::RunNowOrPostTask(const fml::RefPtr<fml::TaskRunner>& runner,
   }
 }
 
+// static
+void TaskRunner::RunNowAndFlushMessages(
+    const fml::RefPtr<fml::TaskRunner>& runner,
+    const fml::closure& task) {
+  FML_DCHECK(runner);
+  if (runner->RunsTasksOnCurrentThread()) {
+    task();
+    // Post an empty task to make the UI message loop run its task observers.
+    // The observers will execute any Dart microtasks queued by the platform
+    // message handler.
+    runner->PostTask([] {});
+  } else {
+    runner->PostTask(task);
+  }
+}
+
 }  // namespace fml

fml/task_runner.h

@@ -62,6 +62,14 @@ class TaskRunner : public fml::RefCountedThreadSafe<TaskRunner>,
   static void RunNowOrPostTask(const fml::RefPtr<fml::TaskRunner>& runner,
                                const fml::closure& task);
 
+  /// Like RunNowOrPostTask, except that if the task can be immediately
+  /// executed, an empty task will still be posted to the runner afterwards.
+  ///
+  /// This is used to ensure that messages posted to Dart from the platform
+  /// thread always flush the Dart event loop.
+  static void RunNowAndFlushMessages(const fml::RefPtr<fml::TaskRunner>& runner,
+                                     const fml::closure& task);
+
  protected:
   explicit TaskRunner(fml::RefPtr<MessageLoopImpl> loop);
 

shell/common/fixtures/shell_test.dart

@@ -640,3 +640,13 @@ void testSemanticsActions() {
     });
   };
 }
+
+@pragma('vm:entry-point')
+void testPointerActions() {
+  PlatformDispatcher.instance.onPointerDataPacket = (PointerDataPacket pointer) async {
+    await null;
+    Future<void>.value().then((_) {
+      notifyNative();
+    });
+  };
+}

shell/common/shell.cc

@@ -2,6 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "fml/task_runner.h"
 #define RAPIDJSON_HAS_STDSTRING 1
 #include "flutter/shell/common/shell.h"
 
@@ -1036,7 +1037,7 @@ void Shell::OnPlatformViewSetViewportMetrics(int64_t view_id,
         }
       });
 
-  fml::TaskRunner::RunNowOrPostTask(
+  fml::TaskRunner::RunNowAndFlushMessages(
       task_runners_.GetUITaskRunner(),
       [engine = engine_->GetWeakPtr(), view_id, metrics]() {
         if (engine) {
@@ -1075,24 +1076,16 @@ void Shell::OnPlatformViewDispatchPlatformMessage(
   }
 #endif  // FLUTTER_RUNTIME_MODE == FLUTTER_RUNTIME_MODE_DEBUG
 
-  if (task_runners_.GetUITaskRunner()->RunsTasksOnCurrentThread()) {
-    engine_->DispatchPlatformMessage(std::move(message));
-
-    // Post an empty task to make the UI message loop run its task observers.
-    // The observers will execute any Dart microtasks queued by the platform
-    // message handler.
-    task_runners_.GetUITaskRunner()->PostTask([] {});
-  } else {
-    // The static leak checker gets confused by the use of fml::MakeCopyable.
-    // NOLINTNEXTLINE(clang-analyzer-cplusplus.NewDeleteLeaks)
-    task_runners_.GetUITaskRunner()->PostTask(
-        fml::MakeCopyable([engine = engine_->GetWeakPtr(),
-                           message = std::move(message)]() mutable {
-          if (engine) {
-            engine->DispatchPlatformMessage(std::move(message));
-          }
-        }));
-  }
+  // The static leak checker gets confused by the use of fml::MakeCopyable.
+  // NOLINTNEXTLINE(clang-analyzer-cplusplus.NewDeleteLeaks)
+  fml::TaskRunner::RunNowAndFlushMessages(
+      task_runners_.GetUITaskRunner(),
+      fml::MakeCopyable([engine = engine_->GetWeakPtr(),
+                         message = std::move(message)]() mutable {
+        if (engine) {
+          engine->DispatchPlatformMessage(std::move(message));
+        }
+      }));
 }
 
 // |PlatformView::Delegate|
@@ -1104,7 +1097,7 @@ void Shell::OnPlatformViewDispatchPointerDataPacket(
   TRACE_FLOW_BEGIN("flutter", "PointerEvent", next_pointer_flow_id_);
   FML_DCHECK(is_set_up_);
   FML_DCHECK(task_runners_.GetPlatformTaskRunner()->RunsTasksOnCurrentThread());
-  fml::TaskRunner::RunNowOrPostTask(
+  fml::TaskRunner::RunNowAndFlushMessages(
       task_runners_.GetUITaskRunner(),
       fml::MakeCopyable([engine = weak_engine_, packet = std::move(packet),
                          flow_id = next_pointer_flow_id_]() mutable {
@@ -1122,7 +1115,8 @@ void Shell::OnPlatformViewDispatchSemanticsAction(int32_t node_id,
   FML_DCHECK(is_set_up_);
   FML_DCHECK(task_runners_.GetPlatformTaskRunner()->RunsTasksOnCurrentThread());
 
-  task_runners_.GetUITaskRunner()->PostTask(
+  fml::TaskRunner::RunNowAndFlushMessages(
+      task_runners_.GetUITaskRunner(),
       fml::MakeCopyable([engine = engine_->GetWeakPtr(), node_id, action,
                          args = std::move(args)]() mutable {
         if (engine) {
@@ -1136,25 +1130,26 @@ void Shell::OnPlatformViewSetSemanticsEnabled(bool enabled) {
   FML_DCHECK(is_set_up_);
   FML_DCHECK(task_runners_.GetPlatformTaskRunner()->RunsTasksOnCurrentThread());
 
-  fml::TaskRunner::RunNowOrPostTask(task_runners_.GetUITaskRunner(),
-                                    [engine = engine_->GetWeakPtr(), enabled] {
-                                      if (engine) {
-                                        engine->SetSemanticsEnabled(enabled);
-                                      }
-                                    });
+  fml::TaskRunner::RunNowAndFlushMessages(
+      task_runners_.GetUITaskRunner(),
+      [engine = engine_->GetWeakPtr(), enabled] {
+        if (engine) {
+          engine->SetSemanticsEnabled(enabled);
+        }
+      });
 }
 
 // |PlatformView::Delegate|
 void Shell::OnPlatformViewSetAccessibilityFeatures(int32_t flags) {
   FML_DCHECK(is_set_up_);
   FML_DCHECK(task_runners_.GetPlatformTaskRunner()->RunsTasksOnCurrentThread());
 
-  fml::TaskRunner::RunNowOrPostTask(task_runners_.GetUITaskRunner(),
-                                    [engine = engine_->GetWeakPtr(), flags] {
-                                      if (engine) {
-                                        engine->SetAccessibilityFeatures(flags);
-                                      }
-                                    });
+  fml::TaskRunner::RunNowAndFlushMessages(
+      task_runners_.GetUITaskRunner(), [engine = engine_->GetWeakPtr(), flags] {
+        if (engine) {
+          engine->SetAccessibilityFeatures(flags);
+        }
+      });
 }
 
 // |PlatformView::Delegate|

shell/common/shell_unittests.cc

@@ -4312,7 +4312,8 @@ TEST_F(ShellTest, NavigationMessageDispachedImmediately) {
   ASSERT_FALSE(DartVMRef::IsInstanceRunning());
 }
 
-TEST_F(ShellTest, SemanticsActionsPostTask) {
+// Verifies a semantics Action will flush the dart event loop.
+TEST_F(ShellTest, SemanticsActionsFlushMessageLoop) {
   Settings settings = CreateSettingsForFixture();
   ThreadHost thread_host("io.flutter.test." + GetCurrentTestName() + ".",
                          ThreadHost::Type::kPlatform);
@@ -4327,20 +4328,49 @@ TEST_F(ShellTest, SemanticsActionsPostTask) {
   configuration.SetEntrypoint("testSemanticsActions");
 
   RunEngine(shell.get(), std::move(configuration));
+  fml::CountDownLatch latch(1);
+  AddNativeCallback(
+      // The Dart native function names aren't very consistent but this is
+      // just the native function name of the second vm entrypoint in the
+      // fixture.
+      "NotifyNative",
+      CREATE_NATIVE_ENTRY([&](auto args) { latch.CountDown(); }));
 
   task_runners.GetPlatformTaskRunner()->PostTask([&] {
     SendSemanticsAction(shell.get(), 0, SemanticsAction::kTap,
                         fml::MallocMapping(nullptr, 0));
   });
+  latch.Wait();
 
-  // Fulfill native function for the second Shell's entrypoint.
+  DestroyShell(std::move(shell), task_runners);
+  ASSERT_FALSE(DartVMRef::IsInstanceRunning());
+}
+
+// Verifies a pointer event will flush the dart event loop.
+TEST_F(ShellTest, PointerPacketFlushMessageLoop) {
+  Settings settings = CreateSettingsForFixture();
+  ThreadHost thread_host("io.flutter.test." + GetCurrentTestName() + ".",
+                         ThreadHost::Type::kPlatform);
+  auto task_runner = thread_host.platform_thread->GetTaskRunner();
+  TaskRunners task_runners("test", task_runner, task_runner, task_runner,
+                           task_runner);
+
+  EXPECT_EQ(task_runners.GetPlatformTaskRunner(),
+            task_runners.GetUITaskRunner());
+  auto shell = CreateShell(settings, task_runners);
+  auto configuration = RunConfiguration::InferFromSettings(settings);
+  configuration.SetEntrypoint("testPointerActions");
+
+  RunEngine(shell.get(), std::move(configuration));
   fml::CountDownLatch latch(1);
   AddNativeCallback(
       // The Dart native function names aren't very consistent but this is
       // just the native function name of the second vm entrypoint in the
       // fixture.
       "NotifyNative",
       CREATE_NATIVE_ENTRY([&](auto args) { latch.CountDown(); }));
+
+  DispatchFakePointerData(shell.get(), 23);
   latch.Wait();
 
   DestroyShell(std::move(shell), task_runners);

shell/common/vsync_waiter.cc

@@ -127,7 +127,6 @@ void VsyncWaiter::FireCallback(fml::TimePoint frame_start_time,
 
     fml::TaskQueueId ui_task_queue_id =
         task_runners_.GetUITaskRunner()->GetTaskQueueId();
-
     task_runners_.GetUITaskRunner()->PostTask(
         [ui_task_queue_id, callback, flow_identifier, frame_start_time,
          frame_target_time, pause_secondary_tasks]() {