From 581c4297d0953fe492b34889e95d0353f4b8a509 Mon Sep 17 00:00:00 2001 From: Thiago Santos Date: Wed, 7 Aug 2013 21:26:01 -0300 Subject: [PATCH] dataqueue: add gst_data_queue_push_force Adds a variant of the _push function that doesn't check the queue limits before adding the new item. It is useful when pushing an element to the queue shouldn't lock the thread. One particular scenario is when the queue is used to serialize buffers and events that are going to be pushed from another thread. The dataqueue should have a limit on the amount of buffers to be stored to avoid large memory consumption, but events can be considered to have negligible impact on memory compared to buffers. So it is useful to be used to push items into the queue that contain events, even though the queue is already full, it shouldn't matter inserting an item that has no significative size. This scenario happens on adaptive elements (dashdemux / mssdemux) as there is a single download thread fetching buffers and putting into the dataqueues for the streams. This same download thread can als generate events in some situations as caps changes, eos or a internal control events. There can be a deadlock at preroll if the first buffer fetched is large enough to fill the dataqueue and the download thread and the next iteration of the download thread decides to push an event to this same dataqueue before fetching buffers to other streams, if this push locks, the pipeline will be stuck in preroll as no more buffers will be downloaded. There is a somewhat common practice in dash streams to have a single very large buffer for audio and one for video, so this will always happen as the download thread will have to push an EOS right after fetching the first buffer for any stream. API: gst_data_queue_push_force https://bugzilla.gnome.org/show_bug.cgi?id=705694 --- docs/libs/gstreamer-libs-sections.txt | 1 + libs/gst/base/gstdataqueue.c | 69 ++++++++++++++++++++++++--- libs/gst/base/gstdataqueue.h | 1 + win32/common/libgstbase.def | 1 + 4 files changed, 66 insertions(+), 6 deletions(-) diff --git a/docs/libs/gstreamer-libs-sections.txt b/docs/libs/gstreamer-libs-sections.txt index cdcd21c1fb..bbf7332f7b 100644 --- a/docs/libs/gstreamer-libs-sections.txt +++ b/docs/libs/gstreamer-libs-sections.txt @@ -761,6 +761,7 @@ GstDataQueueEmptyCallback GstDataQueueFullCallback gst_data_queue_new gst_data_queue_push +gst_data_queue_push_force gst_data_queue_pop gst_data_queue_flush gst_data_queue_set_flushing diff --git a/libs/gst/base/gstdataqueue.c b/libs/gst/base/gstdataqueue.c index 9a3ca9ff1c..0c46f0c2ba 100644 --- a/libs/gst/base/gstdataqueue.c +++ b/libs/gst/base/gstdataqueue.c @@ -410,6 +410,68 @@ gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing) GST_DATA_QUEUE_MUTEX_UNLOCK (queue); } +static void +gst_data_queue_push_force_unlocked (GstDataQueue * queue, + GstDataQueueItem * item) +{ + GstDataQueuePrivate *priv = queue->priv; + + gst_queue_array_push_tail (priv->queue, item); + + if (item->visible) + priv->cur_level.visible++; + priv->cur_level.bytes += item->size; + priv->cur_level.time += item->duration; +} + +/** + * gst_data_queue_push_force: + * @queue: a #GstDataQueue. + * @item: a #GstDataQueueItem. + * + * Pushes a #GstDataQueueItem (or a structure that begins with the same fields) + * on the @queue. It ignores if the @queue is full or not and forces the @item + * to be pushed anyway. + * MT safe. + * + * Note that this function has slightly different semantics than gst_pad_push() + * and gst_pad_push_event(): this function only takes ownership of @item and + * the #GstMiniObject contained in @item if the push was successful. If FALSE + * is returned, the caller is responsible for freeing @item and its contents. + * + * Returns: #TRUE if the @item was successfully pushed on the @queue. + * + * Since: 1.2.0 + */ +gboolean +gst_data_queue_push_force (GstDataQueue * queue, GstDataQueueItem * item) +{ + GstDataQueuePrivate *priv = queue->priv; + + g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE); + g_return_val_if_fail (item != NULL, FALSE); + + GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing); + + STATUS (queue, "before pushing"); + gst_data_queue_push_force_unlocked (queue, item); + STATUS (queue, "after pushing"); + if (priv->waiting_add) + g_cond_signal (&priv->item_add); + + GST_DATA_QUEUE_MUTEX_UNLOCK (queue); + + return TRUE; + + /* ERRORS */ +flushing: + { + GST_DEBUG ("queue:%p, we are flushing", queue); + GST_DATA_QUEUE_MUTEX_UNLOCK (queue); + return FALSE; + } +} + /** * gst_data_queue_push: * @queue: a #GstDataQueue. @@ -460,12 +522,7 @@ gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item) } } - gst_queue_array_push_tail (priv->queue, item); - - if (item->visible) - priv->cur_level.visible++; - priv->cur_level.bytes += item->size; - priv->cur_level.time += item->duration; + gst_data_queue_push_force_unlocked (queue, item); STATUS (queue, "after pushing"); if (priv->waiting_add) diff --git a/libs/gst/base/gstdataqueue.h b/libs/gst/base/gstdataqueue.h index 3bd76f5a93..c707c8b57d 100644 --- a/libs/gst/base/gstdataqueue.h +++ b/libs/gst/base/gstdataqueue.h @@ -139,6 +139,7 @@ GstDataQueue * gst_data_queue_new (GstDataQueueCheckFullFunction chec gpointer checkdata) G_GNUC_MALLOC; gboolean gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item); +gboolean gst_data_queue_push_force (GstDataQueue * queue, GstDataQueueItem * item); gboolean gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item); gboolean gst_data_queue_peek (GstDataQueue * queue, GstDataQueueItem ** item); diff --git a/win32/common/libgstbase.def b/win32/common/libgstbase.def index 09a13ab7c9..1b1db5f6d0 100644 --- a/win32/common/libgstbase.def +++ b/win32/common/libgstbase.def @@ -256,6 +256,7 @@ EXPORTS gst_data_queue_peek gst_data_queue_pop gst_data_queue_push + gst_data_queue_push_force gst_data_queue_set_flushing gst_push_src_get_type gst_queue_array_drop_element