hlsdemux2/preloader: Implement basic request handling

Implement fulfilment of HTTP requests from the active preload downloads by
finding any preload request that can provide the requested data and feeding
bytes from the internal DownloadRequest to the caller provided target
DownloadRequest.

Doesn't yet calculate timestamps to make the target request have a sensible
apparent bitrate.

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

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

@@ -25,6 +25,9 @@
 
 #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
 
@@ -33,7 +36,16 @@ struct _GstHLSDemuxPreloadRequest
 {
   GstHLSDemuxPreloader *preloader;      /* Parent preloader */
   GstM3U8PreloadHint *hint;
+
   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 */
+
+  /* FIXME: Support multiple target requests? I don't think that can happen in practice,
+   * since we only download one segment at a time, and MAP requests are distinct from PART requests. */
+  guint64 target_cur_offset;    /* offset of the next delivered target data */
+  DownloadRequest *target_request;
 };
 
 static GstHLSDemuxPreloadRequest *
@@ -52,16 +64,25 @@ gst_hls_demux_preload_request_free (GstHLSDemuxPreloadRequest * req)
 {
   gst_m3u8_preload_hint_unref (req->hint);
 
-  /* The download request must have been cancelled and removed by the preload helper */
+  if (req->download_request != NULL) {
-  g_assert (req->download_request == NULL);
+    /* The download request must have been cancelled by the preload helper */
+    g_assert (req->download_request->in_use == FALSE);
+    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_cancel_request (GstHLSDemuxPreloader *
+static void gst_hls_demux_preloader_release_request (GstHLSDemuxPreloader *
-    preloader, GstHLSDemuxPreloadRequest * req);
+    preloader, GstHLSDemuxPreloadRequest * preload_req,
+    gboolean cancel_download);
 
 GstHLSDemuxPreloader *
 gst_hls_demux_preloader_new (DownloadHelper * download_helper)
@@ -101,7 +122,7 @@ gst_hls_demux_preloader_load (GstHLSDemuxPreloader * preloader,
         return;                 /* Nothing to do */
       }
 
-      gst_hls_demux_preloader_cancel_request (preloader, req);
+      gst_hls_demux_preloader_release_request (preloader, req, TRUE);
       g_ptr_array_remove_index_fast (preloader->active_preloads, idx);
       break;
     }
@@ -116,7 +137,7 @@ gst_hls_demux_preloader_load (GstHLSDemuxPreloader * preloader,
     g_ptr_array_add (preloader->active_preloads, req);
   } else {
     /* Discard failed request */
-    gst_hls_demux_preloader_cancel_request (preloader, req);
+    gst_hls_demux_preloader_release_request (preloader, req, TRUE);
   }
 }
 
@@ -130,7 +151,7 @@ gst_hls_demux_preloader_cancel (GstHLSDemuxPreloader * preloader,
     GstHLSDemuxPreloadRequest *req =
         g_ptr_array_index (preloader->active_preloads, idx);
     if (hint_types & req->hint->hint_type) {
-      gst_hls_demux_preloader_cancel_request (preloader, req);
+      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 */
     }
@@ -139,10 +160,122 @@ gst_hls_demux_preloader_cancel (GstHLSDemuxPreloader * preloader,
   }
 }
 
+/* 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;
+
+  if (input_is_finished)
+    preload_req->download_is_finished = TRUE;
+  else
+    input_is_finished = preload_req->download_is_finished;
+
+  /* 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;
+
+    download_request_lock (preload_req->target_request);
+    download_request_lock (preload_req->download_request);
+
+    DownloadRequestState target_state = preload_req->download_request->state;
+
+    /* Transfer the http status code */
+    preload_req->target_request->status_code =
+        preload_req->download_request->status_code;
+
+    GstBuffer *target_buf =
+        download_request_take_buffer_range (preload_req->download_request,
+        preload_req->target_cur_offset,
+        preload_req->target_request->range_end);
+
+    if (target_buf != NULL) {
+      DownloadRequest *req = preload_req->target_request;
+
+      /* 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), req->uri, req->range_start,
+          req->range_end);
+
+      download_request_add_buffer (req, target_buf);
+      despatch_progress = TRUE; /* Added a buffer, despatch progress callback */
+
+      if (req->range_end != -1
+          && preload_req->target_cur_offset > 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",
+              req->uri, req->range_start, req->range_end);
+        }
+        output_is_finished = TRUE;
+      }
+    }
+
+    /* Update the target request's state, which may have been adjusted from the
+     * input request's state */
+    preload_req->target_request->state = target_state;
+
+    /* FIXME: Transfer timing from the input download as best we can, so the receiver can
+     * calculate bitrates */
+
+    /* We're done with the input download request . */
+    download_request_unlock (preload_req->download_request);
+
+    if (output_is_finished) {
+      DownloadRequest *req = preload_req->target_request;
+
+      GST_DEBUG ("Finishing target preload request uri: %s, start: %"
+          G_GINT64_FORMAT " end: %" G_GINT64_FORMAT, req->uri, req->range_start,
+          req->range_end);
+
+      download_request_despatch_completion (req);
+      download_request_unlock (req);
+
+      download_request_unref (req);
+      preload_req->target_request = NULL;
+    } else if (despatch_progress) {
+      DownloadRequest *req = preload_req->target_request;
+      download_request_despatch_progress (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 (preload_req->download_request == NULL
+        || download_request_get_bytes_available (preload_req->download_request)
+        == 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
@@ -152,7 +285,13 @@ on_download_error (DownloadRequest * request, DownloadRequestState state,
   GstM3U8PreloadHint *hint = preload_req->hint;
   GST_DEBUG ("preload type %d uri: %s download error", hint->hint_type,
       hint->uri);
-  GST_FIXME ("How to handle failed preload request?");
+
+  /* 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
@@ -160,8 +299,15 @@ on_download_progress (DownloadRequest * request, DownloadRequestState state,
     GstHLSDemuxPreloadRequest * preload_req)
 {
   GstM3U8PreloadHint *hint = preload_req->hint;
-  GST_DEBUG ("preload type %d uri: %s download progress", hint->hint_type,
+
-      hint->uri);
+  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
@@ -169,8 +315,14 @@ on_download_complete (DownloadRequest * request, DownloadRequestState state,
     GstHLSDemuxPreloadRequest * preload_req)
 {
   GstM3U8PreloadHint *hint = preload_req->hint;
-  GST_DEBUG ("preload type %d uri: %s download complete", hint->hint_type,
+  GST_DEBUG ("preload type %d uri: %s download complete. position %"
-      hint->uri);
+      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
@@ -205,24 +357,130 @@ gst_hls_demux_preloader_submit (GstHLSDemuxPreloader * preloader,
     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_cancel_request (GstHLSDemuxPreloader * preloader,
+gst_hls_demux_preloader_release_request (GstHLSDemuxPreloader * preloader,
-    GstHLSDemuxPreloadRequest * preload_req)
+    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);
 
       downloadhelper_cancel_request (preloader->download_helper,
           preload_req->download_request);
-    download_request_unref (preload_req->download_request);
-    preload_req->download_request = NULL;
     }
+  }
+
   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;
+}

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

@@ -42,5 +42,7 @@ void gst_hls_demux_preloader_free (GstHLSDemuxPreloader *preloader);
 void gst_hls_demux_preloader_load (GstHLSDemuxPreloader *preloader, GstM3U8PreloadHint *hint, const gchar *referrer_uri);
 void gst_hls_demux_preloader_cancel (GstHLSDemuxPreloader *preloader, GstM3U8PreloadHintType hint_types);
 
+gboolean gst_hls_demux_preloader_provide_request (GstHLSDemuxPreloader *preloader, DownloadRequest *target_req);
+
 G_END_DECLS
 #endif /* __GST_HLS_DEMUX_PRELOADER_H__ */