Pipelines vend single use continuations that callers can complete at any time...

Pipelines vend single use continuations that callers can complete at any time to signal readiness of a pipeline resource. (#2957)

sky/shell/ui/animator.cc

@@ -43,24 +43,35 @@ void Animator::Start() {
 void Animator::BeginFrame(int64_t time_stamp) {
   pending_frame_semaphore_.Signal();
 
-  LayerTreePipeline::Producer producer = [this]() {
-    renderable_tree_.reset();
-    ftl::Stopwatch stopwatch;
-    stopwatch.Start();
-    engine_->BeginFrame(ftl::TimePoint::Now());
-    if (renderable_tree_) {
-      renderable_tree_->set_construction_time(stopwatch.Elapsed());
+  if (!producer_continuation_) {
+    // We may already have a valid pipeline continuation in case a previous
+    // begin frame did not result in an Animation::Render. Simply reuse that
+    // instead of asking the pipeline for a fresh continuation.
+    producer_continuation_ = layer_tree_pipeline_->Produce();
+
+    if (!producer_continuation_) {
+      // If we still don't have valid continuation, the pipeline is currently
+      // full because the consumer is being too slow. Try again at the next
+      // frame interval.
+      TRACE_EVENT_INSTANT0("flutter", "ConsumerSlowDefer",
+                           TRACE_EVENT_SCOPE_PROCESS);
+      RequestFrame();
+      return;
     }
-    return std::move(renderable_tree_);
-  };
-
-  if (!layer_tree_pipeline_->Produce(producer)) {
-    TRACE_EVENT_INSTANT0("flutter", "ConsumerSlowDefer",
-                         TRACE_EVENT_SCOPE_PROCESS);
-    RequestFrame();
-    return;
   }
 
+  // We have acquired a valid continuation from the pipeline and are ready
+  // to service potential frame.
+  DCHECK(producer_continuation_);
+
+  engine_->BeginFrame(ftl::TimePoint::Now());
+}
+
+void Animator::Render(std::unique_ptr<flow::LayerTree> layer_tree) {
+  // Commit the pending continuation.
+  producer_continuation_.Complete(std::move(layer_tree));
+
+  // Notify the rasterizer that the pipeline has items it may consume.
   auto weak_rasterizer = rasterizer_->GetWeakRasterizerPtr();
   auto pipeline = layer_tree_pipeline_;
 
@@ -72,10 +83,6 @@ void Animator::BeginFrame(int64_t time_stamp) {
   });
 }
 
-void Animator::Render(std::unique_ptr<flow::LayerTree> layer_tree) {
-  renderable_tree_ = std::move(layer_tree);
-}
-
 void Animator::RequestFrame() {
   if (paused_) {
     return;

sky/shell/ui/animator.h

@@ -47,7 +47,7 @@ class Animator {
   vsync::VSyncProviderPtr fallback_vsync_provider_;
   ftl::RefPtr<LayerTreePipeline> layer_tree_pipeline_;
   flutter::Semaphore pending_frame_semaphore_;
-  std::unique_ptr<flow::LayerTree> renderable_tree_;
+  LayerTreePipeline::ProducerContinuation producer_continuation_;
   bool paused_;
 
   base::WeakPtrFactory<Animator> weak_factory_;

synchronization/pipeline.cc

@@ -3,9 +3,3 @@
 // found in the LICENSE file.
 
 #include "flutter/synchronization/pipeline.h"
-
-namespace flutter {
-
-//
-
-}  // namespace flutter

synchronization/pipeline.h

@@ -30,39 +30,60 @@ class Pipeline : public ftl::RefCountedThreadSafe<Pipeline<R>> {
   using Resource = R;
   using ResourcePtr = std::unique_ptr<Resource>;
 
-  explicit Pipeline(uint32_t depth) : empty_(depth), available_(0) {}
-
-  ~Pipeline() {}
-
-  bool IsValid() const { return empty_.IsValid() && available_.IsValid(); }
+  /// Denotes a spot in the pipeline reserved for the producer to finish
+  /// preparing a completed pipeline resource.
+  class ProducerContinuation {
+   public:
+    ProducerContinuation() = default;
+
+    ProducerContinuation(ProducerContinuation&& other)
+        : continuation_(other.continuation_) {
+      other.continuation_ = nullptr;
+    }
 
-  using Producer = std::function<ResourcePtr(void)>;
+    ProducerContinuation& operator=(ProducerContinuation&& other) {
+      std::swap(continuation_, other.continuation_);
+      return *this;
+    }
 
-  FTL_WARN_UNUSED_RESULT
-  bool Produce(Producer producer) {
-    if (producer == nullptr) {
-      return false;
+    ~ProducerContinuation() {
+      if (continuation_) {
+        continuation_(nullptr);
+      }
     }
 
-    if (!empty_.TryWait()) {
-      return false;
+    void Complete(ResourcePtr resource) {
+      if (continuation_) {
+        continuation_(std::move(resource));
+        continuation_ = nullptr;
+      }
     }
 
-    ResourcePtr resource;
+    operator bool() const { return continuation_ != nullptr; }
 
-    {
-      TRACE_EVENT0("flutter", "PipelineProduce");
-      resource = producer();
-    }
+   private:
+    friend class Pipeline;
 
-    {
-      ftl::MutexLocker lock(&queue_mutex_);
-      queue_.emplace(std::move(resource));
-    }
+    std::function<void(ResourcePtr)> continuation_;
 
-    available_.Signal();
+    ProducerContinuation(std::function<void(ResourcePtr)> continuation)
+        : continuation_(continuation) {}
+
+    FTL_DISALLOW_COPY_AND_ASSIGN(ProducerContinuation);
+  };
 
-    return true;
+  explicit Pipeline(uint32_t depth) : empty_(depth), available_(0) {}
+
+  ~Pipeline() = default;
+
+  bool IsValid() const { return empty_.IsValid() && available_.IsValid(); }
+
+  ProducerContinuation Produce() {
+    if (!empty_.TryWait()) {
+      return {};
+    }
+
+    return {std::bind(&Pipeline::ProducerCommit, this, std::placeholders::_1)};
   }
 
   using Consumer = std::function<void(ResourcePtr)>;
@@ -104,6 +125,16 @@ class Pipeline : public ftl::RefCountedThreadSafe<Pipeline<R>> {
   ftl::Mutex queue_mutex_;
   std::queue<ResourcePtr> queue_;
 
+  void ProducerCommit(ResourcePtr resource) {
+    {
+      ftl::MutexLocker lock(&queue_mutex_);
+      queue_.emplace(std::move(resource));
+    }
+
+    // Ensure the queue mutex is not held as that would be a pessimization.
+    available_.Signal();
+  }
+
   FTL_DISALLOW_COPY_AND_ASSIGN(Pipeline);
 };