adaptivedemux2: Rework input download wakeups

Change the way streams are woken up to download more data.

Instead of checking the level on tracks that are being
output as data is dequeued, calculate a 'wakeup time'
at which it should download more data, and wake up
the stream when the global output position crosses
that threshold.

For efficiency, compute the earliest wakeup time
for all streams and store it on the period, so the
output loop can quickly check only a single value
to decide if something needs waking up.

Does the same buffering as the previous method,
but ensures that as we approach the end of
one period, the next period continues incrementally
downloading data so that it is fully buffered when
the period starts.

Fixes issues with multi-period VOD content where
download of the second period resumes only after
the first period is completely drained.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3055>

subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-period.c

@@ -40,6 +40,7 @@ gst_adaptive_demux_period_new (GstAdaptiveDemux * demux)
 
   period->demux = demux;
   period->period_num = demux->priv->n_periods++;
+  period->next_input_wakeup_time = GST_CLOCK_STIME_NONE;
 
   g_queue_push_tail (demux->priv->periods, period);
 
@@ -292,3 +293,39 @@ gst_adaptive_demux_period_has_pending_tracks (GstAdaptiveDemuxPeriod * period)
   }
   return FALSE;
 }
+
+/* Called from the output thread, holding the tracks lock */
+void
+gst_adaptive_demux_period_check_input_wakeup_locked (GstAdaptiveDemuxPeriod *
+    period, GstClockTimeDiff current_output_position)
+{
+  /* Fast case: It's not time to wake up yet */
+  if (period->next_input_wakeup_time == GST_CLOCK_STIME_NONE ||
+      period->next_input_wakeup_time > current_output_position) {
+    return;
+  }
+
+  /* Slow case: Somewhere there's a stream that needs waking up */
+  GList *iter;
+  GstClockTimeDiff next_input_wakeup_time = GST_CLOCK_STIME_NONE;
+
+  for (iter = period->streams; iter; iter = g_list_next (iter)) {
+    GstAdaptiveDemux2Stream *stream = iter->data;
+
+    if (stream->next_input_wakeup_time == GST_CLOCK_STIME_NONE)
+      continue;
+
+    if (stream->next_input_wakeup_time < current_output_position) {
+      GST_LOG_OBJECT (stream, "Waking for more input at time %" GST_TIME_FORMAT,
+          GST_TIME_ARGS (current_output_position));
+      gst_adaptive_demux2_stream_on_output_space_available (stream);
+    } else if (next_input_wakeup_time == GST_CLOCK_STIME_NONE ||
+        /* Otherwise this stream will need waking in the future, accumulate
+         * the earliest stream wakeup time */
+        stream->next_input_wakeup_time < next_input_wakeup_time) {
+      next_input_wakeup_time = stream->next_input_wakeup_time;
+    }
+  }
+
+  period->next_input_wakeup_time = next_input_wakeup_time;
+}

subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-private.h

@@ -217,6 +217,7 @@ GstMiniObject * track_dequeue_data_locked (GstAdaptiveDemux * demux, GstAdaptive
 void gst_adaptive_demux_track_flush (GstAdaptiveDemuxTrack * track);
 void gst_adaptive_demux_track_drain_to (GstAdaptiveDemuxTrack * track, GstClockTime drain_running_time);
 void gst_adaptive_demux_track_update_next_position (GstAdaptiveDemuxTrack * track);
+void gst_adaptive_demux_track_update_level_locked (GstAdaptiveDemuxTrack * track);
 
 /* Period functions */
 GstAdaptiveDemuxPeriod * gst_adaptive_demux_period_new (GstAdaptiveDemux * demux);
@@ -238,5 +239,6 @@ void                     gst_adaptive_demux_period_stop_tasks (GstAdaptiveDemuxP
 GstFlowReturn            gst_adaptive_demux_period_combine_stream_flows (GstAdaptiveDemuxPeriod * period);
 
 gboolean                 gst_adaptive_demux_period_has_pending_tracks (GstAdaptiveDemuxPeriod * period);
+void      gst_adaptive_demux_period_check_input_wakeup_locked (GstAdaptiveDemuxPeriod * period, GstClockTimeDiff current_output_position);
 
 #endif

subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-stream.c

@@ -59,6 +59,7 @@ gst_adaptive_demux2_stream_init (GstAdaptiveDemux2Stream * stream)
   stream->download_request = download_request_new ();
   stream->state = GST_ADAPTIVE_DEMUX2_STREAM_STATE_STOPPED;
   stream->last_ret = GST_FLOW_OK;
+  stream->next_input_wakeup_time = GST_CLOCK_STIME_NONE;
 
   stream->fragment_bitrates =
       g_malloc0 (sizeof (guint64) * NUM_LOOKBACK_FRAGMENTS);
@@ -840,10 +841,10 @@ gst_adaptive_demux2_stream_wait_for_output_space (GstAdaptiveDemux * demux,
     if (track->level_time < high_threshold) {
       if (track->active) {
         need_to_wait = FALSE;
-        GST_DEBUG_OBJECT (demux,
-            "stream %p track %s has level %" GST_TIME_FORMAT
+        GST_DEBUG_OBJECT (stream,
+            "track %s has level %" GST_TIME_FORMAT
             " - needs more data (target %" GST_TIME_FORMAT
-            ") (fragment duration %" GST_TIME_FORMAT ")", stream,
+            ") (fragment duration %" GST_TIME_FORMAT ")",
             track->stream_id, GST_TIME_ARGS (track->level_time),
             GST_TIME_ARGS (high_threshold), GST_TIME_ARGS (fragment_duration));
         continue;
@@ -852,34 +853,67 @@ gst_adaptive_demux2_stream_wait_for_output_space (GstAdaptiveDemux * demux,
       have_filled_inactive = TRUE;
     }
 
-    GST_DEBUG_OBJECT (demux,
-        "stream %p track %s active (%d) has level %" GST_TIME_FORMAT,
-        stream, track->stream_id, track->active,
-        GST_TIME_ARGS (track->level_time));
+    GST_DEBUG_OBJECT (stream,
+        "track %s active (%d) has level %" GST_TIME_FORMAT,
+        track->stream_id, track->active, GST_TIME_ARGS (track->level_time));
   }
 
   /* If there are no tracks, don't wait (we might need data to create them),
    * or if there are active tracks that need more data to hit the threshold,
    * don't wait. Otherwise it means all active tracks are full and we should wait */
   if (!have_any_tracks) {
-    GST_DEBUG_OBJECT (demux, "stream %p has no tracks - not waiting", stream);
+    GST_DEBUG_OBJECT (stream, "no tracks created yet - not waiting");
     need_to_wait = FALSE;
   } else if (!have_active_tracks && !have_filled_inactive) {
-    GST_DEBUG_OBJECT (demux,
-        "stream %p has inactive tracks that need more data - not waiting",
-        stream);
+    GST_DEBUG_OBJECT (stream,
+        "have only inactive tracks that need more data - not waiting");
     need_to_wait = FALSE;
   }
 
   if (need_to_wait) {
+    stream->next_input_wakeup_time = GST_CLOCK_STIME_NONE;
+
     for (iter = stream->tracks; iter; iter = iter->next) {
       GstAdaptiveDemuxTrack *track = (GstAdaptiveDemuxTrack *) iter->data;
-      track->waiting_del_level = high_threshold;
-      GST_DEBUG_OBJECT (demux,
-          "Waiting for queued data on stream %p track %s to drop below %"
+
+      GST_DEBUG_OBJECT (stream,
+          "Waiting for queued data on track %s to drop below %"
           GST_TIME_FORMAT " (fragment duration %" GST_TIME_FORMAT ")",
-          stream, track->stream_id, GST_TIME_ARGS (track->waiting_del_level),
+          track->stream_id, GST_TIME_ARGS (high_threshold),
           GST_TIME_ARGS (fragment_duration));
+
+      /* we want to get woken up when the global output position reaches
+       * a point where the input is closer than "high_threshold" to needing
+       * output, so we can put more data in */
+      GstClockTimeDiff wakeup_time = track->input_time - high_threshold;
+
+      if (stream->next_input_wakeup_time == GST_CLOCK_STIME_NONE ||
+          wakeup_time < stream->next_input_wakeup_time) {
+        stream->next_input_wakeup_time = wakeup_time;
+
+        GST_DEBUG_OBJECT (stream,
+            "Track %s level %" GST_TIME_FORMAT ". Input at position %"
+            GST_TIME_FORMAT " next wakeup should be %" GST_TIME_FORMAT " now %"
+            GST_TIME_FORMAT, track->stream_id,
+            GST_TIME_ARGS (track->level_time),
+            GST_TIME_ARGS (track->input_time), GST_TIME_ARGS (wakeup_time),
+            GST_TIME_ARGS (demux->priv->global_output_position));
+      }
+    }
+
+    if (stream->next_input_wakeup_time != GST_CLOCK_TIME_NONE) {
+      GST_DEBUG_OBJECT (stream,
+          "Next input wakeup time is now %" GST_TIME_FORMAT,
+          GST_TIME_ARGS (stream->next_input_wakeup_time));
+
+      /* If this stream needs waking up sooner than any other current one,
+       * update the period wakeup time, which is what the output loop
+       * will check */
+      GstAdaptiveDemuxPeriod *period = stream->period;
+      if (period->next_input_wakeup_time == GST_CLOCK_STIME_NONE ||
+          period->next_input_wakeup_time > stream->next_input_wakeup_time) {
+        period->next_input_wakeup_time = stream->next_input_wakeup_time;
+      }
     }
   }
 
@@ -1582,6 +1616,26 @@ gst_adaptive_demux2_stream_on_output_space_available_cb (GstAdaptiveDemux2Stream
   if (stream->state != GST_ADAPTIVE_DEMUX2_STREAM_STATE_WAITING_OUTPUT_SPACE)
     return G_SOURCE_REMOVE;
 
+  GstAdaptiveDemux *demux = stream->demux;
+  TRACKS_LOCK (demux);
+
+  GList *iter;
+  for (iter = stream->tracks; iter; iter = iter->next) {
+    GstAdaptiveDemuxTrack *track = (GstAdaptiveDemuxTrack *) iter->data;
+
+    /* We need to recompute the track's level_time value, as the
+     * global output position may have advanced and reduced the
+     * value, even without anything being dequeued yet */
+    gst_adaptive_demux_track_update_level_locked (track);
+
+    GST_DEBUG_OBJECT (stream, "track %s woken level %" GST_TIME_FORMAT
+        " input position %" GST_TIME_FORMAT " at %" GST_TIME_FORMAT,
+        track->stream_id, GST_TIME_ARGS (track->level_time),
+        GST_TIME_ARGS (track->input_time),
+        GST_TIME_ARGS (demux->priv->global_output_position));
+  }
+  TRACKS_UNLOCK (demux);
+
   while (gst_adaptive_demux2_stream_load_a_fragment (stream));
 
   return G_SOURCE_REMOVE;
@@ -1592,12 +1646,8 @@ gst_adaptive_demux2_stream_on_output_space_available (GstAdaptiveDemux2Stream *
     stream)
 {
   GstAdaptiveDemux *demux = stream->demux;
-  GList *iter;
 
-  for (iter = stream->tracks; iter; iter = iter->next) {
-    GstAdaptiveDemuxTrack *tmp_track = (GstAdaptiveDemuxTrack *) iter->data;
-    tmp_track->waiting_del_level = 0;
-  }
+  stream->next_input_wakeup_time = GST_CLOCK_STIME_NONE;
 
   GST_LOG_OBJECT (stream, "Scheduling output_space_available() call");
 
@@ -1959,6 +2009,8 @@ gst_adaptive_demux2_stream_stop (GstAdaptiveDemux2Stream * stream)
   stream->downloading_header = stream->downloading_index = FALSE;
   stream->download_request = download_request_new ();
   stream->download_active = FALSE;
+
+  stream->next_input_wakeup_time = GST_CLOCK_STIME_NONE;
 }
 
 gboolean

subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-track.c

@@ -287,30 +287,13 @@ handle_event:
 
   /* Update track buffering levels */
   if (GST_CLOCK_STIME_IS_VALID (running_time_buffering)) {
-    GstClockTimeDiff output_time;
-
     track->output_time = running_time_buffering;
 
     GST_LOG_OBJECT (demux,
         "track %s buffering time:%" GST_STIME_FORMAT,
         track->stream_id, GST_STIME_ARGS (running_time_buffering));
 
-    if (GST_CLOCK_STIME_IS_VALID (track->output_time))
-      output_time =
-          MAX (track->output_time, demux->priv->global_output_position);
-    else
-      output_time = track->input_time;
-
-    if (track->input_time >= output_time)
-      track->level_time = track->input_time - output_time;
-    else
-      track->level_time = 0;
-
-    GST_LOG_OBJECT (demux,
-        "track %s input_time:%" GST_STIME_FORMAT " output_time:%"
-        GST_STIME_FORMAT " level:%" GST_TIME_FORMAT,
-        track->stream_id, GST_STIME_ARGS (track->input_time),
-        GST_STIME_ARGS (output_time), GST_TIME_ARGS (track->level_time));
+    gst_adaptive_demux_track_update_level_locked (track);
   } else {
     GST_LOG_OBJECT (demux, "track %s popping untimed item %" GST_PTR_FORMAT,
         track->stream_id, res);
@@ -318,17 +301,6 @@ handle_event:
 
   track->level_bytes -= item_size;
 
-  /* FIXME: This logic should be in adaptive demux, not the track */
-  if (track->level_time < track->waiting_del_level) {
-    /* Wake up download loop */
-    GstAdaptiveDemux2Stream *stream =
-        find_stream_for_track_locked (demux, track);
-
-    g_assert (stream != NULL);
-
-    gst_adaptive_demux2_stream_on_output_space_available (stream);
-  }
-
   return res;
 }
 
@@ -444,7 +416,7 @@ track_queue_data_locked (GstAdaptiveDemux * demux,
   item.runningtime_buffering = GST_CLOCK_STIME_NONE;
 
   if (timestamp != GST_CLOCK_TIME_NONE) {
-    GstClockTimeDiff output_time, input_time;
+    GstClockTimeDiff input_time;
 
     /* Set the running time of the item */
     input_time = item.runningtime_end = item.runningtime =
@@ -508,18 +480,7 @@ track_queue_data_locked (GstAdaptiveDemux * demux,
           GST_STIME_ARGS (track->output_time));
     }
 
-    output_time = MAX (track->output_time, demux->priv->global_output_position);
-
-    if (track->input_time >= output_time)
-      track->level_time = track->input_time - output_time;
-    else
-      track->level_time = 0;
-
-    GST_LOG_OBJECT (track->sinkpad,
-        "track %s (period %u) input_time:%" GST_STIME_FORMAT " output_time:%"
-        GST_STIME_FORMAT " level:%" GST_TIME_FORMAT,
-        track->stream_id, track->period_num, GST_STIME_ARGS (track->input_time),
-        GST_STIME_ARGS (track->output_time), GST_TIME_ARGS (track->level_time));
+    gst_adaptive_demux_track_update_level_locked (track);
   }
 
   GST_LOG_OBJECT (track->sinkpad,
@@ -798,6 +759,30 @@ gst_adaptive_demux_track_update_next_position (GstAdaptiveDemuxTrack * track)
       "Track '%s' doesn't have any pending timed data", track->stream_id);
 }
 
+/* TRACKS_LOCK held. Recomputes the level_time for the track */
+void
+gst_adaptive_demux_track_update_level_locked (GstAdaptiveDemuxTrack * track)
+{
+  GstAdaptiveDemux *demux = track->demux;
+  GstClockTimeDiff output_time;
+
+  if (GST_CLOCK_STIME_IS_VALID (track->output_time))
+    output_time = MAX (track->output_time, demux->priv->global_output_position);
+  else
+    output_time = MIN (track->input_time, demux->priv->global_output_position);
+
+  if (track->input_time >= output_time)
+    track->level_time = track->input_time - output_time;
+  else
+    track->level_time = 0;
+
+  GST_LOG_OBJECT (track->sinkpad,
+      "track %s (period %u) input_time:%" GST_STIME_FORMAT " output_time:%"
+      GST_STIME_FORMAT " level:%" GST_TIME_FORMAT,
+      track->stream_id, track->period_num, GST_STIME_ARGS (track->input_time),
+      GST_STIME_ARGS (track->output_time), GST_TIME_ARGS (track->level_time));
+}
+
 static void
 _demux_track_free (GstAdaptiveDemuxTrack * track)
 {
@@ -955,7 +940,6 @@ gst_adaptive_demux_track_new (GstAdaptiveDemux * demux,
   gst_event_store_init (&track->sticky_events);
 
   track->waiting_add = TRUE;
-  track->waiting_del_level = 0;
 
   /* We have no fragment duration yet, so the buffering threshold is just the
    * low watermark in time for now */

subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.c

@@ -3799,9 +3799,14 @@ restart:
   }
 
   /* Store global output position */
-  if (global_output_position != GST_CLOCK_STIME_NONE)
+  if (global_output_position != GST_CLOCK_STIME_NONE) {
     demux->priv->global_output_position = global_output_position;
 
+    /* And see if any streams need to be woken for more input */
+    gst_adaptive_demux_period_check_input_wakeup_locked (demux->input_period,
+        global_output_position);
+  }
+
   if (need_restart)
     goto restart;
 

subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.h

@@ -210,9 +210,6 @@ struct _GstAdaptiveDemuxTrack
   /* The track received EOS */
   gboolean eos;
 
-  /* Level to wait until download can commence */
-  GstClockTime waiting_del_level;
-
   /* Input segment and time (in running time) */
   GstSegment input_segment;
   GstClockTimeDiff input_time;
@@ -324,6 +321,9 @@ struct _GstAdaptiveDemux2Stream
   GstAdaptiveDemux2StreamState state;
   guint pending_cb_id;
   gboolean download_active;
+  /* The (global output) time at which this stream should be woken
+   * to download more input */
+  GstClockTimeDiff next_input_wakeup_time;
 
   guint last_status_code;
 
@@ -395,6 +395,11 @@ struct _GstAdaptiveDemuxPeriod
 
   /* Whether tracks were changed and need re-matching against outputs */
   gboolean tracks_changed;
+
+  /* The time at which to wake up input streams for more
+   * data - the earliest of all waiting input stream thresholds,
+   * or GST_CLOCK_STIME_NONE if noone is waiting */
+  GstClockTimeDiff next_input_wakeup_time;
 };
 
 /**