Skip to content
This repository was archived by the owner on Feb 25, 2025. It is now read-only.

Add synchronization mechanism to avoid deadlock caused by thread merging #30314

Closed
wants to merge 3 commits into from
Closed

Add synchronization mechanism to avoid deadlock caused by thread merging #30314

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c0c591f
Fix platformview deadlock.
0xZOne Dec 23, 2021
7973865
rename global lock
0xZOne Dec 24, 2021
da35807
Add synchronous |PostTask| function.
0xZOne Jan 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions fml/raster_thread_merger.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,20 +50,36 @@ 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.";
{
std::scoped_lock lock(mutex_);
if (IsMergedUnSafe()) {
merged_condition_.notify_one();
return;
}
}

if (IsMergedUnSafe()) {
merged_condition_.notify_one();
return;
// Here we flush all expired tasks in the raster task queue before thread
// merging to ensure these sync tasks the platform thread post has been
// executed.
// TODO(0xZOne): Maybe we should only flush expired sync tasks posted by the
// platform thread.
if (MessageLoop::IsInitializedForCurrentThread()) {
if (!IsOnPlatformThread()) {
MessageLoop::GetCurrent().RunExpiredTasksNow();
}
Comment on lines +72 to +77
Copy link
Contributor

@dnfield dnfield Jan 11, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@iskakaushik @cyanglaz - do we want to specialize this to the PlatformThread?

Do we not also need to flush any raster tasks?

Should we just be flushing both queues first?

Copy link
Contributor

@cyanglaz cyanglaz Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry about the delay. I just cycled back to this today.
We actually cannot flush raster tasks before thread merging. Here's why:

Let's say we have 2 tasks in the raster task queue waiting to be executed. Frame 1 and Frame 2.
Each task represents a frame that will be drawn on the screen. (For example, a scroll animation or a new page transition animation)

Now frame 1 gave us the signal that a thread merge needs to happen (there is a platform view appearing in frame 1).

What we do today is that we drop frame 1 after pre-roll (that's when we know a platform view is in the layer_tree), create a new copy of frame 1, merge the thread, run frame 1 and frame 2 on platform thread.

So we cannot just flush frame 1 and frame 2 on the raster thread because they have to be run on the platform thread. Even when frame 2 doesn't need to run on platform thread (like how the example in the issue is), we still need to run frame 1 on the platform thread.

}

// To avoid data races between the platform thread posting sync tasks to the
// raster thread and the raster thread merging, we add a lock here.
std::scoped_lock lock{mutex_, g_thread_merging_lock};
bool success = shared_merger_->MergeWithLease(this, lease_term);
if (success && merge_unmerge_callback_ != nullptr) {
merge_unmerge_callback_();
Expand Down
89 changes: 81 additions & 8 deletions fml/raster_thread_merger_unittests.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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
18 changes: 18 additions & 0 deletions fml/task_runner.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,12 @@
#include "flutter/fml/message_loop.h"
#include "flutter/fml/message_loop_impl.h"
#include "flutter/fml/message_loop_task_queues.h"
#include "flutter/fml/synchronization/waitable_event.h"

namespace fml {

std::mutex g_thread_merging_lock;
Copy link
Member

@chinmaygarde chinmaygarde Feb 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can't have global locks when working with non-singleton instances.


TaskRunner::TaskRunner(fml::RefPtr<MessageLoopImpl> loop)
: loop_(std::move(loop)) {}

Expand Down Expand Up @@ -62,4 +65,19 @@ void TaskRunner::RunNowOrPostTask(fml::RefPtr<fml::TaskRunner> runner,
}
}

void TaskRunner::RunNowOrPostTaskSync(fml::RefPtr<fml::TaskRunner> runner,
const fml::closure& task) {
FML_DCHECK(runner);
std::scoped_lock lock(g_thread_merging_lock);
if (runner->RunsTasksOnCurrentThread()) {
task();
} else {
fml::AutoResetWaitableEvent latch;
runner->PostTask([&] {
task();
latch.Signal();
});
latch.Wait();
}
}
} // namespace fml
9 changes: 9 additions & 0 deletions fml/task_runner.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,10 @@ namespace fml {

class MessageLoopImpl;

/// To avoid data races between the platform thread posting sync tasks to the
/// raster thread and the raster thread merging.
extern std::mutex g_thread_merging_lock;
Copy link
Member

@chinmaygarde chinmaygarde Feb 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When adding a mutex, I would ask myself what the critical section is. From the documentation, it is not clear what that is.


/// An interface over the ability to schedule tasks on a \p TaskRunner.
class BasicTaskRunner {
public:
Expand Down Expand Up @@ -62,6 +66,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 mutex with thread
/// merging operation and post the \p tast synchronously.
static void RunNowOrPostTaskSync(fml::RefPtr<fml::TaskRunner> runner,
const fml::closure& task);

protected:
explicit TaskRunner(fml::RefPtr<MessageLoopImpl> loop);

Expand Down
8 changes: 3 additions & 5 deletions shell/common/platform_view.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,16 +62,14 @@ void PlatformView::NotifyCreated() {
// Using the weak pointer is illegal. But, we are going to introduce a latch
// so that the platform view is not collected till the surface is obtained.
auto* platform_view = this;
fml::ManualResetWaitableEvent latch;
fml::TaskRunner::RunNowOrPostTask(
task_runners_.GetRasterTaskRunner(), [platform_view, &surface, &latch]() {
fml::TaskRunner::RunNowOrPostTaskSync(
task_runners_.GetRasterTaskRunner(), [platform_view, &surface]() {
surface = platform_view->CreateRenderingSurface();
if (surface && !surface->IsValid()) {
surface.reset();
}
latch.Signal();
});
latch.Wait();

if (!surface) {
FML_LOG(ERROR) << "Failed to create platform view rendering surface";
return;
Expand Down
36 changes: 12 additions & 24 deletions shell/platform/android/platform_view_android.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,15 +97,12 @@ void PlatformViewAndroid::NotifyCreated(
if (android_surface_) {
InstallFirstFrameCallback();

fml::AutoResetWaitableEvent latch;
fml::TaskRunner::RunNowOrPostTask(
fml::TaskRunner::RunNowOrPostTaskSync(
task_runners_.GetRasterTaskRunner(),
[&latch, surface = android_surface_.get(),
[surface = android_surface_.get(),
native_window = std::move(native_window)]() {
surface->SetNativeWindow(native_window);
latch.Signal();
});
latch.Wait();
}

PlatformView::NotifyCreated();
Expand All @@ -114,46 +111,37 @@ void PlatformViewAndroid::NotifyCreated(
void PlatformViewAndroid::NotifySurfaceWindowChanged(
fml::RefPtr<AndroidNativeWindow> native_window) {
if (android_surface_) {
fml::AutoResetWaitableEvent latch;
fml::TaskRunner::RunNowOrPostTask(
fml::TaskRunner::RunNowOrPostTaskSync(
task_runners_.GetRasterTaskRunner(),
[&latch, surface = android_surface_.get(),
[surface = android_surface_.get(),
native_window = std::move(native_window)]() {
surface->TeardownOnScreenContext();
surface->SetNativeWindow(native_window);
latch.Signal();
});
latch.Wait();
}
}

void PlatformViewAndroid::NotifyDestroyed() {
PlatformView::NotifyDestroyed();

if (android_surface_) {
fml::AutoResetWaitableEvent latch;
fml::TaskRunner::RunNowOrPostTask(
task_runners_.GetRasterTaskRunner(),
[&latch, surface = android_surface_.get()]() {
surface->TeardownOnScreenContext();
latch.Signal();
});
latch.Wait();
fml::TaskRunner::RunNowOrPostTaskSync(task_runners_.GetRasterTaskRunner(),
[surface = android_surface_.get()]() {
surface->TeardownOnScreenContext();
});
}
}

void PlatformViewAndroid::NotifyChanged(const SkISize& size) {
if (!android_surface_) {
return;
}
fml::AutoResetWaitableEvent latch;
fml::TaskRunner::RunNowOrPostTask(
task_runners_.GetRasterTaskRunner(), //
[&latch, surface = android_surface_.get(), size]() {

fml::TaskRunner::RunNowOrPostTaskSync(
task_runners_.GetRasterTaskRunner(),
[surface = android_surface_.get(), size]() {
surface->OnScreenSurfaceResize(size);
latch.Signal();
});
latch.Wait();
}

void PlatformViewAndroid::DispatchPlatformMessage(JNIEnv* env,
Expand Down