Introduce time-based caching of encoded data to FFmpegDemuxer (round 2).

The original commit r169004 introduce seek regressions on constrained network connections. For now we'll limit the number of asynchronous pending reads to one. While it limits throughput it shouldn't be a big factor during normal playback.

BUG=160640

Review URL: https://codereview.chromium.org/11308145

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

6 files changed, 401 insertions, 77 deletions

diff --git a/media/base/decoder_buffer_queue.cc b/media/base/decoder_buffer_queue.cc
new file mode 100644
index 0000000..ad91c37
--- /dev/null
+++ b/media/base/decoder_buffer_queue.cc
@@ -0,0 +1,78 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/decoder_buffer_queue.h"
+
+#include "base/logging.h"
+#include "media/base/decoder_buffer.h"
+
+namespace media {
+
+DecoderBufferQueue::DecoderBufferQueue()
+    : earliest_valid_timestamp_(kNoTimestamp()) {
+}
+
+DecoderBufferQueue::~DecoderBufferQueue() {}
+
+void DecoderBufferQueue::Push(const scoped_refptr<DecoderBuffer>& buffer) {
+  CHECK(!buffer->IsEndOfStream());
+
+  queue_.push_back(buffer);
+
+  // TODO(scherkus): FFmpeg returns some packets with no timestamp after
+  // seeking. Fix and turn this into CHECK(). See http://crbug.com/162192
+  if (buffer->GetTimestamp() == kNoTimestamp()) {
+    DVLOG(1) << "Buffer has no timestamp";
+    return;
+  }
+
+  if (earliest_valid_timestamp_ == kNoTimestamp()) {
+    earliest_valid_timestamp_ = buffer->GetTimestamp();
+  }
+
+  if (buffer->GetTimestamp() < earliest_valid_timestamp_) {
+    DVLOG(1)
+        << "Out of order timestamps: "
+        << buffer->GetTimestamp().InMicroseconds()
+        << " vs. "
+        << earliest_valid_timestamp_.InMicroseconds();
+    return;
+  }
+
+  earliest_valid_timestamp_ = buffer->GetTimestamp();
+  in_order_queue_.push_back(buffer);
+}
+
+scoped_refptr<DecoderBuffer> DecoderBufferQueue::Pop() {
+  scoped_refptr<DecoderBuffer> buffer = queue_.front();
+  queue_.pop_front();
+
+  if (!in_order_queue_.empty() &&
+      in_order_queue_.front() == buffer) {
+    in_order_queue_.pop_front();
+  }
+
+  return buffer;
+}
+
+void DecoderBufferQueue::Clear() {
+  queue_.clear();
+  in_order_queue_.clear();
+  earliest_valid_timestamp_ = kNoTimestamp();
+}
+
+bool DecoderBufferQueue::IsEmpty() {
+  return queue_.empty();
+}
+
+base::TimeDelta DecoderBufferQueue::Duration() {
+  if (in_order_queue_.size() < 2)
+    return base::TimeDelta();
+
+  base::TimeDelta start = in_order_queue_.front()->GetTimestamp();
+  base::TimeDelta end = in_order_queue_.back()->GetTimestamp();
+  return end - start;
+}
+
+}  // namespace media
diff --git a/media/base/decoder_buffer_queue.h b/media/base/decoder_buffer_queue.h
new file mode 100644
index 0000000..f75046c
--- /dev/null
+++ b/media/base/decoder_buffer_queue.h
@@ -0,0 +1,70 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MEDIA_BASE_DECODER_BUFFER_QUEUE_H_
+#define MEDIA_BASE_DECODER_BUFFER_QUEUE_H_
+
+#include <deque>
+
+#include "base/memory/ref_counted.h"
+#include "base/time.h"
+#include "media/base/media_export.h"
+
+namespace media {
+
+class DecoderBuffer;
+
+// Maintains a queue of DecoderBuffers in increasing timestamp order.
+//
+// Individual buffer durations are ignored when calculating the duration of the
+// queue i.e., the queue must have at least 2 in-order buffers to calculate
+// duration.
+//
+// Not thread safe: access must be externally synchronized.
+class MEDIA_EXPORT DecoderBufferQueue {
+ public:
+  DecoderBufferQueue();
+  ~DecoderBufferQueue();
+
+  // Push |buffer| to the end of the queue. If |buffer| is queued out of order
+  // it will be excluded from duration calculations.
+  //
+  // It is invalid to push an end-of-stream |buffer|.
+  void Push(const scoped_refptr<DecoderBuffer>& buffer);
+
+  // Pops a DecoderBuffer from the front of the queue.
+  //
+  // It is invalid to call Pop() on an empty queue.
+  scoped_refptr<DecoderBuffer> Pop();
+
+  // Removes all queued buffers.
+  void Clear();
+
+  // Returns true if this queue is empty.
+  bool IsEmpty();
+
+  // Returns the duration of encoded data stored in this queue as measured by
+  // the timestamps of the earliest and latest buffers, ignoring out of order
+  // buffers.
+  //
+  // Returns zero if the queue is empty.
+  base::TimeDelta Duration();
+
+ private:
+  typedef std::deque<scoped_refptr<DecoderBuffer> > Queue;
+  Queue queue_;
+
+  // A subset of |queue_| that contains buffers that are in strictly
+  // increasing timestamp order. Used to calculate Duration() while ignoring
+  // out-of-order buffers.
+  Queue in_order_queue_;
+
+  base::TimeDelta earliest_valid_timestamp_;
+
+  DISALLOW_COPY_AND_ASSIGN(DecoderBufferQueue);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_BASE_DECODER_BUFFER_QUEUE_H_
diff --git a/media/base/decoder_buffer_queue_unittest.cc b/media/base/decoder_buffer_queue_unittest.cc
new file mode 100644
index 0000000..02cd541
--- /dev/null
+++ b/media/base/decoder_buffer_queue_unittest.cc
@@ -0,0 +1,137 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/decoder_buffer.h"
+#include "media/base/decoder_buffer_queue.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace media {
+
+static base::TimeDelta ToTimeDelta(int seconds) {
+  if (seconds < 0)
+    return kNoTimestamp();
+  return base::TimeDelta::FromSeconds(seconds);
+}
+
+// Helper to create buffers with specified timestamp in seconds.
+//
+// Negative numbers will be converted to kNoTimestamp();
+static scoped_refptr<DecoderBuffer> CreateBuffer(int timestamp) {
+  scoped_refptr<DecoderBuffer> buffer = new DecoderBuffer(0);
+  buffer->SetTimestamp(ToTimeDelta(timestamp));
+  buffer->SetDuration(ToTimeDelta(0));
+  return buffer;
+}
+
+TEST(DecoderBufferQueueTest, IsEmpty) {
+  DecoderBufferQueue queue;
+  EXPECT_TRUE(queue.IsEmpty());
+
+  queue.Push(CreateBuffer(0));
+  EXPECT_FALSE(queue.IsEmpty());
+}
+
+TEST(DecoderBufferQueueTest, Clear) {
+  DecoderBufferQueue queue;
+  queue.Push(CreateBuffer(0));
+  queue.Push(CreateBuffer(1));
+  EXPECT_FALSE(queue.IsEmpty());
+  EXPECT_EQ(1, queue.Duration().InSeconds());
+
+  queue.Clear();
+  EXPECT_TRUE(queue.IsEmpty());
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+}
+
+TEST(DecoderBufferQueueTest, Duration) {
+  DecoderBufferQueue queue;
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  queue.Push(CreateBuffer(0));
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  queue.Push(CreateBuffer(1));
+  EXPECT_EQ(1, queue.Duration().InSeconds());
+
+  queue.Push(CreateBuffer(2));
+  EXPECT_EQ(2, queue.Duration().InSeconds());
+
+  queue.Push(CreateBuffer(4));
+  EXPECT_EQ(4, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(3, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(2, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+}
+
+TEST(DecoderBufferQueueTest, Duration_OutOfOrder) {
+  DecoderBufferQueue queue;
+  queue.Push(CreateBuffer(10));
+  queue.Push(CreateBuffer(12));
+  EXPECT_EQ(2, queue.Duration().InSeconds());
+
+  // Out of order: duration shouldn't change.
+  queue.Push(CreateBuffer(8));
+  EXPECT_EQ(2, queue.Duration().InSeconds());
+
+  // Removing first buffer should leave the second buffer as the only buffer
+  // included in the duration calculation.
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  // Removing second buffer leaves the out-of-order buffer. It shouldn't be
+  // included in duration calculations.
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  // Push a still-too-early buffer. It shouldn't be included in duration
+  // calculations.
+  queue.Push(CreateBuffer(11));
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  // Push a buffer that's after the earliest valid time. It's a singular valid
+  // buffer so duration is still zero.
+  queue.Push(CreateBuffer(14));
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  // Push a second valid buffer. We should now have a duration.
+  queue.Push(CreateBuffer(17));
+  EXPECT_EQ(3, queue.Duration().InSeconds());
+}
+
+TEST(DecoderBufferQueueTest, Duration_NoTimestamp) {
+  // Buffers with no timestamp don't affect duration.
+  DecoderBufferQueue queue;
+  queue.Push(CreateBuffer(0));
+  queue.Push(CreateBuffer(4));
+  EXPECT_EQ(4, queue.Duration().InSeconds());
+
+  queue.Push(CreateBuffer(-1));
+  EXPECT_EQ(4, queue.Duration().InSeconds());
+
+  queue.Push(CreateBuffer(6));
+  EXPECT_EQ(6, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(2, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+
+  queue.Pop();
+  EXPECT_EQ(0, queue.Duration().InSeconds());
+}
+
+}  // namespace media
diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index 64f8401..326b881 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -39,6 +39,7 @@ FFmpegDemuxerStream::FFmpegDemuxerStream(
       stream_(stream),
       type_(UNKNOWN),
       stopped_(false),
+      end_of_stream_(false),
       last_packet_timestamp_(kNoTimestamp()),
       bitstream_converter_enabled_(false) {
   DCHECK(demuxer_);
@@ -67,68 +68,65 @@ FFmpegDemuxerStream::FFmpegDemuxerStream(
   }
 }
 
-bool FFmpegDemuxerStream::HasPendingReads() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
-  DCHECK(!stopped_ || read_queue_.empty())
-      << "Read queue should have been emptied if demuxing stream is stopped";
-  return !read_queue_.empty();
-}
-
 void FFmpegDemuxerStream::EnqueuePacket(ScopedAVPacket packet) {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
-  if (stopped_) {
+  if (stopped_ || end_of_stream_) {
     NOTREACHED() << "Attempted to enqueue packet on a stopped stream";
     return;
   }
 
-  scoped_refptr<DecoderBuffer> buffer;
-  if (!packet.get()) {
-    buffer = DecoderBuffer::CreateEOSBuffer();
-  } else {
-    // Convert the packet if there is a bitstream filter.
-    if (packet->data && bitstream_converter_enabled_ &&
-        !bitstream_converter_->ConvertPacket(packet.get())) {
-      LOG(ERROR) << "Format converstion failed.";
-    }
-
-    // If a packet is returned by FFmpeg's av_parser_parse2() the packet will
-    // reference inner memory of FFmpeg.  As such we should transfer the packet
-    // into memory we control.
-    buffer = DecoderBuffer::CopyFrom(packet->data, packet->size);
-    buffer->SetTimestamp(ConvertStreamTimestamp(
-        stream_->time_base, packet->pts));
-    buffer->SetDuration(ConvertStreamTimestamp(
-        stream_->time_base, packet->duration));
-    if (buffer->GetTimestamp() != kNoTimestamp() &&
-        last_packet_timestamp_ != kNoTimestamp() &&
-        last_packet_timestamp_ < buffer->GetTimestamp()) {
-      buffered_ranges_.Add(last_packet_timestamp_, buffer->GetTimestamp());
-      demuxer_->NotifyBufferingChanged();
-    }
-    last_packet_timestamp_ = buffer->GetTimestamp();
+  // Convert the packet if there is a bitstream filter.
+  if (packet->data && bitstream_converter_enabled_ &&
+      !bitstream_converter_->ConvertPacket(packet.get())) {
+    LOG(ERROR) << "Format conversion failed.";
   }
 
-  buffer_queue_.push_back(buffer);
+  // If a packet is returned by FFmpeg's av_parser_parse2() the packet will
+  // reference inner memory of FFmpeg.  As such we should transfer the packet
+  // into memory we control.
+  scoped_refptr<DecoderBuffer> buffer;
+  buffer = DecoderBuffer::CopyFrom(packet->data, packet->size);
+  buffer->SetTimestamp(ConvertStreamTimestamp(
+      stream_->time_base, packet->pts));
+  buffer->SetDuration(ConvertStreamTimestamp(
+      stream_->time_base, packet->duration));
+  if (buffer->GetTimestamp() != kNoTimestamp() &&
+      last_packet_timestamp_ != kNoTimestamp() &&
+      last_packet_timestamp_ < buffer->GetTimestamp()) {
+    buffered_ranges_.Add(last_packet_timestamp_, buffer->GetTimestamp());
+    demuxer_->NotifyBufferingChanged();
+  }
+  last_packet_timestamp_ = buffer->GetTimestamp();
+
+  buffer_queue_.Push(buffer);
+  SatisfyPendingReads();
+}
+
+void FFmpegDemuxerStream::SetEndOfStream() {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  end_of_stream_ = true;
   SatisfyPendingReads();
 }
 
 void FFmpegDemuxerStream::FlushBuffers() {
   DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(read_queue_.empty()) << "Read requests should be empty";
-  buffer_queue_.clear();
+  buffer_queue_.Clear();
+  end_of_stream_ = false;
   last_packet_timestamp_ = kNoTimestamp();
 }
 
 void FFmpegDemuxerStream::Stop() {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  buffer_queue_.clear();
+  buffer_queue_.Clear();
   for (ReadQueue::iterator it = read_queue_.begin();
        it != read_queue_.end(); ++it) {
     it->Run(DemuxerStream::kOk, DecoderBuffer::CreateEOSBuffer());
   }
   read_queue_.clear();
   stopped_ = true;
+  end_of_stream_ = true;
 }
 
 base::TimeDelta FFmpegDemuxerStream::duration() {
@@ -182,7 +180,7 @@ const VideoDecoderConfig& FFmpegDemuxerStream::video_decoder_config() {
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
   DCHECK(stopped_);
   DCHECK(read_queue_.empty());
-  DCHECK(buffer_queue_.empty());
+  DCHECK(buffer_queue_.IsEmpty());
 }
 
 base::TimeDelta FFmpegDemuxerStream::GetElapsedTime() const {
@@ -195,23 +193,36 @@ Ranges<base::TimeDelta> FFmpegDemuxerStream::GetBufferedRanges() const {
 
 void FFmpegDemuxerStream::SatisfyPendingReads() {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  while (!read_queue_.empty() && !buffer_queue_.empty()) {
+  while (!read_queue_.empty()) {
+    scoped_refptr<DecoderBuffer> buffer;
+
+    if (!buffer_queue_.IsEmpty()) {
+      buffer = buffer_queue_.Pop();
+    } else if (end_of_stream_) {
+      buffer = DecoderBuffer::CreateEOSBuffer();
+    } else {
+      break;
+    }
+
+    // Send buffer back on a new execution stack to avoid recursing.
     ReadCB read_cb = read_queue_.front();
     read_queue_.pop_front();
-
-    // Send earliest buffer back on a new execution stack to avoid recursing.
-    scoped_refptr<DecoderBuffer> buffer = buffer_queue_.front();
-    buffer_queue_.pop_front();
     message_loop_->PostTask(FROM_HERE, base::Bind(
         read_cb, DemuxerStream::kOk, buffer));
   }
 
-  // No buffers but pending reads? Ask for more!
-  if (!read_queue_.empty() && buffer_queue_.empty()) {
-    demuxer_->NotifyHasPendingRead();
+  // Have capacity? Ask for more!
+  if (HasAvailableCapacity() && !end_of_stream_) {
+    demuxer_->NotifyCapacityAvailable();
   }
 }
 
+bool FFmpegDemuxerStream::HasAvailableCapacity() {
+  // Try to have one second's worth of encoded data per stream.
+  const base::TimeDelta kCapacity = base::TimeDelta::FromSeconds(1);
+  return buffer_queue_.IsEmpty() || buffer_queue_.Duration() < kCapacity;
+}
+
 // static
 base::TimeDelta FFmpegDemuxerStream::ConvertStreamTimestamp(
     const AVRational& time_base, int64 timestamp) {
@@ -230,6 +241,8 @@ FFmpegDemuxer::FFmpegDemuxer(
     : host_(NULL),
       message_loop_(message_loop),
       blocking_thread_("FFmpegDemuxer"),
+      pending_read_(false),
+      pending_seek_(false),
       data_source_(data_source),
       bitrate_(0),
       start_time_(kNoTimestamp()),
@@ -467,6 +480,11 @@ void FFmpegDemuxer::OnFindStreamInfoDone(const PipelineStatusCB& status_cb,
 
 void FFmpegDemuxer::SeekTask(base::TimeDelta time, const PipelineStatusCB& cb) {
   DCHECK(message_loop_->BelongsToCurrentThread());
+  CHECK(!pending_seek_);
+
+  // TODO(scherkus): Inspect |pending_read_| and cancel IO via |blocking_url_|,
+  // otherwise we can end up waiting for a pre-seek read to complete even though
+  // we know we're going to drop it on the floor.
 
   // Always seek to a timestamp less than or equal to the desired timestamp.
   int flags = AVSEEK_FLAG_BACKWARD;
@@ -474,6 +492,7 @@ void FFmpegDemuxer::SeekTask(base::TimeDelta time, const PipelineStatusCB& cb) {
   // Passing -1 as our stream index lets FFmpeg pick a default stream.  FFmpeg
   // will attempt to use the lowest-index video stream, if present, followed by
   // the lowest-index audio stream.
+  pending_seek_ = true;
   base::PostTaskAndReplyWithResult(
       blocking_thread_.message_loop_proxy(), FROM_HERE,
       base::Bind(&av_seek_frame, glue_->format_context(), -1,
@@ -483,6 +502,9 @@ void FFmpegDemuxer::SeekTask(base::TimeDelta time, const PipelineStatusCB& cb) {
 
 void FFmpegDemuxer::OnSeekFrameDone(const PipelineStatusCB& cb, int result) {
   DCHECK(message_loop_->BelongsToCurrentThread());
+  CHECK(pending_seek_);
+  pending_seek_ = false;
+
   if (!blocking_thread_.IsRunning()) {
     cb.Run(PIPELINE_ERROR_ABORT);
     return;
@@ -502,6 +524,9 @@ void FFmpegDemuxer::OnSeekFrameDone(const PipelineStatusCB& cb, int result) {
       (*iter)->FlushBuffers();
   }
 
+  // Resume demuxing until capacity.
+  DemuxTask();
+
   // Notify we're finished seeking.
   cb.Run(PIPELINE_OK);
 }
@@ -510,7 +535,8 @@ void FFmpegDemuxer::DemuxTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // Make sure we have work to do before demuxing.
-  if (!StreamsHavePendingReads()) {
+  if (!blocking_thread_.IsRunning() || !StreamsHaveAvailableCapacity() ||
+      pending_read_ || pending_seek_) {
     return;
   }
 
@@ -520,6 +546,7 @@ void FFmpegDemuxer::DemuxTask() {
   ScopedAVPacket packet(new AVPacket());
   AVPacket* packet_ptr = packet.get();
 
+  pending_read_ = true;
   base::PostTaskAndReplyWithResult(
       blocking_thread_.message_loop_proxy(), FROM_HERE,
       base::Bind(&av_read_frame, glue_->format_context(), packet_ptr),
@@ -528,7 +555,10 @@ void FFmpegDemuxer::DemuxTask() {
 
 void FFmpegDemuxer::OnReadFrameDone(ScopedAVPacket packet, int result) {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  if (!blocking_thread_.IsRunning()) {
+  DCHECK(pending_read_);
+  pending_read_ = false;
+
+  if (!blocking_thread_.IsRunning() || pending_seek_) {
     return;
   }
 
@@ -570,12 +600,8 @@ void FFmpegDemuxer::OnReadFrameDone(ScopedAVPacket packet, int result) {
     demuxer_stream->EnqueuePacket(packet.Pass());
   }
 
-  // Create a loop by posting another task.  This allows seek and message loop
-  // quit tasks to get processed.
-  if (StreamsHavePendingReads()) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &FFmpegDemuxer::DemuxTask, this));
-  }
+  // Keep demuxing until we've reached capacity.
+  DemuxTask();
 }
 
 void FFmpegDemuxer::StopTask(const base::Closure& callback) {
@@ -613,11 +639,11 @@ void FFmpegDemuxer::DisableAudioStreamTask() {
   }
 }
 
-bool FFmpegDemuxer::StreamsHavePendingReads() {
+bool FFmpegDemuxer::StreamsHaveAvailableCapacity() {
   DCHECK(message_loop_->BelongsToCurrentThread());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
-    if (*iter && (*iter)->HasPendingReads()) {
+    if (*iter && (*iter)->HasAvailableCapacity()) {
       return true;
     }
   }
@@ -632,11 +658,11 @@ void FFmpegDemuxer::StreamHasEnded() {
         (audio_disabled_ && (*iter)->type() == DemuxerStream::AUDIO)) {
       continue;
     }
-    (*iter)->EnqueuePacket(ScopedAVPacket());
+    (*iter)->SetEndOfStream();
   }
 }
 
-void FFmpegDemuxer::NotifyHasPendingRead() {
+void FFmpegDemuxer::NotifyCapacityAvailable() {
   DCHECK(message_loop_->BelongsToCurrentThread());
   DemuxTask();
 }
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index 5f8566f..889cc3b 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -30,6 +30,7 @@
 #include "base/threading/thread.h"
 #include "media/base/audio_decoder_config.h"
 #include "media/base/decoder_buffer.h"
+#include "media/base/decoder_buffer_queue.h"
 #include "media/base/demuxer.h"
 #include "media/base/pipeline.h"
 #include "media/base/video_decoder_config.h"
@@ -55,14 +56,15 @@ class FFmpegDemuxerStream : public DemuxerStream {
   // inside |stream|.  Both parameters must outlive |this|.
   FFmpegDemuxerStream(FFmpegDemuxer* demuxer, AVStream* stream);
 
-  // Returns true is this stream has pending reads, false otherwise.
-  bool HasPendingReads();
-
-  // Enqueues the given AVPacket.  If |packet| is NULL an end of stream packet
-  // is enqueued.
+  // Enqueues the given AVPacket. It is invalid to queue a |packet| after
+  // SetEndOfStream() has been called.
   void EnqueuePacket(ScopedAVPacket packet);
 
-  // Signals to empty the buffer queue and mark next packet as discontinuous.
+  // Enters the end of stream state. After delivering remaining queued buffers
+  // only end of stream buffers will be delivered.
+  void SetEndOfStream();
+
+  // Drops queued buffers and clears end of stream state.
   void FlushBuffers();
 
   // Empties the queues and ignores any additional calls to Read().
@@ -85,6 +87,9 @@ class FFmpegDemuxerStream : public DemuxerStream {
   // Used to determine stream duration when it's not known ahead of time.
   base::TimeDelta GetElapsedTime() const;
 
+  // Returns true if this stream has capacity for additional data.
+  bool HasAvailableCapacity();
+
  protected:
   virtual ~FFmpegDemuxerStream();
 
@@ -107,11 +112,11 @@ class FFmpegDemuxerStream : public DemuxerStream {
   Type type_;
   base::TimeDelta duration_;
   bool stopped_;
+  bool end_of_stream_;
   base::TimeDelta last_packet_timestamp_;
   Ranges<base::TimeDelta> buffered_ranges_;
 
-  typedef std::deque<scoped_refptr<DecoderBuffer> > BufferQueue;
-  BufferQueue buffer_queue_;
+  DecoderBufferQueue buffer_queue_;
 
   typedef std::deque<ReadCB> ReadQueue;
   ReadQueue read_queue_;
@@ -138,9 +143,9 @@ class MEDIA_EXPORT FFmpegDemuxer : public Demuxer {
       DemuxerStream::Type type) OVERRIDE;
   virtual base::TimeDelta GetStartTime() const OVERRIDE;
 
-  // Allow FFmpegDemuxerStream to notify us when it requires more data or has
-  // updated information about what buffered data is available.
-  void NotifyHasPendingRead();
+  // Allow FFmpegDemuxerStream to notify us when there is updated information
+  // about capacity and what buffered data is available.
+  void NotifyCapacityAvailable();
   void NotifyBufferingChanged();
 
  private:
@@ -169,16 +174,11 @@ class MEDIA_EXPORT FFmpegDemuxer : public Demuxer {
   // 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.
-  //
-  // Must be called on the demuxer thread.
-  bool StreamsHavePendingReads();
+  // Returns true iff any stream has additional capacity. Note that streams can
+  // go over capacity depending on how the file is muxed.
+  bool StreamsHaveAvailableCapacity();
 
-  // Signal all FFmpegDemuxerStream that the stream has ended.
-  //
-  // Must be called on the demuxer thread.
+  // Signal all FFmpegDemuxerStreams that the stream has ended.
   void StreamHasEnded();
 
   // Called by |url_protocol_| whenever |data_source_| returns a read error.
@@ -196,6 +196,16 @@ class MEDIA_EXPORT FFmpegDemuxer : public Demuxer {
   // Thread on which all blocking FFmpeg operations are executed.
   base::Thread blocking_thread_;
 
+  // Tracks if there's an outstanding av_read_frame() operation.
+  //
+  // TODO(scherkus): Allow more than one read in flight for higher read
+  // throughput using demuxer_bench to verify improvements.
+  bool pending_read_;
+
+  // Tracks if there's an outstanding av_seek_frame() operation. Used to discard
+  // results of pre-seek av_read_frame() operations.
+  bool pending_seek_;
+
   // |streams_| mirrors the AVStream array in |format_context_|. It contains
   // FFmpegDemuxerStreams encapsluating AVStream objects at the same index.
   //
diff --git a/media/media.gyp b/media/media.gyp
index 552dfee..0bb6e00 100644
--- a/media/media.gyp
+++ b/media/media.gyp
@@ -202,6 +202,8 @@
         'base/data_source.h',
         'base/decoder_buffer.cc',
         'base/decoder_buffer.h',
+        'base/decoder_buffer_queue.cc',
+        'base/decoder_buffer_queue.h',
         'base/decryptor.cc',
         'base/decryptor.h',
         'base/decryptor_client.h',
@@ -684,6 +686,7 @@
         'base/clock_unittest.cc',
         'base/data_buffer_unittest.cc',
         'base/decoder_buffer_unittest.cc',
+        'base/decoder_buffer_queue_unittest.cc',
         'base/djb2_unittest.cc',
         'base/filter_collection_unittest.cc',
         'base/gmock_callback_support_unittest.cc',