From a03f3df626ed965a102c2db0ea64a579471980d0 Mon Sep 17 00:00:00 2001 From: Jan Schmidt Date: Fri, 16 Sep 2022 02:01:58 +1000 Subject: [PATCH] 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: --- .../adaptivedemux2/gstadaptivedemux-period.c | 37 ++++++++ .../adaptivedemux2/gstadaptivedemux-private.h | 2 + .../adaptivedemux2/gstadaptivedemux-stream.c | 92 +++++++++++++++---- .../adaptivedemux2/gstadaptivedemux-track.c | 70 ++++++-------- .../ext/adaptivedemux2/gstadaptivedemux.c | 7 +- .../ext/adaptivedemux2/gstadaptivedemux.h | 11 ++- 6 files changed, 152 insertions(+), 67 deletions(-) diff --git a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-period.c b/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-period.c index 412b551af2..2adf12db5a 100644 --- a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-period.c +++ b/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; +} diff --git a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-private.h b/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-private.h index 2fd4fe3458..17f8778b3c 100644 --- a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-private.h +++ b/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 diff --git a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-stream.c b/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-stream.c index b56f1e9387..0594da2be4 100644 --- a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-stream.c +++ b/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 diff --git a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-track.c b/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-track.c index 590413709c..b620772313 100644 --- a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux-track.c +++ b/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 */ diff --git a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.c b/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.c index f7eae666a7..de51bb1855 100644 --- a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.c +++ b/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; diff --git a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.h b/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.h index 3a5b5f640b..ea4a473146 100644 --- a/subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemux.h +++ b/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; }; /** -- 2.34.1