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 #ifdef __SYMBIAN32__

   116 #include <glib_global.h>

   117 #include <gobject_global.h>

   118 #endif

   119 

   120 /**

   121  * GstSingleQueue:

   122  * @sinkpad: associated sink #GstPad

   123  * @srcpad: associated source #GstPad

   124  *

   125  * Structure containing all information and properties about

   126  * a single queue.

   127  */

   128 typedef struct _GstSingleQueue GstSingleQueue;

   129 

   130 struct _GstSingleQueue

   131 {

   132   /* unique identifier of the queue */

   133   guint id;

   134 

   135   GstMultiQueue *mqueue;

   136 

   137   GstPad *sinkpad;

   138   GstPad *srcpad;

   139 

   140   /* flowreturn of previous srcpad push */

   141   GstFlowReturn srcresult;

   142   GstSegment sink_segment;

   143   GstSegment src_segment;

   144 

   145   /* queue of data */

   146   GstDataQueue *queue;

   147   GstDataQueueSize max_size, extra_size;

   148   GstClockTime cur_time;

   149   gboolean is_eos;

   150   gboolean inextra;             /* TRUE if the queue is currently in extradata mode */

   151 

   152   /* Protected by global lock */

   153   guint32 nextid;               /* ID of the next object waiting to be pushed */

   154   guint32 oldid;                /* ID of the last object pushed (last in a series) */

   155   GCond *turn;                  /* SingleQueue turn waiting conditional */

   156 };

   157 

   158 

   159 /* Extension of GstDataQueueItem structure for our usage */

   160 typedef struct _GstMultiQueueItem GstMultiQueueItem;

   161 

   162 struct _GstMultiQueueItem

   163 {

   164   GstMiniObject *object;

   165   guint size;

   166   guint64 duration;

   167   gboolean visible;

   168 

   169   GDestroyNotify destroy;

   170   guint32 posid;

   171 };

   172 

   173 static GstSingleQueue *gst_single_queue_new (GstMultiQueue * mqueue);

   174 static void gst_single_queue_free (GstSingleQueue * squeue);

   175 

   176 static void wake_up_next_non_linked (GstMultiQueue * mq);

   177 static void compute_high_id (GstMultiQueue * mq);

   178 

   179 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",

   180     GST_PAD_SINK,

   181     GST_PAD_REQUEST,

   182     GST_STATIC_CAPS_ANY);

   183 

   184 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",

   185     GST_PAD_SRC,

   186     GST_PAD_SOMETIMES,

   187     GST_STATIC_CAPS_ANY);

   188 

   189 GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);

   190 #define GST_CAT_DEFAULT (multi_queue_debug)

   191 

   192 /* default limits, we try to keep up to 2 seconds of data and if there is not

   193  * time, up to 10 MB. The number of buffers is dynamically scaled to make sure

   194  * there is data in the queues. Normally, the byte and time limits are not hit

   195  * in theses conditions. */

   196 #define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */

   197 #define DEFAULT_MAX_SIZE_BUFFERS 5

   198 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND

   199 

   200 /* second limits. When we hit one of the above limits we are probably dealing

   201  * with a badly muxed file and we scale the limits to these emergency values.

   202  * This is currently not yet implemented. */

   203 #define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024       /* 10 MB */

   204 #define DEFAULT_EXTRA_SIZE_BUFFERS 5

   205 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND

   206 

   207 /* Signals and args */

   208 enum

   209 {

   210   SIGNAL_UNDERRUN,

   211   SIGNAL_OVERRUN,

   212   LAST_SIGNAL

   213 };

   214 

   215 enum

   216 {

   217   ARG_0,

   218   ARG_EXTRA_SIZE_BYTES,

   219   ARG_EXTRA_SIZE_BUFFERS,

   220   ARG_EXTRA_SIZE_TIME,

   221   ARG_MAX_SIZE_BYTES,

   222   ARG_MAX_SIZE_BUFFERS,

   223   ARG_MAX_SIZE_TIME,

   224 };

   225 

   226 #define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START {                          \

   227   g_mutex_lock (q->qlock);                                              \

   228 } G_STMT_END

   229 

   230 #define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                        \

   231   g_mutex_unlock (q->qlock);                                            \

   232 } G_STMT_END

   233 

   234 static void gst_multi_queue_finalize (GObject * object);

   235 static void gst_multi_queue_set_property (GObject * object,

   236     guint prop_id, const GValue * value, GParamSpec * pspec);

   237 static void gst_multi_queue_get_property (GObject * object,

   238     guint prop_id, GValue * value, GParamSpec * pspec);

   239 

   240 static GstPad *gst_multi_queue_request_new_pad (GstElement * element,

   241     GstPadTemplate * temp, const gchar * name);

   242 static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);

   243 

   244 static void gst_multi_queue_loop (GstPad * pad);

   245 

   246 #define _do_init(bla) \

   247   GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");

   248 

   249 GST_BOILERPLATE_FULL (GstMultiQueue, gst_multi_queue, GstElement,

   250     GST_TYPE_ELEMENT, _do_init);

   251 

   252 static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };

   253 

   254 static void

   255 gst_multi_queue_base_init (gpointer g_class)

   256 {

   257   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);

   258 

   259   gst_element_class_set_details_simple (gstelement_class,

   260       "MultiQueue",

   261       "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");

   262   gst_element_class_add_pad_template (gstelement_class,

   263       gst_static_pad_template_get (&sinktemplate));

   264   gst_element_class_add_pad_template (gstelement_class,

   265       gst_static_pad_template_get (&srctemplate));

   266 }

   267 

   268 static void

   269 gst_multi_queue_class_init (GstMultiQueueClass * klass)

   270 {

   271   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

   272   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);

   273 

   274   gobject_class->set_property =

   275       GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);

   276   gobject_class->get_property =

   277       GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);

   278 

   279   /* SIGNALS */

   280   gst_multi_queue_signals[SIGNAL_UNDERRUN] =

   281       g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

   282       G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,

   283       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

   284 

   285   gst_multi_queue_signals[SIGNAL_OVERRUN] =

   286       g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

   287       G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,

   288       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

   289 

   290   /* PROPERTIES */

   291 

   292   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,

   293       g_param_spec_uint ("max-size-bytes", "Max. size (kB)",

   294           "Max. amount of data in the queue (bytes, 0=disable)",

   295           0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES, G_PARAM_READWRITE));

   296   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,

   297       g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",

   298           "Max. number of buffers in the queue (0=disable)",

   299           0, G_MAXUINT, DEFAULT_MAX_SIZE_BUFFERS, G_PARAM_READWRITE));

   300   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,

   301       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",

   302           "Max. amount of data in the queue (in ns, 0=disable)",

   303           0, G_MAXUINT64, DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE));

   304 

   305   g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,

   306       g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",

   307           "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",

   308           0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BYTES, G_PARAM_READWRITE));

   309   g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,

   310       g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",

   311           "Amount of buffers the queues can grow if one of them is empty (0=disable)",

   312           0, G_MAXUINT, DEFAULT_EXTRA_SIZE_BUFFERS, G_PARAM_READWRITE));

   313   g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,

   314       g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",

   315           "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",

   316           0, G_MAXUINT64, DEFAULT_EXTRA_SIZE_TIME, G_PARAM_READWRITE));

   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;