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 e9b860243762dd0233743ed68f40e8dce3ad4b02..1da520d999aef5cb8153bcf6fc797dbb1bdf35d7 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()),

[sandersd (OOO until July 31), 2016/05/02 18:37:47]

Perhaps this should be explicitly passed, since |client| is explicit and the two
are related?

[xhwang, 2016/05/02 20:58:35]

+1

[alokp, 2016/05/02 22:25:17]

I disagree. IMO passing a task runner makes sense when instance of a class
should be used on a thread different from the one on which it is constructed. It
is implicit that non-thread-safe class must be created and used exclusively on
one thread.

That said, I do not feel strongly about this and am OK with adding it if you
guys do not agree with my argument above.

[sandersd (OOO until July 31), 2016/05/02 22:45:52]

This is reasonable, although I would feel better about it if the client was
passed to the constructor. As a general rule I prefer explicitness, but I'll let
xhwang@ make the call on how important that is.

[alokp, 2016/05/02 23:00:59]

I agree. In fact initial patches passed client in constructor but we decided
against it to keep Pipeline a pure abstract interface.

[sandersd (OOO until July 31), 2016/05/02 23:20:47]

That's true, but the reason to do so was also removed (stashing the pointer in
the base class). We can still switch back without re-introducing that, if that
is better.

[alokp, 2016/05/03 05:22:18]

How do you propose doing that? It will be weird to talk about Client in the base
class without an API on the base class to pass it.

[sandersd (OOO until July 31), 2016/05/03 19:28:55]

A fair point, I'll accept that.

[alokp, 2016/05/03 20:15:23]

One way to add Client in the base class interface without passing it in the
constructor would be a static Create function. Would you prefer that?

+      media_task_runner_(media_task_runner),
       media_log_(media_log),
       running_(false),
       did_loading_progress_(false),
@@ -48,35 +51,29 @@ PipelineImpl::PipelineImpl(
       renderer_ended_(false),
       text_renderer_ended_(false),
       demuxer_(NULL),
+      client_(nullptr),
       cdm_context_(nullptr),
       weak_factory_(this) {
+  DVLOG(2) << __FUNCTION__;
   weak_this_ = weak_factory_.GetWeakPtr();
   media_log_->AddEvent(media_log_->CreatePipelineStateChangedEvent(kCreated));
 }
 
 PipelineImpl::~PipelineImpl() {
-  DCHECK(thread_checker_.CalledOnValidThread())
+  DVLOG(2) << __FUNCTION__;
+  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 +81,116 @@ 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_ = client;
+  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.
+    base::WaitableEvent waiter(false, false);
+    media_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&PipelineImpl::StopTask, weak_this_, &waiter));

[sandersd (OOO until July 31), 2016/05/02 18:37:47]

I'm not a fan of this pattern; it was bad in WMPI and it's worse hidden behind
an API. (Not that I have immediately implementable improvements to offer.)

At the very least, we should harden it against the |media_task_runner_| being
stopped, and document that it synchronously calls to the media thread (since
this could cause deadlocks).

If there is no bug already for replacing it, we should file one
(Hotlist-CodeHealth).

[xhwang, 2016/05/02 20:58:35]

We already have a waiter in WMPI when Pipeline::Stop() is called:
https://code.google.com/p/chromium/codesearch#chromium/src/media/blink/webmed...

We don't like that pattern but it's kinda hard to remove (at least when I looked
at it last time). Therefore, it'd be nice if we could avoid adding another
waiter.

Is there anything particular that you want to solve by using a sync Stop with
waiter instead of the async Stop?

[alokp, 2016/05/02 22:25:17]

This is not a new waiter. The one in WMPI is simply moved here to make the
Pipeline API consistent from threading POV. I think it is better because earlier
stop_cb was the only callback called directly on the media thread. All the other
callbacks were called on the main thread. This is inconsistent from the client
POV.

I have added a TODO to get rid of this synchronous wait in a follow-up patch. I
think it should be easier now that all threading logic is localized in this
class.

+    waiter.Wait();
+  } else {
+    // This path is executed by unittests that share media and main threads.
+    StopTask(nullptr);
+  }
+  client_ = nullptr;
 }
 
 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 +199,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 +222,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 +242,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,8 +278,7 @@ const char* PipelineImpl::GetStateString(State state) {
 #undef RETURN_STRING
 
 PipelineImpl::State PipelineImpl::GetNextState() const {
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  DCHECK(stop_cb_.is_null()) << "State transitions don't happen when stopping";
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK_EQ(status_, PIPELINE_OK)
       << "State transitions don't happen when there's an error: " << status_;
 
@@ -280,34 +312,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 +357,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,
+                                         base::Unretained(client_)));

[xhwang, 2016/05/02 20:58:35]

Could you explain why base::Unretained is safe here and below? Is it possible
that when this task is run, the whole media element and WMPI/Pipeline is already
destroyed?

[alokp, 2016/05/02 22:25:17]

It is not safe. The latest patch uses weak pointer.

 }
 
 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;
   }
 
@@ -404,7 +444,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;
@@ -432,9 +472,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(base::WaitableEvent* waiter) {
   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,
@@ -452,55 +493,27 @@ 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_);
+  // If we stop during initialization/seeking/suspending we don't want to leave
+  // 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();
 
+  SetState(kStopped);
   {
     base::AutoLock auto_lock(lock_);
     running_ = false;
   }
+  weak_factory_.InvalidateWeakPtrs();
 
-  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();
-
-    // Invalid all weak pointers so it's safe to destroy |this| on the render
-    // main thread.
-    weak_factory_.InvalidateWeakPtrs();
-
-    base::ResetAndReturn(&stop_cb_).Run();
-
-    // NOTE: pipeline may be deleted at this point in time as a result of
-    // executing |stop_cb_|.
-    return;
-  }
-  if (!error_cb_.is_null()) {
-    DCHECK_NE(status_, PIPELINE_OK);
-    base::ResetAndReturn(&error_cb_).Run(status_);
-  }
+  if (waiter)
+    waiter->Signal();
 }
 
 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;
@@ -517,8 +530,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,
+                            base::Unretained(client_)));
+}
+
 void PipelineImpl::StartTask() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   CHECK_EQ(kCreated, state_)
       << "Media pipeline cannot be started more than once";
@@ -532,27 +553,15 @@ void PipelineImpl::StartTask() {
   StateTransitionTask(PIPELINE_OK);
 }
 
-void PipelineImpl::StopTask(const base::Closure& stop_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  DCHECK(stop_cb_.is_null());
-
-  if (state_ == kStopped) {
-    // Invalid all weak pointers so it's safe to destroy |this| on the render
-    // main thread.
-    weak_factory_.InvalidateWeakPtrs();
-
-    // NOTE: pipeline may be deleted at this point in time as a result of
-    // executing |stop_cb|.
-    stop_cb.Run();
-
-    return;
-  }
-
-  stop_cb_ = stop_cb;
+void PipelineImpl::StopTask(base::WaitableEvent* waiter) {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // We may already be stopping due to a runtime error.
-  if (state_ == kStopping)
+  if (state_ == kStopped || state_ == kStopping) {
+    if (waiter)
+      waiter->Signal();
     return;
+  }
 
   // Do not report statistics if the pipeline is not fully initialized.
   if (state_ == kSeeking || state_ == kPlaying || state_ == kSuspending ||
@@ -566,33 +575,53 @@ void PipelineImpl::StopTask(const base::Closure& stop_cb) {
 
   SetState(kStopping);
   pending_callbacks_.reset();
-  DoStop(base::Bind(&PipelineImpl::OnStopCompleted, weak_this_));
+  DoStop(waiter);
 }
 
 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
   // 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 status_reported = false;
+  if (!seek_cb_.is_null()) {
+    base::ResetAndReturn(&seek_cb_).Run(status_);

[sandersd (OOO until July 31), 2016/05/02 18:37:47]

Are we sure it's safe to drop these from the Stop() path? It's an overall
improvement to the design, but I worry that other code may still be waiting for
these callbacks. (Maybe it doesn't matter anyway, since the resulting state is
kStopped, but I've not checked every caller.)

[alokp, 2016/05/02 22:25:17]

It is not possible to call these since Stop is blocking. We should fix the
callers if they are waiting for these callbacks.

[sandersd (OOO until July 31), 2016/05/02 22:45:51]

That's fair, but we also can't commit this without doing that (or at least
scoping it) first.

[alokp, 2016/05/02 23:00:59]

PipelineImpl::Seek/Suspend is only called from PipelineController, unittests for
which pass. Also I do not see PipelineController waiting for these
notifications.

[sandersd (OOO until July 31), 2016/05/02 23:20:47]

PipelineController binds them all to OnPipelineStatus(), and does explicit error
handling there.

[alokp, 2016/05/03 05:22:18]

Right. I am sorry I do not see a problem there. Can you elaborate?

PipelineController::OnPipelineStatus will not be called *only* if WMPI stops the
Pipeline before it has a chance to call the seek/resume callback. Is that a
problem?

BTW I find it confusing to control the pipeline from two places - WMPI and
PipelineController. Do we really need PipelineController?

[sandersd (OOO until July 31), 2016/05/03 19:28:55]

I'm not sure if it's a problem, it's probably not, but it's not trivial to
check.

The actual plan is to get rid of everywhere that WMPI controls the Pipeline, and
then likely merge PipelineController into Pipeline. I would say to think of it
as an in-progress refactor of the Pipeline API.

[alokp, 2016/05/03 20:15:23]

SGTM

+    status_reported = true;
+  }
+  if (!suspend_cb_.is_null()) {
+    base::ResetAndReturn(&suspend_cb_).Run(status_);
+    status_reported = true;
+  }
+  if (!status_reported) {
+    DCHECK_NE(status_, PIPELINE_OK);
+    main_task_runner_->PostTask(FROM_HERE,
+                                base::Bind(&Pipeline::Client::OnError,
+                                           base::Unretained(client_), status_));
+  }
 
-  DoStop(base::Bind(&PipelineImpl::OnStopCompleted, weak_this_));
+  // Stop the pipeline.
+  StopTask(nullptr);
 }
 
 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)
@@ -602,7 +631,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)
@@ -612,8 +641,7 @@ void PipelineImpl::VolumeChangedTask(float volume) {
 }
 
 void PipelineImpl::SeekTask(TimeDelta time, const PipelineStatusCB& seek_cb) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  DCHECK(stop_cb_.is_null());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   // Suppress seeking if we're not fully started.
   if (state_ != kPlaying) {
@@ -639,7 +667,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) {
@@ -687,7 +715,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) {
@@ -741,14 +769,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)
@@ -761,7 +789,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)
@@ -774,7 +802,7 @@ void PipelineImpl::OnTextRendererEnded() {
 }
 
 void PipelineImpl::RunEndedCallbackIfNeeded() {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
 
   if (renderer_ && !renderer_ended_)
     return;
@@ -783,23 +811,25 @@ void PipelineImpl::RunEndedCallbackIfNeeded() {
     return;
 
   DCHECK_EQ(status_, PIPELINE_OK);
-  ended_cb_.Run();
+  main_task_runner_->PostTask(FROM_HERE, base::Bind(&Pipeline::Client::OnEnded,
+                                                    base::Unretained(client_)));
 }
 
 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_)
@@ -807,24 +837,29 @@ 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,
+                            base::Unretained(client_), 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)) {
@@ -845,11 +880,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);
@@ -861,13 +897,19 @@ 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,
+                                         base::Unretained(client_), metadata));
 }
 
 void PipelineImpl::BufferingStateChanged(BufferingState new_buffering_state) {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DVLOG(1) << __FUNCTION__ << "(" << new_buffering_state << ") ";
-  DCHECK(task_runner_->BelongsToCurrentThread());
-  buffering_state_cb_.Run(new_buffering_state);
+
+  main_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&Pipeline::Client::OnBufferingStateChange,
+                            base::Unretained(client_), new_buffering_state));
 }
 
 }  // namespace media