Could you guys get on this one ASAP. My video render code depends on these methods. I also slipped in the critical section code stuff here too.

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

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

5 files changed, 216 insertions, 20 deletions

diff --git a/media/base/filter_host.h b/media/base/filter_host.h
index 42f6554..d04d4d4 100644
--- a/media/base/filter_host.h
+++ b/media/base/filter_host.h
@@ -46,6 +46,14 @@ class FilterHost {
   //   base::TimeDelta - the new pipeline time, in microseconds.
   virtual void SetTimeUpdateCallback(Callback1<base::TimeDelta>::Type* cb) = 0;
 
+  // Request that the time callback be called at the specified stream
+  // time.  This will set a timer specific to the filter that will be fired
+  // no sooner than the specified time based on the interpolated time.  Note
+  // that, becuase the callback will be made with the interpolated time, it is
+  // possible for time to move "backward" slightly when the audio device updates
+  // the pipeline time though the SetTime method.
+  virtual void ScheduleTimeUpdateCallback(base::TimeDelta time) = 0;
+
   // Filters must call this method to indicate that their initialization is
   // complete.  They may call this from within their Initialize() method or may
   // choose call it after processing some data.
diff --git a/media/base/filter_host_impl.cc b/media/base/filter_host_impl.cc
index 5e7522e..74eaa16 100644
--- a/media/base/filter_host_impl.cc
+++ b/media/base/filter_host_impl.cc
@@ -1,4 +1,4 @@
-// Copyright (c) 2006-2009 The Chromium Authors. All rights reserved.
+// Copyright (c) 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.
 
@@ -12,8 +12,48 @@ void FilterHostImpl::SetTimeUpdateCallback(
 }
 
 void FilterHostImpl::RunTimeUpdateCallback(base::TimeDelta time) {
-  if (time_update_callback_.get())
+  if (time_update_callback_.get()) {
     time_update_callback_->Run(time);
+  }
+}
+
+// To understand why this method takes the |caller| parameter, see the comments
+// of the TimeUpdatedTask in the file filter_host_impl.h.
+void FilterHostImpl::RunScheduledTimeUpdateCallback(TimeUpdateTask* caller) {
+  time_update_lock_.Acquire();
+  if (caller == scheduled_time_update_task_) {
+    scheduled_time_update_task_ = NULL;
+  }
+  time_update_lock_.Release();
+  RunTimeUpdateCallback(pipeline()->GetInterpolatedTime());
+}
+
+void FilterHostImpl::ScheduleTimeUpdateCallback(base::TimeDelta time) {
+  time_update_lock_.Acquire();
+  if (stopped_) {
+    time_update_lock_.Release();
+  } else {
+    DCHECK(time_update_callback_.get());
+    if (scheduled_time_update_task_) {
+      scheduled_time_update_task_->Cancel();
+    }
+    scheduled_time_update_task_ = new TimeUpdateTask(this);
+    Task* task = scheduled_time_update_task_;
+    time_update_lock_.Release();
+
+    // Here, we compute the delta from now & adjust it by the playback rate.
+    time -= pipeline()->GetInterpolatedTime();
+    int delay = static_cast<int>(time.InMilliseconds());
+    float rate = pipeline()->GetPlaybackRate();
+    if (rate != 1.0f && rate != 0.0f) {
+      delay = static_cast<int>(delay / rate);
+    }
+    if (delay > 0) {
+      pipeline_thread_->message_loop()->PostDelayedTask(FROM_HERE, task, delay);
+    } else {
+      pipeline_thread_->message_loop()->PostTask(FROM_HERE, task);
+    }
+  }
 }
 
 void FilterHostImpl::InitializationComplete() {
@@ -38,19 +78,19 @@ void FilterHostImpl::SetTime(base::TimeDelta time) {
 }
 
 void FilterHostImpl::SetDuration(base::TimeDelta duration) {
-  pipeline()->duration_ = duration;
+  pipeline()->SetDuration(duration);
 }
 
 void FilterHostImpl::SetBufferedTime(base::TimeDelta buffered_time) {
-  pipeline()->buffered_time_ = buffered_time;
+  pipeline()->SetBufferedTime(buffered_time);
 }
 
 void FilterHostImpl::SetTotalBytes(int64 total_bytes) {
-  pipeline()->total_bytes_ = total_bytes;
+  pipeline()->SetTotalBytes(total_bytes);
 }
 
 void FilterHostImpl::SetBufferedBytes(int64 buffered_bytes) {
-  pipeline()->buffered_bytes_ = buffered_bytes;
+  pipeline()->SetBufferedBytes(buffered_bytes);
 }
 
 void FilterHostImpl::SetVideoSize(size_t width, size_t height) {
@@ -64,7 +104,11 @@ const PipelineStatus* FilterHostImpl::GetPipelineStatus() const {
 void FilterHostImpl::Stop() {
   if (!stopped_) {
     filter_->Stop();
+    AutoLock auto_lock(time_update_lock_);
     stopped_ = true;
+    if (scheduled_time_update_task_) {
+      scheduled_time_update_task_->Cancel();
+    }
   }
 }
 
diff --git a/media/base/filter_host_impl.h b/media/base/filter_host_impl.h
index bdbb5e1..df48592 100644
--- a/media/base/filter_host_impl.h
+++ b/media/base/filter_host_impl.h
@@ -18,6 +18,7 @@ class FilterHostImpl : public FilterHost {
   // FilterHost interface.
   virtual const PipelineStatus* GetPipelineStatus() const;
   virtual void SetTimeUpdateCallback(Callback1<base::TimeDelta>::Type* cb);
+  virtual void ScheduleTimeUpdateCallback(base::TimeDelta time);
   virtual void InitializationComplete();
   virtual void PostTask(Task* task);
   virtual void Error(PipelineError error);
@@ -40,6 +41,7 @@ class FilterHostImpl : public FilterHost {
       : pipeline_thread_(pipeline_thread),
         filter_type_(Filter::filter_type()),
         filter_(filter),
+        scheduled_time_update_task_(NULL),
         stopped_(false) {
   }
   ~FilterHostImpl() {}
@@ -66,6 +68,50 @@ class FilterHostImpl : public FilterHost {
   MediaFilter* media_filter() const { return filter_; }
 
  private:
+  // This task class is used to schedule a time update callback for the filter.
+  // Because a filter may call the ScheduleTimeUpdateCallback method from any
+  // thread, and becuase we only want to honor the last call to that method,
+  // we always have only one current task.
+  // We are required to keep a pointer to the host and a boolean that tells
+  // us if the task was canceled because the cancelation could happen on one
+  // thread, just as the pipeline thread is calling the Run method on this task.
+  // So, we can't just NULL out the host_ to cancel this because it could
+  // fault.  Once we have called the host, it needs to enter it's critical
+  // section and make sure that the task that has Run is, in fact, the last one
+  // that was scheduled.
+  // In the case where the filter host is Stopping (or being destroyed), it will
+  // be guaranteed to happen on the pipeline thread, thus making the setting
+  // of the |canceled_| bool thread safe since the task would only execute on
+  // the pipeline thread.  This means that it could be possible for a task to
+  // hold a pointer to a |host_| that has been deleted, but it will never access
+  // that pointer because the task was canceled.
+  class TimeUpdateTask : public CancelableTask {
+   public:
+    explicit TimeUpdateTask(FilterHostImpl* host)
+        : host_(host),
+          canceled_(false) {}
+
+    virtual void Run() {
+      if (!canceled_) {
+        host_->RunScheduledTimeUpdateCallback(this);
+      }
+    }
+
+    virtual void Cancel() {
+      canceled_ = true;
+    }
+
+   private:
+    FilterHostImpl* const host_;
+    bool canceled_;
+
+    DISALLOW_COPY_AND_ASSIGN(TimeUpdateTask);
+  };
+
+  // Method used by the TimeUpdateTask to call back to the filter.
+  void RunScheduledTimeUpdateCallback(TimeUpdateTask* caller);
+
+  // Useful method for getting the pipeline.
   PipelineImpl* pipeline() const { return pipeline_thread_->pipeline(); }
 
   // PipelineThread that owns this FilterHostImpl.
@@ -80,6 +126,12 @@ class FilterHostImpl : public FilterHost {
   // An optional callback that will be called when the time is updated.
   scoped_ptr<Callback1<base::TimeDelta>::Type> time_update_callback_;
 
+  // Critical section used for scheduled time update callbacks.
+  Lock time_update_lock_;
+
+  // Pointer to the current time update callback task.
+  TimeUpdateTask* scheduled_time_update_task_;
+
   // Used to avoid calling Filter's Stop method multiplie times.  It is also
   // used to prevent a filter that has been stopped from calling PostTask.
   bool stopped_;
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index 4f11d62..5792972 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -18,49 +18,100 @@ PipelineImpl::~PipelineImpl() {
 }
 
 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&>(video_size_access_lock_));
+  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::InternalSetError(PipelineError error) {
+  AutoLock auto_lock(lock_);
+  bool changed_error = false;
+  DCHECK(PIPELINE_OK != error);
+  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,
@@ -119,6 +170,7 @@ void PipelineImpl::SetVolume(float volume) {
 }
 
 void PipelineImpl::ResetState() {
+  AutoLock auto_lock(lock_);
   pipeline_thread_  = NULL;
   initialized_      = false;
   duration_         = base::TimeDelta();
@@ -129,12 +181,33 @@ void PipelineImpl::ResetState() {
   video_height_     = 0;
   volume_           = 0.0f;
   playback_rate_    = 0.0f;
-  time_             = base::TimeDelta();
   error_            = PIPELINE_OK;
+  time_             = base::TimeDelta();
+  ticks_at_last_set_time_ = base::TimeTicks::Now();
+}
+
+void PipelineImpl::SetDuration(base::TimeDelta duration) {
+  AutoLock auto_lock(lock_);
+  duration_ = duration;
+}
+
+void PipelineImpl::SetBufferedTime(base::TimeDelta buffered_time) {
+  AutoLock auto_lock(lock_);
+  buffered_time_ = buffered_time;
+}
+
+void PipelineImpl::SetTotalBytes(int64 total_bytes) {
+  AutoLock auto_lock(lock_);
+  total_bytes_ = total_bytes;
+}
+
+void PipelineImpl::SetBufferedBytes(int64 buffered_bytes) {
+  AutoLock auto_lock(lock_);
+  buffered_bytes_ = buffered_bytes;
 }
 
 void PipelineImpl::SetVideoSize(size_t width, size_t height) {
-  AutoLock auto_lock(video_size_access_lock_);
+  AutoLock auto_lock(lock_);
   width = width;
   height = height;
 }
@@ -221,7 +294,7 @@ void PipelineThread::InitializationComplete(FilterHostImpl* host) {
 // 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.
 void PipelineThread::SetTime(base::TimeDelta time) {
-  pipeline()->time_ = time;
+  pipeline()->SetTime(time);
   if (!time_update_callback_scheduled_) {
     time_update_callback_scheduled_ = true;
     PostTask(NewRunnableMethod(this, &PipelineThread::SetTimeTask));
@@ -233,9 +306,9 @@ void PipelineThread::SetTime(base::TimeDelta time) {
 // 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) {
-  DCHECK(PIPELINE_OK != error);
-  if (PIPELINE_OK == pipeline()->error_) {
-    pipeline()->error_ = 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));
   }
 }
@@ -296,7 +369,7 @@ void PipelineThread::StartTask(FilterFactory* filter_factory,
   }
   if (success) {
     pipeline_->initialized_ = true;
-  } else if (PIPELINE_OK == pipeline_->error_) {
+  } else {
     Error(PIPELINE_ERROR_INITIALIZATION_FAILED);
   }
 
@@ -342,7 +415,7 @@ void PipelineThread::InitializationCompleteTask(FilterHostImpl* host) {
 }
 
 void PipelineThread::SetPlaybackRateTask(float rate) {
-  pipeline_->playback_rate_ = rate;
+  pipeline_->InternalSetPlaybackRate(rate);
   FilterHostVector::iterator iter = filter_hosts_.begin();
   while (iter != filter_hosts_.end()) {
     (*iter)->media_filter()->SetPlaybackRate(rate);
@@ -434,7 +507,7 @@ bool PipelineThread::CreateFilter(FilterFactory* filter_factory,
 
   // If this method fails, but no error set, then indicate a general
   // initialization failure.
-  if (PIPELINE_OK == pipeline_->error_ && (!success)) {
+  if (!success) {
     Error(PIPELINE_ERROR_INITIALIZATION_FAILED);
   }
   return success;
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index bf0a756d..362b54e 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -41,6 +41,7 @@ class PipelineImpl : public Pipeline {
   virtual float GetVolume() const;
   virtual float GetPlaybackRate() const;
   virtual base::TimeDelta GetTime() const;
+  virtual base::TimeDelta GetInterpolatedTime() const;
   virtual PipelineError GetError() const;
 
   // Impementation of Pipeline methods.
@@ -67,8 +68,19 @@ class PipelineImpl : public Pipeline {
     return (pipeline_thread_ && initialized_ && PIPELINE_OK == error_);
   }
 
-  // Called directly by the FilterHostImpl object to set the video size.
+  // Methods called by FilterHostImpl to update pipeline state.
+  void SetDuration(base::TimeDelta duration);
+  void SetBufferedTime(base::TimeDelta buffered_time);
+  void SetTotalBytes(int64 total_bytes);
+  void SetBufferedBytes(int64 buffered_bytes);
   void SetVideoSize(size_t width, size_t height);
+  void SetTime(base::TimeDelta time);
+  void InternalSetPlaybackRate(float rate);
+
+  // Sets the error to the new error code only if the current error state is
+  // PIPELINE_OK. Returns true if error set, otherwise leaves current error
+  // alone, and returns false.
+  bool InternalSetError(PipelineError error);
 
   // Holds a ref counted reference to the PipelineThread object associated
   // with this pipeline.  Prior to the call to the Start method, this member
@@ -95,10 +107,12 @@ class PipelineImpl : public Pipeline {
   // of a filter.
   int64 total_bytes_;
 
+  // Lock used to serialize access for getter/setter methods.
+  Lock lock_;
+
   // Video width and height.  Set by a FilterHostImpl object on behalf
   // of a filter.  The video_size_access_lock_ is used to make sure access
   // to the pair of width and height are modified or read in thread safe way.
-  Lock video_size_access_lock_;
   size_t video_width_;
   size_t video_height_;
 
@@ -119,6 +133,10 @@ class PipelineImpl : public Pipeline {
   // audio renderer filter.
   base::TimeDelta time_;
 
+  // Internal system timer at last time the SetTime method was called.  Used to
+  // compute interpolated time.
+  base::TimeTicks ticks_at_last_set_time_;
+
   // Status of the pipeline.  Initialized to PIPELINE_OK which indicates that
   // the pipeline is operating correctly. Any other value indicates that the
   // pipeline is stopped or is stopping.  Clients can call the Stop method to
@@ -172,6 +190,9 @@ class PipelineThread : public base::RefCountedThreadSafe<PipelineThread>,
   // pipeline object.
   PipelineImpl* pipeline() const { return pipeline_; }
 
+  // Accessor used to post messages to thread's message loop.
+  MessageLoop* message_loop() const { return thread_.message_loop(); }
+
  private:
   // Implementation of MessageLoop::DestructionObserver.  StartTask registers
   // this class as a destruction observer on the thread's message loop.
@@ -183,8 +204,6 @@ class PipelineThread : public base::RefCountedThreadSafe<PipelineThread>,
   friend class base::RefCountedThreadSafe<PipelineThread>;
   virtual ~PipelineThread();
 
-  MessageLoop* message_loop() const { return thread_.message_loop(); }
-
   // The following "task" methods correspond to the public methods, but these
   // methods are run as the result of posting a task to the PipelineThread's
   // message loop.  For example, the Start method posts a task to call the