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)

[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.
   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)

[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 (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 =

[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) {
   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;
   }

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

Now we could have multiple Decode() calls in flight, but we still only have one
|read_cb_|. That's why SatisfyRead() always assumes that the |read_cb_| is not
null. So when pending_decode_requests_ > 1 and we call DecoderStream::Reset(), I
assume OnDecodeOutputReady() will be called for multiple times before the
decoder is fully reset. In that case, we'll call SatisfyRead() multiple times,
which won't work. I think what we need to do is only SatisfyRead once.

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

Done. Also added unittests for this case.

 
   if (status == Decoder::kDecryptError) {
     state_ = STATE_ERROR;
     SatisfyRead(DECRYPT_ERROR, NULL);
     return;
   }
 
   if (status == Decoder::kAborted) {
     SatisfyRead(ABORTED, NULL);
     return;
   }
 
   // Any successful decode counts!
   if (buffer_size > 0) {
     StreamTraits::ReportStatistics(statistics_cb_, buffer_size);
   }
 
   // Drop decoding result if Reset() was called during decoding.
   // The resetting process will be handled when the decoder is reset.
   if (!reset_cb_.is_null()) {
     AbortRead();

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

ditto. We only need to AbortRead() once.

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

Removed AbortRead(). Reads are aborted from Reset() now.

     return;
   }
 
   // Decoder flushed. Reinitialize the decoder.
   if (state_ == STATE_FLUSHING_DECODER &&
       status == Decoder::kOk && output->end_of_stream()) {
     ReinitializeDecoder();
     return;
   }
 
   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) {

[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());
 
   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()) {

[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>
 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