* information.
*/
+/*
+MT safety.
+The following rules were observed while implementing MT safety in adaptive demux:
+1. If a variable is accessed from multiple threads and at least one thread
+writes to it, then all the accesses needs to be done from inside a critical section.
+2. If thread A wants to join thread B then at the moment it calls gst_task_join
+it must not hold any mutexes that thread B might take.
+
+Adaptive demux API can be called from several threads. More, adaptive demux
+starts some threads to monitor the download of fragments. In order to protect
+accesses to shared variables (demux and streams) all the API functions that
+can be run in different threads will need to get a mutex (manifest_lock)
+when they start and release it when they end. Because some of those functions
+can indirectly call other API functions (eg they can generate events or messages
+that are processed in the same thread) the manifest_lock must be recursive.
+
+The manifest_lock will serialize the public API making access to shared
+variables safe. But some of these functions will try at some moment to join
+threads created by adaptive demux, or to change the state of src elements
+(which will block trying to join the src element streaming thread). Because
+of rule 2, those functions will need to release the manifest_lock during the
+call of gst_task_join. During this time they can be interrupted by other API calls.
+For example, during the precessing of a seek event, gst_adaptive_demux_stop_tasks
+is called and this will join all threads. In order to prevent interruptions
+during such period, all the API functions will also use a second lock: api_lock.
+This will be taken at the beginning of the function and released at the end,
+but this time this lock will not be temporarily released during join.
+This lock will be used only by API calls (not by gst_adaptive_demux_stream_download_loop
+or gst_adaptive_demux_updates_loop or _src_chain or _src_event) so it is safe
+to hold it while joining the threads or changing the src element state. The
+api_lock will serialise all external requests to adaptive demux. In order to
+avoid deadlocks, if a function needs to acquire both manifest and api locks,
+the api_lock will be taken first and the manifest_lock second.
+
+By using the api_lock a thread is protected against other API calls. But when
+temporarily dropping the manifest_lock, it will be vulnerable to changes from
+threads that use only the manifest_lock and not the api_lock. These threads run
+one of the following functions: gst_adaptive_demux_stream_download_loop,
+gst_adaptive_demux_updates_loop, _src_chain, _src_event. In order to guarantee
+that all operations during an API call are not impacted by other writes, the
+above mentioned functions must check a cancelled flag every time they reacquire
+the manifest_lock. If the flag is set, they must exit immediately, without
+performing any changes on the shared data. In this way, an API call (eg seek
+request) can set the cancel flag before releasing the manifest_lock and be sure
+that the demux object and its streams are not changed by anybody else.
+*/
+
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#define DEFAULT_CONNECTION_SPEED 0
#define DEFAULT_BITRATE_LIMIT 0.8
+#define GST_MANIFEST_GET_LOCK(d) (&(GST_ADAPTIVE_DEMUX_CAST(d)->priv->manifest_lock))
+#define GST_MANIFEST_LOCK(d) g_rec_mutex_lock (GST_MANIFEST_GET_LOCK (d));
+#define GST_MANIFEST_UNLOCK(d) g_rec_mutex_unlock (GST_MANIFEST_GET_LOCK (d));
+
+#define GST_API_GET_LOCK(d) (&(GST_ADAPTIVE_DEMUX_CAST(d)->priv->api_lock))
+#define GST_API_LOCK(d) g_mutex_lock (GST_API_GET_LOCK (d));
+#define GST_API_UNLOCK(d) g_mutex_unlock (GST_API_GET_LOCK (d));
+
enum
{
PROP_0,
struct _GstAdaptiveDemuxPrivate
{
- GstAdapter *input_adapter;
- gboolean have_manifest;
+ GstAdapter *input_adapter; /* protected by manifest_lock */
+ gboolean have_manifest; /* protected by manifest_lock */
- GstUriDownloader *downloader;
+ GList *old_streams; /* protected by manifest_lock */
- GList *old_streams;
-
- GstTask *updates_task;
+ GstTask *updates_task; /* MT safe */
GRecMutex updates_lock;
GMutex updates_timed_lock;
- GCond updates_timed_cond;
- gboolean stop_updates_task;
+ GCond updates_timed_cond; /* protected by updates_timed_lock */
+ gboolean stop_updates_task; /* protected by updates_timed_lock */
+
+ /* used only from updates_task, no need to protect it */
gint update_failed_count;
- gint64 next_update;
+ guint32 segment_seqnum; /* protected by manifest_lock */
+
+ /* main lock used to protect adaptive demux and all its streams.
+ * It serializes the adaptive demux public API.
+ */
+ GRecMutex manifest_lock;
- gboolean exposing;
- guint32 segment_seqnum;
+ /* condition to wait for manifest updates on a live stream.
+ * In order to signal the manifest_cond, the caller needs to hold both
+ * manifest_lock and manifest_update_lock (taken in this order)
+ */
+ GCond manifest_cond;
+ GMutex manifest_update_lock;
+
+ GMutex api_lock;
};
static GstBinClass *parent_class = NULL;
static GstFlowReturn
gst_adaptive_demux_stream_push_event (GstAdaptiveDemuxStream * stream,
GstEvent * event);
-static void gst_adaptive_demux_stream_download_wait (GstAdaptiveDemuxStream *
- stream, GstClockTime time_diff);
static void gst_adaptive_demux_start_tasks (GstAdaptiveDemux * demux);
static void gst_adaptive_demux_stop_tasks (GstAdaptiveDemux * demux);
static GstFlowReturn
gst_adaptive_demux_stream_finish_fragment_default (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream);
+static GstFlowReturn
+gst_adaptive_demux_stream_advance_fragment_unlocked (GstAdaptiveDemux * demux,
+ GstAdaptiveDemuxStream * stream, GstClockTime duration);
/* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
{
GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX (object);
+ GST_API_LOCK (demux);
+ GST_MANIFEST_LOCK (demux);
+
switch (prop_id) {
case PROP_LOOKBACK_FRAGMENTS:
demux->num_lookback_fragments = g_value_get_uint (value);
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
+
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
}
static void
{
GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX (object);
+ GST_MANIFEST_LOCK (demux);
+
switch (prop_id) {
case PROP_LOOKBACK_FRAGMENTS:
g_value_set_uint (value, demux->num_lookback_fragments);
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
+
+ GST_MANIFEST_UNLOCK (demux);
}
static void
g_mutex_init (&demux->priv->updates_timed_lock);
g_cond_init (&demux->priv->updates_timed_cond);
- g_cond_init (&demux->manifest_cond);
- g_mutex_init (&demux->manifest_lock);
+ g_cond_init (&demux->priv->manifest_cond);
+ g_mutex_init (&demux->priv->manifest_update_lock);
+
+ g_rec_mutex_init (&demux->priv->manifest_lock);
+ g_mutex_init (&demux->priv->api_lock);
pad_template =
gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "sink");
g_mutex_clear (&priv->updates_timed_lock);
g_cond_clear (&priv->updates_timed_cond);
- g_cond_clear (&demux->manifest_cond);
+ g_mutex_clear (&demux->priv->manifest_update_lock);
+ g_cond_clear (&demux->priv->manifest_cond);
g_object_unref (priv->updates_task);
g_rec_mutex_clear (&priv->updates_lock);
- g_mutex_clear (&demux->manifest_lock);
+ g_rec_mutex_clear (&demux->priv->manifest_lock);
+ g_mutex_clear (&demux->priv->api_lock);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (element);
GstStateChangeReturn result = GST_STATE_CHANGE_FAILURE;
+ GST_API_LOCK (demux);
+ GST_MANIFEST_LOCK (demux);
+
switch (transition) {
case GST_STATE_CHANGE_PAUSED_TO_READY:
gst_adaptive_demux_reset (demux);
result = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
return result;
}
{
GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent);
GstAdaptiveDemuxClass *demux_class;
+ gboolean ret;
switch (event->type) {
case GST_EVENT_FLUSH_STOP:
+ GST_API_LOCK (demux);
+ GST_MANIFEST_LOCK (demux);
+
gst_adaptive_demux_reset (demux);
- break;
+
+ ret = gst_pad_event_default (pad, parent, event);
+
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
+
+ return ret;
case GST_EVENT_EOS:{
GstQuery *query;
gboolean query_res;
gsize available;
GstBuffer *manifest_buffer;
+ GST_API_LOCK (demux);
+ GST_MANIFEST_LOCK (demux);
+
demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
+
available = gst_adapter_available (demux->priv->input_adapter);
if (available == 0) {
GST_WARNING_OBJECT (demux, "Received EOS without a manifest.");
- break;
+ ret = gst_pad_event_default (pad, parent, event);
+
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
+
+ return ret;
}
GST_DEBUG_OBJECT (demux, "Got EOS on the sink pad: manifest fetched");
* uris */
query = gst_query_new_uri ();
query_res = gst_pad_peer_query (pad, query);
- GST_MANIFEST_LOCK (demux);
if (query_res) {
gchar *uri, *redirect_uri;
gboolean permanent;
}
GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
gst_event_unref (event);
return ret;
GstBuffer * buffer)
{
GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent);
+
+ GST_MANIFEST_LOCK (demux);
+
gst_adapter_push (demux->priv->input_adapter, buffer);
GST_INFO_OBJECT (demux, "Received manifest buffer, total size is %i bytes",
(gint) gst_adapter_available (demux->priv->input_adapter));
+ GST_MANIFEST_UNLOCK (demux);
return GST_FLOW_OK;
}
+/* must be called with manifest_lock taken */
static void
gst_adaptive_demux_reset (GstAdaptiveDemux * demux)
{
GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
GList *iter;
+ GList *old_streams;
+ GstEvent *eos;
+
+ /* take ownership of old_streams before releasing the manifest_lock in
+ * gst_adaptive_demux_stop_tasks
+ */
+ old_streams = demux->priv->old_streams;
+ demux->priv->old_streams = NULL;
gst_adaptive_demux_stop_tasks (demux);
gst_uri_downloader_reset (demux->downloader);
if (klass->reset)
klass->reset (demux);
+ eos = gst_event_new_eos ();
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
- if (stream->pad)
+ if (stream->pad) {
+ gst_pad_push_event (stream->pad, gst_event_ref (eos));
+ gst_pad_set_active (stream->pad, FALSE);
+
gst_element_remove_pad (GST_ELEMENT_CAST (demux), stream->pad);
+ }
gst_adaptive_demux_stream_free (stream);
}
+ gst_event_unref (eos);
g_list_free (demux->streams);
demux->streams = NULL;
- if (demux->priv->old_streams) {
- g_list_free_full (demux->priv->old_streams,
+ if (old_streams) {
+ g_list_free_full (old_streams,
(GDestroyNotify) gst_adaptive_demux_stream_free);
- demux->priv->old_streams = NULL;
}
g_free (demux->manifest_uri);
demux->have_group_id = FALSE;
demux->group_id = G_MAXUINT;
- demux->priv->exposing = FALSE;
demux->priv->segment_seqnum = gst_util_seqnum_next ();
}
gchar *debug = NULL;
gchar *new_error = NULL;
+ GST_MANIFEST_LOCK (demux);
+
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
stream = iter->data;
if (GST_OBJECT_CAST (stream->src) == GST_MESSAGE_SRC (msg)) {
}
}
+ GST_MANIFEST_UNLOCK (demux);
+
gst_message_unref (msg);
msg = NULL;
}
gst_adaptive_demux_set_stream_struct_size (GstAdaptiveDemux * demux,
gsize struct_size)
{
+ GST_API_LOCK (demux);
+ GST_MANIFEST_LOCK (demux);
demux->stream_struct_size = struct_size;
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_expose_stream (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
stream->pending_caps = NULL;
}
- stream->download_finished = FALSE;
stream->discont = FALSE;
gst_object_ref (pad);
return gst_element_add_pad (GST_ELEMENT_CAST (demux), pad);
}
+/* must be called with manifest_lock taken */
static GstClockTime
gst_adaptive_demux_stream_get_presentation_offset (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
return klass->get_presentation_offset (demux, stream);
}
+/* must be called with manifest_lock taken */
static GstClockTime
gst_adaptive_demux_get_period_start_time (GstAdaptiveDemux * demux)
{
return klass->get_period_start_time (demux);
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_expose_streams (GstAdaptiveDemux * demux,
gboolean first_and_live)
g_return_val_if_fail (demux->next_streams != NULL, FALSE);
- demux->priv->exposing = TRUE;
old_streams = demux->streams;
demux->streams = demux->next_streams;
demux->next_streams = NULL;
}
gst_element_no_more_pads (GST_ELEMENT_CAST (demux));
- demux->priv->exposing = FALSE;
if (old_streams) {
GstEvent *eos = gst_event_new_eos ();
+ /* before we put streams in the demux->priv->old_streams list,
+ * we ask the download task to stop. In this way, it will no longer be
+ * allowed to change the demux object.
+ */
for (iter = old_streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
gst_pad_push_event (stream->pad, gst_event_ref (eos));
gst_pad_set_active (stream->pad, FALSE);
gst_element_remove_pad (GST_ELEMENT (demux), stream->pad);
+
+ /* ask the download task to stop.
+ * We will not join it now, because our thread can be one of these tasks.
+ * We will do the joining later, from another stream download task or
+ * from gst_adaptive_demux_stop_tasks.
+ * We also cannot change the state of the stream->src element, because
+ * that will wait on the streaming thread (which could be this thread)
+ * to stop first.
+ * Because we sent an EOS to the downstream element, the stream->src
+ * element should detect this in its streaming task and stop.
+ * Even if it doesn't do that, we will change its state later in
+ * gst_adaptive_demux_stop_tasks.
+ */
+ gst_task_stop (stream->download_task);
+ g_mutex_lock (&stream->fragment_download_lock);
+ stream->cancelled = TRUE;
+ g_cond_signal (&stream->fragment_download_cond);
+ g_mutex_unlock (&stream->fragment_download_lock);
}
gst_event_unref (eos);
/* The list should be freed from another thread as we can't properly
* cleanup a GstTask from itself */
- GST_OBJECT_LOCK (demux);
demux->priv->old_streams =
g_list_concat (demux->priv->old_streams, old_streams);
- GST_OBJECT_UNLOCK (demux);
}
return TRUE;
}
+/* must be called with manifest_lock taken */
GstAdaptiveDemuxStream *
gst_adaptive_demux_stream_new (GstAdaptiveDemux * demux, GstPad * pad)
{
return stream;
}
+/* must be called with manifest_lock taken.
+ * It will temporarily drop the manifest_lock in order to join the task.
+ * It will join only the old_streams (the demux->streams are joined by
+ * gst_adaptive_demux_stop_tasks before gst_adaptive_demux_stream_free is
+ * called)
+ */
static void
gst_adaptive_demux_stream_free (GstAdaptiveDemuxStream * stream)
{
GST_DEBUG_OBJECT (demux, "Leaving streaming task %s:%s",
GST_DEBUG_PAD_NAME (stream->pad));
- g_cond_signal (&stream->fragment_download_cond);
gst_task_stop (stream->download_task);
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ stream->cancelled = TRUE;
+ g_cond_signal (&stream->fragment_download_cond);
+ g_mutex_unlock (&stream->fragment_download_lock);
}
GST_LOG_OBJECT (demux, "Waiting for task to finish");
+
+ /* temporarily drop the manifest lock to join the task */
+ GST_MANIFEST_UNLOCK (demux);
+
gst_task_join (stream->download_task);
+
+ GST_MANIFEST_LOCK (demux);
+
GST_LOG_OBJECT (demux, "Finished");
gst_object_unref (stream->download_task);
g_rec_mutex_clear (&stream->download_lock);
g_free (stream);
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_get_live_seek_range (GstAdaptiveDemux * demux,
gint64 * range_start, gint64 * range_stop)
return klass->get_live_seek_range (demux, range_start, range_stop);
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_can_seek (GstAdaptiveDemux * demux)
{
GST_INFO_OBJECT (demux, "Received seek event");
+ GST_API_LOCK (demux);
GST_MANIFEST_LOCK (demux);
+
if (!gst_adaptive_demux_can_seek (demux)) {
GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
gst_event_unref (event);
return FALSE;
}
- GST_MANIFEST_UNLOCK (demux);
gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start,
&stop_type, &stop);
if (format != GST_FORMAT_TIME) {
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
gst_event_unref (event);
return FALSE;
}
gint64 range_start, range_stop;
if (!gst_adaptive_demux_get_live_seek_range (demux, &range_start,
&range_stop)) {
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
gst_event_unref (event);
return FALSE;
}
if (start < range_start || start >= range_stop) {
+ GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
GST_WARNING_OBJECT (demux, "Seek to invalid position");
gst_event_unref (event);
return FALSE;
GST_DEBUG_OBJECT (demux, "Seeking to segment %" GST_SEGMENT_FORMAT,
&demux->segment);
- GST_MANIFEST_LOCK (demux);
ret = demux_class->seek (demux, event);
if (!ret) {
/* Restart the demux */
gst_adaptive_demux_start_tasks (demux);
GST_MANIFEST_UNLOCK (demux);
+ GST_API_UNLOCK (demux);
gst_event_unref (event);
return ret;
case GST_EVENT_RECONFIGURE:{
GList *iter;
- /* When exposing pads reconfigure events are received as result
- * of the linking of the pads. The exposing and reconfigure happens
- * from the same thread. This prevents a deadlock and, although
- * not beautiful, makes this safe by avoiding that the demux->streams
- * list gets modified while this loop is running */
- if (!demux->priv->exposing)
- GST_MANIFEST_LOCK (demux);
+ GST_MANIFEST_LOCK (demux);
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
if (stream->pad == pad) {
- g_mutex_lock (&stream->fragment_download_lock);
if (stream->last_ret == GST_FLOW_NOT_LINKED) {
stream->last_ret = GST_FLOW_OK;
stream->restart_download = TRUE;
GST_DEBUG_OBJECT (stream->pad, "Restarting download loop");
gst_task_start (stream->download_task);
}
- g_mutex_unlock (&stream->fragment_download_lock);
gst_event_unref (event);
- if (!demux->priv->exposing)
- GST_MANIFEST_UNLOCK (demux);
+ GST_MANIFEST_UNLOCK (demux);
return TRUE;
}
}
- if (!demux->priv->exposing)
- GST_MANIFEST_UNLOCK (demux);
+ GST_MANIFEST_UNLOCK (demux);
}
break;
case GST_EVENT_LATENCY:{
GstClockTime duration = -1;
GstFormat fmt;
- GST_MANIFEST_LOCK (demux);
gst_query_parse_duration (query, &fmt, NULL);
+
+ GST_MANIFEST_LOCK (demux);
+
if (fmt == GST_FORMAT_TIME && demux->priv->have_manifest
&& !gst_adaptive_demux_is_live (demux)) {
duration = demux_class->get_duration (demux);
ret = TRUE;
}
}
+
GST_MANIFEST_UNLOCK (demux);
+
GST_LOG_OBJECT (demux, "GST_QUERY_DURATION returns %s with duration %"
GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration));
break;
gint64 start = 0;
GST_MANIFEST_LOCK (demux);
+
if (!demux->priv->have_manifest) {
GST_MANIFEST_UNLOCK (demux);
GST_INFO_OBJECT (demux,
break;
}
case GST_QUERY_URI:
- /* TODO HLS can answer this differently it seems */
+
GST_MANIFEST_LOCK (demux);
+
+ /* TODO HLS can answer this differently it seems */
if (demux->manifest_uri) {
/* FIXME: (hls) Do we answer with the variant playlist, with the current
* playlist or the the uri of the last downlowaded fragment? */
gst_query_set_uri (query, demux->manifest_uri);
ret = TRUE;
}
+
GST_MANIFEST_UNLOCK (demux);
break;
default:
return ret;
}
+/* must be called with manifest_lock taken */
static void
gst_adaptive_demux_start_tasks (GstAdaptiveDemux * demux)
{
GList *iter;
GST_INFO_OBJECT (demux, "Starting streams' tasks");
- demux->cancelled = FALSE;
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ stream->cancelled = FALSE;
+ stream->download_finished = FALSE;
+ g_mutex_unlock (&stream->fragment_download_lock);
+
stream->last_ret = GST_FLOW_OK;
gst_task_start (stream->download_task);
}
}
+/* must be called with manifest_lock taken
+ * This function will temporarily release manifest_lock in order to join the
+ * download threads.
+ * The api_lock will still protect it against other threads trying to modify
+ * the demux element.
+ */
static void
gst_adaptive_demux_stop_tasks (GstAdaptiveDemux * demux)
{
GList *iter;
- demux->cancelled = TRUE;
+ gst_task_stop (demux->priv->updates_task);
+ g_mutex_lock (&demux->priv->updates_timed_lock);
demux->priv->stop_updates_task = TRUE;
- gst_task_stop (demux->priv->updates_task);
g_cond_signal (&demux->priv->updates_timed_cond);
-
- GST_MANIFEST_LOCK (demux);
- g_cond_broadcast (&demux->manifest_cond);
- GST_MANIFEST_UNLOCK (demux);
+ g_mutex_unlock (&demux->priv->updates_timed_lock);
gst_uri_downloader_cancel (demux->downloader);
+
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
- gst_task_stop (stream->download_task);
- if (stream->src)
- gst_element_set_state (stream->src, GST_STATE_READY);
g_mutex_lock (&stream->fragment_download_lock);
- stream->download_finished = TRUE;
+ stream->cancelled = TRUE;
+ gst_task_stop (stream->download_task);
g_cond_signal (&stream->fragment_download_cond);
g_mutex_unlock (&stream->fragment_download_lock);
}
+ g_mutex_lock (&demux->priv->manifest_update_lock);
+ g_cond_broadcast (&demux->priv->manifest_cond);
+ g_mutex_unlock (&demux->priv->manifest_update_lock);
+
+ /* need to release manifest_lock before stopping the src element.
+ * The streams were asked to cancel, so they will not make any writes to demux
+ * object. Even if we temporarily release manifest_lock, the demux->streams
+ * cannot change and iter cannot be invalidated.
+ */
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
+ GstElement *src = stream->src;
+
+ GST_MANIFEST_UNLOCK (demux);
+ if (src) {
+ gst_element_set_state (src, GST_STATE_READY);
+ }
+
+ /* stream->download_task value never changes, so it is safe to read it
+ * outside critical section
+ */
gst_task_join (stream->download_task);
+
+ GST_MANIFEST_LOCK (demux);
+ }
+
+ GST_MANIFEST_UNLOCK (demux);
+
+ /* demux->priv->updates_task value never changes, so it is safe to read it
+ * outside critical section
+ */
+ gst_task_join (demux->priv->updates_task);
+
+ GST_MANIFEST_LOCK (demux);
+
+ for (iter = demux->streams; iter; iter = g_list_next (iter)) {
+ GstAdaptiveDemuxStream *stream = iter->data;
+
stream->download_error_count = 0;
stream->need_header = TRUE;
gst_adapter_clear (stream->adapter);
}
- gst_task_join (demux->priv->updates_task);
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_push_src_event (GstAdaptiveDemux * demux, GstEvent * event)
{
return ret;
}
+/* must be called with manifest_lock taken */
void
gst_adaptive_demux_stream_set_caps (GstAdaptiveDemuxStream * stream,
GstCaps * caps)
gst_caps_unref (caps);
}
+/* must be called with manifest_lock taken */
void
gst_adaptive_demux_stream_set_tags (GstAdaptiveDemuxStream * stream,
GstTagList * tags)
stream->pending_tags = tags;
}
+/* must be called with manifest_lock taken */
void
gst_adaptive_demux_stream_queue_event (GstAdaptiveDemuxStream * stream,
GstEvent * event)
stream->pending_events = g_list_append (stream->pending_events, event);
}
+/* must be called with manifest_lock taken */
static guint64
_update_average_bitrate (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream, guint64 new_bitrate)
return stream->moving_bitrate / stream->moving_index;
}
+/* must be called with manifest_lock taken */
static guint64
gst_adaptive_demux_stream_update_current_bitrate (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
return stream->current_download_rate;
}
+/* must be called with manifest_lock taken */
static GstFlowReturn
gst_adaptive_demux_combine_flows (GstAdaptiveDemux * demux)
{
for (iter = demux->streams; iter; iter = g_list_next (iter)) {
GstAdaptiveDemuxStream *stream = iter->data;
- g_mutex_lock (&stream->fragment_download_lock);
if (stream->last_ret != GST_FLOW_NOT_LINKED) {
all_notlinked = FALSE;
if (stream->last_ret != GST_FLOW_EOS)
if (stream->last_ret <= GST_FLOW_NOT_NEGOTIATED
|| stream->last_ret == GST_FLOW_FLUSHING) {
- g_mutex_unlock (&stream->fragment_download_lock);
return stream->last_ret;
}
- g_mutex_unlock (&stream->fragment_download_lock);
}
if (all_notlinked)
return GST_FLOW_NOT_LINKED;
return GST_FLOW_OK;
}
+/* must be called with manifest_lock taken.
+ * Temporarily releases manifest_lock
+ */
GstFlowReturn
gst_adaptive_demux_stream_push_buffer (GstAdaptiveDemuxStream * stream,
GstBuffer * buffer)
g_list_delete_link (stream->pending_events, stream->pending_events);
}
+ GST_MANIFEST_UNLOCK (demux);
+
ret = gst_pad_push (stream->pad, buffer);
+
+ GST_MANIFEST_LOCK (demux);
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ ret = stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ return ret;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+
GST_LOG_OBJECT (stream->pad, "Push result: %d %s", ret,
gst_flow_get_name (ret));
return ret;
}
+/* must be called with manifest_lock taken */
static GstFlowReturn
gst_adaptive_demux_stream_finish_fragment_default (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
stream->fragment.duration);
}
+/* must be called with manifest_lock taken.
+ * Can temporarily release manifest_lock
+ */
static GstFlowReturn
gst_adaptive_demux_stream_data_received_default (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
GstFlowReturn ret = GST_FLOW_OK;
+ GST_MANIFEST_LOCK (demux);
+
+ /* do not make any changes if the stream is cancelled */
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ g_mutex_unlock (&stream->fragment_download_lock);
+ gst_buffer_unref (buffer);
+ GST_MANIFEST_UNLOCK (demux);
+ ret = stream->last_ret = GST_FLOW_FLUSHING;
+ return ret;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+
if (stream->starting_fragment) {
GstClockTime offset =
gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
GST_DEBUG_OBJECT (stream->pad, "Received buffer of size %" G_GSIZE_FORMAT
". Now %" G_GSIZE_FORMAT " on adapter", gst_buffer_get_size (buffer),
gst_adapter_available (stream->adapter));
+
ret = klass->data_received (demux, stream);
+
+ if (ret == GST_FLOW_FLUSHING) {
+ /* do not make any changes if the stream is cancelled */
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ g_mutex_unlock (&stream->fragment_download_lock);
+ GST_MANIFEST_UNLOCK (demux);
+ return ret;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+ }
+
stream->download_chunk_start_time = g_get_monotonic_time ();
if (ret != GST_FLOW_OK) {
ret = GST_FLOW_EOS; /* return EOS to make the source stop */
}
+ GST_MANIFEST_UNLOCK (demux);
+
return ret;
}
+/* must be called with manifest_lock taken */
static void
gst_adaptive_demux_stream_fragment_download_finish (GstAdaptiveDemuxStream *
stream, GstFlowReturn ret, GError * err)
{
- g_mutex_lock (&stream->fragment_download_lock);
- stream->download_finished = TRUE;
-
GST_DEBUG_OBJECT (stream->pad, "Download finish: %d %s - err: %p", ret,
gst_flow_get_name (ret), err);
stream->last_error = g_error_copy (err);
}
}
+ g_mutex_lock (&stream->fragment_download_lock);
+ stream->download_finished = TRUE;
g_cond_signal (&stream->fragment_download_cond);
g_mutex_unlock (&stream->fragment_download_lock);
}
{
GstPad *srcpad = GST_PAD_CAST (parent);
GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (srcpad);
+ GstAdaptiveDemux *demux = stream->demux;
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_EOS:{
GstAdaptiveDemuxClass *klass;
GstFlowReturn ret;
- klass = GST_ADAPTIVE_DEMUX_GET_CLASS (stream->demux);
- ret = klass->finish_fragment (stream->demux, stream);
+ GST_MANIFEST_LOCK (demux);
+
+ klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
+ ret = klass->finish_fragment (demux, stream);
gst_adaptive_demux_stream_fragment_download_finish (stream, ret, NULL);
+
+ GST_MANIFEST_UNLOCK (demux);
break;
}
default:
return gst_pad_query_default (pad, parent, query);
}
+/* must be called with manifest_lock taken.
+ * Can temporarily release manifest_lock
+ */
static gboolean
gst_adaptive_demux_stream_wait_manifest_update (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
* us to download in the playlist or the playlist
* became non-live */
while (TRUE) {
- if (demux->cancelled) {
+ GST_DEBUG_OBJECT (demux, "No fragment left but live playlist, wait a bit");
+
+ /* get the manifest_update_lock while still holding the manifest_lock.
+ * This will prevent other threads to signal the condition (they will need
+ * both manifest_lock and manifest_update_lock in order to signal).
+ * It cannot deadlock because all threads always get the manifest_lock first
+ * and manifest_update_lock second.
+ */
+ g_mutex_lock (&demux->priv->manifest_update_lock);
+
+ GST_MANIFEST_UNLOCK (demux);
+
+ g_cond_wait (&demux->priv->manifest_cond,
+ &demux->priv->manifest_update_lock);
+ g_mutex_unlock (&demux->priv->manifest_update_lock);
+
+ GST_MANIFEST_LOCK (demux);
+
+ /* check for cancelled every time we get the manifest_lock */
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
ret = FALSE;
+ stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
break;
}
-
- GST_DEBUG_OBJECT (demux, "No fragment left but live playlist, wait a bit");
- g_cond_wait (&demux->manifest_cond, GST_MANIFEST_GET_LOCK (demux));
+ g_mutex_unlock (&stream->fragment_download_lock);
/* Got a new fragment or not live anymore? */
if (gst_adaptive_demux_stream_has_next_fragment (demux, stream)) {
return TRUE;
}
+/* must be called with manifest_lock taken */
static void
gst_adaptive_demux_stream_clear_eos_and_flush_state (GstAdaptiveDemuxStream *
stream)
gst_object_unref (internal_pad);
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_stream_update_source (GstAdaptiveDemuxStream * stream,
const gchar * uri, const gchar * referer, gboolean refresh,
return TRUE;
}
-/* must be called with the stream's fragment_download_lock */
+/* must be called with manifest_lock taken.
+ * Can temporarily release manifest_lock
+ */
static GstFlowReturn
gst_adaptive_demux_stream_download_uri (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream, const gchar * uri, gint64 start,
GST_DEBUG_OBJECT (stream->pad, "Downloading uri: %s, range:%" G_GINT64_FORMAT
" - %" G_GINT64_FORMAT, uri, start, end);
- stream->download_finished = FALSE;
-
if (!gst_adaptive_demux_stream_update_source (stream, uri, NULL, FALSE, TRUE)) {
- g_mutex_lock (&stream->fragment_download_lock);
ret = stream->last_ret = GST_FLOW_ERROR;
- g_mutex_unlock (&stream->fragment_download_lock);
return ret;
}
}
}
- g_mutex_lock (&stream->fragment_download_lock);
if (G_LIKELY (stream->last_ret == GST_FLOW_OK)) {
stream->download_start_time = g_get_monotonic_time ();
stream->download_chunk_start_time = g_get_monotonic_time ();
- g_mutex_unlock (&stream->fragment_download_lock);
+
+ GST_MANIFEST_UNLOCK (demux);
+
gst_element_sync_state_with_parent (stream->src);
- g_mutex_lock (&stream->fragment_download_lock);
/* wait for the fragment to be completely downloaded */
GST_DEBUG_OBJECT (stream->pad,
"Waiting for fragment download to finish: %s", uri);
- while (!stream->demux->cancelled && !stream->download_finished) {
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ g_mutex_unlock (&stream->fragment_download_lock);
+ GST_MANIFEST_LOCK (demux);
+ ret = stream->last_ret = GST_FLOW_FLUSHING;
+ return ret;
+ }
+ while (!stream->cancelled && !stream->download_finished) {
g_cond_wait (&stream->fragment_download_cond,
&stream->fragment_download_lock);
}
+ g_mutex_unlock (&stream->fragment_download_lock);
+
+ GST_MANIFEST_LOCK (demux);
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ ret = stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ return ret;
+ }
+
+ /* download_finished is set to FALSE when the streams are exposed.
+ * It is set to TRUE when the download is finished or when the task
+ * must stop.
+ * After the task has waited on download_finished, it needs to reset it to
+ * FALSE before it releases the fragment_download_lock. If it releases the
+ * lock and does the reset later, it might overwrite a stop request.
+ */
+ stream->download_finished = FALSE;
+ g_mutex_unlock (&stream->fragment_download_lock);
+
ret = stream->last_ret;
GST_DEBUG_OBJECT (stream->pad, "Fragment download finished: %s %d %s",
uri, stream->last_ret, gst_flow_get_name (stream->last_ret));
}
- g_mutex_unlock (&stream->fragment_download_lock);
} else {
- g_mutex_lock (&stream->fragment_download_lock);
if (stream->last_ret == GST_FLOW_OK)
stream->last_ret = GST_FLOW_CUSTOM_ERROR;
ret = GST_FLOW_CUSTOM_ERROR;
- g_mutex_unlock (&stream->fragment_download_lock);
}
/* flush the proxypads so that the EOS state is reset */
gst_pad_push_event (stream->src_srcpad, gst_event_new_flush_start ());
gst_pad_push_event (stream->src_srcpad, gst_event_new_flush_stop (TRUE));
+ /* changing source state will join the streaming thread. That thread can
+ * try to get the manifest_lock, so we must release it here.
+ */
+ GST_MANIFEST_UNLOCK (demux);
+
gst_element_set_state (stream->src, GST_STATE_READY);
+
+ GST_MANIFEST_LOCK (demux);
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ ret = stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ return ret;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+
gst_adaptive_demux_stream_clear_eos_and_flush_state (stream);
return ret;
}
+/* must be called with manifest_lock taken.
+ * Can temporarily release manifest_lock
+ */
static GstFlowReturn
gst_adaptive_demux_stream_download_header_fragment (GstAdaptiveDemuxStream *
stream)
}
/* check if we have an index */
- if (!demux->cancelled && ret == GST_FLOW_OK) { /* TODO check for other valid types */
+ if (ret == GST_FLOW_OK) { /* TODO check for other valid types */
if (stream->fragment.index_uri != NULL) {
GST_DEBUG_OBJECT (demux,
return ret;
}
+/* must be called with manifest_lock taken.
+ * Can temporarily release manifest_lock
+ */
static GstFlowReturn
gst_adaptive_demux_stream_download_fragment (GstAdaptiveDemuxStream * stream)
{
gchar *url = NULL;
GstFlowReturn ret = GST_FLOW_OK;
- g_mutex_lock (&stream->fragment_download_lock);
stream->starting_fragment = TRUE;
stream->last_ret = GST_FLOW_OK;
stream->first_fragment_buffer = TRUE;
- g_mutex_unlock (&stream->fragment_download_lock);
if (stream->fragment.uri == NULL && stream->fragment.header_uri == NULL &&
stream->fragment.index_uri == NULL)
GST_DEBUG_OBJECT (stream->pad, "Fragment download result: %d %s",
stream->last_ret, gst_flow_get_name (stream->last_ret));
if (ret != GST_FLOW_OK) {
- /* TODO check if we are truly stoping */
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ g_mutex_unlock (&stream->fragment_download_lock);
+ return ret;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+
+ /* TODO check if we are truly stopping */
if (ret == GST_FLOW_CUSTOM_ERROR && gst_adaptive_demux_is_live (demux)) {
if (++stream->download_error_count <= MAX_DOWNLOAD_ERROR_COUNT) {
/* looks like there is no way of knowing when a live stream has ended
}
}
+/* this function will take the manifest_lock and will keep it until the end.
+ * It will release it temporarily only when going to sleep.
+ * Every time it takes the manifest_lock, it will check for cancelled condition
+ */
static void
gst_adaptive_demux_stream_download_loop (GstAdaptiveDemuxStream * stream)
{
GST_LOG_OBJECT (stream->pad, "download loop start");
+ GST_MANIFEST_LOCK (demux);
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ goto cancelled;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+
/* Check if we're done with our segment */
if (demux->segment.rate > 0) {
if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)
}
}
- GST_OBJECT_LOCK (demux);
- if (demux->cancelled) {
- stream->last_ret = GST_FLOW_FLUSHING;
- goto cancelled;
- }
-
/* Cleanup old streams if any */
if (G_UNLIKELY (demux->priv->old_streams != NULL)) {
GList *old_streams = demux->priv->old_streams;
demux->priv->old_streams = NULL;
- GST_OBJECT_UNLOCK (demux);
- /* Need to unlock as it might post messages to the bus */
GST_DEBUG_OBJECT (stream->pad, "Cleaning up old streams");
g_list_free_full (old_streams,
(GDestroyNotify) gst_adaptive_demux_stream_free);
GST_DEBUG_OBJECT (stream->pad, "Cleaning up old streams (done)");
- } else {
- GST_OBJECT_UNLOCK (demux);
+
+ /* gst_adaptive_demux_stream_free had temporarily released the manifest_lock.
+ * Recheck the cancelled flag.
+ */
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ goto cancelled;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
}
- GST_MANIFEST_LOCK (demux);
if (G_UNLIKELY (stream->restart_download)) {
GstSegment segment;
GstEvent *seg_event;
GST_DEBUG_OBJECT (stream->pad, "Restarting stream at "
"position %" GST_TIME_FORMAT, GST_TIME_ARGS (ts));
+
gst_segment_copy_into (&demux->segment, &segment);
if (GST_CLOCK_TIME_IS_VALID (ts)) {
stream->restart_download = FALSE;
}
- GST_OBJECT_LOCK (demux);
- if (demux->cancelled) {
- stream->last_ret = GST_FLOW_FLUSHING;
- GST_MANIFEST_UNLOCK (demux);
- goto cancelled;
- }
- GST_OBJECT_UNLOCK (demux);
live = gst_adaptive_demux_is_live (demux);
if (live) {
gint64 wait_time =
gst_adaptive_demux_stream_get_fragment_waiting_time (demux, stream);
- GST_MANIFEST_UNLOCK (demux);
- if (wait_time > 0)
- gst_adaptive_demux_stream_download_wait (stream, wait_time);
- } else {
- GST_MANIFEST_UNLOCK (demux);
- }
+ if (wait_time > 0) {
+ gint64 end_time = g_get_monotonic_time () + wait_time / GST_USECOND;
- GST_OBJECT_LOCK (demux);
- if (demux->cancelled) {
- stream->last_ret = GST_FLOW_FLUSHING;
- goto cancelled;
+ GST_DEBUG_OBJECT (stream->pad, "Download waiting for %" GST_TIME_FORMAT,
+ GST_TIME_ARGS (wait_time));
+
+ GST_MANIFEST_UNLOCK (demux);
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ g_mutex_unlock (&stream->fragment_download_lock);
+ GST_MANIFEST_LOCK (demux);
+ stream->last_ret = GST_FLOW_FLUSHING;
+ goto cancelled;
+ }
+ g_cond_wait_until (&stream->fragment_download_cond,
+ &stream->fragment_download_lock, end_time);
+ g_mutex_unlock (&stream->fragment_download_lock);
+
+ GST_DEBUG_OBJECT (stream->pad, "Download finished waiting");
+
+ GST_MANIFEST_LOCK (demux);
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ goto cancelled;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+ }
}
- GST_OBJECT_UNLOCK (demux);
stream->last_ret = GST_FLOW_OK;
+
next_download = g_get_monotonic_time ();
ret = gst_adaptive_demux_stream_download_fragment (stream);
- GST_OBJECT_LOCK (demux);
- if (demux->cancelled) {
- stream->last_ret = GST_FLOW_FLUSHING;
- goto cancelled;
+ if (ret == GST_FLOW_FLUSHING) {
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ goto cancelled;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
}
- GST_OBJECT_UNLOCK (demux);
- GST_MANIFEST_LOCK (demux);
} else {
stream->last_ret = ret;
}
/* we push the EOS after releasing the object lock */
if (gst_adaptive_demux_is_live (demux)) {
if (gst_adaptive_demux_stream_wait_manifest_update (demux, stream)) {
- GST_MANIFEST_UNLOCK (demux);
- return;
+ goto end;
}
gst_task_stop (stream->download_task);
} else {
gboolean is_live = gst_adaptive_demux_is_live (demux);
GST_WARNING_OBJECT (demux, "Error while downloading fragment");
if (++stream->download_error_count > MAX_DOWNLOAD_ERROR_COUNT) {
- GST_MANIFEST_UNLOCK (demux);
goto download_error;
}
* a bit to not flood the server */
if (stream->download_error_count == 1 && !is_live) {
/* TODO hlsdemux had more options to this function (boolean and err) */
- GST_MANIFEST_UNLOCK (demux);
+
if (gst_adaptive_demux_update_manifest (demux) == GST_FLOW_OK) {
/* Retry immediately, the playlist actually has changed */
GST_DEBUG_OBJECT (demux, "Updated the playlist");
- return;
+ goto end;
}
- GST_MANIFEST_LOCK (demux);
}
/* Wait half the fragment duration before retrying */
(stream->fragment.duration, G_USEC_PER_SEC, 2 * GST_SECOND);
GST_MANIFEST_UNLOCK (demux);
+
g_mutex_lock (&stream->fragment_download_lock);
- if (demux->cancelled) {
+ if (G_UNLIKELY (stream->cancelled)) {
g_mutex_unlock (&stream->fragment_download_lock);
- return;
+ GST_MANIFEST_LOCK (demux);
+ stream->last_ret = GST_FLOW_FLUSHING;
+ goto cancelled;
}
g_cond_wait_until (&stream->fragment_download_cond,
&stream->fragment_download_lock, next_download);
g_mutex_unlock (&stream->fragment_download_lock);
+
GST_DEBUG_OBJECT (demux, "Retrying now");
+ GST_MANIFEST_LOCK (demux);
+
+ g_mutex_lock (&stream->fragment_download_lock);
+ if (G_UNLIKELY (stream->cancelled)) {
+ stream->last_ret = GST_FLOW_FLUSHING;
+ g_mutex_unlock (&stream->fragment_download_lock);
+ goto cancelled;
+ }
+ g_mutex_unlock (&stream->fragment_download_lock);
+
/* Refetch the playlist now after we waited */
if (!is_live
&& gst_adaptive_demux_update_manifest (demux) == GST_FLOW_OK) {
GST_DEBUG_OBJECT (demux, "Updated the playlist");
}
- return;
+ goto end;
}
break;
}
- GST_MANIFEST_UNLOCK (demux);
end_of_manifest:
if (G_UNLIKELY (ret == GST_FLOW_EOS)) {
gst_adaptive_demux_stream_push_event (stream, gst_event_new_eos ());
}
end:
+ GST_MANIFEST_UNLOCK (demux);
GST_LOG_OBJECT (stream->pad, "download loop end");
return;
cancelled:
{
GST_DEBUG_OBJECT (stream->pad, "Stream has been cancelled");
- GST_OBJECT_UNLOCK (demux);
goto end;
}
download_error:
}
gst_element_post_message (GST_ELEMENT_CAST (demux), msg);
+
goto end;
}
}
static void
gst_adaptive_demux_updates_loop (GstAdaptiveDemux * demux)
{
+ gint64 next_update;
GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
/* Loop for updating of the playlist. This periodically checks if
/* block until the next scheduled update or the signal to quit this thread */
GST_DEBUG_OBJECT (demux, "Started updates task");
- demux->priv->next_update =
- g_get_monotonic_time () + klass->get_manifest_update_interval (demux);
+ GST_MANIFEST_LOCK (demux);
+
+ next_update = g_get_monotonic_time () +
+ klass->get_manifest_update_interval (demux);
/* Updating playlist only needed for live playlists */
while (gst_adaptive_demux_is_live (demux)) {
/* Wait here until we should do the next update or we're cancelled */
GST_DEBUG_OBJECT (demux, "Wait for next playlist update");
+ GST_MANIFEST_UNLOCK (demux);
+
g_mutex_lock (&demux->priv->updates_timed_lock);
if (demux->priv->stop_updates_task) {
g_mutex_unlock (&demux->priv->updates_timed_lock);
goto quit;
}
g_cond_wait_until (&demux->priv->updates_timed_cond,
- &demux->priv->updates_timed_lock, demux->priv->next_update);
+ &demux->priv->updates_timed_lock, next_update);
+ g_mutex_unlock (&demux->priv->updates_timed_lock);
+
+ GST_MANIFEST_LOCK (demux);
+
+ g_mutex_lock (&demux->priv->updates_timed_lock);
if (demux->priv->stop_updates_task) {
g_mutex_unlock (&demux->priv->updates_timed_lock);
+ GST_MANIFEST_UNLOCK (demux);
goto quit;
}
g_mutex_unlock (&demux->priv->updates_timed_lock);
GST_DEBUG_OBJECT (demux, "Updating playlist");
+
ret = gst_adaptive_demux_update_manifest (demux);
+
if (ret == GST_FLOW_EOS) {
} else if (ret != GST_FLOW_OK) {
- if (demux->priv->stop_updates_task)
- goto quit;
+ /* update_failed_count is used only here, no need to protect it */
demux->priv->update_failed_count++;
if (demux->priv->update_failed_count <= DEFAULT_FAILED_COUNT) {
GST_WARNING_OBJECT (demux, "Could not update the playlist");
- demux->priv->next_update =
- g_get_monotonic_time () +
+ next_update = g_get_monotonic_time () +
klass->get_manifest_update_interval (demux);
} else {
GST_ELEMENT_ERROR (demux, STREAM, FAILED,
(_("Internal data stream error.")), ("Could not update playlist"));
- goto error;
+ GST_DEBUG_OBJECT (demux, "Stopped updates task because of error");
+ gst_task_stop (demux->priv->updates_task);
+ GST_MANIFEST_UNLOCK (demux);
+ goto end;
}
} else {
GST_DEBUG_OBJECT (demux, "Updated playlist successfully");
- GST_MANIFEST_LOCK (demux);
- demux->priv->next_update =
- g_get_monotonic_time () + klass->get_manifest_update_interval (demux);
- /* Wake up download task */
- g_cond_broadcast (&demux->manifest_cond);
- GST_MANIFEST_UNLOCK (demux);
+ next_update = g_get_monotonic_time () +
+ klass->get_manifest_update_interval (demux);
+
+ /* Wake up download tasks */
+ g_mutex_lock (&demux->priv->manifest_update_lock);
+ g_cond_broadcast (&demux->priv->manifest_cond);
+ g_mutex_unlock (&demux->priv->manifest_update_lock);
}
}
+ GST_MANIFEST_UNLOCK (demux);
+
quit:
{
- GST_DEBUG_OBJECT (demux, "Stopped updates task");
- gst_task_stop (demux->priv->updates_task);
- return;
+ GST_DEBUG_OBJECT (demux, "Stop updates task request detected.");
}
-error:
+end:
{
- GST_DEBUG_OBJECT (demux, "Stopped updates task because of error");
- gst_task_stop (demux->priv->updates_task);
+ return;
}
}
-
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_stream_push_event (GstAdaptiveDemuxStream * stream,
GstEvent * event)
return ret;
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_is_live (GstAdaptiveDemux * demux)
{
return FALSE;
}
+/* must be called with manifest_lock taken */
static GstFlowReturn
gst_adaptive_demux_stream_seek (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream, GstClockTime ts)
return GST_FLOW_ERROR;
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_stream_has_next_fragment (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
return ret;
}
+/* must be called with manifest_lock taken */
GstFlowReturn
gst_adaptive_demux_stream_advance_fragment (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream, GstClockTime duration)
{
GstFlowReturn ret;
- GST_MANIFEST_LOCK (demux);
if (stream->last_ret == GST_FLOW_OK) {
stream->last_ret =
gst_adaptive_demux_stream_advance_fragment_unlocked (demux, stream,
duration);
}
ret = stream->last_ret;
- GST_MANIFEST_UNLOCK (demux);
return ret;
}
+/* must be called with manifest_lock taken */
GstFlowReturn
gst_adaptive_demux_stream_advance_fragment_unlocked (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream, GstClockTime duration)
return ret;
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_stream_select_bitrate (GstAdaptiveDemux *
demux, GstAdaptiveDemuxStream * stream, guint64 bitrate)
return FALSE;
}
+/* must be called with manifest_lock taken */
static GstFlowReturn
gst_adaptive_demux_stream_update_fragment_info (GstAdaptiveDemux * demux,
GstAdaptiveDemuxStream * stream)
return klass->stream_update_fragment_info (stream);
}
+/* must be called with manifest_lock taken */
static gint64
gst_adaptive_demux_stream_get_fragment_waiting_time (GstAdaptiveDemux *
demux, GstAdaptiveDemuxStream * stream)
return 0;
}
+/* must be called with manifest_lock taken */
static GstFlowReturn
gst_adaptive_demux_update_manifest_default (GstAdaptiveDemux * demux)
{
download = gst_uri_downloader_fetch_uri (demux->downloader,
demux->manifest_uri, NULL, TRUE, TRUE, TRUE, NULL);
if (download) {
- GST_MANIFEST_LOCK (demux);
g_free (demux->manifest_uri);
g_free (demux->manifest_base_uri);
if (download->redirect_permanent && download->redirect_uri) {
g_object_unref (download);
ret = klass->update_manifest_data (demux, buffer);
gst_buffer_unref (buffer);
- GST_MANIFEST_UNLOCK (demux);
/* FIXME: Should the manifest uri vars be reverted to original
* values if updating fails? */
} else {
return ret;
}
+/* must be called with manifest_lock taken */
static GstFlowReturn
gst_adaptive_demux_update_manifest (GstAdaptiveDemux * demux)
{
if (ret == GST_FLOW_OK) {
GstClockTime duration;
- GST_MANIFEST_LOCK (demux);
/* Send an updated duration message */
duration = klass->get_duration (demux);
- GST_MANIFEST_UNLOCK (demux);
if (duration != GST_CLOCK_TIME_NONE) {
GST_DEBUG_OBJECT (demux,
"Sending duration message : %" GST_TIME_FORMAT,
f->index_range_end = -1;
}
+/* must be called with manifest_lock taken */
static gboolean
gst_adaptive_demux_has_next_period (GstAdaptiveDemux * demux)
{
return ret;
}
+/* must be called with manifest_lock taken */
static void
gst_adaptive_demux_advance_period (GstAdaptiveDemux * demux)
{
gst_adaptive_demux_expose_streams (demux, FALSE);
gst_adaptive_demux_start_tasks (demux);
}
-
-static void
-gst_adaptive_demux_stream_download_wait (GstAdaptiveDemuxStream * stream,
- GstClockTime time_diff)
-{
- gint64 end_time = g_get_monotonic_time () + time_diff / GST_USECOND;
-
- GST_DEBUG_OBJECT (stream->pad, "Download waiting for %" GST_TIME_FORMAT,
- GST_TIME_ARGS (time_diff));
- g_mutex_lock (&stream->fragment_download_lock);
- g_cond_wait_until (&stream->fragment_download_cond,
- &stream->fragment_download_lock, end_time);
- g_mutex_unlock (&stream->fragment_download_lock);
- GST_DEBUG_OBJECT (stream->pad, "Download finished waiting");
-}
projects / platform / upstream / gstreamer.git / commitdiff