Replace RunInSeries() and RunInParallel() with CallbackSeries helper class.

media/base/pipeline.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "media/base/pipeline.h"
 
 #include <algorithm>
 
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/callback_helpers.h"
 #include "base/compiler_specific.h"
 #include "base/metrics/histogram.h"
 #include "base/message_loop.h"
 #include "base/stl_util.h"
 #include "base/string_util.h"
 #include "base/synchronization/condition_variable.h"
 #include "media/base/audio_decoder.h"
 #include "media/base/audio_renderer.h"
-#include "media/base/callback_util.h"
 #include "media/base/clock.h"
 #include "media/base/filter_collection.h"
 #include "media/base/media_log.h"
 #include "media/base/video_decoder.h"
 #include "media/base/video_renderer.h"
 
 using base::TimeDelta;
 
 namespace media {
 
@@ skipping 413 matching lines @@
   // Since the byte->time calculation is approximate, fudge the beginning &
   // ending areas to look better.
   TimeDelta epsilon = clock_->Duration() / 100;
   if (time_offset < epsilon)
     return TimeDelta();
   if (time_offset + epsilon > clock_->Duration())
     return clock_->Duration();
   return time_offset;
 }
 
-void Pipeline::DoPause(const base::Closure& done_cb) {
+void Pipeline::DoPause(const PipelineStatusCB& done_cb) {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+  DCHECK(!pending_callbacks_.get());
+  scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs(
+      new std::queue<PipelineStatusCBFunc>());
 
   if (audio_renderer_)

[Ami GONE FROM CHROMIUM, 2012/08/03 04:49:57]

Not reviewing this carefully yet b/c I have hopes for simplification of the API.
But if the verbosity remains, you need to add a bunch of {}'s to conditional
bodies that are no longer single-lines.

[scherkus (not reviewing), 2012/08/03 18:08:50]

Done.

-    closures->push(base::Bind(&AudioRenderer::Pause, audio_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&AudioRenderer::Pause, audio_renderer_)));

[scherkus (not reviewing), 2012/08/02 22:17:04]

I thought of having CallbackSeries::Run() that accepts a queue of ClosureFuncs,
but after this CL I want to combine pause/flush/seek and that requires having a
queue of the same func type, so I'm fine with the RunClosureFunc business

 
   if (video_renderer_)
-    closures->push(base::Bind(&VideoRenderer::Pause, video_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&VideoRenderer::Pause, video_renderer_)));
 
-  RunInSeries(closures.Pass(), done_cb);
+  pending_callbacks_ = CallbackSeries::Run(status_cbs.Pass(), done_cb);
 }
 
-void Pipeline::DoFlush(const base::Closure& done_cb) {
+void Pipeline::DoFlush(const PipelineStatusCB& done_cb) {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+  DCHECK(!pending_callbacks_.get());
+  scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs(
+      new std::queue<PipelineStatusCBFunc>());
 
   if (audio_renderer_)
-    closures->push(base::Bind(&AudioRenderer::Flush, audio_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&AudioRenderer::Flush, audio_renderer_)));
 
   if (video_renderer_)
-    closures->push(base::Bind(&VideoRenderer::Flush, video_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&VideoRenderer::Flush, video_renderer_)));
 
-  RunInParallel(closures.Pass(), done_cb);
+  pending_callbacks_ = CallbackSeries::Run(status_cbs.Pass(), done_cb);
 }
 
-void Pipeline::DoPlay(const base::Closure& done_cb) {
+void Pipeline::DoPlay(const PipelineStatusCB& done_cb) {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+  DCHECK(!pending_callbacks_.get());
+  scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs(
+      new std::queue<PipelineStatusCBFunc>());
 
   if (audio_renderer_)
-    closures->push(base::Bind(&AudioRenderer::Play, audio_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&AudioRenderer::Play, audio_renderer_)));
 
   if (video_renderer_)
-    closures->push(base::Bind(&VideoRenderer::Play, video_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&VideoRenderer::Play, video_renderer_)));
 
-  RunInSeries(closures.Pass(), done_cb);
+  pending_callbacks_ = CallbackSeries::Run(status_cbs.Pass(), done_cb);
 }
 
-void Pipeline::DoStop(const base::Closure& done_cb) {
+void Pipeline::DoStop(const PipelineStatusCB& done_cb) {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+  DCHECK(!pending_callbacks_.get());
+  scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs(
+      new std::queue<PipelineStatusCBFunc>());
 
   if (demuxer_)
-    closures->push(base::Bind(&Demuxer::Stop, demuxer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&Demuxer::Stop, demuxer_)));
 
   if (audio_renderer_)
-    closures->push(base::Bind(&AudioRenderer::Stop, audio_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&AudioRenderer::Stop, audio_renderer_)));
 
   if (video_renderer_)
-    closures->push(base::Bind(&VideoRenderer::Stop, video_renderer_));
+    status_cbs->push(base::Bind(
+        &RunClosureFunc, base::Bind(&VideoRenderer::Stop, video_renderer_)));
 
-  RunInSeries(closures.Pass(), done_cb);
+  pending_callbacks_ = CallbackSeries::Run(status_cbs.Pass(), done_cb);
 }
 
 void Pipeline::AddBufferedByteRange(int64 start, int64 end) {
   DCHECK(IsRunning());
   base::AutoLock auto_lock(lock_);
   buffered_byte_ranges_.Add(start, end);
   did_loading_progress_ = true;
 }
 
 void Pipeline::AddBufferedTimeRange(base::TimeDelta start,
@@ skipping 21 matching lines @@
 }
 
 // Called from any thread.
 void Pipeline::OnFilterInitialize(PipelineStatus status) {
   // Continue the initialize task by proceeding to the next stage.
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::InitializeTask, this, status));
 }
 
 // Called from any thread.
-void Pipeline::OnFilterStateTransition() {
-  message_loop_->PostTask(FROM_HERE, base::Bind(
-      &Pipeline::FilterStateTransitionTask, this));
-}
-
-// Called from any thread.
 // This method makes the PipelineStatusCB behave like a Closure. It
 // makes it look like a host()->SetError() call followed by a call to
 // OnFilterStateTransition() when errors occur.
 //
 // TODO: Revisit this code when SetError() is removed from FilterHost and
 //       all the Closures are converted to PipelineStatusCB.
-void Pipeline::OnFilterStateTransitionWithStatus(PipelineStatus status) {
+void Pipeline::OnFilterStateTransition(PipelineStatus status) {
   if (status != PIPELINE_OK)
     SetError(status);
-  OnFilterStateTransition();
+  message_loop_->PostTask(FROM_HERE, base::Bind(
+      &Pipeline::FilterStateTransitionTask, this));
 }
 
-void Pipeline::OnTeardownStateTransition() {
+void Pipeline::OnTeardownStateTransition(PipelineStatus status) {
+  // Ignore any errors during teardown.
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::TeardownStateTransitionTask, this));
 }
 
 // Called from any thread.
 void Pipeline::OnUpdateStatistics(const PipelineStatistics& stats) {
   base::AutoLock auto_lock(lock_);
   statistics_.audio_bytes_decoded += stats.audio_bytes_decoded;
   statistics_.video_bytes_decoded += stats.video_bytes_decoded;
   statistics_.video_frames_decoded += stats.video_frames_decoded;
@@ skipping 110 matching lines @@
     // to set the initial playback rate and volume.
     PlaybackRateChangedTask(GetPlaybackRate());
     VolumeChangedTask(GetVolume());
 
     // Fire a seek request to get the renderers to preroll. We can skip a seek
     // here as the demuxer should be at the start of the stream.
     seek_pending_ = true;
     SetState(kSeeking);
     seek_timestamp_ = demuxer_->GetStartTime();
     DoSeek(seek_timestamp_, true,
-           base::Bind(&Pipeline::OnFilterStateTransitionWithStatus, this));
+           base::Bind(&Pipeline::OnFilterStateTransition, this));
   }
 }
 
 // This method is called as a result of the client calling Pipeline::Stop() or
 // as the result of an error condition.
 // We stop the filters in the reverse order.
 //
 // TODO(scherkus): beware!  this can get posted multiple times since we post
 // Stop() tasks even if we've already stopped.  Perhaps this should no-op for
 // additional calls, however most of this logic will be changing.
@@ skipping 174 matching lines @@
   demuxer_->OnAudioRendererDisabled();
 
   // Start clock since there is no more audio to
   // trigger clock updates.
   clock_->SetMaxTime(clock_->Duration());
   StartClockIfWaitingForTimeUpdate_Locked();
 }
 
 void Pipeline::FilterStateTransitionTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(pending_callbacks_.get())
+      << "Filter state transitions must be completed via pending_callbacks_";
+  pending_callbacks_.reset();
 
   // No reason transitioning if we've errored or have stopped.
   if (IsPipelineStopped()) {
     return;
   }
 
   // If we are tearing down, don't allow any state changes. Teardown
   // state changes will come in via TeardownStateTransitionTask().
   if (IsPipelineTearingDown()) {
     return;
@@ skipping 14 matching lines @@
   }
 
   // Carry out the action for the current state.
   if (TransientState(state_)) {
     if (state_ == kPausing) {
       DoPause(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kFlushing) {
       DoFlush(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kSeeking) {
       DoSeek(seek_timestamp_, false,
-             base::Bind(&Pipeline::OnFilterStateTransitionWithStatus, this));
+             base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kStarting) {
       DoPlay(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kStopping) {
       DoStop(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else {
       NOTREACHED() << "Unexpected state: " << state_;
     }
   } else if (state_ == kStarted) {
     FinishInitialization();
 
@@ skipping 20 matching lines @@
       // We had a pending stop request need to be honored right now.
       TearDownPipeline();
     }
   } else {
     NOTREACHED() << "Unexpected state: " << state_;
   }
 }
 
 void Pipeline::TeardownStateTransitionTask() {
   DCHECK(IsPipelineTearingDown());
+  DCHECK(pending_callbacks_.get())
+      << "Teardown state transitions must be completed via pending_callbacks_";
+  pending_callbacks_.reset();
+
   switch (state_) {
     case kStopping:
       SetState(error_caused_teardown_ ? kError : kStopped);
       FinishDestroyingFiltersTask();
       break;
     case kPausing:
       SetState(kFlushing);
       DoFlush(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
     case kFlushing:
@@ skipping 187 matching lines @@
   DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK_NE(kStopped, state_);
 
   DCHECK(!tearing_down_ ||  // Teardown on Stop().
          (tearing_down_ && error_caused_teardown_) ||  // Teardown on error.
          (tearing_down_ && stop_pending_));  // Stop during teardown by error.
 
   // Mark that we already start tearing down operation.
   tearing_down_ = true;
 
+  // Cancel any pending operation so we can proceed with teardown.
+  pending_callbacks_.reset();

[scherkus (not reviewing), 2012/08/02 22:17:04]

AFAIK this is the fix -- previously we may have had queued operations that still
ran while we were going through the teardown path

+
   switch (state_) {
     case kCreated:
     case kError:
       SetState(kStopped);
       // Need to put this in the message loop to make sure that it comes
       // after any pending callback tasks that are already queued.
       message_loop_->PostTask(FROM_HERE, base::Bind(
           &Pipeline::FinishDestroyingFiltersTask, this));
       break;
 
@@ skipping 38 matching lines @@
       break;
     // default: intentionally left out to force new states to cause compiler
     // errors.
   };
 }
 
 void Pipeline::DoSeek(base::TimeDelta seek_timestamp,
                       bool skip_demuxer_seek,
                       const PipelineStatusCB& done_cb) {
   DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(!pending_callbacks_.get());
   scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs(
       new std::queue<PipelineStatusCBFunc>());
 
   if (!skip_demuxer_seek)
-    status_cbs->push(base::Bind(&Demuxer::Seek, demuxer_, seek_timestamp));
+    status_cbs->push(base::Bind(
+        &Demuxer::Seek, demuxer_, seek_timestamp));
 
   if (audio_renderer_)
     status_cbs->push(base::Bind(
         &AudioRenderer::Preroll, audio_renderer_, seek_timestamp));
 
   if (video_renderer_)
     status_cbs->push(base::Bind(
         &VideoRenderer::Preroll, video_renderer_, seek_timestamp));
 
-  RunInSeriesWithStatus(status_cbs.Pass(), base::Bind(
-      &Pipeline::ReportStatus, this, done_cb));
+  pending_callbacks_ = CallbackSeries::Run(status_cbs.Pass(), done_cb);
 }
 
 void Pipeline::OnAudioUnderflow() {
   if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnAudioUnderflow, this));
     return;
   }
 
   if (state_ != kStarted)
     return;
 
   if (audio_renderer_)
     audio_renderer_->ResumeAfterUnderflow(true);
 }
 
 void Pipeline::StartClockIfWaitingForTimeUpdate_Locked() {
   lock_.AssertAcquired();
   if (!waiting_for_clock_update_)
     return;
 
   waiting_for_clock_update_ = false;
   clock_->Play();
 }
 
 }  // namespace media