plugins/elements/gstmultiqueue.c

   1 /* GStreamer
   2  * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>
   3  * Copyright (C) 2007 Jan Schmidt <jan@fluendo.com>
   4  * Copyright (C) 2007 Wim Taymans <wim@fluendo.com>
   5  * Copyright (C) 2011 Sebastian Dröge <sebastian.droege@collabora.co.uk>
   6  *
   7  * gstmultiqueue.c:
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  22  * Boston, MA 02111-1307, USA.
  23  */
  24
  25 /**
  26  * SECTION:element-multiqueue
  27  * @see_also: #GstQueue
  28  *
  29  * <refsect2>
  30  * <para>
  31  * Multiqueue is similar to a normal #GstQueue with the following additional
  32  * features:
  33  * <orderedlist>
  34  * <listitem>
  35  *   <itemizedlist><title>Multiple streamhandling</title>
  36  *   <listitem><para>
  37  *     The element handles queueing data on more than one stream at once. To
  38  *     achieve such a feature it has request sink pads (sink&percnt;d) and
  39  *     'sometimes' src pads (src&percnt;d).
  40  *   </para><para>
  41  *     When requesting a given sinkpad with gst_element_get_request_pad(),
  42  *     the associated srcpad for that stream will be created.
  43  *     Example: requesting sink1 will generate src1.
  44  *   </para></listitem>
  45  *   </itemizedlist>
  46  * </listitem>
  47  * <listitem>
  48  *   <itemizedlist><title>Non-starvation on multiple streams</title>
  49  *   <listitem><para>
  50  *     If more than one stream is used with the element, the streams' queues
  51  *     will be dynamically grown (up to a limit), in order to ensure that no
  52  *     stream is risking data starvation. This guarantees that at any given
  53  *     time there are at least N bytes queued and available for each individual
  54  *     stream.
  55  *   </para><para>
  56  *     If an EOS event comes through a srcpad, the associated queue will be
  57  *     considered as 'not-empty' in the queue-size-growing algorithm.
  58  *   </para></listitem>
  59  *   </itemizedlist>
  60  * </listitem>
  61  * <listitem>
  62  *   <itemizedlist><title>Non-linked srcpads graceful handling</title>
  63  *   <listitem><para>
  64  *     In order to better support dynamic switching between streams, the multiqueue
  65  *     (unlike the current GStreamer queue) continues to push buffers on non-linked
  66  *     pads rather than shutting down.
  67  *   </para><para>
  68  *     In addition, to prevent a non-linked stream from very quickly consuming all
  69  *     available buffers and thus 'racing ahead' of the other streams, the element
  70  *     must ensure that buffers and inlined events for a non-linked stream are pushed
  71  *     in the same order as they were received, relative to the other streams
  72  *     controlled by the element. This means that a buffer cannot be pushed to a
  73  *     non-linked pad any sooner than buffers in any other stream which were received
  74  *     before it.
  75  *   </para></listitem>
  76  *   </itemizedlist>
  77  * </listitem>
  78  * </orderedlist>
  79  * </para>
  80  * <para>
  81  *   Data is queued until one of the limits specified by the
  82  *   #GstMultiQueue:max-size-buffers, #GstMultiQueue:max-size-bytes and/or
  83  *   #GstMultiQueue:max-size-time properties has been reached. Any attempt to push
  84  *   more buffers into the queue will block the pushing thread until more space
  85  *   becomes available. #GstMultiQueue:extra-size-buffers,
  86  * </para>
  87  * <para>
  88  *   #GstMultiQueue:extra-size-bytes and #GstMultiQueue:extra-size-time are
  89  *   currently unused.
  90  * </para>
  91  * <para>
  92  *   The default queue size limits are 5 buffers, 10MB of data, or
  93  *   two second worth of data, whichever is reached first. Note that the number
  94  *   of buffers will dynamically grow depending on the fill level of
  95  *   other queues.
  96  * </para>
  97  * <para>
  98  *   The #GstMultiQueue::underrun signal is emitted when all of the queues
  99  *   are empty. The #GstMultiQueue::overrun signal is emitted when one of the
 100  *   queues is filled.
 101  *   Both signals are emitted from the context of the streaming thread.
 102  * </para>
 103  * </refsect2>
 104  *
 105  * Last reviewed on 2008-01-25 (0.10.17)
 106  */
 107
 108 #ifdef HAVE_CONFIG_H
 109 #  include "config.h"
 110 #endif
 111
 112 #include <gst/gst.h>
 113 #include <stdio.h>
 114 #include "gstmultiqueue.h"
 115
 116 /**
 117  * GstSingleQueue:
 118  * @sinkpad: associated sink #GstPad
 119  * @srcpad: associated source #GstPad
 120  *
 121  * Structure containing all information and properties about
 122  * a single queue.
 123  */
 124 typedef struct _GstSingleQueue GstSingleQueue;
 125
 126 struct _GstSingleQueue
 127 {
 128   /* unique identifier of the queue */
 129   guint id;
 130
 131   GstMultiQueue *mqueue;
 132
 133   GstPad *sinkpad;
 134   GstPad *srcpad;
 135
 136   /* flowreturn of previous srcpad push */
 137   GstFlowReturn srcresult;
 138
 139   /* segments */
 140   GstSegment sink_segment;
 141   GstSegment src_segment;
 142
 143   /* position of src/sink */
 144   GstClockTime sinktime, srctime;
 145   /* TRUE if either position needs to be recalculated */
 146   gboolean sink_tainted, src_tainted;
 147
 148   /* queue of data */
 149   GstDataQueue *queue;
 150   GstDataQueueSize max_size, extra_size;
 151   GstClockTime cur_time;
 152   gboolean is_eos;
 153   gboolean flushing;
 154
 155   /* Protected by global lock */
 156   guint32 nextid;               /* ID of the next object waiting to be pushed */
 157   guint32 oldid;                /* ID of the last object pushed (last in a series) */
 158   guint32 last_oldid;           /* Previously observed old_id, reset to MAXUINT32 on flush */
 159   GCond *turn;                  /* SingleQueue turn waiting conditional */
 160 };
 161
 162
 163 /* Extension of GstDataQueueItem structure for our usage */
 164 typedef struct _GstMultiQueueItem GstMultiQueueItem;
 165
 166 struct _GstMultiQueueItem
 167 {
 168   GstMiniObject *object;
 169   guint size;
 170   guint64 duration;
 171   gboolean visible;
 172
 173   GDestroyNotify destroy;
 174   guint32 posid;
 175 };
 176
 177 static GstSingleQueue *gst_single_queue_new (GstMultiQueue * mqueue, gint id);
 178 static void gst_single_queue_free (GstSingleQueue * squeue);
 179
 180 static void wake_up_next_non_linked (GstMultiQueue * mq);
 181 static void compute_high_id (GstMultiQueue * mq);
 182 static void single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq);
 183 static void single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq);
 184
 185 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",
 186     GST_PAD_SINK,
 187     GST_PAD_REQUEST,
 188     GST_STATIC_CAPS_ANY);
 189
 190 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
 191     GST_PAD_SRC,
 192     GST_PAD_SOMETIMES,
 193     GST_STATIC_CAPS_ANY);
 194
 195 GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);
 196 #define GST_CAT_DEFAULT (multi_queue_debug)
 197
 198 /* Signals and args */
 199 enum
 200 {
 201   SIGNAL_UNDERRUN,
 202   SIGNAL_OVERRUN,
 203   LAST_SIGNAL
 204 };
 205
 206 /* default limits, we try to keep up to 2 seconds of data and if there is not
 207  * time, up to 10 MB. The number of buffers is dynamically scaled to make sure
 208  * there is data in the queues. Normally, the byte and time limits are not hit
 209  * in theses conditions. */
 210 #define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */
 211 #define DEFAULT_MAX_SIZE_BUFFERS 5
 212 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND
 213
 214 /* second limits. When we hit one of the above limits we are probably dealing
 215  * with a badly muxed file and we scale the limits to these emergency values.
 216  * This is currently not yet implemented.
 217  * Since we dynamically scale the queue buffer size up to the limits but avoid
 218  * going above the max-size-buffers when we can, we don't really need this
 219  * aditional extra size. */
 220 #define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024       /* 10 MB */
 221 #define DEFAULT_EXTRA_SIZE_BUFFERS 5
 222 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND
 223
 224 #define DEFAULT_USE_BUFFERING FALSE
 225 #define DEFAULT_LOW_PERCENT   10
 226 #define DEFAULT_HIGH_PERCENT  99
 227
 228 enum
 229 {
 230   PROP_0,
 231   PROP_EXTRA_SIZE_BYTES,
 232   PROP_EXTRA_SIZE_BUFFERS,
 233   PROP_EXTRA_SIZE_TIME,
 234   PROP_MAX_SIZE_BYTES,
 235   PROP_MAX_SIZE_BUFFERS,
 236   PROP_MAX_SIZE_TIME,
 237   PROP_USE_BUFFERING,
 238   PROP_LOW_PERCENT,
 239   PROP_HIGH_PERCENT,
 240   PROP_LAST
 241 };
 242
 243 #define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START {                          \
 244   g_mutex_lock (q->qlock);                                              \
 245 } G_STMT_END
 246
 247 #define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                        \
 248   g_mutex_unlock (q->qlock);                                            \
 249 } G_STMT_END
 250
 251 static void gst_multi_queue_finalize (GObject * object);
 252 static void gst_multi_queue_set_property (GObject * object,
 253     guint prop_id, const GValue * value, GParamSpec * pspec);
 254 static void gst_multi_queue_get_property (GObject * object,
 255     guint prop_id, GValue * value, GParamSpec * pspec);
 256
 257 static GstPad *gst_multi_queue_request_new_pad (GstElement * element,
 258     GstPadTemplate * temp, const gchar * name, const GstCaps * caps);
 259 static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);
 260 static GstStateChangeReturn gst_multi_queue_change_state (GstElement *
 261     element, GstStateChange transition);
 262
 263 static void gst_multi_queue_loop (GstPad * pad);
 264
 265 #define _do_init \
 266   GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");
 267 #define gst_multi_queue_parent_class parent_class
 268 G_DEFINE_TYPE_WITH_CODE (GstMultiQueue, gst_multi_queue, GST_TYPE_ELEMENT,
 269     _do_init);
 270
 271 static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };
 272
 273 static void
 274 gst_multi_queue_class_init (GstMultiQueueClass * klass)
 275 {
 276   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 277   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 278
 279   gobject_class->set_property = gst_multi_queue_set_property;
 280   gobject_class->get_property = gst_multi_queue_get_property;
 281
 282   /* SIGNALS */
 283
 284   /**
 285    * GstMultiQueue::underrun:
 286    * @multiqueue: the multqueue instance
 287    *
 288    * This signal is emitted from the streaming thread when there is
 289    * no data in any of the queues inside the multiqueue instance (underrun).
 290    *
 291    * This indicates either starvation or EOS from the upstream data sources.
 292    */
 293   gst_multi_queue_signals[SIGNAL_UNDERRUN] =
 294       g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 295       G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,
 296       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 297
 298   /**
 299    * GstMultiQueue::overrun:
 300    * @multiqueue: the multiqueue instance
 301    *
 302    * Reports that one of the queues in the multiqueue is full (overrun).
 303    * A queue is full if the total amount of data inside it (num-buffers, time,
 304    * size) is higher than the boundary values which can be set through the
 305    * GObject properties.
 306    *
 307    * This can be used as an indicator of pre-roll.
 308    */
 309   gst_multi_queue_signals[SIGNAL_OVERRUN] =
 310       g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 311       G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,
 312       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 313
 314   /* PROPERTIES */
 315
 316   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
 317       g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
 318           "Max. amount of data in the queue (bytes, 0=disable)",
 319           0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
 320           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 321   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BUFFERS,
 322       g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
 323           "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
 324           DEFAULT_MAX_SIZE_BUFFERS,
 325           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 326   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
 327       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 328           "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
 329           DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 330
 331   g_object_class_install_property (gobject_class, PROP_EXTRA_SIZE_BYTES,
 332       g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",
 333           "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)"
 334           " (NOT IMPLEMENTED)",
 335           0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES,
 336           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 337   g_object_class_install_property (gobject_class, PROP_EXTRA_SIZE_BUFFERS,
 338       g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",
 339           "Amount of buffers the queues can grow if one of them is empty (0=disable)"
 340           " (NOT IMPLEMENTED)",
 341           0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS,
 342           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 343   g_object_class_install_property (gobject_class, PROP_EXTRA_SIZE_TIME,
 344       g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",
 345           "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)"
 346           " (NOT IMPLEMENTED)",
 347           0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME,
 348           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 349
 350   /**
 351    * GstMultiQueue:use-buffering
 352    *
 353    * Enable the buffering option in multiqueue so that BUFFERING messages are
 354    * emited based on low-/high-percent thresholds.
 355    *
 356    * Since: 0.10.26
 357    */
 358   g_object_class_install_property (gobject_class, PROP_USE_BUFFERING,
 359       g_param_spec_boolean ("use-buffering", "Use buffering",
 360           "Emit GST_MESSAGE_BUFFERING based on low-/high-percent thresholds",
 361           DEFAULT_USE_BUFFERING, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 362   /**
 363    * GstMultiQueue:low-percent
 364    *
 365    * Low threshold percent for buffering to start.
 366    *
 367    * Since: 0.10.26
 368    */
 369   g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
 370       g_param_spec_int ("low-percent", "Low percent",
 371           "Low threshold for buffering to start", 0, 100,
 372           DEFAULT_LOW_PERCENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 373   /**
 374    * GstMultiQueue:high-percent
 375    *
 376    * High threshold percent for buffering to finish.
 377    *
 378    * Since: 0.10.26
 379    */
 380   g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
 381       g_param_spec_int ("high-percent", "High percent",
 382           "High threshold for buffering to finish", 0, 100,
 383           DEFAULT_HIGH_PERCENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 384
 385
 386   gobject_class->finalize = gst_multi_queue_finalize;
 387
 388   gst_element_class_set_details_simple (gstelement_class,
 389       "MultiQueue",
 390       "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");
 391   gst_element_class_add_pad_template (gstelement_class,
 392       gst_static_pad_template_get (&sinktemplate));
 393   gst_element_class_add_pad_template (gstelement_class,
 394       gst_static_pad_template_get (&srctemplate));
 395
 396   gstelement_class->request_new_pad =
 397       GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);
 398   gstelement_class->release_pad =
 399       GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);
 400   gstelement_class->change_state =
 401       GST_DEBUG_FUNCPTR (gst_multi_queue_change_state);
 402 }
 403
 404 static void
 405 gst_multi_queue_init (GstMultiQueue * mqueue)
 406 {
 407   mqueue->nbqueues = 0;
 408   mqueue->queues = NULL;
 409
 410   mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;
 411   mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;
 412   mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;
 413
 414   mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;
 415   mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;
 416   mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
 417
 418   mqueue->use_buffering = DEFAULT_USE_BUFFERING;
 419   mqueue->low_percent = DEFAULT_LOW_PERCENT;
 420   mqueue->high_percent = DEFAULT_HIGH_PERCENT;
 421
 422   mqueue->counter = 1;
 423   mqueue->highid = -1;
 424
 425   mqueue->qlock = g_mutex_new ();
 426 }
 427
 428 static void
 429 gst_multi_queue_finalize (GObject * object)
 430 {
 431   GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);
 432
 433   g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);
 434   g_list_free (mqueue->queues);
 435   mqueue->queues = NULL;
 436   mqueue->queues_cookie++;
 437
 438   /* free/unref instance data */
 439   g_mutex_free (mqueue->qlock);
 440
 441   G_OBJECT_CLASS (parent_class)->finalize (object);
 442 }
 443
 444 #define SET_CHILD_PROPERTY(mq,format) G_STMT_START {            \
 445     GList * tmp = mq->queues;                                   \
 446     while (tmp) {                                               \
 447       GstSingleQueue *q = (GstSingleQueue*)tmp->data;           \
 448       q->max_size.format = mq->max_size.format;                 \
 449       tmp = g_list_next(tmp);                                   \
 450     };                                                          \
 451 } G_STMT_END
 452
 453 static void
 454 gst_multi_queue_set_property (GObject * object, guint prop_id,
 455     const GValue * value, GParamSpec * pspec)
 456 {
 457   GstMultiQueue *mq = GST_MULTI_QUEUE (object);
 458
 459   switch (prop_id) {
 460     case PROP_MAX_SIZE_BYTES:
 461       GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 462       mq->max_size.bytes = g_value_get_uint (value);
 463       SET_CHILD_PROPERTY (mq, bytes);
 464       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 465       break;
 466     case PROP_MAX_SIZE_BUFFERS:
 467       GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 468       mq->max_size.visible = g_value_get_uint (value);
 469       SET_CHILD_PROPERTY (mq, visible);
 470       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 471       break;
 472     case PROP_MAX_SIZE_TIME:
 473       GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 474       mq->max_size.time = g_value_get_uint64 (value);
 475       SET_CHILD_PROPERTY (mq, time);
 476       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 477       break;
 478     case PROP_EXTRA_SIZE_BYTES:
 479       mq->extra_size.bytes = g_value_get_uint (value);
 480       break;
 481     case PROP_EXTRA_SIZE_BUFFERS:
 482       mq->extra_size.visible = g_value_get_uint (value);
 483       break;
 484     case PROP_EXTRA_SIZE_TIME:
 485       mq->extra_size.time = g_value_get_uint64 (value);
 486       break;
 487     case PROP_USE_BUFFERING:
 488       mq->use_buffering = g_value_get_boolean (value);
 489       break;
 490     case PROP_LOW_PERCENT:
 491       mq->low_percent = g_value_get_int (value);
 492       break;
 493     case PROP_HIGH_PERCENT:
 494       mq->high_percent = g_value_get_int (value);
 495       break;
 496     default:
 497       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 498       break;
 499   }
 500 }
 501
 502 static void
 503 gst_multi_queue_get_property (GObject * object, guint prop_id,
 504     GValue * value, GParamSpec * pspec)
 505 {
 506   GstMultiQueue *mq = GST_MULTI_QUEUE (object);
 507
 508   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 509
 510   switch (prop_id) {
 511     case PROP_EXTRA_SIZE_BYTES:
 512       g_value_set_uint (value, mq->extra_size.bytes);
 513       break;
 514     case PROP_EXTRA_SIZE_BUFFERS:
 515       g_value_set_uint (value, mq->extra_size.visible);
 516       break;
 517     case PROP_EXTRA_SIZE_TIME:
 518       g_value_set_uint64 (value, mq->extra_size.time);
 519       break;
 520     case PROP_MAX_SIZE_BYTES:
 521       g_value_set_uint (value, mq->max_size.bytes);
 522       break;
 523     case PROP_MAX_SIZE_BUFFERS:
 524       g_value_set_uint (value, mq->max_size.visible);
 525       break;
 526     case PROP_MAX_SIZE_TIME:
 527       g_value_set_uint64 (value, mq->max_size.time);
 528       break;
 529     case PROP_USE_BUFFERING:
 530       g_value_set_boolean (value, mq->use_buffering);
 531       break;
 532     case PROP_LOW_PERCENT:
 533       g_value_set_int (value, mq->low_percent);
 534       break;
 535     case PROP_HIGH_PERCENT:
 536       g_value_set_int (value, mq->high_percent);
 537       break;
 538     default:
 539       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 540       break;
 541   }
 542
 543   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 544 }
 545
 546 static GstIterator *
 547 gst_multi_queue_iterate_internal_links (GstPad * pad)
 548 {
 549   GstIterator *it = NULL;
 550   GstPad *opad;
 551   GstSingleQueue *squeue;
 552   GstMultiQueue *mq = GST_MULTI_QUEUE (gst_pad_get_parent (pad));
 553   GValue val = { 0, };
 554
 555   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 556   squeue = gst_pad_get_element_private (pad);
 557   if (!squeue)
 558     goto out;
 559
 560   if (squeue->sinkpad == pad)
 561     opad = gst_object_ref (squeue->srcpad);
 562   else if (squeue->srcpad == pad)
 563     opad = gst_object_ref (squeue->sinkpad);
 564   else
 565     goto out;
 566
 567   g_value_init (&val, GST_TYPE_PAD);
 568   g_value_set_object (&val, opad);
 569   it = gst_iterator_new_single (GST_TYPE_PAD, &val);
 570   g_value_unset (&val);
 571
 572   gst_object_unref (opad);
 573
 574 out:
 575   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 576   gst_object_unref (mq);
 577
 578   return it;
 579 }
 580
 581
 582 /*
 583  * GstElement methods
 584  */
 585
 586 static GstPad *
 587 gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,
 588     const gchar * name, const GstCaps * caps)
 589 {
 590   GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
 591   GstSingleQueue *squeue;
 592   gint temp_id = -1;
 593
 594   if (name) {
 595     sscanf (name + 4, "%d", &temp_id);
 596     GST_LOG_OBJECT (element, "name : %s (id %d)", GST_STR_NULL (name), temp_id);
 597   }
 598
 599   /* Create a new single queue, add the sink and source pad and return the sink pad */
 600   squeue = gst_single_queue_new (mqueue, temp_id);
 601
 602   GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",
 603       GST_DEBUG_PAD_NAME (squeue->sinkpad));
 604
 605   return squeue ? squeue->sinkpad : NULL;
 606 }
 607
 608 static void
 609 gst_multi_queue_release_pad (GstElement * element, GstPad * pad)
 610 {
 611   GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
 612   GstSingleQueue *sq = NULL;
 613   GList *tmp;
 614
 615   GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));
 616
 617   GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
 618   /* Find which single queue it belongs to, knowing that it should be a sinkpad */
 619   for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
 620     sq = (GstSingleQueue *) tmp->data;
 621
 622     if (sq->sinkpad == pad)
 623       break;
 624   }
 625
 626   if (!tmp) {
 627     GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");
 628     GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
 629     return;
 630   }
 631
 632   /* FIXME: The removal of the singlequeue should probably not happen until it
 633    * finishes draining */
 634
 635   /* remove it from the list */
 636   mqueue->queues = g_list_delete_link (mqueue->queues, tmp);
 637   mqueue->queues_cookie++;
 638
 639   /* FIXME : recompute next-non-linked */
 640   GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
 641
 642   /* delete SingleQueue */
 643   gst_data_queue_set_flushing (sq->queue, TRUE);
 644
 645   gst_pad_set_active (sq->srcpad, FALSE);
 646   gst_pad_set_active (sq->sinkpad, FALSE);
 647   gst_pad_set_element_private (sq->srcpad, NULL);
 648   gst_pad_set_element_private (sq->sinkpad, NULL);
 649   gst_element_remove_pad (element, sq->srcpad);
 650   gst_element_remove_pad (element, sq->sinkpad);
 651   gst_single_queue_free (sq);
 652 }
 653
 654 static GstStateChangeReturn
 655 gst_multi_queue_change_state (GstElement * element, GstStateChange transition)
 656 {
 657   GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);
 658   GstSingleQueue *sq = NULL;
 659   GstStateChangeReturn result;
 660
 661   switch (transition) {
 662     case GST_STATE_CHANGE_READY_TO_PAUSED:{
 663       GList *tmp;
 664
 665       /* Set all pads to non-flushing */
 666       GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
 667       for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
 668         sq = (GstSingleQueue *) tmp->data;
 669         sq->flushing = FALSE;
 670       }
 671       GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
 672       break;
 673     }
 674     case GST_STATE_CHANGE_PAUSED_TO_READY:{
 675       GList *tmp;
 676
 677       /* Un-wait all waiting pads */
 678       GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
 679       for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
 680         sq = (GstSingleQueue *) tmp->data;
 681         sq->flushing = TRUE;
 682         g_cond_signal (sq->turn);
 683       }
 684       GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
 685       break;
 686     }
 687     default:
 688       break;
 689   }
 690
 691   result = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 692
 693   switch (transition) {
 694     default:
 695       break;
 696   }
 697
 698   return result;
 699
 700
 701
 702 }
 703
 704 static gboolean
 705 gst_single_queue_flush (GstMultiQueue * mq, GstSingleQueue * sq, gboolean flush)
 706 {
 707   gboolean result;
 708
 709   GST_DEBUG_OBJECT (mq, "flush %s queue %d", (flush ? "start" : "stop"),
 710       sq->id);
 711
 712   if (flush) {
 713     sq->srcresult = GST_FLOW_WRONG_STATE;
 714     gst_data_queue_set_flushing (sq->queue, TRUE);
 715
 716     sq->flushing = TRUE;
 717
 718     /* wake up non-linked task */
 719     GST_LOG_OBJECT (mq, "SingleQueue %d : waking up eventually waiting task",
 720         sq->id);
 721     GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 722     g_cond_signal (sq->turn);
 723     GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 724
 725     GST_LOG_OBJECT (mq, "SingleQueue %d : pausing task", sq->id);
 726     result = gst_pad_pause_task (sq->srcpad);
 727     sq->sink_tainted = sq->src_tainted = TRUE;
 728   } else {
 729     gst_data_queue_flush (sq->queue);
 730     gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
 731     gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
 732     /* All pads start off not-linked for a smooth kick-off */
 733     sq->srcresult = GST_FLOW_OK;
 734     sq->cur_time = 0;
 735     sq->max_size.visible = mq->max_size.visible;
 736     sq->is_eos = FALSE;
 737     sq->nextid = 0;
 738     sq->oldid = 0;
 739     sq->last_oldid = G_MAXUINT32;
 740     gst_data_queue_set_flushing (sq->queue, FALSE);
 741
 742     sq->flushing = FALSE;
 743
 744     GST_LOG_OBJECT (mq, "SingleQueue %d : starting task", sq->id);
 745     result =
 746         gst_pad_start_task (sq->srcpad, (GstTaskFunction) gst_multi_queue_loop,
 747         sq->srcpad);
 748   }
 749   return result;
 750 }
 751
 752 static void
 753 update_buffering (GstMultiQueue * mq, GstSingleQueue * sq)
 754 {
 755   GstDataQueueSize size;
 756   gint percent, tmp;
 757   gboolean post = FALSE;
 758
 759   /* nothing to dowhen we are not in buffering mode */
 760   if (!mq->use_buffering)
 761     return;
 762
 763   gst_data_queue_get_level (sq->queue, &size);
 764
 765   GST_DEBUG_OBJECT (mq,
 766       "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
 767       G_GUINT64_FORMAT, sq->id, size.visible, sq->max_size.visible,
 768       size.bytes, sq->max_size.bytes, sq->cur_time, sq->max_size.time);
 769
 770   /* get bytes and time percentages and take the max */
 771   if (sq->is_eos) {
 772     percent = 100;
 773   } else {
 774     percent = 0;
 775     if (sq->max_size.time > 0) {
 776       tmp = (sq->cur_time * 100) / sq->max_size.time;
 777       percent = MAX (percent, tmp);
 778     }
 779     if (sq->max_size.bytes > 0) {
 780       tmp = (size.bytes * 100) / sq->max_size.bytes;
 781       percent = MAX (percent, tmp);
 782     }
 783   }
 784
 785   if (mq->buffering) {
 786     post = TRUE;
 787     if (percent >= mq->high_percent) {
 788       mq->buffering = FALSE;
 789     }
 790     /* make sure it increases */
 791     percent = MAX (mq->percent, percent);
 792
 793     if (percent == mq->percent)
 794       /* don't post if nothing changed */
 795       post = FALSE;
 796     else
 797       /* else keep last value we posted */
 798       mq->percent = percent;
 799   } else {
 800     if (percent < mq->low_percent) {
 801       mq->buffering = TRUE;
 802       mq->percent = percent;
 803       post = TRUE;
 804     }
 805   }
 806   if (post) {
 807     GstMessage *message;
 808
 809     /* scale to high percent so that it becomes the 100% mark */
 810     percent = percent * 100 / mq->high_percent;
 811     /* clip */
 812     if (percent > 100)
 813       percent = 100;
 814
 815     GST_DEBUG_OBJECT (mq, "buffering %d percent", percent);
 816     message = gst_message_new_buffering (GST_OBJECT_CAST (mq), percent);
 817
 818     gst_element_post_message (GST_ELEMENT_CAST (mq), message);
 819   } else {
 820     GST_DEBUG_OBJECT (mq, "filled %d percent", percent);
 821   }
 822 }
 823
 824 /* calculate the diff between running time on the sink and src of the queue.
 825  * This is the total amount of time in the queue.
 826  * WITH LOCK TAKEN */
 827 static void
 828 update_time_level (GstMultiQueue * mq, GstSingleQueue * sq)
 829 {
 830   gint64 sink_time, src_time;
 831
 832   if (sq->sink_tainted) {
 833     sink_time = sq->sinktime =
 834         gst_segment_to_running_time (&sq->sink_segment, GST_FORMAT_TIME,
 835         sq->sink_segment.last_stop);
 836
 837     if (G_UNLIKELY (sink_time != GST_CLOCK_TIME_NONE))
 838       /* if we have a time, we become untainted and use the time */
 839       sq->sink_tainted = FALSE;
 840   } else
 841     sink_time = sq->sinktime;
 842
 843   if (sq->src_tainted) {
 844     src_time = sq->srctime =
 845         gst_segment_to_running_time (&sq->src_segment, GST_FORMAT_TIME,
 846         sq->src_segment.last_stop);
 847     /* if we have a time, we become untainted and use the time */
 848     if (G_UNLIKELY (src_time != GST_CLOCK_TIME_NONE))
 849       sq->src_tainted = FALSE;
 850   } else
 851     src_time = sq->srctime;
 852
 853   GST_DEBUG_OBJECT (mq,
 854       "queue %d, sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT, sq->id,
 855       GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
 856
 857   /* This allows for streams with out of order timestamping - sometimes the
 858    * emerging timestamp is later than the arriving one(s) */
 859   if (G_LIKELY (sink_time != -1 && src_time != -1 && sink_time > src_time))
 860     sq->cur_time = sink_time - src_time;
 861   else
 862     sq->cur_time = 0;
 863
 864   /* updating the time level can change the buffering state */
 865   update_buffering (mq, sq);
 866
 867   return;
 868 }
 869
 870 /* take a NEWSEGMENT event and apply the values to segment, updating the time
 871  * level of queue. */
 872 static void
 873 apply_segment (GstMultiQueue * mq, GstSingleQueue * sq, GstEvent * event,
 874     GstSegment * segment)
 875 {
 876   gboolean update;
 877   GstFormat format;
 878   gdouble rate, arate;
 879   gint64 start, stop, time;
 880
 881   gst_event_parse_new_segment (event, &update, &rate, &arate,
 882       &format, &start, &stop, &time);
 883
 884   /* now configure the values, we use these to track timestamps on the
 885    * sinkpad. */
 886   if (format != GST_FORMAT_TIME) {
 887     /* non-time format, pretent the current time segment is closed with a
 888      * 0 start and unknown stop time. */
 889     update = FALSE;
 890     format = GST_FORMAT_TIME;
 891     start = 0;
 892     stop = -1;
 893     time = 0;
 894   }
 895
 896   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 897
 898   gst_segment_set_newsegment (segment, update,
 899       rate, arate, format, start, stop, time);
 900
 901   if (segment == &sq->sink_segment)
 902     sq->sink_tainted = TRUE;
 903   else
 904     sq->src_tainted = TRUE;
 905
 906   GST_DEBUG_OBJECT (mq,
 907       "queue %d, configured NEWSEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
 908
 909   /* segment can update the time level of the queue */
 910   update_time_level (mq, sq);
 911
 912   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 913 }
 914
 915 /* take a buffer and update segment, updating the time level of the queue. */
 916 static void
 917 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
 918     GstClockTime duration, GstSegment * segment)
 919 {
 920   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 921
 922   /* if no timestamp is set, assume it's continuous with the previous
 923    * time */
 924   if (timestamp == GST_CLOCK_TIME_NONE)
 925     timestamp = segment->last_stop;
 926
 927   /* add duration */
 928   if (duration != GST_CLOCK_TIME_NONE)
 929     timestamp += duration;
 930
 931   GST_DEBUG_OBJECT (mq, "queue %d, last_stop updated to %" GST_TIME_FORMAT,
 932       sq->id, GST_TIME_ARGS (timestamp));
 933
 934   gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
 935
 936   if (segment == &sq->sink_segment)
 937     sq->sink_tainted = TRUE;
 938   else
 939     sq->src_tainted = TRUE;
 940
 941   /* calc diff with other end */
 942   update_time_level (mq, sq);
 943   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 944 }
 945
 946 static GstFlowReturn
 947 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
 948     GstMiniObject * object)
 949 {
 950   GstFlowReturn result = GST_FLOW_OK;
 951
 952   if (GST_IS_BUFFER (object)) {
 953     GstBuffer *buffer;
 954     GstClockTime timestamp, duration;
 955 #if 0
 956     GstCaps *caps;
 957 #endif
 958
 959     buffer = GST_BUFFER_CAST (object);
 960     timestamp = GST_BUFFER_TIMESTAMP (buffer);
 961     duration = GST_BUFFER_DURATION (buffer);
 962 #if 0
 963     caps = GST_BUFFER_CAPS (buffer);
 964 #endif
 965
 966     apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
 967
 968     /* Applying the buffer may have made the queue non-full again, unblock it if needed */
 969     gst_data_queue_limits_changed (sq->queue);
 970
 971     GST_DEBUG_OBJECT (mq,
 972         "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
 973         sq->id, buffer, GST_TIME_ARGS (timestamp));
 974
 975 #if 0
 976     /* Set caps on pad before pushing, this avoids core calling the acceptcaps
 977      * function on the srcpad, which will call acceptcaps upstream, which might
 978      * not accept these caps (anymore). */
 979     if (caps && caps != GST_PAD_CAPS (sq->srcpad))
 980       gst_pad_set_caps (sq->srcpad, caps);
 981 #endif
 982
 983     result = gst_pad_push (sq->srcpad, buffer);
 984   } else if (GST_IS_EVENT (object)) {
 985     GstEvent *event;
 986
 987     event = GST_EVENT_CAST (object);
 988
 989     switch (GST_EVENT_TYPE (event)) {
 990       case GST_EVENT_EOS:
 991         result = GST_FLOW_UNEXPECTED;
 992         break;
 993       case GST_EVENT_NEWSEGMENT:
 994         apply_segment (mq, sq, event, &sq->src_segment);
 995         /* Applying the segment may have made the queue non-full again, unblock it if needed */
 996         gst_data_queue_limits_changed (sq->queue);
 997         break;
 998       default:
 999         break;
1000     }
1001
1002     GST_DEBUG_OBJECT (mq,
1003         "SingleQueue %d : Pushing event %p of type %s",
1004         sq->id, event, GST_EVENT_TYPE_NAME (event));
1005
1006     gst_pad_push_event (sq->srcpad, event);
1007   } else {
1008     g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
1009         sq->id);
1010   }
1011   return result;
1012
1013   /* ERRORS */
1014 }
1015
1016 static GstMiniObject *
1017 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
1018 {
1019   GstMiniObject *res;
1020
1021   res = item->object;
1022   item->object = NULL;
1023
1024   return res;
1025 }
1026
1027 static void
1028 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
1029 {
1030   if (item->object)
1031     gst_mini_object_unref (item->object);
1032   g_slice_free (GstMultiQueueItem, item);
1033 }
1034
1035 /* takes ownership of passed mini object! */
1036 static GstMultiQueueItem *
1037 gst_multi_queue_buffer_item_new (GstMiniObject * object, guint32 curid)
1038 {
1039   GstMultiQueueItem *item;
1040
1041   item = g_slice_new (GstMultiQueueItem);
1042   item->object = object;
1043   item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
1044   item->posid = curid;
1045
1046   item->size = gst_buffer_get_size (GST_BUFFER_CAST (object));
1047   item->duration = GST_BUFFER_DURATION (object);
1048   if (item->duration == GST_CLOCK_TIME_NONE)
1049     item->duration = 0;
1050   item->visible = TRUE;
1051   return item;
1052 }
1053
1054 static GstMultiQueueItem *
1055 gst_multi_queue_event_item_new (GstMiniObject * object, guint32 curid)
1056 {
1057   GstMultiQueueItem *item;
1058
1059   item = g_slice_new (GstMultiQueueItem);
1060   item->object = object;
1061   item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
1062   item->posid = curid;
1063
1064   item->size = 0;
1065   item->duration = 0;
1066   item->visible = FALSE;
1067   return item;
1068 }
1069
1070 /* Each main loop attempts to push buffers until the return value
1071  * is not-linked. not-linked pads are not allowed to push data beyond
1072  * any linked pads, so they don't 'rush ahead of the pack'.
1073  */
1074 static void
1075 gst_multi_queue_loop (GstPad * pad)
1076 {
1077   GstSingleQueue *sq;
1078   GstMultiQueueItem *item;
1079   GstDataQueueItem *sitem;
1080   GstMultiQueue *mq;
1081   GstMiniObject *object = NULL;
1082   guint32 newid;
1083   GstFlowReturn result;
1084
1085   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1086   mq = sq->mqueue;
1087
1088   GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
1089
1090   if (sq->flushing)
1091     goto out_flushing;
1092
1093   /* Get something from the queue, blocking until that happens, or we get
1094    * flushed */
1095   if (!(gst_data_queue_pop (sq->queue, &sitem)))
1096     goto out_flushing;
1097
1098   item = (GstMultiQueueItem *) sitem;
1099   newid = item->posid;
1100
1101   /* steal the object and destroy the item */
1102   object = gst_multi_queue_item_steal_object (item);
1103   gst_multi_queue_item_destroy (item);
1104
1105   GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
1106       sq->id, newid, sq->last_oldid);
1107
1108   /* If we're not-linked, we do some extra work because we might need to
1109    * wait before pushing. If we're linked but there's a gap in the IDs,
1110    * or it's the first loop, or we just passed the previous highid,
1111    * we might need to wake some sleeping pad up, so there's extra work
1112    * there too */
1113   if (sq->srcresult == GST_FLOW_NOT_LINKED ||
1114       (sq->last_oldid == G_MAXUINT32) || (newid != (sq->last_oldid + 1)) ||
1115       sq->last_oldid > mq->highid) {
1116     GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
1117         gst_flow_get_name (sq->srcresult));
1118
1119     GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1120
1121     /* Check again if we're flushing after the lock is taken,
1122      * the flush flag might have been changed in the meantime */
1123     if (sq->flushing) {
1124       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1125       goto out_flushing;
1126     }
1127
1128     /* Update the nextid so other threads know when to wake us up */
1129     sq->nextid = newid;
1130
1131     /* Update the oldid (the last ID we output) for highid tracking */
1132     if (sq->last_oldid != G_MAXUINT32)
1133       sq->oldid = sq->last_oldid;
1134
1135     if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1136       /* Go to sleep until it's time to push this buffer */
1137
1138       /* Recompute the highid */
1139       compute_high_id (mq);
1140       while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
1141         GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
1142             "newid %u and highid %u", sq->id, newid, mq->highid);
1143
1144
1145         /* Wake up all non-linked pads before we sleep */
1146         wake_up_next_non_linked (mq);
1147
1148         mq->numwaiting++;
1149         g_cond_wait (sq->turn, mq->qlock);
1150         mq->numwaiting--;
1151
1152         if (sq->flushing) {
1153           GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1154           goto out_flushing;
1155         }
1156
1157         GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
1158             "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
1159       }
1160
1161       /* Re-compute the high_id in case someone else pushed */
1162       compute_high_id (mq);
1163     } else {
1164       compute_high_id (mq);
1165       /* Wake up all non-linked pads */
1166       wake_up_next_non_linked (mq);
1167     }
1168     /* We're done waiting, we can clear the nextid */
1169     sq->nextid = 0;
1170
1171     GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1172   }
1173
1174   if (sq->flushing)
1175     goto out_flushing;
1176
1177   GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
1178       gst_flow_get_name (sq->srcresult));
1179
1180   /* Try to push out the new object */
1181   result = gst_single_queue_push_one (mq, sq, object);
1182   sq->srcresult = result;
1183   object = NULL;
1184
1185   if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED
1186       && result != GST_FLOW_UNEXPECTED)
1187     goto out_flushing;
1188
1189   GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
1190       gst_flow_get_name (sq->srcresult));
1191
1192   sq->last_oldid = newid;
1193   return;
1194
1195 out_flushing:
1196   {
1197     if (object)
1198       gst_mini_object_unref (object);
1199
1200     /* Need to make sure wake up any sleeping pads when we exit */
1201     GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1202     compute_high_id (mq);
1203     wake_up_next_non_linked (mq);
1204     GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1205
1206     /* upstream needs to see fatal result ASAP to shut things down,
1207      * but might be stuck in one of our other full queues;
1208      * so empty this one and trigger dynamic queue growth. At
1209      * this point the srcresult is not OK, NOT_LINKED
1210      * or UNEXPECTED, i.e. a real failure */
1211     gst_data_queue_flush (sq->queue);
1212     single_queue_underrun_cb (sq->queue, sq);
1213     gst_data_queue_set_flushing (sq->queue, TRUE);
1214     gst_pad_pause_task (sq->srcpad);
1215     GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
1216         "SingleQueue[%d] task paused, reason:%s",
1217         sq->id, gst_flow_get_name (sq->srcresult));
1218     return;
1219   }
1220 }
1221
1222 /**
1223  * gst_multi_queue_chain:
1224  *
1225  * This is similar to GstQueue's chain function, except:
1226  * _ we don't have leak behavioures,
1227  * _ we push with a unique id (curid)
1228  */
1229 static GstFlowReturn
1230 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
1231 {
1232   GstSingleQueue *sq;
1233   GstMultiQueue *mq;
1234   GstMultiQueueItem *item;
1235   guint32 curid;
1236   GstClockTime timestamp, duration;
1237
1238   sq = gst_pad_get_element_private (pad);
1239   mq = sq->mqueue;
1240
1241   /* if eos, we are always full, so avoid hanging incoming indefinitely */
1242   if (sq->is_eos)
1243     goto was_eos;
1244
1245   /* Get a unique incrementing id */
1246   curid = g_atomic_int_exchange_and_add ((gint *) & mq->counter, 1);
1247
1248   GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
1249       sq->id, buffer, curid);
1250
1251   item = gst_multi_queue_buffer_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
1252
1253   timestamp = GST_BUFFER_TIMESTAMP (buffer);
1254   duration = GST_BUFFER_DURATION (buffer);
1255
1256   if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1257     goto flushing;
1258
1259   /* update time level, we must do this after pushing the data in the queue so
1260    * that we never end up filling the queue first. */
1261   apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
1262
1263 done:
1264   return sq->srcresult;
1265
1266   /* ERRORS */
1267 flushing:
1268   {
1269     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1270         sq->id, gst_flow_get_name (sq->srcresult));
1271     gst_multi_queue_item_destroy (item);
1272     goto done;
1273   }
1274 was_eos:
1275   {
1276     GST_DEBUG_OBJECT (mq, "we are EOS, dropping buffer, return UNEXPECTED");
1277     gst_buffer_unref (buffer);
1278     return GST_FLOW_UNEXPECTED;
1279   }
1280 }
1281
1282 static gboolean
1283 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
1284 {
1285   GstSingleQueue *sq;
1286
1287   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1288
1289   if (active) {
1290     /* All pads start off linked until they push one buffer */
1291     sq->srcresult = GST_FLOW_OK;
1292   } else {
1293     sq->srcresult = GST_FLOW_WRONG_STATE;
1294     gst_data_queue_flush (sq->queue);
1295   }
1296   return TRUE;
1297 }
1298
1299 static gboolean
1300 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
1301 {
1302   GstSingleQueue *sq;
1303   GstMultiQueue *mq;
1304   guint32 curid;
1305   GstMultiQueueItem *item;
1306   gboolean res;
1307   GstEventType type;
1308   GstEvent *sref = NULL;
1309
1310   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1311   mq = (GstMultiQueue *) gst_pad_get_parent (pad);
1312
1313   type = GST_EVENT_TYPE (event);
1314
1315   switch (type) {
1316     case GST_EVENT_FLUSH_START:
1317       GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
1318           sq->id);
1319
1320       res = gst_pad_push_event (sq->srcpad, event);
1321
1322       gst_single_queue_flush (mq, sq, TRUE);
1323       goto done;
1324
1325     case GST_EVENT_FLUSH_STOP:
1326       GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
1327           sq->id);
1328
1329       res = gst_pad_push_event (sq->srcpad, event);
1330
1331       gst_single_queue_flush (mq, sq, FALSE);
1332       goto done;
1333     case GST_EVENT_NEWSEGMENT:
1334       /* take ref because the queue will take ownership and we need the event
1335        * afterwards to update the segment */
1336       sref = gst_event_ref (event);
1337       break;
1338
1339     default:
1340       if (!(GST_EVENT_IS_SERIALIZED (event))) {
1341         res = gst_pad_push_event (sq->srcpad, event);
1342         goto done;
1343       }
1344       break;
1345   }
1346
1347   /* if eos, we are always full, so avoid hanging incoming indefinitely */
1348   if (sq->is_eos)
1349     goto was_eos;
1350
1351   /* Get an unique incrementing id. */
1352   curid = g_atomic_int_exchange_and_add ((gint *) & mq->counter, 1);
1353
1354   item = gst_multi_queue_event_item_new ((GstMiniObject *) event, curid);
1355
1356   GST_DEBUG_OBJECT (mq,
1357       "SingleQueue %d : Enqueuing event %p of type %s with id %d",
1358       sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
1359
1360   if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1361     goto flushing;
1362
1363   /* mark EOS when we received one, we must do that after putting the
1364    * buffer in the queue because EOS marks the buffer as filled. No need to take
1365    * a lock, the _check_full happens from this thread only, right before pushing
1366    * into dataqueue. */
1367   switch (type) {
1368     case GST_EVENT_EOS:
1369       sq->is_eos = TRUE;
1370       /* EOS affects the buffering state */
1371       update_buffering (mq, sq);
1372       single_queue_overrun_cb (sq->queue, sq);
1373       break;
1374     case GST_EVENT_NEWSEGMENT:
1375       apply_segment (mq, sq, sref, &sq->sink_segment);
1376       gst_event_unref (sref);
1377       break;
1378     default:
1379       break;
1380   }
1381 done:
1382   gst_object_unref (mq);
1383   return res;
1384
1385 flushing:
1386   {
1387     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1388         sq->id, gst_flow_get_name (sq->srcresult));
1389     if (sref)
1390       gst_event_unref (sref);
1391     gst_multi_queue_item_destroy (item);
1392     goto done;
1393   }
1394 was_eos:
1395   {
1396     GST_DEBUG_OBJECT (mq, "we are EOS, dropping event, return FALSE");
1397     gst_event_unref (event);
1398     res = FALSE;
1399     goto done;
1400   }
1401 }
1402
1403 static GstCaps *
1404 gst_multi_queue_getcaps (GstPad * pad)
1405 {
1406   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1407   GstPad *otherpad;
1408   GstCaps *result;
1409
1410   otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1411
1412   GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
1413
1414   result = gst_pad_peer_get_caps (otherpad);
1415   if (result == NULL)
1416     result = gst_caps_new_any ();
1417
1418   return result;
1419 }
1420
1421 static gboolean
1422 gst_multi_queue_acceptcaps (GstPad * pad, GstCaps * caps)
1423 {
1424   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1425   GstPad *otherpad;
1426   gboolean result;
1427
1428   otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1429
1430   GST_LOG_OBJECT (otherpad, "Accept caps from the peer of this pad");
1431
1432   result = gst_pad_peer_accept_caps (otherpad, caps);
1433
1434   return result;
1435 }
1436
1437 static gboolean
1438 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1439 {
1440   GstMultiQueue *mq;
1441   GstSingleQueue *sq;
1442   gboolean result = FALSE;
1443
1444   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1445   mq = sq->mqueue;
1446
1447   GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1448
1449   if (active) {
1450     result = gst_single_queue_flush (mq, sq, FALSE);
1451   } else {
1452     result = gst_single_queue_flush (mq, sq, TRUE);
1453     /* make sure streaming finishes */
1454     result |= gst_pad_stop_task (pad);
1455   }
1456   return result;
1457 }
1458
1459 static gboolean
1460 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1461 {
1462   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1463
1464   return gst_pad_push_event (sq->sinkpad, event);
1465 }
1466
1467 static gboolean
1468 gst_multi_queue_src_query (GstPad * pad, GstQuery ** query)
1469 {
1470   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1471   GstPad *peerpad;
1472   gboolean res;
1473
1474   /* FIXME, Handle position offset depending on queue size */
1475
1476   /* default handling */
1477   if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1478     goto no_peer;
1479
1480   res = gst_pad_query (peerpad, query);
1481
1482   gst_object_unref (peerpad);
1483
1484   return res;
1485
1486   /* ERRORS */
1487 no_peer:
1488   {
1489     GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1490     return FALSE;
1491   }
1492 }
1493
1494 /*
1495  * Next-non-linked functions
1496  */
1497
1498 /* WITH LOCK TAKEN */
1499 static void
1500 wake_up_next_non_linked (GstMultiQueue * mq)
1501 {
1502   GList *tmp;
1503
1504   /* maybe no-one is waiting */
1505   if (mq->numwaiting < 1)
1506     return;
1507
1508   /* Else figure out which singlequeue(s) need waking up */
1509   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1510     GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1511
1512     if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1513       if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1514         GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1515         g_cond_signal (sq->turn);
1516       }
1517     }
1518   }
1519 }
1520
1521 /* WITH LOCK TAKEN */
1522 static void
1523 compute_high_id (GstMultiQueue * mq)
1524 {
1525   /* The high-id is either the highest id among the linked pads, or if all
1526    * pads are not-linked, it's the lowest not-linked pad */
1527   GList *tmp;
1528   guint32 lowest = G_MAXUINT32;
1529   guint32 highid = G_MAXUINT32;
1530
1531   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1532     GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1533
1534     GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1535         sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1536
1537     if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1538       /* No need to consider queues which are not waiting */
1539       if (sq->nextid == 0) {
1540         GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1541         continue;
1542       }
1543
1544       if (sq->nextid < lowest)
1545         lowest = sq->nextid;
1546     } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1547       /* If we don't have a global highid, or the global highid is lower than
1548        * this single queue's last outputted id, store the queue's one,
1549        * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1550       if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1551         highid = sq->oldid;
1552     }
1553   }
1554
1555   if (highid == G_MAXUINT32 || lowest < highid)
1556     mq->highid = lowest;
1557   else
1558     mq->highid = highid;
1559
1560   GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1561       lowest);
1562 }
1563
1564 #define IS_FILLED(q, format, value) (((q)->max_size.format) != 0 && \
1565      ((q)->max_size.format) <= (value))
1566
1567 /*
1568  * GstSingleQueue functions
1569  */
1570 static void
1571 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1572 {
1573   GstMultiQueue *mq = sq->mqueue;
1574   GList *tmp;
1575   GstDataQueueSize size;
1576   gboolean filled = FALSE;
1577
1578   gst_data_queue_get_level (sq->queue, &size);
1579
1580   GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1581
1582   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1583   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1584     GstSingleQueue *oq = (GstSingleQueue *) tmp->data;
1585     GstDataQueueSize ssize;
1586
1587     GST_LOG_OBJECT (mq, "Checking Queue %d", oq->id);
1588
1589     if (gst_data_queue_is_empty (oq->queue)) {
1590       GST_LOG_OBJECT (mq, "Queue %d is empty", oq->id);
1591       if (IS_FILLED (sq, visible, size.visible)) {
1592         sq->max_size.visible = size.visible + 1;
1593         GST_DEBUG_OBJECT (mq,
1594             "Another queue is empty, bumping single queue %d max visible to %d",
1595             sq->id, sq->max_size.visible);
1596       }
1597       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1598       goto beach;
1599     }
1600     /* check if we reached the hard time/bytes limits */
1601     gst_data_queue_get_level (oq->queue, &ssize);
1602
1603     GST_DEBUG_OBJECT (mq,
1604         "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1605         G_GUINT64_FORMAT, oq->id, ssize.visible, oq->max_size.visible,
1606         ssize.bytes, oq->max_size.bytes, oq->cur_time, oq->max_size.time);
1607
1608     /* if this queue is filled completely we must signal overrun.
1609      * FIXME, this seems wrong in many ways
1610      *  - we're comparing the filled level of this queue against the
1611      *    values of the other one
1612      *  - we should only do this after we found no empty queues, ie, move
1613      *    this check outside of the loop
1614      *  - the debug statement talks about a different queue than the one
1615      *    we are checking here.
1616      */
1617     if (sq->is_eos || IS_FILLED (sq, bytes, ssize.bytes) ||
1618         IS_FILLED (sq, time, sq->cur_time)) {
1619       GST_LOG_OBJECT (mq, "Queue %d is filled", oq->id);
1620       filled = TRUE;
1621     }
1622   }
1623   /* no queues were empty */
1624   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1625
1626   /* Overrun is always forwarded, since this is blocking the upstream element */
1627   if (filled) {
1628     GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1629     g_signal_emit (mq, gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1630   }
1631
1632 beach:
1633   return;
1634 }
1635
1636 static void
1637 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1638 {
1639   gboolean empty = TRUE;
1640   GstMultiQueue *mq = sq->mqueue;
1641   GList *tmp;
1642
1643   GST_LOG_OBJECT (mq,
1644       "Single Queue %d is empty, Checking other single queues", sq->id);
1645
1646   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1647   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1648     GstSingleQueue *oq = (GstSingleQueue *) tmp->data;
1649
1650     if (gst_data_queue_is_full (oq->queue)) {
1651       GstDataQueueSize size;
1652
1653       gst_data_queue_get_level (oq->queue, &size);
1654       if (IS_FILLED (oq, visible, size.visible)) {
1655         oq->max_size.visible = size.visible + 1;
1656         GST_DEBUG_OBJECT (mq,
1657             "queue %d is filled, bumping its max visible to %d", oq->id,
1658             oq->max_size.visible);
1659         gst_data_queue_limits_changed (oq->queue);
1660       }
1661     }
1662     if (!gst_data_queue_is_empty (oq->queue))
1663       empty = FALSE;
1664   }
1665   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1666
1667   if (empty) {
1668     GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1669     g_signal_emit (mq, gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1670   }
1671 }
1672
1673 static gboolean
1674 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1675     guint64 time, GstSingleQueue * sq)
1676 {
1677   gboolean res;
1678   GstMultiQueue *mq = sq->mqueue;
1679
1680   GST_DEBUG_OBJECT (mq,
1681       "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1682       G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1683       sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1684
1685   /* we are always filled on EOS */
1686   if (sq->is_eos)
1687     return TRUE;
1688
1689   /* we never go past the max visible items unless we are in buffering mode */
1690   if (!mq->use_buffering && IS_FILLED (sq, visible, visible))
1691     return TRUE;
1692
1693   /* check time or bytes */
1694   res = IS_FILLED (sq, time, sq->cur_time) || IS_FILLED (sq, bytes, bytes);
1695
1696   return res;
1697 }
1698
1699 static void
1700 gst_single_queue_free (GstSingleQueue * sq)
1701 {
1702   /* DRAIN QUEUE */
1703   gst_data_queue_flush (sq->queue);
1704   g_object_unref (sq->queue);
1705   g_cond_free (sq->turn);
1706   g_free (sq);
1707 }
1708
1709 static GstSingleQueue *
1710 gst_single_queue_new (GstMultiQueue * mqueue, gint id)
1711 {
1712   GstSingleQueue *sq;
1713   gchar *name;
1714   GList *tmp;
1715   gint temp_id = (id == -1) ? 0 : id;
1716
1717   GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1718
1719   /* Find an unused queue ID, if possible the passed one */
1720   for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
1721     GstSingleQueue *sq2 = (GstSingleQueue *) tmp->data;
1722     /* This works because the IDs are sorted in ascending order */
1723     if (sq2->id == temp_id) {
1724       /* If this ID was requested by the caller return NULL,
1725        * otherwise just get us the next one */
1726       if (id == -1)
1727         temp_id = sq2->id + 1;
1728       else
1729         return NULL;
1730     } else if (sq2->id > temp_id) {
1731       break;
1732     }
1733   }
1734
1735   sq = g_new0 (GstSingleQueue, 1);
1736   mqueue->nbqueues++;
1737   sq->id = temp_id;
1738
1739   mqueue->queues = g_list_insert_before (mqueue->queues, tmp, sq);
1740   mqueue->queues_cookie++;
1741
1742   /* copy over max_size and extra_size so we don't need to take the lock
1743    * any longer when checking if the queue is full. */
1744   sq->max_size.visible = mqueue->max_size.visible;
1745   sq->max_size.bytes = mqueue->max_size.bytes;
1746   sq->max_size.time = mqueue->max_size.time;
1747
1748   sq->extra_size.visible = mqueue->extra_size.visible;
1749   sq->extra_size.bytes = mqueue->extra_size.bytes;
1750   sq->extra_size.time = mqueue->extra_size.time;
1751
1752   GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1753
1754   GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1755
1756   sq->mqueue = mqueue;
1757   sq->srcresult = GST_FLOW_WRONG_STATE;
1758   sq->queue = gst_data_queue_new_full ((GstDataQueueCheckFullFunction)
1759       single_queue_check_full,
1760       (GstDataQueueFullCallback) single_queue_overrun_cb,
1761       (GstDataQueueEmptyCallback) single_queue_underrun_cb, sq);
1762   sq->is_eos = FALSE;
1763   sq->flushing = FALSE;
1764   gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1765   gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1766
1767   sq->nextid = 0;
1768   sq->oldid = 0;
1769   sq->turn = g_cond_new ();
1770
1771   sq->sinktime = GST_CLOCK_TIME_NONE;
1772   sq->srctime = GST_CLOCK_TIME_NONE;
1773   sq->sink_tainted = TRUE;
1774   sq->src_tainted = TRUE;
1775
1776   name = g_strdup_printf ("sink%d", sq->id);
1777   sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, name);
1778   g_free (name);
1779
1780   gst_pad_set_chain_function (sq->sinkpad,
1781       GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1782   gst_pad_set_activatepush_function (sq->sinkpad,
1783       GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1784   gst_pad_set_event_function (sq->sinkpad,
1785       GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1786   gst_pad_set_getcaps_function (sq->sinkpad,
1787       GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1788   gst_pad_set_acceptcaps_function (sq->sinkpad,
1789       GST_DEBUG_FUNCPTR (gst_multi_queue_acceptcaps));
1790   gst_pad_set_iterate_internal_links_function (sq->sinkpad,
1791       GST_DEBUG_FUNCPTR (gst_multi_queue_iterate_internal_links));
1792
1793   name = g_strdup_printf ("src%d", sq->id);
1794   sq->srcpad = gst_pad_new_from_static_template (&srctemplate, name);
1795   g_free (name);
1796
1797   gst_pad_set_activatepush_function (sq->srcpad,
1798       GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1799   gst_pad_set_getcaps_function (sq->srcpad,
1800       GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1801   gst_pad_set_acceptcaps_function (sq->srcpad,
1802       GST_DEBUG_FUNCPTR (gst_multi_queue_acceptcaps));
1803   gst_pad_set_event_function (sq->srcpad,
1804       GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1805   gst_pad_set_query_function (sq->srcpad,
1806       GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1807   gst_pad_set_iterate_internal_links_function (sq->srcpad,
1808       GST_DEBUG_FUNCPTR (gst_multi_queue_iterate_internal_links));
1809
1810   gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1811   gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1812
1813   /* only activate the pads when we are not in the NULL state
1814    * and add the pad under the state_lock to prevend state changes
1815    * between activating and adding */
1816   g_static_rec_mutex_lock (GST_STATE_GET_LOCK (mqueue));
1817   if (GST_STATE_TARGET (mqueue) != GST_STATE_NULL) {
1818     gst_pad_set_active (sq->srcpad, TRUE);
1819     gst_pad_set_active (sq->sinkpad, TRUE);
1820   }
1821   gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1822   gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1823   g_static_rec_mutex_unlock (GST_STATE_GET_LOCK (mqueue));
1824
1825   GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1826       sq->id);
1827
1828   return sq;
1829 }