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_TRACE (data_queue_dataflow, \
81 "locking qlock from thread %p", \
83 g_mutex_lock (&q->priv->qlock); \
84 GST_CAT_TRACE (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_TRACE (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_wait_non_empty (GstDataQueue * queue)
490 GstDataQueuePrivate *priv = queue->priv;
492 while (gst_data_queue_locked_is_empty (queue)) {
493 priv->waiting_add = TRUE;
494 g_cond_wait (&priv->item_add, &priv->qlock);
495 priv->waiting_add = FALSE;
503 * gst_data_queue_pop:
504 * @queue: a #GstDataQueue.
505 * @item: pointer to store the returned #GstDataQueueItem.
507 * Retrieves the first @item available on the @queue. If the queue is currently
508 * empty, the call will block until at least one item is available, OR the
509 * @queue is set to the flushing state.
512 * Returns: #TRUE if an @item was successfully retrieved from the @queue.
517 gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
519 GstDataQueuePrivate *priv = queue->priv;
521 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
522 g_return_val_if_fail (item != NULL, FALSE);
524 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
526 STATUS (queue, "before popping");
528 if (gst_data_queue_locked_is_empty (queue)) {
529 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
530 if (G_LIKELY (priv->emptycallback))
531 priv->emptycallback (queue, priv->checkdata);
533 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
534 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
536 if (!_gst_data_queue_wait_non_empty (queue))
540 /* Get the item from the GQueue */
541 *item = gst_queue_array_pop_head (priv->queue);
543 /* update current level counter */
544 if ((*item)->visible)
545 priv->cur_level.visible--;
546 priv->cur_level.bytes -= (*item)->size;
547 priv->cur_level.time -= (*item)->duration;
549 STATUS (queue, "after popping");
550 if (priv->waiting_del)
551 g_cond_signal (&priv->item_del);
553 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
560 GST_DEBUG ("queue:%p, we are flushing", queue);
561 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
567 is_of_type (gconstpointer a, gconstpointer b)
569 return !G_TYPE_CHECK_INSTANCE_TYPE (a, GPOINTER_TO_SIZE (b));
573 * gst_data_queue_peek:
574 * @queue: a #GstDataQueue.
575 * @item: pointer to store the returned #GstDataQueueItem.
577 * Retrieves the first @item available on the @queue without removing it.
578 * If the queue is currently empty, the call will block until at least
579 * one item is available, OR the @queue is set to the flushing state.
582 * Returns: #TRUE if an @item was successfully retrieved from the @queue.
587 gst_data_queue_peek (GstDataQueue * queue, GstDataQueueItem ** item)
589 GstDataQueuePrivate *priv = queue->priv;
591 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
592 g_return_val_if_fail (item != NULL, FALSE);
594 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
596 STATUS (queue, "before peeking");
598 if (gst_data_queue_locked_is_empty (queue)) {
599 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
600 if (G_LIKELY (priv->emptycallback))
601 priv->emptycallback (queue, priv->checkdata);
603 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
604 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
606 if (!_gst_data_queue_wait_non_empty (queue))
610 /* Get the item from the GQueue */
611 *item = gst_queue_array_peek_head (priv->queue);
613 STATUS (queue, "after peeking");
614 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
621 GST_DEBUG ("queue:%p, we are flushing", queue);
622 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
628 * gst_data_queue_drop_head:
629 * @queue: The #GstDataQueue to drop an item from.
630 * @type: The #GType of the item to drop.
632 * Pop and unref the head-most #GstMiniObject with the given #GType.
634 * Returns: TRUE if an element was removed.
639 gst_data_queue_drop_head (GstDataQueue * queue, GType type)
641 gboolean res = FALSE;
642 GstDataQueueItem *leak = NULL;
644 GstDataQueuePrivate *priv = queue->priv;
646 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
648 GST_DEBUG ("queue:%p", queue);
650 GST_DATA_QUEUE_MUTEX_LOCK (queue);
651 idx = gst_queue_array_find (priv->queue, is_of_type, GSIZE_TO_POINTER (type));
656 leak = gst_queue_array_drop_element (priv->queue, idx);
659 priv->cur_level.visible--;
660 priv->cur_level.bytes -= leak->size;
661 priv->cur_level.time -= leak->duration;
663 leak->destroy (leak);
668 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
670 GST_DEBUG ("queue:%p , res:%d", queue, res);
676 * gst_data_queue_limits_changed:
677 * @queue: The #GstDataQueue
679 * Inform the queue that the limits for the fullness check have changed and that
680 * any blocking gst_data_queue_push() should be unblocked to recheck the limts.
685 gst_data_queue_limits_changed (GstDataQueue * queue)
687 GstDataQueuePrivate *priv = queue->priv;
689 g_return_if_fail (GST_IS_DATA_QUEUE (queue));
691 GST_DATA_QUEUE_MUTEX_LOCK (queue);
692 if (priv->waiting_del) {
693 GST_DEBUG ("signal del");
694 g_cond_signal (&priv->item_del);
696 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
700 * gst_data_queue_get_level:
701 * @queue: The #GstDataQueue
702 * @level: the location to store the result
704 * Get the current level of the queue.
709 gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize * level)
711 GstDataQueuePrivate *priv = queue->priv;
713 memcpy (level, (&priv->cur_level), sizeof (GstDataQueueSize));
717 gst_data_queue_set_property (GObject * object,
718 guint prop_id, const GValue * value, GParamSpec * pspec)
722 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
728 gst_data_queue_get_property (GObject * object,
729 guint prop_id, GValue * value, GParamSpec * pspec)
731 GstDataQueue *queue = GST_DATA_QUEUE (object);
732 GstDataQueuePrivate *priv = queue->priv;
734 GST_DATA_QUEUE_MUTEX_LOCK (queue);
737 case PROP_CUR_LEVEL_BYTES:
738 g_value_set_uint (value, priv->cur_level.bytes);
740 case PROP_CUR_LEVEL_VISIBLE:
741 g_value_set_uint (value, priv->cur_level.visible);
743 case PROP_CUR_LEVEL_TIME:
744 g_value_set_uint64 (value, priv->cur_level.time);
747 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
751 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);