1 /* GStreamer
2 * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>
3 * Copyright (C) 2007 Jan Schmidt <jan@fluendo.com>
4 * Copyright (C) 2007 Wim Taymans <wim@fluendo.com>
5 *
6 * gstmultiqueue.c:
7 *
8 * This library is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Library General Public
10 * License as published by the Free Software Foundation; either
11 * version 2 of the License, or (at your option) any later version.
12 *
13 * This library is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * Library General Public License for more details.
17 *
18 * You should have received a copy of the GNU Library General Public
19 * License along with this library; if not, write to the
20 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
21 * Boston, MA 02111-1307, USA.
22 */
24 /**
25 * SECTION:element-multiqueue
26 * @short_description: Asynchronous data queues
27 * @see_also: #GstQueue
28 *
29 * <refsect2>
30 * <para>
31 * Multiqueue is similar to a normal #GstQueue with the following additional
32 * features:
33 * <orderedlist>
34 * <listitem>
35 * <itemizedlist><title>Multiple streamhandling</title>
36 * <listitem><para>
37 * The element handles queueing data on more than one stream at once. To
38 * achieve such a feature it has request sink pads (sink_%d) and
39 * 'sometimes' src pads (src_%d).
40 * </para><para>
41 * When requesting a given sinkpad with gst_element_get_request_pad(),
42 * the associated srcpad for that stream will be created.
43 * Ex: requesting sink_1 will generate src_1.
44 * </para></listitem>
45 * </itemizedlist>
46 * </listitem>
47 * <listitem>
48 * <itemizedlist><title>Non-starvation on multiple streams</title>
49 * <listitem><para>
50 * If more than one stream is used with the element, the streams' queues
51 * will be dynamically grown (up to a limit), in order to ensure that no
52 * stream is risking data starvation. This guarantees that at any given
53 * time there are at least N bytes queued and available for each individual
54 * stream.
55 * </para><para>
56 * If an EOS event comes through a srcpad, the associated queue will be
57 * considered as 'not-empty' in the queue-size-growing algorithm.
58 * </para></listitem>
59 * </itemizedlist>
60 * </listitem>
61 * <listitem>
62 * <itemizedlist><title>Non-linked srcpads graceful handling</title>
63 * <listitem><para>
64 * In order to better support dynamic switching between streams, the multiqueue
65 * (unlike the current GStreamer queue) continues to push buffers on non-linked
66 * pads rather than shutting down.
67 * </para><para>
68 * In addition, to prevent a non-linked stream from very quickly consuming all
69 * available buffers and thus 'racing ahead' of the other streams, the element
70 * must ensure that buffers and inlined events for a non-linked stream are pushed
71 * in the same order as they were received, relative to the other streams
72 * controlled by the element. This means that a buffer cannot be pushed to a
73 * non-linked pad any sooner than buffers in any other stream which were received
74 * before it.
75 * </para></listitem>
76 * </itemizedlist>
77 * </listitem>
78 * </orderedlist>
79 * </para>
80 * <para>
81 * Data is queued until one of the limits specified by the
82 * #GstMultiQueue:max-size-buffers, #GstMultiQueue:max-size-bytes and/or
83 * #GstMultiQueue:max-size-time properties has been reached. Any attempt to push
84 * more buffers into the queue will block the pushing thread until more space
85 * becomes available. #GstMultiQueue:extra-size-buffers,
86 * </para>
87 * <para>
88 * #GstMultiQueue:extra-size-bytes and #GstMultiQueue:extra-size-time are
89 * currently unused.
90 * </para>
91 * <para>
92 * The default queue size limits are 5 buffers, 10MB of data, or
93 * two second worth of data, whichever is reached first. Note that the number
94 * of buffers will dynamically grow depending on the fill level of
95 * other queues.
96 * </para>
97 * <para>
98 * The #GstMultiQueue::underrun signal is emitted when all of the queues
99 * are empty. The #GstMultiQueue::overrun signal is emitted when one of the
100 * queues is filled.
101 * Both signals are emitted from the context of the streaming thread.
102 * </para>
103 * </refsect2>
104 *
105 * Last reviewed on 2008-01-25 (0.10.17)
106 */
108 #ifdef HAVE_CONFIG_H
109 # include "config.h"
110 #endif
112 #include <gst/gst.h>
113 #include "gstmultiqueue.h"
115 /**
116 * GstSingleQueue:
117 * @sinkpad: associated sink #GstPad
118 * @srcpad: associated source #GstPad
119 *
120 * Structure containing all information and properties about
121 * a single queue.
122 */
123 typedef struct _GstSingleQueue GstSingleQueue;
125 struct _GstSingleQueue
126 {
127 /* unique identifier of the queue */
128 guint id;
130 GstMultiQueue *mqueue;
132 GstPad *sinkpad;
133 GstPad *srcpad;
135 /* flowreturn of previous srcpad push */
136 GstFlowReturn srcresult;
137 GstSegment sink_segment;
138 GstSegment src_segment;
140 /* queue of data */
141 GstDataQueue *queue;
142 GstDataQueueSize max_size, extra_size;
143 GstClockTime cur_time;
144 gboolean is_eos;
145 gboolean inextra; /* TRUE if the queue is currently in extradata mode */
147 /* Protected by global lock */
148 guint32 nextid; /* ID of the next object waiting to be pushed */
149 guint32 oldid; /* ID of the last object pushed (last in a series) */
150 GCond *turn; /* SingleQueue turn waiting conditional */
151 };
154 /* Extension of GstDataQueueItem structure for our usage */
155 typedef struct _GstMultiQueueItem GstMultiQueueItem;
157 struct _GstMultiQueueItem
158 {
159 GstMiniObject *object;
160 guint size;
161 guint64 duration;
162 gboolean visible;
164 GDestroyNotify destroy;
165 guint32 posid;
166 };
168 static GstSingleQueue *gst_single_queue_new (GstMultiQueue * mqueue);
169 static void gst_single_queue_free (GstSingleQueue * squeue);
171 static void wake_up_next_non_linked (GstMultiQueue * mq);
172 static void compute_high_id (GstMultiQueue * mq);
174 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",
175 GST_PAD_SINK,
176 GST_PAD_REQUEST,
177 GST_STATIC_CAPS_ANY);
179 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
180 GST_PAD_SRC,
181 GST_PAD_SOMETIMES,
182 GST_STATIC_CAPS_ANY);
184 GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);
185 #define GST_CAT_DEFAULT (multi_queue_debug)
187 /* default limits, we try to keep up to 2 seconds of data and if there is not
188 * time, up to 10 MB. The number of buffers is dynamically scaled to make sure
189 * there is data in the queues. Normally, the byte and time limits are not hit
190 * in theses conditions. */
191 #define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
192 #define DEFAULT_MAX_SIZE_BUFFERS 5
193 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND
195 /* second limits. When we hit one of the above limits we are probably dealing
196 * with a badly muxed file and we scale the limits to these emergency values.
197 * This is currently not yet implemented. */
198 #define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
199 #define DEFAULT_EXTRA_SIZE_BUFFERS 5
200 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND
202 /* Signals and args */
203 enum
204 {
205 SIGNAL_UNDERRUN,
206 SIGNAL_OVERRUN,
207 LAST_SIGNAL
208 };
210 enum
211 {
212 ARG_0,
213 ARG_EXTRA_SIZE_BYTES,
214 ARG_EXTRA_SIZE_BUFFERS,
215 ARG_EXTRA_SIZE_TIME,
216 ARG_MAX_SIZE_BYTES,
217 ARG_MAX_SIZE_BUFFERS,
218 ARG_MAX_SIZE_TIME,
219 };
221 #define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
222 g_mutex_lock (q->qlock); \
223 } G_STMT_END
225 #define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
226 g_mutex_unlock (q->qlock); \
227 } G_STMT_END
229 static void gst_multi_queue_finalize (GObject * object);
230 static void gst_multi_queue_set_property (GObject * object,
231 guint prop_id, const GValue * value, GParamSpec * pspec);
232 static void gst_multi_queue_get_property (GObject * object,
233 guint prop_id, GValue * value, GParamSpec * pspec);
235 static GstPad *gst_multi_queue_request_new_pad (GstElement * element,
236 GstPadTemplate * temp, const gchar * name);
237 static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);
239 static void gst_multi_queue_loop (GstPad * pad);
241 #define _do_init(bla) \
242 GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");
244 GST_BOILERPLATE_FULL (GstMultiQueue, gst_multi_queue, GstElement,
245 GST_TYPE_ELEMENT, _do_init);
247 static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };
249 static void
250 gst_multi_queue_base_init (gpointer g_class)
251 {
252 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
254 gst_element_class_set_details_simple (gstelement_class,
255 "MultiQueue",
256 "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");
257 gst_element_class_add_pad_template (gstelement_class,
258 gst_static_pad_template_get (&sinktemplate));
259 gst_element_class_add_pad_template (gstelement_class,
260 gst_static_pad_template_get (&srctemplate));
261 }
263 static void
264 gst_multi_queue_class_init (GstMultiQueueClass * klass)
265 {
266 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
267 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
269 gobject_class->set_property =
270 GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);
271 gobject_class->get_property =
272 GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);
274 /* SIGNALS */
275 gst_multi_queue_signals[SIGNAL_UNDERRUN] =
276 g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
277 G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,
278 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
280 gst_multi_queue_signals[SIGNAL_OVERRUN] =
281 g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
282 G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,
283 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
285 /* PROPERTIES */
287 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
288 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
289 "Max. amount of data in the queue (bytes, 0=disable)",
290 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
291 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
292 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
293 g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
294 "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
295 DEFAULT_MAX_SIZE_BUFFERS,
296 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
297 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
298 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
299 "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
300 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
302 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,
303 g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",
304 "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",
305 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES,
306 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
307 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,
308 g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",
309 "Amount of buffers the queues can grow if one of them is empty (0=disable)",
310 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS,
311 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
312 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,
313 g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",
314 "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",
315 0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME,
316 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
318 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_multi_queue_finalize);
320 gstelement_class->request_new_pad =
321 GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);
322 gstelement_class->release_pad =
323 GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);
324 }
326 static void
327 gst_multi_queue_init (GstMultiQueue * mqueue, GstMultiQueueClass * klass)
328 {
329 mqueue->nbqueues = 0;
330 mqueue->queues = NULL;
332 mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;
333 mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;
334 mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;
336 mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;
337 mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;
338 mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
340 mqueue->counter = 1;
341 mqueue->highid = -1;
342 mqueue->nextnotlinked = -1;
344 mqueue->qlock = g_mutex_new ();
345 }
347 static void
348 gst_multi_queue_finalize (GObject * object)
349 {
350 GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);
352 g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);
353 g_list_free (mqueue->queues);
354 mqueue->queues = NULL;
356 /* free/unref instance data */
357 g_mutex_free (mqueue->qlock);
359 G_OBJECT_CLASS (parent_class)->finalize (object);
360 }
362 #define SET_CHILD_PROPERTY(mq,format) G_STMT_START { \
363 GList * tmp = mq->queues; \
364 while (tmp) { \
365 GstSingleQueue *q = (GstSingleQueue*)tmp->data; \
366 q->max_size.format = mq->max_size.format; \
367 tmp = g_list_next(tmp); \
368 }; \
369 } G_STMT_END
371 static void
372 gst_multi_queue_set_property (GObject * object, guint prop_id,
373 const GValue * value, GParamSpec * pspec)
374 {
375 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
377 switch (prop_id) {
378 case ARG_MAX_SIZE_BYTES:
379 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
380 mq->max_size.bytes = g_value_get_uint (value);
381 SET_CHILD_PROPERTY (mq, bytes);
382 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
383 break;
384 case ARG_MAX_SIZE_BUFFERS:
385 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
386 mq->max_size.visible = g_value_get_uint (value);
387 SET_CHILD_PROPERTY (mq, visible);
388 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
389 break;
390 case ARG_MAX_SIZE_TIME:
391 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
392 mq->max_size.time = g_value_get_uint64 (value);
393 SET_CHILD_PROPERTY (mq, time);
394 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
395 break;
396 case ARG_EXTRA_SIZE_BYTES:
397 mq->extra_size.bytes = g_value_get_uint (value);
398 break;
399 case ARG_EXTRA_SIZE_BUFFERS:
400 mq->extra_size.visible = g_value_get_uint (value);
401 break;
402 case ARG_EXTRA_SIZE_TIME:
403 mq->extra_size.time = g_value_get_uint64 (value);
404 break;
405 default:
406 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
407 break;
408 }
409 }
411 static void
412 gst_multi_queue_get_property (GObject * object, guint prop_id,
413 GValue * value, GParamSpec * pspec)
414 {
415 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
417 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
419 switch (prop_id) {
420 case ARG_EXTRA_SIZE_BYTES:
421 g_value_set_uint (value, mq->extra_size.bytes);
422 break;
423 case ARG_EXTRA_SIZE_BUFFERS:
424 g_value_set_uint (value, mq->extra_size.visible);
425 break;
426 case ARG_EXTRA_SIZE_TIME:
427 g_value_set_uint64 (value, mq->extra_size.time);
428 break;
429 case ARG_MAX_SIZE_BYTES:
430 g_value_set_uint (value, mq->max_size.bytes);
431 break;
432 case ARG_MAX_SIZE_BUFFERS:
433 g_value_set_uint (value, mq->max_size.visible);
434 break;
435 case ARG_MAX_SIZE_TIME:
436 g_value_set_uint64 (value, mq->max_size.time);
437 break;
438 default:
439 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
440 break;
441 }
443 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
444 }
446 static GList *
447 gst_multi_queue_get_internal_links (GstPad * pad)
448 {
449 GList *res = NULL;
450 GstMultiQueue *mqueue;
451 GstSingleQueue *sq = NULL;
452 GList *tmp;
454 g_return_val_if_fail (GST_IS_PAD (pad), NULL);
456 mqueue = GST_MULTI_QUEUE (GST_PAD_PARENT (pad));
457 if (!mqueue)
458 goto no_parent;
460 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
461 /* Find which single queue it belongs to */
462 for (tmp = mqueue->queues; tmp && !res; tmp = g_list_next (tmp)) {
463 sq = (GstSingleQueue *) tmp->data;
465 if (sq->sinkpad == pad)
466 res = g_list_prepend (res, sq->srcpad);
467 if (sq->srcpad == pad)
468 res = g_list_prepend (res, sq->sinkpad);
469 }
471 if (!res)
472 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
473 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
475 return res;
477 no_parent:
478 {
479 GST_DEBUG_OBJECT (pad, "no parent");
480 return NULL;
481 }
482 }
485 /*
486 * GstElement methods
487 */
489 static GstPad *
490 gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,
491 const gchar * name)
492 {
493 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
494 GstSingleQueue *squeue;
496 GST_LOG_OBJECT (element, "name : %s", GST_STR_NULL (name));
498 /* Create a new single queue, add the sink and source pad and return the sink pad */
499 squeue = gst_single_queue_new (mqueue);
501 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
502 mqueue->queues = g_list_append (mqueue->queues, squeue);
503 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
505 GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",
506 GST_DEBUG_PAD_NAME (squeue->sinkpad));
508 return squeue->sinkpad;
509 }
511 static void
512 gst_multi_queue_release_pad (GstElement * element, GstPad * pad)
513 {
514 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
515 GstSingleQueue *sq = NULL;
516 GList *tmp;
518 GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));
520 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
521 /* Find which single queue it belongs to, knowing that it should be a sinkpad */
522 for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
523 sq = (GstSingleQueue *) tmp->data;
525 if (sq->sinkpad == pad)
526 break;
527 }
529 if (!tmp) {
530 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
531 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
532 return;
533 }
535 /* FIXME: The removal of the singlequeue should probably not happen until it
536 * finishes draining */
538 /* remove it from the list */
539 mqueue->queues = g_list_delete_link (mqueue->queues, tmp);
541 /* FIXME : recompute next-non-linked */
542 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
544 /* delete SingleQueue */
545 gst_data_queue_set_flushing (sq->queue, TRUE);
547 gst_pad_set_active (sq->srcpad, FALSE);
548 gst_pad_set_active (sq->sinkpad, FALSE);
549 gst_element_remove_pad (element, sq->srcpad);
550 gst_element_remove_pad (element, sq->sinkpad);
551 gst_single_queue_free (sq);
552 }
554 static gboolean
555 gst_single_queue_flush (GstMultiQueue * mq, GstSingleQueue * sq, gboolean flush)
556 {
557 gboolean result;
559 GST_DEBUG_OBJECT (mq, "flush %s queue %d", (flush ? "start" : "stop"),
560 sq->id);
562 if (flush) {
563 sq->srcresult = GST_FLOW_WRONG_STATE;
564 gst_data_queue_set_flushing (sq->queue, TRUE);
566 /* wake up non-linked task */
567 GST_LOG_OBJECT (mq, "SingleQueue %d : waking up eventually waiting task",
568 sq->id);
569 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
570 g_cond_signal (sq->turn);
571 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
573 GST_LOG_OBJECT (mq, "SingleQueue %d : pausing task", sq->id);
574 result = gst_pad_pause_task (sq->srcpad);
575 } else {
576 gst_data_queue_flush (sq->queue);
577 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
578 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
579 /* All pads start off not-linked for a smooth kick-off */
580 sq->srcresult = GST_FLOW_OK;
581 sq->cur_time = 0;
582 sq->max_size.visible = mq->max_size.visible;
583 sq->is_eos = FALSE;
584 sq->inextra = FALSE;
585 sq->nextid = 0;
586 sq->oldid = 0;
587 gst_data_queue_set_flushing (sq->queue, FALSE);
589 GST_LOG_OBJECT (mq, "SingleQueue %d : starting task", sq->id);
590 result =
591 gst_pad_start_task (sq->srcpad, (GstTaskFunction) gst_multi_queue_loop,
592 sq->srcpad);
593 }
594 return result;
595 }
597 /* calculate the diff between running time on the sink and src of the queue.
598 * This is the total amount of time in the queue.
599 * WITH LOCK TAKEN */
600 static void
601 update_time_level (GstMultiQueue * mq, GstSingleQueue * sq)
602 {
603 gint64 sink_time, src_time;
605 sink_time =
606 gst_segment_to_running_time (&sq->sink_segment, GST_FORMAT_TIME,
607 sq->sink_segment.last_stop);
609 src_time = gst_segment_to_running_time (&sq->src_segment, GST_FORMAT_TIME,
610 sq->src_segment.last_stop);
612 GST_DEBUG_OBJECT (mq,
613 "queue %d, sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT, sq->id,
614 GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
616 /* This allows for streams with out of order timestamping - sometimes the
617 * emerging timestamp is later than the arriving one(s) */
618 if (sink_time >= src_time)
619 sq->cur_time = sink_time - src_time;
620 else
621 sq->cur_time = 0;
622 }
624 /* take a NEWSEGMENT event and apply the values to segment, updating the time
625 * level of queue. */
626 static void
627 apply_segment (GstMultiQueue * mq, GstSingleQueue * sq, GstEvent * event,
628 GstSegment * segment)
629 {
630 gboolean update;
631 GstFormat format;
632 gdouble rate, arate;
633 gint64 start, stop, time;
635 gst_event_parse_new_segment_full (event, &update, &rate, &arate,
636 &format, &start, &stop, &time);
638 /* now configure the values, we use these to track timestamps on the
639 * sinkpad. */
640 if (format != GST_FORMAT_TIME) {
641 /* non-time format, pretent the current time segment is closed with a
642 * 0 start and unknown stop time. */
643 update = FALSE;
644 format = GST_FORMAT_TIME;
645 start = 0;
646 stop = -1;
647 time = 0;
648 }
650 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
652 gst_segment_set_newsegment_full (segment, update,
653 rate, arate, format, start, stop, time);
655 GST_DEBUG_OBJECT (mq,
656 "queue %d, configured NEWSEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
658 /* segment can update the time level of the queue */
659 update_time_level (mq, sq);
661 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
662 }
664 /* take a buffer and update segment, updating the time level of the queue. */
665 static void
666 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
667 GstClockTime duration, GstSegment * segment)
668 {
669 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
671 /* if no timestamp is set, assume it's continuous with the previous
672 * time */
673 if (timestamp == GST_CLOCK_TIME_NONE)
674 timestamp = segment->last_stop;
676 /* add duration */
677 if (duration != GST_CLOCK_TIME_NONE)
678 timestamp += duration;
680 GST_DEBUG_OBJECT (mq, "queue %d, last_stop updated to %" GST_TIME_FORMAT,
681 sq->id, GST_TIME_ARGS (timestamp));
683 gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
685 /* calc diff with other end */
686 update_time_level (mq, sq);
687 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
688 }
690 static GstFlowReturn
691 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
692 GstMiniObject * object)
693 {
694 GstFlowReturn result = GST_FLOW_OK;
696 if (GST_IS_BUFFER (object)) {
697 GstBuffer *buffer;
698 GstClockTime timestamp, duration;
699 GstCaps *caps;
701 buffer = GST_BUFFER_CAST (object);
702 timestamp = GST_BUFFER_TIMESTAMP (buffer);
703 duration = GST_BUFFER_DURATION (buffer);
704 caps = GST_BUFFER_CAPS (buffer);
706 apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
708 /* Applying the buffer may have made the queue non-full again, unblock it if needed */
709 gst_data_queue_limits_changed (sq->queue);
711 GST_DEBUG_OBJECT (mq,
712 "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
713 sq->id, buffer, GST_TIME_ARGS (timestamp));
715 /* Set caps on pad before pushing, this avoids core calling the accpetcaps
716 * function on the srcpad, which will call acceptcaps upstream, which might
717 * not accept these caps (anymore). */
718 if (caps && caps != GST_PAD_CAPS (sq->srcpad))
719 gst_pad_set_caps (sq->srcpad, caps);
721 result = gst_pad_push (sq->srcpad, buffer);
722 } else if (GST_IS_EVENT (object)) {
723 GstEvent *event;
725 event = GST_EVENT_CAST (object);
727 switch (GST_EVENT_TYPE (event)) {
728 case GST_EVENT_EOS:
729 result = GST_FLOW_UNEXPECTED;
730 break;
731 case GST_EVENT_NEWSEGMENT:
732 apply_segment (mq, sq, event, &sq->src_segment);
733 /* Applying the segment may have made the queue non-full again, unblock it if needed */
734 gst_data_queue_limits_changed (sq->queue);
735 break;
736 default:
737 break;
738 }
740 GST_DEBUG_OBJECT (mq,
741 "SingleQueue %d : Pushing event %p of type %s",
742 sq->id, event, GST_EVENT_TYPE_NAME (event));
744 gst_pad_push_event (sq->srcpad, event);
745 } else {
746 g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
747 sq->id);
748 }
749 return result;
751 /* ERRORS */
752 }
754 static GstMiniObject *
755 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
756 {
757 GstMiniObject *res;
759 res = item->object;
760 item->object = NULL;
762 return res;
763 }
765 static void
766 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
767 {
768 if (item->object)
769 gst_mini_object_unref (item->object);
770 g_free (item);
771 }
773 /* takes ownership of passed mini object! */
774 static GstMultiQueueItem *
775 gst_multi_queue_item_new (GstMiniObject * object, guint32 curid)
776 {
777 GstMultiQueueItem *item;
779 item = g_new (GstMultiQueueItem, 1);
780 item->object = object;
781 item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
782 item->posid = curid;
784 if (GST_IS_BUFFER (object)) {
785 item->size = GST_BUFFER_SIZE (object);
786 item->duration = GST_BUFFER_DURATION (object);
787 if (item->duration == GST_CLOCK_TIME_NONE)
788 item->duration = 0;
789 item->visible = TRUE;
790 } else {
791 item->size = 0;
792 item->duration = 0;
793 item->visible = FALSE;
794 }
795 return item;
796 }
798 /* Each main loop attempts to push buffers until the return value
799 * is not-linked. not-linked pads are not allowed to push data beyond
800 * any linked pads, so they don't 'rush ahead of the pack'.
801 */
802 static void
803 gst_multi_queue_loop (GstPad * pad)
804 {
805 GstSingleQueue *sq;
806 GstMultiQueueItem *item;
807 GstDataQueueItem *sitem;
808 GstMultiQueue *mq;
809 GstMiniObject *object;
810 guint32 newid;
811 guint32 oldid = G_MAXUINT32;
812 GstFlowReturn result;
814 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
815 mq = sq->mqueue;
817 do {
818 GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
820 /* Get something from the queue, blocking until that happens, or we get
821 * flushed */
822 if (!(gst_data_queue_pop (sq->queue, &sitem)))
823 goto out_flushing;
825 item = (GstMultiQueueItem *) sitem;
826 newid = item->posid;
828 /* steal the object and destroy the item */
829 object = gst_multi_queue_item_steal_object (item);
830 gst_multi_queue_item_destroy (item);
832 GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
833 sq->id, newid, oldid);
835 /* If we're not-linked, we do some extra work because we might need to
836 * wait before pushing. If we're linked but there's a gap in the IDs,
837 * or it's the first loop, or we just passed the previous highid,
838 * we might need to wake some sleeping pad up, so there's extra work
839 * there too */
840 if (sq->srcresult == GST_FLOW_NOT_LINKED ||
841 (oldid == G_MAXUINT32) || (newid != (oldid + 1)) ||
842 oldid > mq->highid) {
843 GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
844 gst_flow_get_name (sq->srcresult));
846 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
848 /* Update the nextid so other threads know when to wake us up */
849 sq->nextid = newid;
851 /* Update the oldid (the last ID we output) for highid tracking */
852 if (oldid != G_MAXUINT32)
853 sq->oldid = oldid;
855 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
856 /* Go to sleep until it's time to push this buffer */
858 /* Recompute the highid */
859 compute_high_id (mq);
860 while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
861 GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
862 "newid %u and highid %u", sq->id, newid, mq->highid);
865 /* Wake up all non-linked pads before we sleep */
866 wake_up_next_non_linked (mq);
868 mq->numwaiting++;
869 g_cond_wait (sq->turn, mq->qlock);
870 mq->numwaiting--;
872 GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
873 "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
874 }
876 /* Re-compute the high_id in case someone else pushed */
877 compute_high_id (mq);
878 } else {
879 compute_high_id (mq);
880 /* Wake up all non-linked pads */
881 wake_up_next_non_linked (mq);
882 }
883 /* We're done waiting, we can clear the nextid */
884 sq->nextid = 0;
886 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
887 }
889 GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
890 gst_flow_get_name (sq->srcresult));
892 /* Try to push out the new object */
893 result = gst_single_queue_push_one (mq, sq, object);
894 sq->srcresult = result;
896 if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED)
897 goto out_flushing;
899 GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
900 gst_flow_get_name (sq->srcresult));
902 oldid = newid;
903 }
904 while (TRUE);
906 out_flushing:
907 {
908 /* Need to make sure wake up any sleeping pads when we exit */
909 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
910 compute_high_id (mq);
911 wake_up_next_non_linked (mq);
912 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
914 gst_data_queue_set_flushing (sq->queue, TRUE);
915 gst_pad_pause_task (sq->srcpad);
916 GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
917 "SingleQueue[%d] task paused, reason:%s",
918 sq->id, gst_flow_get_name (sq->srcresult));
919 return;
920 }
921 }
923 /**
924 * gst_multi_queue_chain:
925 *
926 * This is similar to GstQueue's chain function, except:
927 * _ we don't have leak behavioures,
928 * _ we push with a unique id (curid)
929 */
930 static GstFlowReturn
931 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
932 {
933 GstSingleQueue *sq;
934 GstMultiQueue *mq;
935 GstMultiQueueItem *item;
936 GstFlowReturn ret = GST_FLOW_OK;
937 guint32 curid;
938 GstClockTime timestamp, duration;
940 sq = gst_pad_get_element_private (pad);
941 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
943 /* Get a unique incrementing id */
944 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
945 curid = mq->counter++;
946 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
948 GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
949 sq->id, buffer, curid);
951 item = gst_multi_queue_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
953 timestamp = GST_BUFFER_TIMESTAMP (buffer);
954 duration = GST_BUFFER_DURATION (buffer);
956 if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
957 goto flushing;
959 /* update time level, we must do this after pushing the data in the queue so
960 * that we never end up filling the queue first. */
961 apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
963 done:
964 gst_object_unref (mq);
966 return ret;
968 /* ERRORS */
969 flushing:
970 {
971 ret = sq->srcresult;
972 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
973 sq->id, gst_flow_get_name (ret));
974 gst_multi_queue_item_destroy (item);
975 goto done;
976 }
977 }
979 static gboolean
980 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
981 {
982 GstSingleQueue *sq;
984 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
986 if (active) {
987 /* All pads start off linked until they push one buffer */
988 sq->srcresult = GST_FLOW_OK;
989 } else {
990 sq->srcresult = GST_FLOW_WRONG_STATE;
991 gst_data_queue_flush (sq->queue);
992 }
993 return TRUE;
994 }
996 static gboolean
997 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
998 {
999 GstSingleQueue *sq;
1000 GstMultiQueue *mq;
1001 guint32 curid;
1002 GstMultiQueueItem *item;
1003 gboolean res;
1004 GstEventType type;
1005 GstEvent *sref = NULL;
1007 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1008 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
1010 type = GST_EVENT_TYPE (event);
1012 switch (type) {
1013 case GST_EVENT_FLUSH_START:
1014 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
1015 sq->id);
1017 res = gst_pad_push_event (sq->srcpad, event);
1019 gst_single_queue_flush (mq, sq, TRUE);
1020 goto done;
1022 case GST_EVENT_FLUSH_STOP:
1023 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
1024 sq->id);
1026 res = gst_pad_push_event (sq->srcpad, event);
1028 gst_single_queue_flush (mq, sq, FALSE);
1029 goto done;
1030 case GST_EVENT_NEWSEGMENT:
1031 /* take ref because the queue will take ownership and we need the event
1032 * afterwards to update the segment */
1033 sref = gst_event_ref (event);
1034 break;
1036 default:
1037 if (!(GST_EVENT_IS_SERIALIZED (event))) {
1038 res = gst_pad_push_event (sq->srcpad, event);
1039 goto done;
1040 }
1041 break;
1042 }
1044 /* Get an unique incrementing id */
1045 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1046 curid = mq->counter++;
1047 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1049 item = gst_multi_queue_item_new ((GstMiniObject *) event, curid);
1051 GST_DEBUG_OBJECT (mq,
1052 "SingleQueue %d : Enqueuing event %p of type %s with id %d",
1053 sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
1055 if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1056 goto flushing;
1058 /* mark EOS when we received one, we must do that after putting the
1059 * buffer in the queue because EOS marks the buffer as filled. No need to take
1060 * a lock, the _check_full happens from this thread only, right before pushing
1061 * into dataqueue. */
1062 switch (type) {
1063 case GST_EVENT_EOS:
1064 sq->is_eos = TRUE;
1065 break;
1066 case GST_EVENT_NEWSEGMENT:
1067 apply_segment (mq, sq, sref, &sq->sink_segment);
1068 gst_event_unref (sref);
1069 break;
1070 default:
1071 break;
1072 }
1073 done:
1074 gst_object_unref (mq);
1075 return res;
1077 flushing:
1078 {
1079 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1080 sq->id, gst_flow_get_name (sq->srcresult));
1081 if (sref)
1082 gst_event_unref (sref);
1083 gst_multi_queue_item_destroy (item);
1084 goto done;
1085 }
1086 }
1088 static GstCaps *
1089 gst_multi_queue_getcaps (GstPad * pad)
1090 {
1091 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1092 GstPad *otherpad;
1093 GstCaps *result;
1095 otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1097 GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
1099 result = gst_pad_peer_get_caps (otherpad);
1100 if (result == NULL)
1101 result = gst_caps_new_any ();
1103 return result;
1104 }
1106 static GstFlowReturn
1107 gst_multi_queue_bufferalloc (GstPad * pad, guint64 offset, guint size,
1108 GstCaps * caps, GstBuffer ** buf)
1109 {
1110 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1112 return gst_pad_alloc_buffer (sq->srcpad, offset, size, caps, buf);
1113 }
1115 static gboolean
1116 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1117 {
1118 GstMultiQueue *mq;
1119 GstSingleQueue *sq;
1120 gboolean result = FALSE;
1122 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1123 mq = sq->mqueue;
1125 GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1127 if (active) {
1128 result = gst_single_queue_flush (mq, sq, FALSE);
1129 } else {
1130 result = gst_single_queue_flush (mq, sq, TRUE);
1131 /* make sure streaming finishes */
1132 result |= gst_pad_stop_task (pad);
1133 }
1134 return result;
1135 }
1137 static gboolean
1138 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1139 {
1140 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1142 return gst_pad_push_event (sq->sinkpad, event);
1143 }
1145 static gboolean
1146 gst_multi_queue_src_query (GstPad * pad, GstQuery * query)
1147 {
1148 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1149 GstPad *peerpad;
1150 gboolean res;
1152 /* FIXME, Handle position offset depending on queue size */
1154 /* default handling */
1155 if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1156 goto no_peer;
1158 res = gst_pad_query (peerpad, query);
1160 gst_object_unref (peerpad);
1162 return res;
1164 /* ERRORS */
1165 no_peer:
1166 {
1167 GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1168 return FALSE;
1169 }
1170 }
1172 /*
1173 * Next-non-linked functions
1174 */
1176 /* WITH LOCK TAKEN */
1177 static void
1178 wake_up_next_non_linked (GstMultiQueue * mq)
1179 {
1180 GList *tmp;
1182 /* maybe no-one is waiting */
1183 if (mq->numwaiting < 1)
1184 return;
1186 /* Else figure out which singlequeue(s) need waking up */
1187 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1188 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1190 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1191 if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1192 GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1193 g_cond_signal (sq->turn);
1194 }
1195 }
1196 }
1197 }
1199 /* WITH LOCK TAKEN */
1200 static void
1201 compute_high_id (GstMultiQueue * mq)
1202 {
1203 /* The high-id is either the highest id among the linked pads, or if all
1204 * pads are not-linked, it's the lowest not-linked pad */
1205 GList *tmp;
1206 guint32 lowest = G_MAXUINT32;
1207 guint32 highid = G_MAXUINT32;
1209 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1210 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1212 GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1213 sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1215 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1216 /* No need to consider queues which are not waiting */
1217 if (sq->nextid == 0) {
1218 GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1219 continue;
1220 }
1222 if (sq->nextid < lowest)
1223 lowest = sq->nextid;
1224 } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1225 /* If we don't have a global highid, or the global highid is lower than
1226 * this single queue's last outputted id, store the queue's one,
1227 * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1228 if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1229 highid = sq->oldid;
1230 }
1231 }
1233 if (highid == G_MAXUINT32 || lowest < highid)
1234 mq->highid = lowest;
1235 else
1236 mq->highid = highid;
1238 GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1239 lowest);
1240 }
1242 #define IS_FILLED(format, value) ((sq->max_size.format) != 0 && \
1243 (sq->max_size.format) <= (value))
1245 /*
1246 * GstSingleQueue functions
1247 */
1248 static void
1249 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1250 {
1251 GstMultiQueue *mq = sq->mqueue;
1252 GList *tmp;
1253 GstDataQueueSize size;
1254 gboolean filled = FALSE;
1256 gst_data_queue_get_level (sq->queue, &size);
1258 GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1260 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1261 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1262 GstSingleQueue *ssq = (GstSingleQueue *) tmp->data;
1263 GstDataQueueSize ssize;
1265 GST_LOG_OBJECT (mq, "Checking Queue %d", ssq->id);
1267 if (gst_data_queue_is_empty (ssq->queue)) {
1268 GST_LOG_OBJECT (mq, "Queue %d is empty", ssq->id);
1269 if (IS_FILLED (visible, size.visible)) {
1270 sq->max_size.visible = size.visible + 1;
1271 GST_DEBUG_OBJECT (mq,
1272 "Another queue is empty, bumping single queue %d max visible to %d",
1273 sq->id, sq->max_size.visible);
1274 }
1275 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1276 goto beach;
1277 }
1278 /* check if we reached the hard time/bytes limits */
1279 gst_data_queue_get_level (ssq->queue, &ssize);
1281 GST_DEBUG_OBJECT (mq,
1282 "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1283 G_GUINT64_FORMAT, ssq->id, ssize.visible, sq->max_size.visible,
1284 ssize.bytes, sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1286 /* if this queue is filled completely we must signal overrun */
1287 if (IS_FILLED (bytes, ssize.bytes) || IS_FILLED (time, sq->cur_time)) {
1288 GST_LOG_OBJECT (mq, "Queue %d is filled", ssq->id);
1289 filled = TRUE;
1290 }
1291 }
1292 /* no queues were empty */
1293 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1295 /* Overrun is always forwarded, since this is blocking the upstream element */
1296 if (filled) {
1297 GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1298 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1299 }
1301 beach:
1302 return;
1303 }
1305 static void
1306 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1307 {
1308 gboolean empty = TRUE;
1309 GstMultiQueue *mq = sq->mqueue;
1310 GList *tmp;
1312 GST_LOG_OBJECT (mq,
1313 "Single Queue %d is empty, Checking other single queues", sq->id);
1315 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1316 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1317 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1319 if (gst_data_queue_is_full (sq->queue)) {
1320 GstDataQueueSize size;
1322 gst_data_queue_get_level (sq->queue, &size);
1323 if (IS_FILLED (visible, size.visible)) {
1324 sq->max_size.visible = size.visible + 1;
1325 GST_DEBUG_OBJECT (mq,
1326 "queue %d is filled, bumping its max visible to %d", sq->id,
1327 sq->max_size.visible);
1328 gst_data_queue_limits_changed (sq->queue);
1329 }
1330 }
1331 if (!gst_data_queue_is_empty (sq->queue))
1332 empty = FALSE;
1333 }
1334 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1336 if (empty) {
1337 GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1338 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1339 }
1340 }
1342 static gboolean
1343 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1344 guint64 time, GstSingleQueue * sq)
1345 {
1346 gboolean res;
1348 GST_DEBUG ("queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT
1349 "/%" G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1350 sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1352 /* we are always filled on EOS */
1353 if (sq->is_eos)
1354 return TRUE;
1356 /* we never go past the max visible items */
1357 if (IS_FILLED (visible, visible))
1358 return TRUE;
1360 if (sq->cur_time != 0) {
1361 /* if we have valid time in the queue, check */
1362 res = IS_FILLED (time, sq->cur_time);
1363 } else {
1364 /* no valid time, check bytes */
1365 res = IS_FILLED (bytes, bytes);
1366 }
1367 return res;
1368 }
1370 static void
1371 gst_single_queue_free (GstSingleQueue * sq)
1372 {
1373 /* DRAIN QUEUE */
1374 gst_data_queue_flush (sq->queue);
1375 g_object_unref (sq->queue);
1376 g_cond_free (sq->turn);
1377 g_free (sq);
1378 }
1380 static GstSingleQueue *
1381 gst_single_queue_new (GstMultiQueue * mqueue)
1382 {
1383 GstSingleQueue *sq;
1384 gchar *tmp;
1386 sq = g_new0 (GstSingleQueue, 1);
1388 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1389 sq->id = mqueue->nbqueues++;
1391 /* copy over max_size and extra_size so we don't need to take the lock
1392 * any longer when checking if the queue is full. */
1393 sq->max_size.visible = mqueue->max_size.visible;
1394 sq->max_size.bytes = mqueue->max_size.bytes;
1395 sq->max_size.time = mqueue->max_size.time;
1397 sq->extra_size.visible = mqueue->extra_size.visible;
1398 sq->extra_size.bytes = mqueue->extra_size.bytes;
1399 sq->extra_size.time = mqueue->extra_size.time;
1401 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1403 GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1405 sq->mqueue = mqueue;
1406 sq->srcresult = GST_FLOW_WRONG_STATE;
1407 sq->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction)
1408 single_queue_check_full, sq);
1409 sq->is_eos = FALSE;
1410 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1411 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1413 sq->nextid = 0;
1414 sq->oldid = 0;
1415 sq->turn = g_cond_new ();
1417 /* attach to underrun/overrun signals to handle non-starvation */
1418 g_signal_connect (G_OBJECT (sq->queue), "full",
1419 G_CALLBACK (single_queue_overrun_cb), sq);
1420 g_signal_connect (G_OBJECT (sq->queue), "empty",
1421 G_CALLBACK (single_queue_underrun_cb), sq);
1423 tmp = g_strdup_printf ("sink%d", sq->id);
1424 sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
1425 g_free (tmp);
1427 gst_pad_set_chain_function (sq->sinkpad,
1428 GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1429 gst_pad_set_activatepush_function (sq->sinkpad,
1430 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1431 gst_pad_set_event_function (sq->sinkpad,
1432 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1433 gst_pad_set_getcaps_function (sq->sinkpad,
1434 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1435 gst_pad_set_bufferalloc_function (sq->sinkpad,
1436 GST_DEBUG_FUNCPTR (gst_multi_queue_bufferalloc));
1437 gst_pad_set_internal_link_function (sq->sinkpad,
1438 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1440 tmp = g_strdup_printf ("src%d", sq->id);
1441 sq->srcpad = gst_pad_new_from_static_template (&srctemplate, tmp);
1442 g_free (tmp);
1444 gst_pad_set_activatepush_function (sq->srcpad,
1445 GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1446 gst_pad_set_getcaps_function (sq->srcpad,
1447 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1448 gst_pad_set_event_function (sq->srcpad,
1449 GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1450 gst_pad_set_query_function (sq->srcpad,
1451 GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1452 gst_pad_set_internal_link_function (sq->srcpad,
1453 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1455 gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1456 gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1458 gst_pad_set_active (sq->srcpad, TRUE);
1459 gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1461 gst_pad_set_active (sq->sinkpad, TRUE);
1462 gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1464 GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1465 sq->id);
1467 return sq;
1468 }