Replace Pipeline's ProduceToFront with ProduceIfEmpty to handle thread merging. (#17122)

Author: Chris Yang

shell/common/pipeline.h

@@ -111,16 +111,22 @@ 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() {
+  // Create a `ProducerContinuation` that will only push the task if the queue
+  // is empty.
+  // Prefer using |Produce|. ProducerContinuation returned by this method
+  // doesn't guarantee that the frame will be rendered.
+  ProducerContinuation ProduceIfEmpty() {
+    if (!empty_.TryWait()) {
+      return {};
+    }
+    ++inflight_;
+    FML_TRACE_COUNTER("flutter", "Pipeline Depth",
+                      reinterpret_cast<int64_t>(this),      //
+                      "frames in flight", inflight_.load()  //
+    );
+
     return ProducerContinuation{
-        std::bind(&Pipeline::ProducerCommitFront, this, std::placeholders::_1,
+        std::bind(&Pipeline::ProducerCommitIfEmpty, this, std::placeholders::_1,
                   std::placeholders::_2),  // continuation
         GetNextPipelineTraceID()};         // trace id
   }
@@ -181,13 +187,16 @@ class Pipeline : public fml::RefCountedThreadSafe<Pipeline<R>> {
     available_.Signal();
   }
 
-  void ProducerCommitFront(ResourcePtr resource, size_t trace_id) {
+  void ProducerCommitIfEmpty(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();
+      if (!queue_.empty()) {
+        // Bail if the queue is not empty, opens up spaces to produce other
+        // frames.
+        empty_.Signal();
+        return;
       }
+      queue_.emplace_back(std::move(resource), trace_id);
     }
 
     // Ensure the queue mutex is not held as that would be a pessimization.

shell/common/pipeline_unittests.cc

@@ -89,46 +89,34 @@ TEST(PipelineTest, PushingMultiProcessesInOrder) {
   ASSERT_EQ(consume_result_2, PipelineConsumeResult::Done);
 }
 
-TEST(PipelineTest, PushingToFrontOverridesOrder) {
+TEST(PipelineTest, ProduceIfEmptyDoesNotConsumeWhenQueueIsNotEmpty) {
   const int depth = 2;
   fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(depth);
 
   Continuation continuation_1 = pipeline->Produce();
-  Continuation continuation_2 = pipeline->ProduceToFront();
+  Continuation continuation_2 = pipeline->ProduceIfEmpty();
 
   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);
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::Done);
 }
 
-TEST(PipelineTest, PushingToFrontDropsLastResource) {
-  const int depth = 2;
+TEST(PipelineTest, ProduceIfEmptySuccessfulIfQueueIsEmpty) {
+  const int depth = 1;
   fml::RefPtr<IntPipeline> pipeline = fml::MakeRefCounted<IntPipeline>(depth);
 
-  Continuation continuation_1 = pipeline->Produce();
-  Continuation continuation_2 = pipeline->Produce();
-  Continuation continuation_3 = pipeline->ProduceToFront();
+  Continuation continuation_1 = pipeline->ProduceIfEmpty();
 
-  const int test_val_1 = 1, test_val_2 = 2, test_val_3 = 3;
+  const int test_val_1 = 1;
   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);
+  ASSERT_EQ(consume_result_1, PipelineConsumeResult::Done);
 }
 
 }  // namespace testing

shell/common/rasterizer.cc

@@ -127,9 +127,8 @@ void Rasterizer::Draw(fml::RefPtr<Pipeline<flutter::LayerTree>> pipeline) {
   // if the raster status is to resubmit the frame, we push the frame to the
   // front of the queue and also change the consume status to more available.
   if (raster_status == RasterStatus::kResubmit) {
-    auto front_continuation = pipeline->ProduceToFront();
+    auto front_continuation = pipeline->ProduceIfEmpty();
     front_continuation.Complete(std::move(resubmitted_layer_tree_));
-    consume_result = PipelineConsumeResult::MoreAvailable;
   } else if (raster_status == RasterStatus::kEnqueuePipeline) {
     consume_result = PipelineConsumeResult::MoreAvailable;
   }