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 27 matching lines @@
   return "DecoderStream<AUDIO>::Decode";
 }
 
 template <DemuxerStream::Type StreamType>
 DecoderStream<StreamType>::DecoderStream(
     const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
     ScopedVector<Decoder> decoders,
     const SetDecryptorReadyCB& set_decryptor_ready_cb)
     : task_runner_(task_runner),
       state_(STATE_UNINITIALIZED),
+      error_(OK),
       stream_(NULL),
       decoder_selector_(
           new DecoderSelector<StreamType>(task_runner,
                                           decoders.Pass(),
                                           set_decryptor_ready_cb)),
       active_splice_(false),
+      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,
@@ skipping 14 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_ == STATE_PENDING_DEMUXER_READ)
+      << 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());
 
+  read_cb_ = read_cb;
+
   if (state_ == STATE_ERROR) {
+    DCHECK_NE(error_, OK);
     task_runner_->PostTask(FROM_HERE, base::Bind(
-        read_cb, DECODE_ERROR, scoped_refptr<Output>()));
+        base::ResetAndReturn(&read_cb_), error_, scoped_refptr<Output>()));
     return;
   }
 
-  read_cb_ = read_cb;
+  if (!ready_outputs_.empty()) {
+    task_runner_->PostTask(FROM_HERE, base::Bind(
+        base::ResetAndReturn(&read_cb_), OK, ready_outputs_.front()));
+    ready_outputs_.pop_front();
+  }
+
+  // Decoder may be in reinitializing state as result of the previous Read().
+  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());
 
   reset_cb_ = closure;
 
+  if (!read_cb_.is_null()) {
+    task_runner_->PostTask(FROM_HERE, base::Bind(
+        base::ResetAndReturn(&read_cb_), ABORTED, scoped_refptr<Output>()));
+  }
+
+  ready_outputs_.clear();
+
   // During decoder reinitialization, the Decoder does not need to be and
   // cannot be Reset(). |decrypting_demuxer_stream_| was reset before decoder
   // reinitialization.
   if (state_ == STATE_REINITIALIZING_DECODER)
     return;
 
   // During pending demuxer read and when not using DecryptingDemuxerStream,
   // the Decoder will be reset after demuxer read is returned
   // (in OnBufferReady()).
   if (state_ == STATE_PENDING_DEMUXER_READ && !decrypting_demuxer_stream_)
     return;
 
-  // The Decoder API guarantees that if Decoder::Reset() is called during
-  // a pending decode, the decode callback must be fired before the reset
-  // callback is fired. Therefore, we can call Decoder::Reset() regardless
-  // of if we have a pending decode and always satisfy the reset callback when
-  // the decoder reset is finished.
   if (decrypting_demuxer_stream_) {
     decrypting_demuxer_stream_->Reset(base::Bind(
         &DecoderStream<StreamType>::ResetDecoder, weak_factory_.GetWeakPtr()));
     return;
   }
 
   ResetDecoder();
 }
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::Stop(const base::Closure& closure) {
   FUNCTION_DVLOG(2);
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK_NE(state_, STATE_STOPPED) << state_;
   DCHECK(stop_cb_.is_null());
 
   stop_cb_ = closure;
 
   if (state_ == STATE_INITIALIZING) {
     decoder_selector_->Abort();
     return;
   }
 
   DCHECK(init_cb_.is_null());
 
   // All pending callbacks will be dropped.
   weak_factory_.InvalidateWeakPtrs();
 
   // Post callbacks to prevent reentrance into this object.
-  if (!read_cb_.is_null())
+  if (!read_cb_.is_null()) {
     task_runner_->PostTask(FROM_HERE, base::Bind(
         base::ResetAndReturn(&read_cb_), ABORTED, scoped_refptr<Output>()));
+  }
   if (!reset_cb_.is_null())
     task_runner_->PostTask(FROM_HERE, base::ResetAndReturn(&reset_cb_));
 
   if (decrypting_demuxer_stream_) {
     decrypting_demuxer_stream_->Stop(base::Bind(
         &DecoderStream<StreamType>::StopDecoder, weak_factory_.GetWeakPtr()));
     return;
   }
 
   // We may not have a |decoder_| if Stop() was called during initialization.
@@ skipping 15 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());
+
+  // Limit total number of outputs stored in |ready_outputs_| and being decoded.
+  // It only makes sense to saturate decoder completely when output queue is
+  // empty.
+  // TODO(sergeyu): Is that the best way to throttle decoding requests?

[xhwang, 2014/04/25 00:36:03]

I don't see a better way to do that. The current approach makes sense to me.

[Sergey Ulanov, 2014/04/26 00:59:29]

removed TODO

+  int num_decodes =
+      static_cast<int>(ready_outputs_.size()) + pending_decode_requests_;
+  return num_decodes < decoder_->GetMaxDecodeRequests();
+}
+
+template <>
+bool DecoderStream<DemuxerStream::AUDIO>::CanDecodeAnotherBuffer() const {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  return !pending_decode_requests_ && ready_outputs_.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 22 matching lines @@
 
 template <DemuxerStream::Type StreamType>
 void DecoderStream<StreamType>::SatisfyRead(
     Status status,
     const scoped_refptr<Output>& output) {
   DCHECK(!read_cb_.is_null());
   base::ResetAndReturn(&read_cb_).Run(status, output);
 }
 
 template <DemuxerStream::Type StreamType>
-void DecoderStream<StreamType>::AbortRead() {
-  // Abort read during pending reset. It is safe to fire the |read_cb_| directly
-  // instead of posting it because the renderer won't call into this class
-  // again when it's in kFlushing state.
-  // TODO(xhwang): Improve the resetting process to avoid this dependency on the
-  // 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());
+  if (pending_decode_requests_ == 0)

[xhwang, 2014/04/25 00:36:03]

This also depends on the fact that GVD doesn't have decoding delays, which may
not be true in general. (We need a high level comment about decoding delay
assumption somewhere.)

[Sergey Ulanov, 2014/04/26 00:59:29]

Added condition in VideoDecoder comment requiring that |decode_cb| is called in
the same order in which the decoder gets the requests, i.e. frames cannot be
decoded out of order. That's what I think is important here.

[xhwang, 2014/04/29 20:01:26]

By "decoding delay", I mean the case where VideoDecoder will not return a
decoded frame until more buffers has been fed to the decoder.

For example:
input:  I P B B P B B ...
output:   I B B P B B P ...

Here's what I was worrying about:

Demuxed buffer: 1 2 3 ConfigChange

Then DS will make the following calls:
Decode(1): no frame returned
Decode(2): frame 1 returned
Decode(3): frame 2 returned
FlushDecoder()

Where FlushDecoder() will do nothing because pending_decode_requests_ is not
zero (frame 3 not returned). But without flushing the decoder, the decoder will
not return frame 3, hence a deadlock.

Currently we don't have a decoder that does this so we are fine:
1, GVD doesn't have decoding delay.
2, Decoders that have decoding delay always return kNotEnoughData in DecodeCB so
that we don't accumulate DecodeCB due to decoding delay.

Maybe we should add a comment in VideoDecoder that the decoder must call
DecodeCB some time after Decode() call even if there are no further inputs. In
other words, decoders must call DecodeCB with a frame if it can produce one, or
with a kNotEnoughData if it cannot (e.g. due to decoding delay).

WDYT?

[Sergey Ulanov, 2014/04/30 18:56:51]

Ah, I see your point. Yes, decoders must never wait for following Decode() call
before calling decode_cb. I've added comment about it in video_decoder.h.

On 2014/04/29 20:01:26, xhwang wrote:

+    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());
+  FUNCTION_DVLOG(2) << status << " " << output;
+  DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
+         state_ == STATE_PENDING_DEMUXER_READ || state_ == STATE_ERROR)
+      << state_;
   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 (state_ == STATE_ERROR)

[xhwang, 2014/04/25 00:36:03]

DCHECK(!read_cb_.is_null())?

[Sergey Ulanov, 2014/04/26 00:59:29]

I think you meant  "DCHECK(read_cb_.is_null())"
Done

+    return;
+
   if (status == Decoder::kDecodeError) {
     state_ = STATE_ERROR;
-    SatisfyRead(DECODE_ERROR, NULL);
+    error_ = DECODE_ERROR;

[xhwang, 2014/04/25 00:36:03]

Shall we clear the ready_outputs_ in case of error?

[Sergey Ulanov, 2014/04/26 00:59:29]

Done.

+    if (!read_cb_.is_null())
+      SatisfyRead(DECODE_ERROR, NULL);
     return;
   }
 
   if (status == Decoder::kDecryptError) {
     state_ = STATE_ERROR;
-    SatisfyRead(DECRYPT_ERROR, NULL);
+    error_ = DECRYPT_ERROR;

[xhwang, 2014/04/25 00:36:03]

ditto

[Sergey Ulanov, 2014/04/26 00:59:29]

Done.

+    if (!read_cb_.is_null())
+      SatisfyRead(DECRYPT_ERROR, NULL);
     return;
   }
 
   if (status == Decoder::kAborted) {
-    SatisfyRead(ABORTED, NULL);
+    if (!read_cb_.is_null())

[xhwang, 2014/04/25 00:36:03]

kAborted must be a result of Reset(). In Reset() call we already cleared
ready_outputs_. Also, any frame returned after Reset() won't be added to
ready_outputs_ until Reset() is completed. So we should be able to
DCHECK(ready_outputs_.empty()) here. WDYT?

[Sergey Ulanov, 2014/04/26 00:59:29]

There are some tests that return kAborted from decoder without Reset() being
called. See [Audio|Video]RendererImplTest.AbortPendingRead_* . So I assume there
is a case when stream can be aborted by the decoder. Do you think those tests
are incorrect and should be fixed?

[xhwang, 2014/04/29 20:01:26]

You are right. kAborted can also be a result of DemuxerStream::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();
+  if (!reset_cb_.is_null())
     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_outputs_.push_back(output);
+  scoped_refptr<Output> extra_output;
+  while ((extra_output = decoder_->GetDecodeOutput()) != NULL) {
+    ready_outputs_.push_back(extra_output);
+  }
+
+  // Satisfy outstanding read request, if any.
+  if (!read_cb_.is_null()) {
+    scoped_refptr<Output> read_result = ready_outputs_.front();
+    ready_outputs_.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(state_ == STATE_PENDING_DEMUXER_READ || state_ == STATE_ERROR ||
+         state_ == STATE_STOPPED)
+      << state_;
   DCHECK_EQ(buffer.get() != NULL, status == DemuxerStream::kOk) << status;
-  DCHECK(!read_cb_.is_null());
   DCHECK(stop_cb_.is_null());
 
+  // Decoding has been stopped (e.g due to an error).
+  if (state_ != STATE_PENDING_DEMUXER_READ)

[xhwang, 2014/04/25 00:36:03]

Can we 
DCHECK_EQ(state_, STATE_ERROR);
DCHECK(read_cb_.is_null());
?

[Sergey Ulanov, 2014/04/26 00:59:29]

Done. state_ can be either STATE_ERROR or STATE_STOPPED.

+    return;
+
   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();
 
     state_ = STATE_FLUSHING_DECODER;
     if (!reset_cb_.is_null()) {
-      AbortRead();
       // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
       // which will continue the resetting process in it's callback.
       if (!decrypting_demuxer_stream_)
         Reset(base::ResetAndReturn(&reset_cb_));
       // Reinitialization will continue after Reset() is done.
     } else {
       FlushDecoder();
     }
     return;
   }
 
   if (!reset_cb_.is_null()) {
-    AbortRead();
     // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
     // which will continue the resetting process in it's callback.
     if (!decrypting_demuxer_stream_)
       Reset(base::ResetAndReturn(&reset_cb_));
     return;
   }
 
   if (status == DemuxerStream::kAborted) {
     SatisfyRead(DEMUXER_READ_ABORTED, NULL);
     return;
   }
 
   if (!splice_observer_cb_.is_null() && !buffer->end_of_stream()) {
     const bool has_splice_ts = buffer->splice_timestamp() != kNoTimestamp();
     if (active_splice_ || has_splice_ts) {
       splice_observer_cb_.Run(buffer->splice_timestamp());
       active_splice_ = has_splice_ts;
     }
   }
 
   DCHECK(status == DemuxerStream::kOk) << status;
   Decode(buffer);
+
+  // Read more data if the decoder supports multiple parallel decoding requests.
+  if (CanDecodeAnotherBuffer() && !buffer->end_of_stream())
+    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>
 void DecoderStream<StreamType>::OnDecoderReinitialized(PipelineStatus status) {
   FUNCTION_DVLOG(2);
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK_EQ(state_, STATE_REINITIALIZING_DECODER) << state_;
   DCHECK(stop_cb_.is_null());
 
   // ReinitializeDecoder() can be called in two cases:
   // 1, Flushing decoder finished (see OnDecodeOutputReady()).
   // 2, Reset() was called during flushing decoder (see OnDecoderReset()).
   // Also, Reset() can be called during pending ReinitializeDecoder().
   // This function needs to handle them all!
 
   state_ = (status == PIPELINE_OK) ? STATE_NORMAL : STATE_ERROR;
 
   if (!reset_cb_.is_null()) {
-    if (!read_cb_.is_null())
-      AbortRead();
     base::ResetAndReturn(&reset_cb_).Run();
+    return;
   }
 
   if (read_cb_.is_null())
     return;
 
   if (state_ == STATE_ERROR) {
     SatisfyRead(DECODE_ERROR, NULL);
     return;
   }
 
@@ skipping 47 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