Pipeline & CompositeFilter now use MessageLoopProxy instead of plain MessageLoop.

This prevents a race between initialization and teardown, for example, where
Pipeline tries to use its message_loop_ when FFVD::Initialize() is completed (on
the decoder thread) but the pipeline thread's loop has already been freed.

Because I ran into it during this chase, also made MessageLoopFactory use a list instead of a map.  There are only 1-3 threads that the factory every knows about, and using a list allows destroying them in reverse order of creation, for at least a little more predictability.

BUG=117341


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

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

7 files changed, 81 insertions, 75 deletions

diff --git a/media/base/composite_filter.cc b/media/base/composite_filter.cc
index 4c6c867..59776be 100644
--- a/media/base/composite_filter.cc
+++ b/media/base/composite_filter.cc
@@ -7,7 +7,8 @@
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/callback_helpers.h"
-#include "base/message_loop.h"
+#include "base/location.h"
+#include "base/message_loop_proxy.h"
 #include "base/stl_util.h"
 
 namespace media {
@@ -33,7 +34,8 @@ class CompositeFilter::FilterHostImpl : public FilterHost {
   DISALLOW_COPY_AND_ASSIGN(FilterHostImpl);
 };
 
-CompositeFilter::CompositeFilter(MessageLoop* message_loop)
+CompositeFilter::CompositeFilter(
+    const scoped_refptr<base::MessageLoopProxy>& message_loop)
     : state_(kCreated),
       sequence_index_(0),
       message_loop_(message_loop),
@@ -43,14 +45,14 @@ CompositeFilter::CompositeFilter(MessageLoop* message_loop)
 }
 
 CompositeFilter::~CompositeFilter() {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(state_ == kCreated || state_ == kStopped);
 
   filters_.clear();
 }
 
 void CompositeFilter::AddFilter(scoped_refptr<Filter> filter) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   CHECK(filter && state_ == kCreated && host());
 
   // Register ourselves as the filter's host.
@@ -59,7 +61,7 @@ void CompositeFilter::AddFilter(scoped_refptr<Filter> filter) {
 }
 
 void CompositeFilter::RemoveFilter(scoped_refptr<Filter> filter) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (!filter.get() || state_ != kCreated || !host())
     LOG(FATAL) << "Unknown filter, or in unexpected state.";
 
@@ -75,7 +77,7 @@ void CompositeFilter::RemoveFilter(scoped_refptr<Filter> filter) {
 }
 
 void CompositeFilter::set_host(FilterHost* host) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(host);
   DCHECK(!host_impl_.get());
   host_impl_.reset(new FilterHostImpl(this, host));
@@ -86,7 +88,7 @@ FilterHost* CompositeFilter::host() {
 }
 
 void CompositeFilter::Play(const base::Closure& play_cb) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (IsOperationPending()) {
     SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
     play_cb.Run();
@@ -106,7 +108,7 @@ void CompositeFilter::Play(const base::Closure& play_cb) {
 }
 
 void CompositeFilter::Pause(const base::Closure& pause_cb) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (IsOperationPending()) {
     SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
     pause_cb.Run();
@@ -126,7 +128,7 @@ void CompositeFilter::Pause(const base::Closure& pause_cb) {
 }
 
 void CompositeFilter::Flush(const base::Closure& flush_cb) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (IsOperationPending()) {
     SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
     flush_cb.Run();
@@ -143,7 +145,7 @@ void CompositeFilter::Flush(const base::Closure& flush_cb) {
 }
 
 void CompositeFilter::Stop(const base::Closure& stop_cb) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (!host()) {
     SendErrorToHost(PIPELINE_ERROR_INVALID_STATE);
     stop_cb.Run();
@@ -190,7 +192,7 @@ void CompositeFilter::Stop(const base::Closure& stop_cb) {
 }
 
 void CompositeFilter::SetPlaybackRate(float playback_rate) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   for (FilterVector::iterator iter = filters_.begin();
        iter != filters_.end();
        ++iter) {
@@ -200,7 +202,7 @@ void CompositeFilter::SetPlaybackRate(float playback_rate) {
 
 void CompositeFilter::Seek(base::TimeDelta time,
                            const PipelineStatusCB& seek_cb) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   if (IsOperationPending()) {
     seek_cb.Run(PIPELINE_ERROR_OPERATION_PENDING);
@@ -217,7 +219,7 @@ void CompositeFilter::Seek(base::TimeDelta time,
 }
 
 void CompositeFilter::OnAudioRendererDisabled() {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   for (FilterVector::iterator iter = filters_.begin();
        iter != filters_.end();
        ++iter) {
@@ -226,12 +228,12 @@ void CompositeFilter::OnAudioRendererDisabled() {
 }
 
 void CompositeFilter::ChangeState(State new_state) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   state_ = new_state;
 }
 
 void CompositeFilter::StartSerialCallSequence() {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   status_ = PIPELINE_OK;
   sequence_index_ = 0;
 
@@ -244,7 +246,7 @@ void CompositeFilter::StartSerialCallSequence() {
 }
 
 void CompositeFilter::StartParallelCallSequence() {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   status_ = PIPELINE_OK;
   sequence_index_ = 0;
 
@@ -337,7 +339,7 @@ CompositeFilter::State CompositeFilter::GetNextState(State state) const {
 }
 
 void CompositeFilter::SerialCallback() {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (status_ != PIPELINE_OK) {
     // We encountered an error. Terminate the sequence now.
     ChangeState(kError);
@@ -362,7 +364,7 @@ void CompositeFilter::SerialCallback() {
 }
 
 void CompositeFilter::ParallelCallback() {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   if (!filters_.empty())
     sequence_index_++;
@@ -405,9 +407,10 @@ void CompositeFilter::SendErrorToHost(PipelineStatus error) {
 }
 
 // Execute |closure| if on |message_loop|, otherwise post to it.
-static void TrampolineClosureIfNecessary(MessageLoop* message_loop,
-                                         const base::Closure& closure) {
-  if (MessageLoop::current() == message_loop)
+static void TrampolineClosureIfNecessary(
+    const scoped_refptr<base::MessageLoopProxy>& message_loop,
+    const base::Closure& closure) {
+  if (message_loop->BelongsToCurrentThread())
     closure.Run();
   else
     message_loop->PostTask(FROM_HERE, closure);
@@ -439,13 +442,13 @@ void CompositeFilter::OnStatusCB(const base::Closure& callback,
 }
 
 void CompositeFilter::SetError(PipelineStatus error) {
-  if (message_loop_ != MessageLoop::current()) {
+  if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE,
         base::Bind(&CompositeFilter::SetError, this, error));
     return;
   }
 
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK_NE(state_, kCreated);
 
   // Drop errors recieved while stopping or stopped.
diff --git a/media/base/composite_filter.h b/media/base/composite_filter.h
index e05b826..91aaa68 100644
--- a/media/base/composite_filter.h
+++ b/media/base/composite_filter.h
@@ -12,13 +12,16 @@
 #include "media/base/filter_host.h"
 #include "media/base/filters.h"
 
-class MessageLoop;
+namespace base {
+class MessageLoopProxy;
+}
 
 namespace media {
 
 class MEDIA_EXPORT CompositeFilter : public Filter {
  public:
-  explicit CompositeFilter(MessageLoop* message_loop);
+  explicit CompositeFilter(
+      const scoped_refptr<base::MessageLoopProxy>& message_loop);
 
   // Adds a filter to the composite. This is only allowed after set_host()
   // is called and before the first state changing operation such as Play(),
@@ -131,7 +134,7 @@ class MEDIA_EXPORT CompositeFilter : public Filter {
   unsigned int sequence_index_;
 
   // Message loop passed into the constructor.
-  MessageLoop* message_loop_;
+  scoped_refptr<base::MessageLoopProxy> message_loop_;
 
   // FilterHost implementation passed to Filters owned by this
   // object.
diff --git a/media/base/composite_filter_unittest.cc b/media/base/composite_filter_unittest.cc
index 85bec70..e5442d7 100644
--- a/media/base/composite_filter_unittest.cc
+++ b/media/base/composite_filter_unittest.cc
@@ -141,7 +141,7 @@ class CompositeFilterTest : public testing::Test {
 };
 
 CompositeFilterTest::CompositeFilterTest() :
-    composite_(new CompositeFilter(&message_loop_)),
+    composite_(new CompositeFilter(message_loop_.message_loop_proxy())),
     filter_1_status_(PIPELINE_OK),
     filter_2_status_(PIPELINE_OK),
     mock_filter_host_(new StrictMock<MockFilterHost>()) {
@@ -151,7 +151,7 @@ CompositeFilterTest::~CompositeFilterTest() {}
 
 void CompositeFilterTest::SetupAndAdd2Filters() {
   mock_filter_host_.reset(new StrictMock<MockFilterHost>());
-  composite_ = new CompositeFilter(&message_loop_);
+  composite_ = new CompositeFilter(message_loop_.message_loop_proxy());
   composite_->set_host(mock_filter_host_.get());
 
   // Setup |filter_1_| and arrange for methods to set
diff --git a/media/base/message_loop_factory.cc b/media/base/message_loop_factory.cc
index a62abc5..22ec96c 100644
--- a/media/base/message_loop_factory.cc
+++ b/media/base/message_loop_factory.cc
@@ -11,17 +11,13 @@ namespace media {
 MessageLoopFactory::MessageLoopFactory() {}
 
 MessageLoopFactory::~MessageLoopFactory() {
-  for (ThreadMap::iterator iter = thread_map_.begin();
-       iter != thread_map_.end();
-       ++iter) {
-    base::Thread* thread = (*iter).second;
-
-    if (thread) {
-      thread->Stop();
-      delete thread;
-    }
+  for (ThreadList::reverse_iterator it = threads_.rbegin();
+       it != threads_.rend(); ++it) {
+    base::Thread* thread = it->second;
+    thread->Stop();
+    delete thread;
   }
-  thread_map_.clear();
+  threads_.clear();
 }
 
 MessageLoop* MessageLoopFactory::GetMessageLoop(const std::string& name) {
@@ -37,13 +33,14 @@ base::Thread* MessageLoopFactory::GetThread(const std::string& name) {
   DCHECK(!name.empty());
 
   base::AutoLock auto_lock(lock_);
-  ThreadMap::iterator it = thread_map_.find(name);
-  if (it != thread_map_.end())
-    return (*it).second;
+  for (ThreadList::iterator it = threads_.begin(); it != threads_.end(); ++it) {
+    if (it->first == name)
+      return it->second;
+  }
 
   base::Thread* thread = new base::Thread(name.c_str());
   CHECK(thread->Start()) << "Failed to start thread: " << name;
-  thread_map_[name] = thread;
+  threads_.push_back(std::make_pair(name, thread));
   return thread;
 }
 
diff --git a/media/base/message_loop_factory.h b/media/base/message_loop_factory.h
index d8ecdb5..f5ecda2 100644
--- a/media/base/message_loop_factory.h
+++ b/media/base/message_loop_factory.h
@@ -5,7 +5,7 @@
 #ifndef MEDIA_BASE_MESSAGE_LOOP_FACTORY_H_
 #define MEDIA_BASE_MESSAGE_LOOP_FACTORY_H_
 
-#include <map>
+#include <list>
 #include <string>
 
 #include "base/memory/ref_counted.h"
@@ -53,8 +53,10 @@ class MEDIA_EXPORT MessageLoopFactory {
   // Lock used to serialize access for the following data members.
   base::Lock lock_;
 
-  typedef std::map<std::string, base::Thread*> ThreadMap;
-  ThreadMap thread_map_;
+  // List of pairs of created threads and their names.  We use a list to ensure
+  // threads are stopped & deleted in reverse order of creation.
+  typedef std::list<std::pair<std::string, base::Thread*> > ThreadList;
+  ThreadList threads_;
 
   DISALLOW_COPY_AND_ASSIGN(MessageLoopFactory);
 };
diff --git a/media/base/pipeline.cc b/media/base/pipeline.cc
index 699aba7..d3854e4 100644
--- a/media/base/pipeline.cc
+++ b/media/base/pipeline.cc
@@ -64,7 +64,7 @@ struct Pipeline::PipelineInitState {
 };
 
 Pipeline::Pipeline(MessageLoop* message_loop, MediaLog* media_log)
-    : message_loop_(message_loop),
+    : message_loop_(message_loop->message_loop_proxy()),
       media_log_(media_log),
       clock_(new Clock(&base::Time::Now)),
       waiting_for_clock_update_(false),
@@ -337,22 +337,22 @@ bool Pipeline::IsPipelineOk() {
 }
 
 bool Pipeline::IsPipelineStopped() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   return state_ == kStopped || state_ == kError;
 }
 
 bool Pipeline::IsPipelineTearingDown() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   return tearing_down_;
 }
 
 bool Pipeline::IsPipelineStopPending() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   return stop_pending_;
 }
 
 bool Pipeline::IsPipelineSeeking() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (!seek_pending_)
     return false;
   DCHECK(kSeeking == state_ || kPausing == state_ ||
@@ -371,7 +371,7 @@ void Pipeline::ReportStatus(const PipelineStatusCB& cb, PipelineStatus status) {
 }
 
 void Pipeline::FinishInitialization() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   // Execute the seek callback, if present.  Note that this might be the
   // initial callback passed into Start().
   ReportStatus(seek_cb_, status_);
@@ -596,7 +596,7 @@ void Pipeline::StartTask(scoped_ptr<FilterCollection> filter_collection,
                          const PipelineStatusCB& error_cb,
                          const NetworkEventCB& network_cb,
                          const PipelineStatusCB& start_cb) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK_EQ(kCreated, state_);
   filter_collection_ = filter_collection.Pass();
   ended_cb_ = ended_cb;
@@ -631,7 +631,7 @@ void Pipeline::StartTask(scoped_ptr<FilterCollection> filter_collection,
 // works like a big state change table. If we no longer need to start filters
 // in order, we need to get rid of all the state change.
 void Pipeline::InitializeTask(PipelineStatus last_stage_status) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   if (last_stage_status != PIPELINE_OK) {
     // Currently only VideoDecoders have a recoverable error code.
@@ -733,7 +733,7 @@ void Pipeline::InitializeTask(PipelineStatus last_stage_status) {
 // Stop() tasks even if we've already stopped.  Perhaps this should no-op for
 // additional calls, however most of this logic will be changing.
 void Pipeline::StopTask(const base::Closure& stop_cb) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(!IsPipelineStopPending());
   DCHECK_NE(state_, kStopped);
 
@@ -764,7 +764,7 @@ void Pipeline::StopTask(const base::Closure& stop_cb) {
 }
 
 void Pipeline::ErrorChangedTask(PipelineStatus error) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK_NE(PIPELINE_OK, error) << "PIPELINE_OK isn't an error!";
 
   // Suppress executing additional error logic. Note that if we are currently
@@ -789,7 +789,7 @@ void Pipeline::ErrorChangedTask(PipelineStatus error) {
 }
 
 void Pipeline::PlaybackRateChangedTask(float playback_rate) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   if (!running_ || tearing_down_)
     return;
@@ -817,7 +817,7 @@ void Pipeline::PlaybackRateChangedTask(float playback_rate) {
 }
 
 void Pipeline::VolumeChangedTask(float volume) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (!running_ || tearing_down_)
     return;
 
@@ -827,7 +827,7 @@ void Pipeline::VolumeChangedTask(float volume) {
 
 void Pipeline::SeekTask(base::TimeDelta time,
                         const PipelineStatusCB& seek_cb) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(!IsPipelineStopPending());
 
   // Suppress seeking if we're not fully started.
@@ -864,7 +864,7 @@ void Pipeline::SeekTask(base::TimeDelta time,
 }
 
 void Pipeline::NotifyEndedTask() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   // We can only end if we were actually playing.
   if (state_ != kStarted) {
@@ -901,13 +901,13 @@ void Pipeline::NotifyEndedTask() {
 }
 
 void Pipeline::NotifyNetworkEventTask(NetworkEvent type) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (!network_cb_.is_null())
     network_cb_.Run(type);
 }
 
 void Pipeline::DisableAudioRendererTask() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   base::AutoLock auto_lock(lock_);
   has_audio_ = false;
@@ -929,7 +929,7 @@ void Pipeline::DisableAudioRendererTask() {
 }
 
 void Pipeline::FilterStateTransitionTask() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
   // No reason transitioning if we've errored or have stopped.
   if (IsPipelineStopped()) {
@@ -1057,7 +1057,7 @@ void Pipeline::TeardownStateTransitionTask() {
 }
 
 void Pipeline::FinishDestroyingFiltersTask() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineStopped());
 
   // Clear filter references.
@@ -1082,7 +1082,7 @@ void Pipeline::FinishDestroyingFiltersTask() {
 }
 
 void Pipeline::InitializeDemuxer() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   demuxer_ = filter_collection_->GetDemuxer();
@@ -1095,7 +1095,7 @@ void Pipeline::InitializeDemuxer() {
 }
 
 void Pipeline::OnDemuxerInitialized(PipelineStatus status) {
-  if (MessageLoop::current() != message_loop_) {
+  if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnDemuxerInitialized, this, status));
     return;
@@ -1118,7 +1118,7 @@ void Pipeline::OnDemuxerInitialized(PipelineStatus status) {
 
 bool Pipeline::InitializeAudioDecoder(
     const scoped_refptr<Demuxer>& demuxer) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
   DCHECK(demuxer);
 
@@ -1144,7 +1144,7 @@ bool Pipeline::InitializeAudioDecoder(
 
 bool Pipeline::InitializeVideoDecoder(
     const scoped_refptr<Demuxer>& demuxer) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
   DCHECK(demuxer);
 
@@ -1172,7 +1172,7 @@ bool Pipeline::InitializeVideoDecoder(
 
 bool Pipeline::InitializeAudioRenderer(
     const scoped_refptr<AudioDecoder>& decoder) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   if (!decoder)
@@ -1197,7 +1197,7 @@ bool Pipeline::InitializeAudioRenderer(
 
 bool Pipeline::InitializeVideoRenderer(
     const scoped_refptr<VideoDecoder>& decoder) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   if (!decoder)
@@ -1220,7 +1220,7 @@ bool Pipeline::InitializeVideoRenderer(
 }
 
 void Pipeline::TearDownPipeline() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK_NE(kStopped, state_);
 
   DCHECK(!tearing_down_ ||  // Teardown on Stop().
@@ -1297,7 +1297,7 @@ void Pipeline::DoStop(const base::Closure& callback) {
 }
 
 void Pipeline::OnDemuxerStopDone(const base::Closure& callback) {
-  if (MessageLoop::current() != message_loop_) {
+  if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnDemuxerStopDone, this, callback));
     return;
@@ -1326,7 +1326,7 @@ void Pipeline::DoSeek(base::TimeDelta seek_timestamp) {
 
 void Pipeline::OnDemuxerSeekDone(base::TimeDelta seek_timestamp,
                                  PipelineStatus status) {
-  if (MessageLoop::current() != message_loop_) {
+  if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnDemuxerSeekDone, this, seek_timestamp, status));
     return;
@@ -1344,7 +1344,7 @@ void Pipeline::OnDemuxerSeekDone(base::TimeDelta seek_timestamp,
 }
 
 void Pipeline::OnAudioUnderflow() {
-  if (MessageLoop::current() != message_loop_) {
+  if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnAudioUnderflow, this));
     return;
@@ -1363,7 +1363,7 @@ void Pipeline::OnCanPlayThrough() {
 }
 
 void Pipeline::NotifyCanPlayThrough() {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   NotifyNetworkEventTask(CAN_PLAY_THROUGH);
 }
 
diff --git a/media/base/pipeline.h b/media/base/pipeline.h
index f2c7751..1b10def 100644
--- a/media/base/pipeline.h
+++ b/media/base/pipeline.h
@@ -20,6 +20,7 @@
 class MessageLoop;
 
 namespace base {
+class MessageLoopProxy;
 class TimeDelta;
 }
 
@@ -464,7 +465,7 @@ class MEDIA_EXPORT Pipeline
   void ReportStatus(const PipelineStatusCB& cb, PipelineStatus status);
 
   // Message loop used to execute pipeline tasks.
-  MessageLoop* message_loop_;
+  scoped_refptr<base::MessageLoopProxy> message_loop_;
 
   // MediaLog to which to log events.
   scoped_refptr<MediaLog> media_log_;