Allow multiple concurrent Decode() requests in VideoDecoder interface.

media/filters/decoder_stream.cc

Index: media/filters/decoder_stream.cc
diff --git a/media/filters/decoder_stream.cc b/media/filters/decoder_stream.cc
index 350e47f72cd3887c2fd01ee9d878774791254745..aaf2f0fc795c1b315c3970bd2c1f1f45131790f3 100644
--- a/media/filters/decoder_stream.cc
+++ b/media/filters/decoder_stream.cc
@@ -50,7 +50,9 @@ DecoderStream<StreamType>::DecoderStream(
           new DecoderSelector<StreamType>(task_runner,
                                           decoders.Pass(),
                                           set_decryptor_ready_cb)),
-      weak_factory_(this) {}
+      pending_decode_requests_(0),
+      weak_factory_(this) {
+}
 
 template <DemuxerStream::Type StreamType>
 DecoderStream<StreamType>::~DecoderStream() {
@@ -84,7 +86,8 @@ void DecoderStream<StreamType>::Read(const ReadCB& read_cb) {
   FUNCTION_DVLOG(2);
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
-         state_ == STATE_ERROR) << state_;
+         state_ == STATE_ERROR || state_ == STATE_REINITIALIZING_DECODER)
+      << state_;
   // No two reads in the flight at any time.
   DCHECK(read_cb_.is_null());
   // No read during resetting or stopping process.
@@ -99,21 +102,25 @@ void DecoderStream<StreamType>::Read(const ReadCB& read_cb) {
 
   read_cb_ = read_cb;
 
-  if (state_ == STATE_FLUSHING_DECODER) {
-    FlushDecoder();
-    return;
+  if (!ready_output_buffers_.empty()) {
+    task_runner_->PostTask(FROM_HERE, base::Bind(
+        base::ResetAndReturn(&read_cb_), OK, ready_output_buffers_.front()));
+    ready_output_buffers_.pop_front();
   }
 
-  scoped_refptr<Output> output = decoder_->GetDecodeOutput();
+  if (state_ == STATE_REINITIALIZING_DECODER)
+    return;
+
+  if (!CanDecodeAnotherBuffer())
+    return;
 
-  // If the decoder has queued output ready to go we don't need a demuxer read.
-  if (output) {
-    task_runner_->PostTask(
-        FROM_HERE, base::Bind(base::ResetAndReturn(&read_cb_), OK, output));
+  if (state_ == STATE_FLUSHING_DECODER) {
+    FlushDecoder();
     return;
   }
 
-  ReadFromDemuxerStream();
+  if (state_ != STATE_PENDING_DEMUXER_READ)
+    ReadFromDemuxerStream();
 }
 
 template <DemuxerStream::Type StreamType>
@@ -210,6 +217,20 @@ bool DecoderStream<DemuxerStream::AUDIO>::CanReadWithoutStalling() const {
 }
 
 template <DemuxerStream::Type StreamType>
+bool DecoderStream<StreamType>::CanDecodeAnotherBuffer() const {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  int buffers_in_queue =
+      static_cast<int>(ready_output_buffers_.size()) + pending_decode_requests_;

[Ami GONE FROM CHROMIUM, 2014/04/16 01:00:00]

Why does it matter how many ready_output_buffers_ there are? 
I suspect what you really want is to cap the # of ready buffers with a separate
limit.

[Sergey Ulanov, 2014/04/16 01:44:14]

We need to limit number of output buffers waiting in ready_output_buffers_. It
doesn't make sense to request another decode request if there are already some
buffers in the queue, so we need to take into account size of
ready_output_buffers_ when making decision to decode another frame. If Read() is
never called again then we will end up with ready_output_buffers_.size() +
pending_decode_requests_ total output buffers waiting in the queue.

We could have a separate limit on the size of the output queue, but then it
shouldn't be lower than GetMaxDecodeRequests() (otherwise we wouldn't always
saturate the decoder, depending on when and how Read() is called), and it's
never necessary to have a queue longer than GetMaxDecodeRequests().

I don't feel strongly about this logic so I can change it if you think a
different approach would work better.

+  return buffers_in_queue < decoder_->GetMaxDecodeRequests();
+}
+
+template <>
+bool DecoderStream<DemuxerStream::AUDIO>::CanDecodeAnotherBuffer() const {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  return !pending_decode_requests_ && ready_output_buffers_.empty();
+}
+
+template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::OnDecoderSelected(
     scoped_ptr<Decoder> selected_decoder,
     scoped_ptr<DecryptingDemuxerStream> decrypting_demuxer_stream) {
@@ -267,7 +288,7 @@ void DecoderStream<StreamType>::Decode(
     const scoped_refptr<DecoderBuffer>& buffer) {
   FUNCTION_DVLOG(2);
   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER) << state_;
-  DCHECK(!read_cb_.is_null());
+  DCHECK(CanDecodeAnotherBuffer());
   DCHECK(reset_cb_.is_null());
   DCHECK(stop_cb_.is_null());
   DCHECK(buffer);
@@ -275,6 +296,7 @@ void DecoderStream<StreamType>::Decode(
   int buffer_size = buffer->end_of_stream() ? 0 : buffer->data_size();
 
   TRACE_EVENT_ASYNC_BEGIN0("media", GetTraceString<StreamType>(), this);
+  ++pending_decode_requests_;
   decoder_->Decode(buffer,
                    base::Bind(&DecoderStream<StreamType>::OnDecodeOutputReady,
                               weak_factory_.GetWeakPtr(),
@@ -293,9 +315,11 @@ void DecoderStream<StreamType>::OnDecodeOutputReady(
     const scoped_refptr<Output>& output) {
   FUNCTION_DVLOG(2);
   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER) << state_;
-  DCHECK(!read_cb_.is_null());
   DCHECK(stop_cb_.is_null());
   DCHECK_EQ(status == Decoder::kOk, output != NULL);
+  DCHECK_GT(pending_decode_requests_, 0);
+
+  --pending_decode_requests_;
 
   TRACE_EVENT_ASYNC_END0("media", GetTraceString<StreamType>(), this);
 
@@ -344,14 +368,27 @@ void DecoderStream<StreamType>::OnDecodeOutputReady(
   }
 
   DCHECK(output);
-  SatisfyRead(OK, output);
+
+  // Store decoded output.
+  ready_output_buffers_.push_back(output);
+  scoped_refptr<Output> extra_output;
+  while ((extra_output = decoder_->GetDecodeOutput()) != NULL) {
+    ready_output_buffers_.push_back(extra_output);
+  }
+
+  // Satisfy outstanding read request, if any.
+  if (!read_cb_.is_null()) {
+    scoped_refptr<Output> read_result = ready_output_buffers_.front();
+    ready_output_buffers_.pop_front();
+    SatisfyRead(OK, output);
+  }
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::ReadFromDemuxerStream() {
   FUNCTION_DVLOG(2);
   DCHECK_EQ(state_, STATE_NORMAL) << state_;
-  DCHECK(!read_cb_.is_null());
+  DCHECK(CanDecodeAnotherBuffer());
   DCHECK(reset_cb_.is_null());
   DCHECK(stop_cb_.is_null());
 
@@ -368,7 +405,6 @@ void DecoderStream<StreamType>::OnBufferReady(
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK_EQ(state_, STATE_PENDING_DEMUXER_READ) << state_;
   DCHECK_EQ(buffer.get() != NULL, status == DemuxerStream::kOk) << status;
-  DCHECK(!read_cb_.is_null());
   DCHECK(stop_cb_.is_null());
 
   state_ = STATE_NORMAL;
@@ -415,6 +451,11 @@ void DecoderStream<StreamType>::OnBufferReady(
 
   DCHECK(status == DemuxerStream::kOk) << status;
   Decode(buffer);
+
+  // Read more data if the decoder supports multiple parallel decoding requests.
+  if (state_ == STATE_NORMAL && CanDecodeAnotherBuffer()) {
+    ReadFromDemuxerStream();
+  }
 }
 
 template <DemuxerStream::Type StreamType>
@@ -422,6 +463,7 @@ void DecoderStream<StreamType>::ReinitializeDecoder() {
   FUNCTION_DVLOG(2);
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK_EQ(state_, STATE_FLUSHING_DECODER) << state_;
+  DCHECK_EQ(pending_decode_requests_, 0);
 
   DCHECK(StreamTraits::GetDecoderConfig(*stream_).IsValidConfig());
   state_ = STATE_REINITIALIZING_DECODER;