Change SerialRunner to use OnceCallback.

Bug: 1007769
Change-Id: I3fcf6d89e0bb50fdb9cabe737cb15ebb6a894b19
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1868929Reviewed-by: Dale Curtis <dalecurtis@chromium.org>
Commit-Queue: Takashi Sakamoto <tasak@google.com>
Cr-Commit-Position: refs/heads/master@{#707758}

media/base/pipeline_impl.cc

@@ -256,23 +256,24 @@ void PipelineImpl::RendererWrapper::Start(
   SerialRunner::Queue fns;
 
   // Initialize demuxer.
-  fns.Push(base::BindRepeating(&RendererWrapper::InitializeDemuxer,
-                               weak_factory_.GetWeakPtr()));
+  fns.Push(base::BindOnce(&RendererWrapper::InitializeDemuxer,
+                          weak_factory_.GetWeakPtr()));
 
   // Once the demuxer is initialized successfully, media metadata must be
   // available - report the metadata to client. If starting without a renderer
   // we'll complete initialization at this point.
-  fns.Push(base::BindRepeating(&RendererWrapper::ReportMetadata,
-                               weak_factory_.GetWeakPtr(), start_type));
+  fns.Push(base::BindOnce(&RendererWrapper::ReportMetadata,
+                          weak_factory_.GetWeakPtr(), start_type));
 
   // Initialize renderer.
-  fns.Push(base::BindRepeating(&RendererWrapper::InitializeRenderer,
-                               weak_factory_.GetWeakPtr()));
+  fns.Push(base::BindOnce(&RendererWrapper::InitializeRenderer,
+                          weak_factory_.GetWeakPtr()));
 
   // Run tasks.
   pending_callbacks_ = SerialRunner::Run(
-      fns, base::BindRepeating(&RendererWrapper::CompleteSeek,
-                               weak_factory_.GetWeakPtr(), base::TimeDelta()));
+      std::move(fns),
+      base::BindRepeating(&RendererWrapper::CompleteSeek,
+                          weak_factory_.GetWeakPtr(), base::TimeDelta()));
 }
 
 void PipelineImpl::RendererWrapper::Stop() {
@@ -334,16 +335,16 @@ void PipelineImpl::RendererWrapper::Seek(base::TimeDelta time) {
 
   // Flush.
   DCHECK(shared_state_.renderer);
-  bound_fns.Push(base::BindRepeating(
+  bound_fns.Push(base::BindOnce(
       &Renderer::Flush, base::Unretained(shared_state_.renderer.get())));
 
   // Seek demuxer.
-  bound_fns.Push(base::BindRepeating(&Demuxer::Seek, base::Unretained(demuxer_),
-                                     seek_timestamp));
+  bound_fns.Push(base::BindOnce(&Demuxer::Seek, base::Unretained(demuxer_),
+                                seek_timestamp));
 
   // Run tasks.
   pending_callbacks_ = SerialRunner::Run(
-      bound_fns,
+      std::move(bound_fns),
       base::BindRepeating(&RendererWrapper::CompleteSeek,
                           weak_factory_.GetWeakPtr(), seek_timestamp));
 }
@@ -376,8 +377,8 @@ void PipelineImpl::RendererWrapper::Suspend() {
 
   // No need to flush the renderer since it's going to be destroyed.
   pending_callbacks_ = SerialRunner::Run(
-      fns, base::BindRepeating(&RendererWrapper::CompleteSuspend,
-                               weak_factory_.GetWeakPtr()));
+      std::move(fns), base::BindRepeating(&RendererWrapper::CompleteSuspend,
+                                          weak_factory_.GetWeakPtr()));
 }
 
 void PipelineImpl::RendererWrapper::Resume(std::unique_ptr<Renderer> renderer,
@@ -419,8 +420,9 @@ void PipelineImpl::RendererWrapper::Resume(std::unique_ptr<Renderer> renderer,
                                weak_factory_.GetWeakPtr()));
 
   pending_callbacks_ = SerialRunner::Run(
-      fns, base::BindRepeating(&RendererWrapper::CompleteSeek,
-                               weak_factory_.GetWeakPtr(), start_timestamp));
+      std::move(fns),
+      base::BindRepeating(&RendererWrapper::CompleteSeek,
+                          weak_factory_.GetWeakPtr(), start_timestamp));
 }
 
 void PipelineImpl::RendererWrapper::SetPlaybackRate(double playback_rate) {

media/base/pipeline_status.h

@@ -58,7 +58,7 @@ enum PipelineStatus {
   PIPELINE_STATUS_MAX = DEMUXER_ERROR_DETECTED_HLS,
 };
 
-typedef base::Callback<void(PipelineStatus)> PipelineStatusCB;
+typedef base::RepeatingCallback<void(PipelineStatus)> PipelineStatusCB;
 
 struct PipelineDecoderInfo {
   bool is_platform_decoder = false;

media/base/serial_runner.cc

@@ -13,20 +13,18 @@
 namespace media {
 
 // Converts a Closure into a bound function accepting a PipelineStatusCB.
-static void RunClosure(
-    const base::Closure& closure,
-    const PipelineStatusCB& status_cb) {
-  closure.Run();
+static void RunClosure(base::OnceClosure closure,
+                       const PipelineStatusCB& status_cb) {
+  std::move(closure).Run();
   status_cb.Run(PIPELINE_OK);
 }
 
 // Converts a bound function accepting a Closure into a bound function
 // accepting a PipelineStatusCB. Since closures have no way of reporting a
 // status |status_cb| is executed with PIPELINE_OK.
-static void RunBoundClosure(
-    const SerialRunner::BoundClosure& bound_closure,
-    const PipelineStatusCB& status_cb) {
-  bound_closure.Run(base::Bind(status_cb, PIPELINE_OK));
+static void RunBoundClosure(SerialRunner::BoundClosure bound_closure,
+                            const PipelineStatusCB& status_cb) {
+  std::move(bound_closure).Run(base::BindRepeating(status_cb, PIPELINE_OK));
 }
 
 // Runs |status_cb| with |last_status| on |task_runner|.
@@ -36,30 +34,29 @@ static void RunOnTaskRunner(
     PipelineStatus last_status) {
   // Force post to permit cancellation of a series in the scenario where all
   // bound functions run on the same thread.
-  task_runner->PostTask(FROM_HERE, base::BindOnce(status_cb, last_status));
+  task_runner->PostTask(FROM_HERE, base::BindRepeating(status_cb, last_status));
 }
 
 SerialRunner::Queue::Queue() = default;
-SerialRunner::Queue::Queue(const Queue& other) = default;
+SerialRunner::Queue::Queue(Queue&& other) = default;
 SerialRunner::Queue::~Queue() = default;
 
-void SerialRunner::Queue::Push(const base::Closure& closure) {
-  bound_fns_.push(base::Bind(&RunClosure, closure));
+void SerialRunner::Queue::Push(base::OnceClosure closure) {
+  bound_fns_.push_back(base::BindOnce(&RunClosure, std::move(closure)));
 }
 
-void SerialRunner::Queue::Push(
-    const BoundClosure& bound_closure) {
-  bound_fns_.push(base::Bind(&RunBoundClosure, bound_closure));
+void SerialRunner::Queue::Push(BoundClosure bound_closure) {
+  bound_fns_.push_back(
+      base::BindOnce(&RunBoundClosure, std::move(bound_closure)));
 }
 
-void SerialRunner::Queue::Push(
-    const BoundPipelineStatusCB& bound_status_cb) {
-  bound_fns_.push(bound_status_cb);
+void SerialRunner::Queue::Push(BoundPipelineStatusCB bound_status_cb) {
+  bound_fns_.push_back(std::move(bound_status_cb));
 }
 
 SerialRunner::BoundPipelineStatusCB SerialRunner::Queue::Pop() {
-  BoundPipelineStatusCB bound_fn = bound_fns_.front();
-  bound_fns_.pop();
+  BoundPipelineStatusCB bound_fn = std::move(bound_fns_.front());
+  bound_fns_.pop_front();
   return bound_fn;
 }
 
@@ -67,10 +64,9 @@ bool SerialRunner::Queue::empty() {
   return bound_fns_.empty();
 }
 
-SerialRunner::SerialRunner(const Queue& bound_fns,
-                           const PipelineStatusCB& done_cb)
+SerialRunner::SerialRunner(Queue&& bound_fns, const PipelineStatusCB& done_cb)
     : task_runner_(base::ThreadTaskRunnerHandle::Get()),
-      bound_fns_(bound_fns),
+      bound_fns_(std::move(bound_fns)),
       done_cb_(done_cb) {
   // Respect both cancellation and calling stack guarantees for |done_cb|
   // when empty.
@@ -87,10 +83,10 @@ SerialRunner::SerialRunner(const Queue& bound_fns,
 SerialRunner::~SerialRunner() = default;
 
 std::unique_ptr<SerialRunner> SerialRunner::Run(
-    const Queue& bound_fns,
+    Queue&& bound_fns,
     const PipelineStatusCB& done_cb) {
   std::unique_ptr<SerialRunner> callback_series(
-      new SerialRunner(bound_fns, done_cb));
+      new SerialRunner(std::move(bound_fns), done_cb));
   return callback_series;
 }
 
@@ -104,10 +100,10 @@ void SerialRunner::RunNextInSeries(PipelineStatus last_status) {
   }
 
   BoundPipelineStatusCB bound_fn = bound_fns_.Pop();
-  bound_fn.Run(base::Bind(
-      &RunOnTaskRunner,
-      task_runner_,
-      base::Bind(&SerialRunner::RunNextInSeries, weak_factory_.GetWeakPtr())));
+  std::move(bound_fn).Run(
+      base::BindRepeating(&RunOnTaskRunner, task_runner_,
+                          base::BindRepeating(&SerialRunner::RunNextInSeries,
+                                              weak_factory_.GetWeakPtr())));
 }
 
 }  // namespace media

media/base/serial_runner.h

@@ -8,7 +8,7 @@
 #include <memory>
 
 #include "base/callback.h"
-#include "base/containers/queue.h"
+#include "base/containers/circular_deque.h"
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/weak_ptr.h"
@@ -26,28 +26,31 @@ namespace media {
 // the completion callback as the series progresses.
 class MEDIA_EXPORT SerialRunner {
  public:
-  // TODO(dalecurtis): Change SerialRunner to use OnceCallback.
-  typedef base::Callback<void(const base::Closure&)> BoundClosure;
-  typedef base::Callback<void(const PipelineStatusCB&)> BoundPipelineStatusCB;
+  typedef base::OnceCallback<void(const base::Closure&)> BoundClosure;
+  typedef base::OnceCallback<void(const PipelineStatusCB&)>
+      BoundPipelineStatusCB;
 
   // Serial queue of bound functions to run.
   class MEDIA_EXPORT Queue {
    public:
     Queue();
-    Queue(const Queue& other);
+    Queue(Queue&& other);
     ~Queue();
 
-    void Push(const base::Closure& closure);
-    void Push(const BoundClosure& bound_fn);
-    void Push(const BoundPipelineStatusCB& bound_fn);
+    void Push(base::OnceClosure closure);
+    void Push(BoundClosure bound_fn);
+    void Push(BoundPipelineStatusCB bound_fn);
 
    private:
+    Queue(const Queue&) = delete;
+    Queue& operator=(const Queue&) = delete;
+
     friend class SerialRunner;
 
     BoundPipelineStatusCB Pop();
     bool empty();
 
-    base::queue<BoundPipelineStatusCB> bound_fns_;
+    base::circular_deque<BoundPipelineStatusCB> bound_fns_;
   };
 
   // Executes the bound functions in series, executing |done_cb| when finished.
@@ -64,13 +67,13 @@ class MEDIA_EXPORT SerialRunner {
   //
   // Deleting the object will prevent execution of any unstarted bound
   // functions, including |done_cb|.
-  static std::unique_ptr<SerialRunner> Run(const Queue& bound_fns,
+  static std::unique_ptr<SerialRunner> Run(Queue&& bound_fns,
                                            const PipelineStatusCB& done_cb);
 
  private:
   friend std::default_delete<SerialRunner>;
 
-  SerialRunner(const Queue& bound_fns, const PipelineStatusCB& done_cb);
+  SerialRunner(Queue&& bound_fns, const PipelineStatusCB& done_cb);
   ~SerialRunner();
 
   void RunNextInSeries(PipelineStatus last_status);

media/base/serial_runner_unittest.cc

@@ -25,8 +25,9 @@ class SerialRunnerTest : public ::testing::Test {
 
   void RunSerialRunner() {
     task_environment_.GetMainThreadTaskRunner()->PostTask(
-        FROM_HERE, base::BindOnce(&SerialRunnerTest::StartRunnerInternal,
-                                  base::Unretained(this), bound_fns_));
+        FROM_HERE,
+        base::BindOnce(&SerialRunnerTest::StartRunnerInternal,
+                       base::Unretained(this), std::move(bound_fns_)));
     base::RunLoop().RunUntilIdle();
   }
 
@@ -34,32 +35,29 @@ class SerialRunnerTest : public ::testing::Test {
   // |status|. called(i) returns whether the i'th bound function pushed to the
   // queue was called while running the SerialRunner.
   void PushBoundFunction(PipelineStatus status) {
-    bound_fns_.Push(base::Bind(&SerialRunnerTest::RunBoundFunction,
-                               base::Unretained(this),
-                               status,
-                               called_.size()));
+    bound_fns_.Push(base::BindOnce(&SerialRunnerTest::RunBoundFunction,
+                                   base::Unretained(this), status,
+                                   called_.size()));
     called_.push_back(false);
   }
 
   void PushBoundClosure() {
-    bound_fns_.Push(base::Bind(&SerialRunnerTest::RunBoundClosure,
-                               base::Unretained(this),
-                               called_.size()));
+    bound_fns_.Push(base::BindOnce(&SerialRunnerTest::RunBoundClosure,
+                                   base::Unretained(this), called_.size()));
     called_.push_back(false);
   }
 
   void PushClosure() {
-    bound_fns_.Push(base::Bind(&SerialRunnerTest::RunClosure,
-                               base::Unretained(this),
-                               called_.size()));
+    bound_fns_.Push(base::BindOnce(&SerialRunnerTest::RunClosure,
+                                   base::Unretained(this), called_.size()));
     called_.push_back(false);
   }
 
   // Push a bound function to the queue that will delete the SerialRunner,
   // which should cancel all remaining queued work.
   void PushCancellation() {
-    bound_fns_.Push(base::Bind(&SerialRunnerTest::CancelSerialRunner,
-                               base::Unretained(this)));
+    bound_fns_.Push(base::BindOnce(&SerialRunnerTest::CancelSerialRunner,
+                                   base::Unretained(this)));
   }
 
   // Queries final status of pushed functions and done callback. Valid only
@@ -101,10 +99,12 @@ class SerialRunnerTest : public ::testing::Test {
     called_[index] = true;
   }
 
-  void StartRunnerInternal(const SerialRunner::Queue& bound_fns) {
+  void StartRunnerInternal(SerialRunner::Queue bound_fns) {
     inside_start_ = true;
-    runner_ = SerialRunner::Run(bound_fns_, base::Bind(
-        &SerialRunnerTest::DoneCallback, base::Unretained(this)));
+    runner_ =
+        SerialRunner::Run(std::move(bound_fns),
+                          base::BindRepeating(&SerialRunnerTest::DoneCallback,
+                                              base::Unretained(this)));
     inside_start_ = false;
   }