Reducing the IPC latency for MSE video decoding

media/base/android/media_decoder_job.cc

Index: media/base/android/media_decoder_job.cc
diff --git a/media/base/android/media_decoder_job.cc b/media/base/android/media_decoder_job.cc
index 535b8d114e0a76194d3df307b0534bc3b50ab65f..41be6423180bce07a3807259289d8e6eb7fe4997 100644
--- a/media/base/android/media_decoder_job.cc
+++ b/media/base/android/media_decoder_job.cc
@@ -33,10 +33,14 @@ MediaDecoderJob::MediaDecoderJob(
       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() {}
@@ -44,23 +48,37 @@ 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());
 
-  base::Closure done_cb = base::ResetAndReturn(&on_data_received_cb_);
+  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) {
@@ -89,23 +107,21 @@ bool MediaDecoderJob::Decode(
   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;
 }
 
@@ -120,10 +136,8 @@ void MediaDecoderJob::Flush() {
 
   // 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(
@@ -208,16 +222,11 @@ MediaCodecStatus MediaDecoderJob::QueueInputBuffer(const AccessUnit& unit) {
 
 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) {
@@ -225,26 +234,38 @@ void MediaDecoderJob::RequestData(const base::Closure& done_cb) {
   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),
@@ -256,7 +277,7 @@ void MediaDecoderJob::DecodeNextAccessUnit(
 
   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)))));
@@ -411,7 +432,7 @@ void MediaDecoderJob::OnDecodeCompleted(
     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:
@@ -428,4 +449,44 @@ void MediaDecoderJob::OnDecodeCompleted(
                                         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