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