Change MediaFilter::Stop() to accept a callback so that Stop() is asynchronous.

media/base/pipeline_impl.cc

 // Copyright (c) 2010 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.
 //
 // TODO(scherkus): clean up PipelineImpl... too many crazy function names,
 // potential deadlocks, etc...
 
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/condition_variable.h"
@@ skipping 331 matching lines @@
 bool PipelineImpl::IsPipelineInitializing() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   return state_ == kInitDataSource ||
          state_ == kInitDemuxer ||
          state_ == kInitAudioDecoder ||
          state_ == kInitAudioRenderer ||
          state_ == kInitVideoDecoder ||
          state_ == kInitVideoRenderer;
 }
 
-// static
-bool PipelineImpl::StateTransitionsToStarted(State state) {
-  return state == kPausing || state == kSeeking || state == kStarting;
+bool PipelineImpl::IsPipelineStopped() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  return state_ == kStopped || state_ == kError;
+}
+
+void PipelineImpl::FinishInitialization() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  // Execute the seek callback, if present.  Note that this might be the
+  // initial callback passed into Start().
+  if (seek_callback_.get()) {
+    seek_callback_->Run();
+    seek_callback_.reset();
+  }
+  filter_factory_ = NULL;
 }
 
 // static
+bool PipelineImpl::TransientState(State state) {
+  return state == kPausing ||
+         state == kSeeking ||
+         state == kStarting ||
+         state == kStopping;
+}
+
+// static
 PipelineImpl::State PipelineImpl::FindNextState(State current) {
   // TODO(scherkus): refactor InitializeTask() to make use of this function.
   if (current == kPausing)
     return kSeeking;
   if (current == kSeeking)
     return kStarting;
   if (current == kStarting)
     return kStarted;
+  if (current == kStopping)
+    return kStopped;
   return current;
 }
 
 void PipelineImpl::SetError(PipelineError error) {
   DCHECK(IsRunning());
   DCHECK(error != PIPELINE_OK) << "PIPELINE_OK isn't an error!";
   LOG(INFO) << "Media pipeline error: " << error;
 
-  AutoLock auto_lock(lock_);
-  error_ = error;
   message_loop_->PostTask(FROM_HERE,
      NewRunnableMethod(this, &PipelineImpl::ErrorChangedTask, error));
 }
 
 base::TimeDelta PipelineImpl::GetTime() const {
   DCHECK(IsRunning());
   return GetCurrentTime();
 }
 
 base::TimeDelta PipelineImpl::GetDuration() const {
@@ skipping 138 matching lines @@
 // FilterHost's InitializationComplete() method, the pipeline will update its
 // state to kStarted and |init_callback_|, will be executed.
 //
 // TODO(hclam): InitializeTask() is now starting the pipeline asynchronously. It
 // works like a big state change table. If we no longer need to start filters
 // in order, we need to get rid of all the state change.
 void PipelineImpl::InitializeTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // If we have received the stop or error signal, return immediately.
-  if (state_ == kStopped || state_ == kError)
+  if (state_ == kStopping || IsPipelineStopped())
     return;
 
   DCHECK(state_ == kCreated || IsPipelineInitializing());
 
   // Just created, create data source.
   if (state_ == kCreated) {
     state_ = kInitDataSource;
     CreateDataSource();
     return;
   }
@@ skipping 39 matching lines @@
       return;
     }
   }
 
   if (state_ == kInitVideoRenderer) {
     if (!IsPipelineOk() || !HasRenderedMimeTypes()) {
       SetError(PIPELINE_ERROR_COULD_NOT_RENDER);
       return;
     }
 
-    // We've successfully created and initialized every filter, so we no longer
-    // need the filter factory.
-    filter_factory_ = NULL;
-
     // Initialization was successful, we are now considered paused, so it's safe
     // to set the initial playback rate and volume.
     PlaybackRateChangedTask(GetPlaybackRate());
     VolumeChangedTask(GetVolume());
 
     // Fire the initial seek request to get the filters to preroll.
     state_ = kSeeking;
     remaining_transitions_ = filters_.size();
     seek_timestamp_ = base::TimeDelta();
     filters_.front()->Seek(seek_timestamp_,
         NewCallback(this, &PipelineImpl::OnFilterStateTransition));
   }
 }
 
 // This method is called as a result of the client calling Pipeline::Stop() or
-// as the result of an error condition.  If there is no error, then set the
-// pipeline's |error_| member to PIPELINE_STOPPING.  We stop the filters in the
-// reverse order.
+// 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.
 void PipelineImpl::StopTask(PipelineCallback* stop_callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
+  PipelineError error = GetError();
+
+  if (state_ == kStopped || (state_ == kStopping && error == PIPELINE_OK)) {

[scherkus (not reviewing), 2010/06/02 02:13:05]

this bit of logic is confusing me...

so if we're stopping and then SetError() is called we'll save the error value
but continue stopping?

if we're in error state and stoptask is called we'll overwrite the error_ with
PIPELINE_OK?

also the comment seems to be out of date.. I believe StopTask can only be
executed as a result of calling Stop() -- not an error condition

we can talk about it in person

+    // If we are already stopped or stopping normally, return immediately.
+    delete stop_callback;
+    return;
+  } else if (state_ == kError ||

[scherkus (not reviewing), 2010/06/02 02:13:05]

nit: we return, don't need else

+             (state_ == kStopping && error != PIPELINE_OK)) {
+    // If we are stopping due to SetError(), stop normally instead of
+    // going to error state.
+    AutoLock auto_lock(lock_);
+    error_ = PIPELINE_OK;
+  }
+
   stop_callback_.reset(stop_callback);
 
-  // If we've already stopped, return immediately.
-  if (state_ == kStopped) {
-    return;
+  if (IsPipelineInitializing()) {
+    FinishInitialization();
   }
 
-  // Carry out setting the error, notifying the client and destroying filters.
-  ErrorChangedTask(PIPELINE_STOPPING);
-
-  // We no longer need to examine our previous state, set it to stopped.
-  state_ = kStopped;
-
-  // Reset the pipeline.
-  ResetState();
-
-  // Notify the client that stopping has finished.
-  if (stop_callback_.get()) {
-    stop_callback_->Run();
-    stop_callback_.reset();
-  }
+  StartDestroyingFilters();
 }
 
 void PipelineImpl::ErrorChangedTask(PipelineError error) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK_NE(PIPELINE_OK, error) << "PIPELINE_OK isn't an error!";
 
-  // Suppress executing additional error logic.
-  // TODO(hclam): Remove the condition for kStopped. It is there only because
-  // FFmpegDemuxer submits a read error while reading after it is called to
-  // stop. After FFmpegDemuxer is cleaned up we should remove this condition
-  // and add an extra assert.
-  if (state_ == kError || state_ == kStopped) {
+  // 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) {
     return;
   }
 
+  AutoLock auto_lock(lock_);
+  error_ = error;
+
   // Notify the client that starting did not complete, if necessary.
-  if (IsPipelineInitializing() && seek_callback_.get()) {
-    seek_callback_->Run();
+  if (IsPipelineInitializing()) {
+    FinishInitialization();
   }
-  seek_callback_.reset();
-  filter_factory_ = NULL;
 
-  // We no longer need to examine our previous state, set it to stopped.
-  state_ = kError;
-
-  // Destroy every filter and reset the pipeline as well.
-  DestroyFilters();
-
-  // If our owner has requested to be notified of an error, execute
-  // |error_callback_| unless we have a "good" error.
-  if (error_callback_.get() && error != PIPELINE_STOPPING) {
-    error_callback_->Run();
-  }
+  StartDestroyingFilters();
 }
 
 void PipelineImpl::PlaybackRateChangedTask(float playback_rate) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   {
     AutoLock auto_lock(lock_);
     clock_.SetPlaybackRate(playback_rate);
   }
   for (FilterVector::iterator iter = filters_.begin();
        iter != filters_.end();
@@ skipping 97 matching lines @@
        iter != filters_.end();
        ++iter) {
     (*iter)->OnAudioRendererDisabled();
   }
 }
 
 void PipelineImpl::FilterStateTransitionTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // No reason transitioning if we've errored or have stopped.
-  if (state_ == kError || state_ == kStopped) {
+  if (IsPipelineStopped()) {
     return;
   }
 
-  if (!StateTransitionsToStarted(state_)) {
+  if (!TransientState(state_)) {
     NOTREACHED() << "Invalid current state: " << state_;
     SetError(PIPELINE_ERROR_ABORT);
     return;
   }
 
   // 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);
   if (--remaining_transitions_ == 0) {
     state_ = FindNextState(state_);
     if (state_ == kSeeking) {
       AutoLock auto_lock(lock_);
       clock_.SetTime(seek_timestamp_);
     }
 
-    if (StateTransitionsToStarted(state_)) {
+    if (TransientState(state_)) {
       remaining_transitions_ = filters_.size();
     }
   }
 
   // Carry out the action for the current state.
-  if (StateTransitionsToStarted(state_)) {
+  if (TransientState(state_)) {
     MediaFilter* filter = filters_[filters_.size() - remaining_transitions_];
     if (state_ == kPausing) {
       filter->Pause(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kSeeking) {
       filter->Seek(seek_timestamp_,
           NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kStarting) {
       filter->Play(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+    } else if (state_ == kStopping) {
+      filter->Stop(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else {
       NOTREACHED();
     }
   } else if (state_ == kStarted) {
-    // Execute the seek callback, if present.  Note that this might be the
-    // initial callback passed into Start().
-    if (seek_callback_.get()) {
-      seek_callback_->Run();
-      seek_callback_.reset();
-    }
+    FinishInitialization();
 
     // Finally, reset our seeking timestamp back to zero.
     seek_timestamp_ = base::TimeDelta();
 
     AutoLock auto_lock(lock_);
     // We use audio stream to update the clock. So if there is such a stream,
     // we pause the clock until we receive a valid timestamp.
     waiting_for_clock_update_ =
         rendered_mime_types_.find(mime_type::kMajorTypeAudio) !=
         rendered_mime_types_.end();
     if (!waiting_for_clock_update_)
       clock_.Play();
+  } else if (IsPipelineStopped()) {
+    FinishDestroyingFiltersTask();
   } else {
     NOTREACHED();
   }
 }
 
+void PipelineImpl::FinishDestroyingFiltersTask() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(IsPipelineStopped());
+
+  // Stop every running filter thread.
+  //
+  // TODO(scherkus): can we watchdog this section to detect wedged threads?
+  for (FilterThreadVector::iterator iter = filter_threads_.begin();
+       iter != filter_threads_.end();
+       ++iter) {
+    (*iter)->Stop();
+  }
+
+  // Reset the pipeline, which will decrement a reference to this object.
+  // We will get destroyed as soon as the remaining tasks finish executing.
+  // To be safe, we'll set our pipeline reference to NULL.
+  filters_.clear();
+  filter_types_.clear();
+  STLDeleteElements(&filter_threads_);
+
+  if (PIPELINE_OK == GetError()) {
+    // Destroying filters due to Stop().
+    ResetState();
+
+    // Notify the client that stopping has finished.
+    if (stop_callback_.get()) {
+      stop_callback_->Run();
+      stop_callback_.reset();
+    }
+  } else {
+    // Destroying filters due to SetError().
+    state_ = kError;
+    // If our owner has requested to be notified of an error.
+    if (error_callback_.get()) {
+      error_callback_->Run();
+    }
+  }
+}
+
 template <class Filter, class Source>
 void PipelineImpl::CreateFilter(FilterFactory* filter_factory,
                                 Source source,
                                 const MediaFormat& media_format) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(IsPipelineOk());
 
   // Create the filter.
   scoped_refptr<Filter> filter = filter_factory->Create<Filter>(media_format);
   if (!filter) {
@@ skipping 92 matching lines @@
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   FilterTypeMap::const_iterator ft = filter_types_.find(Filter::filter_type());
   if (ft == filter_types_.end()) {
     *filter_out = NULL;
   } else {
     *filter_out = reinterpret_cast<Filter*>(ft->second.get());
   }
 }
 
-void PipelineImpl::DestroyFilters() {
-  // Stop every filter.
-  for (FilterVector::iterator iter = filters_.begin();
-       iter != filters_.end();
-       ++iter) {
-    (*iter)->Stop();
+void PipelineImpl::StartDestroyingFilters() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK_NE(kStopped, state_);
+
+  if (state_ == kStopping) {
+    return; // Do not call Stop() on filters twice.

[scherkus (not reviewing), 2010/06/02 02:13:05]

nit: two spaces between ; and //

   }
 
-  // Crude blocking counter implementation.
-  Lock lock;
-  ConditionVariable wait_for_zero(&lock);
-  int count = filter_threads_.size();
-
-  // Post a task to every filter's thread to ensure that they've completed their
-  // stopping logic before stopping the threads themselves.
-  //
-  // TODO(scherkus): again, Stop() should either be synchronous or we should
-  // receive a signal from filters that they have indeed stopped.
-  for (FilterThreadVector::iterator iter = filter_threads_.begin();
-       iter != filter_threads_.end();
-       ++iter) {
-    (*iter)->message_loop()->PostTask(FROM_HERE,
-        NewRunnableFunction(&DecrementCounter, &lock, &wait_for_zero, &count));
+  remaining_transitions_ = filters_.size();
+  if (remaining_transitions_ > 0) {
+    state_ = kStopping;
+    filters_.front()->Stop(NewCallback(
+        this, &PipelineImpl::OnFilterStateTransition));
+  } else {
+    state_ = kStopped;
+    message_loop_->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &PipelineImpl::FinishDestroyingFiltersTask));
   }
-
-  // Wait on our "blocking counter".
-  {
-    AutoLock auto_lock(lock);
-    while (count > 0) {
-      wait_for_zero.Wait();
-    }
-  }
-
-  // Stop every running filter thread.
-  //
-  // TODO(scherkus): can we watchdog this section to detect wedged threads?
-  for (FilterThreadVector::iterator iter = filter_threads_.begin();
-       iter != filter_threads_.end();
-       ++iter) {
-    (*iter)->Stop();
-  }
-
-  // Reset the pipeline, which will decrement a reference to this object.
-  // We will get destroyed as soon as the remaining tasks finish executing.
-  // To be safe, we'll set our pipeline reference to NULL.
-  filters_.clear();
-  filter_types_.clear();
-  STLDeleteElements(&filter_threads_);
 }
 
 }  // namespace media