Free demuxed audio packets if audio device fails

BUG=17481
TEST=media_unittests

Introduced a mesage broadcasting mechanism to notify filters of events from
another filter. In particular this is used to notify all filters that the
audio device has failed and special actions should be taken to handle it. In
this case, demuxer should disable the audio stream and should free the demuxed
audio packets instead of caching them. This will fix the memory leak that
happens if audio device fails during playback.

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

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

10 files changed, 161 insertions, 0 deletions

diff --git a/media/base/filter_host.h b/media/base/filter_host.h
index 65aeebb..17e7631 100644
--- a/media/base/filter_host.h
+++ b/media/base/filter_host.h
@@ -18,6 +18,7 @@
 #define MEDIA_BASE_FILTER_HOST_H_
 
 #include "base/task.h"
+#include "media/base/filters.h"
 #include "media/base/pipeline.h"
 
 namespace media {
@@ -56,6 +57,9 @@ class FilterHost {
   // Sets the flag to indicate that we are doing streaming.
   virtual void SetStreaming(bool streaming) = 0;
 
+  // Broadcast a message of type |message| to all other filters from |source|.
+  virtual void BroadcastMessage(FilterMessage message) = 0;
+
  protected:
   virtual ~FilterHost() {}
 };
diff --git a/media/base/filters.h b/media/base/filters.h
index 3faede5..9a63763 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -53,6 +53,12 @@ enum FilterType {
   FILTER_VIDEO_RENDERER
 };
 
+// A filter can broadcast messages to all other filters.  This identifies
+// the type of message to be broadcasted.
+enum FilterMessage {
+  kMsgDisableAudio,
+};
+
 // Used for completing asynchronous methods.
 typedef Callback0::Type FilterCallback;
 
@@ -124,6 +130,12 @@ class MediaFilter : public base::RefCountedThreadSafe<MediaFilter> {
     }
   }
 
+  // This method is called from the pipeline when a message of type |message|
+  // is broadcasted from |source|. Filters can ignore the message if they do
+  // not need to react to events raised from another filter.
+  virtual void OnReceivedMessage(FilterMessage message) {
+  }
+
  protected:
   // Only allow scoped_refptr<> to delete filters.
   friend class base::RefCountedThreadSafe<MediaFilter>;
diff --git a/media/base/mock_filter_host.h b/media/base/mock_filter_host.h
index a7150c2..c463541 100644
--- a/media/base/mock_filter_host.h
+++ b/media/base/mock_filter_host.h
@@ -37,6 +37,7 @@ class MockFilterHost : public FilterHost {
   MOCK_METHOD1(SetBufferedBytes, void(int64 buffered_bytes));
   MOCK_METHOD2(SetVideoSize, void(size_t width, size_t height));
   MOCK_METHOD1(SetStreaming, void(bool streamed));
+  MOCK_METHOD1(BroadcastMessage, void(FilterMessage message));
 
  private:
   DISALLOW_COPY_AND_ASSIGN(MockFilterHost);
diff --git a/media/base/mock_filters.h b/media/base/mock_filters.h
index cf5d2d5..affc0fa 100644
--- a/media/base/mock_filters.h
+++ b/media/base/mock_filters.h
@@ -96,6 +96,7 @@ class MockDataSource : public DataSource {
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(OnReceivedMessage, void(FilterMessage message));
 
   // DataSource implementation.
   MOCK_METHOD2(Initialize, void(const std::string& url,
@@ -123,6 +124,7 @@ class MockDemuxer : public Demuxer {
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(OnReceivedMessage, void(FilterMessage message));
 
   // Demuxer implementation.
   MOCK_METHOD2(Initialize, void(DataSource* data_source,
@@ -176,6 +178,7 @@ class MockVideoDecoder : public VideoDecoder {
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(OnReceivedMessage, void(FilterMessage message));
 
   // VideoDecoder implementation.
   MOCK_METHOD2(Initialize, void(DemuxerStream* stream,
@@ -209,6 +212,7 @@ class MockAudioDecoder : public AudioDecoder {
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(OnReceivedMessage, void(FilterMessage message));
 
   // AudioDecoder implementation.
   MOCK_METHOD2(Initialize, void(DemuxerStream* stream,
@@ -233,6 +237,7 @@ class MockVideoRenderer : public VideoRenderer {
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(OnReceivedMessage, void(FilterMessage message));
 
   // VideoRenderer implementation.
   MOCK_METHOD2(Initialize, void(VideoDecoder* decoder,
@@ -253,6 +258,7 @@ class MockAudioRenderer : public AudioRenderer {
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(OnReceivedMessage, void(FilterMessage message));
 
   // AudioRenderer implementation.
   MOCK_METHOD2(Initialize, void(AudioDecoder* decoder,
@@ -365,6 +371,12 @@ ACTION_P2(SetBufferedBytes, filter, bytes) {
   filter->host()->SetBufferedBytes(bytes);
 }
 
+// Helper gmock action that calls BroadcastMessage() on behalf of the provided
+// filter.
+ACTION_P2(BroadcastMessage, filter, message) {
+  filter->host()->BroadcastMessage(message);
+}
+
 }  // namespace media
 
 #endif  // MEDIA_BASE_MOCK_FILTERS_H_
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index cc1c3b2..ccfce70 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -374,6 +374,15 @@ void PipelineImpl::SetStreaming(bool streaming) {
   streaming_ = streaming;
 }
 
+void PipelineImpl::BroadcastMessage(FilterMessage message) {
+  DCHECK(IsRunning());
+
+  // Broadcast the message on the message loop.
+  message_loop_->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &PipelineImpl::BroadcastMessageTask,
+                        message));
+}
+
 void PipelineImpl::InsertRenderedMimeType(const std::string& major_mime_type) {
   DCHECK(IsRunning());
   AutoLock auto_lock(lock_);
@@ -630,6 +639,17 @@ void PipelineImpl::SeekTask(base::TimeDelta time,
       NewCallback(this, &PipelineImpl::OnFilterStateTransition));
 }
 
+void PipelineImpl::BroadcastMessageTask(FilterMessage message) {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+
+  // Broadcast the message to all filters.
+  for (FilterVector::iterator iter = filters_.begin();
+       iter != filters_.end();
+       ++iter) {
+    (*iter)->OnReceivedMessage(message);
+  }
+}
+
 void PipelineImpl::FilterStateTransitionTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index aee1f59..4332cba 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -136,6 +136,7 @@ class PipelineImpl : public Pipeline, public FilterHost {
   virtual void SetBufferedBytes(int64 buffered_bytes);
   virtual void SetVideoSize(size_t width, size_t height);
   virtual void SetStreaming(bool streamed);
+  virtual void BroadcastMessage(FilterMessage message);
 
   // Method called during initialization to insert a mime type into the
   // |rendered_mime_types_| set.
@@ -180,6 +181,9 @@ class PipelineImpl : public Pipeline, public FilterHost {
   // Carries out notifying filters that we are seeking to a new timestamp.
   void SeekTask(base::TimeDelta time, PipelineCallback* seek_callback);
 
+  // Carries out message broadcasting on the message loop.
+  void BroadcastMessageTask(FilterMessage message);
+
   // Carries out advancing to the next filter during Play()/Pause()/Seek().
   void FilterStateTransitionTask();
 
diff --git a/media/base/pipeline_impl_unittest.cc b/media/base/pipeline_impl_unittest.cc
index 6400049..f3c0297 100644
--- a/media/base/pipeline_impl_unittest.cc
+++ b/media/base/pipeline_impl_unittest.cc
@@ -439,4 +439,45 @@ TEST_F(PipelineImplTest, Properties) {
             pipeline_->GetBufferedTime().ToInternalValue());
 }
 
+TEST_F(PipelineImplTest, BroadcastMessage) {
+  scoped_refptr<StrictMock<MockDemuxerStream> > audio_stream =
+      new StrictMock<MockDemuxerStream>("audio/x-foo");
+  scoped_refptr<StrictMock<MockDemuxerStream> > video_stream =
+      new StrictMock<MockDemuxerStream>("video/x-foo");
+  MockDemuxerStreamVector streams;
+  streams.push_back(audio_stream);
+  streams.push_back(video_stream);
+
+  InitializeDataSource();
+  InitializeDemuxer(&streams, base::TimeDelta());
+  InitializeAudioDecoder(audio_stream);
+  InitializeAudioRenderer();
+  InitializeVideoDecoder(video_stream);
+  InitializeVideoRenderer();
+
+  InitializePipeline();
+  EXPECT_TRUE(pipeline_->IsInitialized());
+  EXPECT_EQ(PIPELINE_OK, pipeline_->GetError());
+  EXPECT_TRUE(pipeline_->IsRendered(mime_type::kMajorTypeAudio));
+  EXPECT_TRUE(pipeline_->IsRendered(mime_type::kMajorTypeVideo));
+
+  EXPECT_CALL(*mocks_->audio_renderer(), SetPlaybackRate(1.0f))
+      .WillOnce(BroadcastMessage(mocks_->audio_renderer(),
+                                 kMsgDisableAudio));
+  EXPECT_CALL(*mocks_->data_source(),
+              OnReceivedMessage(kMsgDisableAudio));
+  EXPECT_CALL(*mocks_->demuxer(),
+              OnReceivedMessage(kMsgDisableAudio));
+  EXPECT_CALL(*mocks_->audio_decoder(),
+              OnReceivedMessage(kMsgDisableAudio));
+  EXPECT_CALL(*mocks_->audio_renderer(),
+              OnReceivedMessage(kMsgDisableAudio));
+  EXPECT_CALL(*mocks_->video_decoder(),
+              OnReceivedMessage(kMsgDisableAudio));
+  EXPECT_CALL(*mocks_->video_renderer(),
+              OnReceivedMessage(kMsgDisableAudio));
+
+  mocks_->audio_renderer()->SetPlaybackRate(1.0f);
+}
+
 }  // namespace media
diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index 00e6531..20f4194 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -279,6 +279,13 @@ void FFmpegDemuxer::Seek(base::TimeDelta time, FilterCallback* callback) {
       NewRunnableMethod(this, &FFmpegDemuxer::SeekTask, time, callback));
 }
 
+void FFmpegDemuxer::OnReceivedMessage(FilterMessage message) {
+  if (message == kMsgDisableAudio) {
+    message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &FFmpegDemuxer::DisableAudioStreamTask));
+  }
+}
+
 void FFmpegDemuxer::Initialize(DataSource* data_source,
                                FilterCallback* callback) {
   message_loop()->PostTask(FROM_HERE,
@@ -527,6 +534,24 @@ void FFmpegDemuxer::StopTask() {
   }
 }
 
+void FFmpegDemuxer::DisableAudioStreamTask() {
+  DCHECK_EQ(MessageLoop::current(), message_loop());
+
+  StreamVector::iterator iter;
+  for (size_t i = 0; i < packet_streams_.size(); ++i) {
+    if (!packet_streams_[i])
+      continue;
+
+    // If the codec type is audio, remove the reference. DemuxTask() will
+    // look for such reference, and this will result in deleting the
+    // audio packets after they are demuxed.
+    if (packet_streams_[i]->GetAVStream()->codec->codec_type ==
+        CODEC_TYPE_AUDIO) {
+      packet_streams_[i] = NULL;
+    }
+  }
+}
+
 bool FFmpegDemuxer::StreamsHavePendingReads() {
   DCHECK_EQ(MessageLoop::current(), message_loop());
   StreamVector::iterator iter;
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index c897ca4..7ab9449 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -125,6 +125,7 @@ class FFmpegDemuxer : public Demuxer,
   // MediaFilter implementation.
   virtual void Stop();
   virtual void Seek(base::TimeDelta time, FilterCallback* callback);
+  virtual void OnReceivedMessage(FilterMessage message);
 
   // Demuxer implementation.
   virtual void Initialize(DataSource* data_source, FilterCallback* callback);
@@ -159,6 +160,9 @@ class FFmpegDemuxer : public Demuxer,
   // Carries out stopping the demuxer streams on the demuxer thread.
   void StopTask();
 
+  // Carries out disabling the audio stream on the demuxer thread.
+  void DisableAudioStreamTask();
+
   // Returns true if any of the streams have pending reads.  Since we lazily
   // post a DemuxTask() for every read, we use this method to quickly terminate
   // the tasks if there is no work to do.
diff --git a/media/filters/ffmpeg_demuxer_unittest.cc b/media/filters/ffmpeg_demuxer_unittest.cc
index 044e8ea..a026a87c 100644
--- a/media/filters/ffmpeg_demuxer_unittest.cc
+++ b/media/filters/ffmpeg_demuxer_unittest.cc
@@ -679,6 +679,44 @@ TEST_F(FFmpegDemuxerTest, Stop) {
   MockFFmpeg::get()->CheckPoint(1);
 }
 
+TEST_F(FFmpegDemuxerTest, DisableAudioStream) {
+  // We are doing the following things here:
+  // 1. Initialize the demuxer with audio and video stream.
+  // 2. Send a "disable audio stream" message to the demuxer.
+  // 3. Demuxer will free audio packets even if audio stream was initialized.
+  {
+    SCOPED_TRACE("");
+    InitializeDemuxer();
+  }
+
+  // Submit a "disable audio stream" message to the demuxer.
+  demuxer_->OnReceivedMessage(kMsgDisableAudio);
+  message_loop_.RunAllPending();
+
+  // Expect all calls in sequence.
+  InSequence s;
+
+  // The demuxer will read an audio packet which will get immediately freed.
+  EXPECT_CALL(*MockFFmpeg::get(), AVReadFrame(&format_context_, _))
+      .WillOnce(CreatePacket(AV_STREAM_AUDIO, kNullData, 0));
+  EXPECT_CALL(*MockFFmpeg::get(), AVFreePacket(_)).WillOnce(FreePacket());
+
+  // Then an end-of-stream packet is read.
+  EXPECT_CALL(*MockFFmpeg::get(), AVReadFrame(&format_context_, _))
+      .WillOnce(Return(AVERROR_IO));
+  EXPECT_CALL(*MockFFmpeg::get(), AVFreePacket(_));
+  EXPECT_CALL(*MockFFmpeg::get(), AVFreePacket(_));
+
+  // Get our streams.
+  scoped_refptr<DemuxerStream> video = demuxer_->GetStream(DS_STREAM_VIDEO);
+  ASSERT_TRUE(video);
+
+  // Attempt a read from the video stream and run the message loop until done.
+  scoped_refptr<DemuxerStreamReader> reader(new DemuxerStreamReader());
+  reader->Read(video);
+  message_loop_.RunAllPending();
+}
+
 class MockFFmpegDemuxer : public FFmpegDemuxer {
  public:
   MockFFmpegDemuxer() {}