Refactor PipelineImpl to use CompositeFilter to manage Filter state transitions.

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"
 #include "base/stl_util-inl.h"
 #include "media/base/clock_impl.h"
 #include "media/base/filter_collection.h"
 #include "media/base/media_format.h"
 #include "media/base/pipeline_impl.h"
 
 namespace media {
 
 class PipelineImpl::PipelineInitState {
  public:
   scoped_refptr<DataSource> data_source_;
   scoped_refptr<Demuxer> demuxer_;
   scoped_refptr<AudioDecoder> audio_decoder_;
   scoped_refptr<VideoDecoder> video_decoder_;
+  scoped_refptr<CompositeFilter> composite_;
 };
 
 PipelineImpl::PipelineImpl(MessageLoop* message_loop)
     : message_loop_(message_loop),
       clock_(new ClockImpl(&base::Time::Now)),
       waiting_for_clock_update_(false),
       state_(kCreated),
-      remaining_transitions_(0),
       current_bytes_(0) {
   ResetState();
 }
 
 PipelineImpl::~PipelineImpl() {
   AutoLock auto_lock(lock_);
   DCHECK(!running_) << "Stop() must complete before destroying object";
   DCHECK(!stop_pending_);
   DCHECK(!seek_pending_);
 }
@@ skipping 267 matching lines @@
   video_height_     = 0;
   volume_           = 1.0f;
   playback_rate_    = 0.0f;
   error_            = PIPELINE_OK;
   waiting_for_clock_update_ = false;
   audio_disabled_   = false;
   clock_->SetTime(kZero);
   rendered_mime_types_.clear();
 }
 
+void PipelineImpl::set_state(State next_state) {
+  state_ = next_state;
+}
+
 bool PipelineImpl::IsPipelineOk() {
   return PIPELINE_OK == GetError();
 }
 
 bool PipelineImpl::IsPipelineInitializing() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   return state_ == kInitDataSource ||
          state_ == kInitDemuxer ||
          state_ == kInitAudioDecoder ||
          state_ == kInitAudioRenderer ||
@@ skipping 235 matching lines @@
   if (IsPipelineStopPending() ||
       IsPipelineStopped() ||
       PIPELINE_OK != GetError()) {
     return;
   }
 
   DCHECK(state_ == kCreated || IsPipelineInitializing());
 
   // Just created, create data source.
   if (state_ == kCreated) {
-    state_ = kInitDataSource;
+    set_state(kInitDataSource);
     pipeline_init_state_.reset(new PipelineInitState());
+    pipeline_init_state_->composite_ = new CompositeFilter(message_loop_);
+    pipeline_init_state_->composite_->set_host(this);
+
     InitializeDataSource();
     return;
   }
 
   // Data source created, create demuxer.
   if (state_ == kInitDataSource) {
-    state_ = kInitDemuxer;
+    set_state(kInitDemuxer);
     InitializeDemuxer(pipeline_init_state_->data_source_);
     return;
   }
 
   // Demuxer created, create audio decoder.
   if (state_ == kInitDemuxer) {
-    state_ = kInitAudioDecoder;
+    set_state(kInitAudioDecoder);
     // If this method returns false, then there's no audio stream.
     if (InitializeAudioDecoder(pipeline_init_state_->demuxer_))
       return;
   }
 
   // Assuming audio decoder was created, create audio renderer.
   if (state_ == kInitAudioDecoder) {
-    state_ = kInitAudioRenderer;
+    set_state(kInitAudioRenderer);
     // Returns false if there's no audio stream.
     if (InitializeAudioRenderer(pipeline_init_state_->audio_decoder_)) {
       InsertRenderedMimeType(mime_type::kMajorTypeAudio);
       return;
     }
   }
 
   // Assuming audio renderer was created, create video decoder.
   if (state_ == kInitAudioRenderer) {
     // Then perform the stage of initialization, i.e. initialize video decoder.
-    state_ = kInitVideoDecoder;
+    set_state(kInitVideoDecoder);
     if (InitializeVideoDecoder(pipeline_init_state_->demuxer_))
       return;
   }
 
   // Assuming video decoder was created, create video renderer.
   if (state_ == kInitVideoDecoder) {
-    state_ = kInitVideoRenderer;
+    set_state(kInitVideoRenderer);
     if (InitializeVideoRenderer(pipeline_init_state_->video_decoder_)) {
       InsertRenderedMimeType(mime_type::kMajorTypeVideo);
       return;
     }
   }
 
   if (state_ == kInitVideoRenderer) {
     if (!IsPipelineOk() || !HasRenderedMimeTypes()) {
       SetError(PIPELINE_ERROR_COULD_NOT_RENDER);
       return;
     }
 
     // Clear the collection of filters.
     filter_collection_->Clear();
 
+    pipeline_filter_ = pipeline_init_state_->composite_;
+
     // Clear init state since we're done initializing.
     pipeline_init_state_.reset();
 
     // 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.
+    // Fire the seek request to get the filters to preroll.
     seek_pending_ = true;
-    state_ = kSeeking;
-    remaining_transitions_ = filters_.size();
+    set_state(kSeeking);
     seek_timestamp_ = base::TimeDelta();
-    filters_.front()->Seek(seek_timestamp_,
+    pipeline_filter_->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.
 // 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
@@ skipping 42 matching lines @@
 
   TearDownPipeline();
 }
 
 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();
-       ++iter) {
-    (*iter)->SetPlaybackRate(playback_rate);
-  }
+  pipeline_filter_->SetPlaybackRate(playback_rate);
 }
 
 void PipelineImpl::VolumeChangedTask(float volume) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   if (audio_renderer_) {
     audio_renderer_->SetVolume(volume);
   }
 }
 
@@ skipping 15 matching lines @@
   DCHECK(!seek_pending_);
   seek_pending_ = true;
 
   // We'll need to pause every filter before seeking.  The state transition
   // is as follows:
   //   kStarted/kEnded
   //   kPausing (for each filter)
   //   kSeeking (for each filter)
   //   kStarting (for each filter)
   //   kStarted
-  state_ = kPausing;
+  set_state(kPausing);
   seek_timestamp_ = time;
   seek_callback_.reset(seek_callback);
-  remaining_transitions_ = filters_.size();
 
   // Kick off seeking!
   {
     AutoLock auto_lock(lock_);
     // If we are waiting for a clock update, the clock hasn't been played yet.
     if (!waiting_for_clock_update_)
       clock_->Pause();
   }
-  filters_.front()->Pause(
+  pipeline_filter_->Pause(
       NewCallback(this, &PipelineImpl::OnFilterStateTransition));
 }
 
 void PipelineImpl::NotifyEndedTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // We can only end if we were actually playing.
   if (state_ != kStarted) {
     return;
   }
@@ skipping 12 matching lines @@
       waiting_for_clock_update_ = false;
       clock_->Play();
     }
   }
 
   if (video_renderer_ && !video_renderer_->HasEnded()) {
     return;
   }
 
   // Transition to ended, executing the callback if present.
-  state_ = kEnded;
+  set_state(kEnded);
   if (ended_callback_.get()) {
     ended_callback_->Run();
   }
 }
 
 void PipelineImpl::NotifyNetworkEventTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   if (network_callback_.get()) {
     network_callback_->Run();
   }
 }
 
 void PipelineImpl::DisableAudioRendererTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // |rendered_mime_types_| is read through public methods so we need to lock
   // this variable.
   AutoLock auto_lock(lock_);
   rendered_mime_types_.erase(mime_type::kMajorTypeAudio);
 
   audio_disabled_ = true;
 
   // Notify all filters of disabled audio renderer.
-  for (FilterVector::iterator iter = filters_.begin();
-       iter != filters_.end();
-       ++iter) {
-    (*iter)->OnAudioRendererDisabled();
-  }
+  pipeline_filter_->OnAudioRendererDisabled();
 }
 
 void PipelineImpl::FilterStateTransitionTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // No reason transitioning if we've errored or have stopped.
   if (IsPipelineStopped()) {
     return;
   }
 
   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.
-  DCHECK(remaining_transitions_ <= filters_.size());
-  DCHECK(remaining_transitions_ > 0u);
-  if (--remaining_transitions_ == 0) {
-    state_ = FindNextState(state_);
-    if (state_ == kSeeking) {
-      AutoLock auto_lock(lock_);
-      clock_->SetTime(seek_timestamp_);
-    }
-
-    if (TransientState(state_)) {
-      remaining_transitions_ = filters_.size();
-    }
+  set_state(FindNextState(state_));
+  if (state_ == kSeeking) {
+    AutoLock auto_lock(lock_);
+    clock_->SetTime(seek_timestamp_);
   }
 
   // Carry out the action for the current state.
   if (TransientState(state_)) {
-    Filter* filter = filters_[filters_.size() - remaining_transitions_];
     if (state_ == kPausing) {
-      filter->Pause(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+      pipeline_filter_->Pause(
+          NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kFlushing) {
-      // 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));
-        }
-      }
+      pipeline_filter_->Flush(
+          NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kSeeking) {
-      filter->Seek(seek_timestamp_,
+      pipeline_filter_->Seek(seek_timestamp_,
           NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kStarting) {
-      filter->Play(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+      pipeline_filter_->Play(
+          NewCallback(this,&PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kStopping) {
-      filter->Stop(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+      pipeline_filter_->Stop(
+          NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else {
-      NOTREACHED();
+      NOTREACHED() << "Unexpected state: " << state_;
     }
   } else if (state_ == kStarted) {
     FinishInitialization();
 
     // 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,
     // 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();
 
     if (IsPipelineStopPending()) {
       // We had a pending stop request need to be honored right now.
       TearDownPipeline();
     }
   } else if (IsPipelineStopped()) {
     FinishDestroyingFiltersTask();
   } else {
-    NOTREACHED();
+    NOTREACHED() << "Unexpected state: " << state_;
   }
 }
 
 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();
-  }
-
   // Clear renderer references.
   audio_renderer_ = NULL;
   video_renderer_ = NULL;
 
-  // 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();
-  STLDeleteElements(&filter_threads_);
+  pipeline_filter_ = NULL;
 
   stop_pending_ = false;
   tearing_down_ = false;
 
   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;
+    set_state(kError);
     // If our owner has requested to be notified of an error.
     if (error_callback_.get()) {
       error_callback_->Run();
     }
   }
 }
 
 bool PipelineImpl::PrepareFilter(scoped_refptr<Filter> filter) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
-  DCHECK(IsPipelineOk());
+  bool ret = pipeline_init_state_->composite_->AddFilter(filter.get());
 
-  // Create a dedicated thread for this filter if applicable.
-  if (filter->requires_message_loop()) {
-    scoped_ptr<base::Thread> thread(
-        new base::Thread(filter->message_loop_name()));
-    if (!thread.get() || !thread->Start()) {
-      NOTREACHED() << "Could not start filter thread";
-      SetError(PIPELINE_ERROR_INITIALIZATION_FAILED);
-      return false;
-    }
-
-    filter->set_message_loop(thread->message_loop());
-    filter_threads_.push_back(thread.release());
+  if (!ret) {
+    SetError(PIPELINE_ERROR_INITIALIZATION_FAILED);
   }
-
-  // Register ourselves as the filter's host.
-  DCHECK(IsPipelineOk());
-  filter->set_host(this);
-  filters_.push_back(make_scoped_refptr(filter.get()));
-  return true;
+  return ret;
 }
 
 void PipelineImpl::InitializeDataSource() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(IsPipelineOk());
 
   scoped_refptr<DataSource> data_source;
   while (true) {
     filter_collection_->SelectDataSource(&data_source);
     if (!data_source || data_source->IsUrlSupported(url_))
@@ skipping 156 matching lines @@
 }
 
 void PipelineImpl::TearDownPipeline() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK_NE(kStopped, state_);
 
   // Mark that we already start tearing down operation.
   tearing_down_ = true;
 
   if (IsPipelineInitializing()) {
-    // Notify the client that starting did not complete, if necessary.
+    // Make it look like initialization was successful.
+    pipeline_filter_ = pipeline_init_state_->composite_;
+    pipeline_init_state_.reset();
+
+    set_state(kStopping);
+    pipeline_filter_->Stop(NewCallback(
+        this, &PipelineImpl::OnFilterStateTransition));
+
     FinishInitialization();
-  }
-
-  remaining_transitions_ = filters_.size();
-  if (remaining_transitions_ > 0) {
-    if (IsPipelineInitializing()) {
-      state_ = kStopping;
-      filters_.front()->Stop(NewCallback(
-          this, &PipelineImpl::OnFilterStateTransition));
-    } else {
-      state_ = kPausing;
-      filters_.front()->Pause(NewCallback(
-          this, &PipelineImpl::OnFilterStateTransition));
-    }
+  } else if (pipeline_filter_.get()) {
+    set_state(kPausing);
+    pipeline_filter_->Pause(NewCallback(
+        this, &PipelineImpl::OnFilterStateTransition));
   } else {
-    state_ = kStopped;
+    set_state(kStopped);
     message_loop_->PostTask(FROM_HERE,
         NewRunnableMethod(this, &PipelineImpl::FinishDestroyingFiltersTask));
   }
 }
 
 }  // namespace media