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 #ifdef HAVE_CONFIG_H
25 # include "config.h"
26 #endif
28 #include <gst/gst.h>
29 #include "gstmultiqueue.h"
31 /**
32 * GstSingleQueue:
33 * @sinkpad: associated sink #GstPad
34 * @srcpad: associated source #GstPad
35 *
36 * Structure containing all information and properties about
37 * a single queue.
38 */
39 typedef struct _GstSingleQueue GstSingleQueue;
41 struct _GstSingleQueue
42 {
43 /* unique identifier of the queue */
44 guint id;
46 GstMultiQueue *mqueue;
48 GstPad *sinkpad;
49 GstPad *srcpad;
51 /* flowreturn of previous srcpad push */
52 GstFlowReturn srcresult;
53 GstSegment sink_segment;
54 GstSegment src_segment;
56 /* queue of data */
57 GstDataQueue *queue;
58 GstDataQueueSize max_size, extra_size;
59 GstClockTime cur_time;
60 gboolean is_eos;
61 gboolean inextra; /* TRUE if the queue is currently in extradata mode */
63 /* Protected by global lock */
64 guint32 nextid; /* ID of the next object waiting to be pushed */
65 guint32 oldid; /* ID of the last object pushed (last in a series) */
66 GCond *turn; /* SingleQueue turn waiting conditional */
67 };
70 /* Extension of GstDataQueueItem structure for our usage */
71 typedef struct _GstMultiQueueItem GstMultiQueueItem;
73 struct _GstMultiQueueItem
74 {
75 GstMiniObject *object;
76 guint size;
77 guint64 duration;
78 gboolean visible;
80 GDestroyNotify destroy;
81 guint32 posid;
82 };
84 static GstSingleQueue *gst_single_queue_new (GstMultiQueue * mqueue);
85 static void gst_single_queue_free (GstSingleQueue * squeue);
87 static void wake_up_next_non_linked (GstMultiQueue * mq);
88 static void compute_high_id (GstMultiQueue * mq);
90 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",
91 GST_PAD_SINK,
92 GST_PAD_REQUEST,
93 GST_STATIC_CAPS_ANY);
95 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
96 GST_PAD_SRC,
97 GST_PAD_SOMETIMES,
98 GST_STATIC_CAPS_ANY);
100 GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);
101 #define GST_CAT_DEFAULT (multi_queue_debug)
103 /* default limits, we try to keep up to 2 seconds of data and if there is not
104 * time, up to 10 MB. The number of buffers is dynamically scaled to make sure
105 * there is data in the queues. Normally, the byte and time limits are not hit
106 * in theses conditions. */
107 #define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
108 #define DEFAULT_MAX_SIZE_BUFFERS 5
109 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND
111 /* second limits. When we hit one of the above limits we are probably dealing
112 * with a badly muxed file and we scale the limits to these emergency values.
113 * This is currently not yet implemented. */
114 #define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
115 #define DEFAULT_EXTRA_SIZE_BUFFERS 5
116 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND
118 /* Signals and args */
119 enum
120 {
121 SIGNAL_UNDERRUN,
122 SIGNAL_OVERRUN,
123 LAST_SIGNAL
124 };
126 enum
127 {
128 ARG_0,
129 ARG_EXTRA_SIZE_BYTES,
130 ARG_EXTRA_SIZE_BUFFERS,
131 ARG_EXTRA_SIZE_TIME,
132 ARG_MAX_SIZE_BYTES,
133 ARG_MAX_SIZE_BUFFERS,
134 ARG_MAX_SIZE_TIME,
135 };
137 #define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
138 g_mutex_lock (q->qlock); \
139 } G_STMT_END
141 #define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
142 g_mutex_unlock (q->qlock); \
143 } G_STMT_END
145 static void gst_multi_queue_finalize (GObject * object);
146 static void gst_multi_queue_set_property (GObject * object,
147 guint prop_id, const GValue * value, GParamSpec * pspec);
148 static void gst_multi_queue_get_property (GObject * object,
149 guint prop_id, GValue * value, GParamSpec * pspec);
151 static GstPad *gst_multi_queue_request_new_pad (GstElement * element,
152 GstPadTemplate * temp, const gchar * name);
153 static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);
155 static void gst_multi_queue_loop (GstPad * pad);
157 #define _do_init(bla) \
158 GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");
160 GST_BOILERPLATE_FULL (GstMultiQueue, gst_multi_queue, GstElement,
161 GST_TYPE_ELEMENT, _do_init);
163 static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };
165 static void
166 gst_multi_queue_base_init (gpointer g_class)
167 {
168 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
170 gst_element_class_set_details_simple (gstelement_class,
171 "MultiQueue",
172 "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");
173 gst_element_class_add_pad_template (gstelement_class,
174 gst_static_pad_template_get (&sinktemplate));
175 gst_element_class_add_pad_template (gstelement_class,
176 gst_static_pad_template_get (&srctemplate));
177 }
179 static void
180 gst_multi_queue_class_init (GstMultiQueueClass * klass)
181 {
182 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
183 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
185 gobject_class->set_property =
186 GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);
187 gobject_class->get_property =
188 GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);
190 /* SIGNALS */
191 gst_multi_queue_signals[SIGNAL_UNDERRUN] =
192 g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
193 G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,
194 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
196 gst_multi_queue_signals[SIGNAL_OVERRUN] =
197 g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
198 G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,
199 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
201 /* PROPERTIES */
203 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
204 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
205 "Max. amount of data in the queue (bytes, 0=disable)",
206 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES, G_PARAM_READWRITE));
207 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
208 g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
209 "Max. number of buffers in the queue (0=disable)",
210 0, G_MAXUINT, DEFAULT_MAX_SIZE_BUFFERS, G_PARAM_READWRITE));
211 g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
212 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
213 "Max. amount of data in the queue (in ns, 0=disable)",
214 0, G_MAXUINT64, DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE));
216 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,
217 g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",
218 "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",
219 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES, G_PARAM_READWRITE));
220 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,
221 g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",
222 "Amount of buffers the queues can grow if one of them is empty (0=disable)",
223 0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS, G_PARAM_READWRITE));
224 g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,
225 g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",
226 "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",
227 0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME, G_PARAM_READWRITE));
229 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_multi_queue_finalize);
231 gstelement_class->request_new_pad =
232 GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);
233 gstelement_class->release_pad =
234 GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);
235 }
237 static void
238 gst_multi_queue_init (GstMultiQueue * mqueue, GstMultiQueueClass * klass)
239 {
240 mqueue->nbqueues = 0;
241 mqueue->queues = NULL;
243 mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;
244 mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;
245 mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;
247 mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;
248 mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;
249 mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
251 mqueue->counter = 1;
252 mqueue->highid = -1;
253 mqueue->nextnotlinked = -1;
255 mqueue->qlock = g_mutex_new ();
256 }
258 static void
259 gst_multi_queue_finalize (GObject * object)
260 {
261 GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);
263 g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);
264 g_list_free (mqueue->queues);
265 mqueue->queues = NULL;
267 /* free/unref instance data */
268 g_mutex_free (mqueue->qlock);
270 G_OBJECT_CLASS (parent_class)->finalize (object);
271 }
273 #define SET_CHILD_PROPERTY(mq,format) G_STMT_START { \
274 GList * tmp = mq->queues; \
275 while (tmp) { \
276 GstSingleQueue *q = (GstSingleQueue*)tmp->data; \
277 q->max_size.format = mq->max_size.format; \
278 tmp = g_list_next(tmp); \
279 }; \
280 } G_STMT_END
282 static void
283 gst_multi_queue_set_property (GObject * object, guint prop_id,
284 const GValue * value, GParamSpec * pspec)
285 {
286 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
288 switch (prop_id) {
289 case ARG_MAX_SIZE_BYTES:
290 mq->max_size.bytes = g_value_get_uint (value);
291 SET_CHILD_PROPERTY (mq, bytes);
292 break;
293 case ARG_MAX_SIZE_BUFFERS:
294 mq->max_size.visible = g_value_get_uint (value);
295 SET_CHILD_PROPERTY (mq, visible);
296 break;
297 case ARG_MAX_SIZE_TIME:
298 mq->max_size.time = g_value_get_uint64 (value);
299 SET_CHILD_PROPERTY (mq, time);
300 break;
301 case ARG_EXTRA_SIZE_BYTES:
302 mq->extra_size.bytes = g_value_get_uint (value);
303 break;
304 case ARG_EXTRA_SIZE_BUFFERS:
305 mq->extra_size.visible = g_value_get_uint (value);
306 break;
307 case ARG_EXTRA_SIZE_TIME:
308 mq->extra_size.time = g_value_get_uint64 (value);
309 break;
310 default:
311 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
312 break;
313 }
314 }
316 static void
317 gst_multi_queue_get_property (GObject * object, guint prop_id,
318 GValue * value, GParamSpec * pspec)
319 {
320 GstMultiQueue *mq = GST_MULTI_QUEUE (object);
322 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
324 switch (prop_id) {
325 case ARG_EXTRA_SIZE_BYTES:
326 g_value_set_uint (value, mq->extra_size.bytes);
327 break;
328 case ARG_EXTRA_SIZE_BUFFERS:
329 g_value_set_uint (value, mq->extra_size.visible);
330 break;
331 case ARG_EXTRA_SIZE_TIME:
332 g_value_set_uint64 (value, mq->extra_size.time);
333 break;
334 case ARG_MAX_SIZE_BYTES:
335 g_value_set_uint (value, mq->max_size.bytes);
336 break;
337 case ARG_MAX_SIZE_BUFFERS:
338 g_value_set_uint (value, mq->max_size.visible);
339 break;
340 case ARG_MAX_SIZE_TIME:
341 g_value_set_uint64 (value, mq->max_size.time);
342 break;
343 default:
344 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
345 break;
346 }
348 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
349 }
351 GList *
352 gst_multi_queue_get_internal_links (GstPad * pad)
353 {
354 GList *res = NULL;
355 GstMultiQueue *mqueue;
356 GstSingleQueue *sq = NULL;
357 GList *tmp;
359 g_return_val_if_fail (GST_IS_PAD (pad), NULL);
361 mqueue = GST_MULTI_QUEUE (GST_PAD_PARENT (pad));
362 if (!mqueue)
363 goto no_parent;
365 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
366 /* Find which single queue it belongs to */
367 for (tmp = mqueue->queues; tmp && !res; tmp = g_list_next (tmp)) {
368 sq = (GstSingleQueue *) tmp->data;
370 if (sq->sinkpad == pad)
371 res = g_list_prepend (res, sq->srcpad);
372 if (sq->srcpad == pad)
373 res = g_list_prepend (res, sq->sinkpad);
374 }
376 if (!res)
377 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
378 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
380 return res;
382 no_parent:
383 {
384 GST_DEBUG_OBJECT (pad, "no parent");
385 return NULL;
386 }
387 }
390 /*
391 * GstElement methods
392 */
394 static GstPad *
395 gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,
396 const gchar * name)
397 {
398 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
399 GstSingleQueue *squeue;
401 GST_LOG_OBJECT (element, "name : %s", name);
403 /* Create a new single queue, add the sink and source pad and return the sink pad */
404 squeue = gst_single_queue_new (mqueue);
406 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
407 mqueue->queues = g_list_append (mqueue->queues, squeue);
408 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
410 GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",
411 GST_DEBUG_PAD_NAME (squeue->sinkpad));
413 return squeue->sinkpad;
414 }
416 static void
417 gst_multi_queue_release_pad (GstElement * element, GstPad * pad)
418 {
419 GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
420 GstSingleQueue *sq = NULL;
421 GList *tmp;
423 GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));
425 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
426 /* Find which single queue it belongs to, knowing that it should be a sinkpad */
427 for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
428 sq = (GstSingleQueue *) tmp->data;
430 if (sq->sinkpad == pad)
431 break;
432 }
434 if (!tmp) {
435 GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
436 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
437 return;
438 }
440 /* FIXME: The removal of the singlequeue should probably not happen until it
441 * finishes draining */
443 /* remove it from the list */
444 mqueue->queues = g_list_delete_link (mqueue->queues, tmp);
446 /* FIXME : recompute next-non-linked */
447 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
449 /* delete SingleQueue */
450 gst_data_queue_set_flushing (sq->queue, TRUE);
452 gst_pad_set_active (sq->srcpad, FALSE);
453 gst_pad_set_active (sq->sinkpad, FALSE);
454 gst_element_remove_pad (element, sq->srcpad);
455 gst_element_remove_pad (element, sq->sinkpad);
456 gst_single_queue_free (sq);
457 }
459 static gboolean
460 gst_single_queue_flush (GstMultiQueue * mq, GstSingleQueue * sq, gboolean flush)
461 {
462 gboolean result;
464 GST_DEBUG_OBJECT (mq, "flush %s queue %d", (flush ? "start" : "stop"),
465 sq->id);
467 if (flush) {
468 sq->srcresult = GST_FLOW_WRONG_STATE;
469 gst_data_queue_set_flushing (sq->queue, TRUE);
471 /* wake up non-linked task */
472 GST_LOG_OBJECT (mq, "SingleQueue %d : waking up eventually waiting task",
473 sq->id);
474 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
475 g_cond_signal (sq->turn);
476 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
478 GST_LOG_OBJECT (mq, "SingleQueue %d : pausing task", sq->id);
479 result = gst_pad_pause_task (sq->srcpad);
480 } else {
481 gst_data_queue_flush (sq->queue);
482 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
483 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
484 /* All pads start off not-linked for a smooth kick-off */
485 sq->srcresult = GST_FLOW_NOT_LINKED;
486 sq->cur_time = 0;
487 sq->max_size.visible = mq->max_size.visible;
488 sq->is_eos = FALSE;
489 sq->inextra = FALSE;
490 sq->nextid = 0;
491 sq->oldid = 0;
492 gst_data_queue_set_flushing (sq->queue, FALSE);
494 GST_LOG_OBJECT (mq, "SingleQueue %d : starting task", sq->id);
495 result =
496 gst_pad_start_task (sq->srcpad, (GstTaskFunction) gst_multi_queue_loop,
497 sq->srcpad);
498 }
499 return result;
500 }
502 /* calculate the diff between running time on the sink and src of the queue.
503 * This is the total amount of time in the queue.
504 * WITH LOCK TAKEN */
505 static void
506 update_time_level (GstMultiQueue * mq, GstSingleQueue * sq)
507 {
508 gint64 sink_time, src_time;
510 sink_time =
511 gst_segment_to_running_time (&sq->sink_segment, GST_FORMAT_TIME,
512 sq->sink_segment.last_stop);
514 src_time = gst_segment_to_running_time (&sq->src_segment, GST_FORMAT_TIME,
515 sq->src_segment.last_stop);
517 GST_DEBUG_OBJECT (mq,
518 "queue %d, sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT, sq->id,
519 GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
521 /* This allows for streams with out of order timestamping - sometimes the
522 * emerging timestamp is later than the arriving one(s) */
523 if (sink_time >= src_time)
524 sq->cur_time = sink_time - src_time;
525 else
526 sq->cur_time = 0;
527 }
529 /* take a NEWSEGMENT event and apply the values to segment, updating the time
530 * level of queue. */
531 static void
532 apply_segment (GstMultiQueue * mq, GstSingleQueue * sq, GstEvent * event,
533 GstSegment * segment)
534 {
535 gboolean update;
536 GstFormat format;
537 gdouble rate, arate;
538 gint64 start, stop, time;
540 gst_event_parse_new_segment_full (event, &update, &rate, &arate,
541 &format, &start, &stop, &time);
543 /* now configure the values, we use these to track timestamps on the
544 * sinkpad. */
545 if (format != GST_FORMAT_TIME) {
546 /* non-time format, pretent the current time segment is closed with a
547 * 0 start and unknown stop time. */
548 update = FALSE;
549 format = GST_FORMAT_TIME;
550 start = 0;
551 stop = -1;
552 time = 0;
553 }
555 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
557 gst_segment_set_newsegment_full (segment, update,
558 rate, arate, format, start, stop, time);
560 GST_DEBUG_OBJECT (mq,
561 "queue %d, configured NEWSEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
563 /* segment can update the time level of the queue */
564 update_time_level (mq, sq);
566 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
567 }
569 /* take a buffer and update segment, updating the time level of the queue. */
570 static void
571 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
572 GstClockTime duration, GstSegment * segment)
573 {
574 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
576 /* if no timestamp is set, assume it's continuous with the previous
577 * time */
578 if (timestamp == GST_CLOCK_TIME_NONE)
579 timestamp = segment->last_stop;
581 /* add duration */
582 if (duration != GST_CLOCK_TIME_NONE)
583 timestamp += duration;
585 GST_DEBUG_OBJECT (mq, "queue %d, last_stop updated to %" GST_TIME_FORMAT,
586 sq->id, GST_TIME_ARGS (timestamp));
588 gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
590 /* calc diff with other end */
591 update_time_level (mq, sq);
592 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
593 }
595 static GstFlowReturn
596 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
597 GstMiniObject * object)
598 {
599 GstFlowReturn result = GST_FLOW_OK;
601 if (GST_IS_BUFFER (object)) {
602 GstBuffer *buffer;
603 GstClockTime timestamp, duration;
605 buffer = GST_BUFFER_CAST (object);
606 timestamp = GST_BUFFER_TIMESTAMP (buffer);
607 duration = GST_BUFFER_DURATION (buffer);
609 apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
611 /* Applying the buffer may have made the queue non-full again, unblock it if needed */
612 gst_data_queue_limits_changed (sq->queue);
614 GST_DEBUG_OBJECT (mq,
615 "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
616 sq->id, buffer, GST_TIME_ARGS (timestamp));
618 result = gst_pad_push (sq->srcpad, buffer);
619 } else if (GST_IS_EVENT (object)) {
620 GstEvent *event;
622 event = GST_EVENT_CAST (object);
624 switch (GST_EVENT_TYPE (event)) {
625 case GST_EVENT_EOS:
626 result = GST_FLOW_UNEXPECTED;
627 break;
628 case GST_EVENT_NEWSEGMENT:
629 apply_segment (mq, sq, event, &sq->src_segment);
630 /* Applying the segment may have made the queue non-full again, unblock it if needed */
631 gst_data_queue_limits_changed (sq->queue);
632 break;
633 default:
634 break;
635 }
637 GST_DEBUG_OBJECT (mq,
638 "SingleQueue %d : Pushing event %p of type %s",
639 sq->id, event, GST_EVENT_TYPE_NAME (event));
641 gst_pad_push_event (sq->srcpad, event);
642 } else {
643 g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
644 sq->id);
645 }
646 return result;
648 /* ERRORS */
649 }
651 static GstMiniObject *
652 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
653 {
654 GstMiniObject *res;
656 res = item->object;
657 item->object = NULL;
659 return res;
660 }
662 static void
663 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
664 {
665 if (item->object)
666 gst_mini_object_unref (item->object);
667 g_free (item);
668 }
670 /* takes ownership of passed mini object! */
671 static GstMultiQueueItem *
672 gst_multi_queue_item_new (GstMiniObject * object, guint32 curid)
673 {
674 GstMultiQueueItem *item;
676 item = g_new (GstMultiQueueItem, 1);
677 item->object = object;
678 item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
679 item->posid = curid;
681 if (GST_IS_BUFFER (object)) {
682 item->size = GST_BUFFER_SIZE (object);
683 item->duration = GST_BUFFER_DURATION (object);
684 if (item->duration == GST_CLOCK_TIME_NONE)
685 item->duration = 0;
686 item->visible = TRUE;
687 } else {
688 item->size = 0;
689 item->duration = 0;
690 item->visible = FALSE;
691 }
692 return item;
693 }
695 /* Each main loop attempts to push buffers until the return value
696 * is not-linked. not-linked pads are not allowed to push data beyond
697 * any linked pads, so they don't 'rush ahead of the pack'.
698 */
699 static void
700 gst_multi_queue_loop (GstPad * pad)
701 {
702 GstSingleQueue *sq;
703 GstMultiQueueItem *item;
704 GstDataQueueItem *sitem;
705 GstMultiQueue *mq;
706 GstMiniObject *object;
707 guint32 newid;
708 guint32 oldid = -1;
709 GstFlowReturn result;
711 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
712 mq = sq->mqueue;
714 do {
715 GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
717 /* Get something from the queue, blocking until that happens, or we get
718 * flushed */
719 if (!(gst_data_queue_pop (sq->queue, &sitem)))
720 goto out_flushing;
722 item = (GstMultiQueueItem *) sitem;
723 newid = item->posid;
725 /* steal the object and destroy the item */
726 object = gst_multi_queue_item_steal_object (item);
727 gst_multi_queue_item_destroy (item);
729 GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
730 sq->id, newid, oldid);
732 /* If we're not-linked, we do some extra work because we might need to
733 * wait before pushing. If we're linked but there's a gap in the IDs,
734 * or it's the first loop, or we just passed the previous highid,
735 * we might need to wake some sleeping pad up, so there's extra work
736 * there too */
737 if (sq->srcresult == GST_FLOW_NOT_LINKED ||
738 (oldid == -1) || (newid != (oldid + 1)) || oldid > mq->highid) {
739 GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
740 gst_flow_get_name (sq->srcresult));
742 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
744 /* Update the nextid so other threads know when to wake us up */
745 sq->nextid = newid;
747 /* Update the oldid (the last ID we output) for highid tracking */
748 if (oldid != -1)
749 sq->oldid = oldid;
751 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
752 /* Go to sleep until it's time to push this buffer */
754 /* Recompute the highid */
755 compute_high_id (mq);
756 while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
757 GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
758 "newid %u and highid %u", sq->id, newid, mq->highid);
761 /* Wake up all non-linked pads before we sleep */
762 wake_up_next_non_linked (mq);
764 mq->numwaiting++;
765 g_cond_wait (sq->turn, mq->qlock);
766 mq->numwaiting--;
768 GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
769 "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
770 }
772 /* Re-compute the high_id in case someone else pushed */
773 compute_high_id (mq);
774 } else {
775 compute_high_id (mq);
776 /* Wake up all non-linked pads */
777 wake_up_next_non_linked (mq);
778 }
779 /* We're done waiting, we can clear the nextid */
780 sq->nextid = 0;
782 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
783 }
785 GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
786 gst_flow_get_name (sq->srcresult));
788 /* Try to push out the new object */
789 result = gst_single_queue_push_one (mq, sq, object);
790 sq->srcresult = result;
792 if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED)
793 goto out_flushing;
795 GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
796 gst_flow_get_name (sq->srcresult));
798 oldid = newid;
799 }
800 while (TRUE);
802 out_flushing:
803 {
804 /* Need to make sure wake up any sleeping pads when we exit */
805 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
806 compute_high_id (mq);
807 wake_up_next_non_linked (mq);
808 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
810 gst_data_queue_set_flushing (sq->queue, TRUE);
811 gst_pad_pause_task (sq->srcpad);
812 GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
813 "SingleQueue[%d] task paused, reason:%s",
814 sq->id, gst_flow_get_name (sq->srcresult));
815 return;
816 }
817 }
819 /**
820 * gst_multi_queue_chain:
821 *
822 * This is similar to GstQueue's chain function, except:
823 * _ we don't have leak behavioures,
824 * _ we push with a unique id (curid)
825 */
826 static GstFlowReturn
827 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
828 {
829 GstSingleQueue *sq;
830 GstMultiQueue *mq;
831 GstMultiQueueItem *item;
832 GstFlowReturn ret = GST_FLOW_OK;
833 guint32 curid;
834 GstClockTime timestamp, duration;
836 sq = gst_pad_get_element_private (pad);
837 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
839 /* Get a unique incrementing id */
840 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
841 curid = mq->counter++;
842 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
844 GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
845 sq->id, buffer, curid);
847 item = gst_multi_queue_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
849 timestamp = GST_BUFFER_TIMESTAMP (buffer);
850 duration = GST_BUFFER_DURATION (buffer);
852 if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
853 goto flushing;
855 /* update time level, we must do this after pushing the data in the queue so
856 * that we never end up filling the queue first. */
857 apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
859 done:
860 gst_object_unref (mq);
862 return ret;
864 /* ERRORS */
865 flushing:
866 {
867 ret = sq->srcresult;
868 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
869 sq->id, gst_flow_get_name (ret));
870 gst_multi_queue_item_destroy (item);
871 goto done;
872 }
873 }
875 static gboolean
876 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
877 {
878 GstSingleQueue *sq;
880 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
882 if (active) {
883 /* All pads start off not-linked for a smooth kick-off */
884 sq->srcresult = GST_FLOW_NOT_LINKED;
885 } else {
886 sq->srcresult = GST_FLOW_WRONG_STATE;
887 gst_data_queue_flush (sq->queue);
888 }
889 return TRUE;
890 }
892 static gboolean
893 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
894 {
895 GstSingleQueue *sq;
896 GstMultiQueue *mq;
897 guint32 curid;
898 GstMultiQueueItem *item;
899 gboolean res;
900 GstEventType type;
901 GstEvent *sref = NULL;
903 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
904 mq = (GstMultiQueue *) gst_pad_get_parent (pad);
906 type = GST_EVENT_TYPE (event);
908 switch (type) {
909 case GST_EVENT_FLUSH_START:
910 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
911 sq->id);
913 res = gst_pad_push_event (sq->srcpad, event);
915 gst_single_queue_flush (mq, sq, TRUE);
916 goto done;
918 case GST_EVENT_FLUSH_STOP:
919 GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
920 sq->id);
922 res = gst_pad_push_event (sq->srcpad, event);
924 gst_single_queue_flush (mq, sq, FALSE);
925 goto done;
926 case GST_EVENT_NEWSEGMENT:
927 /* take ref because the queue will take ownership and we need the event
928 * afterwards to update the segment */
929 sref = gst_event_ref (event);
930 break;
932 default:
933 if (!(GST_EVENT_IS_SERIALIZED (event))) {
934 res = gst_pad_push_event (sq->srcpad, event);
935 goto done;
936 }
937 break;
938 }
940 /* Get an unique incrementing id */
941 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
942 curid = mq->counter++;
943 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
945 item = gst_multi_queue_item_new ((GstMiniObject *) event, curid);
947 GST_DEBUG_OBJECT (mq,
948 "SingleQueue %d : Enqueuing event %p of type %s with id %d",
949 sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
951 if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
952 goto flushing;
954 /* mark EOS when we received one, we must do that after putting the
955 * buffer in the queue because EOS marks the buffer as filled. No need to take
956 * a lock, the _check_full happens from this thread only, right before pushing
957 * into dataqueue. */
958 switch (type) {
959 case GST_EVENT_EOS:
960 sq->is_eos = TRUE;
961 break;
962 case GST_EVENT_NEWSEGMENT:
963 apply_segment (mq, sq, sref, &sq->sink_segment);
964 gst_event_unref (sref);
965 break;
966 default:
967 break;
968 }
969 done:
970 gst_object_unref (mq);
971 return res;
973 flushing:
974 {
975 GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
976 sq->id, gst_flow_get_name (sq->srcresult));
977 if (sref)
978 gst_event_unref (sref);
979 gst_multi_queue_item_destroy (item);
980 goto done;
981 }
982 }
984 static GstCaps *
985 gst_multi_queue_getcaps (GstPad * pad)
986 {
987 GstSingleQueue *sq = gst_pad_get_element_private (pad);
988 GstPad *otherpad;
989 GstCaps *result;
991 otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
993 GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
995 result = gst_pad_peer_get_caps (otherpad);
996 if (result == NULL)
997 result = gst_caps_new_any ();
999 return result;
1000 }
1002 static GstFlowReturn
1003 gst_multi_queue_bufferalloc (GstPad * pad, guint64 offset, guint size,
1004 GstCaps * caps, GstBuffer ** buf)
1005 {
1006 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1008 return gst_pad_alloc_buffer (sq->srcpad, offset, size, caps, buf);
1009 }
1011 static gboolean
1012 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1013 {
1014 GstMultiQueue *mq;
1015 GstSingleQueue *sq;
1016 gboolean result = FALSE;
1018 sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1019 mq = sq->mqueue;
1021 GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1023 if (active) {
1024 result = gst_single_queue_flush (mq, sq, FALSE);
1025 } else {
1026 result = gst_single_queue_flush (mq, sq, TRUE);
1027 /* make sure streaming finishes */
1028 result |= gst_pad_stop_task (pad);
1029 }
1030 return result;
1031 }
1033 static gboolean
1034 gst_multi_queue_acceptcaps (GstPad * pad, GstCaps * caps)
1035 {
1036 return TRUE;
1037 }
1039 static gboolean
1040 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1041 {
1042 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1044 return gst_pad_push_event (sq->sinkpad, event);
1045 }
1047 static gboolean
1048 gst_multi_queue_src_query (GstPad * pad, GstQuery * query)
1049 {
1050 GstSingleQueue *sq = gst_pad_get_element_private (pad);
1051 GstPad *peerpad;
1052 gboolean res;
1054 /* FIXME, Handle position offset depending on queue size */
1056 /* default handling */
1057 if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1058 goto no_peer;
1060 res = gst_pad_query (peerpad, query);
1062 gst_object_unref (peerpad);
1064 return res;
1066 /* ERRORS */
1067 no_peer:
1068 {
1069 GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1070 return FALSE;
1071 }
1072 }
1074 /*
1075 * Next-non-linked functions
1076 */
1078 /* WITH LOCK TAKEN */
1079 static void
1080 wake_up_next_non_linked (GstMultiQueue * mq)
1081 {
1082 GList *tmp;
1084 /* maybe no-one is waiting */
1085 if (mq->numwaiting < 1)
1086 return;
1088 /* Else figure out which singlequeue(s) need waking up */
1089 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1090 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1092 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1093 if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1094 GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1095 g_cond_signal (sq->turn);
1096 }
1097 }
1098 }
1099 }
1101 /* WITH LOCK TAKEN */
1102 static void
1103 compute_high_id (GstMultiQueue * mq)
1104 {
1105 /* The high-id is either the highest id among the linked pads, or if all
1106 * pads are not-linked, it's the lowest not-linked pad */
1107 GList *tmp;
1108 guint32 lowest = G_MAXUINT32;
1109 guint32 highid = G_MAXUINT32;
1111 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1112 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1114 GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1115 sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1117 if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1118 /* No need to consider queues which are not waiting */
1119 if (sq->nextid == 0) {
1120 GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1121 continue;
1122 }
1124 if (sq->nextid < lowest)
1125 lowest = sq->nextid;
1126 } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1127 /* If we don't have a global highid, or the global highid is lower than
1128 * this single queue's last outputted id, store the queue's one,
1129 * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1130 if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1131 highid = sq->oldid;
1132 }
1133 }
1135 if (highid == G_MAXUINT32 || lowest < highid)
1136 mq->highid = lowest;
1137 else
1138 mq->highid = highid;
1140 GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1141 lowest);
1142 }
1144 #define IS_FILLED(format, value) ((sq->max_size.format) != 0 && \
1145 (sq->max_size.format) <= (value))
1147 /*
1148 * GstSingleQueue functions
1149 */
1150 static void
1151 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1152 {
1153 GstMultiQueue *mq = sq->mqueue;
1154 GList *tmp;
1155 GstDataQueueSize size;
1156 gboolean filled = FALSE;
1158 gst_data_queue_get_level (sq->queue, &size);
1160 GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1162 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1163 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1164 GstSingleQueue *ssq = (GstSingleQueue *) tmp->data;
1165 GstDataQueueSize ssize;
1167 GST_LOG_OBJECT (mq, "Checking Queue %d", ssq->id);
1169 if (gst_data_queue_is_empty (ssq->queue)) {
1170 GST_LOG_OBJECT (mq, "Queue %d is empty", ssq->id);
1171 if (IS_FILLED (visible, size.visible)) {
1172 sq->max_size.visible++;
1173 GST_DEBUG_OBJECT (mq,
1174 "Another queue is empty, bumping single queue %d max visible to %d",
1175 sq->id, sq->max_size.visible);
1176 }
1177 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1178 goto beach;
1179 }
1180 /* check if we reached the hard time/bytes limits */
1181 gst_data_queue_get_level (ssq->queue, &ssize);
1183 GST_DEBUG_OBJECT (mq,
1184 "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1185 G_GUINT64_FORMAT, ssq->id, ssize.visible, sq->max_size.visible,
1186 ssize.bytes, sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1188 /* if this queue is filled completely we must signal overrun */
1189 if (IS_FILLED (bytes, ssize.bytes) || IS_FILLED (time, sq->cur_time)) {
1190 GST_LOG_OBJECT (mq, "Queue %d is filled", ssq->id);
1191 filled = TRUE;
1192 }
1193 }
1194 /* no queues were empty */
1195 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1197 /* Overrun is always forwarded, since this is blocking the upstream element */
1198 if (filled) {
1199 GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1200 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1201 }
1203 beach:
1204 return;
1205 }
1207 static void
1208 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1209 {
1210 gboolean empty = TRUE;
1211 GstMultiQueue *mq = sq->mqueue;
1212 GList *tmp;
1214 GST_LOG_OBJECT (mq,
1215 "Single Queue %d is empty, Checking other single queues", sq->id);
1217 GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1218 for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1219 GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1221 if (gst_data_queue_is_full (sq->queue)) {
1222 GstDataQueueSize size;
1224 gst_data_queue_get_level (sq->queue, &size);
1225 if (IS_FILLED (visible, size.visible)) {
1226 sq->max_size.visible++;
1227 GST_DEBUG_OBJECT (mq,
1228 "queue %d is filled, bumping its max visible to %d", sq->id,
1229 sq->max_size.visible);
1230 gst_data_queue_limits_changed (sq->queue);
1231 }
1232 }
1233 if (!gst_data_queue_is_empty (sq->queue))
1234 empty = FALSE;
1235 }
1236 GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1238 if (empty) {
1239 GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1240 g_signal_emit (G_OBJECT (mq), gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1241 }
1242 }
1244 static gboolean
1245 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1246 guint64 time, GstSingleQueue * sq)
1247 {
1248 gboolean res;
1250 GST_DEBUG ("queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT
1251 "/%" G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1252 sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1254 /* we are always filled on EOS */
1255 if (sq->is_eos)
1256 return TRUE;
1258 /* we never go past the max visible items */
1259 if (IS_FILLED (visible, visible))
1260 return TRUE;
1262 if (sq->cur_time != 0) {
1263 /* if we have valid time in the queue, check */
1264 res = IS_FILLED (time, sq->cur_time);
1265 } else {
1266 /* no valid time, check bytes */
1267 res = IS_FILLED (bytes, bytes);
1268 }
1269 return res;
1270 }
1272 static void
1273 gst_single_queue_free (GstSingleQueue * sq)
1274 {
1275 /* DRAIN QUEUE */
1276 gst_data_queue_flush (sq->queue);
1277 g_object_unref (sq->queue);
1278 g_cond_free (sq->turn);
1279 g_free (sq);
1280 }
1282 static GstSingleQueue *
1283 gst_single_queue_new (GstMultiQueue * mqueue)
1284 {
1285 GstSingleQueue *sq;
1286 gchar *tmp;
1288 sq = g_new0 (GstSingleQueue, 1);
1290 GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1291 sq->id = mqueue->nbqueues++;
1293 /* copy over max_size and extra_size so we don't need to take the lock
1294 * any longer when checking if the queue is full. */
1295 sq->max_size.visible = mqueue->max_size.visible;
1296 sq->max_size.bytes = mqueue->max_size.bytes;
1297 sq->max_size.time = mqueue->max_size.time;
1299 sq->extra_size.visible = mqueue->extra_size.visible;
1300 sq->extra_size.bytes = mqueue->extra_size.bytes;
1301 sq->extra_size.time = mqueue->extra_size.time;
1303 GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1305 GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1307 sq->mqueue = mqueue;
1308 sq->srcresult = GST_FLOW_WRONG_STATE;
1309 sq->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction)
1310 single_queue_check_full, sq);
1311 sq->is_eos = FALSE;
1312 gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1313 gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1315 sq->nextid = 0;
1316 sq->oldid = 0;
1317 sq->turn = g_cond_new ();
1319 /* attach to underrun/overrun signals to handle non-starvation */
1320 g_signal_connect (G_OBJECT (sq->queue), "full",
1321 G_CALLBACK (single_queue_overrun_cb), sq);
1322 g_signal_connect (G_OBJECT (sq->queue), "empty",
1323 G_CALLBACK (single_queue_underrun_cb), sq);
1325 tmp = g_strdup_printf ("sink%d", sq->id);
1326 sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
1327 g_free (tmp);
1329 gst_pad_set_chain_function (sq->sinkpad,
1330 GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1331 gst_pad_set_activatepush_function (sq->sinkpad,
1332 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1333 gst_pad_set_event_function (sq->sinkpad,
1334 GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1335 gst_pad_set_getcaps_function (sq->sinkpad,
1336 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1337 gst_pad_set_bufferalloc_function (sq->sinkpad,
1338 GST_DEBUG_FUNCPTR (gst_multi_queue_bufferalloc));
1339 gst_pad_set_internal_link_function (sq->sinkpad,
1340 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1342 tmp = g_strdup_printf ("src%d", sq->id);
1343 sq->srcpad = gst_pad_new_from_static_template (&srctemplate, tmp);
1344 g_free (tmp);
1346 gst_pad_set_activatepush_function (sq->srcpad,
1347 GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1348 gst_pad_set_acceptcaps_function (sq->srcpad,
1349 GST_DEBUG_FUNCPTR (gst_multi_queue_acceptcaps));
1350 gst_pad_set_getcaps_function (sq->srcpad,
1351 GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1352 gst_pad_set_event_function (sq->srcpad,
1353 GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1354 gst_pad_set_query_function (sq->srcpad,
1355 GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1356 gst_pad_set_internal_link_function (sq->srcpad,
1357 GST_DEBUG_FUNCPTR (gst_multi_queue_get_internal_links));
1359 gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1360 gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1362 gst_pad_set_active (sq->srcpad, TRUE);
1363 gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1365 gst_pad_set_active (sq->sinkpad, TRUE);
1366 gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1368 GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1369 sq->id);
1371 return sq;
1372 }