Switching decoders to use the injected message loop.

TEST=test should continue to pass, movies stay in sync
BUG=none

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

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

7 files changed, 205 insertions, 210 deletions

diff --git a/media/base/filters.h b/media/base/filters.h
index 7f59138..43f24fd 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -174,6 +174,8 @@ class DemuxerStream : public base::RefCountedThreadSafe<DemuxerStream> {
 
   // Schedules a read.  When the |read_callback| is called, the downstream
   // filter takes ownership of the buffer by AddRef()'ing the buffer.
+  //
+  // TODO(scherkus): switch Read() callback to scoped_refptr<>.
   virtual void Read(Callback1<Buffer*>::Type* read_callback) = 0;
 
   // Given a class that supports the |Interface| and a related static method
@@ -218,6 +220,8 @@ class VideoDecoder : public MediaFilter {
   virtual const MediaFormat& media_format() = 0;
 
   // Schedules a read.  Decoder takes ownership of the callback.
+  //
+  // TODO(scherkus): switch Read() callback to scoped_refptr<>.
   virtual void Read(Callback1<VideoFrame*>::Type* read_callback) = 0;
 };
 
@@ -239,6 +243,8 @@ class AudioDecoder : public MediaFilter {
   virtual const MediaFormat& media_format() = 0;
 
   // Schedules a read.  Decoder takes ownership of the callback.
+  //
+  // TODO(scherkus): switch Read() callback to scoped_refptr<>.
   virtual void Read(Callback1<Buffer*>::Type* read_callbasck) = 0;
 };
 
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index 8a43382..5ed70e3 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -23,6 +23,8 @@ template <class Filter>
 bool SupportsSetMessageLoop() {
   switch (Filter::filter_type()) {
     case FILTER_DEMUXER:
+    case FILTER_AUDIO_DECODER:
+    case FILTER_VIDEO_DECODER:
       return true;
     default:
       return false;
@@ -456,7 +458,9 @@ void PipelineThread::StartTask(FilterFactory* filter_factory,
 // pipeline's error_ member to PIPELINE_STOPPING.  We stop the filters in the
 // reverse order.
 //
-// TODO(scherkus): beware!  this can get posted multiple times!  it shouldn't!
+// TODO(scherkus): beware!  this can get posted multiple times since we post
+// 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 PipelineThread::StopTask() {
   if (PipelineOk()) {
     pipeline_->error_ = PIPELINE_STOPPING;
@@ -635,6 +639,8 @@ scoped_refptr<Filter> PipelineThread::CreateFilter(
     } else {
       // Create a dedicated thread for this filter.
       if (SupportsSetMessageLoop<Filter>()) {
+        // TODO(scherkus): figure out a way to name these threads so it matches
+        // the filter type.
         scoped_ptr<base::Thread> thread(new base::Thread("FilterThread"));
         if (!thread.get() || !thread->Start()) {
           NOTREACHED() << "Could not start filter thread";
diff --git a/media/filters/decoder_base.h b/media/filters/decoder_base.h
index 4f70f7a..1508439 100644
--- a/media/filters/decoder_base.h
+++ b/media/filters/decoder_base.h
@@ -10,6 +10,7 @@
 #include <deque>
 
 #include "base/lock.h"
+#include "base/stl_util-inl.h"
 #include "base/task.h"
 #include "base/thread.h"
 #include "media/base/buffers.h"
@@ -25,120 +26,61 @@ class DecoderBase : public Decoder {
 
   // MediaFilter implementation.
   virtual void Stop() {
-    OnStop();
-    {
-      AutoLock auto_lock(lock_);
-      running_ = false;
-      if (process_task_) {
-        process_task_->Cancel();
-        process_task_ = NULL;
-      }
-      DiscardQueues_Locked();
-    }
-
-    // Stop our decoding thread.
-    thread_.Stop();
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &DecoderBase::StopTask));
   }
 
   virtual void Seek(base::TimeDelta time) {
-    // Delegate to the subclass first.
-    OnSeek(time);
-
-    // Flush the result queue.
-    AutoLock auto_lock(lock_);
-    result_queue_.clear();
-
-    // Flush the input queue. This will trigger more reads from the demuxer.
-    input_queue_.clear();
-
-    // Turn on the seeking flag so that we can discard buffers until a
-    // discontinuous buffer is received.
-    seeking_ = true;
-
-    // Trigger more reads and keep the process loop rolling.
-    ScheduleProcessTask_Locked();
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &DecoderBase::SeekTask, time));
   }
 
   // Decoder implementation.
   virtual bool Initialize(DemuxerStream* demuxer_stream) {
-    demuxer_stream_ = demuxer_stream;
-
-    // Start our internal decoding thread.
-    if (!thread_.Start()) {
-      host()->Error(PIPELINE_ERROR_DECODE);
-      return false;
-    }
-
-    if (!OnInitialize(demuxer_stream)) {
-      // Release our resources and stop our thread.
-      // TODO(scherkus): shouldn't stop a thread inside Initialize(), but until I
-      // figure out proper error signaling semantics we're going to do it anyway!!
-      host()->Error(PIPELINE_ERROR_DECODE);
-      demuxer_stream_ = NULL;
-      thread_.Stop();
-      return false;
-    }
-
-    DCHECK(!media_format_.empty());
-    host()->InitializationComplete();
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &DecoderBase::InitializeTask, demuxer_stream));
     return true;
   }
 
   virtual const MediaFormat& media_format() { return media_format_; }
 
-  // Audio or Video decoder.
+  // Audio or video decoder.
   virtual void Read(ReadCallback* read_callback) {
-    AutoLock auto_lock(lock_);
-    if (IsRunning()) {
-      read_queue_.push_back(read_callback);
-      ScheduleProcessTask_Locked();
-    } else {
-      delete read_callback;
-    }
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &DecoderBase::ReadTask, read_callback));
   }
 
   void OnReadComplete(Buffer* buffer) {
-    AutoLock auto_lock(lock_);
-    if (IsRunning()) {
-      // Once the |seeking_| flag is set we ignore every buffers here
-      // until we receive a discontinuous buffer and we will turn off the
-      // |seeking_| flag.
-      if (buffer->IsDiscontinuous()) {
-        // TODO(hclam): put a DCHECK here to assert |seeking_| being true.
-        // I cannot do this now because seek operation is not fully
-        // asynchronous. There may be pending seek requests even before the
-        // previous was finished.
-        seeking_ = false;
-      }
-      if (!seeking_)
-        input_queue_.push_back(buffer);
-      --pending_reads_;
-      ScheduleProcessTask_Locked();
-    }
+    // Little bit of magic here to get NewRunnableMethod() to generate a Task
+    // that holds onto a reference via scoped_refptr<>.
+    //
+    // TODO(scherkus): change the callback format to pass a scoped_refptr<> or
+    // better yet see if we can get away with not using reference counting.
+    scoped_refptr<Buffer> buffer_ref = buffer;
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &DecoderBase::ReadCompleteTask, buffer_ref));
   }
 
  protected:
-  // |thread_name| is mandatory and is used to identify the thread in debuggers.
-  explicit DecoderBase(const char* thread_name)
-      : running_(true),
-        demuxer_stream_(NULL),
-        thread_(thread_name),
-        pending_reads_(0),
-        process_task_(NULL),
-        seeking_(false) {
+  DecoderBase()
+      : pending_reads_(0),
+        seeking_(false),
+        state_(UNINITIALIZED),
+        thread_id_(NULL) {
   }
 
   virtual ~DecoderBase() {
-    DCHECK(!thread_.IsRunning());
-    DCHECK(!process_task_);
+    DCHECK(state_ == UNINITIALIZED || state_ == STOPPED);
+    DCHECK(result_queue_.empty());
+    DCHECK(read_queue_.empty());
   }
 
   // This method is called by the derived class from within the OnDecode method.
   // It places an output buffer in the result queue.  It must be called from
   // within the OnDecode method.
   void EnqueueResult(Output* output) {
-    AutoLock auto_lock(lock_);
-    if (IsRunning()) {
+    DCHECK_EQ(PlatformThread::CurrentId(), thread_id_);
+    if (!stopped_) {
       result_queue_.push_back(output);
     }
   }
@@ -166,156 +108,185 @@ class DecoderBase : public Decoder {
   // the EnequeueResult() method from within this method.
   virtual void OnDecode(Buffer* input) = 0;
 
-  bool IsRunning() const { return running_; }
+  // Used for subclasses who friend unit tests and need to set the thread id.
+  virtual void set_thread_id(PlatformThreadId thread_id) {
+    thread_id_ = thread_id;
+  }
 
   MediaFormat media_format_;
 
  private:
-  // The GCL compiler does not like .cc files that directly access members of
-  // a base class.  This inline method helps.
+  // GCC doesn't let us access superclass member variables directly, so use
+  // a helper to get around the situation.
+  //
+  // TODO(scherkus): another reason to add protected accessors to MediaFilter.
   FilterHost* host() const { return Decoder::host_; }
+  MessageLoop* message_loop() const { return Decoder::message_loop_; }
 
-  // Schedules a task that will execute the ProcessTask method.
-  void ScheduleProcessTask_Locked() {
-    lock_.AssertAcquired();
-    DCHECK(IsRunning());
-    if (!process_task_) {
-      process_task_ = NewRunnableMethod(this, &DecoderBase::ProcessTask);
-      thread_.message_loop()->PostTask(FROM_HERE, process_task_);
-    }
+  void StopTask() {
+    DCHECK_EQ(PlatformThread::CurrentId(), thread_id_);
+    // Delegate to the subclass first.
+    OnStop();
+
+    // Throw away all buffers in all queues.
+    result_queue_.clear();
+    STLDeleteElements(&read_queue_);
+    state_ = STOPPED;
   }
 
-  // The core work loop of the decoder base.  This method will run the methods
-  // SubmitReads_Locked(), ProcessInput_Locked(), and ProcessOutput_Locked() in
-  // a loop until they either produce no further work, or the filter is stopped.
-  // Once there is no further work to do, the method returns.  A later call to
-  // the ScheduleProcessTask_Locked() method will start this task again.
-  void ProcessTask() {
-    AutoLock auto_lock(lock_);
-    bool did_some_work;
-    do {
-      did_some_work  = SubmitReads_Locked();
-      did_some_work |= ProcessInput_Locked();
-      did_some_work |= ProcessOutput_Locked();
-    } while (IsRunning() && did_some_work);
-    DCHECK(process_task_ || !IsRunning());
-    process_task_ = NULL;
+  void SeekTask(base::TimeDelta time) {
+    DCHECK_EQ(PlatformThread::CurrentId(), thread_id_);
+    // Delegate to the subclass first.
+    OnSeek(time);
+
+    // Flush the result queue.
+    result_queue_.clear();
+
+    // Turn on the seeking flag so that we can discard buffers until a
+    // discontinuous buffer is received.
+    seeking_ = true;
   }
 
-  // If necessary, calls the |demuxer_stream_| to read buffers.  Returns true
-  // if reads have happened, else false.  This method must be called with
-  // |lock_| acquired.  If the method submits any reads, then it will Release()
-  // the |lock_| when calling the demuxer and then re-Acquire() the |lock_|.
-  bool SubmitReads_Locked() {
-    lock_.AssertAcquired();
-    bool did_read = false;
-    if (IsRunning() &&
-        pending_reads_ + input_queue_.size() < read_queue_.size()) {
-      did_read = true;
-      size_t read = read_queue_.size() - pending_reads_ - input_queue_.size();
-      pending_reads_ += read;
-      {
-        AutoUnlock unlock(lock_);
-        while (read) {
-          demuxer_stream_->
-              Read(NewCallback(this, &DecoderBase::OnReadComplete));
-          --read;
-        }
-      }
+  void InitializeTask(DemuxerStream* demuxer_stream) {
+    DCHECK(state_ == UNINITIALIZED);
+    DCHECK(!demuxer_stream_);
+    DCHECK(!thread_id_ || thread_id_ == PlatformThread::CurrentId());
+    demuxer_stream_ = demuxer_stream;
+
+    // Grab the thread id for debugging.
+    thread_id_ = PlatformThread::CurrentId();
+
+    // Delegate to subclass first.
+    if (!OnInitialize(demuxer_stream_)) {
+      host()->Error(PIPELINE_ERROR_DECODE);
+      return;
     }
-    return did_read;
+
+    // TODO(scherkus): subclass shouldn't mutate superclass media format.
+    DCHECK(!media_format_.empty()) << "Subclass did not set media_format_";
+    state_ = INITIALIZED;
+    host()->InitializationComplete();
   }
 
-  // If the |input_queue_| has any buffers, this method will call the derived
-  // class's OnDecode() method.
-  bool ProcessInput_Locked() {
-    lock_.AssertAcquired();
-    bool did_decode = false;
-    while (IsRunning() && !input_queue_.empty()) {
-      did_decode = true;
-      scoped_refptr<Buffer> input = input_queue_.front();
-      input_queue_.pop_front();
-      // Release |lock_| before calling the derived class to do the decode.
-      {
-        AutoUnlock unlock(lock_);
-        OnDecode(input);
-      }
+  void ReadTask(ReadCallback* read_callback) {
+    DCHECK_EQ(PlatformThread::CurrentId(), thread_id_);
+    // TODO(scherkus): should reply with a null operation (empty buffer).
+    if (stopped_) {
+      delete read_callback;
+      return;
     }
-    return did_decode;
-  }
 
-  // Removes any buffers from the |result_queue_| and calls the next callback
-  // in the |read_queue_|.
-  bool ProcessOutput_Locked() {
-    lock_.AssertAcquired();
-    bool called_renderer = false;
-    while (IsRunning() && !read_queue_.empty() && !result_queue_.empty()) {
-      called_renderer = true;
-      scoped_refptr<Output> output = result_queue_.front();
-      result_queue_.pop_front();
-      scoped_ptr<ReadCallback> read_callback(read_queue_.front());
-      read_queue_.pop_front();
-      // Release |lock_| before calling the renderer.
-      {
-        AutoUnlock unlock(lock_);
-        read_callback->Run(output);
-      }
+    // Enqueue the callback and attempt to fulfill it immediately.
+    read_queue_.push_back(read_callback);
+    FulfillPendingRead();
+
+    // Issue reads as necessary.
+    while (pending_reads_ < read_queue_.size()) {
+      demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
+      ++pending_reads_;
     }
-    return called_renderer;
   }
 
-  // Throw away all buffers in all queues.
-  void DiscardQueues_Locked() {
-    lock_.AssertAcquired();
-    input_queue_.clear();
-    result_queue_.clear();
-    while (!read_queue_.empty()) {
-      delete read_queue_.front();
-      read_queue_.pop_front();
+  void ReadCompleteTask(scoped_refptr<Buffer> buffer) {
+    DCHECK_EQ(PlatformThread::CurrentId(), thread_id_);
+    DCHECK_GT(pending_reads_, 0u);
+    --pending_reads_;
+    if (stopped_) {
+      return;
+    }
+
+    // Once the |seeking_| flag is set we ignore every buffers here
+    // until we receive a discontinuous buffer and we will turn off the
+    // |seeking_| flag.
+    if (buffer->IsDiscontinuous()) {
+      // TODO(hclam): put a DCHECK here to assert |seeking_| being true.
+      // I cannot do this now because seek operation is not fully
+      // asynchronous. There may be pending seek requests even before the
+      // previous was finished.
+      seeking_ = false;
+    }
+    if (seeking_) {
+      return;
     }
-  }
 
-  // The critical section for the decoder.
-  Lock lock_;
+    // Decode the frame right away.
+    OnDecode(buffer);
+
+    // Attempt to fulfill a pending read callback and schedule additional reads
+    // if necessary.
+    FulfillPendingRead();
+
+    // Issue reads as necessary.
+    //
+    // Note that it's possible for us to decode but not produce a frame, in
+    // which case |pending_reads_| will remain less than |read_queue_| so we
+    // need to schedule an additional read.
+    DCHECK_LE(pending_reads_, read_queue_.size());
+    while (pending_reads_ < read_queue_.size()) {
+      demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
+      ++pending_reads_;
+    }
+  }
 
-  // If false, then the Stop() method has been called, and no further processing
-  // of buffers should occur.
-  bool running_;
+  // Attempts to fulfill a single pending read by dequeuing a buffer and read
+  // callback pair and executing the callback.
+  void FulfillPendingRead() {
+    DCHECK_EQ(PlatformThread::CurrentId(), thread_id_);
+    if (read_queue_.empty() || result_queue_.empty()) {
+      return;
+    }
 
-  // Pointer to the demuxer stream that will feed us compressed buffers.
-  scoped_refptr<DemuxerStream> demuxer_stream_;
+    // Dequeue a frame and read callback pair.
+    scoped_refptr<Output> output = result_queue_.front();
+    scoped_ptr<ReadCallback> read_callback(read_queue_.front());
+    result_queue_.pop_front();
+    read_queue_.pop_front();
 
-  // The dedicated decoding thread for this filter.
-  base::Thread thread_;
+    // Execute the callback!
+    read_callback->Run(output);
+  }
 
-  // Number of times we have called Read() on the demuxer that have not yet
-  // been satisfied.
+  // Tracks the number of asynchronous reads issued to |demuxer_stream_|.
+  // Using size_t since it is always compared against deque::size().
   size_t pending_reads_;
 
-  CancelableTask* process_task_;
+  // If true, then Stop() has been called and no further processing of buffers
+  // should occur.
+  bool stopped_;
+
+  // An internal state of the decoder that indicates that are waiting for seek
+  // to complete. We expect to receive a discontinuous frame/packet from the
+  // demuxer to signal that seeking is completed.
+  bool seeking_;
 
-  // Queue of buffers read from the |demuxer_stream_|.
-  typedef std::deque< scoped_refptr<Buffer> > InputQueue;
-  InputQueue input_queue_;
+  // Pointer to the demuxer stream that will feed us compressed buffers.
+  scoped_refptr<DemuxerStream> demuxer_stream_;
 
   // Queue of decoded samples produced in the OnDecode() method of the decoder.
   // Any samples placed in this queue will be assigned to the OutputQueue
   // buffers once the OnDecode() method returns.
+  //
   // TODO(ralphl): Eventually we want to have decoders get their destination
   // buffer from the OutputQueue and write to it directly.  Until we change
   // from the Assignable buffer to callbacks and renderer-allocated buffers,
   // we need this extra queue.
-  typedef std::deque< scoped_refptr<Output> > ResultQueue;
+  typedef std::deque<scoped_refptr<Output> > ResultQueue;
   ResultQueue result_queue_;
 
   // Queue of callbacks supplied by the renderer through the Read() method.
   typedef std::deque<ReadCallback*> ReadQueue;
   ReadQueue read_queue_;
 
-  // An internal state of the decoder that indicates that are waiting for seek
-  // to complete. We expect to receive a discontinuous frame/packet from the
-  // demuxer to signal that seeking is completed.
-  bool seeking_;
+  // Simple state tracking variable.
+  enum State {
+    UNINITIALIZED,
+    INITIALIZED,
+    STOPPED,
+  };
+  State state_;
+
+  // Used for debugging.
+  PlatformThreadId thread_id_;
 
   DISALLOW_COPY_AND_ASSIGN(DecoderBase);
 };
diff --git a/media/filters/ffmpeg_audio_decoder.cc b/media/filters/ffmpeg_audio_decoder.cc
index 7462fa5..12e7be8 100644
--- a/media/filters/ffmpeg_audio_decoder.cc
+++ b/media/filters/ffmpeg_audio_decoder.cc
@@ -14,8 +14,7 @@ const size_t FFmpegAudioDecoder::kOutputBufferSize =
     AVCODEC_MAX_AUDIO_FRAME_SIZE;
 
 FFmpegAudioDecoder::FFmpegAudioDecoder()
-    : DecoderBase<AudioDecoder, Buffer>("AudioDecoderThread"),
-      codec_context_(NULL) {
+    : codec_context_(NULL) {
 }
 
 FFmpegAudioDecoder::~FFmpegAudioDecoder() {
diff --git a/media/filters/ffmpeg_video_decoder.cc b/media/filters/ffmpeg_video_decoder.cc
index 115f97b..6423db4 100644
--- a/media/filters/ffmpeg_video_decoder.cc
+++ b/media/filters/ffmpeg_video_decoder.cc
@@ -37,8 +37,7 @@ namespace media {
 static const int kDecodeThreads = 2;
 
 FFmpegVideoDecoder::FFmpegVideoDecoder()
-    : DecoderBase<VideoDecoder, VideoFrame>("VideoDecoderThread"),
-      width_(0),
+    : width_(0),
       height_(0),
       time_base_(new AVRational()),
       state_(kNormal),
diff --git a/media/filters/ffmpeg_video_decoder_unittest.cc b/media/filters/ffmpeg_video_decoder_unittest.cc
index ea236a4..40b122d 100644
--- a/media/filters/ffmpeg_video_decoder_unittest.cc
+++ b/media/filters/ffmpeg_video_decoder_unittest.cc
@@ -70,9 +70,15 @@ class FFmpegVideoDecoderTest : public testing::Test {
 
     // Create an FFmpegVideoDecoder.
     factory_ = FFmpegVideoDecoder::CreateFactory();
-    decoder_ = factory_->Create<VideoDecoder>(media_format);
+    decoder_ = factory_->Create<FFmpegVideoDecoder>(media_format);
     DCHECK(decoder_);
 
+    // Provide a message loop.
+    decoder_->SetMessageLoop(&message_loop_);
+
+    // Manually set the thread id for tests that don't initialize the decoder.
+    decoder_->set_thread_id(PlatformThread::CurrentId());
+
     // Prepare a filter host, pipeline and demuxer for the video decoder.
     pipeline_.reset(new MockPipeline());
     filter_host_.reset(new MockFilterHost<VideoDecoder>(pipeline_.get(),
@@ -100,18 +106,22 @@ class FFmpegVideoDecoderTest : public testing::Test {
     // Call Stop() to shut down internal threads.
     decoder_->Stop();
 
+    // Finish up any remaining tasks.
+    message_loop_.RunAllPending();
+
     // Reset MockFFmpeg.
     MockFFmpeg::set(NULL);
   }
 
   // Fixture members.
   scoped_refptr<FilterFactory> factory_;
-  scoped_refptr<VideoDecoder> decoder_;
+  scoped_refptr<FFmpegVideoDecoder> decoder_;
   scoped_ptr<MockPipeline> pipeline_;
   scoped_ptr<MockFilterHost<VideoDecoder> > filter_host_;
   scoped_refptr<MockDemuxerStream> demuxer_;
   scoped_refptr<DataBuffer> buffer_;
   scoped_refptr<DataBuffer> end_of_stream_buffer_;
+  MessageLoop message_loop_;
 
   // FFmpeg fixtures.
   AVStream stream_;
@@ -155,7 +165,8 @@ TEST_F(FFmpegVideoDecoderTest, Initialize_QueryInterfaceFails) {
   EXPECT_CALL(*demuxer_, QueryInterface(AVStreamProvider::interface_id()))
       .WillOnce(ReturnNull());
 
-  EXPECT_FALSE(decoder_->Initialize(demuxer_));
+  EXPECT_TRUE(decoder_->Initialize(demuxer_));
+  message_loop_.RunAllPending();
   EXPECT_TRUE(filter_host_->WaitForError(PIPELINE_ERROR_DECODE));
   EXPECT_FALSE(filter_host_->IsInitialized());
 }
@@ -170,7 +181,8 @@ TEST_F(FFmpegVideoDecoderTest, Initialize_FindDecoderFails) {
   EXPECT_CALL(*MockFFmpeg::get(), AVCodecFindDecoder(CODEC_ID_NONE))
       .WillOnce(ReturnNull());
 
-  EXPECT_FALSE(decoder_->Initialize(demuxer_));
+  EXPECT_TRUE(decoder_->Initialize(demuxer_));
+  message_loop_.RunAllPending();
   EXPECT_TRUE(filter_host_->WaitForError(PIPELINE_ERROR_DECODE));
   EXPECT_FALSE(filter_host_->IsInitialized());
 }
@@ -187,7 +199,8 @@ TEST_F(FFmpegVideoDecoderTest, Initialize_InitThreadFails) {
   EXPECT_CALL(*MockFFmpeg::get(), AVCodecThreadInit(&codec_context_, 2))
       .WillOnce(Return(-1));
 
-  EXPECT_FALSE(decoder_->Initialize(demuxer_));
+  EXPECT_TRUE(decoder_->Initialize(demuxer_));
+  message_loop_.RunAllPending();
   EXPECT_TRUE(filter_host_->WaitForError(PIPELINE_ERROR_DECODE));
   EXPECT_FALSE(filter_host_->IsInitialized());
 }
@@ -206,7 +219,8 @@ TEST_F(FFmpegVideoDecoderTest, Initialize_OpenDecoderFails) {
   EXPECT_CALL(*MockFFmpeg::get(), AVCodecOpen(&codec_context_, &codec_))
       .WillOnce(Return(-1));
 
-  EXPECT_FALSE(decoder_->Initialize(demuxer_));
+  EXPECT_TRUE(decoder_->Initialize(demuxer_));
+  message_loop_.RunAllPending();
   EXPECT_TRUE(filter_host_->WaitForError(PIPELINE_ERROR_DECODE));
   EXPECT_FALSE(filter_host_->IsInitialized());
 }
@@ -226,6 +240,7 @@ TEST_F(FFmpegVideoDecoderTest, Initialize_Successful) {
       .WillOnce(Return(0));
 
   EXPECT_TRUE(decoder_->Initialize(demuxer_));
+  message_loop_.RunAllPending();
   EXPECT_TRUE(filter_host_->WaitForInitialized());
   EXPECT_TRUE(filter_host_->IsInitialized());
   EXPECT_EQ(PIPELINE_OK, pipeline_->GetError());
@@ -301,7 +316,7 @@ TEST_F(FFmpegVideoDecoderTest, GetSurfaceFormat) {
   // YV12 formats.
   context.pix_fmt = PIX_FMT_YUV420P;
   EXPECT_EQ(VideoSurface::YV12, decoder->GetSurfaceFormat(context));
-  context.pix_fmt =  PIX_FMT_YUVJ420P;
+  context.pix_fmt = PIX_FMT_YUVJ420P;
   EXPECT_EQ(VideoSurface::YV12, decoder->GetSurfaceFormat(context));
 
   // YV16 formats.
@@ -379,7 +394,7 @@ TEST_F(FFmpegVideoDecoderTest, OnDecode_TestStateTransition) {
   // Simulates a input sequence of three buffers, and six decode requests to
   // exercise the state transitions, and bookkeeping logic of OnDecode.
   //
-  // We try verify the folowing:
+  // We try to verify the following:
   //   1) Non-EoS buffer timestamps are pushed into the pts_queue.
   //   2) Timestamps are popped for each decoded frame.
   //   3) The last_pts_ is updated for each decoded frame.
diff --git a/media/filters/test_video_decoder.h b/media/filters/test_video_decoder.h
index ca442bf..bb82c36 100644
--- a/media/filters/test_video_decoder.h
+++ b/media/filters/test_video_decoder.h
@@ -20,8 +20,7 @@ namespace media {
 class TestVideoDecoder : public DecoderBase<VideoDecoder, VideoFrame> {
  public:
   TestVideoDecoder()
-      : DecoderBase<VideoDecoder, VideoFrame>("VideoDecoderThread"),
-        video_width_(0),
+      : video_width_(0),
         video_height_(0) {
   }