Pipeline allows continuations that can produce to front (flutter/engine#9652)

* Make pipeline hold a deque so we can push_front

- Also assert that all the DoDraw tasks are executed
  on the GPU thread.

* Add tests for pipeline

* Add support for pipeline to push resources to the front

engine/src/flutter/ci/licenses_golden/licenses_flutter

@@ -454,6 +454,7 @@ FILE: ../../../flutter/shell/common/persistent_cache.cc
 FILE: ../../../flutter/shell/common/persistent_cache.h
 FILE: ../../../flutter/shell/common/pipeline.cc
 FILE: ../../../flutter/shell/common/pipeline.h
+FILE: ../../../flutter/shell/common/pipeline_unittests.cc
 FILE: ../../../flutter/shell/common/platform_view.cc
 FILE: ../../../flutter/shell/common/platform_view.h
 FILE: ../../../flutter/shell/common/rasterizer.cc

engine/src/flutter/shell/common/BUILD.gn

@@ -152,6 +152,7 @@ test_fixtures("shell_unittests_fixtures") {
 
 shell_host_executable("shell_unittests") {
   sources = [
+    "pipeline_unittests.cc",
     "shell_test.cc",
     "shell_test.h",
     "shell_unittests.cc",

engine/src/flutter/shell/common/pipeline.h

@@ -10,9 +10,9 @@
 #include "flutter/fml/synchronization/semaphore.h"
 #include "flutter/fml/trace_event.h"
 
+#include <deque>
 #include <memory>
 #include <mutex>
-#include <queue>
 
 namespace flutter {
 
@@ -88,7 +88,8 @@ class Pipeline : public fml::RefCountedThreadSafe<Pipeline<R>> {
     FML_DISALLOW_COPY_AND_ASSIGN(ProducerContinuation);
   };
 
-  explicit Pipeline(uint32_t depth) : empty_(depth), available_(0) {}
+  explicit Pipeline(uint32_t depth)
+      : depth_(depth), empty_(depth), available_(0) {}
 
   ~Pipeline() = default;
 
@@ -105,6 +106,20 @@ class Pipeline : public fml::RefCountedThreadSafe<Pipeline<R>> {
         GetNextPipelineTraceID()};         // trace id
   }
 
+  // Pushes task to the front of the pipeline.
+  //
+  // If we exceed the depth completing this continuation, we drop the
+  // last frame to preserve the depth of the pipeline.
+  //
+  // Note: Use |Pipeline::Produce| where possible. This should only be
+  // used to en-queue high-priority resources.
+  ProducerContinuation ProduceToFront() {
+    return ProducerContinuation{
+        std::bind(&Pipeline::ProducerCommitFront, this, std::placeholders::_1,
+                  std::placeholders::_2),  // continuation
+        GetNextPipelineTraceID()};         // trace id
+  }
+
   using Consumer = std::function<void(ResourcePtr)>;
 
   FML_WARN_UNUSED_RESULT
@@ -124,7 +139,7 @@ class Pipeline : public fml::RefCountedThreadSafe<Pipeline<R>> {
     {
       std::scoped_lock lock(queue_mutex_);
       std::tie(resource, trace_id) = std::move(queue_.front());
-      queue_.pop();
+      queue_.pop_front();
       items_count = queue_.size();
     }
 
@@ -143,15 +158,29 @@ class Pipeline : public fml::RefCountedThreadSafe<Pipeline<R>> {
   }
 
  private:
+  uint32_t depth_;
   fml::Semaphore empty_;
   fml::Semaphore available_;
   std::mutex queue_mutex_;
-  std::queue<std::pair<ResourcePtr, size_t>> queue_;
+  std::deque<std::pair<ResourcePtr, size_t>> queue_;
 
   void ProducerCommit(ResourcePtr resource, size_t trace_id) {
     {
       std::scoped_lock lock(queue_mutex_);
-      queue_.emplace(std::move(resource), trace_id);
+      queue_.emplace_back(std::move(resource), trace_id);
+    }
+
+    // Ensure the queue mutex is not held as that would be a pessimization.
+    available_.Signal();
+  }
+
+  void ProducerCommitFront(ResourcePtr resource, size_t trace_id) {
+    {
+      std::scoped_lock lock(queue_mutex_);
+      queue_.emplace_front(std::move(resource), trace_id);
+      while (queue_.size() > depth_) {
+        queue_.pop_back();
+      }
     }
 
     // Ensure the queue mutex is not held as that would be a pessimization.

engine/src/flutter/shell/common/pipeline_unittests.cc

@@ -0,0 +1,135 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#define FML_USED_ON_EMBEDDER
+
+#include <functional>
+#include <future>
+#include <memory>
+
+#include "flutter/shell/common/pipeline.h"
+#include "gtest/gtest.h"
+
+namespace flutter {
+namespace testing {
+
+using IntPipeline = Pipeline<int>;
+using Continuation = IntPipeline::ProducerContinuation;
+
+TEST(PipelineTest, ConsumeOneVal) {
+  fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(2);
+
+  Continuation continuation = pipeline->Produce();
+
+  const int test_val = 1;
+  continuation.Complete(std::make_unique<int>(test_val));
+
+  PipelineConsumeResult consume_result = pipeline->Consume(
+      [&test_val](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val); });
+
+  ASSERT_EQ(consume_result, PipelineConsumeResult::Done);
+}
+
+TEST(PipelineTest, ContinuationCanOnlyBeUsedOnce) {
+  fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(2);
+
+  Continuation continuation = pipeline->Produce();
+
+  const int test_val = 1;
+  continuation.Complete(std::make_unique<int>(test_val));
+
+  PipelineConsumeResult consume_result_1 = pipeline->Consume(
+      [&test_val](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val); });
+
+  continuation.Complete(std::make_unique<int>(test_val));
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::Done);
+
+  PipelineConsumeResult consume_result_2 =
+      pipeline->Consume([](std::unique_ptr<int> v) { FAIL(); });
+
+  continuation.Complete(std::make_unique<int>(test_val));
+  ASSERT_EQ(consume_result_2, PipelineConsumeResult::NoneAvailable);
+}
+
+TEST(PipelineTest, PushingMoreThanDepthCompletesFirstSubmission) {
+  const int depth = 1;
+  fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(depth);
+
+  Continuation continuation_1 = pipeline->Produce();
+  Continuation continuation_2 = pipeline->Produce();
+
+  const int test_val_1 = 1, test_val_2 = 2;
+  continuation_1.Complete(std::make_unique<int>(test_val_1));
+  continuation_2.Complete(std::make_unique<int>(test_val_2));
+
+  PipelineConsumeResult consume_result_1 = pipeline->Consume(
+      [&test_val_1](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_1); });
+
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::Done);
+}
+
+TEST(PipelineTest, PushingMultiProcessesInOrder) {
+  const int depth = 2;
+  fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(depth);
+
+  Continuation continuation_1 = pipeline->Produce();
+  Continuation continuation_2 = pipeline->Produce();
+
+  const int test_val_1 = 1, test_val_2 = 2;
+  continuation_1.Complete(std::make_unique<int>(test_val_1));
+  continuation_2.Complete(std::make_unique<int>(test_val_2));
+
+  PipelineConsumeResult consume_result_1 = pipeline->Consume(
+      [&test_val_1](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_1); });
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::MoreAvailable);
+
+  PipelineConsumeResult consume_result_2 = pipeline->Consume(
+      [&test_val_2](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_2); });
+  ASSERT_EQ(consume_result_2, PipelineConsumeResult::Done);
+}
+
+TEST(PipelineTest, PushingToFrontOverridesOrder) {
+  const int depth = 2;
+  fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(depth);
+
+  Continuation continuation_1 = pipeline->Produce();
+  Continuation continuation_2 = pipeline->ProduceToFront();
+
+  const int test_val_1 = 1, test_val_2 = 2;
+  continuation_1.Complete(std::make_unique<int>(test_val_1));
+  continuation_2.Complete(std::make_unique<int>(test_val_2));
+
+  PipelineConsumeResult consume_result_1 = pipeline->Consume(
+      [&test_val_2](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_2); });
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::MoreAvailable);
+
+  PipelineConsumeResult consume_result_2 = pipeline->Consume(
+      [&test_val_1](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_1); });
+  ASSERT_EQ(consume_result_2, PipelineConsumeResult::Done);
+}
+
+TEST(PipelineTest, PushingToFrontDropsLastResource) {
+  const int depth = 2;
+  fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(depth);
+
+  Continuation continuation_1 = pipeline->Produce();
+  Continuation continuation_2 = pipeline->Produce();
+  Continuation continuation_3 = pipeline->ProduceToFront();
+
+  const int test_val_1 = 1, test_val_2 = 2, test_val_3 = 3;
+  continuation_1.Complete(std::make_unique<int>(test_val_1));
+  continuation_2.Complete(std::make_unique<int>(test_val_2));
+  continuation_3.Complete(std::make_unique<int>(test_val_3));
+
+  PipelineConsumeResult consume_result_1 = pipeline->Consume(
+      [&test_val_3](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_3); });
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::MoreAvailable);
+
+  PipelineConsumeResult consume_result_2 = pipeline->Consume(
+      [&test_val_1](std::unique_ptr<int> v) { ASSERT_EQ(*v, test_val_1); });
+  ASSERT_EQ(consume_result_2, PipelineConsumeResult::Done);
+}
+
+}  // namespace testing
+}  // namespace flutter

engine/src/flutter/shell/common/rasterizer.cc

@@ -173,6 +173,8 @@ sk_sp<SkImage> Rasterizer::MakeRasterSnapshot(sk_sp<SkPicture> picture,
 }
 
 void Rasterizer::DoDraw(std::unique_ptr<flutter::LayerTree> layer_tree) {
+  FML_DCHECK(task_runners_.GetGPUTaskRunner()->RunsTasksOnCurrentThread());
+
   if (!layer_tree || !surface_) {
     return;
   }