2 * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Library General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Library General Public License for more details.
16 * You should have received a copy of the GNU Library General Public
17 * License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
23 * SECTION:gstdataqueue
24 * @short_description: Threadsafe queueing object
26 * #GstDataQueue is an object that handles threadsafe queueing of objects. It
27 * also provides size-related functionality. This object should be used for
28 * any #GstElement that wishes to provide some sort of queueing functionality.
33 #include "gstdataqueue.h"
34 #include "gstqueuearray.h"
35 #include "gst/glib-compat-private.h"
37 GST_DEBUG_CATEGORY_STATIC (data_queue_debug);
38 #define GST_CAT_DEFAULT (data_queue_debug)
39 GST_DEBUG_CATEGORY_STATIC (data_queue_dataflow);
42 /* Queue signals and args */
53 PROP_CUR_LEVEL_VISIBLE,
59 struct _GstDataQueuePrivate
61 /* the array of data we're keeping our grubby hands on */
64 GstDataQueueSize cur_level; /* size of the queue */
65 GstDataQueueCheckFullFunction checkfull; /* Callback to check if the queue is full */
68 GMutex qlock; /* lock for queue (vs object lock) */
70 GCond item_add; /* signals buffers now available for reading */
72 GCond item_del; /* signals space now available for writing */
73 gboolean flushing; /* indicates whether conditions where signalled because
74 * of external flushing */
75 GstDataQueueFullCallback fullcallback;
76 GstDataQueueEmptyCallback emptycallback;
79 #define GST_DATA_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
80 GST_CAT_LOG (data_queue_dataflow, \
81 "locking qlock from thread %p", \
83 g_mutex_lock (&q->priv->qlock); \
84 GST_CAT_LOG (data_queue_dataflow, \
85 "locked qlock from thread %p", \
89 #define GST_DATA_QUEUE_MUTEX_LOCK_CHECK(q, label) G_STMT_START { \
90 GST_DATA_QUEUE_MUTEX_LOCK (q); \
91 if (q->priv->flushing) \
95 #define GST_DATA_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
96 GST_CAT_LOG (data_queue_dataflow, \
97 "unlocking qlock from thread %p", \
99 g_mutex_unlock (&q->priv->qlock); \
102 #define STATUS(q, msg) \
103 GST_CAT_LOG (data_queue_dataflow, \
104 "queue:%p " msg ": %u visible items, %u " \
105 "bytes, %"G_GUINT64_FORMAT \
106 " ns, %u elements", \
108 q->priv->cur_level.visible, \
109 q->priv->cur_level.bytes, \
110 q->priv->cur_level.time, \
111 gst_queue_array_get_length (q->priv->queue))
113 static void gst_data_queue_finalize (GObject * object);
115 static void gst_data_queue_set_property (GObject * object,
116 guint prop_id, const GValue * value, GParamSpec * pspec);
117 static void gst_data_queue_get_property (GObject * object,
118 guint prop_id, GValue * value, GParamSpec * pspec);
120 static guint gst_data_queue_signals[LAST_SIGNAL] = { 0 };
124 GST_DEBUG_CATEGORY_INIT (data_queue_debug, "dataqueue", 0, \
125 "data queue object"); \
126 GST_DEBUG_CATEGORY_INIT (data_queue_dataflow, "data_queue_dataflow", 0, \
127 "dataflow inside the data queue object"); \
130 #define parent_class gst_data_queue_parent_class
131 G_DEFINE_TYPE_WITH_CODE (GstDataQueue, gst_data_queue, G_TYPE_OBJECT, _do_init);
134 gst_data_queue_class_init (GstDataQueueClass * klass)
136 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
138 g_type_class_add_private (klass, sizeof (GstDataQueuePrivate));
140 gobject_class->set_property = gst_data_queue_set_property;
141 gobject_class->get_property = gst_data_queue_get_property;
145 * GstDataQueue::empty:
146 * @queue: the queue instance
148 * Reports that the queue became empty (empty).
149 * A queue is empty if the total amount of visible items inside it (num-visible, time,
150 * size) is lower than the boundary values which can be set through the GObject
153 gst_data_queue_signals[SIGNAL_EMPTY] =
154 g_signal_new ("empty", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
155 G_STRUCT_OFFSET (GstDataQueueClass, empty), NULL, NULL,
156 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
159 * GstDataQueue::full:
160 * @queue: the queue instance
162 * Reports that the queue became full (full).
163 * A queue is full if the total amount of data inside it (num-visible, time,
164 * size) is higher than the boundary values which can be set through the GObject
167 gst_data_queue_signals[SIGNAL_FULL] =
168 g_signal_new ("full", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
169 G_STRUCT_OFFSET (GstDataQueueClass, full), NULL, NULL,
170 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
173 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BYTES,
174 g_param_spec_uint ("current-level-bytes", "Current level (kB)",
175 "Current amount of data in the queue (bytes)",
176 0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
177 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_VISIBLE,
178 g_param_spec_uint ("current-level-visible",
179 "Current level (visible items)",
180 "Current number of visible items in the queue", 0, G_MAXUINT, 0,
181 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
182 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_TIME,
183 g_param_spec_uint64 ("current-level-time", "Current level (ns)",
184 "Current amount of data in the queue (in ns)", 0, G_MAXUINT64, 0,
185 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
187 gobject_class->finalize = gst_data_queue_finalize;
191 gst_data_queue_init (GstDataQueue * queue)
194 G_TYPE_INSTANCE_GET_PRIVATE (queue, GST_TYPE_DATA_QUEUE,
195 GstDataQueuePrivate);
197 queue->priv->cur_level.visible = 0; /* no content */
198 queue->priv->cur_level.bytes = 0; /* no content */
199 queue->priv->cur_level.time = 0; /* no content */
201 queue->priv->checkfull = NULL;
203 g_mutex_init (&queue->priv->qlock);
204 g_cond_init (&queue->priv->item_add);
205 g_cond_init (&queue->priv->item_del);
206 queue->priv->queue = gst_queue_array_new (50);
208 GST_DEBUG ("initialized queue's not_empty & not_full conditions");
212 * gst_data_queue_new:
213 * @checkfull: the callback used to tell if the element considers the queue full
215 * @fullcallback: the callback which will be called when the queue is considered full.
216 * @emptycallback: the callback which will be called when the queue is considered empty.
217 * @checkdata: a #gpointer that will be given in the @checkfull callback.
219 * Creates a new #GstDataQueue. The difference with @gst_data_queue_new is that it will
220 * not emit the 'full' and 'empty' signals, but instead calling directly @fullcallback
223 * Returns: a new #GstDataQueue.
228 gst_data_queue_new (GstDataQueueCheckFullFunction checkfull,
229 GstDataQueueFullCallback fullcallback,
230 GstDataQueueEmptyCallback emptycallback, gpointer checkdata)
234 g_return_val_if_fail (checkfull != NULL, NULL);
236 ret = g_object_newv (GST_TYPE_DATA_QUEUE, 0, NULL);
237 ret->priv->checkfull = checkfull;
238 ret->priv->checkdata = checkdata;
239 ret->priv->fullcallback = fullcallback;
240 ret->priv->emptycallback = emptycallback;
246 gst_data_queue_cleanup (GstDataQueue * queue)
248 GstDataQueuePrivate *priv = queue->priv;
250 while (!gst_queue_array_is_empty (priv->queue)) {
251 GstDataQueueItem *item = gst_queue_array_pop_head (priv->queue);
253 /* Just call the destroy notify on the item */
254 item->destroy (item);
256 priv->cur_level.visible = 0;
257 priv->cur_level.bytes = 0;
258 priv->cur_level.time = 0;
261 /* called only once, as opposed to dispose */
263 gst_data_queue_finalize (GObject * object)
265 GstDataQueue *queue = GST_DATA_QUEUE (object);
266 GstDataQueuePrivate *priv = queue->priv;
268 GST_DEBUG ("finalizing queue");
270 gst_data_queue_cleanup (queue);
271 gst_queue_array_free (priv->queue);
273 GST_DEBUG ("free mutex");
274 g_mutex_clear (&priv->qlock);
275 GST_DEBUG ("done free mutex");
277 g_cond_clear (&priv->item_add);
278 g_cond_clear (&priv->item_del);
280 G_OBJECT_CLASS (parent_class)->finalize (object);
284 gst_data_queue_locked_flush (GstDataQueue * queue)
286 GstDataQueuePrivate *priv = queue->priv;
288 STATUS (queue, "before flushing");
289 gst_data_queue_cleanup (queue);
290 STATUS (queue, "after flushing");
291 /* we deleted something... */
292 if (priv->waiting_del)
293 g_cond_signal (&priv->item_del);
296 static inline gboolean
297 gst_data_queue_locked_is_empty (GstDataQueue * queue)
299 GstDataQueuePrivate *priv = queue->priv;
301 return (gst_queue_array_get_length (priv->queue) == 0);
304 static inline gboolean
305 gst_data_queue_locked_is_full (GstDataQueue * queue)
307 GstDataQueuePrivate *priv = queue->priv;
309 return priv->checkfull (queue, priv->cur_level.visible,
310 priv->cur_level.bytes, priv->cur_level.time, priv->checkdata);
314 * gst_data_queue_flush:
315 * @queue: a #GstDataQueue.
317 * Flushes all the contents of the @queue. Any call to #gst_data_queue_push and
318 * #gst_data_queue_pop will be released.
324 gst_data_queue_flush (GstDataQueue * queue)
326 GST_DEBUG ("queue:%p", queue);
327 GST_DATA_QUEUE_MUTEX_LOCK (queue);
328 gst_data_queue_locked_flush (queue);
329 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
333 * gst_data_queue_is_empty:
334 * @queue: a #GstDataQueue.
336 * Queries if there are any items in the @queue.
339 * Returns: #TRUE if @queue is empty.
344 gst_data_queue_is_empty (GstDataQueue * queue)
348 GST_DATA_QUEUE_MUTEX_LOCK (queue);
349 res = gst_data_queue_locked_is_empty (queue);
350 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
356 * gst_data_queue_is_full:
357 * @queue: a #GstDataQueue.
359 * Queries if @queue is full. This check will be done using the
360 * #GstDataQueueCheckFullFunction registered with @queue.
363 * Returns: #TRUE if @queue is full.
368 gst_data_queue_is_full (GstDataQueue * queue)
372 GST_DATA_QUEUE_MUTEX_LOCK (queue);
373 res = gst_data_queue_locked_is_full (queue);
374 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
380 * gst_data_queue_set_flushing:
381 * @queue: a #GstDataQueue.
382 * @flushing: a #gboolean stating if the queue will be flushing or not.
384 * Sets the queue to flushing state if @flushing is #TRUE. If set to flushing
385 * state, any incoming data on the @queue will be discarded. Any call currently
386 * blocking on #gst_data_queue_push or #gst_data_queue_pop will return straight
387 * away with a return value of #FALSE. While the @queue is in flushing state,
388 * all calls to those two functions will return #FALSE.
395 gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
397 GstDataQueuePrivate *priv = queue->priv;
399 GST_DEBUG ("queue:%p , flushing:%d", queue, flushing);
401 GST_DATA_QUEUE_MUTEX_LOCK (queue);
402 priv->flushing = flushing;
404 /* release push/pop functions */
405 if (priv->waiting_add)
406 g_cond_signal (&priv->item_add);
407 if (priv->waiting_del)
408 g_cond_signal (&priv->item_del);
410 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
414 * gst_data_queue_push:
415 * @queue: a #GstDataQueue.
416 * @item: a #GstDataQueueItem.
418 * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
419 * on the @queue. If the @queue is full, the call will block until space is
420 * available, OR the @queue is set to flushing state.
423 * Note that this function has slightly different semantics than gst_pad_push()
424 * and gst_pad_push_event(): this function only takes ownership of @item and
425 * the #GstMiniObject contained in @item if the push was successful. If FALSE
426 * is returned, the caller is responsible for freeing @item and its contents.
428 * Returns: #TRUE if the @item was successfully pushed on the @queue.
433 gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item)
435 GstDataQueuePrivate *priv = queue->priv;
437 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
438 g_return_val_if_fail (item != NULL, FALSE);
440 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
442 STATUS (queue, "before pushing");
444 /* We ALWAYS need to check for queue fillness */
445 if (gst_data_queue_locked_is_full (queue)) {
446 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
447 if (G_LIKELY (priv->fullcallback))
448 priv->fullcallback (queue, priv->checkdata);
450 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_FULL], 0);
451 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
453 /* signal might have removed some items */
454 while (gst_data_queue_locked_is_full (queue)) {
455 priv->waiting_del = TRUE;
456 g_cond_wait (&priv->item_del, &priv->qlock);
457 priv->waiting_del = FALSE;
463 gst_queue_array_push_tail (priv->queue, item);
466 priv->cur_level.visible++;
467 priv->cur_level.bytes += item->size;
468 priv->cur_level.time += item->duration;
470 STATUS (queue, "after pushing");
471 if (priv->waiting_add)
472 g_cond_signal (&priv->item_add);
474 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
481 GST_DEBUG ("queue:%p, we are flushing", queue);
482 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
488 * gst_data_queue_pop:
489 * @queue: a #GstDataQueue.
490 * @item: pointer to store the returned #GstDataQueueItem.
492 * Retrieves the first @item available on the @queue. If the queue is currently
493 * empty, the call will block until at least one item is available, OR the
494 * @queue is set to the flushing state.
497 * Returns: #TRUE if an @item was successfully retrieved from the @queue.
502 gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
504 GstDataQueuePrivate *priv = queue->priv;
506 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
507 g_return_val_if_fail (item != NULL, FALSE);
509 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
511 STATUS (queue, "before popping");
513 if (gst_data_queue_locked_is_empty (queue)) {
514 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
515 if (G_LIKELY (priv->emptycallback))
516 priv->emptycallback (queue, priv->checkdata);
518 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
519 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
521 while (gst_data_queue_locked_is_empty (queue)) {
522 priv->waiting_add = TRUE;
523 g_cond_wait (&priv->item_add, &priv->qlock);
524 priv->waiting_add = FALSE;
530 /* Get the item from the GQueue */
531 *item = gst_queue_array_pop_head (priv->queue);
533 /* update current level counter */
534 if ((*item)->visible)
535 priv->cur_level.visible--;
536 priv->cur_level.bytes -= (*item)->size;
537 priv->cur_level.time -= (*item)->duration;
539 STATUS (queue, "after popping");
540 if (priv->waiting_del)
541 g_cond_signal (&priv->item_del);
543 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
550 GST_DEBUG ("queue:%p, we are flushing", queue);
551 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
557 is_of_type (gconstpointer a, gconstpointer b)
559 return !G_TYPE_CHECK_INSTANCE_TYPE (a, GPOINTER_TO_INT (b));
563 * gst_data_queue_drop_head:
564 * @queue: The #GstDataQueue to drop an item from.
565 * @type: The #GType of the item to drop.
567 * Pop and unref the head-most #GstMiniObject with the given #GType.
569 * Returns: TRUE if an element was removed.
574 gst_data_queue_drop_head (GstDataQueue * queue, GType type)
576 gboolean res = FALSE;
577 GstDataQueueItem *leak = NULL;
579 GstDataQueuePrivate *priv = queue->priv;
581 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
583 GST_DEBUG ("queue:%p", queue);
585 GST_DATA_QUEUE_MUTEX_LOCK (queue);
586 idx = gst_queue_array_find (priv->queue, is_of_type, GINT_TO_POINTER (type));
591 leak = gst_queue_array_drop_element (priv->queue, idx);
594 priv->cur_level.visible--;
595 priv->cur_level.bytes -= leak->size;
596 priv->cur_level.time -= leak->duration;
598 leak->destroy (leak);
603 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
605 GST_DEBUG ("queue:%p , res:%d", queue, res);
611 * gst_data_queue_limits_changed:
612 * @queue: The #GstDataQueue
614 * Inform the queue that the limits for the fullness check have changed and that
615 * any blocking gst_data_queue_push() should be unblocked to recheck the limts.
620 gst_data_queue_limits_changed (GstDataQueue * queue)
622 GstDataQueuePrivate *priv = queue->priv;
624 g_return_if_fail (GST_IS_DATA_QUEUE (queue));
626 GST_DATA_QUEUE_MUTEX_LOCK (queue);
627 if (priv->waiting_del) {
628 GST_DEBUG ("signal del");
629 g_cond_signal (&priv->item_del);
631 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
635 * gst_data_queue_get_level:
636 * @queue: The #GstDataQueue
637 * @level: the location to store the result
639 * Get the current level of the queue.
644 gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize * level)
646 GstDataQueuePrivate *priv = queue->priv;
648 memcpy (level, (&priv->cur_level), sizeof (GstDataQueueSize));
652 gst_data_queue_set_property (GObject * object,
653 guint prop_id, const GValue * value, GParamSpec * pspec)
657 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
663 gst_data_queue_get_property (GObject * object,
664 guint prop_id, GValue * value, GParamSpec * pspec)
666 GstDataQueue *queue = GST_DATA_QUEUE (object);
667 GstDataQueuePrivate *priv = queue->priv;
669 GST_DATA_QUEUE_MUTEX_LOCK (queue);
672 case PROP_CUR_LEVEL_BYTES:
673 g_value_set_uint (value, priv->cur_level.bytes);
675 case PROP_CUR_LEVEL_VISIBLE:
676 g_value_set_uint (value, priv->cur_level.visible);
678 case PROP_CUR_LEVEL_TIME:
679 g_value_set_uint64 (value, priv->cur_level.time);
682 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
686 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);