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.position);
 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.position);
 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 SEGMENT 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   gst_event_parse_segment (event, segment);
 877
 878   /* now configure the values, we use these to track timestamps on the
 879    * sinkpad. */
 880   if (segment->format != GST_FORMAT_TIME) {
 881     /* non-time format, pretent the current time segment is closed with a
 882      * 0 start and unknown stop time. */
 883     segment->format = GST_FORMAT_TIME;
 884     segment->start = 0;
 885     segment->stop = -1;
 886     segment->time = 0;
 887   }
 888   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 889
 890   if (segment == &sq->sink_segment)
 891     sq->sink_tainted = TRUE;
 892   else
 893     sq->src_tainted = TRUE;
 894
 895   GST_DEBUG_OBJECT (mq,
 896       "queue %d, configured SEGMENT %" GST_SEGMENT_FORMAT, sq->id, segment);
 897
 898   /* segment can update the time level of the queue */
 899   update_time_level (mq, sq);
 900
 901   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 902 }
 903
 904 /* take a buffer and update segment, updating the time level of the queue. */
 905 static void
 906 apply_buffer (GstMultiQueue * mq, GstSingleQueue * sq, GstClockTime timestamp,
 907     GstClockTime duration, GstSegment * segment)
 908 {
 909   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 910
 911   /* if no timestamp is set, assume it's continuous with the previous
 912    * time */
 913   if (timestamp == GST_CLOCK_TIME_NONE)
 914     timestamp = segment->position;
 915
 916   /* add duration */
 917   if (duration != GST_CLOCK_TIME_NONE)
 918     timestamp += duration;
 919
 920   GST_DEBUG_OBJECT (mq, "queue %d, position updated to %" GST_TIME_FORMAT,
 921       sq->id, GST_TIME_ARGS (timestamp));
 922
 923   segment->position = timestamp;
 924
 925   if (segment == &sq->sink_segment)
 926     sq->sink_tainted = TRUE;
 927   else
 928     sq->src_tainted = TRUE;
 929
 930   /* calc diff with other end */
 931   update_time_level (mq, sq);
 932   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
 933 }
 934
 935 static GstFlowReturn
 936 gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
 937     GstMiniObject * object)
 938 {
 939   GstFlowReturn result = GST_FLOW_OK;
 940
 941   if (GST_IS_BUFFER (object)) {
 942     GstBuffer *buffer;
 943     GstClockTime timestamp, duration;
 944 #if 0
 945     GstCaps *caps;
 946 #endif
 947
 948     buffer = GST_BUFFER_CAST (object);
 949     timestamp = GST_BUFFER_TIMESTAMP (buffer);
 950     duration = GST_BUFFER_DURATION (buffer);
 951 #if 0
 952     caps = GST_BUFFER_CAPS (buffer);
 953 #endif
 954
 955     apply_buffer (mq, sq, timestamp, duration, &sq->src_segment);
 956
 957     /* Applying the buffer may have made the queue non-full again, unblock it if needed */
 958     gst_data_queue_limits_changed (sq->queue);
 959
 960     GST_DEBUG_OBJECT (mq,
 961         "SingleQueue %d : Pushing buffer %p with ts %" GST_TIME_FORMAT,
 962         sq->id, buffer, GST_TIME_ARGS (timestamp));
 963
 964 #if 0
 965     /* Set caps on pad before pushing, this avoids core calling the acceptcaps
 966      * function on the srcpad, which will call acceptcaps upstream, which might
 967      * not accept these caps (anymore). */
 968     if (caps && caps != GST_PAD_CAPS (sq->srcpad))
 969       gst_pad_set_caps (sq->srcpad, caps);
 970 #endif
 971
 972     result = gst_pad_push (sq->srcpad, buffer);
 973   } else if (GST_IS_EVENT (object)) {
 974     GstEvent *event;
 975
 976     event = GST_EVENT_CAST (object);
 977
 978     switch (GST_EVENT_TYPE (event)) {
 979       case GST_EVENT_EOS:
 980         result = GST_FLOW_UNEXPECTED;
 981         break;
 982       case GST_EVENT_SEGMENT:
 983         apply_segment (mq, sq, event, &sq->src_segment);
 984         /* Applying the segment may have made the queue non-full again, unblock it if needed */
 985         gst_data_queue_limits_changed (sq->queue);
 986         break;
 987       default:
 988         break;
 989     }
 990
 991     GST_DEBUG_OBJECT (mq,
 992         "SingleQueue %d : Pushing event %p of type %s",
 993         sq->id, event, GST_EVENT_TYPE_NAME (event));
 994
 995     gst_pad_push_event (sq->srcpad, event);
 996   } else {
 997     g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
 998         sq->id);
 999   }
1000   return result;
1001
1002   /* ERRORS */
1003 }
1004
1005 static GstMiniObject *
1006 gst_multi_queue_item_steal_object (GstMultiQueueItem * item)
1007 {
1008   GstMiniObject *res;
1009
1010   res = item->object;
1011   item->object = NULL;
1012
1013   return res;
1014 }
1015
1016 static void
1017 gst_multi_queue_item_destroy (GstMultiQueueItem * item)
1018 {
1019   if (item->object)
1020     gst_mini_object_unref (item->object);
1021   g_slice_free (GstMultiQueueItem, item);
1022 }
1023
1024 /* takes ownership of passed mini object! */
1025 static GstMultiQueueItem *
1026 gst_multi_queue_buffer_item_new (GstMiniObject * object, guint32 curid)
1027 {
1028   GstMultiQueueItem *item;
1029
1030   item = g_slice_new (GstMultiQueueItem);
1031   item->object = object;
1032   item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
1033   item->posid = curid;
1034
1035   item->size = gst_buffer_get_size (GST_BUFFER_CAST (object));
1036   item->duration = GST_BUFFER_DURATION (object);
1037   if (item->duration == GST_CLOCK_TIME_NONE)
1038     item->duration = 0;
1039   item->visible = TRUE;
1040   return item;
1041 }
1042
1043 static GstMultiQueueItem *
1044 gst_multi_queue_event_item_new (GstMiniObject * object, guint32 curid)
1045 {
1046   GstMultiQueueItem *item;
1047
1048   item = g_slice_new (GstMultiQueueItem);
1049   item->object = object;
1050   item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
1051   item->posid = curid;
1052
1053   item->size = 0;
1054   item->duration = 0;
1055   item->visible = FALSE;
1056   return item;
1057 }
1058
1059 /* Each main loop attempts to push buffers until the return value
1060  * is not-linked. not-linked pads are not allowed to push data beyond
1061  * any linked pads, so they don't 'rush ahead of the pack'.
1062  */
1063 static void
1064 gst_multi_queue_loop (GstPad * pad)
1065 {
1066   GstSingleQueue *sq;
1067   GstMultiQueueItem *item;
1068   GstDataQueueItem *sitem;
1069   GstMultiQueue *mq;
1070   GstMiniObject *object = NULL;
1071   guint32 newid;
1072   GstFlowReturn result;
1073
1074   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1075   mq = sq->mqueue;
1076
1077   GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
1078
1079   if (sq->flushing)
1080     goto out_flushing;
1081
1082   /* Get something from the queue, blocking until that happens, or we get
1083    * flushed */
1084   if (!(gst_data_queue_pop (sq->queue, &sitem)))
1085     goto out_flushing;
1086
1087   item = (GstMultiQueueItem *) sitem;
1088   newid = item->posid;
1089
1090   /* steal the object and destroy the item */
1091   object = gst_multi_queue_item_steal_object (item);
1092   gst_multi_queue_item_destroy (item);
1093
1094   GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
1095       sq->id, newid, sq->last_oldid);
1096
1097   /* If we're not-linked, we do some extra work because we might need to
1098    * wait before pushing. If we're linked but there's a gap in the IDs,
1099    * or it's the first loop, or we just passed the previous highid,
1100    * we might need to wake some sleeping pad up, so there's extra work
1101    * there too */
1102   if (sq->srcresult == GST_FLOW_NOT_LINKED ||
1103       (sq->last_oldid == G_MAXUINT32) || (newid != (sq->last_oldid + 1)) ||
1104       sq->last_oldid > mq->highid) {
1105     GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
1106         gst_flow_get_name (sq->srcresult));
1107
1108     GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1109
1110     /* Check again if we're flushing after the lock is taken,
1111      * the flush flag might have been changed in the meantime */
1112     if (sq->flushing) {
1113       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1114       goto out_flushing;
1115     }
1116
1117     /* Update the nextid so other threads know when to wake us up */
1118     sq->nextid = newid;
1119
1120     /* Update the oldid (the last ID we output) for highid tracking */
1121     if (sq->last_oldid != G_MAXUINT32)
1122       sq->oldid = sq->last_oldid;
1123
1124     if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1125       /* Go to sleep until it's time to push this buffer */
1126
1127       /* Recompute the highid */
1128       compute_high_id (mq);
1129       while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
1130         GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
1131             "newid %u and highid %u", sq->id, newid, mq->highid);
1132
1133
1134         /* Wake up all non-linked pads before we sleep */
1135         wake_up_next_non_linked (mq);
1136
1137         mq->numwaiting++;
1138         g_cond_wait (sq->turn, mq->qlock);
1139         mq->numwaiting--;
1140
1141         if (sq->flushing) {
1142           GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1143           goto out_flushing;
1144         }
1145
1146         GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
1147             "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
1148       }
1149
1150       /* Re-compute the high_id in case someone else pushed */
1151       compute_high_id (mq);
1152     } else {
1153       compute_high_id (mq);
1154       /* Wake up all non-linked pads */
1155       wake_up_next_non_linked (mq);
1156     }
1157     /* We're done waiting, we can clear the nextid */
1158     sq->nextid = 0;
1159
1160     GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1161   }
1162
1163   if (sq->flushing)
1164     goto out_flushing;
1165
1166   GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
1167       gst_flow_get_name (sq->srcresult));
1168
1169   /* Try to push out the new object */
1170   result = gst_single_queue_push_one (mq, sq, object);
1171   sq->srcresult = result;
1172   object = NULL;
1173
1174   if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED
1175       && result != GST_FLOW_UNEXPECTED)
1176     goto out_flushing;
1177
1178   GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
1179       gst_flow_get_name (sq->srcresult));
1180
1181   sq->last_oldid = newid;
1182   return;
1183
1184 out_flushing:
1185   {
1186     if (object)
1187       gst_mini_object_unref (object);
1188
1189     /* Need to make sure wake up any sleeping pads when we exit */
1190     GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1191     compute_high_id (mq);
1192     wake_up_next_non_linked (mq);
1193     GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1194
1195     /* upstream needs to see fatal result ASAP to shut things down,
1196      * but might be stuck in one of our other full queues;
1197      * so empty this one and trigger dynamic queue growth. At
1198      * this point the srcresult is not OK, NOT_LINKED
1199      * or UNEXPECTED, i.e. a real failure */
1200     gst_data_queue_flush (sq->queue);
1201     single_queue_underrun_cb (sq->queue, sq);
1202     gst_data_queue_set_flushing (sq->queue, TRUE);
1203     gst_pad_pause_task (sq->srcpad);
1204     GST_CAT_LOG_OBJECT (multi_queue_debug, mq,
1205         "SingleQueue[%d] task paused, reason:%s",
1206         sq->id, gst_flow_get_name (sq->srcresult));
1207     return;
1208   }
1209 }
1210
1211 /**
1212  * gst_multi_queue_chain:
1213  *
1214  * This is similar to GstQueue's chain function, except:
1215  * _ we don't have leak behavioures,
1216  * _ we push with a unique id (curid)
1217  */
1218 static GstFlowReturn
1219 gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
1220 {
1221   GstSingleQueue *sq;
1222   GstMultiQueue *mq;
1223   GstMultiQueueItem *item;
1224   guint32 curid;
1225   GstClockTime timestamp, duration;
1226
1227   sq = gst_pad_get_element_private (pad);
1228   mq = sq->mqueue;
1229
1230   /* if eos, we are always full, so avoid hanging incoming indefinitely */
1231   if (sq->is_eos)
1232     goto was_eos;
1233
1234   /* Get a unique incrementing id */
1235   curid = g_atomic_int_exchange_and_add ((gint *) & mq->counter, 1);
1236
1237   GST_LOG_OBJECT (mq, "SingleQueue %d : about to enqueue buffer %p with id %d",
1238       sq->id, buffer, curid);
1239
1240   item = gst_multi_queue_buffer_item_new (GST_MINI_OBJECT_CAST (buffer), curid);
1241
1242   timestamp = GST_BUFFER_TIMESTAMP (buffer);
1243   duration = GST_BUFFER_DURATION (buffer);
1244
1245   if (!(gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1246     goto flushing;
1247
1248   /* update time level, we must do this after pushing the data in the queue so
1249    * that we never end up filling the queue first. */
1250   apply_buffer (mq, sq, timestamp, duration, &sq->sink_segment);
1251
1252 done:
1253   return sq->srcresult;
1254
1255   /* ERRORS */
1256 flushing:
1257   {
1258     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1259         sq->id, gst_flow_get_name (sq->srcresult));
1260     gst_multi_queue_item_destroy (item);
1261     goto done;
1262   }
1263 was_eos:
1264   {
1265     GST_DEBUG_OBJECT (mq, "we are EOS, dropping buffer, return UNEXPECTED");
1266     gst_buffer_unref (buffer);
1267     return GST_FLOW_UNEXPECTED;
1268   }
1269 }
1270
1271 static gboolean
1272 gst_multi_queue_sink_activate_push (GstPad * pad, gboolean active)
1273 {
1274   GstSingleQueue *sq;
1275
1276   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1277
1278   if (active) {
1279     /* All pads start off linked until they push one buffer */
1280     sq->srcresult = GST_FLOW_OK;
1281   } else {
1282     sq->srcresult = GST_FLOW_WRONG_STATE;
1283     gst_data_queue_flush (sq->queue);
1284   }
1285   return TRUE;
1286 }
1287
1288 static gboolean
1289 gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
1290 {
1291   GstSingleQueue *sq;
1292   GstMultiQueue *mq;
1293   guint32 curid;
1294   GstMultiQueueItem *item;
1295   gboolean res;
1296   GstEventType type;
1297   GstEvent *sref = NULL;
1298
1299   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1300   mq = (GstMultiQueue *) gst_pad_get_parent (pad);
1301
1302   type = GST_EVENT_TYPE (event);
1303
1304   switch (type) {
1305     case GST_EVENT_FLUSH_START:
1306       GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush start event",
1307           sq->id);
1308
1309       res = gst_pad_push_event (sq->srcpad, event);
1310
1311       gst_single_queue_flush (mq, sq, TRUE);
1312       goto done;
1313
1314     case GST_EVENT_FLUSH_STOP:
1315       GST_DEBUG_OBJECT (mq, "SingleQueue %d : received flush stop event",
1316           sq->id);
1317
1318       res = gst_pad_push_event (sq->srcpad, event);
1319
1320       gst_single_queue_flush (mq, sq, FALSE);
1321       goto done;
1322     case GST_EVENT_SEGMENT:
1323       /* take ref because the queue will take ownership and we need the event
1324        * afterwards to update the segment */
1325       sref = gst_event_ref (event);
1326       break;
1327
1328     default:
1329       if (!(GST_EVENT_IS_SERIALIZED (event))) {
1330         res = gst_pad_push_event (sq->srcpad, event);
1331         goto done;
1332       }
1333       break;
1334   }
1335
1336   /* if eos, we are always full, so avoid hanging incoming indefinitely */
1337   if (sq->is_eos)
1338     goto was_eos;
1339
1340   /* Get an unique incrementing id. */
1341   curid = g_atomic_int_exchange_and_add ((gint *) & mq->counter, 1);
1342
1343   item = gst_multi_queue_event_item_new ((GstMiniObject *) event, curid);
1344
1345   GST_DEBUG_OBJECT (mq,
1346       "SingleQueue %d : Enqueuing event %p of type %s with id %d",
1347       sq->id, event, GST_EVENT_TYPE_NAME (event), curid);
1348
1349   if (!(res = gst_data_queue_push (sq->queue, (GstDataQueueItem *) item)))
1350     goto flushing;
1351
1352   /* mark EOS when we received one, we must do that after putting the
1353    * buffer in the queue because EOS marks the buffer as filled. No need to take
1354    * a lock, the _check_full happens from this thread only, right before pushing
1355    * into dataqueue. */
1356   switch (type) {
1357     case GST_EVENT_EOS:
1358       sq->is_eos = TRUE;
1359       /* EOS affects the buffering state */
1360       update_buffering (mq, sq);
1361       single_queue_overrun_cb (sq->queue, sq);
1362       break;
1363     case GST_EVENT_SEGMENT:
1364       apply_segment (mq, sq, sref, &sq->sink_segment);
1365       gst_event_unref (sref);
1366       break;
1367     default:
1368       break;
1369   }
1370 done:
1371   gst_object_unref (mq);
1372   return res;
1373
1374 flushing:
1375   {
1376     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
1377         sq->id, gst_flow_get_name (sq->srcresult));
1378     if (sref)
1379       gst_event_unref (sref);
1380     gst_multi_queue_item_destroy (item);
1381     goto done;
1382   }
1383 was_eos:
1384   {
1385     GST_DEBUG_OBJECT (mq, "we are EOS, dropping event, return FALSE");
1386     gst_event_unref (event);
1387     res = FALSE;
1388     goto done;
1389   }
1390 }
1391
1392 static GstCaps *
1393 gst_multi_queue_getcaps (GstPad * pad, GstCaps * filter)
1394 {
1395   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1396   GstPad *otherpad;
1397   GstCaps *result;
1398
1399   otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1400
1401   GST_LOG_OBJECT (otherpad, "Getting caps from the peer of this pad");
1402
1403   result = gst_pad_peer_get_caps (otherpad, filter);
1404   if (result == NULL)
1405     result = (filter ? gst_caps_ref (filter) : gst_caps_new_any ());
1406
1407   return result;
1408 }
1409
1410 static gboolean
1411 gst_multi_queue_acceptcaps (GstPad * pad, GstCaps * caps)
1412 {
1413   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1414   GstPad *otherpad;
1415   gboolean result;
1416
1417   otherpad = (pad == sq->srcpad) ? sq->sinkpad : sq->srcpad;
1418
1419   GST_LOG_OBJECT (otherpad, "Accept caps from the peer of this pad");
1420
1421   result = gst_pad_peer_accept_caps (otherpad, caps);
1422
1423   return result;
1424 }
1425
1426 static gboolean
1427 gst_multi_queue_src_activate_push (GstPad * pad, gboolean active)
1428 {
1429   GstMultiQueue *mq;
1430   GstSingleQueue *sq;
1431   gboolean result = FALSE;
1432
1433   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
1434   mq = sq->mqueue;
1435
1436   GST_DEBUG_OBJECT (mq, "SingleQueue %d", sq->id);
1437
1438   if (active) {
1439     result = gst_single_queue_flush (mq, sq, FALSE);
1440   } else {
1441     result = gst_single_queue_flush (mq, sq, TRUE);
1442     /* make sure streaming finishes */
1443     result |= gst_pad_stop_task (pad);
1444   }
1445   return result;
1446 }
1447
1448 static gboolean
1449 gst_multi_queue_src_event (GstPad * pad, GstEvent * event)
1450 {
1451   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1452
1453   return gst_pad_push_event (sq->sinkpad, event);
1454 }
1455
1456 static gboolean
1457 gst_multi_queue_src_query (GstPad * pad, GstQuery ** query)
1458 {
1459   GstSingleQueue *sq = gst_pad_get_element_private (pad);
1460   GstPad *peerpad;
1461   gboolean res;
1462
1463   /* FIXME, Handle position offset depending on queue size */
1464
1465   /* default handling */
1466   if (!(peerpad = gst_pad_get_peer (sq->sinkpad)))
1467     goto no_peer;
1468
1469   res = gst_pad_query (peerpad, query);
1470
1471   gst_object_unref (peerpad);
1472
1473   return res;
1474
1475   /* ERRORS */
1476 no_peer:
1477   {
1478     GST_LOG_OBJECT (sq->sinkpad, "Couldn't send query because we have no peer");
1479     return FALSE;
1480   }
1481 }
1482
1483 /*
1484  * Next-non-linked functions
1485  */
1486
1487 /* WITH LOCK TAKEN */
1488 static void
1489 wake_up_next_non_linked (GstMultiQueue * mq)
1490 {
1491   GList *tmp;
1492
1493   /* maybe no-one is waiting */
1494   if (mq->numwaiting < 1)
1495     return;
1496
1497   /* Else figure out which singlequeue(s) need waking up */
1498   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1499     GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1500
1501     if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1502       if (sq->nextid != 0 && sq->nextid <= mq->highid) {
1503         GST_LOG_OBJECT (mq, "Waking up singlequeue %d", sq->id);
1504         g_cond_signal (sq->turn);
1505       }
1506     }
1507   }
1508 }
1509
1510 /* WITH LOCK TAKEN */
1511 static void
1512 compute_high_id (GstMultiQueue * mq)
1513 {
1514   /* The high-id is either the highest id among the linked pads, or if all
1515    * pads are not-linked, it's the lowest not-linked pad */
1516   GList *tmp;
1517   guint32 lowest = G_MAXUINT32;
1518   guint32 highid = G_MAXUINT32;
1519
1520   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1521     GstSingleQueue *sq = (GstSingleQueue *) tmp->data;
1522
1523     GST_LOG_OBJECT (mq, "inspecting sq:%d , nextid:%d, oldid:%d, srcresult:%s",
1524         sq->id, sq->nextid, sq->oldid, gst_flow_get_name (sq->srcresult));
1525
1526     if (sq->srcresult == GST_FLOW_NOT_LINKED) {
1527       /* No need to consider queues which are not waiting */
1528       if (sq->nextid == 0) {
1529         GST_LOG_OBJECT (mq, "sq:%d is not waiting - ignoring", sq->id);
1530         continue;
1531       }
1532
1533       if (sq->nextid < lowest)
1534         lowest = sq->nextid;
1535     } else if (sq->srcresult != GST_FLOW_UNEXPECTED) {
1536       /* If we don't have a global highid, or the global highid is lower than
1537        * this single queue's last outputted id, store the queue's one,
1538        * unless the singlequeue is at EOS (srcresult = UNEXPECTED) */
1539       if ((highid == G_MAXUINT32) || (sq->oldid > highid))
1540         highid = sq->oldid;
1541     }
1542   }
1543
1544   if (highid == G_MAXUINT32 || lowest < highid)
1545     mq->highid = lowest;
1546   else
1547     mq->highid = highid;
1548
1549   GST_LOG_OBJECT (mq, "Highid is now : %u, lowest non-linked %u", mq->highid,
1550       lowest);
1551 }
1552
1553 #define IS_FILLED(q, format, value) (((q)->max_size.format) != 0 && \
1554      ((q)->max_size.format) <= (value))
1555
1556 /*
1557  * GstSingleQueue functions
1558  */
1559 static void
1560 single_queue_overrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1561 {
1562   GstMultiQueue *mq = sq->mqueue;
1563   GList *tmp;
1564   GstDataQueueSize size;
1565   gboolean filled = FALSE;
1566
1567   gst_data_queue_get_level (sq->queue, &size);
1568
1569   GST_LOG_OBJECT (mq, "Single Queue %d is full", sq->id);
1570
1571   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1572   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1573     GstSingleQueue *oq = (GstSingleQueue *) tmp->data;
1574     GstDataQueueSize ssize;
1575
1576     GST_LOG_OBJECT (mq, "Checking Queue %d", oq->id);
1577
1578     if (gst_data_queue_is_empty (oq->queue)) {
1579       GST_LOG_OBJECT (mq, "Queue %d is empty", oq->id);
1580       if (IS_FILLED (sq, visible, size.visible)) {
1581         sq->max_size.visible = size.visible + 1;
1582         GST_DEBUG_OBJECT (mq,
1583             "Another queue is empty, bumping single queue %d max visible to %d",
1584             sq->id, sq->max_size.visible);
1585       }
1586       GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1587       goto beach;
1588     }
1589     /* check if we reached the hard time/bytes limits */
1590     gst_data_queue_get_level (oq->queue, &ssize);
1591
1592     GST_DEBUG_OBJECT (mq,
1593         "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1594         G_GUINT64_FORMAT, oq->id, ssize.visible, oq->max_size.visible,
1595         ssize.bytes, oq->max_size.bytes, oq->cur_time, oq->max_size.time);
1596
1597     /* if this queue is filled completely we must signal overrun.
1598      * FIXME, this seems wrong in many ways
1599      *  - we're comparing the filled level of this queue against the
1600      *    values of the other one
1601      *  - we should only do this after we found no empty queues, ie, move
1602      *    this check outside of the loop
1603      *  - the debug statement talks about a different queue than the one
1604      *    we are checking here.
1605      */
1606     if (sq->is_eos || IS_FILLED (sq, bytes, ssize.bytes) ||
1607         IS_FILLED (sq, time, sq->cur_time)) {
1608       GST_LOG_OBJECT (mq, "Queue %d is filled", oq->id);
1609       filled = TRUE;
1610     }
1611   }
1612   /* no queues were empty */
1613   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1614
1615   /* Overrun is always forwarded, since this is blocking the upstream element */
1616   if (filled) {
1617     GST_DEBUG_OBJECT (mq, "A queue is filled, signalling overrun");
1618     g_signal_emit (mq, gst_multi_queue_signals[SIGNAL_OVERRUN], 0);
1619   }
1620
1621 beach:
1622   return;
1623 }
1624
1625 static void
1626 single_queue_underrun_cb (GstDataQueue * dq, GstSingleQueue * sq)
1627 {
1628   gboolean empty = TRUE;
1629   GstMultiQueue *mq = sq->mqueue;
1630   GList *tmp;
1631
1632   GST_LOG_OBJECT (mq,
1633       "Single Queue %d is empty, Checking other single queues", sq->id);
1634
1635   GST_MULTI_QUEUE_MUTEX_LOCK (mq);
1636   for (tmp = mq->queues; tmp; tmp = g_list_next (tmp)) {
1637     GstSingleQueue *oq = (GstSingleQueue *) tmp->data;
1638
1639     if (gst_data_queue_is_full (oq->queue)) {
1640       GstDataQueueSize size;
1641
1642       gst_data_queue_get_level (oq->queue, &size);
1643       if (IS_FILLED (oq, visible, size.visible)) {
1644         oq->max_size.visible = size.visible + 1;
1645         GST_DEBUG_OBJECT (mq,
1646             "queue %d is filled, bumping its max visible to %d", oq->id,
1647             oq->max_size.visible);
1648         gst_data_queue_limits_changed (oq->queue);
1649       }
1650     }
1651     if (!gst_data_queue_is_empty (oq->queue))
1652       empty = FALSE;
1653   }
1654   GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
1655
1656   if (empty) {
1657     GST_DEBUG_OBJECT (mq, "All queues are empty, signalling it");
1658     g_signal_emit (mq, gst_multi_queue_signals[SIGNAL_UNDERRUN], 0);
1659   }
1660 }
1661
1662 static gboolean
1663 single_queue_check_full (GstDataQueue * dataq, guint visible, guint bytes,
1664     guint64 time, GstSingleQueue * sq)
1665 {
1666   gboolean res;
1667   GstMultiQueue *mq = sq->mqueue;
1668
1669   GST_DEBUG_OBJECT (mq,
1670       "queue %d: visible %u/%u, bytes %u/%u, time %" G_GUINT64_FORMAT "/%"
1671       G_GUINT64_FORMAT, sq->id, visible, sq->max_size.visible, bytes,
1672       sq->max_size.bytes, sq->cur_time, sq->max_size.time);
1673
1674   /* we are always filled on EOS */
1675   if (sq->is_eos)
1676     return TRUE;
1677
1678   /* we never go past the max visible items unless we are in buffering mode */
1679   if (!mq->use_buffering && IS_FILLED (sq, visible, visible))
1680     return TRUE;
1681
1682   /* check time or bytes */
1683   res = IS_FILLED (sq, time, sq->cur_time) || IS_FILLED (sq, bytes, bytes);
1684
1685   return res;
1686 }
1687
1688 static void
1689 gst_single_queue_free (GstSingleQueue * sq)
1690 {
1691   /* DRAIN QUEUE */
1692   gst_data_queue_flush (sq->queue);
1693   g_object_unref (sq->queue);
1694   g_cond_free (sq->turn);
1695   g_free (sq);
1696 }
1697
1698 static GstSingleQueue *
1699 gst_single_queue_new (GstMultiQueue * mqueue, gint id)
1700 {
1701   GstSingleQueue *sq;
1702   gchar *name;
1703   GList *tmp;
1704   gint temp_id = (id == -1) ? 0 : id;
1705
1706   GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);
1707
1708   /* Find an unused queue ID, if possible the passed one */
1709   for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {
1710     GstSingleQueue *sq2 = (GstSingleQueue *) tmp->data;
1711     /* This works because the IDs are sorted in ascending order */
1712     if (sq2->id == temp_id) {
1713       /* If this ID was requested by the caller return NULL,
1714        * otherwise just get us the next one */
1715       if (id == -1)
1716         temp_id = sq2->id + 1;
1717       else
1718         return NULL;
1719     } else if (sq2->id > temp_id) {
1720       break;
1721     }
1722   }
1723
1724   sq = g_new0 (GstSingleQueue, 1);
1725   mqueue->nbqueues++;
1726   sq->id = temp_id;
1727
1728   mqueue->queues = g_list_insert_before (mqueue->queues, tmp, sq);
1729   mqueue->queues_cookie++;
1730
1731   /* copy over max_size and extra_size so we don't need to take the lock
1732    * any longer when checking if the queue is full. */
1733   sq->max_size.visible = mqueue->max_size.visible;
1734   sq->max_size.bytes = mqueue->max_size.bytes;
1735   sq->max_size.time = mqueue->max_size.time;
1736
1737   sq->extra_size.visible = mqueue->extra_size.visible;
1738   sq->extra_size.bytes = mqueue->extra_size.bytes;
1739   sq->extra_size.time = mqueue->extra_size.time;
1740
1741   GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);
1742
1743   GST_DEBUG_OBJECT (mqueue, "Creating GstSingleQueue id:%d", sq->id);
1744
1745   sq->mqueue = mqueue;
1746   sq->srcresult = GST_FLOW_WRONG_STATE;
1747   sq->queue = gst_data_queue_new_full ((GstDataQueueCheckFullFunction)
1748       single_queue_check_full,
1749       (GstDataQueueFullCallback) single_queue_overrun_cb,
1750       (GstDataQueueEmptyCallback) single_queue_underrun_cb, sq);
1751   sq->is_eos = FALSE;
1752   sq->flushing = FALSE;
1753   gst_segment_init (&sq->sink_segment, GST_FORMAT_TIME);
1754   gst_segment_init (&sq->src_segment, GST_FORMAT_TIME);
1755
1756   sq->nextid = 0;
1757   sq->oldid = 0;
1758   sq->turn = g_cond_new ();
1759
1760   sq->sinktime = GST_CLOCK_TIME_NONE;
1761   sq->srctime = GST_CLOCK_TIME_NONE;
1762   sq->sink_tainted = TRUE;
1763   sq->src_tainted = TRUE;
1764
1765   name = g_strdup_printf ("sink%d", sq->id);
1766   sq->sinkpad = gst_pad_new_from_static_template (&sinktemplate, name);
1767   g_free (name);
1768
1769   gst_pad_set_chain_function (sq->sinkpad,
1770       GST_DEBUG_FUNCPTR (gst_multi_queue_chain));
1771   gst_pad_set_activatepush_function (sq->sinkpad,
1772       GST_DEBUG_FUNCPTR (gst_multi_queue_sink_activate_push));
1773   gst_pad_set_event_function (sq->sinkpad,
1774       GST_DEBUG_FUNCPTR (gst_multi_queue_sink_event));
1775   gst_pad_set_getcaps_function (sq->sinkpad,
1776       GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1777   gst_pad_set_acceptcaps_function (sq->sinkpad,
1778       GST_DEBUG_FUNCPTR (gst_multi_queue_acceptcaps));
1779   gst_pad_set_iterate_internal_links_function (sq->sinkpad,
1780       GST_DEBUG_FUNCPTR (gst_multi_queue_iterate_internal_links));
1781
1782   name = g_strdup_printf ("src%d", sq->id);
1783   sq->srcpad = gst_pad_new_from_static_template (&srctemplate, name);
1784   g_free (name);
1785
1786   gst_pad_set_activatepush_function (sq->srcpad,
1787       GST_DEBUG_FUNCPTR (gst_multi_queue_src_activate_push));
1788   gst_pad_set_getcaps_function (sq->srcpad,
1789       GST_DEBUG_FUNCPTR (gst_multi_queue_getcaps));
1790   gst_pad_set_acceptcaps_function (sq->srcpad,
1791       GST_DEBUG_FUNCPTR (gst_multi_queue_acceptcaps));
1792   gst_pad_set_event_function (sq->srcpad,
1793       GST_DEBUG_FUNCPTR (gst_multi_queue_src_event));
1794   gst_pad_set_query_function (sq->srcpad,
1795       GST_DEBUG_FUNCPTR (gst_multi_queue_src_query));
1796   gst_pad_set_iterate_internal_links_function (sq->srcpad,
1797       GST_DEBUG_FUNCPTR (gst_multi_queue_iterate_internal_links));
1798
1799   gst_pad_set_element_private (sq->sinkpad, (gpointer) sq);
1800   gst_pad_set_element_private (sq->srcpad, (gpointer) sq);
1801
1802   /* only activate the pads when we are not in the NULL state
1803    * and add the pad under the state_lock to prevend state changes
1804    * between activating and adding */
1805   g_static_rec_mutex_lock (GST_STATE_GET_LOCK (mqueue));
1806   if (GST_STATE_TARGET (mqueue) != GST_STATE_NULL) {
1807     gst_pad_set_active (sq->srcpad, TRUE);
1808     gst_pad_set_active (sq->sinkpad, TRUE);
1809   }
1810   gst_element_add_pad (GST_ELEMENT (mqueue), sq->srcpad);
1811   gst_element_add_pad (GST_ELEMENT (mqueue), sq->sinkpad);
1812   g_static_rec_mutex_unlock (GST_STATE_GET_LOCK (mqueue));
1813
1814   GST_DEBUG_OBJECT (mqueue, "GstSingleQueue [%d] created and pads added",
1815       sq->id);
1816
1817   return sq;
1818 }