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