Ensure threads are merged when tearing down the Rasterizer

fml/message_loop_task_queues.cc

@@ -246,7 +246,7 @@ bool MessageLoopTaskQueues::Unmerge(TaskQueueId owner) {
 bool MessageLoopTaskQueues::Owns(TaskQueueId owner,
                                  TaskQueueId subsumed) const {
   std::lock_guard guard(queue_mutex_);
-  return subsumed == queue_entries_.at(owner)->owner_of || owner == subsumed;
+  return subsumed == queue_entries_.at(owner)->owner_of;
 }
 
 // Subsumed queues will never have pending tasks.

fml/message_loop_task_queues_unittests.cc

@@ -173,6 +173,13 @@ TEST(MessageLoopTaskQueue, NotifyObserversWhileCreatingQueues) {
   before_second_observer.Signal();
   notify_observers.join();
 }
+
+TEST(MessageLoopTaskQueue, QueueDoNotOwnItself) {
+  auto task_queue = fml::MessageLoopTaskQueues::GetInstance();
+  auto queue_id = task_queue->CreateTaskQueue();
+  ASSERT_FALSE(task_queue->Owns(queue_id, queue_id));
+}
+
 // TODO(chunhtai): This unit-test is flaky and sometimes fails asynchronizely
 // after the test has finished.
 // https://github.com/flutter/flutter/issues/43858

fml/raster_thread_merger.cc

@@ -21,6 +21,10 @@ RasterThreadMerger::RasterThreadMerger(fml::TaskQueueId platform_queue_id,
 }
 
 void RasterThreadMerger::MergeWithLease(size_t lease_term) {
+  if (TaskQueuesAreSame()) {
+    return;
+  }
+
   FML_DCHECK(lease_term > 0) << "lease_term should be positive.";
   std::scoped_lock lock(lease_term_mutex_);
   if (!IsMergedUnSafe()) {
@@ -32,6 +36,10 @@ void RasterThreadMerger::MergeWithLease(size_t lease_term) {
 }
 
 void RasterThreadMerger::UnMergeNow() {
+  if (TaskQueuesAreSame()) {
+    return;
+  }
+
   std::scoped_lock lock(lease_term_mutex_);
   lease_term_ = 0;
   bool success = task_queues_->Unmerge(platform_queue_id_);
@@ -51,6 +59,9 @@ bool RasterThreadMerger::IsOnRasterizingThread() {
 }
 
 void RasterThreadMerger::ExtendLeaseTo(size_t lease_term) {
+  if (TaskQueuesAreSame()) {
+    return;
+  }
   std::scoped_lock lock(lease_term_mutex_);
   FML_DCHECK(IsMergedUnSafe()) << "lease_term should be positive.";
   if (lease_term_ != kLeaseNotSet &&
@@ -65,16 +76,26 @@ bool RasterThreadMerger::IsMerged() {
 }
 
 bool RasterThreadMerger::IsMergedUnSafe() {
-  return lease_term_ > 0;
+  return lease_term_ > 0 || TaskQueuesAreSame();
+}
+
+bool RasterThreadMerger::TaskQueuesAreSame() {
+  return platform_queue_id_ == gpu_queue_id_;
 }
 
 void RasterThreadMerger::WaitUntilMerged() {
+  if (TaskQueuesAreSame()) {
+    return;
+  }
   FML_CHECK(IsOnPlatformThread());
   std::unique_lock<std::mutex> lock(lease_term_mutex_);
   merged_condition_.wait(lock, [&] { return IsMergedUnSafe(); });
 }
 
 RasterThreadStatus RasterThreadMerger::DecrementLease() {
+  if (TaskQueuesAreSame()) {
+    return RasterThreadStatus::kRemainsMerged;
+  }
   std::unique_lock<std::mutex> lock(lease_term_mutex_);
   if (!IsMergedUnSafe()) {
     return RasterThreadStatus::kRemainsUnmerged;

fml/raster_thread_merger.h

@@ -30,17 +30,29 @@ class RasterThreadMerger
   // When the caller merges with a lease term of say 2. The threads
   // are going to remain merged until 2 invocations of |DecreaseLease|,
   // unless an |ExtendLeaseTo| gets called.
+  //
+  // If the task queues are the same, we consider them statically merged.
+  // When task queues are statically merged this method becomes no-op.
   void MergeWithLease(size_t lease_term);
 
   // Un-merges the threads now, and resets the lease term to 0.
   //
   // Must be executed on the raster task runner.
+  //
+  // If the task queues are the same, we consider them statically merged.
+  // When task queues are statically merged, we never unmerge them and
+  // this method becomes no-op.
   void UnMergeNow();
 
+  // If the task queues are the same, we consider them statically merged.
+  // When task queues are statically merged this method becomes no-op.
   void ExtendLeaseTo(size_t lease_term);
 
   // Returns |RasterThreadStatus::kUnmergedNow| if this call resulted in
   // splitting the raster and platform threads. Reduces the lease term by 1.
+  //
+  // If the task queues are the same, we consider them statically merged.
+  // When task queues are statically merged this method becomes no-op.
   RasterThreadStatus DecrementLease();
 
   bool IsMerged();
@@ -71,6 +83,9 @@ class RasterThreadMerger
   std::mutex lease_term_mutex_;
 
   bool IsMergedUnSafe();
+  // The platform_queue_id and gpu_queue_id are exactly the same.
+  // We consider the threads are always merged and cannot be unmerged.
+  bool TaskQueuesAreSame();
 
   FML_FRIEND_REF_COUNTED_THREAD_SAFE(RasterThreadMerger);
   FML_FRIEND_MAKE_REF_COUNTED(RasterThreadMerger);

fml/raster_thread_merger_unittests.cc

@@ -262,55 +262,42 @@ TEST(RasterThreadMerger, WaitUntilMerged) {
   thread_raster.join();
 }
 
-TEST(RasterThreadMerger, WaitUntilMergedReturnsIfAlreadyMerged) {
-  fml::RefPtr<fml::RasterThreadMerger> raster_thread_merger;
-
-  fml::AutoResetWaitableEvent thread_merged_latch;
-  fml::MessageLoop* loop_platform = nullptr;
-  fml::AutoResetWaitableEvent latch_platform;
-  fml::AutoResetWaitableEvent term_platform;
-  fml::AutoResetWaitableEvent latch_wait_until_merged;
-  std::thread thread_platform([&]() {
+TEST(RasterThreadMerger, HandleTaskQueuesAreTheSame) {
+  fml::MessageLoop* loop1 = nullptr;
+  fml::AutoResetWaitableEvent latch1;
+  fml::AutoResetWaitableEvent term1;
+  std::thread thread1([&loop1, &latch1, &term1]() {
     fml::MessageLoop::EnsureInitializedForCurrentThread();
-    loop_platform = &fml::MessageLoop::GetCurrent();
-    latch_platform.Signal();
-    thread_merged_latch.Wait();
-    raster_thread_merger->WaitUntilMerged();
-    latch_wait_until_merged.Signal();
-    term_platform.Wait();
+    loop1 = &fml::MessageLoop::GetCurrent();
+    latch1.Signal();
+    term1.Wait();
   });
 
-  const int kNumFramesMerged = 5;
-  fml::MessageLoop* loop_raster = nullptr;
-  fml::AutoResetWaitableEvent term_raster;
-  std::thread thread_raster([&]() {
-    fml::MessageLoop::EnsureInitializedForCurrentThread();
-    loop_raster = &fml::MessageLoop::GetCurrent();
-    latch_platform.Wait();
-    fml::TaskQueueId qid_platform =
-        loop_platform->GetTaskRunner()->GetTaskQueueId();
-    fml::TaskQueueId qid_raster =
-        loop_raster->GetTaskRunner()->GetTaskQueueId();
-    raster_thread_merger =
-        fml::MakeRefCounted<fml::RasterThreadMerger>(qid_platform, qid_raster);
-    ASSERT_FALSE(raster_thread_merger->IsMerged());
-    raster_thread_merger->MergeWithLease(kNumFramesMerged);
-    thread_merged_latch.Signal();
-    term_raster.Wait();
-  });
+  latch1.Wait();
 
-  latch_wait_until_merged.Wait();
-  ASSERT_TRUE(raster_thread_merger->IsMerged());
+  fml::TaskQueueId qid1 = loop1->GetTaskRunner()->GetTaskQueueId();
+  fml::TaskQueueId qid2 = qid1;
+  const auto raster_thread_merger_ =
+      fml::MakeRefCounted<fml::RasterThreadMerger>(qid1, qid2);
+  // Statically merged.
+  ASSERT_TRUE(raster_thread_merger_->IsMerged());
+
+  // Test decrement lease and unmerge are both no-ops.
+  // The task queues should be always merged.
+  const int kNumFramesMerged = 5;
+  raster_thread_merger_->MergeWithLease(kNumFramesMerged);
 
   for (int i = 0; i < kNumFramesMerged; i++) {
-    ASSERT_TRUE(raster_thread_merger->IsMerged());
-    raster_thread_merger->DecrementLease();
+    ASSERT_TRUE(raster_thread_merger_->IsMerged());
+    raster_thread_merger_->DecrementLease();
   }
 
-  ASSERT_FALSE(raster_thread_merger->IsMerged());
+  ASSERT_TRUE(raster_thread_merger_->IsMerged());
 
-  term_platform.Signal();
-  term_raster.Signal();
-  thread_platform.join();
-  thread_raster.join();
+  // Wait until merged should also return immediately.
+  raster_thread_merger_->WaitUntilMerged();
+  ASSERT_TRUE(raster_thread_merger_->IsMerged());
+
+  term1.Signal();
+  thread1.join();
 }

shell/common/rasterizer.cc

@@ -62,11 +62,7 @@ void Rasterizer::Setup(std::unique_ptr<Surface> surface) {
   // TODO(sanjayc77): https://github.com/flutter/flutter/issues/53179. Add
   // support for raster thread merger for Fuchsia.
   if (surface_->GetExternalViewEmbedder() &&
-      surface_->GetExternalViewEmbedder()->SupportsDynamicThreadMerging() &&
-      // Don't create raster_thread_merger if platform and raster task runners
-      // are the same.
-      delegate_.GetTaskRunners().GetRasterTaskRunner() !=
-          delegate_.GetTaskRunners().GetPlatformTaskRunner()) {
+      surface_->GetExternalViewEmbedder()->SupportsDynamicThreadMerging()) {
     const auto platform_id =
         delegate_.GetTaskRunners().GetPlatformTaskRunner()->GetTaskQueueId();
     const auto gpu_id =
@@ -668,11 +664,6 @@ std::optional<size_t> Rasterizer::GetResourceCacheMaxBytes() const {
 }
 
 bool Rasterizer::EnsureThreadsAreMerged() {
-  // If threads are merged statically, always return true.
-  if (delegate_.GetTaskRunners().GetRasterTaskRunner() ==
-      delegate_.GetTaskRunners().GetPlatformTaskRunner()) {
-    return true;
-  }
   if (surface_ == nullptr || raster_thread_merger_.get() == nullptr) {
     return false;
   }