Move Pipeline permanent callbacks into Pipeline::Client interface.

media/base/pipeline_impl.cc

Index: media/base/pipeline_impl.cc
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index 8289f7366b55fadd4f545464de4e742a043f8142..ba7dc895883658fa61d6d5749c04e4f5fd8e5710 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -20,7 +20,9 @@
 #include "base/stl_util.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_util.h"
-#include "base/synchronization/condition_variable.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/thread_task_runner_handle.h"
+#include "media/base/bind_to_current_loop.h"
 #include "media/base/media_log.h"
 #include "media/base/media_switches.h"
 #include "media/base/renderer.h"
@@ -34,9 +36,10 @@ using base::TimeDelta;
 namespace media {
 
 PipelineImpl::PipelineImpl(
-    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& media_task_runner,
     MediaLog* media_log)
-    : task_runner_(task_runner),
+    : main_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      media_task_runner_(media_task_runner),
       media_log_(media_log),
       running_(false),
       did_loading_progress_(false),
@@ -55,28 +58,19 @@ PipelineImpl::PipelineImpl(
 }
 
 PipelineImpl::~PipelineImpl() {
-  DCHECK(thread_checker_.CalledOnValidThread())
+  DCHECK(main_task_runner_->BelongsToCurrentThread())
       << "Pipeline must be destroyed on same thread that created it";
   DCHECK(!running_) << "Stop() must complete before destroying object";
-  DCHECK(stop_cb_.is_null());
   DCHECK(seek_cb_.is_null());
 }
 
 void PipelineImpl::Start(Demuxer* demuxer,
                          std::unique_ptr<Renderer> renderer,
-                         const base::Closure& ended_cb,
-                         const PipelineStatusCB& error_cb,
-                         const PipelineStatusCB& seek_cb,
-                         const PipelineMetadataCB& metadata_cb,
-                         const BufferingStateCB& buffering_state_cb,
-                         const base::Closure& duration_change_cb,
-                         const AddTextTrackCB& add_text_track_cb,
-                         const base::Closure& waiting_for_decryption_key_cb) {
-  DCHECK(!ended_cb.is_null());
-  DCHECK(!error_cb.is_null());
+                         Client* client,
+                         const PipelineStatusCB& seek_cb) {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  DCHECK(client);
   DCHECK(!seek_cb.is_null());
-  DCHECK(!metadata_cb.is_null());
-  DCHECK(!buffering_state_cb.is_null());
 
   base::AutoLock auto_lock(lock_);
   CHECK(!running_) << "Media pipeline is already running";
@@ -84,91 +78,123 @@ void PipelineImpl::Start(Demuxer* demuxer,
 
   demuxer_ = demuxer;
   renderer_ = std::move(renderer);
-  ended_cb_ = ended_cb;
-  error_cb_ = error_cb;
-  seek_cb_ = seek_cb;
-  metadata_cb_ = metadata_cb;
-  buffering_state_cb_ = buffering_state_cb;
-  duration_change_cb_ = duration_change_cb;
-  add_text_track_cb_ = add_text_track_cb;
-  waiting_for_decryption_key_cb_ = waiting_for_decryption_key_cb;
-
-  task_runner_->PostTask(FROM_HERE,
-                         base::Bind(&PipelineImpl::StartTask, weak_this_));
+  client_weak_factory_.reset(new base::WeakPtrFactory<Client>(client));

[xhwang, 2016/05/04 06:32:22]

wez: This is a bit new to me and it seems working well. But the documentation
only recommended "primitive types" for this kind of use. Do you see any issues?
If not, shall we update the documentation? 

https://code.google.com/p/chromium/codesearch#chromium/src/base/memory/weak_p...

Ulitmately I think we want to split this class into two different classes, one
lives on the main thread and one lives on the media thread, to avoid the
complexity here.

[Wez, 2016/05/04 15:56:26]

Agreed re there being too much complexity due to having the object spanning
threads!

This is a perfectly valid use of WeakPtrFactory(). However, I'd strongly
recommend that you take a weak_client_ WeakPtr here, rather than calling
GetWeakPtr() at each Bind(), for both thread-safety and efficiency.

[alokp, 2016/05/04 17:40:01]

Agreed about thread-safety, so created a weak_client_. For my knowledge, can you
explain why calling GetWeakPtr is less efficient than the copy constructor for
WeakPtr? On the surface they look equivalent.

I already have a TODO in Stop for splitting the class along thread lines.

+  seek_cb_ = media::BindToCurrentLoop(seek_cb);
+  media_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&PipelineImpl::StartTask, weak_this_));
 }
 
-void PipelineImpl::Stop(const base::Closure& stop_cb) {
+void PipelineImpl::Stop() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
   DVLOG(2) << __FUNCTION__;
-  task_runner_->PostTask(
-      FROM_HERE, base::Bind(&PipelineImpl::StopTask, weak_this_, stop_cb));
+
+  if (media_task_runner_ != main_task_runner_) {
+    // This path is executed by production code where the two task runners -
+    // main and media - live on different threads.
+    // TODO(alokp): It may be possible to not have to wait for StopTask by
+    // moving the members accessed on media thread into a class/struct and
+    // DeleteSoon the instance on the media thread.
+    base::WaitableEvent waiter(false, false);
+    base::Closure stop_cb =
+        base::Bind(&base::WaitableEvent::Signal, base::Unretained(&waiter));
+    CHECK(media_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&PipelineImpl::StopTask, weak_this_, stop_cb)));

[Wez, 2016/05/04 15:56:26]

What is your intent in CHECKing the result of PostTask() here?

[alokp, 2016/05/04 17:40:01]

Clarified in comment.

+    waiter.Wait();
+  } else {
+    // This path is executed by unittests that share media and main threads.
+    StopTask(base::Bind(&base::DoNothing));
+  }
+  // Invalidate client weak pointer effectively canceling all pending client
+  // notifications in the message queue.
+  client_weak_factory_.reset();
 }
 
 void PipelineImpl::Seek(TimeDelta time, const PipelineStatusCB& seek_cb) {
-  base::AutoLock auto_lock(lock_);
-  if (!running_) {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
+  if (!IsRunning()) {
     DLOG(ERROR) << "Media pipeline isn't running. Ignoring Seek().";
     return;
   }
 
-  task_runner_->PostTask(FROM_HERE, base::Bind(&PipelineImpl::SeekTask,
-                                               weak_this_, time, seek_cb));
+  media_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&PipelineImpl::SeekTask, weak_this_, time,
+                            media::BindToCurrentLoop(seek_cb)));
 }
 
 bool PipelineImpl::IsRunning() const {
+  // TODO(alokp): Add thread DCHECK after removing the internal usage on
+  // media thread.
   base::AutoLock auto_lock(lock_);
   return running_;
 }
 
 double PipelineImpl::GetPlaybackRate() const {
+  // TODO(alokp): Add thread DCHECK after removing the internal usage on
+  // media thread.
   base::AutoLock auto_lock(lock_);
   return playback_rate_;
 }
 
 void PipelineImpl::SetPlaybackRate(double playback_rate) {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
   if (playback_rate < 0.0)
     return;
 
   base::AutoLock auto_lock(lock_);
   playback_rate_ = playback_rate;
   if (running_) {
-    task_runner_->PostTask(FROM_HERE,
-                           base::Bind(&PipelineImpl::PlaybackRateChangedTask,
-                                      weak_this_, playback_rate));
+    media_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&PipelineImpl::PlaybackRateChangedTask,
+                              weak_this_, playback_rate));
   }
 }
 
 void PipelineImpl::Suspend(const PipelineStatusCB& suspend_cb) {
-  task_runner_->PostTask(FROM_HERE, base::Bind(&PipelineImpl::SuspendTask,
-                                               weak_this_, suspend_cb));
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
+  media_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&PipelineImpl::SuspendTask, weak_this_,
+                            media::BindToCurrentLoop(suspend_cb)));
 }
 
 void PipelineImpl::Resume(std::unique_ptr<Renderer> renderer,
                           base::TimeDelta timestamp,
                           const PipelineStatusCB& seek_cb) {
-  task_runner_->PostTask(
-      FROM_HERE, base::Bind(&PipelineImpl::ResumeTask, weak_this_,
-                            base::Passed(&renderer), timestamp, seek_cb));
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
+  media_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&PipelineImpl::ResumeTask, weak_this_, base::Passed(&renderer),
+                 timestamp, media::BindToCurrentLoop(seek_cb)));
 }
 
 float PipelineImpl::GetVolume() const {
+  // TODO(alokp): Add thread DCHECK after removing the internal usage on
+  // media thread.
   base::AutoLock auto_lock(lock_);
   return volume_;
 }
 
 void PipelineImpl::SetVolume(float volume) {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
   if (volume < 0.0f || volume > 1.0f)
     return;
 
   base::AutoLock auto_lock(lock_);
   volume_ = volume;
   if (running_) {
-    task_runner_->PostTask(
+    media_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&PipelineImpl::VolumeChangedTask, weak_this_, volume));
   }
 }
 
 TimeDelta PipelineImpl::GetMediaTime() const {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
   base::AutoLock auto_lock(lock_);
   if (suspend_timestamp_ != kNoTimestamp())
     return suspend_timestamp_;
@@ -177,16 +203,22 @@ TimeDelta PipelineImpl::GetMediaTime() const {
 }
 
 Ranges<TimeDelta> PipelineImpl::GetBufferedTimeRanges() const {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
   base::AutoLock auto_lock(lock_);
   return buffered_time_ranges_;
 }
 
 TimeDelta PipelineImpl::GetMediaDuration() const {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
   base::AutoLock auto_lock(lock_);
   return duration_;
 }
 
 bool PipelineImpl::DidLoadingProgress() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
   base::AutoLock auto_lock(lock_);
   bool ret = did_loading_progress_;
   did_loading_progress_ = false;
@@ -194,13 +226,17 @@ bool PipelineImpl::DidLoadingProgress() {
 }
 
 PipelineStatistics PipelineImpl::GetStatistics() const {
+  // TODO(alokp): Add thread DCHECK after removing the internal usage on
+  // media thread.
   base::AutoLock auto_lock(lock_);
   return statistics_;
 }
 
 void PipelineImpl::SetCdm(CdmContext* cdm_context,
                           const CdmAttachedCB& cdm_attached_cb) {
-  task_runner_->PostTask(
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
+  media_task_runner_->PostTask(
       FROM_HERE, base::Bind(&PipelineImpl::SetCdmTask, weak_this_, cdm_context,
                             cdm_attached_cb));
 }
@@ -210,11 +246,12 @@ void PipelineImpl::SetErrorForTesting(PipelineStatus status) {
 }
 
 bool PipelineImpl::HasWeakPtrsForTesting() const {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   return weak_factory_.HasWeakPtrs();
 }
 
 void PipelineImpl::SetState(State next_state) {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DVLOG(1) << GetStateString(state_) << " -> " << GetStateString(next_state);
 
   state_ = next_state;
@@ -245,7 +282,7 @@ const char* PipelineImpl::GetStateString(State state) {
 #undef RETURN_STRING
 
 PipelineImpl::State PipelineImpl::GetNextState() const {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(stop_cb_.is_null()) << "State transitions don't happen when stopping";
   DCHECK_EQ(status_, PIPELINE_OK)
       << "State transitions don't happen when there's an error: " << status_;
@@ -280,34 +317,44 @@ PipelineImpl::State PipelineImpl::GetNextState() const {
 }
 
 void PipelineImpl::OnDemuxerError(PipelineStatus error) {
-  task_runner_->PostTask(FROM_HERE, base::Bind(&PipelineImpl::ErrorChangedTask,
-                                               weak_this_, error));
+  // TODO(alokp): Add thread DCHECK after ensuring that all Demuxer
+  // implementations call DemuxerHost on the media thread.
+  media_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&PipelineImpl::ErrorChangedTask, weak_this_, error));
 }
 
 void PipelineImpl::AddTextStream(DemuxerStream* text_stream,
                                  const TextTrackConfig& config) {
-  task_runner_->PostTask(
+  // TODO(alokp): Add thread DCHECK after ensuring that all Demuxer
+  // implementations call DemuxerHost on the media thread.
+  media_task_runner_->PostTask(
       FROM_HERE, base::Bind(&PipelineImpl::AddTextStreamTask, weak_this_,
                             text_stream, config));
 }
 
 void PipelineImpl::RemoveTextStream(DemuxerStream* text_stream) {
-  task_runner_->PostTask(
+  // TODO(alokp): Add thread DCHECK after ensuring that all Demuxer
+  // implementations call DemuxerHost on the media thread.
+  media_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&PipelineImpl::RemoveTextStreamTask, weak_this_, text_stream));
 }
 
 void PipelineImpl::OnError(PipelineStatus error) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(IsRunning());
   DCHECK_NE(PIPELINE_OK, error);
   VLOG(1) << "Media pipeline error: " << error;
 
-  task_runner_->PostTask(FROM_HERE, base::Bind(&PipelineImpl::ErrorChangedTask,
-                                               weak_this_, error));
+  media_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&PipelineImpl::ErrorChangedTask, weak_this_, error));
 }
 
 void PipelineImpl::SetDuration(TimeDelta duration) {
+  // TODO(alokp): Add thread DCHECK after ensuring that all Demuxer
+  // implementations call DemuxerHost on the media thread.
   DCHECK(IsRunning());
   media_log_->AddEvent(media_log_->CreateTimeEvent(MediaLogEvent::DURATION_SET,
                                                    "duration", duration));
@@ -315,23 +362,21 @@ void PipelineImpl::SetDuration(TimeDelta duration) {
 
   base::AutoLock auto_lock(lock_);
   duration_ = duration;
-  if (!duration_change_cb_.is_null())
-    duration_change_cb_.Run();
+  main_task_runner_->PostTask(FROM_HERE,
+                              base::Bind(&Pipeline::Client::OnDurationChange,
+                                         client_weak_factory_->GetWeakPtr()));
 }
 
 void PipelineImpl::StateTransitionTask(PipelineStatus status) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // No-op any state transitions if we're stopping.
   if (state_ == kStopping || state_ == kStopped)
     return;
 
-  // Preserve existing abnormal status, otherwise update based on the result of
-  // the previous operation.
-  status_ = (status_ != PIPELINE_OK ? status_ : status);
-
-  if (status_ != PIPELINE_OK) {
-    ErrorChangedTask(status_);
+  // Report error from the previous operation.
+  if (status != PIPELINE_OK) {
+    ErrorChangedTask(status);
     return;
   }
 
@@ -407,7 +452,7 @@ void PipelineImpl::StateTransitionTask(PipelineStatus status) {
 // running on the media thread would result in crashes.
 void PipelineImpl::DoSeek(TimeDelta seek_timestamp,
                           const PipelineStatusCB& done_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(!pending_callbacks_.get());
   DCHECK_EQ(state_, kSeeking);
   SerialRunner::Queue bound_fns;
@@ -435,9 +480,10 @@ void PipelineImpl::DoSeek(TimeDelta seek_timestamp,
   pending_callbacks_ = SerialRunner::Run(bound_fns, done_cb);
 }
 
-void PipelineImpl::DoStop(const PipelineStatusCB& done_cb) {
+void PipelineImpl::DoStop() {
   DVLOG(2) << __FUNCTION__;
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+  DCHECK_EQ(state_, kStopping);
   DCHECK(!pending_callbacks_.get());
 
   // TODO(scherkus): Enforce that Renderer is only called on a single thread,
@@ -455,37 +501,19 @@ void PipelineImpl::DoStop(const PipelineStatusCB& done_cb) {
     demuxer_ = NULL;
   }
 
-  task_runner_->PostTask(FROM_HERE, base::Bind(done_cb, PIPELINE_OK));
-}
-
-void PipelineImpl::OnStopCompleted(PipelineStatus status) {
-  DVLOG(2) << __FUNCTION__;
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  DCHECK_EQ(state_, kStopping);
-  DCHECK(!renderer_);
-  DCHECK(!text_renderer_);
-
   {
     base::AutoLock auto_lock(lock_);
     running_ = false;
   }
-
   SetState(kStopped);
-  demuxer_ = NULL;
 
   // If we stop during initialization/seeking/suspending we don't want to leave
-  // outstanding callbacks around.
-  if (!seek_cb_.is_null()) {
-    base::ResetAndReturn(&seek_cb_).Run(status_);
-    error_cb_.Reset();
-  }
-  if (!suspend_cb_.is_null()) {
-    base::ResetAndReturn(&suspend_cb_).Run(status_);
-    error_cb_.Reset();
-  }
-  if (!stop_cb_.is_null()) {
-    error_cb_.Reset();
+  // outstanding callbacks around. The callbacks also do not get run if the
+  // pipeline is stopped before it had a chance to complete outstanding tasks.
+  seek_cb_.Reset();
+  suspend_cb_.Reset();
 
+  if (!stop_cb_.is_null()) {
     // Invalid all weak pointers so it's safe to destroy |this| on the render
     // main thread.
     weak_factory_.InvalidateWeakPtrs();
@@ -496,14 +524,12 @@ void PipelineImpl::OnStopCompleted(PipelineStatus status) {
     // executing |stop_cb_|.
     return;
   }
-  if (!error_cb_.is_null()) {
-    DCHECK_NE(status_, PIPELINE_OK);
-    base::ResetAndReturn(&error_cb_).Run(status_);
-  }
 }
 
 void PipelineImpl::OnBufferedTimeRangesChanged(
     const Ranges<base::TimeDelta>& ranges) {
+  // TODO(alokp): Add thread DCHECK after ensuring that all Demuxer
+  // implementations call DemuxerHost on the media thread.
   base::AutoLock auto_lock(lock_);
   buffered_time_ranges_ = ranges;
   did_loading_progress_ = true;
@@ -520,8 +546,16 @@ void PipelineImpl::OnUpdateStatistics(const PipelineStatistics& stats_delta) {
   statistics_.video_memory_usage += stats_delta.video_memory_usage;
 }
 
+void PipelineImpl::OnWaitingForDecryptionKey() {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+
+  main_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&Pipeline::Client::OnWaitingForDecryptionKey,
+                            client_weak_factory_->GetWeakPtr()));

[sandersd (OOO until July 31), 2016/05/03 21:51:43]

It is not safe to vend weak pointers from the same factory on multiple threads.
(In fact the key reason that we invalidate in Stop() is to be able to vend and
deref on separate threads, but we also never call GetWeakPtr on the media
thread.)

[alokp, 2016/05/03 23:00:35]

This factory is for the Client, not PipelineImpl. Vending happens exclusively on
the media thread. Deref/Invalidation happens on the main thread.

[sandersd (OOO until July 31), 2016/05/03 23:03:39]

SetDuration() is not on the media thread.

[Wez, 2016/05/04 15:56:26]

Looks like SetDuration() should be, but isn't currently in all impls, so yes,
that is potentially dangerous. Even if it weren't, it's more obviously correct,
and slightly more efficient, to create a WeakPtr once and Bind() it into
multiple calls than to keep creating new ones.

[alokp, 2016/05/04 17:40:01]

Done.

+}
+
 void PipelineImpl::StartTask() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   CHECK_EQ(kCreated, state_)
       << "Media pipeline cannot be started more than once";
@@ -536,7 +570,7 @@ void PipelineImpl::StartTask() {
 }
 
 void PipelineImpl::StopTask(const base::Closure& stop_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(stop_cb_.is_null());
 
   if (state_ == kStopped) {
@@ -569,33 +603,54 @@ void PipelineImpl::StopTask(const base::Closure& stop_cb) {
 
   SetState(kStopping);
   pending_callbacks_.reset();
-  DoStop(base::Bind(&PipelineImpl::OnStopCompleted, weak_this_));
+  DoStop();
 }
 
 void PipelineImpl::ErrorChangedTask(PipelineStatus error) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK_NE(PIPELINE_OK, error) << "PIPELINE_OK isn't an error!";
 
+  // Preserve existing abnormal status.
+  if (status_ != PIPELINE_OK)
+    return;
+
   // Don't report pipeline error events to the media log here. The embedder will
   // log this when |error_cb_| is called. If the pipeline is already stopped or

[xhwang, 2016/05/04 06:32:22]

There's no |error_cb_| now. Please update this comment.

[alokp, 2016/05/04 17:40:01]

Done.

   // stopping we also don't want to log any event. In case we are suspending or
   // suspended, the error may be recoverable, so don't propagate it now, instead
   // let the subsequent seek during resume propagate it if it's unrecoverable.
-
   if (state_ == kStopping || state_ == kStopped || state_ == kSuspending ||
       state_ == kSuspended) {
     return;
   }
 
-  SetState(kStopping);
-  pending_callbacks_.reset();
+  // Once we enter |kStopping| state, nothing is reported back to the client.
+  // If we encounter an error during initialization/seeking/suspending,
+  // report the error using the completion callbacks for those tasks.
   status_ = error;
+  bool error_reported = false;
+  if (!seek_cb_.is_null()) {
+    base::ResetAndReturn(&seek_cb_).Run(status_);
+    error_reported = true;
+  }
+  if (!suspend_cb_.is_null()) {
+    base::ResetAndReturn(&suspend_cb_).Run(status_);
+    error_reported = true;
+  }
+  if (!error_reported) {
+    DCHECK_NE(status_, PIPELINE_OK);
+    main_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&Pipeline::Client::OnError,
+                              client_weak_factory_->GetWeakPtr(), status_));
+  }
 
-  DoStop(base::Bind(&PipelineImpl::OnStopCompleted, weak_this_));
+  SetState(kStopping);
+  pending_callbacks_.reset();
+  DoStop();
 }
 
 void PipelineImpl::PlaybackRateChangedTask(double playback_rate) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // Playback rate changes are only carried out while playing.
   if (state_ != kPlaying)
@@ -605,7 +660,7 @@ void PipelineImpl::PlaybackRateChangedTask(double playback_rate) {
 }
 
 void PipelineImpl::VolumeChangedTask(float volume) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // Volume changes are only carried out while playing.
   if (state_ != kPlaying)
@@ -615,7 +670,7 @@ void PipelineImpl::VolumeChangedTask(float volume) {
 }
 
 void PipelineImpl::SeekTask(TimeDelta time, const PipelineStatusCB& seek_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(stop_cb_.is_null());
 
   // Suppress seeking if we're not fully started.
@@ -642,7 +697,7 @@ void PipelineImpl::SeekTask(TimeDelta time, const PipelineStatusCB& seek_cb) {
 }
 
 void PipelineImpl::SuspendTask(const PipelineStatusCB& suspend_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // Suppress suspending if we're not playing.
   if (state_ != kPlaying) {
@@ -690,7 +745,7 @@ void PipelineImpl::SuspendTask(const PipelineStatusCB& suspend_cb) {
 void PipelineImpl::ResumeTask(std::unique_ptr<Renderer> renderer,
                               base::TimeDelta timestamp,
                               const PipelineStatusCB& seek_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // Suppress resuming if we're not suspended.
   if (state_ != kSuspended) {
@@ -744,14 +799,14 @@ void PipelineImpl::SetCdmTask(CdmContext* cdm_context,
 void PipelineImpl::OnCdmAttached(const CdmAttachedCB& cdm_attached_cb,
                                  CdmContext* cdm_context,
                                  bool success) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   if (success)
     cdm_context_ = cdm_context;
   cdm_attached_cb.Run(success);
 }
 
 void PipelineImpl::OnRendererEnded() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   media_log_->AddEvent(media_log_->CreateEvent(MediaLogEvent::ENDED));
 
   if (state_ != kPlaying)
@@ -764,7 +819,7 @@ void PipelineImpl::OnRendererEnded() {
 }
 
 void PipelineImpl::OnTextRendererEnded() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   media_log_->AddEvent(media_log_->CreateEvent(MediaLogEvent::TEXT_ENDED));
 
   if (state_ != kPlaying)
@@ -777,7 +832,7 @@ void PipelineImpl::OnTextRendererEnded() {
 }
 
 void PipelineImpl::RunEndedCallbackIfNeeded() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   if (renderer_ && !renderer_ended_)
     return;
@@ -786,23 +841,26 @@ void PipelineImpl::RunEndedCallbackIfNeeded() {
     return;
 
   DCHECK_EQ(status_, PIPELINE_OK);
-  ended_cb_.Run();
+  main_task_runner_->PostTask(FROM_HERE,
+                              base::Bind(&Pipeline::Client::OnEnded,
+                                         client_weak_factory_->GetWeakPtr()));
 }
 
 std::unique_ptr<TextRenderer> PipelineImpl::CreateTextRenderer() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   const base::CommandLine* cmd_line = base::CommandLine::ForCurrentProcess();
   if (!cmd_line->HasSwitch(switches::kEnableInbandTextTracks))
     return nullptr;
 
   return base::WrapUnique(new media::TextRenderer(
-      task_runner_, base::Bind(&PipelineImpl::OnAddTextTrack, weak_this_)));
+      media_task_runner_,
+      base::Bind(&PipelineImpl::OnAddTextTrack, weak_this_)));
 }
 
 void PipelineImpl::AddTextStreamTask(DemuxerStream* text_stream,
                                      const TextTrackConfig& config) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   // TODO(matthewjheaney): fix up text_ended_ when text stream
   // is added (http://crbug.com/321446).
   if (text_renderer_)
@@ -810,24 +868,30 @@ void PipelineImpl::AddTextStreamTask(DemuxerStream* text_stream,
 }
 
 void PipelineImpl::RemoveTextStreamTask(DemuxerStream* text_stream) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+
   if (text_renderer_)
     text_renderer_->RemoveTextStream(text_stream);
 }
 
 void PipelineImpl::OnAddTextTrack(const TextTrackConfig& config,
                                   const AddTextTrackDoneCB& done_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  add_text_track_cb_.Run(config, done_cb);
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+
+  main_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&Pipeline::Client::OnAddTextTrack,
+                 client_weak_factory_->GetWeakPtr(), config, done_cb));
 }
 
 void PipelineImpl::InitializeDemuxer(const PipelineStatusCB& done_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+
   demuxer_->Initialize(this, done_cb, !!text_renderer_);
 }
 
 void PipelineImpl::InitializeRenderer(const PipelineStatusCB& done_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   if (!demuxer_->GetStream(DemuxerStream::AUDIO) &&
       !demuxer_->GetStream(DemuxerStream::VIDEO)) {
@@ -848,11 +912,12 @@ void PipelineImpl::InitializeRenderer(const PipelineStatusCB& done_cb) {
       base::Bind(&PipelineImpl::BufferingStateChanged, weak_this_),
       base::Bind(&PipelineImpl::OnRendererEnded, weak_this_),
       base::Bind(&PipelineImpl::OnError, weak_this_),
-      waiting_for_decryption_key_cb_);
+      base::Bind(&PipelineImpl::OnWaitingForDecryptionKey, weak_this_));
 }
 
 void PipelineImpl::ReportMetadata() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+
   PipelineMetadata metadata;
   metadata.timeline_offset = demuxer_->GetTimelineOffset();
   DemuxerStream* stream = demuxer_->GetStream(DemuxerStream::VIDEO);
@@ -864,13 +929,20 @@ void PipelineImpl::ReportMetadata() {
   if (demuxer_->GetStream(DemuxerStream::AUDIO)) {
     metadata.has_audio = true;
   }
-  metadata_cb_.Run(metadata);
+
+  main_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&Pipeline::Client::OnMetadata,
+                            client_weak_factory_->GetWeakPtr(), metadata));
 }
 
 void PipelineImpl::BufferingStateChanged(BufferingState new_buffering_state) {
   DVLOG(1) << __FUNCTION__ << "(" << new_buffering_state << ") ";
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  buffering_state_cb_.Run(new_buffering_state);
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+
+  main_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&Pipeline::Client::OnBufferingStateChange,
+                 client_weak_factory_->GetWeakPtr(), new_buffering_state));
 }
 
 }  // namespace media