plugins/elements/gstmultiqueue.c

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