Allow multiple concurrent Decode() requests in VideoDecoder interface.

media/filters/decoder_stream.cc

 // Copyright 2014 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/filters/decoder_stream.h"
 
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "base/debug/trace_event.h"
 #include "base/location.h"
@@ skipping 32 matching lines @@
     const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
     ScopedVector<Decoder> decoders,
     const SetDecryptorReadyCB& set_decryptor_ready_cb)
     : task_runner_(task_runner),
       state_(STATE_UNINITIALIZED),
       stream_(NULL),
       decoder_selector_(
           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() {
   DCHECK(state_ == STATE_UNINITIALIZED || state_ == STATE_STOPPED) << state_;
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::Initialize(DemuxerStream* stream,
                                            const StatisticsCB& statistics_cb,
                                            const InitCB& init_cb) {
@@ skipping 13 matching lines @@
       stream,
       base::Bind(&DecoderStream<StreamType>::OnDecoderSelected,
                  weak_factory_.GetWeakPtr()));
 }
 
 template <DemuxerStream::Type StreamType>
 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.
   DCHECK(reset_cb_.is_null());
   DCHECK(stop_cb_.is_null());
 
   if (state_ == STATE_ERROR) {
     task_runner_->PostTask(FROM_HERE, base::Bind(
         read_cb, DECODE_ERROR, scoped_refptr<Output>()));
     return;
   }
 
   read_cb_ = read_cb;
 
+  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();
+  }
+
+  if (state_ == STATE_REINITIALIZING_DECODER)
+    return;
+
+  if (!CanDecodeAnotherBuffer())
+    return;
+
   if (state_ == STATE_FLUSHING_DECODER) {
     FlushDecoder();
     return;
   }
 
-  scoped_refptr<Output> output = decoder_->GetDecodeOutput();
-
-  // 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));
-    return;
-  }
-
-  ReadFromDemuxerStream();
+  if (state_ != STATE_PENDING_DEMUXER_READ)
+    ReadFromDemuxerStream();
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::Reset(const base::Closure& closure) {
   FUNCTION_DVLOG(2);
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK(state_ != STATE_UNINITIALIZED && state_ != STATE_STOPPED) << state_;
   DCHECK(reset_cb_.is_null());
   DCHECK(stop_cb_.is_null());
 
@@ skipping 76 matching lines @@
   return decoder_->CanReadWithoutStalling();
 }
 
 template <>
 bool DecoderStream<DemuxerStream::AUDIO>::CanReadWithoutStalling() const {
   DCHECK(task_runner_->BelongsToCurrentThread());
   return true;
 }
 
 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) {
   FUNCTION_DVLOG(2);
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK_EQ(state_, STATE_INITIALIZING) << state_;
   DCHECK(!init_cb_.is_null());
   DCHECK(read_cb_.is_null());
   DCHECK(reset_cb_.is_null());
 
@@ skipping 37 matching lines @@
   // caller.
   DCHECK(!reset_cb_.is_null());
   SatisfyRead(ABORTED, NULL);
 }
 
 template <DemuxerStream::Type StreamType>
 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);
 
   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(),
                               buffer_size));
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::FlushDecoder() {
   Decode(DecoderBuffer::CreateEOSBuffer());
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::OnDecodeOutputReady(
     int buffer_size,
     typename Decoder::Status status,
     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);
 
   if (status == Decoder::kDecodeError) {
     state_ = STATE_ERROR;
     SatisfyRead(DECODE_ERROR, NULL);
     return;
   }
 
   if (status == Decoder::kDecryptError) {
@@ skipping 28 matching lines @@
 
   if (status == Decoder::kNotEnoughData) {
     if (state_ == STATE_NORMAL)
       ReadFromDemuxerStream();
     else if (state_ == STATE_FLUSHING_DECODER)
       FlushDecoder();
     return;
   }
 
   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());
 
   state_ = STATE_PENDING_DEMUXER_READ;
   stream_->Read(base::Bind(&DecoderStream<StreamType>::OnBufferReady,
                            weak_factory_.GetWeakPtr()));
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::OnBufferReady(
     DemuxerStream::Status status,
     const scoped_refptr<DecoderBuffer>& buffer) {
   FUNCTION_DVLOG(2) << ": " << status;
   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;
 
   if (status == DemuxerStream::kConfigChanged) {
     FUNCTION_DVLOG(2) << ": " << "ConfigChanged";
     DCHECK(stream_->SupportsConfigChanges());
 
     if (!config_change_observer_cb_.is_null())
       config_change_observer_cb_.Run();
@@ skipping 26 matching lines @@
     return;
   }
 
   if (!splice_observer_cb_.is_null() && !buffer->end_of_stream() &&
       buffer->splice_timestamp() != kNoTimestamp()) {
     splice_observer_cb_.Run(buffer->splice_timestamp());
   }
 
   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>
 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;
   decoder_->Initialize(
       StreamTraits::GetDecoderConfig(*stream_),
       base::Bind(&DecoderStream<StreamType>::OnDecoderReinitialized,
                  weak_factory_.GetWeakPtr()));
 }
 
 template <DemuxerStream::Type StreamType>
@@ skipping 75 matching lines @@
   decrypting_demuxer_stream_.reset();
   // Post |stop_cb_| because pending |read_cb_| and/or |reset_cb_| are also
   // posted in Stop().
   task_runner_->PostTask(FROM_HERE, base::ResetAndReturn(&stop_cb_));
 }
 
 template class DecoderStream<DemuxerStream::VIDEO>;
 template class DecoderStream<DemuxerStream::AUDIO>;
 
 }  // namespace media