Big media::Pipeline cleanup.

media/base/pipeline_impl.cc

 // Copyright (c) 2008-2009 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/compiler_specific.h"
 #include "base/condition_variable.h"
 #include "base/stl_util-inl.h"
@@ skipping 36 matching lines @@
 }  // namespace
 
 PipelineImpl::PipelineImpl() {
   ResetState();
 }
 
 PipelineImpl::~PipelineImpl() {
   Stop();
 }
 
-bool PipelineImpl::IsInitialized() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return initialized_;
-}
-
-base::TimeDelta PipelineImpl::GetDuration() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return duration_;
-}
-
-base::TimeDelta PipelineImpl::GetBufferedTime() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return buffered_time_;
-}
-
-int64 PipelineImpl::GetTotalBytes() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return total_bytes_;
-}
-
-int64 PipelineImpl::GetBufferedBytes() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return buffered_bytes_;
-}
-
-void PipelineImpl::GetVideoSize(size_t* width_out, size_t* height_out) const {
-  DCHECK(width_out);
-  DCHECK(height_out);
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  *width_out = video_width_;
-  *height_out = video_height_;
-}
-
-float PipelineImpl::GetVolume() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return volume_;
-}
-
-float PipelineImpl::GetPlaybackRate() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return playback_rate_;
-}
-
-base::TimeDelta PipelineImpl::GetTime() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return time_;
-}
-
-base::TimeDelta PipelineImpl::GetInterpolatedTime() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  base::TimeDelta time = time_;
-  if (playback_rate_ > 0.0f) {
-    base::TimeDelta delta = base::TimeTicks::Now() - ticks_at_last_set_time_;
-    if (playback_rate_ == 1.0f) {
-      time += delta;
-    } else {
-      int64 adjusted_delta = static_cast<int64>(delta.InMicroseconds() *
-                                                playback_rate_);
-      time += base::TimeDelta::FromMicroseconds(adjusted_delta);
-    }
-  }
-  return time;
-}
-
-void PipelineImpl::SetTime(base::TimeDelta time) {
-  AutoLock auto_lock(lock_);
-  time_ = time;
-  ticks_at_last_set_time_ = base::TimeTicks::Now();
-}
-
-void PipelineImpl::InternalSetPlaybackRate(float rate) {
-  AutoLock auto_lock(lock_);
-  if (playback_rate_ == 0.0f && rate > 0.0f) {
-    ticks_at_last_set_time_ = base::TimeTicks::Now();
-  }
-  playback_rate_ = rate;
-}
-
-PipelineError PipelineImpl::GetError() const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  return error_;
-}
-
-bool PipelineImpl::IsRendered(const std::string& major_mime_type) const {
-  AutoLock auto_lock(const_cast<Lock&>(lock_));
-  bool is_rendered = (rendered_mime_types_.find(major_mime_type) !=
-                      rendered_mime_types_.end());
-  return is_rendered;
-}
-
-
-bool PipelineImpl::InternalSetError(PipelineError error) {
-  // Don't want callers to set an error of "OK".  STOPPING is a special value
-  // that should only be used internally by the StopTask() method.
-  DCHECK(PIPELINE_OK != error && PIPELINE_STOPPING != error);
-  AutoLock auto_lock(lock_);
-  bool changed_error = false;
-  if (PIPELINE_OK == error_) {
-    error_ = error;
-    changed_error = true;
-  }
-  return changed_error;
-}
-
 // Creates the PipelineThread and calls it's start method.
 bool PipelineImpl::Start(FilterFactory* factory,
                          const std::string& url,
                          PipelineCallback* init_complete_callback) {
   DCHECK(!pipeline_thread_);
   DCHECK(factory);
   DCHECK(!initialized_);
   DCHECK(!IsPipelineThread());
   if (!pipeline_thread_ && factory) {
     pipeline_thread_ = new PipelineThread(this);
@@ skipping 14 matching lines @@
 // created PipelineImpl object.
 void PipelineImpl::Stop() {
   DCHECK(!IsPipelineThread());
 
   if (pipeline_thread_) {
     pipeline_thread_->Stop();
   }
   ResetState();
 }
 
+void PipelineImpl::Seek(base::TimeDelta time,
+                        PipelineCallback* seek_callback) {
+  DCHECK(!IsPipelineThread());
+
+  if (IsPipelineOk()) {
+    pipeline_thread_->Seek(time, seek_callback);
+  } else {
+    NOTREACHED();
+  }
+}
+
+bool PipelineImpl::IsInitialized() const {
+  AutoLock auto_lock(lock_);
+  return initialized_;
+}
+
+bool PipelineImpl::IsRendered(const std::string& major_mime_type) const {
+  AutoLock auto_lock(lock_);
+  bool is_rendered = (rendered_mime_types_.find(major_mime_type) !=
+                      rendered_mime_types_.end());
+  return is_rendered;
+}
+
+float PipelineImpl::GetPlaybackRate() const {
+  AutoLock auto_lock(lock_);
+  return playback_rate_;
+}
+
 void PipelineImpl::SetPlaybackRate(float rate) {
   DCHECK(!IsPipelineThread());
 
   if (IsPipelineOk() && rate >= 0.0f) {
     pipeline_thread_->SetPlaybackRate(rate);
   } else {
     // It's OK for a client to call SetPlaybackRate(0.0f) if we're stopped.
     DCHECK(rate == 0.0f && playback_rate_ == 0.0f);
   }
 }
 
-void PipelineImpl::Seek(base::TimeDelta time,
-                        PipelineCallback* seek_callback) {
-  DCHECK(!IsPipelineThread());
-
-  if (IsPipelineOk()) {
-    pipeline_thread_->Seek(time, seek_callback);
-  } else {
-    NOTREACHED();
-  }
+float PipelineImpl::GetVolume() const {
+  AutoLock auto_lock(lock_);
+  return volume_;
 }
 
 void PipelineImpl::SetVolume(float volume) {
   DCHECK(!IsPipelineThread());
 
   if (IsPipelineOk() && volume >= 0.0f && volume <= 1.0f) {
     pipeline_thread_->SetVolume(volume);
   } else {
     NOTREACHED();
   }
 }
 
+base::TimeDelta PipelineImpl::GetTime() const {
+  AutoLock auto_lock(lock_);
+  return time_;
+}
+
+base::TimeDelta PipelineImpl::GetBufferedTime() const {
+  AutoLock auto_lock(lock_);
+  return buffered_time_;
+}
+
+base::TimeDelta PipelineImpl::GetDuration() const {
+  AutoLock auto_lock(lock_);
+  return duration_;
+}
+
+int64 PipelineImpl::GetBufferedBytes() const {
+  AutoLock auto_lock(lock_);
+  return buffered_bytes_;
+}
+
+int64 PipelineImpl::GetTotalBytes() const {
+  AutoLock auto_lock(lock_);
+  return total_bytes_;
+}
+
+void PipelineImpl::GetVideoSize(size_t* width_out, size_t* height_out) const {
+  CHECK(width_out);
+  CHECK(height_out);
+  AutoLock auto_lock(lock_);
+  *width_out = video_width_;
+  *height_out = video_height_;
+}
+
+PipelineError PipelineImpl::GetError() const {
+  AutoLock auto_lock(lock_);
+  return error_;
+}
+
 void PipelineImpl::ResetState() {
   AutoLock auto_lock(lock_);
   pipeline_thread_  = NULL;
   initialized_      = false;
   duration_         = base::TimeDelta();
   buffered_time_    = base::TimeDelta();
   buffered_bytes_   = 0;
   total_bytes_      = 0;
   video_width_      = 0;
   video_height_     = 0;
   volume_           = 0.0f;
   playback_rate_    = 0.0f;
   error_            = PIPELINE_OK;
   time_             = base::TimeDelta();
-  ticks_at_last_set_time_ = base::TimeTicks::Now();
   rendered_mime_types_.clear();
 }
 
 bool PipelineImpl::IsPipelineOk() const {
   return pipeline_thread_ && initialized_ && PIPELINE_OK == error_;
 }
 
 bool PipelineImpl::IsPipelineThread() const {
   return pipeline_thread_ &&
       PlatformThread::CurrentId() == pipeline_thread_->thread_id();
@@ skipping 18 matching lines @@
   AutoLock auto_lock(lock_);
   buffered_bytes_ = buffered_bytes;
 }
 
 void PipelineImpl::SetVideoSize(size_t width, size_t height) {
   AutoLock auto_lock(lock_);
   video_width_ = width;
   video_height_ = height;
 }
 
+void PipelineImpl::SetTime(base::TimeDelta time) {
+  AutoLock auto_lock(lock_);
+  time_ = time;
+}
+
+void PipelineImpl::InternalSetPlaybackRate(float rate) {
+  AutoLock auto_lock(lock_);
+  playback_rate_ = rate;
+}
+
+bool PipelineImpl::InternalSetError(PipelineError error) {
+  // Don't want callers to set an error of "OK".  STOPPING is a special value
+  // that should only be used internally by the StopTask() method.
+  DCHECK(PIPELINE_OK != error && PIPELINE_STOPPING != error);
+  AutoLock auto_lock(lock_);
+  bool changed_error = false;
+  if (PIPELINE_OK == error_) {
+    error_ = error;
+    changed_error = true;
+  }
+  return changed_error;
+}
+
 void PipelineImpl::InsertRenderedMimeType(const std::string& major_mime_type) {
   AutoLock auto_lock(lock_);
   rendered_mime_types_.insert(major_mime_type);
 }
 
 
 //-----------------------------------------------------------------------------
 
 PipelineThread::PipelineThread(PipelineImpl* pipeline)
     : pipeline_(pipeline),
       thread_("PipelineThread"),
-      time_update_callback_scheduled_(false),
       state_(kCreated) {
 }
 
 PipelineThread::~PipelineThread() {
   Stop();
   DCHECK(state_ == kStopped || state_ == kError);
 }
 
 // This method is called on the client's thread.  It starts the pipeline's
 // dedicated thread and posts a task to call the StartTask() method on that
 // thread.
 bool PipelineThread::Start(FilterFactory* filter_factory,
                            const std::string& url,
                            PipelineCallback* init_complete_callback) {
   DCHECK_EQ(kCreated, state_);
   if (thread_.Start()) {
     filter_factory_ = filter_factory;
     url_ = url;
     init_callback_.reset(init_complete_callback);
-    PostTask(NewRunnableMethod(this, &PipelineThread::StartTask));
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &PipelineThread::StartTask));
     return true;
   }
   return false;
 }
 
 // Called on the client's thread.  If the thread has been started, then posts
 // a task to call the StopTask() method, then waits until the thread has
 // stopped.
 void PipelineThread::Stop() {
   if (thread_.IsRunning()) {
-    PostTask(NewRunnableMethod(this, &PipelineThread::StopTask));
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &PipelineThread::StopTask));
     thread_.Stop();
   }
   DCHECK(filter_hosts_.empty());
   DCHECK(filter_threads_.empty());
 }
 
 // Called on client's thread.
-void PipelineThread::SetPlaybackRate(float rate) {
-  PostTask(NewRunnableMethod(this, &PipelineThread::SetPlaybackRateTask, rate));
-}
-
-// Called on client's thread.
 void PipelineThread::Seek(base::TimeDelta time,
                           PipelineCallback* seek_callback) {
-  PostTask(NewRunnableMethod(this, &PipelineThread::SeekTask, time,
-                             seek_callback));
+  message_loop()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &PipelineThread::SeekTask, time, seek_callback));
+}
+
+// Called on client's thread.
+void PipelineThread::SetPlaybackRate(float rate) {
+  message_loop()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &PipelineThread::SetPlaybackRateTask, rate));
 }
 
 // Called on client's thread.
 void PipelineThread::SetVolume(float volume) {
-  PostTask(NewRunnableMethod(this, &PipelineThread::SetVolumeTask, volume));
+  message_loop()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &PipelineThread::SetVolumeTask, volume));
 }
 
+// Called from any thread.
 void PipelineThread::InitializationComplete(FilterHostImpl* host) {
   if (IsPipelineOk()) {
     // Continue the start task by proceeding to the next stage.
-    PostTask(NewRunnableMethod(this, &PipelineThread::StartTask));
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &PipelineThread::StartTask));
   }
 }
 
-// Called from any thread.  Updates the pipeline time and schedules a task to
-// call back to filters that have registered a callback for time updates.
+// Called from any thread.  Updates the pipeline time.
 void PipelineThread::SetTime(base::TimeDelta time) {
   pipeline()->SetTime(time);
-  if (!time_update_callback_scheduled_) {
-    time_update_callback_scheduled_ = true;
-    PostTask(NewRunnableMethod(this, &PipelineThread::SetTimeTask));
-  }
 }
 
 // Called from any thread.  Sets the pipeline |error_| member and schedules a
 // task to stop all the filters in the pipeline.  Note that the thread will
 // continue to run until the client calls Pipeline::Stop(), but nothing will
 // be processed since filters will not be able to post tasks.
 void PipelineThread::Error(PipelineError error) {
   // If this method returns false, then an error has already happened, so no
   // reason to run the StopTask again.  It's going to happen.
   if (pipeline()->InternalSetError(error)) {
-    PostTask(NewRunnableMethod(this, &PipelineThread::StopTask));
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &PipelineThread::StopTask));
   }
 }
 
-// This is a helper method to post task on message_loop(). This method is only
-// called from this class or from Pipeline.
-void PipelineThread::PostTask(Task* task) {
-  message_loop()->PostTask(FROM_HERE, task);
+// Called as a result of destruction of the thread.
+//
+// TODO(scherkus): this can block the client due to synchronous Stop() API call.
+void PipelineThread::WillDestroyCurrentMessageLoop() {
+  STLDeleteElements(&filter_hosts_);
+  STLDeleteElements(&filter_threads_);
 }
 
-
 // Main initialization method called on the pipeline thread.  This code attempts
 // to use the specified filter factory to build a pipeline.
 // Initialization step performed in this method depends on current state of this
 // object, indicated by |state_|.  After each step of initialization, this
 // object transits to the next stage.  It starts by creating a DataSource,
 // connects it to a Demuxer, and then connects the Demuxer's audio stream to an
 // AudioDecoder which is then connected to an AudioRenderer.  If the media has
 // video, then it connects a VideoDecoder to the Demuxer's video stream, and
 // then connects the VideoDecoder to a VideoRenderer.
 //
@@ skipping 204 matching lines @@
   DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
 
   pipeline_->volume_ = volume;
   scoped_refptr<AudioRenderer> audio_renderer;
   GetFilter(&audio_renderer);
   if (audio_renderer) {
     audio_renderer->SetVolume(volume);
   }
 }
 
-void PipelineThread::SetTimeTask() {
-  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
-
-  time_update_callback_scheduled_ = false;
-  for (FilterHostVector::iterator iter = filter_hosts_.begin();
-       iter != filter_hosts_.end();
-       ++iter) {
-    (*iter)->RunTimeUpdateCallback(pipeline_->time_);
-  }
-}
-
-template <class Filter>
-void PipelineThread::GetFilter(scoped_refptr<Filter>* filter_out) const {
-  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
-
-  *filter_out = NULL;
-  for (FilterHostVector::const_iterator iter = filter_hosts_.begin();
-       iter != filter_hosts_.end() && NULL == *filter_out;
-       iter++) {
-    (*iter)->GetFilter(filter_out);
-  }
-}
-
 template <class Filter, class Source>
 void PipelineThread::CreateFilter(FilterFactory* filter_factory,
                                   Source source,
                                   const MediaFormat& media_format) {
   DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
   DCHECK(IsPipelineOk());
 
   scoped_refptr<Filter> filter = filter_factory->Create<Filter>(media_format);
   if (!filter) {
     Error(PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
@@ skipping 77 matching lines @@
 
   if (decoder) {
     // If the decoder was created.
     const std::string major_mime_type = Decoder::major_mime_type();
     CreateFilter<Renderer, Decoder>(filter_factory_, decoder);
     return true;
   }
   return false;
 }
 
-// Called as a result of destruction of the thread.
-//
-// TODO(scherkus): this can block the client due to synchronous Stop() API call.
-void PipelineThread::WillDestroyCurrentMessageLoop() {
-  STLDeleteElements(&filter_hosts_);
-  STLDeleteElements(&filter_threads_);
+template <class Filter>
+void PipelineThread::GetFilter(scoped_refptr<Filter>* filter_out) const {
+  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
+
+  *filter_out = NULL;
+  for (FilterHostVector::const_iterator iter = filter_hosts_.begin();
+       iter != filter_hosts_.end() && NULL == *filter_out;
+       iter++) {
+    (*iter)->GetFilter(filter_out);
+  }
 }
 
 }  // namespace media