Fix platformview deadlock.

Author: 0xZOne

fml/message_loop_task_queues.cc

@@ -11,11 +11,14 @@
 #include <optional>
 
 #include "flutter/fml/make_copyable.h"
+#include "flutter/fml/message_loop.h"
 #include "flutter/fml/task_source.h"
 #include "flutter/fml/thread_local.h"
 
 namespace fml {
 
+std::mutex gThreadMergingLock;
+
 std::mutex MessageLoopTaskQueues::creation_mutex_;
 
 const size_t TaskQueueId::kUnmerged = ULONG_MAX;
@@ -127,29 +130,7 @@ bool MessageLoopTaskQueues::HasPendingTasks(TaskQueueId queue_id) const {
 fml::closure MessageLoopTaskQueues::GetNextTaskToRun(TaskQueueId queue_id,
                                                      fml::TimePoint from_time) {
   std::lock_guard guard(queue_mutex_);
-  if (!HasPendingTasksUnlocked(queue_id)) {
-    return nullptr;
-  }
-  TaskSource::TopTask top = PeekNextTaskUnlocked(queue_id);
-
-  if (!HasPendingTasksUnlocked(queue_id)) {
-    WakeUpUnlocked(queue_id, fml::TimePoint::Max());
-  } else {
-    WakeUpUnlocked(queue_id, GetNextWakeTimeUnlocked(queue_id));
-  }
-
-  if (top.task.GetTargetTime() > from_time) {
-    return nullptr;
-  }
-  fml::closure invocation = top.task.GetTask();
-  queue_entries_.at(top.task_queue_id)
-      ->task_source->PopTask(top.task.GetTaskSourceGrade());
-  {
-    std::scoped_lock creation(creation_mutex_);
-    const auto task_source_grade = top.task.GetTaskSourceGrade();
-    tls_task_source_grade.reset(new TaskSourceGradeHolder{task_source_grade});
-  }
-  return invocation;
+  return GetNextTaskToRunUnlocked(queue_id, from_time);
 }
 
 void MessageLoopTaskQueues::WakeUpUnlocked(TaskQueueId queue_id,
@@ -194,24 +175,7 @@ void MessageLoopTaskQueues::RemoveTaskObserver(TaskQueueId queue_id,
 std::vector<fml::closure> MessageLoopTaskQueues::GetObserversToNotify(
     TaskQueueId queue_id) const {
   std::lock_guard guard(queue_mutex_);
-  std::vector<fml::closure> observers;
-
-  if (queue_entries_.at(queue_id)->subsumed_by != _kUnmerged) {
-    return observers;
-  }
-
-  for (const auto& observer : queue_entries_.at(queue_id)->task_observers) {
-    observers.push_back(observer.second);
-  }
-
-  auto& subsumed_set = queue_entries_.at(queue_id)->owner_of;
-  for (auto& subsumed : subsumed_set) {
-    for (const auto& observer : queue_entries_.at(subsumed)->task_observers) {
-      observers.push_back(observer.second);
-    }
-  }
-
-  return observers;
+  return GetObserversToNotifyUnlocked(queue_id);
 }
 
 void MessageLoopTaskQueues::SetWakeable(TaskQueueId queue_id,
@@ -262,7 +226,9 @@ bool MessageLoopTaskQueues::Merge(TaskQueueId owner, TaskQueueId subsumed) {
                      << subsumed_entry->subsumed_by;
     return false;
   }
+
   // All checking is OK, set merged state.
+  std::scoped_lock lock(gThreadMergingLock);
   owner_entry->owner_of.insert(subsumed);
   subsumed_entry->subsumed_by = owner;
 
@@ -319,11 +285,7 @@ bool MessageLoopTaskQueues::Unmerge(TaskQueueId owner, TaskQueueId subsumed) {
 bool MessageLoopTaskQueues::Owns(TaskQueueId owner,
                                  TaskQueueId subsumed) const {
   std::lock_guard guard(queue_mutex_);
-  if (owner == _kUnmerged || subsumed == _kUnmerged) {
-    return false;
-  }
-  auto& subsumed_set = queue_entries_.at(owner)->owner_of;
-  return subsumed_set.find(subsumed) != subsumed_set.end();
+  return OwnsUnlocked(owner, subsumed);
 }
 
 std::set<TaskQueueId> MessageLoopTaskQueues::GetSubsumedTaskQueueId(
@@ -346,6 +308,33 @@ void MessageLoopTaskQueues::ResumeSecondarySource(TaskQueueId queue_id) {
   }
 }
 
+void MessageLoopTaskQueues::FlushExpiredTasksNow(TaskQueueId queue_id) {
+  std::lock_guard guard(queue_mutex_);
+  if (!MessageLoop::IsInitializedForCurrentThread()) {
+    return;
+  }
+
+  TaskQueueId current_queue_id = MessageLoop::GetCurrentTaskQueueId();
+  if (current_queue_id == queue_id ||
+      OwnsUnlocked(current_queue_id, queue_id) ||
+      OwnsUnlocked(queue_id, current_queue_id)) {
+    const auto now = fml::TimePoint::Now();
+    fml::closure invocation;
+    do {
+      invocation = GetNextTaskToRunUnlocked(queue_id, now);
+      if (!invocation) {
+        return;
+      }
+      invocation();
+      std::vector<fml::closure> observers =
+          GetObserversToNotifyUnlocked(queue_id);
+      for (const auto& observer : observers) {
+        observer();
+      }
+    } while (invocation);
+  }
+}
+
 // Subsumed queues will never have pending tasks.
 // Owning queues will consider both their and their subsumed tasks.
 bool MessageLoopTaskQueues::HasPendingTasksUnlocked(
@@ -407,4 +396,64 @@ TaskSource::TopTask MessageLoopTaskQueues::PeekNextTaskUnlocked(
   return top_task.value();
 }
 
+bool MessageLoopTaskQueues::OwnsUnlocked(TaskQueueId owner,
+                                         TaskQueueId subsumed) const {
+  if (owner == _kUnmerged || subsumed == _kUnmerged) {
+    return false;
+  }
+  auto& subsumed_set = queue_entries_.at(owner)->owner_of;
+  return subsumed_set.find(subsumed) != subsumed_set.end();
+}
+
+fml::closure MessageLoopTaskQueues::GetNextTaskToRunUnlocked(
+    TaskQueueId queue_id,
+    fml::TimePoint from_time) {
+  if (!HasPendingTasksUnlocked(queue_id)) {
+    return nullptr;
+  }
+
+  TaskSource::TopTask top = PeekNextTaskUnlocked(queue_id);
+
+  if (!HasPendingTasksUnlocked(queue_id)) {
+    WakeUpUnlocked(queue_id, fml::TimePoint::Max());
+  } else {
+    WakeUpUnlocked(queue_id, GetNextWakeTimeUnlocked(queue_id));
+  }
+
+  if (top.task.GetTargetTime() > from_time) {
+    return nullptr;
+  }
+
+  fml::closure invocation = top.task.GetTask();
+  queue_entries_.at(top.task_queue_id)
+      ->task_source->PopTask(top.task.GetTaskSourceGrade());
+  {
+    std::scoped_lock creation(creation_mutex_);
+    const auto task_source_grade = top.task.GetTaskSourceGrade();
+    tls_task_source_grade.reset(new TaskSourceGradeHolder{task_source_grade});
+  }
+  return invocation;
+}
+
+std::vector<fml::closure> MessageLoopTaskQueues::GetObserversToNotifyUnlocked(
+    TaskQueueId queue_id) const {
+  std::vector<fml::closure> observers;
+
+  if (queue_entries_.at(queue_id)->subsumed_by != _kUnmerged) {
+    return observers;
+  }
+
+  for (const auto& observer : queue_entries_.at(queue_id)->task_observers) {
+    observers.push_back(observer.second);
+  }
+
+  auto& subsumed_set = queue_entries_.at(queue_id)->owner_of;
+  for (auto& subsumed : subsumed_set) {
+    for (const auto& observer : queue_entries_.at(subsumed)->task_observers) {
+      observers.push_back(observer.second);
+    }
+  }
+
+  return observers;
+}
 }  // namespace fml

fml/message_loop_task_queues.h

@@ -23,6 +23,10 @@ namespace fml {
 
 static const TaskQueueId _kUnmerged = TaskQueueId(TaskQueueId::kUnmerged);
 
+/// To avoid data races between the platform thread posting sync tasks to the
+/// raster thread and the raster thread merging.
+extern std::mutex gThreadMergingLock;
+
 /// A collection of tasks and observers associated with one TaskQueue.
 ///
 /// Often a TaskQueue has a one-to-one relationship with a fml::MessageLoop,
@@ -137,6 +141,8 @@ class MessageLoopTaskQueues
 
   void ResumeSecondarySource(TaskQueueId queue_id);
 
+  void FlushExpiredTasksNow(TaskQueueId queue_id);
+
  private:
   class MergedQueuesRunner;
 
@@ -152,6 +158,14 @@ class MessageLoopTaskQueues
 
   fml::TimePoint GetNextWakeTimeUnlocked(TaskQueueId queue_id) const;
 
+  bool OwnsUnlocked(TaskQueueId owner, TaskQueueId subsumed) const;
+
+  fml::closure GetNextTaskToRunUnlocked(TaskQueueId queue_id,
+                                        fml::TimePoint from_time);
+
+  std::vector<fml::closure> GetObserversToNotifyUnlocked(
+      TaskQueueId queue_id) const;
+
   static std::mutex creation_mutex_;
   static fml::RefPtr<MessageLoopTaskQueues> instance_;
 

fml/raster_thread_merger.cc

@@ -50,15 +50,19 @@ RasterThreadMerger::CreateOrShareThreadMerger(
 }
 
 void RasterThreadMerger::MergeWithLease(size_t lease_term) {
-  std::scoped_lock lock(mutex_);
   if (TaskQueuesAreSame()) {
     return;
   }
-  if (!IsEnabledUnSafe()) {
+  if (!IsEnabled()) {
     return;
   }
-  FML_DCHECK(lease_term > 0) << "lease_term should be positive.";
 
+  // Here we flush the raster task runner before thead merging to make
+  // sure these sync tasks the platform thread post has been executed.
+  shared_merger_->FlushSubsumedTaskQueueNow();
+
+  std::scoped_lock lock(mutex_);
+  FML_DCHECK(lease_term > 0) << "lease_term should be positive.";
   if (IsMergedUnSafe()) {
     merged_condition_.notify_one();
     return;

fml/raster_thread_merger_unittests.cc

@@ -413,14 +413,15 @@ TEST(RasterThreadMerger, RunExpiredTasksWhileFirstTaskMergesThreads) {
       raster_thread_merger->MergeWithLease(1);
       post_merge.CountDown();
     });
-
-    loop_raster->GetTaskRunner()->PostTask([&]() {
-      ASSERT_TRUE(raster_thread_merger->IsOnRasterizingThread());
-      ASSERT_TRUE(raster_thread_merger->IsOnPlatformThread());
-      ASSERT_EQ(fml::MessageLoop::GetCurrentTaskQueueId(), qid_platform);
-      raster_thread_merger->DecrementLease();
-      post_merge.CountDown();
-    });
+    loop_raster->GetTaskRunner()->PostDelayedTask(
+        [&]() {
+          ASSERT_TRUE(raster_thread_merger->IsOnRasterizingThread());
+          ASSERT_TRUE(raster_thread_merger->IsOnPlatformThread());
+          ASSERT_EQ(fml::MessageLoop::GetCurrentTaskQueueId(), qid_platform);
+          raster_thread_merger->DecrementLease();
+          post_merge.CountDown();
+        },
+        fml::TimeDelta::FromMilliseconds(100));
 
     loop_raster->RunExpiredTasksNow();
     post_merge.Wait();
@@ -657,5 +658,77 @@ TEST(RasterThreadMerger,
   ASSERT_FALSE(merger_from_3_to_1->IsMerged());
 }
 
+TEST(RasterThreadMerger, FlushAllExpiredTasksBeforeThreadMerging) {
+  fml::MessageLoop* loop_platform = nullptr;
+  fml::AutoResetWaitableEvent latch1;
+  std::thread thread_platform([&loop_platform, &latch1]() {
+    fml::MessageLoop::EnsureInitializedForCurrentThread();
+    loop_platform = &fml::MessageLoop::GetCurrent();
+    loop_platform->GetTaskRunner()->PostTask([&]() { latch1.Signal(); });
+    loop_platform->Run();
+  });
+
+  fml::MessageLoop* loop_raster = nullptr;
+  fml::AutoResetWaitableEvent latch2;
+  std::thread thread_raster([&loop_raster, &loop_platform, &latch1, &latch2]() {
+    latch1.Wait();
+
+    fml::MessageLoop::EnsureInitializedForCurrentThread();
+    loop_raster = &fml::MessageLoop::GetCurrent();
+    fml::TaskQueueId qid_platform =
+        loop_platform->GetTaskRunner()->GetTaskQueueId();
+    fml::TaskQueueId qid_raster =
+        loop_raster->GetTaskRunner()->GetTaskQueueId();
+    fml::CountDownLatch post_merge(4);
+    const auto raster_thread_merger =
+        fml::MakeRefCounted<fml::RasterThreadMerger>(qid_platform, qid_raster);
+
+    int order = 0;
+    loop_raster->GetTaskRunner()->PostTask([&]() {
+      ASSERT_TRUE(raster_thread_merger->IsOnRasterizingThread());
+      ASSERT_FALSE(raster_thread_merger->IsOnPlatformThread());
+      ASSERT_EQ(fml::MessageLoop::GetCurrentTaskQueueId(), qid_raster);
+      raster_thread_merger->MergeWithLease(1);
+      ASSERT_EQ(3, ++order);
+      post_merge.CountDown();
+    });
+
+    loop_raster->GetTaskRunner()->PostTask([&]() {
+      ASSERT_TRUE(raster_thread_merger->IsOnRasterizingThread());
+      ASSERT_FALSE(raster_thread_merger->IsOnPlatformThread());
+      ASSERT_EQ(1, ++order);
+      post_merge.CountDown();
+    });
+
+    loop_raster->GetTaskRunner()->PostTask([&]() {
+      ASSERT_TRUE(raster_thread_merger->IsOnRasterizingThread());
+      ASSERT_FALSE(raster_thread_merger->IsOnPlatformThread());
+      ASSERT_EQ(2, ++order);
+      post_merge.CountDown();
+    });
+
+    loop_raster->GetTaskRunner()->PostDelayedTask(
+        [&]() {
+          ASSERT_TRUE(raster_thread_merger->IsOnRasterizingThread());
+          ASSERT_TRUE(raster_thread_merger->IsOnPlatformThread());
+          ASSERT_EQ(fml::MessageLoop::GetCurrentTaskQueueId(), qid_platform);
+          ASSERT_EQ(4, ++order);
+          raster_thread_merger->DecrementLease();
+          post_merge.CountDown();
+        },
+        fml::TimeDelta::FromMilliseconds(100));
+
+    loop_raster->RunExpiredTasksNow();
+    post_merge.Wait();
+    latch2.Signal();
+  });
+
+  latch2.Wait();
+  loop_platform->GetTaskRunner()->PostTask(
+      [&]() { loop_platform->Terminate(); });
+
+  thread_platform.join();
+  thread_raster.join();
+}
 }  // namespace testing
 }  // namespace fml

fml/shared_thread_merger.cc

@@ -31,6 +31,14 @@ bool SharedThreadMerger::MergeWithLease(RasterThreadMergerId caller,
   return success;
 }
 
+void SharedThreadMerger::FlushSubsumedTaskQueueNow() {
+  std::scoped_lock lock(mutex_);
+  if (IsMergedUnSafe()) {
+    return;
+  }
+  task_queues_->FlushExpiredTasksNow(subsumed_);
+}
+
 bool SharedThreadMerger::UnMergeNowUnSafe() {
   FML_CHECK(IsAllLeaseTermsZeroUnSafe())
       << "all lease term records must be zero before calling "

fml/shared_thread_merger.h

@@ -54,6 +54,8 @@ class SharedThreadMerger
   // See the doc of |RasterThreadMerger::DecrementLease|.
   bool DecrementLease(RasterThreadMergerId caller);
 
+  void FlushSubsumedTaskQueueNow();
+
  private:
   fml::TaskQueueId owner_;
   fml::TaskQueueId subsumed_;

fml/task_runner.cc

@@ -62,4 +62,9 @@ void TaskRunner::RunNowOrPostTask(fml::RefPtr<fml::TaskRunner> runner,
   }
 }
 
+void TaskRunner::RunNowOrPostSyncTask(fml::RefPtr<fml::TaskRunner> runner,
+                                      const fml::closure& task) {
+  std::scoped_lock lock(gThreadMergingLock);
+  RunNowOrPostTask(runner, task);
+}
 }  // namespace fml

fml/task_runner.h

@@ -62,6 +62,11 @@ class TaskRunner : public fml::RefCountedThreadSafe<TaskRunner>,
   static void RunNowOrPostTask(fml::RefPtr<fml::TaskRunner> runner,
                                const fml::closure& task);
 
+  /// Similar to |RunNowOrPostTask|, but it needs to synchronize with thread
+  /// merging operation.
+  static void RunNowOrPostSyncTask(fml::RefPtr<fml::TaskRunner> runner,
+                                   const fml::closure& task);
+
  protected:
   explicit TaskRunner(fml::RefPtr<MessageLoopImpl> loop);