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 * @see_also: #GstQueue
27 *
28 * <refsect2>
29 * <para>
30 * Multiqueue is similar to a normal #GstQueue with the following additional
31 * features:
32 * <orderedlist>
33 * <listitem>
34 * <itemizedlist><title>Multiple streamhandling</title>
35 * <listitem><para>
36 * The element handles queueing data on more than one stream at once. To
37 * achieve such a feature it has request sink pads (sink%d) and
38 * 'sometimes' src pads (src%d).
39 * </para><para>
40 * When requesting a given sinkpad with gst_element_get_request_pad(),
41 * the associated srcpad for that stream will be created.
42 * Example: requesting sink1 will generate src1.
43 * </para></listitem>
44 * </itemizedlist>
45 * </listitem>
46 * <listitem>
47 * <itemizedlist><title>Non-starvation on multiple streams</title>
48 * <listitem><para>
49 * If more than one stream is used with the element, the streams' queues
50 * will be dynamically grown (up to a limit), in order to ensure that no
51 * stream is risking data starvation. This guarantees that at any given
52 * time there are at least N bytes queued and available for each individual
53 * stream.
54 * </para><para>
55 * If an EOS event comes through a srcpad, the associated queue will be
56 * considered as 'not-empty' in the queue-size-growing algorithm.
57 * </para></listitem>
58 * </itemizedlist>
59 * </listitem>
60 * <listitem>
61 * <itemizedlist><title>Non-linked srcpads graceful handling</title>
62 * <listitem><para>
63 * In order to better support dynamic switching between streams, the multiqueue
64 * (unlike the current GStreamer queue) continues to push buffers on non-linked
65 * pads rather than shutting down.
66 * </para><para>
67 * In addition, to prevent a non-linked stream from very quickly consuming all
68 * available buffers and thus 'racing ahead' of the other streams, the element
69 * must ensure that buffers and inlined events for a non-linked stream are pushed
70 * in the same order as they were received, relative to the other streams
71 * controlled by the element. This means that a buffer cannot be pushed to a
72 * non-linked pad any sooner than buffers in any other stream which were received
73 * before it.
74 * </para></listitem>
75 * </itemizedlist>
76 * </listitem>
77 * </orderedlist>
78 * </para>
79 * <para>
80 * Data is queued until one of the limits specified by the
81 * #GstMultiQueue:max-size-buffers, #GstMultiQueue:max-size-bytes and/or
82 * #GstMultiQueue:max-size-time properties has been reached. Any attempt to push
83 * more buffers into the queue will block the pushing thread until more space
84 * becomes available. #GstMultiQueue:extra-size-buffers,
85 * </para>
86 * <para>
87 * #GstMultiQueue:extra-size-bytes and #GstMultiQueue:extra-size-time are
88 * currently unused.
89 * </para>
90 * <para>
91 * The default queue size limits are 5 buffers, 10MB of data, or
92 * two second worth of data, whichever is reached first. Note that the number
93 * of buffers will dynamically grow depending on the fill level of
94 * other queues.
95 * </para>
96 * <para>
97 * The #GstMultiQueue::underrun signal is emitted when all of the queues
98 * are empty. The #GstMultiQueue::overrun signal is emitted when one of the
99 * queues is filled.
100 * Both signals are emitted from the context of the streaming thread.
101 * </para>
102 * </refsect2>
103 *
104 * Last reviewed on 2008-01-25 (0.10.17)
105 */
107 #ifdef HAVE_CONFIG_H
108 # include "config.h"
109 #endif
111 #include <gst/gst.h>
112 #include "gstmultiqueue.h"
114 /**
115 * GstSingleQueue:
116 * @sinkpad: associated sink #GstPad
117 * @srcpad: associated source #GstPad
118 *
119 * Structure containing all information and properties about
120 * a single queue.
121 */
122 typedef struct _GstSingleQueue GstSingleQueue;
124 struct _GstSingleQueue
125 {
126 /* unique identifier of the queue */
127 guint id;
129 GstMultiQueue *mqueue;
131 GstPad *sinkpad;
132 GstPad *srcpad;
134 /* flowreturn of previous srcpad push */
135 GstFlowReturn srcresult;
136 GstSegment sink_segment;
137 GstSegment src_segment;
139 /* queue of data */
140 GstDataQueue *queue;
141 GstDataQueueSize max_size, extra_size;
142 GstClockTime cur_time;
143 gboolean is_eos;
144 gboolean inextra; /* TRUE if the queue is currently in extradata mode */
146 /* Protected by global lock */
147 guint32 nextid; /* ID of the next object waiting to be pushed */
148 guint32 oldid; /* ID of the last object pushed (last in a series) */
149 GCond *turn; /* SingleQueue turn waiting conditional */
150 };
153 /* Extension of GstDataQueueItem structure for our usage */
154 typedef struct _GstMultiQueueItem GstMultiQueueItem;
156 struct _GstMultiQueueItem
157 {
158 GstMiniObject *object;
159 guint size;
160 guint64 duration;
161 gboolean visible;
163 GDestroyNotify destroy;
164 guint32 posid;
165 };
167 static GstSingleQueue *gst_single_queue_new (GstMultiQueue * mqueue);
168 static void gst_single_queue_free (GstSingleQueue * squeue);
170 static void wake_up_next_non_linked (GstMultiQueue * mq);
171 static void compute_high_id (GstMultiQueue * mq);
173 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",
174 GST_PAD_SINK,
175 GST_PAD_REQUEST,
176 GST_STATIC_CAPS_ANY);
178 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
179 GST_PAD_SRC,
180 GST_PAD_SOMETIMES,
181 GST_STATIC_CAPS_ANY);
183 GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);
184 #define GST_CAT_DEFAULT (multi_queue_debug)
186 /* default limits, we try to keep up to 2 seconds of data and if there is not
187 * time, up to 10 MB. The number of buffers is dynamically scaled to make sure
188 * there is data in the queues. Normally, the byte and time limits are not hit
189 * in theses conditions. */
190 #define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
191 #define DEFAULT_MAX_SIZE_BUFFERS 5
192 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND
194 /* second limits. When we hit one of the above limits we are probably dealing
195 * with a badly muxed file and we scale the limits to these emergency values.
196 * This is currently not yet implemented. */
197 #define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
198 #define DEFAULT_EXTRA_SIZE_BUFFERS 5
199 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND
201 /* Signals and args */
202 enum
203 {
204 SIGNAL_UNDERRUN,
205 SIGNAL_OVERRUN,
206 LAST_SIGNAL
207 };
209 enum
210 {
211 ARG_0,
212 ARG_EXTRA_SIZE_BYTES,
213 ARG_EXTRA_SIZE_BUFFERS,
214 ARG_EXTRA_SIZE_TIME,
215 ARG_MAX_SIZE_BYTES,
216 ARG_MAX_SIZE_BUFFERS,
217 ARG_MAX_SIZE_TIME,
218 };
220 #define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
221 g_mutex_lock (q->qlock); \
222 } G_STMT_END
224 #define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
225 g_mutex_unlock (q->qlock); \
226 } G_STMT_END
228 static void gst_multi_queue_finalize (GObject * object);
229 static void gst_multi_queue_set_property (GObject * object,
230 guint prop_id, const GValue * value, GParamSpec * pspec);
231 static void gst_multi_queue_get_property (GObject * object,
232 guint prop_id, GValue * value, GParamSpec * pspec);
234 static GstPad *gst_multi_queue_request_new_pad (GstElement * element,
235 GstPadTemplate * temp, const gchar * name);
236 static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);
238 static void gst_multi_queue_loop (GstPad * pad);
240 #define _do_init(bla) \
241 GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");
243 GST_BOILERPLATE_FULL (GstMultiQueue, gst_multi_queue, GstElement,
244 GST_TYPE_ELEMENT, _do_init);
246 static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };
248 static void
249 gst_multi_queue_base_init (gpointer g_class)
250 {
251 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
253 gst_element_class_set_details_simple (gstelement_class,
254 "MultiQueue",
255 "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");
256 gst_element_class_add_pad_template (gstelement_class,
257 gst_static_pad_template_get (&sinktemplate));
258 gst_element_class_add_pad_template (gstelement_class,
259 gst_static_pad_template_get (&srctemplate));
260 }
262 static void
263 gst_multi_queue_class_init (GstMultiQueueClass * klass)
264 {
265 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
266 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
268 gobject_class->set_property =
269 GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);
270 gobject_class->get_property =
271 GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);
273 /* SIGNALS */
275 /**
276 * GstMultiQueue::underrun:
277 * @multiqueue: the multqueue instance
278 *
279 * This signal is emitted from the streaming thread when there is
280 * no data in any of the queues inside the multiqueue instance (underrun).
281 *
282 * This indicates either starvation or EOS from the upstream data sources.
283 */
284 gst_multi_queue_signals[SIGNAL_UNDERRUN] =
285 g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
286 G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,
287 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
289 /**
290 * GstMultiQueue::overrun:
291 * @multiqueue: the multiqueue instance
292 *
293 * Reports that one of the queues in the multiqueue is full (overrun).
294 * A queue is full if the total amount of data inside it (num-buffers, time,
295 * size) is higher than the boundary values which can be set through the
296 * GObject properties.
297 *
298 * This can be used as an indicator of pre-roll.
299 */
300 gst_multi_queue_signals[SIGNAL_OVERRUN] =
301 g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
302 G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,
303 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
305 /* PROPERTIES */
307 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
308 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
309 "Max. amount of data in the queue (bytes, 0=disable)",
310 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
311 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
312 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
313 g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
314 "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
315 DEFAULT_MAX_SIZE_BUFFERS,
316 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
317 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
318 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
319 "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
320 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
322 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,
323 g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",
324 "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",
325 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES,
326 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
327 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,
328 g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",
329 "Amount of buffers the queues can grow if one of them is empty (0=disable)",
330 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS,
331 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
332 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,
333 g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",
334 "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",
335 0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME,
336 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
338 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_multi_queue_finalize);
340 gstelement_class->request_new_pad =
341 GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);
342 gstelement_class->release_pad =
343 GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);
344 }
346 static void
347 gst_multi_queue_init (GstMultiQueue * mqueue, GstMultiQueueClass * klass)
348 {
349 mqueue->nbqueues = 0;
350 mqueue->queues = NULL;
352 mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;
353 mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;
354 mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;
356 mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;
357 mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;
358 mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
360 mqueue->counter = 1;
361 mqueue->highid = -1;
362 mqueue->nextnotlinked = -1;
364 mqueue->qlock = g_mutex_new ();
365 }
367 static void
368 gst_multi_queue_finalize (GObject * object)
369 {
370 GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);
372 g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);
373 g_list_free (mqueue->queues);
374 mqueue->queues = NULL;
376 /* free/unref instance data */
377 g_mutex_free (mqueue->qlock);
379 G_OBJECT_CLASS (parent_class)->finalize (object);
380 }
382 #define SET_CHILD_PROPERTY(mq,format) G_STMT_START { \
383 GList * tmp = mq->queues; \
384 while (tmp) { \
385 GstSingleQueue *q = (GstSingleQueue*)tmp->data; \
386 q->max_size.format = mq->max_size.format; \
387 tmp = g_list_next(tmp); \
388 }; \
389 } G_STMT_END
391 static void
392 gst_multi_queue_set_property (GObject * object, guint prop_id,
393 const GValue * value, GParamSpec * pspec)
394 {
395 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
397 switch (prop_id) {
398 case ARG_MAX_SIZE_BYTES:
399 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
400 mq->max_size.bytes = g_value_get_uint (value);
401 SET_CHILD_PROPERTY (mq, bytes);
402 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
403 break;
404 case ARG_MAX_SIZE_BUFFERS:
405 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
406 mq->max_size.visible = g_value_get_uint (value);
407 SET_CHILD_PROPERTY (mq, visible);
408 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
409 break;
410 case ARG_MAX_SIZE_TIME:
411 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
412 mq->max_size.time = g_value_get_uint64 (value);
413 SET_CHILD_PROPERTY (mq, time);
414 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
415 break;
416 case ARG_EXTRA_SIZE_BYTES:
417 mq->extra_size.bytes = g_value_get_uint (value);
418 break;
419 case ARG_EXTRA_SIZE_BUFFERS:
420 mq->extra_size.visible = g_value_get_uint (value);
421 break;
422 case ARG_EXTRA_SIZE_TIME:
423 mq->extra_size.time = g_value_get_uint64 (value);
424 break;
425 default:
426 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
427 break;
428 }
429 }
431 static void
432 gst_multi_queue_get_property (GObject * object, guint prop_id,
433 GValue * value, GParamSpec * pspec)
434 {
435 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
437 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
439 switch (prop_id) {
440 case ARG_EXTRA_SIZE_BYTES:
441 g_value_set_uint (value, mq->extra_size.bytes);
442 break;
443 case ARG_EXTRA_SIZE_BUFFERS:
444 g_value_set_uint (value, mq->extra_size.visible);
445 break;
446 case ARG_EXTRA_SIZE_TIME:
447 g_value_set_uint64 (value, mq->extra_size.time);
448 break;
449 case ARG_MAX_SIZE_BYTES:
450 g_value_set_uint (value, mq->max_size.bytes);
451 break;
452 case ARG_MAX_SIZE_BUFFERS:
453 g_value_set_uint (value, mq->max_size.visible);
454 break;
455 case ARG_MAX_SIZE_TIME:
456 g_value_set_uint64 (value, mq->max_size.time);
457 break;
458 default:
459 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
460 break;
461 }
463 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
464 }
466 static GList *
467 gst_multi_queue_get_internal_links (GstPad * pad)
468 {
469 GList *res = NULL;
470 GstMultiQueue *mqueue;
471 GstSingleQueue *sq = NULL;
472 GList *tmp;
474 g_return_val_if_fail (GST_IS_PAD (pad), NULL);
476 mqueue = GST_MULTI_QUEUE (GST_PAD_PARENT (pad));
477 if (!mqueue)
478 goto no_parent;
480 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
481 /* Find which single queue it belongs to */
482 for (tmp = mqueue->queues; tmp && !res; tmp = g_list_next (tmp)) {
483 sq = (GstSingleQueue *) tmp->data;
485 if (sq->sinkpad == pad)
486 res = g_list_prepend (res, sq->srcpad);
487 if (sq->srcpad == pad)
488 res = g_list_prepend (res, sq->sinkpad);
489 }
491 if (!res)
492 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
493 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
495 return res;
497 no_parent:
498 {
499 GST_DEBUG_OBJECT (pad, "no parent");
500 return NULL;
501 }
502 }
505 /*
506 * GstElement methods
507 */
509 static GstPad *
510 gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,
511 const gchar * name)
512 {
513 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
514 GstSingleQueue *squeue;
516 GST_LOG_OBJECT (element, "name : %s", GST_STR_NULL (name));
518 /* Create a new single queue, add the sink and source pad and return the sink pad */
519 squeue = gst_single_queue_new (mqueue);
521 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
522 mqueue->queues = g_list_append (mqueue->queues, squeue);
523 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
525 GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",
526 GST_DEBUG_PAD_NAME (squeue->sinkpad));
528 return squeue->sinkpad;
529 }
531 static void
532 gst_multi_queue_release_pad (GstElement * element, GstPad * pad)
533 {
534 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
535 GstSingleQueue *sq = NULL;
536 GList *tmp;
538 GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));
540 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
541 /* Find which single queue it belongs to, knowing that it should be a sinkpad */
542 for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
543 sq = (GstSingleQueue *) tmp->data;
545 if (sq->sinkpad == pad)
546 break;
547 }
549 if (!tmp) {
550 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
551 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
552 return;
553 }
555 /* FIXME: The removal of the singlequeue should probably not happen until it
556 * finishes draining */
558 /* remove it from the list */
559 mqueue->queues = g_list_delete_link (mqueue->queues, tmp);
561 /* FIXME : recompute next-non-linked */
562 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
564 /* delete SingleQueue */
565 gst_data_queue_set_flushing (sq->queue, TRUE);
567 gst_pad_set_active (sq->srcpad, FALSE);
568 gst_pad_set_active (sq->sinkpad, FALSE);
569 gst_element_remove_pad (element, sq->srcpad);
570 gst_element_remove_pad (element, sq->sinkpad);
571 gst_single_queue_free (sq);
572 }
574 static gboolean
575 gst_single_queue_flush (GstMultiQueue * mq, GstSingleQueue * sq, gboolean flush)
576 {
577 gboolean result;
579 GST_DEBUG_OBJECT (mq, "flush %s queue %d", (flush ? "start" : "stop"),
580 sq->id);
582 if (flush) {
583 sq->srcresult = GST_FLOW_WRONG_STATE;
584 gst_data_queue_set_flushing (sq->queue, TRUE);
586 /* wake up non-linked task */
587 GST_LOG_OBJECT (mq, "SingleQueue %d : waking up eventually waiting task",
588 sq->id);
589 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
590 g_cond_signal (sq->turn);
591 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
593 GST_LOG_OBJECT (mq, "SingleQueue %d : pausing task", sq->id);
594 result = gst_pad_pause_task (sq->srcpad);
595 } else {
596 gst_data_queue_flush (sq->queue);
597 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
598 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
599 /* All pads start off not-linked for a smooth kick-off */
600 sq->srcresult = GST_FLOW_OK;
601 sq->cur_time = 0;
602 sq->max_size.visible = mq->max_size.visible;
603 sq->is_eos = FALSE;
604 sq->inextra = FALSE;
605 sq->nextid = 0;
606 sq->oldid = 0;
607 gst_data_queue_set_flushing (sq->queue, FALSE);
609 GST_LOG_OBJECT (mq, "SingleQueue %d : starting task", sq->id);
610 result =
611 gst_pad_start_task (sq->srcpad, (GstTaskFunction) gst_multi_queue_loop,
612 sq->srcpad);
613 }
614 return result;
615 }
617 /* calculate the diff between running time on the sink and src of the queue.
618 * This is the total amount of time in the queue.
619 * WITH LOCK TAKEN */
620 static void
621 update_time_level (GstMultiQueue * mq, GstSingleQueue * sq)
622 {
623 gint64 sink_time, src_time;
625 sink_time =
626 gst_segment_to_running_time (&sq->sink_segment, GST_FORMAT_TIME,
627 sq->sink_segment.last_stop);
629 src_time = gst_segment_to_running_time (&sq->src_segment, GST_FORMAT_TIME,
630 sq->src_segment.last_stop);
632 GST_DEBUG_OBJECT (mq,
633 "queue %d, sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT, sq->id,
634 GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
636 /* This allows for streams with out of order timestamping - sometimes the
637 * emerging timestamp is later than the arriving one(s) */
638 if (sink_time >= src_time)
639 sq->cur_time = sink_time - src_time;
640 else
641 sq->cur_time = 0;
642 }
644 /* take a NEWSEGMENT event and apply the values to segment, updating the time
645 * level of queue. */
646 static void
647 apply_segment (GstMultiQueue * mq, GstSingleQueue * sq, GstEvent * event,
648 GstSegment * segment)
649 {
650 gboolean update;
651 GstFormat format;
652 gdouble rate, arate;
653 gint64 start, stop, time;
655 gst_event_parse_new_segment_full (event, &update, &rate, &arate,
656 &format, &start, &stop, &time);
658 /* now configure the values, we use these to track timestamps on the
659 * sinkpad. */
660 if (format != GST_FORMAT_TIME) {
661 /* non-time format, pretent the current time segment is closed with a
662 * 0 start and unknown stop time. */
663 update = FALSE;
664 format = GST_FORMAT_TIME;
665 start = 0;
666 stop = -1;
667 time = 0;
668 }
670 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
672 gst_segment_set_newsegment_full (segment, update,
673 rate, arate, format, start, stop, time);
675 GST_DEBUG_OBJECT (mq,
676 "queue %d, configured NEWSEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
678 /* segment can update the time level of the queue */
679 update_time_level (mq, sq);
681 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
682 }
684 /* take a buffer and update segment, updating the time level of the queue. */
685 static void
686 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
687 GstClockTime duration, GstSegment * segment)
688 {
689 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
691 /* if no timestamp is set, assume it's continuous with the previous
692 * time */
693 if (timestamp == GST_CLOCK_TIME_NONE)
694 timestamp = segment->last_stop;
696 /* add duration */
697 if (duration != GST_CLOCK_TIME_NONE)
698 timestamp += duration;
700 GST_DEBUG_OBJECT (mq, "queue %d, last_stop updated to %" GST_TIME_FORMAT,
701 sq->id, GST_TIME_ARGS (timestamp));
703 gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
705 /* calc diff with other end */
706 update_time_level (mq, sq);
707 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
708 }
710 static GstFlowReturn
711 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
712 GstMiniObject * object)
713 {
714 GstFlowReturn result = GST_FLOW_OK;
716 if (GST_IS_BUFFER (object)) {
717 GstBuffer *buffer;
718 GstClockTime timestamp, duration;
719 GstCaps *caps;
721 buffer = GST_BUFFER_CAST (object);
722 timestamp = GST_BUFFER_TIMESTAMP (buffer);
723 duration = GST_BUFFER_DURATION (buffer);
724 caps = GST_BUFFER_CAPS (buffer);
726 apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
728 /* Applying the buffer may have made the queue non-full again, unblock it if needed */
729 gst_data_queue_limits_changed (sq->queue);
731 GST_DEBUG_OBJECT (mq,
732 "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
733 sq->id, buffer, GST_TIME_ARGS (timestamp));
735 /* Set caps on pad before pushing, this avoids core calling the accpetcaps
736 * function on the srcpad, which will call acceptcaps upstream, which might
737 * not accept these caps (anymore). */
738 if (caps && caps != GST_PAD_CAPS (sq->srcpad))
739 gst_pad_set_caps (sq->srcpad, caps);
741 result = gst_pad_push (sq->srcpad, buffer);
742 } else if (GST_IS_EVENT (object)) {
743 GstEvent *event;
745 event = GST_EVENT_CAST (object);
747 switch (GST_EVENT_TYPE (event)) {
748 case GST_EVENT_EOS:
749 result = GST_FLOW_UNEXPECTED;
750 break;
751 case GST_EVENT_NEWSEGMENT:
752 apply_segment (mq, sq, event, &sq->src_segment);
753 /* Applying the segment may have made the queue non-full again, unblock it if needed */
754 gst_data_queue_limits_changed (sq->queue);
755 break;
756 default:
757 break;
758 }
760 GST_DEBUG_OBJECT (mq,
761 "SingleQueue %d : Pushing event %p of type %s",
762 sq->id, event, GST_EVENT_TYPE_NAME (event));
764 gst_pad_push_event (sq->srcpad, event);
765 } else {
766 g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
767 sq->id);
768 }
769 return result;
771 /* ERRORS */
772 }
774 static GstMiniObject *
775 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
776 {
777 GstMiniObject *res;
779 res = item->object;
780 item->object = NULL;
782 return res;
783 }
785 static void
786 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
787 {
788 if (item->object)
789 gst_mini_object_unref (item->object);
790 g_free (item);
791 }
793 /* takes ownership of passed mini object! */
794 static GstMultiQueueItem *
795 gst_multi_queue_item_new (GstMiniObject * object, guint32 curid)
796 {
797 GstMultiQueueItem *item;
799 item = g_new (GstMultiQueueItem, 1);
800 item->object = object;
801 item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
802 item->posid = curid;
804 if (GST_IS_BUFFER (object)) {
805 item->size = GST_BUFFER_SIZE (object);
806 item->duration = GST_BUFFER_DURATION (object);
807 if (item->duration == GST_CLOCK_TIME_NONE)
808 item->duration = 0;
809 item->visible = TRUE;
810 } else {
811 item->size = 0;
812 item->duration = 0;
813 item->visible = FALSE;
814 }
815 return item;
816 }
818 /* Each main loop attempts to push buffers until the return value
819 * is not-linked. not-linked pads are not allowed to push data beyond
820 * any linked pads, so they don't 'rush ahead of the pack'.
821 */
822 static void
823 gst_multi_queue_loop (GstPad * pad)
824 {
825 GstSingleQueue *sq;
826 GstMultiQueueItem *item;
827 GstDataQueueItem *sitem;
828 GstMultiQueue *mq;
829 GstMiniObject *object;
830 guint32 newid;
831 guint32 oldid = G_MAXUINT32;
832 GstFlowReturn result;
834 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
835 mq = sq->mqueue;
837 do {
838 GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
840 /* Get something from the queue, blocking until that happens, or we get
841 * flushed */
842 if (!(gst_data_queue_pop (sq->queue, &sitem)))
843 goto out_flushing;
845 item = (GstMultiQueueItem *) sitem;
846 newid = item->posid;
848 /* steal the object and destroy the item */
849 object = gst_multi_queue_item_steal_object (item);
850 gst_multi_queue_item_destroy (item);
852 GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
853 sq->id, newid, oldid);
855 /* If we're not-linked, we do some extra work because we might need to
856 * wait before pushing. If we're linked but there's a gap in the IDs,
857 * or it's the first loop, or we just passed the previous highid,
858 * we might need to wake some sleeping pad up, so there's extra work
859 * there too */
860 if (sq->srcresult == GST_FLOW_NOT_LINKED ||
861 (oldid == G_MAXUINT32) || (newid != (oldid + 1)) ||
862 oldid > mq->highid) {
863 GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
864 gst_flow_get_name (sq->srcresult));
866 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
868 /* Update the nextid so other threads know when to wake us up */
869 sq->nextid = newid;
871 /* Update the oldid (the last ID we output) for highid tracking */
872 if (oldid != G_MAXUINT32)
873 sq->oldid = oldid;
875 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
876 /* Go to sleep until it's time to push this buffer */
878 /* Recompute the highid */
879 compute_high_id (mq);
880 while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
881 GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
882 "newid %u and highid %u", sq->id, newid, mq->highid);
885 /* Wake up all non-linked pads before we sleep */
886 wake_up_next_non_linked (mq);
888 mq->numwaiting++;
889 g_cond_wait (sq->turn, mq->qlock);
890 mq->numwaiting--;
892 GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
893 "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
894 }
896 /* Re-compute the high_id in case someone else pushed */
897 compute_high_id (mq);
898 } else {
899 compute_high_id (mq);
900 /* Wake up all non-linked pads */
901 wake_up_next_non_linked (mq);
902 }
903 /* We're done waiting, we can clear the nextid */
904 sq->nextid = 0;
906 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
907 }
909 GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
910 gst_flow_get_name (sq->srcresult));
912 /* Try to push out the new object */
913 result = gst_single_queue_push_one (mq, sq, object);
914 sq->srcresult = result;
916 if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED)
917 goto out_flushing;
919 GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
920 gst_flow_get_name (sq->srcresult));
922 oldid = newid;
923 }
924 while (TRUE);
926 out_flushing:
927 {
928 /* Need to make sure wake up any sleeping pads when we exit */
929 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
930 compute_high_id (mq);
931 wake_up_next_non_linked (mq);
932 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
934 gst_data_queue_set_flushing (sq->queue, TRUE);
935 gst_pad_pause_task (sq->srcpad);
936 GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
937 "SingleQueue[%d] task paused, reason:%s",
938 sq->id, gst_flow_get_name (sq->srcresult));
939 return;
940 }
941 }
943 /**
944 * gst_multi_queue_chain:
945 *
946 * This is similar to GstQueue's chain function, except:
947 * _ we don't have leak behavioures,
948 * _ we push with a unique id (curid)
949 */
950 static GstFlowReturn
951 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
952 {
953 GstSingleQueue *sq;
954 GstMultiQueue *mq;
955 GstMultiQueueItem *item;
956 GstFlowReturn ret = GST_FLOW_OK;
957 guint32 curid;
958 GstClockTime timestamp, duration;
960 sq = gst_pad_get_element_private (pad);
961 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
963 /* Get a unique incrementing id */
964 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
965 curid = mq->counter++;
966 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
968 GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
969 sq->id, buffer, curid);
971 item = gst_multi_queue_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
973 timestamp = GST_BUFFER_TIMESTAMP (buffer);
974 duration = GST_BUFFER_DURATION (buffer);
976 if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
977 goto flushing;
979 /* update time level, we must do this after pushing the data in the queue so
980 * that we never end up filling the queue first. */
981 apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
983 done:
984 gst_object_unref (mq);
986 return ret;
988 /* ERRORS */
989 flushing:
990 {
991 ret = sq->srcresult;
992 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
993 sq->id, gst_flow_get_name (ret));
994 gst_multi_queue_item_destroy (item);
995 goto done;
996 }
997 }
999 static gboolean
1000 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
1001 {
1002 GstSingleQueue *sq;
1004 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1006 if (active) {
1007 /* All pads start off linked until they push one buffer */
1008 sq->srcresult = GST_FLOW_OK;
1009 } else {
1010 sq->srcresult = GST_FLOW_WRONG_STATE;
1011 gst_data_queue_flush (sq->queue);
1012 }
1013 return TRUE;
1014 }
1016 static gboolean
1017 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
1018 {
1019 GstSingleQueue *sq;
1020 GstMultiQueue *mq;
1021 guint32 curid;
1022 GstMultiQueueItem *item;
1023 gboolean res;
1024 GstEventType type;
1025 GstEvent *sref = NULL;
1027 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1028 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
1030 type = GST_EVENT_TYPE (event);
1032 switch (type) {
1033 case GST_EVENT_FLUSH_START:
1034 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
1035 sq->id);
1037 res = gst_pad_push_event (sq->srcpad, event);
1039 gst_single_queue_flush (mq, sq, TRUE);
1040 goto done;
1042 case GST_EVENT_FLUSH_STOP:
1043 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
1044 sq->id);
1046 res = gst_pad_push_event (sq->srcpad, event);
1048 gst_single_queue_flush (mq, sq, FALSE);
1049 goto done;
1050 case GST_EVENT_NEWSEGMENT:
1051 /* take ref because the queue will take ownership and we need the event
1052 * afterwards to update the segment */
1053 sref = gst_event_ref (event);
1054 break;
1056 default:
1057 if (!(GST_EVENT_IS_SERIALIZED (event))) {
1058 res = gst_pad_push_event (sq->srcpad, event);
1059 goto done;
1060 }
1061 break;
1062 }
1064 /* Get an unique incrementing id */
1065 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1066 curid = mq->counter++;
1067 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1069 item = gst_multi_queue_item_new ((GstMiniObject *) event, curid);
1071 GST_DEBUG_OBJECT (mq,
1072 "SingleQueue %d : Enqueuing event %p of type %s with id %d",
1073 sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
1075 if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1076 goto flushing;
1078 /* mark EOS when we received one, we must do that after putting the
1079 * buffer in the queue because EOS marks the buffer as filled. No need to take
1080 * a lock, the _check_full happens from this thread only, right before pushing
1081 * into dataqueue. */
1082 switch (type) {
1083 case GST_EVENT_EOS:
1084 sq->is_eos = TRUE;
1085 break;
1086 case GST_EVENT_NEWSEGMENT:
1087 apply_segment (mq, sq, sref, &sq->sink_segment);
1088 gst_event_unref (sref);
1089 break;
1090 default:
1091 break;
1092 }
1093 done:
1094 gst_object_unref (mq);
1095 return res;
1097 flushing:
1098 {
1099 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1100 sq->id, gst_flow_get_name (sq->srcresult));
1101 if (sref)
1102 gst_event_unref (sref);
1103 gst_multi_queue_item_destroy (item);
1104 goto done;
1105 }
1106 }
1108 static GstCaps *
1109 gst_multi_queue_getcaps (GstPad * pad)
1110 {
1111 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1112 GstPad *otherpad;
1113 GstCaps *result;
1115 otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1117 GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
1119 result = gst_pad_peer_get_caps (otherpad);
1120 if (result == NULL)
1121 result = gst_caps_new_any ();
1123 return result;
1124 }
1126 static GstFlowReturn
1127 gst_multi_queue_bufferalloc (GstPad * pad, guint64 offset, guint size,
1128 GstCaps * caps, GstBuffer ** buf)
1129 {
1130 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1132 return gst_pad_alloc_buffer (sq->srcpad, offset, size, caps, buf);
1133 }
1135 static gboolean
1136 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1137 {
1138 GstMultiQueue *mq;
1139 GstSingleQueue *sq;
1140 gboolean result = FALSE;
1142 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1143 mq = sq->mqueue;
1145 GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1147 if (active) {
1148 result = gst_single_queue_flush (mq, sq, FALSE);
1149 } else {
1150 result = gst_single_queue_flush (mq, sq, TRUE);
1151 /* make sure streaming finishes */
1152 result |= gst_pad_stop_task (pad);
1153 }
1154 return result;
1155 }
1157 static gboolean
1158 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1159 {
1160 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1162 return gst_pad_push_event (sq->sinkpad, event);
1163 }
1165 static gboolean
1166 gst_multi_queue_src_query (GstPad * pad, GstQuery * query)
1167 {
1168 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1169 GstPad *peerpad;
1170 gboolean res;
1172 /* FIXME, Handle position offset depending on queue size */
1174 /* default handling */
1175 if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1176 goto no_peer;
1178 res = gst_pad_query (peerpad, query);
1180 gst_object_unref (peerpad);
1182 return res;
1184 /* ERRORS */
1185 no_peer:
1186 {
1187 GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1188 return FALSE;
1189 }
1190 }
1192 /*
1193 * Next-non-linked functions
1194 */
1196 /* WITH LOCK TAKEN */
1197 static void
1198 wake_up_next_non_linked (GstMultiQueue * mq)
1199 {
1200 GList *tmp;
1202 /* maybe no-one is waiting */
1203 if (mq->numwaiting < 1)
1204 return;
1206 /* Else figure out which singlequeue(s) need waking up */
1207 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1208 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1210 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1211 if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1212 GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1213 g_cond_signal (sq->turn);
1214 }
1215 }
1216 }
1217 }
1219 /* WITH LOCK TAKEN */
1220 static void
1221 compute_high_id (GstMultiQueue * mq)
1222 {
1223 /* The high-id is either the highest id among the linked pads, or if all
1224 * pads are not-linked, it's the lowest not-linked pad */
1225 GList *tmp;
1226 guint32 lowest = G_MAXUINT32;
1227 guint32 highid = G_MAXUINT32;
1229 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1230 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1232 GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1233 sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1235 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1236 /* No need to consider queues which are not waiting */
1237 if (sq->nextid == 0) {
1238 GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1239 continue;
1240 }
1242 if (sq->nextid < lowest)
1243 lowest = sq->nextid;
1244 } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1245 /* If we don't have a global highid, or the global highid is lower than
1246 * this single queue's last outputted id, store the queue's one,
1247 * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1248 if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1249 highid = sq->oldid;
1250 }
1251 }
1253 if (highid == G_MAXUINT32 || lowest < highid)
1254 mq->highid = lowest;
1255 else
1256 mq->highid = highid;
1258 GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1259 lowest);
1260 }
1262 #define IS_FILLED(format, value) ((sq->max_size.format) != 0 && \
1263 (sq->max_size.format) <= (value))
1265 /*
1266 * GstSingleQueue functions
1267 */
1268 static void
1269 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1270 {
1271 GstMultiQueue *mq = sq->mqueue;
1272 GList *tmp;
1273 GstDataQueueSize size;
1274 gboolean filled = FALSE;
1276 gst_data_queue_get_level (sq->queue, &size);
1278 GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1280 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1281 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1282 GstSingleQueue *ssq = (GstSingleQueue *) tmp->data;
1283 GstDataQueueSize ssize;
1285 GST_LOG_OBJECT (mq, "Checking Queue %d", ssq->id);
1287 if (gst_data_queue_is_empty (ssq->queue)) {
1288 GST_LOG_OBJECT (mq, "Queue %d is empty", ssq->id);
1289 if (IS_FILLED (visible, size.visible)) {
1290 sq->max_size.visible = size.visible + 1;
1291 GST_DEBUG_OBJECT (mq,
1292 "Another queue is empty, bumping single queue %d max visible to %d",
1293 sq->id, sq->max_size.visible);
1294 }
1295 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1296 goto beach;
1297 }
1298 /* check if we reached the hard time/bytes limits */
1299 gst_data_queue_get_level (ssq->queue, &ssize);
1301 GST_DEBUG_OBJECT (mq,
1302 "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1303 G_GUINT64_FORMAT, ssq->id, ssize.visible, sq->max_size.visible,
1304 ssize.bytes, sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1306 /* if this queue is filled completely we must signal overrun */
1307 if (IS_FILLED (bytes, ssize.bytes) || IS_FILLED (time, sq->cur_time)) {
1308 GST_LOG_OBJECT (mq, "Queue %d is filled", ssq->id);
1309 filled = TRUE;
1310 }
1311 }
1312 /* no queues were empty */
1313 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1315 /* Overrun is always forwarded, since this is blocking the upstream element */
1316 if (filled) {
1317 GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1318 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1319 }
1321 beach:
1322 return;
1323 }
1325 static void
1326 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1327 {
1328 gboolean empty = TRUE;
1329 GstMultiQueue *mq = sq->mqueue;
1330 GList *tmp;
1332 GST_LOG_OBJECT (mq,
1333 "Single Queue %d is empty, Checking other single queues", sq->id);
1335 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1336 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1337 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1339 if (gst_data_queue_is_full (sq->queue)) {
1340 GstDataQueueSize size;
1342 gst_data_queue_get_level (sq->queue, &size);
1343 if (IS_FILLED (visible, size.visible)) {
1344 sq->max_size.visible = size.visible + 1;
1345 GST_DEBUG_OBJECT (mq,
1346 "queue %d is filled, bumping its max visible to %d", sq->id,
1347 sq->max_size.visible);
1348 gst_data_queue_limits_changed (sq->queue);
1349 }
1350 }
1351 if (!gst_data_queue_is_empty (sq->queue))
1352 empty = FALSE;
1353 }
1354 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1356 if (empty) {
1357 GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1358 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1359 }
1360 }
1362 static gboolean
1363 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1364 guint64 time, GstSingleQueue * sq)
1365 {
1366 gboolean res;
1368 GST_DEBUG ("queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT
1369 "/%" G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1370 sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1372 /* we are always filled on EOS */
1373 if (sq->is_eos)
1374 return TRUE;
1376 /* we never go past the max visible items */
1377 if (IS_FILLED (visible, visible))
1378 return TRUE;
1380 if (sq->cur_time != 0) {
1381 /* if we have valid time in the queue, check */
1382 res = IS_FILLED (time, sq->cur_time);
1383 } else {
1384 /* no valid time, check bytes */
1385 res = IS_FILLED (bytes, bytes);
1386 }
1387 return res;
1388 }
1390 static void
1391 gst_single_queue_free (GstSingleQueue * sq)
1392 {
1393 /* DRAIN QUEUE */
1394 gst_data_queue_flush (sq->queue);
1395 g_object_unref (sq->queue);
1396 g_cond_free (sq->turn);
1397 g_free (sq);
1398 }
1400 static GstSingleQueue *
1401 gst_single_queue_new (GstMultiQueue * mqueue)
1402 {
1403 GstSingleQueue *sq;
1404 gchar *tmp;
1406 sq = g_new0 (GstSingleQueue, 1);
1408 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1409 sq->id = mqueue->nbqueues++;
1411 /* copy over max_size and extra_size so we don't need to take the lock
1412 * any longer when checking if the queue is full. */
1413 sq->max_size.visible = mqueue->max_size.visible;
1414 sq->max_size.bytes = mqueue->max_size.bytes;
1415 sq->max_size.time = mqueue->max_size.time;
1417 sq->extra_size.visible = mqueue->extra_size.visible;
1418 sq->extra_size.bytes = mqueue->extra_size.bytes;
1419 sq->extra_size.time = mqueue->extra_size.time;
1421 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1423 GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1425 sq->mqueue = mqueue;
1426 sq->srcresult = GST_FLOW_WRONG_STATE;
1427 sq->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction)
1428 single_queue_check_full, sq);
1429 sq->is_eos = FALSE;
1430 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1431 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1433 sq->nextid = 0;
1434 sq->oldid = 0;
1435 sq->turn = g_cond_new ();
1437 /* attach to underrun/overrun signals to handle non-starvation */
1438 g_signal_connect (G_OBJECT (sq->queue), "full",
1439 G_CALLBACK (single_queue_overrun_cb), sq);
1440 g_signal_connect (G_OBJECT (sq->queue), "empty",
1441 G_CALLBACK (single_queue_underrun_cb), sq);
1443 tmp = g_strdup_printf ("sink%d", sq->id);
1444 sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
1445 g_free (tmp);
1447 gst_pad_set_chain_function (sq->sinkpad,
1448 GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1449 gst_pad_set_activatepush_function (sq->sinkpad,
1450 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1451 gst_pad_set_event_function (sq->sinkpad,
1452 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1453 gst_pad_set_getcaps_function (sq->sinkpad,
1454 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1455 gst_pad_set_bufferalloc_function (sq->sinkpad,
1456 GST_DEBUG_FUNCPTR (gst_multi_queue_bufferalloc));
1457 gst_pad_set_internal_link_function (sq->sinkpad,
1458 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1460 tmp = g_strdup_printf ("src%d", sq->id);
1461 sq->srcpad = gst_pad_new_from_static_template (&srctemplate, tmp);
1462 g_free (tmp);
1464 gst_pad_set_activatepush_function (sq->srcpad,
1465 GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1466 gst_pad_set_getcaps_function (sq->srcpad,
1467 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1468 gst_pad_set_event_function (sq->srcpad,
1469 GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1470 gst_pad_set_query_function (sq->srcpad,
1471 GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1472 gst_pad_set_internal_link_function (sq->srcpad,
1473 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1475 gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1476 gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1478 gst_pad_set_active (sq->srcpad, TRUE);
1479 gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1481 gst_pad_set_active (sq->sinkpad, TRUE);
1482 gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1484 GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1485 sq->id);
1487 return sq;
1488 }