Cherry pick PR #8185: media: Implement flow control for MediaDecoder (#9585)

Refer to the original PR: youtube/cobalt#8185

Integrates `DecoderStateTracker` into the Android `MediaDecoder` to
throttle input processing when the decoder has too many pending frames.
This helps manage memory usage and prevents the decoder from falling
too far behind.

Key changes:
*   Instantiates `DecoderStateTracker` in `MediaDecoder` if enabled.
* Blocks input processing in `DecoderThreadFunc` when `CanAcceptMore()`
returns false.
*   Updates frame tracking via `SetFrameAdded` and `SetFrameDecoded`.

Bug: 455938352

---------

Co-authored-by: Kyujung Youn <kjyoun@google.com>

Author: cobalt-github-releaser-bot

starboard/android/shared/media_decoder.cc

@@ -122,15 +122,16 @@ MediaCodecDecoder::CreateForVideo(
     int tunnel_mode_audio_session_id,
     bool force_big_endian_hdr_metadata,
     int max_video_input_size,
-    int64_t flush_delay_usec) {
+    int64_t flush_delay_usec,
+    std::optional<int> initial_max_frames) {
   std::string error_message;
   auto decoder = std::make_unique<MediaCodecDecoder>(
       PassKey<MediaCodecDecoder>(), job_queue, host, video_codec,
       frame_size_hint, max_frame_size, fps, j_output_surface, drm_system,
       color_metadata, require_software_codec, frame_rendered_cb,
       first_tunnel_frame_ready_cb, tunnel_mode_audio_session_id,
       force_big_endian_hdr_metadata, max_video_input_size, flush_delay_usec,
-      &error_message);
+      initial_max_frames, &error_message);
   if (!decoder->media_codec_bridge_) {
     return Failure(error_message);
   }
@@ -191,6 +192,7 @@ MediaCodecDecoder::MediaCodecDecoder(
     bool force_big_endian_hdr_metadata,
     int max_video_input_size,
     int64_t flush_delay_usec,
+    std::optional<int> initial_max_frames,
     std::string* error_message)
     : JobOwner(job_queue),
 
@@ -200,7 +202,16 @@ MediaCodecDecoder::MediaCodecDecoder(
       frame_rendered_cb_(frame_rendered_cb),
       first_tunnel_frame_ready_cb_(first_tunnel_frame_ready_cb),
       tunnel_mode_enabled_(tunnel_mode_audio_session_id != -1),
-      flush_delay_usec_(flush_delay_usec) {
+      flush_delay_usec_(flush_delay_usec),
+      decoder_state_tracker_([&]() -> std::unique_ptr<DecoderStateTracker> {
+        if (!initial_max_frames || tunnel_mode_enabled_) {
+          return nullptr;
+        }
+        return std::make_unique<DecoderStateTracker>(*initial_max_frames);
+      }()) {
+  if (initial_max_frames && tunnel_mode_enabled_) {
+    SB_LOG(INFO) << "DecoderStateTracker is disabled for tunnel mode.";
+  }
   SB_DCHECK(frame_rendered_cb_);
   SB_DCHECK(first_tunnel_frame_ready_cb_);
 
@@ -368,6 +379,14 @@ void MediaCodecDecoder::DecoderThreadFunc() {
     std::vector<int> input_buffer_indices;
     std::vector<DequeueOutputResult> dequeue_output_results;
 
+    auto can_process_input = [this, &pending_inputs, &input_buffer_indices] {
+      if (decoder_state_tracker_ && !decoder_state_tracker_->CanAcceptMore()) {
+        return false;
+      }
+      return pending_input_to_retry_ ||
+             (!pending_inputs.empty() && !input_buffer_indices.empty());
+    };
+
     while (!destroying_.load()) {
       // TODO(b/329686979): access to `ending_input_to_retry_` should be
       //                    synchronized.
@@ -408,10 +427,7 @@ void MediaCodecDecoder::DecoderThreadFunc() {
         host_->Tick(media_codec_bridge_.get());
       }
 
-      bool can_process_input =
-          pending_input_to_retry_ ||
-          (!pending_inputs.empty() && !input_buffer_indices.empty());
-      if (can_process_input) {
+      if (can_process_input()) {
         ProcessOneInputBuffer(&pending_inputs, &input_buffer_indices);
       }
 
@@ -421,16 +437,11 @@ void MediaCodecDecoder::DecoderThreadFunc() {
         ticked = host_->Tick(media_codec_bridge_.get());
       }
 
-      can_process_input =
-          pending_input_to_retry_ ||
-          (!pending_inputs.empty() && !input_buffer_indices.empty());
-      if (!ticked && !can_process_input && dequeue_output_results.empty()) {
+      if (!ticked && !can_process_input() && dequeue_output_results.empty()) {
         std::unique_lock lock(mutex_);
         CollectPendingData_Locked(&pending_inputs, &input_buffer_indices,
                                   &dequeue_output_results);
-        can_process_input =
-            !pending_inputs.empty() && !input_buffer_indices.empty();
-        if (!can_process_input && dequeue_output_results.empty()) {
+        if (!can_process_input() && dequeue_output_results.empty()) {
           // Wait for a signal or a timeout. We don't use a predicate here
           // because the complex conditions are already checked by the
           // surrounding loop, which will re-evaluate the state when this wait
@@ -606,6 +617,14 @@ bool MediaCodecDecoder::ProcessOneInputBuffer(
     return false;
   }
 
+  if (decoder_state_tracker_) {
+    if (pending_input.type == PendingInput::kWriteEndOfStream) {
+      decoder_state_tracker_->MarkEosReached();
+    } else {
+      decoder_state_tracker_->TrackNewFrame(input_buffer->timestamp());
+    }
+  }
+
   is_output_restricted_ = false;
   return true;
 }
@@ -727,6 +746,10 @@ void MediaCodecDecoder::OnMediaCodecOutputBufferAvailable(
     return;
   }
 
+  if (size > 0 && decoder_state_tracker_) {
+    decoder_state_tracker_->MarkFrameDecoded(presentation_time_us);
+  }
+
   DequeueOutputResult dequeue_output_result;
   dequeue_output_result.status = 0;
   dequeue_output_result.index = buffer_index;
@@ -792,6 +815,10 @@ bool MediaCodecDecoder::Flush() {
   dequeue_output_results_.clear();
   pending_input_to_retry_ = std::nullopt;
 
+  if (decoder_state_tracker_) {
+    decoder_state_tracker_->Reset();
+  }
+
   // 2.3. Add OutputFormatChanged to get current output format after Flush().
   DequeueOutputResult dequeue_output_result = {};
   dequeue_output_result.index = -1;

starboard/android/shared/media_decoder.h

@@ -31,6 +31,7 @@
 #include "starboard/common/thread.h"
 #include "starboard/media.h"
 #include "starboard/shared/internal_only.h"
+#include "starboard/shared/starboard/media/decoder_state_tracker.h"
 #include "starboard/shared/starboard/media/media_util.h"
 #include "starboard/shared/starboard/player/filter/common.h"
 #include "starboard/shared/starboard/player/input_buffer_internal.h"
@@ -97,7 +98,8 @@ class MediaCodecDecoder final : private MediaCodecBridge::Handler,
       int tunnel_mode_audio_session_id,
       bool force_big_endian_hdr_metadata,
       int max_video_input_size,
-      int64_t flush_delay_usec);
+      int64_t flush_delay_usec,
+      std::optional<int> initial_max_frames);
 
   MediaCodecDecoder(PassKey<MediaCodecDecoder>,
                     JobQueue* job_queue,
@@ -125,6 +127,7 @@ class MediaCodecDecoder final : private MediaCodecBridge::Handler,
       bool force_big_endian_hdr_metadata,
       int max_video_input_size,
       int64_t flush_delay_usec,
+      std::optional<int> initial_max_frames,
       std::string* error_message);
   ~MediaCodecDecoder();
 
@@ -140,6 +143,10 @@ class MediaCodecDecoder final : private MediaCodecBridge::Handler,
 
   bool Flush();
 
+  DecoderStateTracker* decoder_state_tracker() {
+    return decoder_state_tracker_.get();
+  }
+
  private:
   // Holding inputs to be processed.  They are mostly InputBuffer objects, but
   // can also be codec configs or end of streams.
@@ -236,6 +243,8 @@ class MediaCodecDecoder final : private MediaCodecBridge::Handler,
   std::vector<int> input_buffer_indices_;
   std::vector<DequeueOutputResult> dequeue_output_results_;
 
+  const std::unique_ptr<DecoderStateTracker> decoder_state_tracker_;
+
   bool is_output_restricted_ = false;
   bool first_call_on_handler_thread_ = true;
 

starboard/android/shared/player_components_factory.cc

@@ -501,6 +501,8 @@ class PlayerComponentsFactory : public PlayerComponents::Factory {
         << "`kResetDelayUsec` is set to > 0, force a delay of "
         << reset_delay_usec << "us during Reset().";
 
+    // TODO: b/455938352 - Connect this options to h5vcc settings.
+    MediaCodecVideoDecoder::FlowControlOptions flow_control_options;
     auto result = MediaCodecVideoDecoder::Create(
         creation_parameters.job_queue(),
         creation_parameters.video_stream_info(),
@@ -510,7 +512,8 @@ class PlayerComponentsFactory : public PlayerComponents::Factory {
         tunnel_mode_audio_session_id, force_secure_pipeline_under_tunnel_mode,
         force_reset_surface, force_big_endian_hdr_metadata,
         max_video_input_size, creation_parameters.surface_view(),
-        enable_flush_during_seek, reset_delay_usec, flush_delay_usec);
+        enable_flush_during_seek, reset_delay_usec, flush_delay_usec,
+        flow_control_options);
     if (!result) {
       return Failure(result.error());
     }

starboard/android/shared/video_decoder.cc

@@ -162,7 +162,8 @@ std::optional<Size> ParseMaxResolution(
 
 class VideoFrameImpl : public VideoFrame {
  public:
-  typedef std::function<void()> VideoFrameReleaseCallback;
+  typedef std::function<void(int64_t pts_us, int64_t release_at_us)>
+      VideoFrameReleaseCallback;
 
   VideoFrameImpl(const DequeueOutputResult& dequeue_output_result,
                  MediaCodecBridge* media_codec_bridge,
@@ -183,7 +184,7 @@ class VideoFrameImpl : public VideoFrame {
       media_codec_bridge_->ReleaseOutputBuffer(dequeue_output_result_.index,
                                                false);
       if (!is_end_of_stream()) {
-        release_callback_();
+        release_callback_(timestamp(), CurrentMonotonicTime());
       }
     }
   }
@@ -194,7 +195,7 @@ class VideoFrameImpl : public VideoFrame {
     released_ = true;
     media_codec_bridge_->ReleaseOutputBufferAtTimestamp(
         dequeue_output_result_.index, release_time_in_nanoseconds);
-    release_callback_();
+    release_callback_(timestamp(), release_time_in_nanoseconds / 1'000);
   }
 
  private:
@@ -212,7 +213,7 @@ const int kNonInitialPrerollFrameCount = 1;
 
 const int kSeekingPrerollPendingWorkSizeInTunnelMode =
     16 + kInitialPrerollFrameCount;
-const int kMaxPendingInputsSize = 128;
+const int kDefaultMaxPendingInputsSize = 128;
 
 const int kFpsGuesstimateRequiredInputBufferCount = 3;
 
@@ -344,7 +345,8 @@ MediaCodecVideoDecoder::Create(JobQueue* job_queue,
                                void* surface_view,
                                bool enable_flush_during_seek,
                                int64_t reset_delay_usec,
-                               int64_t flush_delay_usec) {
+                               int64_t flush_delay_usec,
+                               const FlowControlOptions& flow_control_options) {
   std::string error_message;
   auto video_decoder = std::make_unique<MediaCodecVideoDecoder>(
       PassKey<MediaCodecVideoDecoder>(), job_queue, video_stream_info,
@@ -353,7 +355,7 @@ MediaCodecVideoDecoder::Create(JobQueue* job_queue,
       force_secure_pipeline_under_tunnel_mode, force_reset_surface,
       force_big_endian_hdr_metadata, max_input_size, surface_view,
       enable_flush_during_seek, reset_delay_usec, flush_delay_usec,
-      &error_message);
+      flow_control_options, &error_message);
 
   if (!error_message.empty()) {
     return Failure(error_message);
@@ -387,6 +389,7 @@ MediaCodecVideoDecoder::MediaCodecVideoDecoder(
     bool enable_flush_during_seek,
     int64_t reset_delay_usec,
     int64_t flush_delay_usec,
+    const FlowControlOptions& flow_control_options,
     std::string* error_message)
     : JobOwner(job_queue),
       video_codec_(video_stream_info.codec),
@@ -395,6 +398,11 @@ MediaCodecVideoDecoder::MediaCodecVideoDecoder(
       decode_target_graphics_context_provider_(
           decode_target_graphics_context_provider),
       max_video_capabilities_(max_video_capabilities),
+      initial_max_frames_in_decoder_(
+          flow_control_options.initial_max_frames_in_decoder),
+      max_pending_inputs_size_(
+          flow_control_options.max_pending_input_frames.value_or(
+              kDefaultMaxPendingInputsSize)),
       require_software_codec_(IsSoftwareDecodeRequired(max_video_capabilities)),
       force_big_endian_hdr_metadata_(force_big_endian_hdr_metadata),
       tunnel_mode_audio_session_id_(tunnel_mode_audio_session_id),
@@ -428,7 +436,7 @@ MediaCodecVideoDecoder::MediaCodecVideoDecoder(
 
   if (is_video_frame_tracker_enabled_) {
     video_frame_tracker_ =
-        std::make_unique<VideoFrameTracker>(kMaxPendingInputsSize * 2);
+        std::make_unique<VideoFrameTracker>(max_pending_inputs_size_ * 2);
   }
 
   if (require_software_codec_) {
@@ -448,6 +456,7 @@ MediaCodecVideoDecoder::MediaCodecVideoDecoder(
   SB_LOG(INFO) << "Created VideoDecoder for codec "
                << GetMediaVideoCodecName(video_codec_) << ", with output mode "
                << GetPlayerOutputModeName(output_mode_)
+               << ", max pending input size " << max_pending_inputs_size_
                << ", max video capabilities \"" << max_video_capabilities_
                << "\", and tunnel mode audio session id "
                << tunnel_mode_audio_session_id_;
@@ -780,7 +789,7 @@ Result<void> MediaCodecVideoDecoder::InitializeCodec(
       std::bind(&MediaCodecVideoDecoder::OnFrameRendered, this, _1),
       std::bind(&MediaCodecVideoDecoder::OnFirstTunnelFrameReady, this),
       tunnel_mode_audio_session_id_, force_big_endian_hdr_metadata_,
-      max_video_input_size_, flush_delay_usec_);
+      max_video_input_size_, flush_delay_usec_, initial_max_frames_in_decoder_);
   if (result) {
     media_decoder_ = std::move(result.value());
     if (error_cb_) {
@@ -884,7 +893,7 @@ void MediaCodecVideoDecoder::WriteInputBuffersInternal(
   }
 
   media_decoder_->WriteInputBuffers(input_buffers);
-  if (media_decoder_->GetNumberOfPendingInputs() < kMaxPendingInputsSize) {
+  if (media_decoder_->GetNumberOfPendingInputs() < max_pending_inputs_size_) {
     decoder_status_cb_(kNeedMoreInput, NULL);
   } else if (tunnel_mode_audio_session_id_ != -1) {
     // In tunnel mode playback when need data is not signaled above, it is
@@ -921,8 +930,8 @@ void MediaCodecVideoDecoder::WriteInputBuffersInternal(
             max_timestamp >= video_frame_tracker_->seek_to_time();
       }
 
-      bool cache_full =
-          media_decoder_->GetNumberOfPendingInputs() >= kMaxPendingInputsSize;
+      bool cache_full = media_decoder_->GetNumberOfPendingInputs() >=
+                        max_pending_inputs_size_;
       bool prerolled = tunnel_mode_frame_rendered_.load() > 0 ||
                        enough_buffers_written_to_media_codec || cache_full;
 
@@ -960,9 +969,10 @@ void MediaCodecVideoDecoder::ProcessOutputBuffer(
   }
   decoder_status_cb_(
       is_end_of_stream ? kBufferFull : kNeedMoreInput,
-      new VideoFrameImpl(
-          dequeue_output_result, media_codec_bridge,
-          std::bind(&MediaCodecVideoDecoder::OnVideoFrameRelease, this)));
+      new VideoFrameImpl(dequeue_output_result, media_codec_bridge,
+                         [this](int64_t pts_us, int64_t release_at_us) {
+                           OnVideoFrameRelease(pts_us, release_at_us);
+                         }));
 }
 
 void MediaCodecVideoDecoder::RefreshOutputFormat(
@@ -1279,7 +1289,7 @@ void MediaCodecVideoDecoder::OnTunnelModeCheckForNeedMoreInput() {
     return;
   }
 
-  if (media_decoder_->GetNumberOfPendingInputs() < kMaxPendingInputsSize) {
+  if (media_decoder_->GetNumberOfPendingInputs() < max_pending_inputs_size_) {
     decoder_status_cb_(kNeedMoreInput, NULL);
     return;
   }
@@ -1289,11 +1299,17 @@ void MediaCodecVideoDecoder::OnTunnelModeCheckForNeedMoreInput() {
            kNeedMoreInputCheckIntervalInTunnelMode);
 }
 
-void MediaCodecVideoDecoder::OnVideoFrameRelease() {
+void MediaCodecVideoDecoder::OnVideoFrameRelease(int64_t pts_us,
+                                                 int64_t release_at_us) {
   if (output_format_) {
     --buffered_output_frames_;
     SB_DCHECK_GE(buffered_output_frames_, 0);
   }
+
+  if (media_decoder_ && media_decoder_->decoder_state_tracker()) {
+    media_decoder_->decoder_state_tracker()->MarkFrameReleased(pts_us,
+                                                               release_at_us);
+  }
 }
 
 void MediaCodecVideoDecoder::OnSurfaceDestroyed() {