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