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 * Copyright (C) 2008 Thijs Vermeir <thijsvermeir@gmail.com>
6 *
7 * gstmultiqueue.c:
8 *
9 * This library is free software; you can redistribute it and/or
10 * modify it under the terms of the GNU Library General Public
11 * License as published by the Free Software Foundation; either
12 * version 2 of the License, or (at your option) any later version.
13 *
14 * This library is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * Library General Public License for more details.
18 *
19 * You should have received a copy of the GNU Library General Public
20 * License along with this library; if not, write to the
21 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
22 * Boston, MA 02111-1307, USA.
23 */
25 /**
26 * SECTION:element-multiqueue
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 * Example: requesting sink1 will generate src1.
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>\n"
257 "Thijs Vermeir <thijsvermeir@gmail.com>");
258 gst_element_class_add_pad_template (gstelement_class,
259 gst_static_pad_template_get (&sinktemplate));
260 gst_element_class_add_pad_template (gstelement_class,
261 gst_static_pad_template_get (&srctemplate));
262 }
264 static void
265 gst_multi_queue_class_init (GstMultiQueueClass * klass)
266 {
267 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
268 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
270 gobject_class->set_property =
271 GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);
272 gobject_class->get_property =
273 GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);
275 /* SIGNALS */
277 /**
278 * GstMultiQueue::underrun:
279 * @multiqueue: the multqueue instance
280 *
281 * This signal is emitted from the streaming thread when there is
282 * no data in any of the queues inside the multiqueue instance (underrun).
283 *
284 * This indicates either starvation or EOS from the upstream data sources.
285 */
286 gst_multi_queue_signals[SIGNAL_UNDERRUN] =
287 g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
288 G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,
289 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
291 /**
292 * GstMultiQueue::overrun:
293 * @multiqueue: the multiqueue instance
294 *
295 * Reports that one of the queues in the multiqueue is full (overrun).
296 * A queue is full if the total amount of data inside it (num-buffers, time,
297 * size) is higher than the boundary values which can be set through the
298 * GObject properties.
299 *
300 * This can be used as an indicator of pre-roll.
301 */
302 gst_multi_queue_signals[SIGNAL_OVERRUN] =
303 g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
304 G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,
305 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
307 /* PROPERTIES */
309 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
310 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
311 "Max. amount of data in the queue (bytes, 0=disable)",
312 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
313 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
314 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
315 g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
316 "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
317 DEFAULT_MAX_SIZE_BUFFERS,
318 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
319 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
320 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
321 "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
322 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
324 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,
325 g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",
326 "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",
327 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES,
328 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
329 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,
330 g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",
331 "Amount of buffers the queues can grow if one of them is empty (0=disable)",
332 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS,
333 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
334 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,
335 g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",
336 "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",
337 0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME,
338 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
340 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_multi_queue_finalize);
342 gstelement_class->request_new_pad =
343 GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);
344 gstelement_class->release_pad =
345 GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);
346 }
348 static void
349 gst_multi_queue_init (GstMultiQueue * mqueue, GstMultiQueueClass * klass)
350 {
351 mqueue->nbqueues = 0;
352 mqueue->queues = NULL;
354 mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;
355 mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;
356 mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;
358 mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;
359 mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;
360 mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
362 mqueue->counter = 1;
363 mqueue->highid = -1;
364 mqueue->nextnotlinked = -1;
366 mqueue->qlock = g_mutex_new ();
367 }
369 static void
370 gst_multi_queue_finalize (GObject * object)
371 {
372 GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);
374 g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);
375 g_list_free (mqueue->queues);
376 mqueue->queues = NULL;
378 /* free/unref instance data */
379 g_mutex_free (mqueue->qlock);
381 G_OBJECT_CLASS (parent_class)->finalize (object);
382 }
384 #define SET_CHILD_PROPERTY(mq,format) G_STMT_START { \
385 GList * tmp = mq->queues; \
386 while (tmp) { \
387 GstSingleQueue *q = (GstSingleQueue*)tmp->data; \
388 q->max_size.format = mq->max_size.format; \
389 q->extra_size.format = mq->extra_size.format; \
390 tmp = g_list_next(tmp); \
391 }; \
392 } G_STMT_END
394 static void
395 gst_multi_queue_set_property (GObject * object, guint prop_id,
396 const GValue * value, GParamSpec * pspec)
397 {
398 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
400 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
401 switch (prop_id) {
402 case ARG_MAX_SIZE_BYTES:
403 mq->max_size.bytes = g_value_get_uint (value);
404 SET_CHILD_PROPERTY (mq, bytes);
405 break;
406 case ARG_MAX_SIZE_BUFFERS:
407 mq->max_size.visible = g_value_get_uint (value);
408 SET_CHILD_PROPERTY (mq, visible);
409 break;
410 case ARG_MAX_SIZE_TIME:
411 mq->max_size.time = g_value_get_uint64 (value);
412 SET_CHILD_PROPERTY (mq, time);
413 break;
414 case ARG_EXTRA_SIZE_BYTES:
415 mq->extra_size.bytes = g_value_get_uint (value);
416 SET_CHILD_PROPERTY (mq, bytes);
417 break;
418 case ARG_EXTRA_SIZE_BUFFERS:
419 mq->extra_size.visible = g_value_get_uint (value);
420 SET_CHILD_PROPERTY (mq, visible);
421 break;
422 case ARG_EXTRA_SIZE_TIME:
423 mq->extra_size.time = g_value_get_uint64 (value);
424 SET_CHILD_PROPERTY (mq, time);
425 break;
426 default:
427 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
428 break;
429 }
430 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
431 }
433 static void
434 gst_multi_queue_get_property (GObject * object, guint prop_id,
435 GValue * value, GParamSpec * pspec)
436 {
437 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
439 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
441 switch (prop_id) {
442 case ARG_EXTRA_SIZE_BYTES:
443 g_value_set_uint (value, mq->extra_size.bytes);
444 break;
445 case ARG_EXTRA_SIZE_BUFFERS:
446 g_value_set_uint (value, mq->extra_size.visible);
447 break;
448 case ARG_EXTRA_SIZE_TIME:
449 g_value_set_uint64 (value, mq->extra_size.time);
450 break;
451 case ARG_MAX_SIZE_BYTES:
452 g_value_set_uint (value, mq->max_size.bytes);
453 break;
454 case ARG_MAX_SIZE_BUFFERS:
455 g_value_set_uint (value, mq->max_size.visible);
456 break;
457 case ARG_MAX_SIZE_TIME:
458 g_value_set_uint64 (value, mq->max_size.time);
459 break;
460 default:
461 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
462 break;
463 }
465 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
466 }
468 static GList *
469 gst_multi_queue_get_internal_links (GstPad * pad)
470 {
471 GList *res = NULL;
472 GstMultiQueue *mqueue;
473 GstSingleQueue *sq = NULL;
474 GList *tmp;
476 g_return_val_if_fail (GST_IS_PAD (pad), NULL);
478 mqueue = GST_MULTI_QUEUE (GST_PAD_PARENT (pad));
479 if (!mqueue)
480 goto no_parent;
482 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
483 /* Find which single queue it belongs to */
484 for (tmp = mqueue->queues; tmp && !res; tmp = g_list_next (tmp)) {
485 sq = (GstSingleQueue *) tmp->data;
487 if (sq->sinkpad == pad)
488 res = g_list_prepend (res, sq->srcpad);
489 if (sq->srcpad == pad)
490 res = g_list_prepend (res, sq->sinkpad);
491 }
493 if (!res)
494 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
495 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
497 return res;
499 no_parent:
500 {
501 GST_DEBUG_OBJECT (pad, "no parent");
502 return NULL;
503 }
504 }
506 /*
507 * GstElement methods
508 */
510 static GstPad *
511 gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,
512 const gchar * name)
513 {
514 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
515 GstSingleQueue *squeue;
517 GST_LOG_OBJECT (element, "name : %s", GST_STR_NULL (name));
519 /* Create a new single queue, add the sink and source pad and return the sink pad */
520 squeue = gst_single_queue_new (mqueue);
522 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
523 mqueue->queues = g_list_append (mqueue->queues, squeue);
524 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
526 GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",
527 GST_DEBUG_PAD_NAME (squeue->sinkpad));
529 return squeue->sinkpad;
530 }
532 static void
533 gst_multi_queue_release_pad (GstElement * element, GstPad * pad)
534 {
535 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
536 GstSingleQueue *sq = NULL;
537 GList *tmp;
539 GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));
541 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
542 /* Find which single queue it belongs to, knowing that it should be a sinkpad */
543 for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
544 sq = (GstSingleQueue *) tmp->data;
546 if (sq->sinkpad == pad)
547 break;
548 }
550 if (!tmp) {
551 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
552 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
553 return;
554 }
556 /* FIXME: The removal of the singlequeue should probably not happen until it
557 * finishes draining */
559 /* remove it from the list */
560 mqueue->queues = g_list_delete_link (mqueue->queues, tmp);
562 /* FIXME : recompute next-non-linked */
563 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
565 /* delete SingleQueue */
566 gst_data_queue_set_flushing (sq->queue, TRUE);
568 gst_pad_set_active (sq->srcpad, FALSE);
569 gst_pad_set_active (sq->sinkpad, FALSE);
570 gst_element_remove_pad (element, sq->srcpad);
571 gst_element_remove_pad (element, sq->sinkpad);
572 gst_single_queue_free (sq);
573 }
575 static gboolean
576 gst_single_queue_flush (GstMultiQueue * mq, GstSingleQueue * sq, gboolean flush)
577 {
578 gboolean result;
580 GST_DEBUG_OBJECT (mq, "flush %s queue %d", (flush ? "start" : "stop"),
581 sq->id);
583 if (flush) {
584 sq->srcresult = GST_FLOW_WRONG_STATE;
585 gst_data_queue_set_flushing (sq->queue, TRUE);
587 /* wake up non-linked task */
588 GST_LOG_OBJECT (mq, "SingleQueue %d : waking up eventually waiting task",
589 sq->id);
590 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
591 g_cond_signal (sq->turn);
592 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
594 GST_LOG_OBJECT (mq, "SingleQueue %d : pausing task", sq->id);
595 result = gst_pad_pause_task (sq->srcpad);
596 } else {
597 gst_data_queue_flush (sq->queue);
598 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
599 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
600 /* All pads start off not-linked for a smooth kick-off */
601 sq->srcresult = GST_FLOW_OK;
602 sq->cur_time = 0;
603 sq->max_size.visible = mq->max_size.visible;
604 sq->extra_size.visible = mq->extra_size.visible;
605 sq->is_eos = FALSE;
606 sq->inextra = FALSE;
607 sq->nextid = 0;
608 sq->oldid = 0;
609 gst_data_queue_set_flushing (sq->queue, FALSE);
611 GST_LOG_OBJECT (mq, "SingleQueue %d : starting task", sq->id);
612 result =
613 gst_pad_start_task (sq->srcpad, (GstTaskFunction) gst_multi_queue_loop,
614 sq->srcpad);
615 }
616 return result;
617 }
619 /* calculate the diff between running time on the sink and src of the queue.
620 * This is the total amount of time in the queue.
621 * WITH LOCK TAKEN */
622 static void
623 update_time_level (GstMultiQueue * mq, GstSingleQueue * sq)
624 {
625 gint64 sink_time, src_time;
627 sink_time =
628 gst_segment_to_running_time (&sq->sink_segment, GST_FORMAT_TIME,
629 sq->sink_segment.last_stop);
630 if (sink_time == GST_CLOCK_TIME_NONE)
631 goto beach;
633 src_time = gst_segment_to_running_time (&sq->src_segment, GST_FORMAT_TIME,
634 sq->src_segment.last_stop);
635 if (src_time == GST_CLOCK_TIME_NONE)
636 goto beach;
638 GST_DEBUG_OBJECT (mq,
639 "queue %d, sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT, sq->id,
640 GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
642 /* This allows for streams with out of order timestamping - sometimes the
643 * emerging timestamp is later than the arriving one(s) */
644 if (sink_time < src_time)
645 goto beach;
647 sq->cur_time = sink_time - src_time;
648 return;
650 beach:
651 sq->cur_time = 0;
652 }
654 /* take a NEWSEGMENT event and apply the values to segment, updating the time
655 * level of queue. */
656 static void
657 apply_segment (GstMultiQueue * mq, GstSingleQueue * sq, GstEvent * event,
658 GstSegment * segment)
659 {
660 gboolean update;
661 GstFormat format;
662 gdouble rate, arate;
663 gint64 start, stop, time;
665 gst_event_parse_new_segment_full (event, &update, &rate, &arate,
666 &format, &start, &stop, &time);
668 /* now configure the values, we use these to track timestamps on the
669 * sinkpad. */
670 if (format != GST_FORMAT_TIME) {
671 /* non-time format, pretent the current time segment is closed with a
672 * 0 start and unknown stop time. */
673 update = FALSE;
674 format = GST_FORMAT_TIME;
675 start = 0;
676 stop = -1;
677 time = 0;
678 }
680 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
682 gst_segment_set_newsegment_full (segment, update,
683 rate, arate, format, start, stop, time);
685 GST_DEBUG_OBJECT (mq,
686 "queue %d, configured NEWSEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
688 /* segment can update the time level of the queue */
689 update_time_level (mq, sq);
691 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
692 }
694 /* take a buffer and update segment, updating the time level of the queue. */
695 static void
696 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
697 GstClockTime duration, GstSegment * segment)
698 {
699 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
701 /* if no timestamp is set, assume it's continuous with the previous
702 * time */
703 if (timestamp == GST_CLOCK_TIME_NONE)
704 timestamp = segment->last_stop;
706 /* add duration */
707 if (duration != GST_CLOCK_TIME_NONE)
708 timestamp += duration;
710 GST_DEBUG_OBJECT (mq, "queue %d, last_stop updated to %" GST_TIME_FORMAT,
711 sq->id, GST_TIME_ARGS (timestamp));
713 gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
715 /* calc diff with other end */
716 update_time_level (mq, sq);
717 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
718 }
720 static GstFlowReturn
721 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
722 GstMiniObject * object)
723 {
724 GstFlowReturn result = GST_FLOW_OK;
726 if (GST_IS_BUFFER (object)) {
727 GstBuffer *buffer;
728 GstClockTime timestamp, duration;
729 GstCaps *caps;
731 buffer = GST_BUFFER_CAST (object);
732 timestamp = GST_BUFFER_TIMESTAMP (buffer);
733 duration = GST_BUFFER_DURATION (buffer);
734 caps = GST_BUFFER_CAPS (buffer);
736 apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
738 /* Applying the buffer may have made the queue non-full again, unblock it if needed */
739 gst_data_queue_limits_changed (sq->queue);
741 GST_DEBUG_OBJECT (mq,
742 "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
743 sq->id, buffer, GST_TIME_ARGS (timestamp));
745 /* Set caps on pad before pushing, this avoids core calling the accpetcaps
746 * function on the srcpad, which will call acceptcaps upstream, which might
747 * not accept these caps (anymore). */
748 if (caps && caps != GST_PAD_CAPS (sq->srcpad))
749 gst_pad_set_caps (sq->srcpad, caps);
751 result = gst_pad_push (sq->srcpad, buffer);
752 } else if (GST_IS_EVENT (object)) {
753 GstEvent *event;
755 event = GST_EVENT_CAST (object);
757 switch (GST_EVENT_TYPE (event)) {
758 case GST_EVENT_EOS:
759 result = GST_FLOW_UNEXPECTED;
760 break;
761 case GST_EVENT_NEWSEGMENT:
762 apply_segment (mq, sq, event, &sq->src_segment);
763 /* Applying the segment may have made the queue non-full again, unblock it if needed */
764 gst_data_queue_limits_changed (sq->queue);
765 break;
766 default:
767 break;
768 }
770 GST_DEBUG_OBJECT (mq,
771 "SingleQueue %d : Pushing event %p of type %s",
772 sq->id, event, GST_EVENT_TYPE_NAME (event));
774 gst_pad_push_event (sq->srcpad, event);
775 } else {
776 g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
777 sq->id);
778 }
779 return result;
781 /* ERRORS */
782 }
784 static GstMiniObject *
785 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
786 {
787 GstMiniObject *res;
789 res = item->object;
790 item->object = NULL;
792 return res;
793 }
795 static void
796 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
797 {
798 if (item->object)
799 gst_mini_object_unref (item->object);
800 g_free (item);
801 }
803 /* takes ownership of passed mini object! */
804 static GstMultiQueueItem *
805 gst_multi_queue_item_new (GstMiniObject * object, guint32 curid)
806 {
807 GstMultiQueueItem *item;
809 item = g_new (GstMultiQueueItem, 1);
810 item->object = object;
811 item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
812 item->posid = curid;
814 if (GST_IS_BUFFER (object)) {
815 item->size = GST_BUFFER_SIZE (object);
816 item->duration = GST_BUFFER_DURATION (object);
817 if (item->duration == GST_CLOCK_TIME_NONE)
818 item->duration = 0;
819 item->visible = TRUE;
820 } else {
821 item->size = 0;
822 item->duration = 0;
823 item->visible = FALSE;
824 }
825 return item;
826 }
828 /* Each main loop attempts to push buffers until the return value
829 * is not-linked. not-linked pads are not allowed to push data beyond
830 * any linked pads, so they don't 'rush ahead of the pack'.
831 */
832 static void
833 gst_multi_queue_loop (GstPad * pad)
834 {
835 GstSingleQueue *sq;
836 GstMultiQueueItem *item;
837 GstDataQueueItem *sitem;
838 GstMultiQueue *mq;
839 GstMiniObject *object;
840 guint32 newid;
841 guint32 oldid = G_MAXUINT32;
842 GstFlowReturn result;
844 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
845 mq = sq->mqueue;
847 do {
848 GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
850 /* Get something from the queue, blocking until that happens, or we get
851 * flushed */
852 if (!(gst_data_queue_pop (sq->queue, &sitem)))
853 goto out_flushing;
855 item = (GstMultiQueueItem *) sitem;
856 newid = item->posid;
858 /* steal the object and destroy the item */
859 object = gst_multi_queue_item_steal_object (item);
860 gst_multi_queue_item_destroy (item);
862 GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
863 sq->id, newid, oldid);
865 /* If we're not-linked, we do some extra work because we might need to
866 * wait before pushing. If we're linked but there's a gap in the IDs,
867 * or it's the first loop, or we just passed the previous highid,
868 * we might need to wake some sleeping pad up, so there's extra work
869 * there too */
870 if (sq->srcresult == GST_FLOW_NOT_LINKED ||
871 (oldid == G_MAXUINT32) || (newid != (oldid + 1)) ||
872 oldid > mq->highid) {
873 GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
874 gst_flow_get_name (sq->srcresult));
876 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
878 /* Update the nextid so other threads know when to wake us up */
879 sq->nextid = newid;
881 /* Update the oldid (the last ID we output) for highid tracking */
882 if (oldid != G_MAXUINT32)
883 sq->oldid = oldid;
885 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
886 /* Go to sleep until it's time to push this buffer */
888 /* Recompute the highid */
889 compute_high_id (mq);
890 while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
891 GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
892 "newid %u and highid %u", sq->id, newid, mq->highid);
895 /* Wake up all non-linked pads before we sleep */
896 wake_up_next_non_linked (mq);
898 mq->numwaiting++;
899 g_cond_wait (sq->turn, mq->qlock);
900 mq->numwaiting--;
902 GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
903 "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
904 }
906 /* Re-compute the high_id in case someone else pushed */
907 compute_high_id (mq);
908 } else {
909 compute_high_id (mq);
910 /* Wake up all non-linked pads */
911 wake_up_next_non_linked (mq);
912 }
913 /* We're done waiting, we can clear the nextid */
914 sq->nextid = 0;
916 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
917 }
919 GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
920 gst_flow_get_name (sq->srcresult));
922 /* Try to push out the new object */
923 result = gst_single_queue_push_one (mq, sq, object);
924 sq->srcresult = result;
926 if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED)
927 goto out_flushing;
929 GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
930 gst_flow_get_name (sq->srcresult));
932 oldid = newid;
933 }
934 while (TRUE);
936 out_flushing:
937 {
938 /* Need to make sure wake up any sleeping pads when we exit */
939 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
940 compute_high_id (mq);
941 wake_up_next_non_linked (mq);
942 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
944 gst_data_queue_set_flushing (sq->queue, TRUE);
945 gst_pad_pause_task (sq->srcpad);
946 GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
947 "SingleQueue[%d] task paused, reason:%s",
948 sq->id, gst_flow_get_name (sq->srcresult));
949 return;
950 }
951 }
953 /**
954 * gst_multi_queue_chain:
955 *
956 * This is similar to GstQueue's chain function, except:
957 * _ we don't have leak behavioures,
958 * _ we push with a unique id (curid)
959 */
960 static GstFlowReturn
961 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
962 {
963 GstSingleQueue *sq;
964 GstMultiQueue *mq;
965 GstMultiQueueItem *item;
966 GstFlowReturn ret = GST_FLOW_OK;
967 guint32 curid;
968 GstClockTime timestamp, duration;
970 sq = gst_pad_get_element_private (pad);
971 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
973 /* Get a unique incrementing id */
974 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
975 curid = mq->counter++;
976 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
978 GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
979 sq->id, buffer, curid);
981 item = gst_multi_queue_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
983 timestamp = GST_BUFFER_TIMESTAMP (buffer);
984 duration = GST_BUFFER_DURATION (buffer);
986 if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
987 goto flushing;
989 /* update time level, we must do this after pushing the data in the queue so
990 * that we never end up filling the queue first. */
991 apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
993 done:
994 gst_object_unref (mq);
996 return ret;
998 /* ERRORS */
999 flushing:
1000 {
1001 ret = sq->srcresult;
1002 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1003 sq->id, gst_flow_get_name (ret));
1004 gst_multi_queue_item_destroy (item);
1005 goto done;
1006 }
1007 }
1009 static gboolean
1010 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
1011 {
1012 GstSingleQueue *sq;
1014 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1016 if (active) {
1017 /* All pads start off linked until they push one buffer */
1018 sq->srcresult = GST_FLOW_OK;
1019 } else {
1020 sq->srcresult = GST_FLOW_WRONG_STATE;
1021 gst_data_queue_flush (sq->queue);
1022 }
1023 return TRUE;
1024 }
1026 static gboolean
1027 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
1028 {
1029 GstSingleQueue *sq;
1030 GstMultiQueue *mq;
1031 guint32 curid;
1032 GstMultiQueueItem *item;
1033 gboolean res;
1034 GstEventType type;
1035 GstEvent *sref = NULL;
1037 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1038 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
1040 type = GST_EVENT_TYPE (event);
1042 switch (type) {
1043 case GST_EVENT_FLUSH_START:
1044 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
1045 sq->id);
1047 res = gst_pad_push_event (sq->srcpad, event);
1049 gst_single_queue_flush (mq, sq, TRUE);
1050 goto done;
1052 case GST_EVENT_FLUSH_STOP:
1053 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
1054 sq->id);
1056 res = gst_pad_push_event (sq->srcpad, event);
1058 gst_single_queue_flush (mq, sq, FALSE);
1059 goto done;
1060 case GST_EVENT_NEWSEGMENT:
1061 /* take ref because the queue will take ownership and we need the event
1062 * afterwards to update the segment */
1063 sref = gst_event_ref (event);
1064 break;
1066 default:
1067 if (!(GST_EVENT_IS_SERIALIZED (event))) {
1068 res = gst_pad_push_event (sq->srcpad, event);
1069 goto done;
1070 }
1071 break;
1072 }
1074 /* Get an unique incrementing id */
1075 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1076 curid = mq->counter++;
1077 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1079 item = gst_multi_queue_item_new ((GstMiniObject *) event, curid);
1081 GST_DEBUG_OBJECT (mq,
1082 "SingleQueue %d : Enqueuing event %p of type %s with id %d",
1083 sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
1085 if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1086 goto flushing;
1088 /* mark EOS when we received one, we must do that after putting the
1089 * buffer in the queue because EOS marks the buffer as filled. No need to take
1090 * a lock, the _check_full happens from this thread only, right before pushing
1091 * into dataqueue. */
1092 switch (type) {
1093 case GST_EVENT_EOS:
1094 sq->is_eos = TRUE;
1095 break;
1096 case GST_EVENT_NEWSEGMENT:
1097 apply_segment (mq, sq, sref, &sq->sink_segment);
1098 gst_event_unref (sref);
1099 break;
1100 default:
1101 break;
1102 }
1103 done:
1104 gst_object_unref (mq);
1105 return res;
1107 flushing:
1108 {
1109 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1110 sq->id, gst_flow_get_name (sq->srcresult));
1111 if (sref)
1112 gst_event_unref (sref);
1113 gst_multi_queue_item_destroy (item);
1114 goto done;
1115 }
1116 }
1118 static GstCaps *
1119 gst_multi_queue_getcaps (GstPad * pad)
1120 {
1121 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1122 GstPad *otherpad;
1123 GstCaps *result;
1125 otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1127 GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
1129 result = gst_pad_peer_get_caps (otherpad);
1130 if (result == NULL)
1131 result = gst_caps_new_any ();
1133 return result;
1134 }
1136 static GstFlowReturn
1137 gst_multi_queue_bufferalloc (GstPad * pad, guint64 offset, guint size,
1138 GstCaps * caps, GstBuffer ** buf)
1139 {
1140 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1142 return gst_pad_alloc_buffer (sq->srcpad, offset, size, caps, buf);
1143 }
1145 static gboolean
1146 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1147 {
1148 GstMultiQueue *mq;
1149 GstSingleQueue *sq;
1150 gboolean result = FALSE;
1152 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1153 mq = sq->mqueue;
1155 GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1157 if (active) {
1158 result = gst_single_queue_flush (mq, sq, FALSE);
1159 } else {
1160 result = gst_single_queue_flush (mq, sq, TRUE);
1161 /* make sure streaming finishes */
1162 result |= gst_pad_stop_task (pad);
1163 }
1164 return result;
1165 }
1167 static gboolean
1168 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1169 {
1170 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1172 return gst_pad_push_event (sq->sinkpad, event);
1173 }
1175 static gboolean
1176 gst_multi_queue_src_query (GstPad * pad, GstQuery * query)
1177 {
1178 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1179 GstPad *peerpad;
1180 gboolean res;
1182 /* FIXME, Handle position offset depending on queue size */
1184 /* default handling */
1185 if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1186 goto no_peer;
1188 res = gst_pad_query (peerpad, query);
1190 gst_object_unref (peerpad);
1192 return res;
1194 /* ERRORS */
1195 no_peer:
1196 {
1197 GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1198 return FALSE;
1199 }
1200 }
1202 /*
1203 * Next-non-linked functions
1204 */
1206 /* WITH LOCK TAKEN */
1207 static void
1208 wake_up_next_non_linked (GstMultiQueue * mq)
1209 {
1210 GList *tmp;
1212 /* maybe no-one is waiting */
1213 if (mq->numwaiting < 1)
1214 return;
1216 /* Else figure out which singlequeue(s) need waking up */
1217 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1218 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1220 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1221 if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1222 GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1223 g_cond_signal (sq->turn);
1224 }
1225 }
1226 }
1227 }
1229 /* WITH LOCK TAKEN */
1230 static void
1231 compute_high_id (GstMultiQueue * mq)
1232 {
1233 /* The high-id is either the highest id among the linked pads, or if all
1234 * pads are not-linked, it's the lowest not-linked pad */
1235 GList *tmp;
1236 guint32 lowest = G_MAXUINT32;
1237 guint32 highid = G_MAXUINT32;
1239 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1240 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1242 GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1243 sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1245 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1246 /* No need to consider queues which are not waiting */
1247 if (sq->nextid == 0) {
1248 GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1249 continue;
1250 }
1252 if (sq->nextid < lowest)
1253 lowest = sq->nextid;
1254 } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1255 /* If we don't have a global highid, or the global highid is lower than
1256 * this single queue's last outputted id, store the queue's one,
1257 * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1258 if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1259 highid = sq->oldid;
1260 }
1261 }
1263 if (highid == G_MAXUINT32 || lowest < highid)
1264 mq->highid = lowest;
1265 else
1266 mq->highid = highid;
1268 GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1269 lowest);
1270 }
1272 #define IS_FILLED(format, value) ((sq->max_size.format) != 0 && \
1273 (sq->max_size.format) <= (value))
1275 static void
1276 single_queue_overrun_cb_unlocked (GstDataQueue * dq, GstSingleQueue * sq)
1277 {
1278 GstMultiQueue *mq = sq->mqueue;
1279 GList *tmp;
1280 GstDataQueueSize size;
1282 gst_data_queue_get_level (sq->queue, &size);
1283 mq->filled = FALSE;
1285 /* if we have reached max visible we can maybe bump this
1286 * if another queue is empty, skip this if we can't grow anymore
1287 */
1288 if (IS_FILLED (visible, size.visible) && sq->extra_size.visible > 0) {
1289 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1290 GstSingleQueue *ssq = (GstSingleQueue *) tmp->data;
1292 if (ssq == sq)
1293 continue;
1295 if (gst_data_queue_is_empty (ssq->queue)) {
1296 sq->max_size.visible++;
1297 sq->extra_size.visible--;
1298 GST_DEBUG_OBJECT (mq,
1299 "queue %d is empty, bumping single queue %d max visible to %d",
1300 ssq->id, sq->id, sq->max_size.visible);
1301 break;
1302 }
1303 }
1304 /* check if the queue is still full */
1305 mq->filled = gst_data_queue_is_full (sq->queue);
1306 }
1307 }
1309 /*
1310 * GstSingleQueue functions
1311 */
1312 static void
1313 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1314 {
1315 GstMultiQueue *mq = sq->mqueue;
1317 GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1319 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1321 single_queue_overrun_cb_unlocked (dq, sq);
1323 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1325 if (mq->filled) {
1326 GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1327 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1328 mq->filled = FALSE;
1329 }
1330 }
1332 static void
1333 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1334 {
1335 gboolean all_empty = TRUE;
1336 GstMultiQueue *mq = sq->mqueue;
1337 GList *tmp;
1339 GST_LOG_OBJECT (mq,
1340 "Single Queue %d is empty, Checking other single queues", sq->id);
1342 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1343 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1344 GstSingleQueue *ssq = (GstSingleQueue *) tmp->data;
1346 if (sq == ssq)
1347 continue;
1349 /* prevent data starvation */
1350 if (gst_data_queue_is_full (ssq->queue)) {
1351 single_queue_overrun_cb_unlocked (dq, ssq);
1352 goto check_filled;
1353 }
1355 if (!gst_data_queue_is_empty (ssq->queue)) {
1356 all_empty = FALSE;
1357 break;
1358 }
1359 }
1360 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1362 if (all_empty) {
1363 GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1364 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1365 }
1366 return;
1368 check_filled:
1369 if (mq->filled) {
1370 GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1371 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1372 mq->filled = FALSE;
1373 }
1374 }
1376 static gboolean
1377 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1378 guint64 time, GstSingleQueue * sq)
1379 {
1380 gboolean res;
1382 GST_DEBUG ("queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT
1383 "/%" G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1384 sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1386 /* we are always filled on EOS */
1387 if (sq->is_eos)
1388 return TRUE;
1390 /* we never go past the max visible items */
1391 if (IS_FILLED (visible, visible))
1392 return TRUE;
1394 if (sq->cur_time != 0) {
1395 /* if we have valid time in the queue, check */
1396 res = IS_FILLED (time, sq->cur_time);
1397 } else {
1398 /* no valid time, check bytes */
1399 res = IS_FILLED (bytes, bytes);
1400 }
1401 return res;
1402 }
1404 static void
1405 gst_single_queue_free (GstSingleQueue * sq)
1406 {
1407 /* DRAIN QUEUE */
1408 gst_data_queue_flush (sq->queue);
1409 g_object_unref (sq->queue);
1410 g_cond_free (sq->turn);
1411 g_free (sq);
1412 }
1414 static GstSingleQueue *
1415 gst_single_queue_new (GstMultiQueue * mqueue)
1416 {
1417 GstSingleQueue *sq;
1418 gchar *tmp;
1420 sq = g_new0 (GstSingleQueue, 1);
1422 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1423 sq->id = mqueue->nbqueues++;
1425 /* copy over max_size and extra_size so we don't need to take the lock
1426 * any longer when checking if the queue is full. */
1427 sq->max_size.visible = mqueue->max_size.visible;
1428 sq->max_size.bytes = mqueue->max_size.bytes;
1429 sq->max_size.time = mqueue->max_size.time;
1431 sq->extra_size.visible = mqueue->extra_size.visible;
1432 sq->extra_size.bytes = mqueue->extra_size.bytes;
1433 sq->extra_size.time = mqueue->extra_size.time;
1435 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1437 GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1439 sq->mqueue = mqueue;
1440 sq->srcresult = GST_FLOW_WRONG_STATE;
1441 sq->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction)
1442 single_queue_check_full, sq);
1443 sq->is_eos = FALSE;
1444 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1445 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1447 sq->nextid = 0;
1448 sq->oldid = 0;
1449 sq->turn = g_cond_new ();
1451 /* attach to underrun/overrun signals to handle non-starvation */
1452 g_signal_connect (G_OBJECT (sq->queue), "full",
1453 G_CALLBACK (single_queue_overrun_cb), sq);
1454 g_signal_connect (G_OBJECT (sq->queue), "empty",
1455 G_CALLBACK (single_queue_underrun_cb), sq);
1457 tmp = g_strdup_printf ("sink%d", sq->id);
1458 sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
1459 g_free (tmp);
1461 gst_pad_set_chain_function (sq->sinkpad,
1462 GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1463 gst_pad_set_activatepush_function (sq->sinkpad,
1464 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1465 gst_pad_set_event_function (sq->sinkpad,
1466 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1467 gst_pad_set_getcaps_function (sq->sinkpad,
1468 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1469 gst_pad_set_bufferalloc_function (sq->sinkpad,
1470 GST_DEBUG_FUNCPTR (gst_multi_queue_bufferalloc));
1471 gst_pad_set_internal_link_function (sq->sinkpad,
1472 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1474 tmp = g_strdup_printf ("src%d", sq->id);
1475 sq->srcpad = gst_pad_new_from_static_template (&srctemplate, tmp);
1476 g_free (tmp);
1478 gst_pad_set_activatepush_function (sq->srcpad,
1479 GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1480 gst_pad_set_getcaps_function (sq->srcpad,
1481 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1482 gst_pad_set_event_function (sq->srcpad,
1483 GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1484 gst_pad_set_query_function (sq->srcpad,
1485 GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1486 gst_pad_set_internal_link_function (sq->srcpad,
1487 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1489 gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1490 gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1492 /* only activate the pads when we are not in the NULL state
1493 * and add the pad under the state_lock to prevend state changes
1494 * between activating and adding */
1495 g_static_rec_mutex_lock (GST_STATE_GET_LOCK (mqueue));
1496 if (GST_STATE_TARGET (mqueue) != GST_STATE_NULL) {
1497 gst_pad_set_active (sq->srcpad, TRUE);
1498 gst_pad_set_active (sq->sinkpad, TRUE);
1499 }
1500 gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1501 gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1502 g_static_rec_mutex_unlock (GST_STATE_GET_LOCK (mqueue));
1504 GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1505 sq->id);
1507 return sq;
1508 }