Reducing the IPC latency for MSE video decoding

media/base/android/media_decoder_job.cc

 // Copyright 2013 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/base/android/media_decoder_job.h"
 
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "base/debug/trace_event.h"
 #include "base/message_loop/message_loop_proxy.h"
@@ skipping 15 matching lines @@
     : ui_task_runner_(base::MessageLoopProxy::current()),
       decoder_task_runner_(decoder_task_runner),
       media_codec_bridge_(media_codec_bridge),
       needs_flush_(false),
       input_eos_encountered_(false),
       output_eos_encountered_(false),
       skip_eos_enqueue_(true),
       prerolling_(true),
       weak_this_(this),
       request_data_cb_(request_data_cb),
-      access_unit_index_(0),
+      current_demuxer_data_index_(0),
       input_buf_index_(-1),
       stop_decode_pending_(false),
-      destroy_pending_(false) {
+      destroy_pending_(false),
+      is_requesting_demuxer_data_(false),
+      is_incoming_data_invalid_(false) {
+  for (size_t i = 0; i < 2; ++i)
+    access_unit_index_[i] = 0;
 }
 
 MediaDecoderJob::~MediaDecoderJob() {}
 
 void MediaDecoderJob::OnDataReceived(const DemuxerData& data) {
   DVLOG(1) << __FUNCTION__ << ": " << data.access_units.size() << " units";
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
-  DCHECK(!on_data_received_cb_.is_null());
+  DCHECK(IsDemuxerDataEmpty(false));
 
   TRACE_EVENT_ASYNC_END2(
       "media", "MediaDecoderJob::RequestData", this,
       "Data type", data.type == media::DemuxerStream::AUDIO ? "AUDIO" : "VIDEO",
       "Units read", data.access_units.size());
 
+  if (is_incoming_data_invalid_) {
+    is_incoming_data_invalid_ = false;
+    // If there is a pending callback, need to request the data again to get
+    // valid data.
+    if (!on_data_received_cb_.is_null())
+      request_data_cb_.Run();
+    else
+      is_requesting_demuxer_data_ = false;
+    return;
+  }
+
+  size_t next_demuxer_data_index = 1 - current_demuxer_data_index_;
+  received_data_[next_demuxer_data_index] = data;
+  access_unit_index_[next_demuxer_data_index] = 0;
+  is_requesting_demuxer_data_ = false;
+
   base::Closure done_cb = base::ResetAndReturn(&on_data_received_cb_);
-
   if (stop_decode_pending_) {
     OnDecodeCompleted(MEDIA_CODEC_STOPPED, kNoTimestamp(), 0);
     return;
   }
 
-  access_unit_index_ = 0;
-  received_data_ = data;
-  done_cb.Run();
+  if (!done_cb.is_null())
+    done_cb.Run();
 }
 
 void MediaDecoderJob::Prefetch(const base::Closure& prefetch_cb) {
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
   DCHECK(on_data_received_cb_.is_null());
   DCHECK(decode_cb_.is_null());
 
   if (HasData()) {
     DVLOG(1) << __FUNCTION__ << " : using previously received data";
     ui_task_runner_->PostTask(FROM_HERE, prefetch_cb);
     return;
   }
 
   DVLOG(1) << __FUNCTION__ << " : requesting data";
   RequestData(prefetch_cb);
 }
 
 bool MediaDecoderJob::Decode(
     const base::TimeTicks& start_time_ticks,
     const base::TimeDelta& start_presentation_timestamp,
     const DecoderCallback& callback) {
   DCHECK(decode_cb_.is_null());
   DCHECK(on_data_received_cb_.is_null());
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
 
   decode_cb_ = callback;
 
   if (!HasData()) {
-    RequestData(base::Bind(&MediaDecoderJob::DecodeNextAccessUnit,
+    RequestData(base::Bind(&MediaDecoderJob::DecodeCurrentAccessUnit,
                            base::Unretained(this),
                            start_time_ticks,
                            start_presentation_timestamp));
     return true;
   }
 
-  if (DemuxerStream::kConfigChanged ==
-      received_data_.access_units[access_unit_index_].status) {
+  if (DemuxerStream::kConfigChanged == CurrentAccessUnit().status) {
     // Clear received data because we need to handle a config change.
     decode_cb_.Reset();
-    received_data_ = DemuxerData();
-    access_unit_index_ = 0;
+    ClearData();
     return false;
   }
 
-  DecodeNextAccessUnit(start_time_ticks, start_presentation_timestamp);
+  DecodeCurrentAccessUnit(start_time_ticks, start_presentation_timestamp);
   return true;
 }
 
 void MediaDecoderJob::StopDecode() {
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
   DCHECK(is_decoding());
   stop_decode_pending_ = true;
 }
 
 void MediaDecoderJob::Flush() {
   DCHECK(decode_cb_.is_null());
 
   // Do nothing, flush when the next Decode() happens.
   needs_flush_ = true;
-  received_data_ = DemuxerData();
+  ClearData();
   input_eos_encountered_ = false;
-  access_unit_index_ = 0;
-  on_data_received_cb_.Reset();
 }
 
 void MediaDecoderJob::BeginPrerolling(
     const base::TimeDelta& preroll_timestamp) {
   DVLOG(1) << __FUNCTION__ << "(" << preroll_timestamp.InSecondsF() << ")";
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
   DCHECK(!is_decoding());
 
   preroll_timestamp_ = preroll_timestamp;
   prerolling_ = true;
@@ skipping 64 matching lines @@
   // In case of MEDIA_CODEC_NO_KEY, we must reuse the |input_buf_index_|.
   // Otherwise MediaDrm will report errors.
   if (status == MEDIA_CODEC_NO_KEY)
     input_buf_index_ = input_buf_index;
 
   return status;
 }
 
 bool MediaDecoderJob::HasData() const {
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
-  // When |input_eos_encountered_| is set, |access_units| must not be empty and
-  // |access_unit_index_| must be pointing to an EOS unit. We'll reuse this
-  // unit to flush the decoder until we hit output EOS.
-  DCHECK(!input_eos_encountered_ ||
-         (received_data_.access_units.size() > 0 &&
-          access_unit_index_ < received_data_.access_units.size()))
-      << " (access_units.size(): " << received_data_.access_units.size()
-      << ", access_unit_index_: " << access_unit_index_ << ")";
-  return access_unit_index_ < received_data_.access_units.size() ||
-      input_eos_encountered_;
+  // When |input_eos_encountered_| is set, |access_unit| must be pointing to an

[wolenetz, 2014/03/17 19:51:00]

s/access_unit/access_unit_index_/

Also, mention dependency on |current_demuxer_data_index_| (which should also be
held constant when |input_eos_encoundered_| is set).

Does this condition not hold true for ClearData(), which changes
|current_demuxer_data_index_|, so there's a bug maybe?

[qinmin, 2014/03/18 18:58:58]

Done. 
Good point. ClearData() was called in 2 functions： flush() and Decode(). In
flush(), it is immediately followed by a input_eos_encountered = false. In
decode(), the access unit should be a config change. So we are not reaching EOS
yet.
Anyway, we should make the assumption consistent before and after calling
ClearData(). Setting input_eos_encountered to false in ClearData().

On 2014/03/17 19:51:00, wolenetz wrote:

+  // EOS unit. We'll reuse this unit to flush the decoder until we hit output
+  // EOS.
+  DCHECK(!input_eos_encountered_ || !IsDemuxerDataEmpty(true));
+  return !IsDemuxerDataEmpty(true) || !IsDemuxerDataEmpty(false);
 }
 
 void MediaDecoderJob::RequestData(const base::Closure& done_cb) {
   DVLOG(1) << __FUNCTION__;
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
   DCHECK(on_data_received_cb_.is_null());
   DCHECK(!input_eos_encountered_);
+  DCHECK(IsDemuxerDataEmpty(false));
 
   TRACE_EVENT_ASYNC_BEGIN0("media", "MediaDecoderJob::RequestData", this);
 
-  received_data_ = DemuxerData();
-  access_unit_index_ = 0;
   on_data_received_cb_ = done_cb;
 
+  // If we are already expecting new data, just set the callback and do
+  // nothing.
+  if (is_requesting_demuxer_data_)
+    return;
+
+  // The new incoming data will be stored as the next demuxer data chunk, since
+  // the decoder might still be decoding the current one.
+  size_t next_demuxer_data_index = 1 - current_demuxer_data_index_;
+  received_data_[next_demuxer_data_index] = DemuxerData();
+  access_unit_index_[next_demuxer_data_index] = 0;
+  is_requesting_demuxer_data_ = true;
+
   request_data_cb_.Run();
 }
 
-void MediaDecoderJob::DecodeNextAccessUnit(
+void MediaDecoderJob::DecodeCurrentAccessUnit(
     const base::TimeTicks& start_time_ticks,
     const base::TimeDelta& start_presentation_timestamp) {
   DCHECK(ui_task_runner_->BelongsToCurrentThread());
   DCHECK(!decode_cb_.is_null());
 
+  RequestCurrentChunkIfEmpty();
+  const AccessUnit& access_unit = CurrentAccessUnit();
   // If the first access unit is a config change, request the player to dequeue
   // the input buffer again so that it can request config data.
-  if (received_data_.access_units[access_unit_index_].status ==
-      DemuxerStream::kConfigChanged) {
+  if (access_unit.status == DemuxerStream::kConfigChanged) {
     ui_task_runner_->PostTask(FROM_HERE,
                               base::Bind(&MediaDecoderJob::OnDecodeCompleted,
                                          base::Unretained(this),
                                          MEDIA_CODEC_DEQUEUE_INPUT_AGAIN_LATER,
                                          kNoTimestamp(),
                                          0));
     return;
   }
 
   decoder_task_runner_->PostTask(FROM_HERE, base::Bind(
       &MediaDecoderJob::DecodeInternal, base::Unretained(this),
-      received_data_.access_units[access_unit_index_],
+      access_unit,

[wolenetz, 2014/03/17 19:51:00]

Is it possible for ClearData() to occur while the decoder thread still has
reference |access_unit|? If so, does this introduce UAF race bug?
Who owns the lifetime of each DemuxerData() in received_data_[]?

[qinmin, 2014/03/18 18:58:58]

No, ClearData() will be called only when the decoder thread is idle.
ClearData() is called inside either Decode() or Flush() function. Flush() is
called in ProcessPendingEvents() in MSP, Decode() is called when decoder is
idle. So for both cases, the decoder thread should be idle. 
On 2014/03/17 19:51:00, wolenetz wrote:

       start_time_ticks, start_presentation_timestamp, needs_flush_,
       media::BindToCurrentLoop(base::Bind(
           &MediaDecoderJob::OnDecodeCompleted, base::Unretained(this)))));
   needs_flush_ = false;
 }
 
 void MediaDecoderJob::DecodeInternal(
     const AccessUnit& unit,
     const base::TimeTicks& start_time_ticks,
     const base::TimeDelta& start_presentation_timestamp,
@@ skipping 134 matching lines @@
   if (presentation_timestamp != kNoTimestamp())
     prerolling_ = false;
 
   switch (status) {
     case MEDIA_CODEC_OK:
     case MEDIA_CODEC_DEQUEUE_OUTPUT_AGAIN_LATER:
     case MEDIA_CODEC_OUTPUT_BUFFERS_CHANGED:
     case MEDIA_CODEC_OUTPUT_FORMAT_CHANGED:
     case MEDIA_CODEC_OUTPUT_END_OF_STREAM:
       if (!input_eos_encountered_)
-        access_unit_index_++;
+        access_unit_index_[current_demuxer_data_index_]++;
       break;
 
     case MEDIA_CODEC_DEQUEUE_INPUT_AGAIN_LATER:
     case MEDIA_CODEC_INPUT_END_OF_STREAM:
     case MEDIA_CODEC_NO_KEY:
     case MEDIA_CODEC_STOPPED:
     case MEDIA_CODEC_ERROR:
       // Do nothing.
       break;
   };
 
   stop_decode_pending_ = false;
   base::ResetAndReturn(&decode_cb_).Run(status, presentation_timestamp,
                                         audio_output_bytes);
 }
 
+const AccessUnit& MediaDecoderJob::CurrentAccessUnit() const {

[wolenetz, 2014/03/17 19:51:00]

nit: clarify with DCHECK any assumption of which thread this method is allowed
to be called on.

[qinmin, 2014/03/18 18:58:58]

Done.

+  DCHECK(HasData());
+  int index = IsDemuxerDataEmpty(true) ? 1- current_demuxer_data_index_ :

[wolenetz, 2014/03/17 19:51:00]

nit: s/1-/1 -/

[wolenetz, 2014/03/17 19:51:00]

nit: extract "1 - current_demuxer_data_index_" to a(n inline) function like
size_t inactive_demuxer_data_index_() ?

[qinmin, 2014/03/18 18:58:58]

Done.

[qinmin, 2014/03/18 18:58:58]

Done.

+      current_demuxer_data_index_;
+  return received_data_[index].access_units[access_unit_index_[index]];
+}
+
+bool MediaDecoderJob::IsDemuxerDataEmpty(bool current_chunk) const {

[wolenetz, 2014/03/17 19:51:00]

nit: ditto thread dcheck

[qinmin, 2014/03/18 18:58:58]

Done.

+  size_t index = current_chunk ? current_demuxer_data_index_ :
+      1 - current_demuxer_data_index_;
+  return received_data_[index].access_units.size() <= access_unit_index_[index];
+}
+
+void MediaDecoderJob::ClearData() {

[wolenetz, 2014/03/17 19:51:00]

nit: ditto thread dcheck

[qinmin, 2014/03/18 18:58:58]

Done.

+  current_demuxer_data_index_ = 0;

[wolenetz, 2014/03/17 19:51:00]

ditto: If |input_eos_encountered_| is set, we should not change
|current_demuxer_data_index_|.

[qinmin, 2014/03/18 18:58:58]

changed the function to set |input_eos_encountered_| to false.

On 2014/03/17 19:51:00, wolenetz wrote:

+  for (size_t i = 0; i < 2; ++i) {

[wolenetz, 2014/03/17 19:51:00]

nit: Elevate this to a helper function, also use in constructor?

[qinmin, 2014/03/18 18:58:58]

Done.

+    received_data_[i] = DemuxerData();
+    access_unit_index_[i] = 0;
+  }
+  on_data_received_cb_.Reset();
+  if (is_requesting_demuxer_data_)
+    is_incoming_data_invalid_ = true;
+}
+
+void MediaDecoderJob::RequestCurrentChunkIfEmpty() {

[wolenetz, 2014/03/17 19:51:00]

nit: ditto thread dcheck

[qinmin, 2014/03/18 18:58:58]

Done.

+  DCHECK(HasData());
+  if (!IsDemuxerDataEmpty(true))
+    return;
+
+  // Requests new data if the the last access unit of the next chunk is not EOS.
+  current_demuxer_data_index_ = 1 - current_demuxer_data_index_;
+  const AccessUnit last_access_unit =
+      received_data_[current_demuxer_data_index_].access_units.back();
+  if (!last_access_unit.end_of_stream &&
+      last_access_unit.status != DemuxerStream::kConfigChanged &&
+      last_access_unit.status != DemuxerStream::kAborted) {
+    RequestData(base::Closure());
+  }
+}
+
 }  // namespace media