[Impeller] Maintain separate queues of GLES operations for each thread in the reactor

impeller/renderer/backend/gles/command_buffer_gles.cc

@@ -39,8 +39,8 @@ bool CommandBufferGLES::OnSubmitCommands(CompletionCallback callback) {
 }
 
 // |CommandBuffer|
-void CommandBufferGLES::OnWaitUntilScheduled() {
-  reactor_->GetProcTable().Flush();
+void CommandBufferGLES::OnWaitUntilCompleted() {
+  reactor_->GetProcTable().Finish();
 }
 
 // |CommandBuffer|

impeller/renderer/backend/gles/command_buffer_gles.h

@@ -35,7 +35,7 @@ class CommandBufferGLES final : public CommandBuffer {
   bool OnSubmitCommands(CompletionCallback callback) override;
 
   // |CommandBuffer|
-  void OnWaitUntilScheduled() override;
+  void OnWaitUntilCompleted() override;
 
   // |CommandBuffer|
   std::shared_ptr<RenderPass> OnCreateRenderPass(RenderTarget target) override;

impeller/renderer/backend/gles/proc_table_gles.h

@@ -169,6 +169,7 @@ struct GLProc {
   PROC(Enable);                              \
   PROC(EnableVertexAttribArray);             \
   PROC(Flush);                               \
+  PROC(Finish);                              \
   PROC(FramebufferRenderbuffer);             \
   PROC(FramebufferTexture2D);                \
   PROC(FrontFace);                           \

impeller/renderer/backend/gles/reactor_gles.cc

@@ -105,8 +105,10 @@ bool ReactorGLES::RemoveWorker(WorkerID worker) {
 }
 
 bool ReactorGLES::HasPendingOperations() const {
+  auto thread_id = std::this_thread::get_id();
   Lock ops_lock(ops_mutex_);
-  return !ops_.empty();
+  auto it = ops_.find(thread_id);
+  return it != ops_.end() ? !it->second.empty() : false;
 }
 
 const ProcTableGLES& ReactorGLES::GetProcTable() const {
@@ -136,9 +138,10 @@ bool ReactorGLES::AddOperation(Operation operation) {
   if (!operation) {
     return false;
   }
+  auto thread_id = std::this_thread::get_id();
   {
     Lock ops_lock(ops_mutex_);
-    ops_.emplace_back(std::move(operation));
+    ops_[thread_id].emplace_back(std::move(operation));
   }
   // Attempt a reaction if able but it is not an error if this isn't possible.
   [[maybe_unused]] auto result = React();
@@ -191,10 +194,6 @@ bool ReactorGLES::React() {
   }
   TRACE_EVENT0("impeller", "ReactorGLES::React");
   while (HasPendingOperations()) {
-    // Both the raster thread and the IO thread can flush queued operations.
-    // Ensure that execution of the ops is serialized.
-    Lock execution_lock(ops_execution_mutex_);
-
     if (!ReactOnce()) {
       return false;
     }
@@ -279,10 +278,11 @@ bool ReactorGLES::FlushOps() {
 
   // Do NOT hold the ops or handles locks while performing operations in case
   // the ops enqueue more ops.
-  decltype(ops_) ops;
+  decltype(ops_)::mapped_type ops;
+  auto thread_id = std::this_thread::get_id();
   {
     Lock ops_lock(ops_mutex_);
-    std::swap(ops_, ops);
+    std::swap(ops_[thread_id], ops);
   }
   for (const auto& op : ops) {
     TRACE_EVENT0("impeller", "ReactorGLES::Operation");

impeller/renderer/backend/gles/reactor_gles.h

@@ -260,9 +260,9 @@ class ReactorGLES {
 
   std::unique_ptr<ProcTableGLES> proc_table_;
 
-  Mutex ops_execution_mutex_;
   mutable Mutex ops_mutex_;
-  std::vector<Operation> ops_ IPLR_GUARDED_BY(ops_mutex_);
+  std::map<std::thread::id, std::vector<Operation>> ops_ IPLR_GUARDED_BY(
+      ops_mutex_);
 
   // Make sure the container is one where erasing items during iteration doesn't
   // invalidate other iterators.
@@ -280,7 +280,7 @@ class ReactorGLES {
   bool can_set_debug_labels_ = false;
   bool is_valid_ = false;
 
-  bool ReactOnce() IPLR_REQUIRES(ops_execution_mutex_);
+  bool ReactOnce();
 
   bool HasPendingOperations() const;
 

impeller/renderer/backend/gles/test/reactor_unittests.cc

@@ -3,6 +3,7 @@
 // found in the LICENSE file.
 
 #include <memory>
+#include "flutter/fml/synchronization/waitable_event.h"
 #include "flutter/testing/testing.h"  // IWYU pragma: keep
 #include "gtest/gtest.h"
 #include "impeller/renderer/backend/gles/handle_gles.h"
@@ -60,5 +61,37 @@ TEST(ReactorGLES, DeletesHandlesDuringShutdown) {
               calls.end());
 }
 
+TEST(ReactorGLES, PerThreadOperationQueues) {
+  auto mock_gles = MockGLES::Init();
+  ProcTableGLES::Resolver resolver = kMockResolverGLES;
+  auto proc_table = std::make_unique<ProcTableGLES>(resolver);
+  auto worker = std::make_shared<TestWorker>();
+  auto reactor = std::make_shared<ReactorGLES>(std::move(proc_table));
+  reactor->AddWorker(worker);
+
+  bool op1_called = false;
+  EXPECT_TRUE(
+      reactor->AddOperation([&](const ReactorGLES&) { op1_called = true; }));
+
+  fml::AutoResetWaitableEvent event;
+  bool op2_called = false;
+  std::thread thread([&] {
+    EXPECT_TRUE(
+        reactor->AddOperation([&](const ReactorGLES&) { op2_called = true; }));
+    event.Wait();
+    EXPECT_TRUE(reactor->React());
+  });
+
+  // Reacting on the main thread should only run the main thread's operation.
+  EXPECT_TRUE(reactor->React());
+  EXPECT_TRUE(op1_called);
+  EXPECT_FALSE(op2_called);
+
+  // Reacting on the second thread will run the second thread's operation.
+  event.Signal();
+  thread.join();
+  EXPECT_TRUE(op2_called);
+}
+
 }  // namespace testing
 }  // namespace impeller

impeller/renderer/backend/metal/command_buffer_mtl.h

@@ -37,7 +37,7 @@ class CommandBufferMTL final : public CommandBuffer {
   bool OnSubmitCommands(CompletionCallback callback) override;
 
   // |CommandBuffer|
-  void OnWaitUntilScheduled() override;
+  void OnWaitUntilCompleted() override;
 
   // |CommandBuffer|
   std::shared_ptr<RenderPass> OnCreateRenderPass(RenderTarget target) override;

impeller/renderer/backend/metal/command_buffer_mtl.mm

@@ -187,7 +187,7 @@ static bool LogMTLCommandBufferErrorIfPresent(id<MTLCommandBuffer> buffer) {
   return true;
 }
 
-void CommandBufferMTL::OnWaitUntilScheduled() {}
+void CommandBufferMTL::OnWaitUntilCompleted() {}
 
 std::shared_ptr<RenderPass> CommandBufferMTL::OnCreateRenderPass(
     RenderTarget target) {

impeller/renderer/backend/vulkan/command_buffer_vk.cc

@@ -47,7 +47,7 @@ bool CommandBufferVK::OnSubmitCommands(CompletionCallback callback) {
   FML_UNREACHABLE()
 }
 
-void CommandBufferVK::OnWaitUntilScheduled() {}
+void CommandBufferVK::OnWaitUntilCompleted() {}
 
 std::shared_ptr<RenderPass> CommandBufferVK::OnCreateRenderPass(
     RenderTarget target) {

impeller/renderer/backend/vulkan/command_buffer_vk.h

@@ -100,7 +100,7 @@ class CommandBufferVK final
   bool OnSubmitCommands(CompletionCallback callback) override;
 
   // |CommandBuffer|
-  void OnWaitUntilScheduled() override;
+  void OnWaitUntilCompleted() override;
 
   // |CommandBuffer|
   std::shared_ptr<RenderPass> OnCreateRenderPass(RenderTarget target) override;

impeller/renderer/command_buffer.cc

@@ -30,8 +30,8 @@ bool CommandBuffer::SubmitCommands() {
   return SubmitCommands(nullptr);
 }
 
-void CommandBuffer::WaitUntilScheduled() {
-  return OnWaitUntilScheduled();
+void CommandBuffer::WaitUntilCompleted() {
+  return OnWaitUntilCompleted();
 }
 
 std::shared_ptr<RenderPass> CommandBuffer::CreateRenderPass(

impeller/renderer/command_buffer.h

@@ -61,9 +61,10 @@ class CommandBuffer {
   virtual void SetLabel(std::string_view label) const = 0;
 
   //----------------------------------------------------------------------------
-  /// @brief      Force execution of pending GPU commands.
+  /// @brief      Block the current thread until the GPU has completed execution
+  ///             of the commands.
   ///
-  void WaitUntilScheduled();
+  void WaitUntilCompleted();
 
   //----------------------------------------------------------------------------
   /// @brief      Create a render pass to record render commands into.
@@ -102,7 +103,7 @@ class CommandBuffer {
 
   [[nodiscard]] virtual bool OnSubmitCommands(CompletionCallback callback) = 0;
 
-  virtual void OnWaitUntilScheduled() = 0;
+  virtual void OnWaitUntilCompleted() = 0;
 
   virtual std::shared_ptr<ComputePass> OnCreateComputePass() = 0;
 

impeller/renderer/testing/mocks.h

@@ -128,7 +128,7 @@ class MockCommandBuffer : public CommandBuffer {
               OnSubmitCommands,
               (CompletionCallback callback),
               (override));
-  MOCK_METHOD(void, OnWaitUntilScheduled, (), (override));
+  MOCK_METHOD(void, OnWaitUntilCompleted, (), (override));
   MOCK_METHOD(std::shared_ptr<ComputePass>,
               OnCreateComputePass,
               (),

lib/ui/painting/image_decoder_impeller.cc

@@ -382,6 +382,10 @@ ImageDecoderImpeller::UnsafeUploadTextureToPrivate(
     return std::make_pair(nullptr, decode_error);
   }
 
+  // Flush the pending command buffer to ensure that its output becomes visible
+  // to the raster thread.
+  command_buffer->WaitUntilCompleted();
+
   context->DisposeThreadLocalCachedResources();
 
   return std::make_pair(