Make buffer_queue_ and read_queue_ from DemuxerStream shared resources.

Added lock calls around buffer_queue_, read_queue_, and stopped_ in demuxer stream. This is so the Read call will not block if the DemuxTask is blocked and the buffer queue is not empty.

BUG=75483
TEST=Play a live WebM stream and make sure the AV sync is good and
the video plays back smoothly.

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

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

2 files changed, 43 insertions, 4 deletions

diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index b7d53bc..9387668 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -80,6 +80,7 @@ FFmpegDemuxerStream::FFmpegDemuxerStream(FFmpegDemuxer* demuxer,
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
+  base::AutoLock auto_lock(lock_);
   DCHECK(stopped_);
   DCHECK(read_queue_.empty());
   DCHECK(buffer_queue_.empty());
@@ -87,6 +88,7 @@ FFmpegDemuxerStream::~FFmpegDemuxerStream() {
 
 bool FFmpegDemuxerStream::HasPendingReads() {
   DCHECK_EQ(MessageLoop::current(), demuxer_->message_loop());
+  base::AutoLock auto_lock(lock_);
   DCHECK(!stopped_ || read_queue_.empty())
       << "Read queue should have been emptied if demuxing stream is stopped";
   return !read_queue_.empty();
@@ -99,6 +101,7 @@ void FFmpegDemuxerStream::EnqueuePacket(AVPacket* packet) {
   base::TimeDelta duration =
       ConvertStreamTimestamp(stream_->time_base, packet->duration);
 
+  base::AutoLock auto_lock(lock_);
   if (stopped_) {
     NOTREACHED() << "Attempted to enqueue packet on a stopped stream";
     return;
@@ -124,12 +127,14 @@ void FFmpegDemuxerStream::EnqueuePacket(AVPacket* packet) {
 
 void FFmpegDemuxerStream::FlushBuffers() {
   DCHECK_EQ(MessageLoop::current(), demuxer_->message_loop());
+  base::AutoLock auto_lock(lock_);
   DCHECK(read_queue_.empty()) << "Read requests should be empty";
   buffer_queue_.clear();
 }
 
 void FFmpegDemuxerStream::Stop() {
   DCHECK_EQ(MessageLoop::current(), demuxer_->message_loop());
+  base::AutoLock auto_lock(lock_);
   buffer_queue_.clear();
   STLDeleteElements(&read_queue_);
   stopped_ = true;
@@ -149,13 +154,31 @@ const MediaFormat& FFmpegDemuxerStream::media_format() {
 
 void FFmpegDemuxerStream::Read(Callback1<Buffer*>::Type* read_callback) {
   DCHECK(read_callback);
-  demuxer_->message_loop()->PostTask(FROM_HERE,
+
+  base::AutoLock auto_lock(lock_);
+  if (!buffer_queue_.empty()) {
+    scoped_ptr<Callback1<Buffer*>::Type> read_callback_deleter(read_callback);
+
+    // Dequeue a buffer send back.
+    scoped_refptr<Buffer> buffer = buffer_queue_.front();
+    buffer_queue_.pop_front();
+
+    // Execute the callback.
+    read_callback_deleter->Run(buffer);
+
+    if (!read_queue_.empty())
+      demuxer_->PostDemuxTask();
+
+  } else {
+    demuxer_->message_loop()->PostTask(FROM_HERE,
       NewRunnableMethod(this, &FFmpegDemuxerStream::ReadTask, read_callback));
+  }
 }
 
 void FFmpegDemuxerStream::ReadTask(Callback1<Buffer*>::Type* read_callback) {
   DCHECK_EQ(MessageLoop::current(), demuxer_->message_loop());
 
+  base::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.
@@ -168,14 +191,15 @@ void FFmpegDemuxerStream::ReadTask(Callback1<Buffer*>::Type* read_callback) {
   read_queue_.push_back(read_callback);
   FulfillPendingRead();
 
-  // There are still pending reads, demux some more.
-  if (HasPendingReads()) {
+  // Check if there are still pending reads, demux some more.
+  if (!read_queue_.empty()) {
     demuxer_->PostDemuxTask();
   }
 }
 
 void FFmpegDemuxerStream::FulfillPendingRead() {
   DCHECK_EQ(MessageLoop::current(), demuxer_->message_loop());
+  lock_.AssertAcquired();
   if (buffer_queue_.empty() || read_queue_.empty()) {
     return;
   }
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index e65cb34..674abca 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -74,6 +74,13 @@ class FFmpegDemuxerStream : public DemuxerStream {
   // DemuxerStream implementation.
   virtual Type type();
   virtual const MediaFormat& media_format();
+
+  // If |buffer_queue_| is not empty will execute on caller's thread, otherwise
+  // will post ReadTask to execute on demuxer's thread. Read will acquire
+  // |lock_| for the life of the function so that means |read_callback| must
+  // not make calls into FFmpegDemuxerStream directly or that may cause a
+  // deadlock. |read_callback| should execute as quickly as possible because
+  // |lock_| is held throughout the life of the callback.
   virtual void Read(Callback1<Buffer*>::Type* read_callback);
   // Bitstream converter to convert input packet.
   virtual void EnableBitstreamConverter();
@@ -86,7 +93,8 @@ class FFmpegDemuxerStream : public DemuxerStream {
   void ReadTask(Callback1<Buffer*>::Type* read_callback);
 
   // Attempts to fulfill a single pending read by dequeueing a buffer and read
-  // callback pair and executing the callback.
+  // callback pair and executing the callback. The calling function must
+  // acquire |lock_| before calling this function.
   void FulfillPendingRead();
 
   // Converts an FFmpeg stream timestamp into a base::TimeDelta.
@@ -110,6 +118,13 @@ class FFmpegDemuxerStream : public DemuxerStream {
   // Used to translate bitstream formats.
   scoped_ptr<BitstreamConverter> bitstream_converter_;
 
+  // Used to synchronize access to |buffer_queue_|, |read_queue_|, and
+  // |stopped_|. This is so other threads can get access to buffers that have
+  // already been demuxed without having the demuxer thread sending the
+  // buffers. |lock_| must be acquired before any access to |buffer_queue_|,
+  // |read_queue_|, or |stopped_|.
+  base::Lock lock_;
+
   DISALLOW_COPY_AND_ASSIGN(FFmpegDemuxerStream);
 };