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