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

Rework image & texture management to use concurrent message queues. #9486

Merged
merged 4 commits into from
Jul 9, 2019
Merged

Rework image & texture management to use concurrent message queues. #9486

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1619752
Rework image & texture management to use concurrent message queues.
chinmaygarde May 31, 2019
0e39907
Licenses
chinmaygarde Jul 9, 2019
125e056
Dont use MakeFromFD
chinmaygarde Jul 9, 2019
d3281f3
PR
chinmaygarde Jul 9, 2019
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
1 change: 1 addition & 0 deletions BUILD.gn
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ group("flutter") {
public_deps += [
"$flutter_root/flow:flow_unittests",
"$flutter_root/fml:fml_unittests",
"$flutter_root/lib/ui:ui_unittests",
"$flutter_root/runtime:runtime_unittests",
"$flutter_root/shell/common:shell_unittests",
"$flutter_root/shell/platform/common/cpp/client_wrapper:client_wrapper_unittests",
Expand Down
9 changes: 9 additions & 0 deletions ci/licenses_golden/licenses_flutter
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,8 @@ FILE: ../../../flutter/lib/ui/dart_runtime_hooks.h
FILE: ../../../flutter/lib/ui/dart_ui.cc
FILE: ../../../flutter/lib/ui/dart_ui.h
FILE: ../../../flutter/lib/ui/dart_wrapper.h
FILE: ../../../flutter/lib/ui/fixtures/DashInNooglerHat.jpg
FILE: ../../../flutter/lib/ui/fixtures/ui_test.dart
FILE: ../../../flutter/lib/ui/geometry.dart
FILE: ../../../flutter/lib/ui/hash_codes.dart
FILE: ../../../flutter/lib/ui/hooks.dart
Expand All @@ -342,6 +344,9 @@ FILE: ../../../flutter/lib/ui/painting/gradient.cc
FILE: ../../../flutter/lib/ui/painting/gradient.h
FILE: ../../../flutter/lib/ui/painting/image.cc
FILE: ../../../flutter/lib/ui/painting/image.h
FILE: ../../../flutter/lib/ui/painting/image_decoder.cc
FILE: ../../../flutter/lib/ui/painting/image_decoder.h
FILE: ../../../flutter/lib/ui/painting/image_decoder_unittests.cc
FILE: ../../../flutter/lib/ui/painting/image_encoding.cc
FILE: ../../../flutter/lib/ui/painting/image_encoding.h
FILE: ../../../flutter/lib/ui/painting/image_filter.cc
Expand All @@ -350,6 +355,8 @@ FILE: ../../../flutter/lib/ui/painting/image_shader.cc
FILE: ../../../flutter/lib/ui/painting/image_shader.h
FILE: ../../../flutter/lib/ui/painting/matrix.cc
FILE: ../../../flutter/lib/ui/painting/matrix.h
FILE: ../../../flutter/lib/ui/painting/multi_frame_codec.cc
FILE: ../../../flutter/lib/ui/painting/multi_frame_codec.h
FILE: ../../../flutter/lib/ui/painting/paint.cc
FILE: ../../../flutter/lib/ui/painting/paint.h
FILE: ../../../flutter/lib/ui/painting/path.cc
Expand All @@ -364,6 +371,8 @@ FILE: ../../../flutter/lib/ui/painting/rrect.cc
FILE: ../../../flutter/lib/ui/painting/rrect.h
FILE: ../../../flutter/lib/ui/painting/shader.cc
FILE: ../../../flutter/lib/ui/painting/shader.h
FILE: ../../../flutter/lib/ui/painting/single_frame_codec.cc
FILE: ../../../flutter/lib/ui/painting/single_frame_codec.h
FILE: ../../../flutter/lib/ui/painting/vertices.cc
FILE: ../../../flutter/lib/ui/painting/vertices.h
FILE: ../../../flutter/lib/ui/plugins.dart
Expand Down
112 changes: 84 additions & 28 deletions fml/concurrent_message_loop.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,14 @@

namespace fml {

ConcurrentMessageLoop::ConcurrentMessageLoop()
: worker_count_(std::max(std::thread::hardware_concurrency(), 1u)),
shutdown_latch_(worker_count_),
shutdown_(false) {
std::shared_ptr<ConcurrentMessageLoop> ConcurrentMessageLoop::Create(
size_t worker_count) {
return std::shared_ptr<ConcurrentMessageLoop>{
new ConcurrentMessageLoop(worker_count)};
}

ConcurrentMessageLoop::ConcurrentMessageLoop(size_t worker_count)
: worker_count_(std::max<size_t>(worker_count, 1ul)) {
for (size_t i = 0; i < worker_count_; ++i) {
workers_.emplace_back([i, this]() {
fml::Thread::SetCurrentThreadName(
Expand All @@ -26,45 +30,97 @@ ConcurrentMessageLoop::ConcurrentMessageLoop()

ConcurrentMessageLoop::~ConcurrentMessageLoop() {
Terminate();
shutdown_latch_.Wait();
for (auto& worker : workers_) {
worker.join();
}
}

// |fml::MessageLoopImpl|
void ConcurrentMessageLoop::Run() {
FML_CHECK(false);
size_t ConcurrentMessageLoop::GetWorkerCount() const {
return worker_count_;
}

// |fml::MessageLoopImpl|
void ConcurrentMessageLoop::Terminate() {
std::scoped_lock lock(wait_condition_mutex_);
shutdown_ = true;
wait_condition_.notify_all();
std::shared_ptr<ConcurrentTaskRunner> ConcurrentMessageLoop::GetTaskRunner() {
return std::make_shared<ConcurrentTaskRunner>(weak_from_this());
}

// |fml::MessageLoopImpl|
void ConcurrentMessageLoop::WakeUp(fml::TimePoint time_point) {
// Assume that the clocks are not the same.
const auto duration = std::chrono::nanoseconds(
(time_point - fml::TimePoint::Now()).ToNanoseconds());
next_wake_ = std::chrono::high_resolution_clock::now() + duration;
wait_condition_.notify_all();
void ConcurrentMessageLoop::PostTask(fml::closure task) {
if (!task) {
return;
Copy link
Member

Choose a reason for hiding this comment

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

You should probably log this in debug cases since asking for a task to post and nothing happening is probably a logical error.

Copy link
Member Author

Choose a reason for hiding this comment

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

The task does not exist. So the call not doing anything makes sense I think.

}

std::unique_lock lock(tasks_mutex_);

// Don't just drop tasks on the floor in case of shutdown.
if (shutdown_) {
FML_DLOG(WARNING)
<< "Tried to post a task to shutdown concurrent message "
"loop. The task will be executed on the callers thread.";
lock.unlock();
task();
return;
}

tasks_.push(task);

// Unlock the mutex before notifying the condition variable because that mutex
// has to be acquired on the other thread anyway. Waiting in this scope till
// it is acquired there is a pessimization.
lock.unlock();

tasks_condition_.notify_one();
}

void ConcurrentMessageLoop::WorkerMain() {
while (!shutdown_) {
std::unique_lock<std::mutex> lock(wait_condition_mutex_);
if (!shutdown_) {
wait_condition_.wait(lock);
while (true) {
std::unique_lock lock(tasks_mutex_);
tasks_condition_.wait(lock,
[&]() { return tasks_.size() > 0 || shutdown_; });

if (tasks_.size() == 0) {
// This can only be caused by shutdown.
FML_DCHECK(shutdown_);
break;
}
TRACE_EVENT0("fml", "ConcurrentWorkerWake");
RunSingleExpiredTaskNow();

auto task = tasks_.front();
tasks_.pop();

// Don't hold onto the mutex while the task is being executed as it could
// itself try to post another tasks to this message loop.
lock.unlock();

TRACE_EVENT0("flutter", "ConcurrentWorkerWake");
// Execute the one tasks we woke up for.
task();
}
}

void ConcurrentMessageLoop::Terminate() {
std::scoped_lock lock(tasks_mutex_);
shutdown_ = true;
tasks_condition_.notify_all();
}

ConcurrentTaskRunner::ConcurrentTaskRunner(
std::weak_ptr<ConcurrentMessageLoop> weak_loop)
: weak_loop_(std::move(weak_loop)) {}

ConcurrentTaskRunner::~ConcurrentTaskRunner() = default;

void ConcurrentTaskRunner::PostTask(fml::closure task) {
if (!task) {
return;
Copy link
Member

Choose a reason for hiding this comment

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

Probably should log in debug build here too.

Copy link
Member Author

Choose a reason for hiding this comment

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

Same reasoning as above.

}

if (auto loop = weak_loop_.lock()) {
loop->PostTask(task);
return;
}

RunExpiredTasksNow();
shutdown_latch_.CountDown();
FML_DLOG(WARNING)
<< "Tried to post to a concurrent message loop that has already died. "
"Executing the task on the callers thread.";
task();
}

} // namespace fml
64 changes: 40 additions & 24 deletions fml/concurrent_message_loop.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,51 +5,67 @@
#ifndef FLUTTER_FML_CONCURRENT_MESSAGE_LOOP_H_
#define FLUTTER_FML_CONCURRENT_MESSAGE_LOOP_H_

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <queue>
#include <thread>
#include <vector>

#include "flutter/fml/closure.h"
#include "flutter/fml/macros.h"
#include "flutter/fml/message_loop_impl.h"
#include "flutter/fml/synchronization/count_down_latch.h"
#include "flutter/fml/synchronization/thread_annotations.h"

namespace fml {

class ConcurrentMessageLoop : public MessageLoopImpl {
private:
const size_t worker_count_;
std::mutex wait_condition_mutex_;
std::condition_variable wait_condition_;
std::vector<std::thread> workers_;
CountDownLatch shutdown_latch_;
std::chrono::high_resolution_clock::time_point next_wake_;
std::atomic_bool shutdown_;
class ConcurrentTaskRunner;

ConcurrentMessageLoop();
class ConcurrentMessageLoop
: public std::enable_shared_from_this<ConcurrentMessageLoop> {
public:
static std::shared_ptr<ConcurrentMessageLoop> Create(
size_t worker_count = std::thread::hardware_concurrency());

~ConcurrentMessageLoop();

// |fml::MessageLoopImpl|
void Run() override;
size_t GetWorkerCount() const;

std::shared_ptr<ConcurrentTaskRunner> GetTaskRunner();

void Terminate();

// |fml::MessageLoopImpl|
void Terminate() override;
private:
friend ConcurrentTaskRunner;

// |fml::MessageLoopImpl|
void WakeUp(fml::TimePoint time_point) override;
size_t worker_count_ = 0;
std::vector<std::thread> workers_;
std::mutex tasks_mutex_;
std::condition_variable tasks_condition_;
std::queue<fml::closure> tasks_;
Copy link
Member

Choose a reason for hiding this comment

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

There is no threadsafe queue class available to us? Maybe this is an opportunity to make one?

Copy link
Member Author

Choose a reason for hiding this comment

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

This class is thread safe. Is there a specific issue I missed? Unfortunately, I could not add thread safety annotation here because the the condition variable wait from libcxx does not have the appropriate annotations.

If you meant a lockless queue instead, I didn't want to write one because we needed the mutex for the condition variable anyway. Authoring, testing and then not needing the lockless nature of the queue seemed unnecessary in this instance.

bool shutdown_ = false;

static void WorkerMain(ConcurrentMessageLoop* loop);
ConcurrentMessageLoop(size_t worker_count);

void WorkerMain();

FML_FRIEND_MAKE_REF_COUNTED(ConcurrentMessageLoop);
FML_FRIEND_REF_COUNTED_THREAD_SAFE(ConcurrentMessageLoop);
void PostTask(fml::closure task);

FML_DISALLOW_COPY_AND_ASSIGN(ConcurrentMessageLoop);
};

class ConcurrentTaskRunner {
public:
ConcurrentTaskRunner(std::weak_ptr<ConcurrentMessageLoop> weak_loop);

~ConcurrentTaskRunner();

void PostTask(fml::closure task);

private:
friend ConcurrentMessageLoop;

std::weak_ptr<ConcurrentMessageLoop> weak_loop_;

FML_DISALLOW_COPY_AND_ASSIGN(ConcurrentTaskRunner);
};

} // namespace fml

#endif // FLUTTER_FML_CONCURRENT_MESSAGE_LOOP_H_
8 changes: 0 additions & 8 deletions fml/message_loop.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@

#include <utility>

#include "flutter/fml/concurrent_message_loop.h"
#include "flutter/fml/memory/ref_counted.h"
#include "flutter/fml/memory/ref_ptr.h"
#include "flutter/fml/message_loop_impl.h"
Expand Down Expand Up @@ -44,13 +43,6 @@ MessageLoop::MessageLoop()
FML_CHECK(task_runner_);
}

MessageLoop::MessageLoop(Type)
: loop_(fml::MakeRefCounted<ConcurrentMessageLoop>()),
task_runner_(fml::MakeRefCounted<fml::TaskRunner>(loop_)) {
FML_CHECK(loop_);
FML_CHECK(task_runner_);
}

MessageLoop::~MessageLoop() = default;

void MessageLoop::Run() {
Expand Down
4 changes: 0 additions & 4 deletions fml/message_loop.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,6 @@ class MessageLoop {
FML_EMBEDDER_ONLY
static MessageLoop& GetCurrent();

enum class Type { kConcurrent };

MessageLoop(Type type);

bool IsValid() const;

void Run();
Expand Down
21 changes: 17 additions & 4 deletions fml/message_loop_unittests.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
#include <iostream>
#include <thread>

#include "flutter/fml/concurrent_message_loop.h"
#include "flutter/fml/message_loop.h"
#include "flutter/fml/synchronization/count_down_latch.h"
#include "flutter/fml/synchronization/waitable_event.h"
Expand Down Expand Up @@ -281,19 +282,31 @@ TEST(MessageLoop, TaskObserverFire) {
ASSERT_TRUE(terminated);
}

TEST(MessageLoop, CanCreateAndShutdownConcurrentMessageLoopsOverAndOver) {
for (size_t i = 0; i < 10; ++i) {
auto loop = fml::ConcurrentMessageLoop::Create(i + 1);
ASSERT_EQ(loop->GetWorkerCount(), i + 1);
}
}

TEST(MessageLoop, CanCreateConcurrentMessageLoop) {
fml::MessageLoop loop(fml::MessageLoop::Type::kConcurrent);
auto task_runner = loop.GetTaskRunner();
auto loop = fml::ConcurrentMessageLoop::Create();
auto task_runner = loop->GetTaskRunner();
const size_t kCount = 10;
fml::CountDownLatch latch(kCount);
std::mutex thread_ids_mutex;
std::set<std::thread::id> thread_ids;
for (size_t i = 0; i < kCount; ++i) {
task_runner->PostTask([&latch]() {
std::this_thread::sleep_for(std::chrono::milliseconds(5));
task_runner->PostTask([&]() {
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Ran on thread: " << std::this_thread::get_id() << std::endl;
std::scoped_lock lock(thread_ids_mutex);
thread_ids.insert(std::this_thread::get_id());
latch.CountDown();
});
}
latch.Wait();
ASSERT_GE(thread_ids.size(), 1u);
}

TEST(MessageLoop, CanSwapMessageLoopsAndPreserveThreadConfiguration) {
Expand Down
35 changes: 35 additions & 0 deletions fml/trace_event.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,41 @@ class ScopedInstantEnd {
FML_DISALLOW_COPY_AND_ASSIGN(ScopedInstantEnd);
};

// A move-only utility object that creates a new flow with a unique ID and
// automatically ends it when it goes out of scope. When tracing using multiple
// overlapping flows, it often gets hard to make sure to end the flow
// (especially with early returns), or, end/step on the wrong flow. This
// leads to corrupted or missing traces in the UI.
class TraceFlow {
Copy link
Member

Choose a reason for hiding this comment

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

Please add docstring for class.

Why is this inlined?

Copy link
Member Author

Choose a reason for hiding this comment

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

Done.

Copy link
Member

Choose a reason for hiding this comment

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

lgtm, just make sure it's it doxygen format ( http://www.doxygen.nl/manual/docblocks.html )

Copy link
Member

Choose a reason for hiding this comment

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

We're using triple slashes in other places of the code methinks.

public:
TraceFlow(const char* label) : label_(label), nonce_(TraceNonce()) {
Copy link
Member

Choose a reason for hiding this comment

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

Shouldn't you memcpy the label?

Copy link
Member Author

Choose a reason for hiding this comment

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

No, all strings using in tracing need to be const. Or, more precisely, the lifecycle of the labels must extend past the shutdown on the Dart VM which is used as the trace processor.

Using a memcpy here would make the VM read from garbage memory as it is constructing the buffer used to collect trace information.

Copy link
Member

Choose a reason for hiding this comment

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

Trace event labels are passed to the Dart timeline API, which requires that the label buffer remain valid until the Dart VM is shut down. Effectively this means that the labels need to be string literals that are not allocated dynamically.

Copy link
Member

Choose a reason for hiding this comment

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

Ok, I think what you want to use is constexpr then to assert statically that you are using a string literal: https://en.cppreference.com/w/cpp/language/constexpr

Copy link
Member

Choose a reason for hiding this comment

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

I played around with this and I'm mistaken, it isn't straightforward to do: https://stackoverflow.com/questions/7613528/restrict-passed-parameter-to-a-string-literal, feel free to ignore my suggestion =).

TraceEventFlowBegin0("flutter", label_, nonce_);
}

~TraceFlow() { End(label_); }

TraceFlow(TraceFlow&& other) : label_(other.label_), nonce_(other.nonce_) {
other.nonce_ = 0;
}

void Step(const char* label) const {
TraceEventFlowStep0("flutter", label, nonce_);
}

void End(const char* label = nullptr) {
if (nonce_ != 0) {
TraceEventFlowEnd0("flutter", label == nullptr ? label_ : label, nonce_);
nonce_ = 0;
}
}

private:
const char* label_;
size_t nonce_;

FML_DISALLOW_COPY_AND_ASSIGN(TraceFlow);
};

} // namespace tracing
} // namespace fml

Expand Down
Loading