media: add flush stage before stop().

to make stop more clean without race condiction, we need to flush therefore when stop happen, there are no buffer exchange.

1. add seek_pending_ to track a seek operation in transition.
2. add tearing_down_ to track a stop operation in transition.
3. add stop_pending_ to track a stop that could be delayed by a seek.
4. an error while initialization will trigger a short teardown. ( stopping => stopped )
5. an error after initialization will trigger a full tear down. ( pausing => flushing =>stopping => stopped. ).

Review URL: http://codereview.chromium.org/3192008

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@57564 0039d316-1c4b-4281-b951-d872f2087c98

2 files changed, 133 insertions, 54 deletions

diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index d8b3192..73af164 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -72,6 +72,8 @@ PipelineImpl::PipelineImpl(MessageLoop* message_loop)
 PipelineImpl::~PipelineImpl() {
   AutoLock auto_lock(lock_);
   DCHECK(!running_) << "Stop() must complete before destroying object";
+  DCHECK(!stop_pending_);
+  DCHECK(!seek_pending_);
 }
 
 // Creates the PipelineInternal and calls it's start method.
@@ -321,6 +323,9 @@ void PipelineImpl::ResetState() {
   AutoLock auto_lock(lock_);
   const base::TimeDelta kZero;
   running_          = false;
+  stop_pending_     = false;
+  seek_pending_     = false;
+  tearing_down_     = false;
   duration_         = kZero;
   buffered_time_    = kZero;
   buffered_bytes_   = 0;
@@ -356,6 +361,25 @@ bool PipelineImpl::IsPipelineStopped() {
   return state_ == kStopped || state_ == kError;
 }
 
+bool PipelineImpl::IsPipelineTearingDown() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  return tearing_down_;
+}
+
+bool PipelineImpl::IsPipelineStopPending() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  return stop_pending_;
+}
+
+bool PipelineImpl::IsPipelineSeeking() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  if (!seek_pending_)
+    return false;
+  DCHECK(kSeeking == state_ || kPausing == state_ ||
+         kFlushing == state_ || kStarting == state_);
+  return true;
+}
+
 void PipelineImpl::FinishInitialization() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   // Execute the seek callback, if present.  Note that this might be the
@@ -379,17 +403,22 @@ bool PipelineImpl::TransientState(State state) {
 // static
 PipelineImpl::State PipelineImpl::FindNextState(State current) {
   // TODO(scherkus): refactor InitializeTask() to make use of this function.
-  if (current == kPausing)
+  if (current == kPausing) {
     return kFlushing;
-  if (current == kFlushing)
-    return kSeeking;
-  if (current == kSeeking)
+  } else if (current == kFlushing) {
+    // We will always honor Seek() before Stop(). This is based on the
+    // assumption that we never accept Seek() after Stop().
+    DCHECK(IsPipelineSeeking() || IsPipelineStopPending());
+    return IsPipelineSeeking() ? kSeeking : kStopping;
+  } else if (current == kSeeking) {
     return kStarting;
-  if (current == kStarting)
+  } else if (current == kStarting) {
     return kStarted;
-  if (current == kStopping)
+  } else if (current == kStopping) {
     return kStopped;
-  return current;
+  } else {
+    return current;
+  }
 }
 
 void PipelineImpl::SetError(PipelineError error) {
@@ -559,8 +588,11 @@ void PipelineImpl::InitializeTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // If we have received the stop or error signal, return immediately.
-  if (state_ == kStopping || IsPipelineStopped())
+  if (IsPipelineStopPending() ||
+      IsPipelineStopped() ||
+      PIPELINE_OK != GetError()) {
     return;
+  }
 
   DCHECK(state_ == kCreated || IsPipelineInitializing());
 
@@ -625,6 +657,7 @@ void PipelineImpl::InitializeTask() {
     VolumeChangedTask(GetVolume());
 
     // Fire the initial seek request to get the filters to preroll.
+    seek_pending_ = true;
     state_ = kSeeking;
     remaining_transitions_ = filters_.size();
     seek_timestamp_ = base::TimeDelta();
@@ -644,12 +677,12 @@ void PipelineImpl::StopTask(PipelineCallback* stop_callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   PipelineError error = GetError();
 
-  if (state_ == kStopped || (state_ == kStopping && error == PIPELINE_OK)) {
+  if (state_ == kStopped || (IsPipelineStopPending() && error == PIPELINE_OK)) {
     // If we are already stopped or stopping normally, return immediately.
     delete stop_callback;
     return;
   } else if (state_ == kError ||
-             (state_ == kStopping && error != PIPELINE_OK)) {
+             (IsPipelineStopPending() && error != PIPELINE_OK)) {
     // If we are stopping due to SetError(), stop normally instead of
     // going to error state.
     AutoLock auto_lock(lock_);
@@ -658,11 +691,14 @@ void PipelineImpl::StopTask(PipelineCallback* stop_callback) {
 
   stop_callback_.reset(stop_callback);
 
-  if (IsPipelineInitializing()) {
-    FinishInitialization();
+  stop_pending_ = true;
+  if (!IsPipelineSeeking()) {
+    // We will tear down pipeline immediately when there is no seek operation
+    // pending. This should include the case where we are partially initialized.
+    // Ideally this case should use SetError() rather than Stop() to tear down.
+    DCHECK(!IsPipelineTearingDown());
+    TearDownPipeline();
   }
-
-  StartDestroyingFilters();
 }
 
 void PipelineImpl::ErrorChangedTask(PipelineError error) {
@@ -672,19 +708,14 @@ void PipelineImpl::ErrorChangedTask(PipelineError error) {
   // Suppress executing additional error logic. Note that if we are currently
   // performing a normal stop, then we return immediately and continue the
   // normal stop.
-  if (IsPipelineStopped() || state_ == kStopping) {
+  if (IsPipelineStopped() || IsPipelineTearingDown()) {
     return;
   }
 
   AutoLock auto_lock(lock_);
   error_ = error;
 
-  // Notify the client that starting did not complete, if necessary.
-  if (IsPipelineInitializing()) {
-    FinishInitialization();
-  }
-
-  StartDestroyingFilters();
+  TearDownPipeline();
 }
 
 void PipelineImpl::PlaybackRateChangedTask(float playback_rate) {
@@ -713,6 +744,7 @@ void PipelineImpl::VolumeChangedTask(float volume) {
 void PipelineImpl::SeekTask(base::TimeDelta time,
                             PipelineCallback* seek_callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(!IsPipelineStopPending());
 
   // Suppress seeking if we're not fully started.
   if (state_ != kStarted && state_ != kEnded) {
@@ -724,6 +756,9 @@ void PipelineImpl::SeekTask(base::TimeDelta time,
     return;
   }
 
+  DCHECK(!seek_pending_);
+  seek_pending_ = true;
+
   // We'll need to pause every filter before seeking.  The state transition
   // is as follows:
   //   kStarted/kEnded
@@ -814,8 +849,8 @@ void PipelineImpl::FilterStateTransitionTask() {
 
   // Decrement the number of remaining transitions, making sure to transition
   // to the next state if needed.
-  CHECK(remaining_transitions_ <= filters_.size());
-  CHECK(remaining_transitions_ > 0u);
+  DCHECK(remaining_transitions_ <= filters_.size());
+  DCHECK(remaining_transitions_ > 0u);
   if (--remaining_transitions_ == 0) {
     state_ = FindNextState(state_);
     if (state_ == kSeeking) {
@@ -834,7 +869,13 @@ void PipelineImpl::FilterStateTransitionTask() {
     if (state_ == kPausing) {
       filter->Pause(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kFlushing) {
-      filter->Flush(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+      // We had to use parallel flushing all filters.
+      if (remaining_transitions_ == filters_.size()) {
+        for (size_t i = 0; i < filters_.size(); i++) {
+          filters_[i]->Flush(
+              NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+        }
+      }
     } else if (state_ == kSeeking) {
       filter->Seek(seek_timestamp_,
           NewCallback(this, &PipelineImpl::OnFilterStateTransition));
@@ -850,6 +891,7 @@ void PipelineImpl::FilterStateTransitionTask() {
 
     // Finally, reset our seeking timestamp back to zero.
     seek_timestamp_ = base::TimeDelta();
+    seek_pending_ = false;
 
     AutoLock auto_lock(lock_);
     // We use audio stream to update the clock. So if there is such a stream,
@@ -859,6 +901,11 @@ void PipelineImpl::FilterStateTransitionTask() {
         rendered_mime_types_.end();
     if (!waiting_for_clock_update_)
       clock_.Play();
+
+    if (IsPipelineStopPending()) {
+      // We had a pending stop request need to be honored right now.
+      TearDownPipeline();
+    }
   } else if (IsPipelineStopped()) {
     FinishDestroyingFiltersTask();
   } else {
@@ -886,6 +933,9 @@ void PipelineImpl::FinishDestroyingFiltersTask() {
   filter_types_.clear();
   STLDeleteElements(&filter_threads_);
 
+  stop_pending_ = false;
+  tearing_down_ = false;
+
   if (PIPELINE_OK == GetError()) {
     // Destroying filters due to Stop().
     ResetState();
@@ -1017,19 +1067,29 @@ void PipelineImpl::GetFilter(scoped_refptr<Filter>* filter_out) const {
   }
 }
 
-void PipelineImpl::StartDestroyingFilters() {
+void PipelineImpl::TearDownPipeline() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK_NE(kStopped, state_);
 
-  if (state_ == kStopping) {
-    return; // Do not call Stop() on filters twice.
+  // Mark that we already start tearing down operation.
+  tearing_down_ = true;
+
+  if (IsPipelineInitializing()) {
+    // Notify the client that starting did not complete, if necessary.
+    FinishInitialization();
   }
 
   remaining_transitions_ = filters_.size();
   if (remaining_transitions_ > 0) {
-    state_ = kStopping;
-    filters_.front()->Stop(NewCallback(
-        this, &PipelineImpl::OnFilterStateTransition));
+    if (IsPipelineInitializing()) {
+      state_ = kStopping;
+      filters_.front()->Stop(NewCallback(
+          this, &PipelineImpl::OnFilterStateTransition));
+    } else {
+      state_ = kPausing;
+      filters_.front()->Pause(NewCallback(
+          this, &PipelineImpl::OnFilterStateTransition));
+    }
   } else {
     state_ = kStopped;
     message_loop_->PostTask(FROM_HERE,
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index 2486f0a..48db653 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -30,31 +30,26 @@ namespace media {
 //
 // Here's a state diagram that describes the lifetime of this object.
 //
-//   [ *Created ]
-//         | Start()
-//         V
-//   [ InitXXX (for each filter) ]
-//         |
-//         V
-//   [ Seeking (for each filter) ] <----------------------.
-//         |                                              |
+//   [ *Created ]                                    [ Stopped ]
+//         | Start()                                      ^
+//         V                       SetError()             |
+//   [ InitXXX (for each filter) ] -------->[ Stopping (for each filter) ]
+//         |                                              ^
+//         V                                              | if Stop
+//   [ Seeking (for each filter) ] <--------[ Flushing (for each filter) ]
+//         |                         if Seek              ^
 //         V                                              |
 //   [ Starting (for each filter) ]                       |
 //         |                                              |
-//         V      Seek()                                  |
-//   [ Started ] --------> [ Pausing (for each filter) ] -'
-//         |                                              |
-//         |   NotifyEnded()                Seek()        |
+//         V      Seek()/Stop()                           |
+//   [ Started ] -------------------------> [ Pausing (for each filter) ]
+//         |                                              ^
+//         |   NotifyEnded()             Seek()/Stop()    |
 //         `-------------> [ Ended ] ---------------------'
-//
-//                  SetError()
-//   [ Any State ] -------------> [ Stopping (for each filter)]
-//         | Stop()                             |
-//         V                                    V
-//   [ Stopping (for each filter) ]         [ Error ]
-//         |
-//         V
-//   [ Stopped ]
+//                                                        ^  SetError()
+//                                                        |
+//                                         [ Any State Other Than InitXXX ]
+
 //
 // Initialization is a series of state transitions from "Created" through each
 // filter initialization state.  When all filter initialization states have
@@ -142,6 +137,15 @@ class PipelineImpl : public Pipeline, public FilterHost {
   // Helper method to tell whether we are stopped or in error.
   bool IsPipelineStopped();
 
+  // Helper method to tell whether we are in transition to stop state.
+  bool IsPipelineTearingDown();
+
+  // We could also be delayed by a transition during seek is performed.
+  bool IsPipelineStopPending();
+
+  // Helper method to tell whether we are in transition to seek state.
+  bool IsPipelineSeeking();
+
   // Helper method to execute callback from Start() and reset
   // |filter_factory_|. Called when initialization completes
   // normally or when pipeline is stopped or error occurs during
@@ -153,7 +157,7 @@ class PipelineImpl : public Pipeline, public FilterHost {
   static bool TransientState(State state);
 
   // Given the current state, returns the next state.
-  static State FindNextState(State current);
+  State FindNextState(State current);
 
   // FilterHost implementation.
   virtual void SetError(PipelineError error);
@@ -296,8 +300,14 @@ class PipelineImpl : public Pipeline, public FilterHost {
   template <class Filter>
   void GetFilter(scoped_refptr<Filter>* filter_out) const;
 
-  // Kicks off stopping filters. Called by StopTask() and ErrorChangedTask().
-  void StartDestroyingFilters();
+  // Kicks off destroying filters. Called by StopTask() and ErrorChangedTask().
+  // When we start to tear down the pipeline, we will consider two cases:
+  // 1. when pipeline has not been initialized, we will transit to stopping
+  // state first.
+  // 2. when pipeline has been initialized, we will first transit to pausing
+  // => flushing => stopping => stopped state.
+  // This will remove the race condition during stop between filters.
+  void TearDownPipeline();
 
   // Message loop used to execute pipeline tasks.
   MessageLoop* message_loop_;
@@ -308,6 +318,15 @@ class PipelineImpl : public Pipeline, public FilterHost {
   // Whether or not the pipeline is running.
   bool running_;
 
+  // Whether or not the pipeline is in transition for a seek operation.
+  bool seek_pending_;
+
+  // Whether or not the pipeline is pending a stop operation.
+  bool stop_pending_;
+
+  // Whether or not the pipeline is perform a stop operation.
+  bool tearing_down_;
+
   // Duration of the media in microseconds.  Set by filters.
   base::TimeDelta duration_;