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);
172 static void single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq);
174 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",
175 GST_PAD_SINK,
176 GST_PAD_REQUEST,
177 GST_STATIC_CAPS_ANY);
179 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
180 GST_PAD_SRC,
181 GST_PAD_SOMETIMES,
182 GST_STATIC_CAPS_ANY);
184 GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);
185 #define GST_CAT_DEFAULT (multi_queue_debug)
187 /* default limits, we try to keep up to 2 seconds of data and if there is not
188 * time, up to 10 MB. The number of buffers is dynamically scaled to make sure
189 * there is data in the queues. Normally, the byte and time limits are not hit
190 * in theses conditions. */
191 #define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
192 #define DEFAULT_MAX_SIZE_BUFFERS 5
193 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND
195 /* second limits. When we hit one of the above limits we are probably dealing
196 * with a badly muxed file and we scale the limits to these emergency values.
197 * This is currently not yet implemented. */
198 #define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
199 #define DEFAULT_EXTRA_SIZE_BUFFERS 5
200 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND
202 /* Signals and args */
203 enum
204 {
205 SIGNAL_UNDERRUN,
206 SIGNAL_OVERRUN,
207 LAST_SIGNAL
208 };
210 enum
211 {
212 ARG_0,
213 ARG_EXTRA_SIZE_BYTES,
214 ARG_EXTRA_SIZE_BUFFERS,
215 ARG_EXTRA_SIZE_TIME,
216 ARG_MAX_SIZE_BYTES,
217 ARG_MAX_SIZE_BUFFERS,
218 ARG_MAX_SIZE_TIME,
219 };
221 #define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
222 g_mutex_lock (q->qlock); \
223 } G_STMT_END
225 #define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
226 g_mutex_unlock (q->qlock); \
227 } G_STMT_END
229 static void gst_multi_queue_finalize (GObject * object);
230 static void gst_multi_queue_set_property (GObject * object,
231 guint prop_id, const GValue * value, GParamSpec * pspec);
232 static void gst_multi_queue_get_property (GObject * object,
233 guint prop_id, GValue * value, GParamSpec * pspec);
235 static GstPad *gst_multi_queue_request_new_pad (GstElement * element,
236 GstPadTemplate * temp, const gchar * name);
237 static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);
239 static void gst_multi_queue_loop (GstPad * pad);
241 #define _do_init(bla) \
242 GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");
244 GST_BOILERPLATE_FULL (GstMultiQueue, gst_multi_queue, GstElement,
245 GST_TYPE_ELEMENT, _do_init);
247 static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };
249 static void
250 gst_multi_queue_base_init (gpointer g_class)
251 {
252 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
254 gst_element_class_set_details_simple (gstelement_class,
255 "MultiQueue",
256 "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");
257 gst_element_class_add_pad_template (gstelement_class,
258 gst_static_pad_template_get (&sinktemplate));
259 gst_element_class_add_pad_template (gstelement_class,
260 gst_static_pad_template_get (&srctemplate));
261 }
263 static void
264 gst_multi_queue_class_init (GstMultiQueueClass * klass)
265 {
266 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
267 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
269 gobject_class->set_property =
270 GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);
271 gobject_class->get_property =
272 GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);
274 /* SIGNALS */
276 /**
277 * GstMultiQueue::underrun:
278 * @multiqueue: the multqueue instance
279 *
280 * This signal is emitted from the streaming thread when there is
281 * no data in any of the queues inside the multiqueue instance (underrun).
282 *
283 * This indicates either starvation or EOS from the upstream data sources.
284 */
285 gst_multi_queue_signals[SIGNAL_UNDERRUN] =
286 g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
287 G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,
288 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
290 /**
291 * GstMultiQueue::overrun:
292 * @multiqueue: the multiqueue instance
293 *
294 * Reports that one of the queues in the multiqueue is full (overrun).
295 * A queue is full if the total amount of data inside it (num-buffers, time,
296 * size) is higher than the boundary values which can be set through the
297 * GObject properties.
298 *
299 * This can be used as an indicator of pre-roll.
300 */
301 gst_multi_queue_signals[SIGNAL_OVERRUN] =
302 g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
303 G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,
304 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
306 /* PROPERTIES */
308 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
309 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
310 "Max. amount of data in the queue (bytes, 0=disable)",
311 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
312 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
313 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
314 g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
315 "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
316 DEFAULT_MAX_SIZE_BUFFERS,
317 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
318 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
319 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
320 "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
321 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
323 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,
324 g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",
325 "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",
326 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES,
327 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
328 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,
329 g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",
330 "Amount of buffers the queues can grow if one of them is empty (0=disable)",
331 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS,
332 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
333 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,
334 g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",
335 "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",
336 0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME,
337 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
339 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_multi_queue_finalize);
341 gstelement_class->request_new_pad =
342 GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);
343 gstelement_class->release_pad =
344 GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);
345 }
347 static void
348 gst_multi_queue_init (GstMultiQueue * mqueue, GstMultiQueueClass * klass)
349 {
350 mqueue->nbqueues = 0;
351 mqueue->queues = NULL;
353 mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;
354 mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;
355 mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;
357 mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;
358 mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;
359 mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
361 mqueue->counter = 1;
362 mqueue->highid = -1;
363 mqueue->nextnotlinked = -1;
365 mqueue->qlock = g_mutex_new ();
366 }
368 static void
369 gst_multi_queue_finalize (GObject * object)
370 {
371 GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);
373 g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);
374 g_list_free (mqueue->queues);
375 mqueue->queues = NULL;
376 mqueue->queues_cookie++;
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 tmp = g_list_next(tmp); \
390 }; \
391 } G_STMT_END
393 static void
394 gst_multi_queue_set_property (GObject * object, guint prop_id,
395 const GValue * value, GParamSpec * pspec)
396 {
397 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
399 switch (prop_id) {
400 case ARG_MAX_SIZE_BYTES:
401 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
402 mq->max_size.bytes = g_value_get_uint (value);
403 SET_CHILD_PROPERTY (mq, bytes);
404 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
405 break;
406 case ARG_MAX_SIZE_BUFFERS:
407 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
408 mq->max_size.visible = g_value_get_uint (value);
409 SET_CHILD_PROPERTY (mq, visible);
410 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
411 break;
412 case ARG_MAX_SIZE_TIME:
413 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
414 mq->max_size.time = g_value_get_uint64 (value);
415 SET_CHILD_PROPERTY (mq, time);
416 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
417 break;
418 case ARG_EXTRA_SIZE_BYTES:
419 mq->extra_size.bytes = g_value_get_uint (value);
420 break;
421 case ARG_EXTRA_SIZE_BUFFERS:
422 mq->extra_size.visible = g_value_get_uint (value);
423 break;
424 case ARG_EXTRA_SIZE_TIME:
425 mq->extra_size.time = g_value_get_uint64 (value);
426 break;
427 default:
428 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
429 break;
430 }
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 GstIterator *
469 gst_multi_queue_iterate_internal_links (GstPad * pad)
470 {
471 GstIterator *it = NULL;
472 GstPad *opad;
473 GstSingleQueue *squeue;
474 GstMultiQueue *mq = GST_MULTI_QUEUE (gst_pad_get_parent (pad));
476 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
477 squeue = gst_pad_get_element_private (pad);
478 if (!squeue)
479 goto out;
481 if (squeue->sinkpad == pad)
482 opad = gst_object_ref (squeue->srcpad);
483 else if (squeue->srcpad == pad)
484 opad = gst_object_ref (squeue->sinkpad);
485 else
486 goto out;
488 it = gst_iterator_new_single (GST_TYPE_PAD, opad,
489 (GstCopyFunction) gst_object_ref, (GFreeFunc) gst_object_unref);
491 gst_object_unref (opad);
493 out:
494 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
495 gst_object_unref (mq);
497 return it;
498 }
501 /*
502 * GstElement methods
503 */
505 static GstPad *
506 gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,
507 const gchar * name)
508 {
509 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
510 GstSingleQueue *squeue;
512 GST_LOG_OBJECT (element, "name : %s", GST_STR_NULL (name));
514 /* Create a new single queue, add the sink and source pad and return the sink pad */
515 squeue = gst_single_queue_new (mqueue);
517 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
518 mqueue->queues = g_list_append (mqueue->queues, squeue);
519 mqueue->queues_cookie++;
520 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
522 GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",
523 GST_DEBUG_PAD_NAME (squeue->sinkpad));
525 return squeue->sinkpad;
526 }
528 static void
529 gst_multi_queue_release_pad (GstElement * element, GstPad * pad)
530 {
531 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
532 GstSingleQueue *sq = NULL;
533 GList *tmp;
535 GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));
537 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
538 /* Find which single queue it belongs to, knowing that it should be a sinkpad */
539 for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
540 sq = (GstSingleQueue *) tmp->data;
542 if (sq->sinkpad == pad)
543 break;
544 }
546 if (!tmp) {
547 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
548 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
549 return;
550 }
552 /* FIXME: The removal of the singlequeue should probably not happen until it
553 * finishes draining */
555 /* remove it from the list */
556 mqueue->queues = g_list_delete_link (mqueue->queues, tmp);
557 mqueue->queues_cookie++;
559 /* FIXME : recompute next-non-linked */
560 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
562 /* delete SingleQueue */
563 gst_data_queue_set_flushing (sq->queue, TRUE);
565 gst_pad_set_active (sq->srcpad, FALSE);
566 gst_pad_set_active (sq->sinkpad, FALSE);
567 gst_pad_set_element_private (sq->srcpad, NULL);
568 gst_pad_set_element_private (sq->sinkpad, NULL);
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);
628 if (sink_time == GST_CLOCK_TIME_NONE)
629 goto beach;
631 src_time = gst_segment_to_running_time (&sq->src_segment, GST_FORMAT_TIME,
632 sq->src_segment.last_stop);
633 if (src_time == GST_CLOCK_TIME_NONE)
634 goto beach;
636 GST_DEBUG_OBJECT (mq,
637 "queue %d, sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT, sq->id,
638 GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
640 /* This allows for streams with out of order timestamping - sometimes the
641 * emerging timestamp is later than the arriving one(s) */
642 if (sink_time < src_time)
643 goto beach;
645 sq->cur_time = sink_time - src_time;
646 return;
648 beach:
649 sq->cur_time = 0;
650 }
652 /* take a NEWSEGMENT event and apply the values to segment, updating the time
653 * level of queue. */
654 static void
655 apply_segment (GstMultiQueue * mq, GstSingleQueue * sq, GstEvent * event,
656 GstSegment * segment)
657 {
658 gboolean update;
659 GstFormat format;
660 gdouble rate, arate;
661 gint64 start, stop, time;
663 gst_event_parse_new_segment_full (event, &update, &rate, &arate,
664 &format, &start, &stop, &time);
666 /* now configure the values, we use these to track timestamps on the
667 * sinkpad. */
668 if (format != GST_FORMAT_TIME) {
669 /* non-time format, pretent the current time segment is closed with a
670 * 0 start and unknown stop time. */
671 update = FALSE;
672 format = GST_FORMAT_TIME;
673 start = 0;
674 stop = -1;
675 time = 0;
676 }
678 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
680 gst_segment_set_newsegment_full (segment, update,
681 rate, arate, format, start, stop, time);
683 GST_DEBUG_OBJECT (mq,
684 "queue %d, configured NEWSEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
686 /* segment can update the time level of the queue */
687 update_time_level (mq, sq);
689 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
690 }
692 /* take a buffer and update segment, updating the time level of the queue. */
693 static void
694 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
695 GstClockTime duration, GstSegment * segment)
696 {
697 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
699 /* if no timestamp is set, assume it's continuous with the previous
700 * time */
701 if (timestamp == GST_CLOCK_TIME_NONE)
702 timestamp = segment->last_stop;
704 /* add duration */
705 if (duration != GST_CLOCK_TIME_NONE)
706 timestamp += duration;
708 GST_DEBUG_OBJECT (mq, "queue %d, last_stop updated to %" GST_TIME_FORMAT,
709 sq->id, GST_TIME_ARGS (timestamp));
711 gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
713 /* calc diff with other end */
714 update_time_level (mq, sq);
715 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
716 }
718 static GstFlowReturn
719 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
720 GstMiniObject * object)
721 {
722 GstFlowReturn result = GST_FLOW_OK;
724 if (GST_IS_BUFFER (object)) {
725 GstBuffer *buffer;
726 GstClockTime timestamp, duration;
727 GstCaps *caps;
729 buffer = GST_BUFFER_CAST (object);
730 timestamp = GST_BUFFER_TIMESTAMP (buffer);
731 duration = GST_BUFFER_DURATION (buffer);
732 caps = GST_BUFFER_CAPS (buffer);
734 apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
736 /* Applying the buffer may have made the queue non-full again, unblock it if needed */
737 gst_data_queue_limits_changed (sq->queue);
739 GST_DEBUG_OBJECT (mq,
740 "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
741 sq->id, buffer, GST_TIME_ARGS (timestamp));
743 /* Set caps on pad before pushing, this avoids core calling the accpetcaps
744 * function on the srcpad, which will call acceptcaps upstream, which might
745 * not accept these caps (anymore). */
746 if (caps && caps != GST_PAD_CAPS (sq->srcpad))
747 gst_pad_set_caps (sq->srcpad, caps);
749 result = gst_pad_push (sq->srcpad, buffer);
750 } else if (GST_IS_EVENT (object)) {
751 GstEvent *event;
753 event = GST_EVENT_CAST (object);
755 switch (GST_EVENT_TYPE (event)) {
756 case GST_EVENT_EOS:
757 result = GST_FLOW_UNEXPECTED;
758 break;
759 case GST_EVENT_NEWSEGMENT:
760 apply_segment (mq, sq, event, &sq->src_segment);
761 /* Applying the segment may have made the queue non-full again, unblock it if needed */
762 gst_data_queue_limits_changed (sq->queue);
763 break;
764 default:
765 break;
766 }
768 GST_DEBUG_OBJECT (mq,
769 "SingleQueue %d : Pushing event %p of type %s",
770 sq->id, event, GST_EVENT_TYPE_NAME (event));
772 gst_pad_push_event (sq->srcpad, event);
773 } else {
774 g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
775 sq->id);
776 }
777 return result;
779 /* ERRORS */
780 }
782 static GstMiniObject *
783 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
784 {
785 GstMiniObject *res;
787 res = item->object;
788 item->object = NULL;
790 return res;
791 }
793 static void
794 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
795 {
796 if (item->object)
797 gst_mini_object_unref (item->object);
798 g_slice_free (GstMultiQueueItem, item);
799 }
801 /* takes ownership of passed mini object! */
802 static GstMultiQueueItem *
803 gst_multi_queue_item_new (GstMiniObject * object, guint32 curid)
804 {
805 GstMultiQueueItem *item;
807 item = g_slice_new (GstMultiQueueItem);
808 item->object = object;
809 item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
810 item->posid = curid;
812 if (GST_IS_BUFFER (object)) {
813 item->size = GST_BUFFER_SIZE (object);
814 item->duration = GST_BUFFER_DURATION (object);
815 if (item->duration == GST_CLOCK_TIME_NONE)
816 item->duration = 0;
817 item->visible = TRUE;
818 } else {
819 item->size = 0;
820 item->duration = 0;
821 item->visible = FALSE;
822 }
823 return item;
824 }
826 /* Each main loop attempts to push buffers until the return value
827 * is not-linked. not-linked pads are not allowed to push data beyond
828 * any linked pads, so they don't 'rush ahead of the pack'.
829 */
830 static void
831 gst_multi_queue_loop (GstPad * pad)
832 {
833 GstSingleQueue *sq;
834 GstMultiQueueItem *item;
835 GstDataQueueItem *sitem;
836 GstMultiQueue *mq;
837 GstMiniObject *object;
838 guint32 newid;
839 guint32 oldid = G_MAXUINT32;
840 GstFlowReturn result;
842 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
843 mq = sq->mqueue;
845 do {
846 GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
848 /* Get something from the queue, blocking until that happens, or we get
849 * flushed */
850 if (!(gst_data_queue_pop (sq->queue, &sitem)))
851 goto out_flushing;
853 item = (GstMultiQueueItem *) sitem;
854 newid = item->posid;
856 /* steal the object and destroy the item */
857 object = gst_multi_queue_item_steal_object (item);
858 gst_multi_queue_item_destroy (item);
860 GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
861 sq->id, newid, oldid);
863 /* If we're not-linked, we do some extra work because we might need to
864 * wait before pushing. If we're linked but there's a gap in the IDs,
865 * or it's the first loop, or we just passed the previous highid,
866 * we might need to wake some sleeping pad up, so there's extra work
867 * there too */
868 if (sq->srcresult == GST_FLOW_NOT_LINKED ||
869 (oldid == G_MAXUINT32) || (newid != (oldid + 1)) ||
870 oldid > mq->highid) {
871 GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
872 gst_flow_get_name (sq->srcresult));
874 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
876 /* Update the nextid so other threads know when to wake us up */
877 sq->nextid = newid;
879 /* Update the oldid (the last ID we output) for highid tracking */
880 if (oldid != G_MAXUINT32)
881 sq->oldid = oldid;
883 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
884 /* Go to sleep until it's time to push this buffer */
886 /* Recompute the highid */
887 compute_high_id (mq);
888 while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
889 GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
890 "newid %u and highid %u", sq->id, newid, mq->highid);
893 /* Wake up all non-linked pads before we sleep */
894 wake_up_next_non_linked (mq);
896 mq->numwaiting++;
897 g_cond_wait (sq->turn, mq->qlock);
898 mq->numwaiting--;
900 GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
901 "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
902 }
904 /* Re-compute the high_id in case someone else pushed */
905 compute_high_id (mq);
906 } else {
907 compute_high_id (mq);
908 /* Wake up all non-linked pads */
909 wake_up_next_non_linked (mq);
910 }
911 /* We're done waiting, we can clear the nextid */
912 sq->nextid = 0;
914 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
915 }
917 GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
918 gst_flow_get_name (sq->srcresult));
920 /* Try to push out the new object */
921 result = gst_single_queue_push_one (mq, sq, object);
922 sq->srcresult = result;
924 if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED)
925 goto out_flushing;
927 GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
928 gst_flow_get_name (sq->srcresult));
930 oldid = newid;
931 }
932 while (TRUE);
934 out_flushing:
935 {
936 /* Need to make sure wake up any sleeping pads when we exit */
937 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
938 compute_high_id (mq);
939 wake_up_next_non_linked (mq);
940 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
942 gst_data_queue_set_flushing (sq->queue, TRUE);
943 gst_pad_pause_task (sq->srcpad);
944 GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
945 "SingleQueue[%d] task paused, reason:%s",
946 sq->id, gst_flow_get_name (sq->srcresult));
947 return;
948 }
949 }
951 /**
952 * gst_multi_queue_chain:
953 *
954 * This is similar to GstQueue's chain function, except:
955 * _ we don't have leak behavioures,
956 * _ we push with a unique id (curid)
957 */
958 static GstFlowReturn
959 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
960 {
961 GstSingleQueue *sq;
962 GstMultiQueue *mq;
963 GstMultiQueueItem *item;
964 GstFlowReturn ret = GST_FLOW_OK;
965 guint32 curid;
966 GstClockTime timestamp, duration;
968 sq = gst_pad_get_element_private (pad);
969 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
971 /* Get a unique incrementing id */
972 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
973 curid = mq->counter++;
974 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
976 GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
977 sq->id, buffer, curid);
979 item = gst_multi_queue_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
981 timestamp = GST_BUFFER_TIMESTAMP (buffer);
982 duration = GST_BUFFER_DURATION (buffer);
984 if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
985 goto flushing;
987 /* update time level, we must do this after pushing the data in the queue so
988 * that we never end up filling the queue first. */
989 apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
991 done:
992 gst_object_unref (mq);
994 return ret;
996 /* ERRORS */
997 flushing:
998 {
999 ret = sq->srcresult;
1000 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1001 sq->id, gst_flow_get_name (ret));
1002 gst_multi_queue_item_destroy (item);
1003 goto done;
1004 }
1005 }
1007 static gboolean
1008 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
1009 {
1010 GstSingleQueue *sq;
1012 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1014 if (active) {
1015 /* All pads start off linked until they push one buffer */
1016 sq->srcresult = GST_FLOW_OK;
1017 } else {
1018 sq->srcresult = GST_FLOW_WRONG_STATE;
1019 gst_data_queue_flush (sq->queue);
1020 }
1021 return TRUE;
1022 }
1024 static gboolean
1025 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
1026 {
1027 GstSingleQueue *sq;
1028 GstMultiQueue *mq;
1029 guint32 curid;
1030 GstMultiQueueItem *item;
1031 gboolean res;
1032 GstEventType type;
1033 GstEvent *sref = NULL;
1035 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1036 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
1038 type = GST_EVENT_TYPE (event);
1040 switch (type) {
1041 case GST_EVENT_FLUSH_START:
1042 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
1043 sq->id);
1045 res = gst_pad_push_event (sq->srcpad, event);
1047 gst_single_queue_flush (mq, sq, TRUE);
1048 goto done;
1050 case GST_EVENT_FLUSH_STOP:
1051 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
1052 sq->id);
1054 res = gst_pad_push_event (sq->srcpad, event);
1056 gst_single_queue_flush (mq, sq, FALSE);
1057 goto done;
1058 case GST_EVENT_NEWSEGMENT:
1059 /* take ref because the queue will take ownership and we need the event
1060 * afterwards to update the segment */
1061 sref = gst_event_ref (event);
1062 break;
1064 default:
1065 if (!(GST_EVENT_IS_SERIALIZED (event))) {
1066 res = gst_pad_push_event (sq->srcpad, event);
1067 goto done;
1068 }
1069 break;
1070 }
1072 /* Get an unique incrementing id */
1073 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1074 curid = mq->counter++;
1075 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1077 item = gst_multi_queue_item_new ((GstMiniObject *) event, curid);
1079 GST_DEBUG_OBJECT (mq,
1080 "SingleQueue %d : Enqueuing event %p of type %s with id %d",
1081 sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
1083 if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1084 goto flushing;
1086 /* mark EOS when we received one, we must do that after putting the
1087 * buffer in the queue because EOS marks the buffer as filled. No need to take
1088 * a lock, the _check_full happens from this thread only, right before pushing
1089 * into dataqueue. */
1090 switch (type) {
1091 case GST_EVENT_EOS:
1092 sq->is_eos = TRUE;
1093 single_queue_overrun_cb (sq->queue, sq);
1094 break;
1095 case GST_EVENT_NEWSEGMENT:
1096 apply_segment (mq, sq, sref, &sq->sink_segment);
1097 gst_event_unref (sref);
1098 break;
1099 default:
1100 break;
1101 }
1102 done:
1103 gst_object_unref (mq);
1104 return res;
1106 flushing:
1107 {
1108 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1109 sq->id, gst_flow_get_name (sq->srcresult));
1110 if (sref)
1111 gst_event_unref (sref);
1112 gst_multi_queue_item_destroy (item);
1113 goto done;
1114 }
1115 }
1117 static GstCaps *
1118 gst_multi_queue_getcaps (GstPad * pad)
1119 {
1120 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1121 GstPad *otherpad;
1122 GstCaps *result;
1124 otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1126 GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
1128 result = gst_pad_peer_get_caps (otherpad);
1129 if (result == NULL)
1130 result = gst_caps_new_any ();
1132 return result;
1133 }
1135 static GstFlowReturn
1136 gst_multi_queue_bufferalloc (GstPad * pad, guint64 offset, guint size,
1137 GstCaps * caps, GstBuffer ** buf)
1138 {
1139 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1141 return gst_pad_alloc_buffer (sq->srcpad, offset, size, caps, buf);
1142 }
1144 static gboolean
1145 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1146 {
1147 GstMultiQueue *mq;
1148 GstSingleQueue *sq;
1149 gboolean result = FALSE;
1151 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1152 mq = sq->mqueue;
1154 GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1156 if (active) {
1157 result = gst_single_queue_flush (mq, sq, FALSE);
1158 } else {
1159 result = gst_single_queue_flush (mq, sq, TRUE);
1160 /* make sure streaming finishes */
1161 result |= gst_pad_stop_task (pad);
1162 }
1163 return result;
1164 }
1166 static gboolean
1167 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1168 {
1169 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1171 return gst_pad_push_event (sq->sinkpad, event);
1172 }
1174 static gboolean
1175 gst_multi_queue_src_query (GstPad * pad, GstQuery * query)
1176 {
1177 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1178 GstPad *peerpad;
1179 gboolean res;
1181 /* FIXME, Handle position offset depending on queue size */
1183 /* default handling */
1184 if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1185 goto no_peer;
1187 res = gst_pad_query (peerpad, query);
1189 gst_object_unref (peerpad);
1191 return res;
1193 /* ERRORS */
1194 no_peer:
1195 {
1196 GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1197 return FALSE;
1198 }
1199 }
1201 /*
1202 * Next-non-linked functions
1203 */
1205 /* WITH LOCK TAKEN */
1206 static void
1207 wake_up_next_non_linked (GstMultiQueue * mq)
1208 {
1209 GList *tmp;
1211 /* maybe no-one is waiting */
1212 if (mq->numwaiting < 1)
1213 return;
1215 /* Else figure out which singlequeue(s) need waking up */
1216 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1217 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1219 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1220 if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1221 GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1222 g_cond_signal (sq->turn);
1223 }
1224 }
1225 }
1226 }
1228 /* WITH LOCK TAKEN */
1229 static void
1230 compute_high_id (GstMultiQueue * mq)
1231 {
1232 /* The high-id is either the highest id among the linked pads, or if all
1233 * pads are not-linked, it's the lowest not-linked pad */
1234 GList *tmp;
1235 guint32 lowest = G_MAXUINT32;
1236 guint32 highid = G_MAXUINT32;
1238 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1239 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1241 GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1242 sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1244 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1245 /* No need to consider queues which are not waiting */
1246 if (sq->nextid == 0) {
1247 GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1248 continue;
1249 }
1251 if (sq->nextid < lowest)
1252 lowest = sq->nextid;
1253 } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1254 /* If we don't have a global highid, or the global highid is lower than
1255 * this single queue's last outputted id, store the queue's one,
1256 * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1257 if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1258 highid = sq->oldid;
1259 }
1260 }
1262 if (highid == G_MAXUINT32 || lowest < highid)
1263 mq->highid = lowest;
1264 else
1265 mq->highid = highid;
1267 GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1268 lowest);
1269 }
1271 #define IS_FILLED(format, value) ((sq->max_size.format) != 0 && \
1272 (sq->max_size.format) <= (value))
1274 /*
1275 * GstSingleQueue functions
1276 */
1277 static void
1278 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1279 {
1280 GstMultiQueue *mq = sq->mqueue;
1281 GList *tmp;
1282 GstDataQueueSize size;
1283 gboolean filled = FALSE;
1285 gst_data_queue_get_level (sq->queue, &size);
1287 GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1289 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1290 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1291 GstSingleQueue *ssq = (GstSingleQueue *) tmp->data;
1292 GstDataQueueSize ssize;
1294 GST_LOG_OBJECT (mq, "Checking Queue %d", ssq->id);
1296 if (gst_data_queue_is_empty (ssq->queue)) {
1297 GST_LOG_OBJECT (mq, "Queue %d is empty", ssq->id);
1298 if (IS_FILLED (visible, size.visible)) {
1299 sq->max_size.visible = size.visible + 1;
1300 GST_DEBUG_OBJECT (mq,
1301 "Another queue is empty, bumping single queue %d max visible to %d",
1302 sq->id, sq->max_size.visible);
1303 }
1304 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1305 goto beach;
1306 }
1307 /* check if we reached the hard time/bytes limits */
1308 gst_data_queue_get_level (ssq->queue, &ssize);
1310 GST_DEBUG_OBJECT (mq,
1311 "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1312 G_GUINT64_FORMAT, ssq->id, ssize.visible, sq->max_size.visible,
1313 ssize.bytes, sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1315 /* if this queue is filled completely we must signal overrun */
1316 if (sq->is_eos || IS_FILLED (bytes, ssize.bytes) ||
1317 IS_FILLED (time, sq->cur_time)) {
1318 GST_LOG_OBJECT (mq, "Queue %d is filled", ssq->id);
1319 filled = TRUE;
1320 }
1321 }
1322 /* no queues were empty */
1323 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1325 /* Overrun is always forwarded, since this is blocking the upstream element */
1326 if (filled) {
1327 GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1328 g_signal_emit (mq, gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1329 }
1331 beach:
1332 return;
1333 }
1335 static void
1336 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1337 {
1338 gboolean empty = TRUE;
1339 GstMultiQueue *mq = sq->mqueue;
1340 GList *tmp;
1342 GST_LOG_OBJECT (mq,
1343 "Single Queue %d is empty, Checking other single queues", sq->id);
1345 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1346 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1347 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1349 if (gst_data_queue_is_full (sq->queue)) {
1350 GstDataQueueSize size;
1352 gst_data_queue_get_level (sq->queue, &size);
1353 if (IS_FILLED (visible, size.visible)) {
1354 sq->max_size.visible = size.visible + 1;
1355 GST_DEBUG_OBJECT (mq,
1356 "queue %d is filled, bumping its max visible to %d", sq->id,
1357 sq->max_size.visible);
1358 gst_data_queue_limits_changed (sq->queue);
1359 }
1360 }
1361 if (!gst_data_queue_is_empty (sq->queue))
1362 empty = FALSE;
1363 }
1364 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1366 if (empty) {
1367 GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1368 g_signal_emit (mq, gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1369 }
1370 }
1372 static gboolean
1373 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1374 guint64 time, GstSingleQueue * sq)
1375 {
1376 gboolean res;
1378 GST_DEBUG ("queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT
1379 "/%" G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1380 sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1382 /* we are always filled on EOS */
1383 if (sq->is_eos)
1384 return TRUE;
1386 /* we never go past the max visible items */
1387 if (IS_FILLED (visible, visible))
1388 return TRUE;
1390 /* check time or bytes */
1391 res = IS_FILLED (time, sq->cur_time) || IS_FILLED (bytes, bytes);
1393 return res;
1394 }
1396 static void
1397 gst_single_queue_free (GstSingleQueue * sq)
1398 {
1399 /* DRAIN QUEUE */
1400 gst_data_queue_flush (sq->queue);
1401 g_object_unref (sq->queue);
1402 g_cond_free (sq->turn);
1403 g_free (sq);
1404 }
1406 static GstSingleQueue *
1407 gst_single_queue_new (GstMultiQueue * mqueue)
1408 {
1409 GstSingleQueue *sq;
1410 gchar *tmp;
1412 sq = g_new0 (GstSingleQueue, 1);
1414 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1415 sq->id = mqueue->nbqueues++;
1417 /* copy over max_size and extra_size so we don't need to take the lock
1418 * any longer when checking if the queue is full. */
1419 sq->max_size.visible = mqueue->max_size.visible;
1420 sq->max_size.bytes = mqueue->max_size.bytes;
1421 sq->max_size.time = mqueue->max_size.time;
1423 sq->extra_size.visible = mqueue->extra_size.visible;
1424 sq->extra_size.bytes = mqueue->extra_size.bytes;
1425 sq->extra_size.time = mqueue->extra_size.time;
1427 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1429 GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1431 sq->mqueue = mqueue;
1432 sq->srcresult = GST_FLOW_WRONG_STATE;
1433 sq->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction)
1434 single_queue_check_full, sq);
1435 sq->is_eos = FALSE;
1436 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1437 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1439 sq->nextid = 0;
1440 sq->oldid = 0;
1441 sq->turn = g_cond_new ();
1443 /* attach to underrun/overrun signals to handle non-starvation */
1444 g_signal_connect (sq->queue, "full",
1445 G_CALLBACK (single_queue_overrun_cb), sq);
1446 g_signal_connect (sq->queue, "empty",
1447 G_CALLBACK (single_queue_underrun_cb), sq);
1449 tmp = g_strdup_printf ("sink%d", sq->id);
1450 sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
1451 g_free (tmp);
1453 gst_pad_set_chain_function (sq->sinkpad,
1454 GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1455 gst_pad_set_activatepush_function (sq->sinkpad,
1456 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1457 gst_pad_set_event_function (sq->sinkpad,
1458 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1459 gst_pad_set_getcaps_function (sq->sinkpad,
1460 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1461 gst_pad_set_bufferalloc_function (sq->sinkpad,
1462 GST_DEBUG_FUNCPTR (gst_multi_queue_bufferalloc));
1463 gst_pad_set_iterate_internal_links_function (sq->sinkpad,
1464 GST_DEBUG_FUNCPTR (gst_multi_queue_iterate_internal_links));
1466 tmp = g_strdup_printf ("src%d", sq->id);
1467 sq->srcpad = gst_pad_new_from_static_template (&srctemplate, tmp);
1468 g_free (tmp);
1470 gst_pad_set_activatepush_function (sq->srcpad,
1471 GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1472 gst_pad_set_getcaps_function (sq->srcpad,
1473 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1474 gst_pad_set_event_function (sq->srcpad,
1475 GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1476 gst_pad_set_query_function (sq->srcpad,
1477 GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1478 gst_pad_set_iterate_internal_links_function (sq->sinkpad,
1479 GST_DEBUG_FUNCPTR (gst_multi_queue_iterate_internal_links));
1481 gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1482 gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1484 /* only activate the pads when we are not in the NULL state
1485 * and add the pad under the state_lock to prevend state changes
1486 * between activating and adding */
1487 g_static_rec_mutex_lock (GST_STATE_GET_LOCK (mqueue));
1488 if (GST_STATE_TARGET (mqueue) != GST_STATE_NULL) {
1489 gst_pad_set_active (sq->srcpad, TRUE);
1490 gst_pad_set_active (sq->sinkpad, TRUE);
1491 }
1492 gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1493 gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1494 g_static_rec_mutex_unlock (GST_STATE_GET_LOCK (mqueue));
1496 GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1497 sq->id);
1499 return sq;
1500 }