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