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)

[xhwang, 2014/04/17 01:06:47]

|state_| can also be STATE_PENDING_DEMUXER_READ. Imagine:

DS::Read() {
Read satisfied because ready_output_buffers isn't empty() (line 106).
ReadFromDemuxerStream();  // line 123; state_ changed to
STATE_PENDING_DEMUXER_READ
}

// Since previous Read() was satisfied.

DS::Read() {
// Now state is STATE_PENDING_DEMUXER_READ.
}

Actually you have the check against STATE_PENDING_DEMUXER_READ on line 122 :)

[Sergey Ulanov, 2014/04/23 02:44:21]

Done. Also added a unittest that coverts this case.

+      << 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)

[xhwang, 2014/04/17 01:06:47]

Might worth a comment how this could happen.

[Sergey Ulanov, 2014/04/23 02:44:21]

Done.

+    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 =

[xhwang, 2014/04/17 01:06:47]

In media/ we use "buffer" for compressed data, e.g. DecoderBuffer. In this
class, we use "output" to refer to decoded output. Maybe we should rename
buffers_in_queue and ready_output_buffers...

[Sergey Ulanov, 2014/04/23 02:44:21]

Renamed
  buffers_in_queue -> num_decodes
  ready_output_buffers_ -> ready_outputs_

+      static_cast<int>(ready_output_buffers_.size()) + pending_decode_requests_;
+  return buffers_in_queue < decoder_->GetMaxDecodeRequests();

[xhwang, 2014/04/17 01:06:47]

Probably add a comment about why we choose to do this. My understanding is: if
decoding is slow and DS's buffers_in_queue is running low, then we can trigger
to do parallel decoding at the maximum speed.

[Sergey Ulanov, 2014/04/23 02:44:21]

Done.

+}
+
+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) {

[xhwang, 2014/04/17 01:06:47]

Now GetDecodeOutput() will always return NULL for GVD. So this function is only
used by the audio decoders (where one input can result in multiple outputs). We
may need to clean this up a bit in the future.

[Sergey Ulanov, 2014/04/23 02:44:21]

Added TODO in video_decoder.h.

+    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()) {

[xhwang, 2014/04/17 01:06:47]

We set state_ to be STATE_NORMAL on line 410, so here state_ has to be
STATE_NORMAL?

[Sergey Ulanov, 2014/04/23 02:44:21]

Done.

+    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;