gstreamer/subprojects/gst-plugins-good/ext/adaptivedemux2/hls/gsthlsdemux-preloader.c

599 lines
23 KiB
C
Raw Normal View History

/* GStreamer
Copyright (C) 2022 Jan Schmidt <jan@centricular.com>
*
* gsthlsdemux-preloader.c:
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifdef HAVE_CONFIG_H
# include "config.h"
#endif
#include "gsthlsdemux-preloader.h"
/* Everything is called from the scheduler thread, including
* download handling callbacks */
GST_DEBUG_CATEGORY_EXTERN (gst_hls_demux2_debug);
#define GST_CAT_DEFAULT gst_hls_demux2_debug
typedef struct _GstHLSDemuxPreloadRequest GstHLSDemuxPreloadRequest;
struct _GstHLSDemuxPreloadRequest
{
GstHLSDemuxPreloader *preloader; /* Parent preloader */
GstM3U8PreloadHint *hint;
/* Incoming download tracking for the resource */
DownloadRequest *download_request;
gboolean download_is_finished; /* TRUE if the input download request completed / failed */
guint64 download_cur_offset; /* offset of the next expected received data */
guint64 download_content_length; /* Content length (filled in when response headers arrive) */
GstClockTime request_latency; /* original HTTP request to data latency */
GstClockTime download_first_data_time; /* Arrival timestamp of the first data in the download chunk */
guint64 download_first_data_offset; /* First data byte offset of the download chunk */
/* Target tracking for the stream download to deliver data blocks to */
/* Each active preload only needs one target to output to at a time,
* since we only download one segment at a time, and MAP requests are distinct from PART requests,
* so 1 preload = 1 download request by the stream */
guint64 target_cur_offset; /* offset of the next delivered target data */
DownloadRequest *target_request;
};
static GstHLSDemuxPreloadRequest *
gst_hls_demux_preload_request_new (GstHLSDemuxPreloader * preloader,
GstM3U8PreloadHint * hint)
{
GstHLSDemuxPreloadRequest *req = g_new0 (GstHLSDemuxPreloadRequest, 1);
req->preloader = preloader;
req->hint = gst_m3u8_preload_hint_ref (hint);
req->request_latency = GST_CLOCK_TIME_NONE;
req->download_first_data_time = GST_CLOCK_TIME_NONE;
req->download_first_data_offset = GST_BUFFER_OFFSET_NONE;
return req;
};
static void
gst_hls_demux_preload_request_free (GstHLSDemuxPreloadRequest * req)
{
gst_m3u8_preload_hint_unref (req->hint);
if (req->download_request != NULL) {
/* The download request must have been cancelled by the preload helper,
* but cancellation is async, so we can't verify */
download_request_unref (req->download_request);
}
if (req->target_request != NULL) {
download_request_unref (req->target_request);
}
g_free (req);
};
static gboolean
gst_hls_demux_preloader_submit (GstHLSDemuxPreloader * preloader,
GstHLSDemuxPreloadRequest * preload_req, const gchar * referrer_uri);
static void gst_hls_demux_preloader_release_request (GstHLSDemuxPreloader *
preloader, GstHLSDemuxPreloadRequest * preload_req,
gboolean cancel_download);
GstHLSDemuxPreloader *
gst_hls_demux_preloader_new (DownloadHelper * download_helper)
{
GstHLSDemuxPreloader *preloader = g_new0 (GstHLSDemuxPreloader, 1);
preloader->download_helper = download_helper;
preloader->active_preloads = g_ptr_array_new ();
return preloader;
}
void
gst_hls_demux_preloader_free (GstHLSDemuxPreloader * preloader)
{
gst_hls_demux_preloader_cancel (preloader, M3U8_PRELOAD_HINT_ALL);
g_ptr_array_free (preloader->active_preloads, TRUE);
g_free (preloader);
}
void
gst_hls_demux_preloader_load (GstHLSDemuxPreloader * preloader,
GstM3U8PreloadHint * hint, const gchar * referrer_uri)
{
/* Check if we have an active preload already for this hint */
guint idx;
for (idx = 0; idx < preloader->active_preloads->len; idx++) {
GstHLSDemuxPreloadRequest *req =
g_ptr_array_index (preloader->active_preloads, idx);
if (hint->hint_type == req->hint->hint_type) {
/* We already have an active hint of this type. If this new one is different, cancel
* the active preload before starting this one */
if (gst_m3u8_preload_hint_equal (hint, req->hint)) {
GST_LOG ("Ignoring pre-existing preload of type %d uri: %s, range:%"
G_GINT64_FORMAT " size %" G_GINT64_FORMAT, hint->hint_type,
hint->uri, hint->offset, hint->size);
return; /* Nothing to do */
}
gst_hls_demux_preloader_release_request (preloader, req, TRUE);
g_ptr_array_remove_index_fast (preloader->active_preloads, idx);
break;
}
}
/* If we get here, then there's no preload of this type. Create one */
GstHLSDemuxPreloadRequest *req =
gst_hls_demux_preload_request_new (preloader, hint);
/* Submit the request */
if (gst_hls_demux_preloader_submit (preloader, req, referrer_uri)) {
g_ptr_array_add (preloader->active_preloads, req);
} else {
/* Discard failed request */
gst_hls_demux_preloader_release_request (preloader, req, TRUE);
}
}
void
gst_hls_demux_preloader_cancel (GstHLSDemuxPreloader * preloader,
GstM3U8PreloadHintType hint_types)
{
/* Go through the active downloads and remove/cancel any with the matching type */
guint idx;
for (idx = 0; idx < preloader->active_preloads->len;) {
GstHLSDemuxPreloadRequest *req =
g_ptr_array_index (preloader->active_preloads, idx);
if (hint_types & req->hint->hint_type) {
gst_hls_demux_preloader_release_request (preloader, req, TRUE);
g_ptr_array_remove_index_fast (preloader->active_preloads, idx);
continue; /* Don't increment idx++, as we just removed an item */
}
idx++;
}
}
/* This function transfers any available data to the target request, and possibly
* completes it and removes it from the preload */
static void
gst_hls_demux_preloader_despatch (GstHLSDemuxPreloadRequest * preload_req,
gboolean input_is_finished)
{
GstHLSDemuxPreloader *preloader = preload_req->preloader;
DownloadRequest *download_req = preload_req->download_request;
if (input_is_finished)
preload_req->download_is_finished = TRUE;
else
input_is_finished = preload_req->download_is_finished;
download_request_lock (download_req);
/* Update timestamp tracking */
if (preload_req->request_latency == GST_CLOCK_TIME_NONE) {
if (GST_CLOCK_TIME_IS_VALID (download_req->download_request_time) &&
GST_CLOCK_TIME_IS_VALID (download_req->download_start_time)) {
preload_req->request_latency =
download_req->download_start_time -
download_req->download_request_time;
}
}
if (preload_req->download_first_data_time == GST_CLOCK_TIME_NONE &&
download_request_get_bytes_available (download_req) > 0) {
/* Got the first data of this download burst */
preload_req->download_first_data_time = download_req->download_start_time;
preload_req->download_first_data_offset =
download_request_get_cur_offset (download_req);
}
download_request_unlock (download_req);
/* If there is a target request, see if any of our data should be
* transferred to it, and if it should be despatched as complete */
if (preload_req->target_request != NULL) {
gboolean output_is_finished = input_is_finished;
gboolean despatch_progress = FALSE;
DownloadRequest *target_req = preload_req->target_request;
download_request_lock (target_req);
download_request_lock (download_req);
DownloadRequestState target_state = download_req->state;
/* Transfer the http status code */
target_req->status_code = download_req->status_code;
GstBuffer *target_buf = download_request_take_buffer_range (download_req,
preload_req->target_cur_offset,
target_req->range_end);
if (target_buf != NULL) {
/* Deliver data to the target, and update our tracked output position */
preload_req->target_cur_offset =
GST_BUFFER_OFFSET (target_buf) + gst_buffer_get_size (target_buf);
GST_LOG ("Adding %" G_GSIZE_FORMAT " bytes at offset %" G_GUINT64_FORMAT
" to target download request uri %s range %" G_GINT64_FORMAT " - %"
G_GINT64_FORMAT, gst_buffer_get_size (target_buf),
GST_BUFFER_OFFSET (target_buf), target_req->uri,
target_req->range_start, target_req->range_end);
download_request_add_buffer (target_req, target_buf);
despatch_progress = TRUE; /* Added a buffer, despatch progress callback */
/* Transfer timing from the input download as best we can, so the receiver can
* calculate bitrates. If all preload requests filled one target download,
* we could just transfer the timestamps, but to handle the case of an
* ongoing chunked connection needs fancier accounting based on the
* arrival times of each data burst */
if (target_req->download_start_time == GST_CLOCK_TIME_NONE) {
if (preload_req->download_first_data_time >
preload_req->request_latency) {
target_req->download_request_time =
preload_req->download_first_data_time -
preload_req->request_latency;
} else {
target_req->download_request_time = 0;
}
target_req->download_start_time = preload_req->download_first_data_time;
target_req->download_newest_data_time =
download_req->download_newest_data_time;
}
if (target_req->range_end != -1
&& preload_req->target_cur_offset > target_req->range_end) {
/* We've delivered all data to satisfy the requested byte range - the target request is complete */
if (target_state == DOWNLOAD_REQUEST_STATE_LOADING) {
target_state = DOWNLOAD_REQUEST_STATE_COMPLETE;
GST_LOG ("target download request uri %s range %" G_GINT64_FORMAT
" - %" G_GINT64_FORMAT " is fully satisfied. Completing",
target_req->uri, target_req->range_start, target_req->range_end);
}
output_is_finished = TRUE;
/* If there's unconsumed data left in the input download, then update
* our variable that tracks the first data arrival time in in a prorata
* fashion (because there's more partial segment data already downloaded
* and we need to preserve a reasonable bitrate estimate. If there's no
* data, but the connection is continuing, then it's returned to a blocking
* read state that'll send more data in the future when
* a new live segment becomes available, so reset our variable as if that
* download was starting again */
guint64 data_avail =
download_request_get_bytes_available (download_req);
if (data_avail > 0) {
/* burst first data offset must have been set by now */
g_assert (preload_req->download_first_data_offset !=
GST_BUFFER_OFFSET_NONE);
/* Calculate how long it took to download the data we have output/discarded
* based on the average bitrate so far.
* time_to_download = total_download_time * consumed_bytes / total_download_bytes */
guint64 new_cur_offset =
download_request_get_cur_offset (download_req);
GstClockTime data_time_offset =
gst_util_uint64_scale (download_req->download_newest_data_time -
preload_req->download_first_data_time,
new_cur_offset - preload_req->download_first_data_offset,
new_cur_offset + data_avail -
preload_req->download_first_data_offset);
preload_req->download_first_data_time += data_time_offset;
preload_req->download_first_data_offset = new_cur_offset;
GST_LOG ("Advancing request timing tracking by %" GST_TIMEP_FORMAT
" to time %" GST_TIMEP_FORMAT " @ offset %" G_GUINT64_FORMAT,
&data_time_offset,
&preload_req->download_first_data_time,
preload_req->download_first_data_offset);
/* Say that this target download finished when the first
* byte of the remaining data arrived */
target_req->download_end_time = preload_req->download_first_data_time;
} else {
/* Reset the download start time */
preload_req->download_first_data_time = GST_CLOCK_TIME_NONE;
preload_req->download_first_data_offset = GST_BUFFER_OFFSET_NONE;
/* Say that this request finished when the most recent data arrived */
target_req->download_end_time =
download_req->download_newest_data_time;
}
}
}
if (input_is_finished
&& target_req->download_end_time == GST_CLOCK_TIME_NONE) {
/* No download end time was set yet - use the input download end time */
target_req->download_end_time = download_req->download_end_time;
}
/* Update the target request's state, which may have been adjusted from the
* input request's state */
target_req->state = target_state;
if (target_req->headers == NULL && download_req->headers != NULL) {
target_req->headers = gst_structure_copy (download_req->headers);
}
if (target_req->redirect_uri == NULL && download_req->redirect_uri != NULL) {
target_req->redirect_uri = g_strdup (download_req->redirect_uri);
target_req->redirect_permanent = download_req->redirect_permanent;
}
/* We're done with the input download request . */
download_request_unlock (download_req);
if (output_is_finished) {
GST_DEBUG ("Finishing target preload request uri: %s, start: %"
G_GINT64_FORMAT " end: %" G_GINT64_FORMAT, target_req->uri,
target_req->range_start, target_req->range_end);
download_request_despatch_completion (target_req);
download_request_unlock (target_req);
download_request_unref (target_req);
preload_req->target_request = NULL;
} else if (despatch_progress) {
download_request_despatch_progress (target_req);
}
/* Unlock if the target request didn't get released above */
if (preload_req->target_request != NULL) {
download_request_unlock (preload_req->target_request);
}
}
if (input_is_finished) {
if (download_req == NULL
|| download_request_get_bytes_available (download_req)
== 0) {
GstM3U8PreloadHint *hint = preload_req->hint;
GST_DEBUG ("Removing finished+drained preload type %d uri: %s, start: %"
G_GINT64_FORMAT " size: %" G_GINT64_FORMAT, hint->hint_type,
hint->uri, hint->offset, hint->size);
/* The incoming request is complete and the data is drained. Remove this preload request from the list */
g_ptr_array_remove_fast (preloader->active_preloads, preload_req);
gst_hls_demux_preloader_release_request (preloader, preload_req, FALSE);
}
}
}
static void
on_download_cancellation (DownloadRequest * request, DownloadRequestState state,
GstHLSDemuxPreloadRequest * preload_req)
{
gst_hls_demux_preloader_despatch (preload_req, TRUE);
}
static void
on_download_error (DownloadRequest * request, DownloadRequestState state,
GstHLSDemuxPreloadRequest * preload_req)
{
GstM3U8PreloadHint *hint = preload_req->hint;
GST_DEBUG ("preload type %d uri: %s download error", hint->hint_type,
hint->uri);
/* FIXME: Should we attempt to re-request a preload? Should we check if
* any part was transferred to the target request already? Should we
* attempt to request a byte range with a new start position if we
* already despatched data to other requests?
*/
gst_hls_demux_preloader_despatch (preload_req, TRUE);
}
static void
on_download_progress (DownloadRequest * request, DownloadRequestState state,
GstHLSDemuxPreloadRequest * preload_req)
{
GstM3U8PreloadHint *hint = preload_req->hint;
GST_DEBUG ("preload type %d uri: %s download progress. position %"
G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT " bytes", hint->hint_type,
hint->uri,
preload_req->download_cur_offset +
download_request_get_bytes_available (request), request->content_length);
preload_req->download_content_length = request->content_length;
gst_hls_demux_preloader_despatch (preload_req, FALSE);
}
static void
on_download_complete (DownloadRequest * request, DownloadRequestState state,
GstHLSDemuxPreloadRequest * preload_req)
{
GstM3U8PreloadHint *hint = preload_req->hint;
GST_DEBUG ("preload type %d uri: %s download complete. position %"
G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT " bytes", hint->hint_type,
hint->uri,
preload_req->download_cur_offset +
download_request_get_bytes_available (request), request->content_length);
preload_req->download_content_length = request->content_length;
gst_hls_demux_preloader_despatch (preload_req, TRUE);
}
static gboolean
gst_hls_demux_preloader_submit (GstHLSDemuxPreloader * preloader,
GstHLSDemuxPreloadRequest * preload_req, const gchar * referrer_uri)
{
g_assert (preload_req->download_request == NULL);
DownloadRequest *download_req = download_request_new ();
GstM3U8PreloadHint *hint = preload_req->hint;
/* Configure our download request */
gint64 end = RFC8673_LAST_BYTE_POS;
if (hint->size > 0) {
end = hint->offset + hint->size - 1;
}
download_request_set_uri (download_req, hint->uri, hint->offset, end);
download_request_set_callbacks (download_req,
(DownloadRequestEventCallback) on_download_complete,
(DownloadRequestEventCallback) on_download_error,
(DownloadRequestEventCallback) on_download_cancellation,
(DownloadRequestEventCallback) on_download_progress, preload_req);
GST_DEBUG ("Submitting preload type %d uri: %s, range:%" G_GINT64_FORMAT
" - %" G_GINT64_FORMAT, hint->hint_type, hint->uri, hint->offset, end);
if (!downloadhelper_submit_request (preloader->download_helper,
referrer_uri, DOWNLOAD_FLAG_NONE, download_req, NULL)) {
/* Abandon the request */
download_request_unref (download_req);
return FALSE;
}
/* Store the current read offset */
preload_req->download_cur_offset = hint->offset;
preload_req->download_request = download_req;
preload_req->download_is_finished = FALSE;
return TRUE;
}
static void
gst_hls_demux_preloader_release_request (GstHLSDemuxPreloader * preloader,
GstHLSDemuxPreloadRequest * preload_req, gboolean cancel_download)
{
if (preload_req->download_request) {
if (cancel_download) {
GstM3U8PreloadHint *hint = preload_req->hint;
GST_DEBUG ("Cancelling preload type %d uri: %s, range start:%"
G_GINT64_FORMAT " size %" G_GINT64_FORMAT, hint->hint_type, hint->uri,
hint->offset, hint->size);
/* We don't want any callbacks to happen after we cancel here */
download_request_set_callbacks (preload_req->download_request,
NULL, NULL, NULL, NULL, NULL);
downloadhelper_cancel_request (preloader->download_helper,
preload_req->download_request);
}
}
gst_hls_demux_preload_request_free (preload_req);
}
/* See if we can satisfy a download request from a preload, and fulfil it if so.
* There are several cases:
* * The URI and range exactly match one of our preloads -> OK
* * The URI matches, and the requested range is a subset of the preload -> OK
* * The URI matches, but the requested range is outside what's available in the preload
* and can't be provided.
*
* Within those options, there are sub-possibilities:
* * The preload request is ongoing. It might have enough data already to completely provide
* the requested range.
* * The preload request is ongoing, but has already moved past the requested range (no longer available)
* * The preload request is ongoing, will feed data to the target req as it arrives
* * The preload request is complete already, so can either provide the requested range or not, but
* also needs to mark the target_req as completed once it has passed the required data.
*/
gboolean
gst_hls_demux_preloader_provide_request (GstHLSDemuxPreloader * preloader,
DownloadRequest * target_req)
{
guint idx;
for (idx = 0; idx < preloader->active_preloads->len; idx++) {
GstHLSDemuxPreloadRequest *preload_req =
g_ptr_array_index (preloader->active_preloads, idx);
GstM3U8PreloadHint *hint = preload_req->hint;
if (!g_str_equal (hint->uri, target_req->uri))
continue;
GST_LOG ("Possible matching preload type %d uri: %s, range start:%"
G_GINT64_FORMAT " size %" G_GINT64_FORMAT " (download position %"
G_GUINT64_FORMAT ") for req with range %" G_GINT64_FORMAT " to %"
G_GINT64_FORMAT, hint->hint_type, hint->uri, hint->offset, hint->size,
preload_req->download_cur_offset, target_req->range_start,
target_req->range_end);
if (target_req->range_start > preload_req->download_cur_offset) {
/* This preload request is for a byte range beyond the desired
* position (or something already consumed the target data) */
GST_LOG ("Range start didn't match");
continue;
}
if (target_req->range_end != -1) {
/* The target request does not want the entire rest of the preload
* stream, so check that the end is satisfiable */
gint64 content_length = preload_req->download_content_length;
if (content_length == 0) {
/* We don't have information from the preload download's response headers yet,
* so check against the requested length and error out later if the server
* doesn't provide all the desired response */
if (hint->size != -1)
content_length = hint->size;
}
if (content_length != 0) {
/* We have some idea of the content length. Check if it will provide the requested
* range */
if (target_req->range_end > hint->offset + content_length - 1) {
GST_LOG ("Range end %" G_GINT64_FORMAT " is beyond the end (%"
G_GINT64_FORMAT ") of this preload", target_req->range_end,
hint->offset + content_length - 1);
continue;
}
}
}
GST_DEBUG ("Found a matching preload type %d uri: %s, range start:%"
G_GINT64_FORMAT " size %" G_GINT64_FORMAT, hint->hint_type, hint->uri,
hint->offset, hint->size);
if (preload_req->target_request != NULL) {
DownloadRequest *old_request = preload_req->target_request;
/* Detach the existing target request */
if (old_request != target_req) {
download_request_lock (old_request);
old_request->state = DOWNLOAD_REQUEST_STATE_UNSENT;
download_request_despatch_completion (old_request);
download_request_unlock (old_request);
}
download_request_unref (old_request);
preload_req->target_request = NULL;
}
/* Attach the new target request and despatch any available data */
preload_req->target_cur_offset = target_req->range_start;
preload_req->target_request = download_request_ref (target_req);
download_request_lock (target_req);
target_req->state = DOWNLOAD_REQUEST_STATE_UNSENT;
download_request_begin_download (target_req);
download_request_unlock (target_req);
gst_hls_demux_preloader_despatch (preload_req, FALSE);
return TRUE;
}
return FALSE;
}