FFmpegDemuxerStream::Read() now functions properly while stopped.

BUG=13907
TEST=some layout tests might start passing

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

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

3 files changed, 154 insertions, 29 deletions

diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index d4c7fc55..2e8a4c6 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -3,6 +3,7 @@
 // found in the LICENSE file.
 
 #include "base/scoped_ptr.h"
+#include "base/stl_util-inl.h"
 #include "base/string_util.h"
 #include "base/time.h"
 #include "media/base/filter_host.h"
@@ -10,6 +11,25 @@
 #include "media/filters/ffmpeg_demuxer.h"
 #include "media/filters/ffmpeg_glue.h"
 
+namespace {
+
+// Helper function to deep copy an AVPacket's data, size and timestamps.
+// Returns NULL if a packet could not be cloned (i.e., out of memory).
+AVPacket* ClonePacket(AVPacket* packet) {
+  scoped_ptr<AVPacket> clone(new AVPacket());
+  if (!clone.get() || av_new_packet(clone.get(), packet->size) < 0) {
+    return NULL;
+  }
+  DCHECK_EQ(clone->size, packet->size);
+  clone->dts = packet->dts;
+  clone->pts = packet->pts;
+  clone->duration = packet->duration;
+  memcpy(clone->data, packet->data, clone->size);
+  return clone.release();
+}
+
+}  // namespace
+
 namespace media {
 
 //
@@ -51,7 +71,8 @@ FFmpegDemuxerStream::FFmpegDemuxerStream(FFmpegDemuxer* demuxer,
                                          AVStream* stream)
     : demuxer_(demuxer),
       stream_(stream),
-      discontinuous_(false) {
+      discontinuous_(false),
+      stopped_(false) {
   DCHECK(demuxer_);
 
   // Determine our media format.
@@ -74,11 +95,10 @@ FFmpegDemuxerStream::FFmpegDemuxerStream(FFmpegDemuxer* demuxer,
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
-  // Since |buffer_queue_| uses scoped_refptr everything will get released.
-  while (!read_queue_.empty()) {
-    delete read_queue_.front();
-    read_queue_.pop_front();
-  }
+  AutoLock auto_lock(lock_);
+  DCHECK(stopped_);
+  DCHECK(read_queue_.empty());
+  DCHECK(buffer_queue_.empty());
 }
 
 void* FFmpegDemuxerStream::QueryInterface(const char* id) {
@@ -98,10 +118,15 @@ bool FFmpegDemuxerStream::HasPendingReads() {
 base::TimeDelta FFmpegDemuxerStream::EnqueuePacket(AVPacket* packet) {
   base::TimeDelta timestamp = ConvertTimestamp(packet->pts);
   base::TimeDelta duration = ConvertTimestamp(packet->duration);
-  Buffer* buffer = new AVPacketBuffer(packet, timestamp, duration);
+  scoped_refptr<Buffer> buffer =
+      new AVPacketBuffer(packet, timestamp, duration);
   DCHECK(buffer);
   {
     AutoLock auto_lock(lock_);
+    if (stopped_) {
+      NOTREACHED() << "Attempted to enqueue packet on a stopped stream.";
+      return timestamp;
+    }
     buffer_queue_.push_back(buffer);
   }
   FulfillPendingReads();
@@ -114,6 +139,21 @@ void FFmpegDemuxerStream::FlushBuffers() {
   discontinuous_ = true;
 }
 
+void FFmpegDemuxerStream::Stop() {
+  // Since |buffer_queue_| can be very large, we swap queues and delete outside
+  // of the lock.
+  BufferQueue tmp_buffer_queue;
+  ReadQueue tmp_read_queue;
+  {
+    AutoLock auto_lock(lock_);
+    buffer_queue_.swap(tmp_buffer_queue);
+    read_queue_.swap(tmp_read_queue);
+    stopped_ = true;
+  }
+  tmp_buffer_queue.clear();
+  STLDeleteElements(&tmp_read_queue);
+}
+
 const MediaFormat& FFmpegDemuxerStream::media_format() {
   return media_format_;
 }
@@ -122,9 +162,28 @@ void FFmpegDemuxerStream::Read(Callback1<Buffer*>::Type* read_callback) {
   DCHECK(read_callback);
   {
     AutoLock auto_lock(lock_);
+    // Don't accept any additional reads if we've been told to stop.
+    //
+    // TODO(scherkus): it would be cleaner if we replied with an error message.
+    if (stopped_) {
+      delete read_callback;
+      return;
+    }
+
+    // Otherwise enqueue the request.
     read_queue_.push_back(read_callback);
   }
-  if (FulfillPendingReads()) {
+
+  // See if we can immediately fulfill this read and whether we need to demux.
+  bool post_demux_task = FulfillPendingReads();
+
+  // Since Read() may be executed on any thread, it's possible that StopTask()
+  // finishes executing on the demuxer thread and the message loop goes away.
+  //
+  // To prevent that we'll grab the lock and post a task at the same time, which
+  // will keep the message loop alive.
+  AutoLock auto_lock(lock_);
+  if (post_demux_task && !stopped_) {
     demuxer_->PostDemuxTask();
   }
 }
@@ -162,7 +221,6 @@ base::TimeDelta FFmpegDemuxerStream::ConvertTimestamp(int64 timestamp) {
   return base::TimeDelta::FromMicroseconds(microseconds);
 }
 
-
 //
 // FFmpegDemuxer
 //
@@ -186,7 +244,12 @@ void FFmpegDemuxer::PostDemuxTask() {
 }
 
 void FFmpegDemuxer::Stop() {
-  thread_.Stop();
+  // Stop our thread and post a task notifying the streams to stop as well.
+  if (thread_.IsRunning()) {
+    thread_.message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &FFmpegDemuxer::StopTask));
+    thread_.Stop();
+  }
   format_context_.reset();
 }
 
@@ -364,7 +427,15 @@ void FFmpegDemuxer::DemuxTask() {
   }
 }
 
+void FFmpegDemuxer::StopTask() {
+  StreamVector::iterator iter;
+  for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
+    (*iter)->Stop();
+  }
+}
+
 bool FFmpegDemuxer::StreamsHavePendingReads() {
+  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if ((*iter)->HasPendingReads()) {
@@ -375,6 +446,7 @@ bool FFmpegDemuxer::StreamsHavePendingReads() {
 }
 
 void FFmpegDemuxer::StreamHasEnded() {
+  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     AVPacket* packet = new AVPacket();
@@ -383,17 +455,4 @@ void FFmpegDemuxer::StreamHasEnded() {
   }
 }
 
-AVPacket* FFmpegDemuxer::ClonePacket(AVPacket* packet) {
-  scoped_ptr<AVPacket> clone(new AVPacket());
-  if (!clone.get() || av_new_packet(clone.get(), packet->size) < 0) {
-    return NULL;
-  }
-  DCHECK_EQ(clone->size, packet->size);
-  clone->dts = packet->dts;
-  clone->pts = packet->pts;
-  clone->duration = packet->duration;
-  memcpy(clone->data, packet->data, clone->size);
-  return clone.release();
-}
-
 }  // namespace media
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index 509333a..2ffbf50 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -14,6 +14,10 @@
 // NOTE: since FFmpegDemuxer reads packets sequentially without seeking, media
 // files with very large drift between audio/video streams may result in
 // excessive memory consumption.
+//
+// When stopped, FFmpegDemuxer and FFmpegDemuxerStream release all callbacks
+// and buffered packets and shuts down its internal thread.  Reads from a
+// stopped FFmpegDemuxerStream will not be replied to.
 
 #ifndef MEDIA_FILTERS_FFMPEG_DEMUXER_H_
 #define MEDIA_FILTERS_FFMPEG_DEMUXER_H_
@@ -61,9 +65,12 @@ class FFmpegDemuxerStream : public DemuxerStream, public AVStreamProvider {
   // of the enqueued packet.
   base::TimeDelta EnqueuePacket(AVPacket* packet);
 
-  // Signals to empty queue and mark next packet as discontinuous.
+  // Signals to empty the buffer queue and mark next packet as discontinuous.
   void FlushBuffers();
 
+  // Empties the queues and ignores any additional calls to Read().
+  void Stop();
+
   // Returns the duration of this stream.
   base::TimeDelta duration() { return duration_; }
 
@@ -89,6 +96,7 @@ class FFmpegDemuxerStream : public DemuxerStream, public AVStreamProvider {
   MediaFormat media_format_;
   base::TimeDelta duration_;
   bool discontinuous_;
+  bool stopped_;
 
   Lock lock_;
 
@@ -138,20 +146,21 @@ class FFmpegDemuxer : public Demuxer {
   // Carries out demuxing and satisfying stream reads on the demuxer thread.
   void DemuxTask();
 
+  // Carries out stopping the demuxer streams on the demuxer thread.
+  void StopTask();
+
   // 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.
   //
-  // Safe to call on any thread.
+  // Must be called on the demuxer thread.
   bool StreamsHavePendingReads();
 
   // Signal all FFmpegDemuxerStream that the stream has ended.
+  //
+  // Must be called on the demuxer thread.
   void StreamHasEnded();
 
-  // Helper function to deep copy an AVPacket's data, size and timestamps.
-  // Returns NULL if a packet could not be cloned (i.e., out of memory).
-  AVPacket* ClonePacket(AVPacket* packet);
-
   // FFmpeg context handle.
   scoped_ptr_malloc<AVFormatContext, ScopedPtrAVFree> format_context_;
 
@@ -166,6 +175,9 @@ class FFmpegDemuxer : public Demuxer {
   //     representing unsupported streams where we throw away the data.
   //
   // Ownership is handled via reference counting.
+  //
+  // Once initialized, operations on FFmpegDemuxerStreams should be carried out
+  // on the demuxer thread.
   typedef std::vector< scoped_refptr<FFmpegDemuxerStream> > StreamVector;
   StreamVector streams_;
   StreamVector packet_streams_;
diff --git a/media/filters/ffmpeg_demuxer_unittest.cc b/media/filters/ffmpeg_demuxer_unittest.cc
index df427f8..619851c 100644
--- a/media/filters/ffmpeg_demuxer_unittest.cc
+++ b/media/filters/ffmpeg_demuxer_unittest.cc
@@ -21,6 +21,7 @@ using ::testing::DoAll;
 using ::testing::InSequence;
 using ::testing::Return;
 using ::testing::SetArgumentPointee;
+using ::testing::StrictMock;
 
 namespace media {
 
@@ -613,4 +614,57 @@ TEST_F(FFmpegDemuxerTest, MP3Hack) {
   MockFFmpeg::get()->CheckPoint(2);
 }
 
+// A mocked callback specialization for calling Read().  Since RunWithParams()
+// is mocked we don't need to pass in object or method pointers.
+typedef CallbackImpl<FFmpegDemuxerTest, void (FFmpegDemuxerTest::*)(Buffer*),
+                     Tuple1<Buffer*> > ReadCallback;
+class MockReadCallback : public ReadCallback {
+ public:
+  MockReadCallback()
+      : ReadCallback(NULL, NULL) {
+  }
+
+  virtual ~MockReadCallback() {
+    OnDelete();
+  }
+
+  MOCK_METHOD0(OnDelete, void());
+  MOCK_METHOD1(RunWithParams, void(const Tuple1<Buffer*>& params));
+};
+
+TEST_F(FFmpegDemuxerTest, Stop) {
+  // Tests that calling Read() on a stopped demuxer immediately deletes the
+  // callback.
+  {
+    SCOPED_TRACE("");
+    InitializeDemuxer();
+  }
+
+  // Create our mocked callback.  The demuxer will take ownership of this
+  // pointer.
+  scoped_ptr<StrictMock<MockReadCallback> > callback(
+      new StrictMock<MockReadCallback>());
+
+  // Get our stream.
+  scoped_refptr<DemuxerStream> audio = demuxer_->GetStream(DS_STREAM_AUDIO);
+  ASSERT_TRUE(audio);
+
+  // Stop the demuxer.
+  demuxer_->Stop();
+
+  // Expect all calls in sequence.
+  InSequence s;
+
+  // The callback should be immediately deleted.  We'll use a checkpoint to
+  // verify that it has indeed been deleted.
+  EXPECT_CALL(*callback, OnDelete());
+  EXPECT_CALL(*MockFFmpeg::get(), CheckPoint(1));
+
+  // Attempt the read...
+  audio->Read(callback.release());
+
+  // ...and verify that |callback| was deleted.
+  MockFFmpeg::get()->CheckPoint(1);
+}
+
 }  // namespace media