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., 59 Temple Place - Suite 330,
19 * Boston, MA 02111-1307, 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.
32 #include "gstdataqueue.h"
34 GST_DEBUG_CATEGORY_STATIC (data_queue_debug);
35 #define GST_CAT_DEFAULT (data_queue_debug)
36 GST_DEBUG_CATEGORY_STATIC (data_queue_dataflow);
39 /* Queue signals and args */
50 ARG_CUR_LEVEL_VISIBLE,
56 #define GST_DATA_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
57 GST_CAT_LOG (data_queue_dataflow, \
58 "locking qlock from thread %p", \
60 g_mutex_lock (q->qlock); \
61 GST_CAT_LOG (data_queue_dataflow, \
62 "locked qlock from thread %p", \
66 #define GST_DATA_QUEUE_MUTEX_LOCK_CHECK(q, label) G_STMT_START { \
67 GST_DATA_QUEUE_MUTEX_LOCK (q); \
72 #define GST_DATA_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
73 GST_CAT_LOG (data_queue_dataflow, \
74 "unlocking qlock from thread %p", \
76 g_mutex_unlock (q->qlock); \
79 #define STATUS(q, msg) \
80 GST_CAT_LOG (data_queue_dataflow, \
81 "queue:%p " msg ": %u visible items, %u " \
82 "bytes, %"G_GUINT64_FORMAT \
85 q->cur_level.visible, \
90 static void gst_data_queue_finalize (GObject * object);
92 static void gst_data_queue_set_property (GObject * object,
93 guint prop_id, const GValue * value, GParamSpec * pspec);
94 static void gst_data_queue_get_property (GObject * object,
95 guint prop_id, GValue * value, GParamSpec * pspec);
97 static GObjectClass *parent_class = NULL;
98 static guint gst_data_queue_signals[LAST_SIGNAL] = { 0 };
102 GST_DEBUG_CATEGORY_INIT (data_queue_debug, "dataqueue", 0, \
103 "data queue object"); \
104 GST_DEBUG_CATEGORY_INIT (data_queue_dataflow, "data_queue_dataflow", 0, \
105 "dataflow inside the data queue object"); \
109 G_DEFINE_TYPE_WITH_CODE (GstDataQueue, gst_data_queue, G_TYPE_OBJECT, _do_init);
112 gst_data_queue_class_init (GstDataQueueClass * klass)
114 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
116 parent_class = g_type_class_peek_parent (klass);
118 gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_data_queue_set_property);
119 gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_data_queue_get_property);
123 * GstDataQueue::empty:
124 * @queue: the queue instance
126 * Reports that the queue became empty (empty).
127 * A queue is empty if the total amount of visible items inside it (num-visible, time,
128 * size) is lower than the boundary values which can be set through the GObject
131 gst_data_queue_signals[SIGNAL_EMPTY] =
132 g_signal_new ("empty", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
133 G_STRUCT_OFFSET (GstDataQueueClass, empty), NULL, NULL,
134 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
137 * GstDataQueue::full:
138 * @queue: the queue instance
140 * Reports that the queue became full (full).
141 * A queue is full if the total amount of data inside it (num-visible, time,
142 * size) is higher than the boundary values which can be set through the GObject
145 gst_data_queue_signals[SIGNAL_FULL] =
146 g_signal_new ("full", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
147 G_STRUCT_OFFSET (GstDataQueueClass, full), NULL, NULL,
148 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
151 g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BYTES,
152 g_param_spec_uint ("current-level-bytes", "Current level (kB)",
153 "Current amount of data in the queue (bytes)",
154 0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
155 g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_VISIBLE,
156 g_param_spec_uint ("current-level-visible",
157 "Current level (visible items)",
158 "Current number of visible items in the queue", 0, G_MAXUINT, 0,
159 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
160 g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_TIME,
161 g_param_spec_uint64 ("current-level-time", "Current level (ns)",
162 "Current amount of data in the queue (in ns)", 0, G_MAXUINT64, 0,
163 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
165 /* set several parent class virtual functions */
166 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_data_queue_finalize);
171 gst_data_queue_init (GstDataQueue * queue)
173 queue->cur_level.visible = 0; /* no content */
174 queue->cur_level.bytes = 0; /* no content */
175 queue->cur_level.time = 0; /* no content */
177 queue->checkfull = NULL;
179 queue->qlock = g_mutex_new ();
180 queue->item_add = g_cond_new ();
181 queue->item_del = g_cond_new ();
182 queue->queue = g_queue_new ();
184 GST_DEBUG ("initialized queue's not_empty & not_full conditions");
188 * gst_data_queue_new_full:
189 * @checkfull: the callback used to tell if the element considers the queue full
191 * @fullcallback: the callback which will be called when the queue is considered full.
192 * @emptycallback: the callback which will be called when the queue is considered empty.
193 * @checkdata: a #gpointer that will be given in the @checkfull callback.
195 * Creates a new #GstDataQueue. The difference with @gst_data_queue_new is that it will
196 * not emit the 'full' and 'empty' signals, but instead calling directly @fullcallback
199 * Returns: a new #GstDataQueue.
203 gst_data_queue_new_full (GstDataQueueCheckFullFunction checkfull,
204 GstDataQueueFullCallback fullcallback,
205 GstDataQueueEmptyCallback emptycallback, gpointer checkdata)
209 g_return_val_if_fail (checkfull != NULL, NULL);
211 ret = g_object_new (GST_TYPE_DATA_QUEUE, NULL);
212 ret->checkfull = checkfull;
213 ret->checkdata = checkdata;
214 ret->fullcallback = fullcallback;
215 ret->emptycallback = emptycallback;
221 * gst_data_queue_new:
222 * @checkfull: the callback used to tell if the element considers the queue full
224 * @checkdata: a #gpointer that will be given in the @checkfull callback.
226 * Returns: a new #GstDataQueue.
230 gst_data_queue_new (GstDataQueueCheckFullFunction checkfull, gpointer checkdata)
232 return gst_data_queue_new_full (checkfull, NULL, NULL, checkdata);
236 gst_data_queue_cleanup (GstDataQueue * queue)
238 while (!g_queue_is_empty (queue->queue)) {
239 GstDataQueueItem *item = g_queue_pop_head (queue->queue);
241 /* Just call the destroy notify on the item */
242 item->destroy (item);
244 queue->cur_level.visible = 0;
245 queue->cur_level.bytes = 0;
246 queue->cur_level.time = 0;
249 /* called only once, as opposed to dispose */
251 gst_data_queue_finalize (GObject * object)
253 GstDataQueue *queue = GST_DATA_QUEUE (object);
255 GST_DEBUG ("finalizing queue");
257 gst_data_queue_cleanup (queue);
258 g_queue_free (queue->queue);
260 GST_DEBUG ("free mutex");
261 g_mutex_free (queue->qlock);
262 GST_DEBUG ("done free mutex");
264 g_cond_free (queue->item_add);
265 g_cond_free (queue->item_del);
267 G_OBJECT_CLASS (parent_class)->finalize (object);
271 gst_data_queue_locked_flush (GstDataQueue * queue)
273 STATUS (queue, "before flushing");
274 gst_data_queue_cleanup (queue);
275 STATUS (queue, "after flushing");
276 /* we deleted something... */
277 g_cond_signal (queue->item_del);
281 gst_data_queue_locked_is_empty (GstDataQueue * queue)
283 return (queue->queue->length == 0);
287 gst_data_queue_locked_is_full (GstDataQueue * queue)
289 return queue->checkfull (queue, queue->cur_level.visible,
290 queue->cur_level.bytes, queue->cur_level.time, queue->checkdata);
294 * gst_data_queue_flush:
295 * @queue: a #GstDataQueue.
297 * Flushes all the contents of the @queue. Any call to #gst_data_queue_push and
298 * #gst_data_queue_pop will be released.
302 gst_data_queue_flush (GstDataQueue * queue)
304 GST_DEBUG ("queue:%p", queue);
305 GST_DATA_QUEUE_MUTEX_LOCK (queue);
306 gst_data_queue_locked_flush (queue);
307 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
311 * gst_data_queue_is_empty:
312 * @queue: a #GstDataQueue.
314 * Queries if there are any items in the @queue.
317 * Returns: #TRUE if @queue is empty.
320 gst_data_queue_is_empty (GstDataQueue * queue)
324 GST_DATA_QUEUE_MUTEX_LOCK (queue);
325 res = gst_data_queue_locked_is_empty (queue);
326 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
332 * gst_data_queue_is_full:
333 * @queue: a #GstDataQueue.
335 * Queries if @queue is full. This check will be done using the
336 * #GstDataQueueCheckFullCallback registered with @queue.
339 * Returns: #TRUE if @queue is full.
342 gst_data_queue_is_full (GstDataQueue * queue)
346 GST_DATA_QUEUE_MUTEX_LOCK (queue);
347 res = gst_data_queue_locked_is_full (queue);
348 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
354 * gst_data_queue_set_flushing:
355 * @queue: a #GstDataQueue.
356 * @flushing: a #gboolean stating if the queue will be flushing or not.
358 * Sets the queue to flushing state if @flushing is #TRUE. If set to flushing
359 * state, any incoming data on the @queue will be discarded. Any call currently
360 * blocking on #gst_data_queue_push or #gst_data_queue_pop will return straight
361 * away with a return value of #FALSE. While the @queue is in flushing state,
362 * all calls to those two functions will return #FALSE.
367 gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
369 GST_DEBUG ("queue:%p , flushing:%d", queue, flushing);
371 GST_DATA_QUEUE_MUTEX_LOCK (queue);
372 queue->flushing = flushing;
374 /* release push/pop functions */
375 g_cond_signal (queue->item_add);
376 g_cond_signal (queue->item_del);
378 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
382 * gst_data_queue_push:
383 * @queue: a #GstDataQueue.
384 * @item: a #GstDataQueueItem.
386 * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
387 * on the @queue. If the @queue is full, the call will block until space is
388 * available, OR the @queue is set to flushing state.
391 * Note that this function has slightly different semantics than gst_pad_push()
392 * and gst_pad_push_event(): this function only takes ownership of @item and
393 * the #GstMiniObject contained in @item if the push was successful. If FALSE
394 * is returned, the caller is responsible for freeing @item and its contents.
396 * Returns: #TRUE if the @item was successfully pushed on the @queue.
399 gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item)
401 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
402 g_return_val_if_fail (item != NULL, FALSE);
404 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
406 STATUS (queue, "before pushing");
408 /* We ALWAYS need to check for queue fillness */
409 if (gst_data_queue_locked_is_full (queue)) {
410 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
411 if (G_LIKELY (queue->fullcallback))
412 queue->fullcallback (queue, queue->checkdata);
414 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_FULL], 0);
415 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
417 /* signal might have removed some items */
418 while (gst_data_queue_locked_is_full (queue)) {
419 g_cond_wait (queue->item_del, queue->qlock);
425 g_queue_push_tail (queue->queue, item);
428 queue->cur_level.visible++;
429 queue->cur_level.bytes += item->size;
430 queue->cur_level.time += item->duration;
432 STATUS (queue, "after pushing");
433 g_cond_signal (queue->item_add);
435 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
442 GST_DEBUG ("queue:%p, we are flushing", queue);
443 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
449 * gst_data_queue_pop:
450 * @queue: a #GstDataQueue.
451 * @item: pointer to store the returned #GstDataQueueItem.
453 * Retrieves the first @item available on the @queue. If the queue is currently
454 * empty, the call will block until at least one item is available, OR the
455 * @queue is set to the flushing state.
458 * Returns: #TRUE if an @item was successfully retrieved from the @queue.
461 gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
463 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
464 g_return_val_if_fail (item != NULL, FALSE);
466 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
468 STATUS (queue, "before popping");
470 if (gst_data_queue_locked_is_empty (queue)) {
471 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
472 if (G_LIKELY (queue->emptycallback))
473 queue->emptycallback (queue, queue->checkdata);
475 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
476 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
478 while (gst_data_queue_locked_is_empty (queue)) {
479 g_cond_wait (queue->item_add, queue->qlock);
485 /* Get the item from the GQueue */
486 *item = g_queue_pop_head (queue->queue);
488 /* update current level counter */
489 if ((*item)->visible)
490 queue->cur_level.visible--;
491 queue->cur_level.bytes -= (*item)->size;
492 queue->cur_level.time -= (*item)->duration;
494 STATUS (queue, "after popping");
495 g_cond_signal (queue->item_del);
497 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
504 GST_DEBUG ("queue:%p, we are flushing", queue);
505 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
511 * gst_data_queue_drop_head:
512 * @queue: The #GstDataQueue to drop an item from.
513 * @type: The #GType of the item to drop.
515 * Pop and unref the head-most #GstMiniObject with the given #GType.
517 * Returns: TRUE if an element was removed.
520 gst_data_queue_drop_head (GstDataQueue * queue, GType type)
522 gboolean res = FALSE;
524 GstDataQueueItem *leak = NULL;
526 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
528 GST_DEBUG ("queue:%p", queue);
530 GST_DATA_QUEUE_MUTEX_LOCK (queue);
531 for (item = g_queue_peek_head_link (queue->queue); item; item = item->next) {
532 GstDataQueueItem *tmp = (GstDataQueueItem *) item->data;
534 if (G_TYPE_CHECK_INSTANCE_TYPE (tmp->object, type)) {
543 g_queue_delete_link (queue->queue, item);
546 queue->cur_level.visible--;
547 queue->cur_level.bytes -= leak->size;
548 queue->cur_level.time -= leak->duration;
550 leak->destroy (leak);
555 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
557 GST_DEBUG ("queue:%p , res:%d", queue, res);
563 * gst_data_queue_limits_changed:
564 * @queue: The #GstDataQueue
566 * Inform the queue that the limits for the fullness check have changed and that
567 * any blocking gst_data_queue_push() should be unblocked to recheck the limts.
570 gst_data_queue_limits_changed (GstDataQueue * queue)
572 g_return_if_fail (GST_IS_DATA_QUEUE (queue));
574 GST_DATA_QUEUE_MUTEX_LOCK (queue);
575 GST_DEBUG ("signal del");
576 g_cond_signal (queue->item_del);
577 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
581 * gst_data_queue_get_level:
582 * @queue: The #GstDataQueue
583 * @level: the location to store the result
585 * Get the current level of the queue.
588 gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize * level)
590 level->visible = queue->cur_level.visible;
591 level->bytes = queue->cur_level.bytes;
592 level->time = queue->cur_level.time;
596 gst_data_queue_set_property (GObject * object,
597 guint prop_id, const GValue * value, GParamSpec * pspec)
601 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
607 gst_data_queue_get_property (GObject * object,
608 guint prop_id, GValue * value, GParamSpec * pspec)
610 GstDataQueue *queue = GST_DATA_QUEUE (object);
612 GST_DATA_QUEUE_MUTEX_LOCK (queue);
615 case ARG_CUR_LEVEL_BYTES:
616 g_value_set_uint (value, queue->cur_level.bytes);
618 case ARG_CUR_LEVEL_VISIBLE:
619 g_value_set_uint (value, queue->cur_level.visible);
621 case ARG_CUR_LEVEL_TIME:
622 g_value_set_uint64 (value, queue->cur_level.time);
625 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
629 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);