Move FFmpegDemuxer initialization onto the demuxer thread.

We now have all FFmpeg function calls executing on the demuxer thread, meaning DataSource will now always receive IO operations on the demuxer thread.

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

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

4 files changed, 80 insertions, 33 deletions

diff --git a/media/base/mock_filter_host.h b/media/base/mock_filter_host.h
index 0a41397..8bef47e 100644
--- a/media/base/mock_filter_host.h
+++ b/media/base/mock_filter_host.h
@@ -13,6 +13,7 @@
 #include <string>
 
 #include "base/scoped_ptr.h"
+#include "base/waitable_event.h"
 #include "media/base/factory.h"
 #include "media/base/filter_host.h"
 #include "media/base/filters.h"
@@ -28,7 +29,10 @@ class MockFilterHost : public FilterHost {
   MockFilterHost(MockPipeline* mock_pipeline, Filter* filter)
       : mock_pipeline_(mock_pipeline),
         filter_(filter),
-        initialized_(false) {
+        initialized_(false),
+        error_(PIPELINE_OK),
+        wait_for_initialized_(false, false),
+        wait_for_error_(false, false) {
     EXPECT_TRUE(mock_pipeline_);
     EXPECT_TRUE(filter_);
     filter_->SetFilterHost(this);
@@ -51,6 +55,7 @@ class MockFilterHost : public FilterHost {
   virtual void InitializationComplete() {
     EXPECT_FALSE(initialized_);
     initialized_ = true;
+    wait_for_initialized_.Signal();
   }
 
   virtual void PostTask(Task* task) {
@@ -58,7 +63,9 @@ class MockFilterHost : public FilterHost {
   }
 
   virtual void Error(PipelineError error) {
+    error_ = error;
     mock_pipeline_->Error(error);
+    wait_for_error_.Signal();
   }
 
   virtual void SetTime(base::TimeDelta time) {
@@ -98,6 +105,26 @@ class MockFilterHost : public FilterHost {
     return initialized_;
   }
 
+  bool WaitForInitialized() {
+    const base::TimeDelta kTimedWait = base::TimeDelta::FromMilliseconds(500);
+    while (!initialized_) {
+      if (!wait_for_initialized_.TimedWait(kTimedWait)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  bool WaitForError(PipelineError error) {
+    const base::TimeDelta kTimedWait = base::TimeDelta::FromMilliseconds(500);
+    while (error_ != error) {
+      if (!wait_for_error_.TimedWait(kTimedWait)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
  private:
   MockPipeline* mock_pipeline_;
   scoped_refptr<Filter> filter_;
@@ -109,6 +136,13 @@ class MockFilterHost : public FilterHost {
   // Tracks if the filter has executed InitializationComplete().
   bool initialized_;
 
+  // Tracks the last pipeline error set by the filter.
+  PipelineError error_;
+
+  // Allows unit tests to wait for particular conditions before asserting.
+  base::WaitableEvent wait_for_initialized_;
+  base::WaitableEvent wait_for_error_;
+
   DISALLOW_COPY_AND_ASSIGN(MockFilterHost);
 };
 
diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index f3e4b79..08433b5 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -196,8 +196,8 @@ FFmpegDemuxer::~FFmpegDemuxer() {
 }
 
 void FFmpegDemuxer::PostDemuxTask() {
-  thread_.message_loop()->PostTask(
-      FROM_HERE, NewRunnableMethod(this, &FFmpegDemuxer::DemuxTask));
+  thread_.message_loop()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &FFmpegDemuxer::DemuxTask));
 }
 
 void FFmpegDemuxer::Stop() {
@@ -205,11 +205,33 @@ void FFmpegDemuxer::Stop() {
 }
 
 void FFmpegDemuxer::Seek(base::TimeDelta time) {
-  thread_.message_loop()->PostTask(
-      FROM_HERE, NewRunnableMethod(this, &FFmpegDemuxer::SeekTask, time));
+  thread_.message_loop()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &FFmpegDemuxer::SeekTask, time));
 }
 
 bool FFmpegDemuxer::Initialize(DataSource* data_source) {
+  // Start our internal demuxing thread.
+  if (!thread_.Start()) {
+    host_->Error(DEMUXER_ERROR_COULD_NOT_CREATE_THREAD);
+    return false;
+  }
+
+  thread_.message_loop()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &FFmpegDemuxer::InititalizeTask, data_source));
+  return true;
+}
+
+size_t FFmpegDemuxer::GetNumberOfStreams() {
+  return streams_.size();
+}
+
+scoped_refptr<DemuxerStream> FFmpegDemuxer::GetStream(int stream) {
+  DCHECK(stream >= 0);
+  DCHECK(stream < static_cast<int>(streams_.size()));
+  return streams_[stream].get();
+}
+
+void FFmpegDemuxer::InititalizeTask(DataSource* data_source) {
   // In order to get FFmpeg to use |data_source| for file IO we must transfer
   // ownership via FFmpegGlue.  We'll add |data_source| to FFmpegGlue and pass
   // the resulting key to FFmpeg.  FFmpeg will pass the key to FFmpegGlue which
@@ -232,7 +254,7 @@ bool FFmpegDemuxer::Initialize(DataSource* data_source) {
 
   if (result < 0) {
     host_->Error(DEMUXER_ERROR_COULD_NOT_OPEN);
-    return false;
+    return;
   }
 
   // Assign to our scoped_ptr_malloc.
@@ -243,7 +265,7 @@ bool FFmpegDemuxer::Initialize(DataSource* data_source) {
   result = av_find_stream_info(format_context_.get());
   if (result < 0) {
     host_->Error(DEMUXER_ERROR_COULD_NOT_PARSE);
-    return false;
+    return;
   }
 
   // Create demuxer streams for all supported streams.
@@ -261,29 +283,12 @@ bool FFmpegDemuxer::Initialize(DataSource* data_source) {
   }
   if (streams_.empty()) {
     host_->Error(DEMUXER_ERROR_NO_SUPPORTED_STREAMS);
-    return false;
-  }
-
-  // We have some streams to demux so create our thread.
-  if (!thread_.Start()) {
-    host_->Error(DEMUXER_ERROR_COULD_NOT_CREATE_THREAD);
-    return false;
+    return;
   }
 
   // Good to go: set the duration and notify we're done initializing.
   host_->SetDuration(max_duration);
   host_->InitializationComplete();
-  return true;
-}
-
-size_t FFmpegDemuxer::GetNumberOfStreams() {
-  return streams_.size();
-}
-
-scoped_refptr<DemuxerStream> FFmpegDemuxer::GetStream(int stream) {
-  DCHECK(stream >= 0);
-  DCHECK(stream < static_cast<int>(streams_.size()));
-  return streams_[stream].get();
 }
 
 void FFmpegDemuxer::SeekTask(base::TimeDelta time) {
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index 72866af..2c113be 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -127,6 +127,9 @@ class FFmpegDemuxer : public Demuxer {
   FFmpegDemuxer();
   virtual ~FFmpegDemuxer();
 
+  // Carries out initialization on the demuxer thread.
+  void InititalizeTask(DataSource* data_source);
+
   // Carries out a seek on the demuxer thread.
   void SeekTask(base::TimeDelta time);
 
diff --git a/media/filters/ffmpeg_demuxer_unittest.cc b/media/filters/ffmpeg_demuxer_unittest.cc
index d9cabfa..8e7d3f3 100644
--- a/media/filters/ffmpeg_demuxer_unittest.cc
+++ b/media/filters/ffmpeg_demuxer_unittest.cc
@@ -279,37 +279,40 @@ TEST(FFmpegDemuxerTest, InitializeFailure) {
   // Simulate av_open_input_fail failing.
   g_av_open_input_file = AVERROR_IO;
   g_av_find_stream_info = 0;
-  EXPECT_FALSE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(filter_host->WaitForError(DEMUXER_ERROR_COULD_NOT_OPEN));
   EXPECT_FALSE(filter_host->IsInitialized());
-  EXPECT_EQ(DEMUXER_ERROR_COULD_NOT_OPEN, pipeline.GetError());
 
   // Simulate av_find_stream_info failing.
   g_av_open_input_file = 0;
   g_av_find_stream_info = AVERROR_IO;
+  pipeline.Reset(false);
   demuxer = factory->Create<Demuxer>(media_format);
   filter_host.reset(new MockFilterHost<Demuxer>(&pipeline, demuxer));
-  EXPECT_FALSE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(filter_host->WaitForError(DEMUXER_ERROR_COULD_NOT_PARSE));
   EXPECT_FALSE(filter_host->IsInitialized());
-  EXPECT_EQ(DEMUXER_ERROR_COULD_NOT_PARSE, pipeline.GetError());
 
   // Simulate media with no parseable streams.
   InitializeFFmpegMocks();
+  pipeline.Reset(false);
   demuxer = factory->Create<Demuxer>(media_format);
   filter_host.reset(new MockFilterHost<Demuxer>(&pipeline, demuxer));
-  EXPECT_FALSE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(filter_host->WaitForError(DEMUXER_ERROR_NO_SUPPORTED_STREAMS));
   EXPECT_FALSE(filter_host->IsInitialized());
-  EXPECT_EQ(DEMUXER_ERROR_NO_SUPPORTED_STREAMS, pipeline.GetError());
 
   // Simulate media with a data stream but no audio or video streams.
   g_format.nb_streams = 1;
   g_format.streams[0] = &g_streams[0];
   g_streams[0].codec = &g_data_codec;
   g_streams[0].duration = 10;
+  pipeline.Reset(false);
   demuxer = factory->Create<Demuxer>(media_format);
   filter_host.reset(new MockFilterHost<Demuxer>(&pipeline, demuxer));
-  EXPECT_FALSE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(filter_host->WaitForError(DEMUXER_ERROR_NO_SUPPORTED_STREAMS));
   EXPECT_FALSE(filter_host->IsInitialized());
-  EXPECT_EQ(DEMUXER_ERROR_NO_SUPPORTED_STREAMS, pipeline.GetError());
 }
 
 TEST(FFmpegDemuxerTest, InitializeStreams) {
@@ -343,6 +346,7 @@ TEST(FFmpegDemuxerTest, InitializeStreams) {
   EXPECT_TRUE(demuxer);
   MockFilterHost<Demuxer> filter_host_b(&pipeline, demuxer);
   EXPECT_TRUE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(filter_host_b.WaitForInitialized());
   EXPECT_TRUE(filter_host_b.IsInitialized());
   EXPECT_EQ(PIPELINE_OK, pipeline.GetError());
 
@@ -434,6 +438,7 @@ TEST(FFmpegDemuxerTest, ReadAndSeek) {
   EXPECT_TRUE(demuxer);
   MockFilterHost<Demuxer> filter_host_b(&pipeline, demuxer);
   EXPECT_TRUE(demuxer->Initialize(data_source));
+  EXPECT_TRUE(filter_host_b.WaitForInitialized());
   EXPECT_TRUE(filter_host_b.IsInitialized());
   EXPECT_EQ(PIPELINE_OK, pipeline.GetError());