Revert 70267 - Refactor PipelineImpl to use CompositeFilter to manage Filter ...

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) {
-    set_state(kInitDataSource);
+    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) {
-    set_state(kInitDemuxer);
+    state_ = kInitDemuxer;
     InitializeDemuxer(pipeline_init_state_->data_source_);
     return;
   }
 
   // Demuxer created, create audio decoder.
   if (state_ == kInitDemuxer) {
-    set_state(kInitAudioDecoder);
+    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) {
-    set_state(kInitAudioRenderer);
+    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.
-    set_state(kInitVideoDecoder);
+    state_ = kInitVideoDecoder;
     if (InitializeVideoDecoder(pipeline_init_state_->demuxer_))
       return;
   }
 
   // Assuming video decoder was created, create video renderer.
   if (state_ == kInitVideoDecoder) {
-    set_state(kInitVideoRenderer);
+    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 seek request to get the filters to preroll.
+    // Fire the initial seek request to get the filters to preroll.
     seek_pending_ = true;
-    set_state(kSeeking);
+    state_ = kSeeking;
+    remaining_transitions_ = filters_.size();
     seek_timestamp_ = base::TimeDelta();
-    pipeline_filter_->Seek(seek_timestamp_,
+    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.
 // 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);
   }
-  pipeline_filter_->SetPlaybackRate(playback_rate);
+  for (FilterVector::iterator iter = filters_.begin();
+       iter != filters_.end();
+       ++iter) {
+    (*iter)->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
-  set_state(kPausing);
+  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();
   }
-  pipeline_filter_->Pause(
+  filters_.front()->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.
-  set_state(kEnded);
+  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.
-  pipeline_filter_->OnAudioRendererDisabled();
+  for (FilterVector::iterator iter = filters_.begin();
+       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 (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.
-  set_state(FindNextState(state_));
-  if (state_ == kSeeking) {
-    AutoLock auto_lock(lock_);
-    clock_->SetTime(seek_timestamp_);
+  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();
+    }
   }
 
   // Carry out the action for the current state.
   if (TransientState(state_)) {
+    Filter* filter = filters_[filters_.size() - remaining_transitions_];
     if (state_ == kPausing) {
-      pipeline_filter_->Pause(
-          NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+      filter->Pause(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kFlushing) {
-      pipeline_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) {
-      pipeline_filter_->Seek(seek_timestamp_,
+      filter->Seek(seek_timestamp_,
           NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kStarting) {
-      pipeline_filter_->Play(
-          NewCallback(this,&PipelineImpl::OnFilterStateTransition));
+      filter->Play(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else if (state_ == kStopping) {
-      pipeline_filter_->Stop(
-          NewCallback(this, &PipelineImpl::OnFilterStateTransition));
+      filter->Stop(NewCallback(this, &PipelineImpl::OnFilterStateTransition));
     } else {
-      NOTREACHED() << "Unexpected state: " << state_;
+      NOTREACHED();
     }
   } 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() << "Unexpected state: " << state_;
+    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();
+  }
+
   // Clear renderer references.
   audio_renderer_ = NULL;
   video_renderer_ = NULL;
 
-  pipeline_filter_ = 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_);
 
   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().
-    set_state(kError);
+    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) {
-  bool ret = pipeline_init_state_->composite_->AddFilter(filter.get());
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(IsPipelineOk());
 
-  if (!ret) {
-    SetError(PIPELINE_ERROR_INITIALIZATION_FAILED);
+  // 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());
   }
-  return ret;
+
+  // Register ourselves as the filter's host.
+  DCHECK(IsPipelineOk());
+  filter->set_host(this);
+  filters_.push_back(make_scoped_refptr(filter.get()));
+  return true;
 }
 
 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()) {
-    // Make it look like initialization was successful.
-    pipeline_filter_ = pipeline_init_state_->composite_;
-    pipeline_init_state_.reset();
+    // Notify the client that starting did not complete, if necessary.
+    FinishInitialization();
+  }
 
-    set_state(kStopping);
-    pipeline_filter_->Stop(NewCallback(
-        this, &PipelineImpl::OnFilterStateTransition));
-
-    FinishInitialization();
-  } else if (pipeline_filter_.get()) {
-    set_state(kPausing);
-    pipeline_filter_->Pause(NewCallback(
-        this, &PipelineImpl::OnFilterStateTransition));
+  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 {
-    set_state(kStopped);
+    state_ = kStopped;
     message_loop_->PostTask(FROM_HERE,
         NewRunnableMethod(this, &PipelineImpl::FinishDestroyingFiltersTask));
   }
 }
 
 }  // namespace media