gst/gstqueue.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *                    2003 Colin Walters <cwalters@gnome.org>
   5  *                    2005 Wim Taymans <wim@fluendo.com>
   6  *
   7  * gstqueue.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  * SECTION:gstqueue
  26  * @short_description: Simple asynchronous data queue.
  27  *
  28  * Data is queued till max_level buffers have been stored. Any subsequent
  29  * buffers sent to this filter will block until free space becomes available in
  30  * the buffer. The queue is typically used in conjunction with a thread.
  31  *
  32  * You can query how many buffers are queued with the level argument.
  33  *
  34  * The default queue length is set to 100.
  35  *
  36  * The queue blocks by default.
  37  */
  38
  39
  40 #include "gst_private.h"
  41
  42 #include "gstqueue.h"
  43 #include "gstevent.h"
  44 #include "gstinfo.h"
  45 #include "gsterror.h"
  46 #include "gstutils.h"
  47
  48 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  49     GST_PAD_SINK,
  50     GST_PAD_ALWAYS,
  51     GST_STATIC_CAPS_ANY);
  52
  53 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  54     GST_PAD_SRC,
  55     GST_PAD_ALWAYS,
  56     GST_STATIC_CAPS_ANY);
  57
  58 GST_DEBUG_CATEGORY_STATIC (queue_dataflow);
  59 #define GST_CAT_DEFAULT (queue_dataflow)
  60
  61 #define STATUS(queue, msg) \
  62   GST_CAT_LOG_OBJECT (queue_dataflow, queue, \
  63                       "(%s:%s) " msg ": %u of %u-%u buffers, %u of %u-%u " \
  64                       "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
  65                       "-%" G_GUINT64_FORMAT " ns, %u elements", \
  66                       GST_DEBUG_PAD_NAME (pad), \
  67                       queue->cur_level.buffers, \
  68                       queue->min_threshold.buffers, \
  69                       queue->max_size.buffers, \
  70                       queue->cur_level.bytes, \
  71                       queue->min_threshold.bytes, \
  72                       queue->max_size.bytes, \
  73                       queue->cur_level.time, \
  74                       queue->min_threshold.time, \
  75                       queue->max_size.time, \
  76                       queue->queue->length)
  77
  78 static GstElementDetails gst_queue_details = GST_ELEMENT_DETAILS ("Queue",
  79     "Generic",
  80     "Simple data queue",
  81     "Erik Walthinsen <omega@cse.ogi.edu>");
  82
  83
  84 /* Queue signals and args */
  85 enum
  86 {
  87   SIGNAL_UNDERRUN,
  88   SIGNAL_RUNNING,
  89   SIGNAL_OVERRUN,
  90   LAST_SIGNAL
  91 };
  92
  93 enum
  94 {
  95   ARG_0,
  96   /* FIXME: don't we have another way of doing this
  97    * "Gstreamer format" (frame/byte/time) queries? */
  98   ARG_CUR_LEVEL_BUFFERS,
  99   ARG_CUR_LEVEL_BYTES,
 100   ARG_CUR_LEVEL_TIME,
 101   ARG_MAX_SIZE_BUFFERS,
 102   ARG_MAX_SIZE_BYTES,
 103   ARG_MAX_SIZE_TIME,
 104   ARG_MIN_THRESHOLD_BUFFERS,
 105   ARG_MIN_THRESHOLD_BYTES,
 106   ARG_MIN_THRESHOLD_TIME,
 107   ARG_LEAKY,
 108   ARG_MAY_DEADLOCK,
 109   ARG_BLOCK_TIMEOUT
 110       /* FILL ME */
 111 };
 112
 113 #define GST_QUEUE_MUTEX_LOCK(q) G_STMT_START {                          \
 114   GST_CAT_LOG_OBJECT (queue_dataflow, q,                                \
 115       "locking qlock from thread %p",                                   \
 116       g_thread_self ());                                                \
 117   g_mutex_lock (q->qlock);                                              \
 118   GST_CAT_LOG_OBJECT (queue_dataflow, q,                                \
 119       "locked qlock from thread %p",                                    \
 120       g_thread_self ());                                                \
 121 } G_STMT_END
 122
 123 #define GST_QUEUE_MUTEX_LOCK_CHECK(q,label) G_STMT_START {              \
 124   GST_QUEUE_MUTEX_LOCK (q);                                             \
 125   if (q->srcresult != GST_FLOW_OK)                                      \
 126     goto label;                                                         \
 127 } G_STMT_END
 128
 129 #define GST_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                        \
 130   GST_CAT_LOG_OBJECT (queue_dataflow, q,                                \
 131       "unlocking qlock from thread %p",                                 \
 132       g_thread_self ());                                                \
 133   g_mutex_unlock (q->qlock);                                            \
 134 } G_STMT_END
 135
 136
 137 static void gst_queue_base_init (GstQueueClass * klass);
 138 static void gst_queue_class_init (GstQueueClass * klass);
 139 static void gst_queue_init (GstQueue * queue);
 140 static void gst_queue_finalize (GObject * object);
 141
 142 static void gst_queue_set_property (GObject * object,
 143     guint prop_id, const GValue * value, GParamSpec * pspec);
 144 static void gst_queue_get_property (GObject * object,
 145     guint prop_id, GValue * value, GParamSpec * pspec);
 146
 147 static GstFlowReturn gst_queue_chain (GstPad * pad, GstBuffer * buffer);
 148 static GstFlowReturn gst_queue_bufferalloc (GstPad * pad, guint64 offset,
 149     guint size, GstCaps * caps, GstBuffer ** buf);
 150 static void gst_queue_loop (GstPad * pad);
 151
 152 static gboolean gst_queue_handle_sink_event (GstPad * pad, GstEvent * event);
 153
 154 static gboolean gst_queue_handle_src_event (GstPad * pad, GstEvent * event);
 155 static gboolean gst_queue_handle_src_query (GstPad * pad, GstQuery * query);
 156
 157 static GstCaps *gst_queue_getcaps (GstPad * pad);
 158 static GstPadLinkReturn gst_queue_link_sink (GstPad * pad, GstPad * peer);
 159 static GstPadLinkReturn gst_queue_link_src (GstPad * pad, GstPad * peer);
 160 static void gst_queue_locked_flush (GstQueue * queue);
 161
 162 static gboolean gst_queue_src_activate_push (GstPad * pad, gboolean active);
 163 static gboolean gst_queue_sink_activate_push (GstPad * pad, gboolean active);
 164 static GstStateChangeReturn gst_queue_change_state (GstElement * element,
 165     GstStateChange transition);
 166
 167
 168 #define GST_TYPE_QUEUE_LEAKY (queue_leaky_get_type ())
 169
 170 static GType
 171 queue_leaky_get_type (void)
 172 {
 173   static GType queue_leaky_type = 0;
 174   static GEnumValue queue_leaky[] = {
 175     {GST_QUEUE_NO_LEAK, "0", "Not Leaky"},
 176     {GST_QUEUE_LEAK_UPSTREAM, "1", "Leaky on Upstream"},
 177     {GST_QUEUE_LEAK_DOWNSTREAM, "2", "Leaky on Downstream"},
 178     {0, NULL, NULL},
 179   };
 180
 181   if (!queue_leaky_type) {
 182     queue_leaky_type = g_enum_register_static ("GstQueueLeaky", queue_leaky);
 183   }
 184   return queue_leaky_type;
 185 }
 186
 187 static GstElementClass *parent_class = NULL;
 188 static guint gst_queue_signals[LAST_SIGNAL] = { 0 };
 189
 190 GType
 191 gst_queue_get_type (void)
 192 {
 193   static GType queue_type = 0;
 194
 195   if (!queue_type) {
 196     static const GTypeInfo queue_info = {
 197       sizeof (GstQueueClass),
 198       (GBaseInitFunc) gst_queue_base_init,
 199       NULL,
 200       (GClassInitFunc) gst_queue_class_init,
 201       NULL,
 202       NULL,
 203       sizeof (GstQueue),
 204       0,
 205       (GInstanceInitFunc) gst_queue_init,
 206       NULL
 207     };
 208
 209     queue_type = g_type_register_static (GST_TYPE_ELEMENT,
 210         "GstQueue", &queue_info, 0);
 211     GST_DEBUG_CATEGORY_INIT (queue_dataflow, "queue_dataflow", 0,
 212         "dataflow inside the queue element");
 213   }
 214
 215   return queue_type;
 216 }
 217
 218 static void
 219 gst_queue_base_init (GstQueueClass * klass)
 220 {
 221   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 222
 223   gst_element_class_add_pad_template (gstelement_class,
 224       gst_static_pad_template_get (&srctemplate));
 225   gst_element_class_add_pad_template (gstelement_class,
 226       gst_static_pad_template_get (&sinktemplate));
 227   gst_element_class_set_details (gstelement_class, &gst_queue_details);
 228 }
 229
 230 static void
 231 gst_queue_class_init (GstQueueClass * klass)
 232 {
 233   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 234   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 235
 236   parent_class = g_type_class_peek_parent (klass);
 237
 238   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_queue_set_property);
 239   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_queue_get_property);
 240
 241   /* signals */
 242   /**
 243    * GstQueue::underrun:
 244    * @queue: the queue instance
 245    *
 246    * Reports that the buffer became empty (underrun).
 247    * A buffer is empty if the total amount of data inside it (num-buffers, time,
 248    * size) is lower than the boundary values which can be set through the GObject
 249    * properties.
 250    */
 251   gst_queue_signals[SIGNAL_UNDERRUN] =
 252       g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 253       G_STRUCT_OFFSET (GstQueueClass, underrun), NULL, NULL,
 254       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 255   /**
 256    * GstQueue::running:
 257    * @queue: the queue instance
 258    *
 259    * Reports that enough (min-threshold) data is in the queue. Use this signal
 260    * together with the underrun signal to pause the pipeline on underrun and wait
 261    * for the queue to fill-up before resume playback.
 262    */
 263   gst_queue_signals[SIGNAL_RUNNING] =
 264       g_signal_new ("running", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 265       G_STRUCT_OFFSET (GstQueueClass, running), NULL, NULL,
 266       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 267   /**
 268    * GstQueue::overrun:
 269    * @queue: the queue instance
 270    *
 271    * Reports that the buffer became full (overrun).
 272    * A buffer is full if the total amount of data inside it (num-buffers, time,
 273    * size) is higher than the boundary values which can be set through the GObject
 274    * properties.
 275    */
 276   gst_queue_signals[SIGNAL_OVERRUN] =
 277       g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 278       G_STRUCT_OFFSET (GstQueueClass, overrun), NULL, NULL,
 279       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 280
 281   /* properties */
 282   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BYTES,
 283       g_param_spec_uint ("current-level-bytes", "Current level (kB)",
 284           "Current amount of data in the queue (bytes)",
 285           0, G_MAXUINT, 0, G_PARAM_READABLE));
 286   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BUFFERS,
 287       g_param_spec_uint ("current-level-buffers", "Current level (buffers)",
 288           "Current number of buffers in the queue",
 289           0, G_MAXUINT, 0, G_PARAM_READABLE));
 290   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_TIME,
 291       g_param_spec_uint64 ("current-level-time", "Current level (ns)",
 292           "Current amount of data in the queue (in ns)",
 293           0, G_MAXUINT64, 0, G_PARAM_READABLE));
 294
 295   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
 296       g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
 297           "Max. amount of data in the queue (bytes, 0=disable)",
 298           0, G_MAXUINT, 0, G_PARAM_READWRITE));
 299   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
 300       g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
 301           "Max. number of buffers in the queue (0=disable)",
 302           0, G_MAXUINT, 0, G_PARAM_READWRITE));
 303   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
 304       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 305           "Max. amount of data in the queue (in ns, 0=disable)",
 306           0, G_MAXUINT64, 0, G_PARAM_READWRITE));
 307
 308   g_object_class_install_property (gobject_class, ARG_MIN_THRESHOLD_BYTES,
 309       g_param_spec_uint ("min-threshold-bytes", "Min. threshold (kB)",
 310           "Min. amount of data in the queue to allow reading (bytes, 0=disable)",
 311           0, G_MAXUINT, 0, G_PARAM_READWRITE));
 312   g_object_class_install_property (gobject_class, ARG_MIN_THRESHOLD_BUFFERS,
 313       g_param_spec_uint ("min-threshold-buffers", "Min. threshold (buffers)",
 314           "Min. number of buffers in the queue to allow reading (0=disable)",
 315           0, G_MAXUINT, 0, G_PARAM_READWRITE));
 316   g_object_class_install_property (gobject_class, ARG_MIN_THRESHOLD_TIME,
 317       g_param_spec_uint64 ("min-threshold-time", "Min. threshold (ns)",
 318           "Min. amount of data in the queue to allow reading (in ns, 0=disable)",
 319           0, G_MAXUINT64, 0, G_PARAM_READWRITE));
 320
 321   g_object_class_install_property (gobject_class, ARG_LEAKY,
 322       g_param_spec_enum ("leaky", "Leaky",
 323           "Where the queue leaks, if at all",
 324           GST_TYPE_QUEUE_LEAKY, GST_QUEUE_NO_LEAK, G_PARAM_READWRITE));
 325   g_object_class_install_property (gobject_class, ARG_MAY_DEADLOCK,
 326       g_param_spec_boolean ("may_deadlock", "May Deadlock",
 327           "The queue may deadlock if it's full and not PLAYING",
 328           TRUE, G_PARAM_READWRITE));
 329   g_object_class_install_property (gobject_class, ARG_BLOCK_TIMEOUT,
 330       g_param_spec_uint64 ("block_timeout", "Timeout for Block",
 331           "Nanoseconds until blocked queue times out and returns filler event. "
 332           "Value of -1 disables timeout",
 333           0, G_MAXUINT64, -1, G_PARAM_READWRITE));
 334
 335   /* set several parent class virtual functions */
 336   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_queue_finalize);
 337
 338   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_queue_change_state);
 339 }
 340
 341 static void
 342 gst_queue_init (GstQueue * queue)
 343 {
 344   queue->sinkpad =
 345       gst_pad_new_from_template (gst_static_pad_template_get (&sinktemplate),
 346       "sink");
 347   gst_pad_set_chain_function (queue->sinkpad,
 348       GST_DEBUG_FUNCPTR (gst_queue_chain));
 349   gst_pad_set_activatepush_function (queue->sinkpad,
 350       GST_DEBUG_FUNCPTR (gst_queue_sink_activate_push));
 351   gst_pad_set_event_function (queue->sinkpad,
 352       GST_DEBUG_FUNCPTR (gst_queue_handle_sink_event));
 353   gst_pad_set_link_function (queue->sinkpad,
 354       GST_DEBUG_FUNCPTR (gst_queue_link_sink));
 355   gst_pad_set_getcaps_function (queue->sinkpad,
 356       GST_DEBUG_FUNCPTR (gst_queue_getcaps));
 357   gst_pad_set_bufferalloc_function (queue->sinkpad,
 358       GST_DEBUG_FUNCPTR (gst_queue_bufferalloc));
 359   gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
 360
 361   queue->srcpad =
 362       gst_pad_new_from_template (gst_static_pad_template_get (&srctemplate),
 363       "src");
 364   gst_pad_set_activatepush_function (queue->srcpad,
 365       GST_DEBUG_FUNCPTR (gst_queue_src_activate_push));
 366   gst_pad_set_link_function (queue->srcpad,
 367       GST_DEBUG_FUNCPTR (gst_queue_link_src));
 368   gst_pad_set_getcaps_function (queue->srcpad,
 369       GST_DEBUG_FUNCPTR (gst_queue_getcaps));
 370   gst_pad_set_event_function (queue->srcpad,
 371       GST_DEBUG_FUNCPTR (gst_queue_handle_src_event));
 372   gst_pad_set_query_function (queue->srcpad,
 373       GST_DEBUG_FUNCPTR (gst_queue_handle_src_query));
 374   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 375
 376   queue->cur_level.buffers = 0; /* no content */
 377   queue->cur_level.bytes = 0;   /* no content */
 378   queue->cur_level.time = 0;    /* no content */
 379   queue->max_size.buffers = 200;        /* 200 buffers */
 380   queue->max_size.bytes = 10 * 1024 * 1024;     /* 10 MB */
 381   queue->max_size.time = GST_SECOND;    /* 1 s. */
 382   queue->min_threshold.buffers = 0;     /* no threshold */
 383   queue->min_threshold.bytes = 0;       /* no threshold */
 384   queue->min_threshold.time = 0;        /* no threshold */
 385
 386   queue->leaky = GST_QUEUE_NO_LEAK;
 387   queue->may_deadlock = TRUE;
 388   queue->block_timeout = GST_CLOCK_TIME_NONE;
 389   queue->srcresult = GST_FLOW_WRONG_STATE;
 390
 391   queue->qlock = g_mutex_new ();
 392   queue->item_add = g_cond_new ();
 393   queue->item_del = g_cond_new ();
 394   queue->queue = g_queue_new ();
 395
 396   GST_CAT_DEBUG_OBJECT (GST_CAT_THREAD, queue,
 397       "initialized queue's not_empty & not_full conditions");
 398 }
 399
 400 /* called only once, as opposed to dispose */
 401 static void
 402 gst_queue_finalize (GObject * object)
 403 {
 404   GstQueue *queue = GST_QUEUE (object);
 405
 406   GST_DEBUG_OBJECT (queue, "finalizing queue");
 407
 408   while (!g_queue_is_empty (queue->queue)) {
 409     GstMiniObject *data = g_queue_pop_head (queue->queue);
 410
 411     gst_mini_object_unref (data);
 412   }
 413   g_queue_free (queue->queue);
 414   GST_CAT_DEBUG_OBJECT (GST_CAT_THREAD, queue, "free mutex");
 415   g_mutex_free (queue->qlock);
 416   GST_CAT_DEBUG_OBJECT (GST_CAT_THREAD, queue, "done free mutex");
 417   g_cond_free (queue->item_add);
 418   g_cond_free (queue->item_del);
 419
 420   if (G_OBJECT_CLASS (parent_class)->finalize)
 421     G_OBJECT_CLASS (parent_class)->finalize (object);
 422 }
 423
 424 static GstCaps *
 425 gst_queue_getcaps (GstPad * pad)
 426 {
 427   GstQueue *queue;
 428   GstPad *otherpad;
 429   GstCaps *result;
 430
 431   queue = GST_QUEUE (GST_PAD_PARENT (pad));
 432
 433   otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);
 434   result = gst_pad_peer_get_caps (otherpad);
 435   if (result == NULL)
 436     result = gst_caps_new_any ();
 437
 438   return result;
 439 }
 440
 441 static GstPadLinkReturn
 442 gst_queue_link_sink (GstPad * pad, GstPad * peer)
 443 {
 444   return GST_PAD_LINK_OK;
 445 }
 446
 447 static GstPadLinkReturn
 448 gst_queue_link_src (GstPad * pad, GstPad * peer)
 449 {
 450   GstPadLinkReturn result = GST_PAD_LINK_OK;
 451   GstQueue *queue;
 452
 453   queue = GST_QUEUE (gst_pad_get_parent (pad));
 454
 455   GST_DEBUG ("queue linking source pad");
 456
 457   if (GST_PAD_LINKFUNC (peer)) {
 458     result = GST_PAD_LINKFUNC (peer) (peer, pad);
 459   }
 460
 461   if (GST_PAD_LINK_SUCCESSFUL (result)) {
 462     GST_QUEUE_MUTEX_LOCK (queue);
 463     if (queue->srcresult == GST_FLOW_OK) {
 464       gst_pad_start_task (pad, (GstTaskFunction) gst_queue_loop, pad);
 465       GST_DEBUG ("starting task as pad is linked");
 466     } else {
 467       GST_DEBUG ("not starting task reason %s",
 468           gst_flow_get_name (queue->srcresult));
 469     }
 470     GST_QUEUE_MUTEX_UNLOCK (queue);
 471   }
 472   gst_object_unref (queue);
 473
 474   return result;
 475 }
 476
 477 static GstFlowReturn
 478 gst_queue_bufferalloc (GstPad * pad, guint64 offset, guint size, GstCaps * caps,
 479     GstBuffer ** buf)
 480 {
 481   GstQueue *queue;
 482   GstFlowReturn result;
 483
 484   queue = GST_QUEUE (GST_PAD_PARENT (pad));
 485
 486   result = gst_pad_alloc_buffer (queue->srcpad, offset, size, caps, buf);
 487
 488   return result;
 489 }
 490
 491
 492 static void
 493 gst_queue_locked_flush (GstQueue * queue)
 494 {
 495   while (!g_queue_is_empty (queue->queue)) {
 496     GstMiniObject *data = g_queue_pop_head (queue->queue);
 497
 498     /* Then loose another reference because we are supposed to destroy that
 499        data when flushing */
 500     gst_mini_object_unref (data);
 501   }
 502   queue->cur_level.buffers = 0;
 503   queue->cur_level.bytes = 0;
 504   queue->cur_level.time = 0;
 505
 506   /* we deleted something... */
 507   g_cond_signal (queue->item_del);
 508 }
 509
 510 #define STATUS(queue, msg) \
 511   GST_CAT_LOG_OBJECT (queue_dataflow, queue, \
 512                       "(%s:%s) " msg ": %u of %u-%u buffers, %u of %u-%u " \
 513                       "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
 514                       "-%" G_GUINT64_FORMAT " ns, %u elements", \
 515                       GST_DEBUG_PAD_NAME (pad), \
 516                       queue->cur_level.buffers, \
 517                       queue->min_threshold.buffers, \
 518                       queue->max_size.buffers, \
 519                       queue->cur_level.bytes, \
 520                       queue->min_threshold.bytes, \
 521                       queue->max_size.bytes, \
 522                       queue->cur_level.time, \
 523                       queue->min_threshold.time, \
 524                       queue->max_size.time, \
 525                       queue->queue->length)
 526
 527 static gboolean
 528 gst_queue_handle_sink_event (GstPad * pad, GstEvent * event)
 529 {
 530   GstQueue *queue;
 531
 532   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 533
 534   switch (GST_EVENT_TYPE (event)) {
 535     case GST_EVENT_FLUSH_START:
 536       STATUS (queue, "received flush start event");
 537       /* forward event, re first as we're going to use it still */
 538       gst_event_ref (event);
 539       gst_pad_push_event (queue->srcpad, event);
 540
 541       /* now unblock the chain function */
 542       GST_QUEUE_MUTEX_LOCK (queue);
 543       queue->srcresult = GST_FLOW_WRONG_STATE;
 544       /* unblock the loop and chain functions */
 545       g_cond_signal (queue->item_add);
 546       g_cond_signal (queue->item_del);
 547       GST_QUEUE_MUTEX_UNLOCK (queue);
 548
 549       /* make sure it pauses */
 550       gst_pad_pause_task (queue->srcpad);
 551       GST_CAT_LOG_OBJECT (queue_dataflow, queue, "loop stopped");
 552       gst_event_unref (event);
 553       goto done;
 554     case GST_EVENT_FLUSH_STOP:
 555       STATUS (queue, "received flush stop event");
 556       /* forward event, re first as we're going to use it still */
 557       gst_event_ref (event);
 558       gst_pad_push_event (queue->srcpad, event);
 559
 560       GST_QUEUE_MUTEX_LOCK (queue);
 561       gst_queue_locked_flush (queue);
 562       queue->srcresult = GST_FLOW_OK;
 563       if (gst_pad_is_linked (queue->srcpad)) {
 564         gst_pad_start_task (queue->srcpad, (GstTaskFunction) gst_queue_loop,
 565             queue->srcpad);
 566       } else {
 567         GST_DEBUG_OBJECT (queue, "not re-starting task as pad is not linked");
 568       }
 569       GST_QUEUE_MUTEX_UNLOCK (queue);
 570
 571       STATUS (queue, "after flush");
 572       gst_event_unref (event);
 573       goto done;
 574     case GST_EVENT_EOS:
 575       STATUS (queue, "received EOS");
 576       break;
 577     default:
 578       if (GST_EVENT_IS_SERIALIZED (event)) {
 579         /* we put the event in the queue, we don't have to act ourselves */
 580         GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 581             "adding event %p of type %d", event, GST_EVENT_TYPE (event));
 582       } else {
 583         gst_event_ref (event);
 584         gst_pad_push_event (queue->srcpad, event);
 585         gst_event_unref (event);
 586         goto done;
 587       }
 588       break;
 589   }
 590
 591   GST_QUEUE_MUTEX_LOCK (queue);
 592   g_queue_push_tail (queue->queue, event);
 593   g_cond_signal (queue->item_add);
 594   GST_QUEUE_MUTEX_UNLOCK (queue);
 595
 596 done:
 597
 598   return TRUE;
 599 }
 600
 601 static gboolean
 602 gst_queue_is_empty (GstQueue * queue)
 603 {
 604   return (queue->queue->length == 0 ||
 605       (queue->min_threshold.buffers > 0 &&
 606           queue->cur_level.buffers < queue->min_threshold.buffers) ||
 607       (queue->min_threshold.bytes > 0 &&
 608           queue->cur_level.bytes < queue->min_threshold.bytes) ||
 609       (queue->min_threshold.time > 0 &&
 610           queue->cur_level.time < queue->min_threshold.time));
 611 }
 612
 613 static gboolean
 614 gst_queue_is_filled (GstQueue * queue)
 615 {
 616   return (((queue->max_size.buffers > 0 &&
 617               queue->cur_level.buffers >= queue->max_size.buffers) ||
 618           (queue->max_size.bytes > 0 &&
 619               queue->cur_level.bytes >= queue->max_size.bytes) ||
 620           (queue->max_size.time > 0 &&
 621               queue->cur_level.time >= queue->max_size.time)));
 622 }
 623
 624
 625 static GstFlowReturn
 626 gst_queue_chain (GstPad * pad, GstBuffer * buffer)
 627 {
 628   GstQueue *queue;
 629
 630   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 631
 632   /* we have to lock the queue since we span threads */
 633   GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 634
 635   GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 636       "adding buffer %p of size %d", buffer, GST_BUFFER_SIZE (buffer));
 637
 638   /* We make space available if we're "full" according to whatever
 639    * the user defined as "full". Note that this only applies to buffers.
 640    * We always handle events and they don't count in our statistics. */
 641   while (gst_queue_is_filled (queue)) {
 642     GST_QUEUE_MUTEX_UNLOCK (queue);
 643     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_OVERRUN], 0);
 644     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 645
 646     /* how are we going to make space for this buffer? */
 647     switch (queue->leaky) {
 648         /* leak current buffer */
 649       case GST_QUEUE_LEAK_UPSTREAM:
 650         GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
 651             "queue is full, leaking buffer on upstream end");
 652         /* now we can clean up and exit right away */
 653         goto out_unref;
 654
 655         /* leak first buffer in the queue */
 656       case GST_QUEUE_LEAK_DOWNSTREAM:{
 657         /* this is a bit hacky. We'll manually iterate the list
 658          * and find the first buffer from the head on. We'll
 659          * unref that and "fix up" the GQueue object... */
 660         GList *item;
 661         GstMiniObject *leak = NULL;
 662
 663         GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
 664             "queue is full, leaking buffer on downstream end");
 665
 666         for (item = queue->queue->head; item != NULL; item = item->next) {
 667           if (GST_IS_BUFFER (item->data)) {
 668             leak = item->data;
 669             break;
 670           }
 671         }
 672
 673         /* if we didn't find anything, it means we have no buffers
 674          * in here. That cannot happen, since we had >= 1 bufs */
 675         g_assert (leak);
 676
 677         /* Now remove it from the list, fixing up the GQueue
 678          * CHECKME: is a queue->head the first or the last item? */
 679         item = g_list_delete_link (queue->queue->head, item);
 680         queue->queue->head = g_list_first (item);
 681         queue->queue->tail = g_list_last (item);
 682         queue->queue->length--;
 683
 684         /* and unref the buffer at the end. Twice, because we keep a ref
 685          * to make things read-only. Also keep our list uptodate. */
 686         queue->cur_level.bytes -= GST_BUFFER_SIZE (buffer);
 687         queue->cur_level.buffers--;
 688         if (GST_BUFFER_DURATION (buffer) != GST_CLOCK_TIME_NONE)
 689           queue->cur_level.time -= GST_BUFFER_DURATION (buffer);
 690
 691         gst_buffer_unref (buffer);
 692         gst_buffer_unref (buffer);
 693         break;
 694       }
 695
 696       default:
 697         g_warning ("Unknown leaky type, using default");
 698         /* fall-through */
 699
 700         /* don't leak. Instead, wait for space to be available */
 701       case GST_QUEUE_NO_LEAK:
 702         STATUS (queue, "pre-full wait");
 703
 704         while (gst_queue_is_filled (queue)) {
 705           STATUS (queue, "waiting for item_del signal from thread using qlock");
 706           g_cond_wait (queue->item_del, queue->qlock);
 707
 708           if (queue->srcresult != GST_FLOW_OK)
 709             goto out_flushing;
 710
 711           /* if there's a pending state change for this queue
 712            * or its manager, switch back to iterator so bottom
 713            * half of state change executes */
 714           STATUS (queue, "received item_del signal from thread using qlock");
 715         }
 716
 717         STATUS (queue, "post-full wait");
 718         GST_QUEUE_MUTEX_UNLOCK (queue);
 719         g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_RUNNING], 0);
 720         GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 721
 722         break;
 723     }
 724   }
 725
 726   g_queue_push_tail (queue->queue, buffer);
 727
 728   /* add buffer to the statistics */
 729   queue->cur_level.buffers++;
 730   queue->cur_level.bytes += GST_BUFFER_SIZE (buffer);
 731   if (GST_BUFFER_DURATION (buffer) != GST_CLOCK_TIME_NONE)
 732     queue->cur_level.time += GST_BUFFER_DURATION (buffer);
 733
 734   STATUS (queue, "+ level");
 735
 736   GST_CAT_LOG_OBJECT (queue_dataflow, queue, "signalling item_add");
 737   g_cond_signal (queue->item_add);
 738   GST_QUEUE_MUTEX_UNLOCK (queue);
 739
 740   return GST_FLOW_OK;
 741
 742   /* special conditions */
 743 out_unref:
 744   {
 745     GST_QUEUE_MUTEX_UNLOCK (queue);
 746
 747     gst_buffer_unref (buffer);
 748
 749     return GST_FLOW_OK;
 750   }
 751 out_flushing:
 752   {
 753     GstFlowReturn ret = queue->srcresult;
 754     const gchar *flowname = gst_flow_get_name (ret);
 755
 756     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 757         "exit because task paused, reason: %s", flowname);
 758     GST_QUEUE_MUTEX_UNLOCK (queue);
 759
 760     gst_buffer_unref (buffer);
 761
 762     return ret;
 763   }
 764 }
 765
 766 static void
 767 gst_queue_loop (GstPad * pad)
 768 {
 769   GstQueue *queue;
 770   GstMiniObject *data;
 771   gboolean restart = TRUE;
 772
 773   queue = GST_QUEUE (GST_PAD_PARENT (pad));
 774
 775   /* have to lock for thread-safety */
 776   GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 777
 778 restart:
 779   while (gst_queue_is_empty (queue)) {
 780     GST_QUEUE_MUTEX_UNLOCK (queue);
 781     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_UNDERRUN], 0);
 782     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 783
 784     STATUS (queue, "pre-empty wait");
 785     while (gst_queue_is_empty (queue)) {
 786       STATUS (queue, "waiting for item_add");
 787
 788       GST_LOG_OBJECT (queue, "doing g_cond_wait using qlock from thread %p",
 789           g_thread_self ());
 790       g_cond_wait (queue->item_add, queue->qlock);
 791
 792       /* we released the lock in the g_cond above so we might be
 793        * flushing now */
 794       if (queue->srcresult != GST_FLOW_OK)
 795         goto out_flushing;
 796
 797       GST_LOG_OBJECT (queue, "done g_cond_wait using qlock from thread %p",
 798           g_thread_self ());
 799       STATUS (queue, "got item_add signal");
 800     }
 801
 802     STATUS (queue, "post-empty wait");
 803     GST_QUEUE_MUTEX_UNLOCK (queue);
 804     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_RUNNING], 0);
 805     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 806   }
 807
 808   /* There's something in the list now, whatever it is */
 809   data = g_queue_pop_head (queue->queue);
 810   GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 811       "retrieved data %p from queue", data);
 812
 813   if (GST_IS_BUFFER (data)) {
 814     GstFlowReturn result;
 815
 816     /* Update statistics */
 817     queue->cur_level.buffers--;
 818     queue->cur_level.bytes -= GST_BUFFER_SIZE (data);
 819     if (GST_BUFFER_DURATION (data) != GST_CLOCK_TIME_NONE)
 820       queue->cur_level.time -= GST_BUFFER_DURATION (data);
 821
 822     GST_QUEUE_MUTEX_UNLOCK (queue);
 823     result = gst_pad_push (pad, GST_BUFFER (data));
 824     GST_QUEUE_MUTEX_LOCK (queue);
 825     /* can opt to check for srcresult here but the push should
 826      * return an error value that is more accurate */
 827     if (result != GST_FLOW_OK) {
 828       const gchar *flowname;
 829
 830       flowname = gst_flow_get_name (result);
 831
 832       queue->srcresult = result;
 833       if (GST_FLOW_IS_FATAL (result)) {
 834         GST_ELEMENT_ERROR (queue, STREAM, STOPPED,
 835             ("streaming stopped, reason %s", flowname),
 836             ("streaming stopped, reason %s", flowname));
 837         gst_pad_push_event (queue->srcpad, gst_event_new_eos ());
 838       }
 839       GST_DEBUG ("pausing queue, reason %s", flowname);
 840       gst_pad_pause_task (queue->srcpad);
 841     }
 842   } else {
 843     if (GST_EVENT_TYPE (data) == GST_EVENT_EOS) {
 844       /* all incomming data is now unexpected */
 845       queue->srcresult = GST_FLOW_UNEXPECTED;
 846       /* and we don't need to process anymore */
 847       GST_DEBUG ("pausing queue, we're EOS now");
 848       gst_pad_pause_task (queue->srcpad);
 849       restart = FALSE;
 850     }
 851     GST_QUEUE_MUTEX_UNLOCK (queue);
 852     gst_pad_push_event (queue->srcpad, GST_EVENT (data));
 853     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 854     if (restart == TRUE)
 855       goto restart;
 856   }
 857
 858   STATUS (queue, "after _get()");
 859
 860   GST_CAT_LOG_OBJECT (queue_dataflow, queue, "signalling item_del");
 861   g_cond_signal (queue->item_del);
 862   GST_QUEUE_MUTEX_UNLOCK (queue);
 863
 864   return;
 865
 866 out_flushing:
 867   {
 868     const gchar *flowname = gst_flow_get_name (queue->srcresult);
 869
 870     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 871         "exit because task paused, reason:  %s", flowname);
 872     GST_QUEUE_MUTEX_UNLOCK (queue);
 873
 874     return;
 875   }
 876 }
 877
 878
 879 static gboolean
 880 gst_queue_handle_src_event (GstPad * pad, GstEvent * event)
 881 {
 882   gboolean res = TRUE;
 883   GstQueue *queue = GST_QUEUE (GST_PAD_PARENT (pad));
 884
 885 #ifndef GST_DISABLE_GST_DEBUG
 886   GST_CAT_DEBUG_OBJECT (queue_dataflow, queue, "got event %p (%d)",
 887       event, GST_EVENT_TYPE (event));
 888 #endif
 889
 890   res = gst_pad_push_event (queue->sinkpad, event);
 891
 892   return res;
 893 }
 894
 895 static gboolean
 896 gst_queue_handle_src_query (GstPad * pad, GstQuery * query)
 897 {
 898   GstQueue *queue = GST_QUEUE (GST_PAD_PARENT (pad));
 899   GstPad *peer;
 900   gboolean res;
 901
 902   if (!(peer = gst_pad_get_peer (queue->sinkpad)))
 903     return FALSE;
 904
 905   res = gst_pad_query (peer, query);
 906   gst_object_unref (peer);
 907   if (!res)
 908     return FALSE;
 909
 910   switch (GST_QUERY_TYPE (query)) {
 911     case GST_QUERY_POSITION:
 912     {
 913       gint64 peer_pos, peer_total;
 914       GstFormat format;
 915
 916       /* get peer position */
 917       gst_query_parse_position (query, &format, &peer_pos, &peer_total);
 918
 919       /* FIXME: this code assumes that there's no discont in the queue */
 920       switch (format) {
 921         case GST_FORMAT_BYTES:
 922           peer_pos -= queue->cur_level.bytes;
 923           break;
 924         case GST_FORMAT_TIME:
 925           peer_pos -= queue->cur_level.time;
 926           break;
 927         default:
 928           /* FIXME */
 929           break;
 930       }
 931       /* set updated positions */
 932       gst_query_set_position (query, format, peer_pos, peer_total);
 933       break;
 934     }
 935     default:
 936       break;
 937   }
 938
 939   return TRUE;
 940 }
 941
 942 static gboolean
 943 gst_queue_sink_activate_push (GstPad * pad, gboolean active)
 944 {
 945   gboolean result = TRUE;
 946   GstQueue *queue;
 947
 948   queue = GST_QUEUE (gst_pad_get_parent (pad));
 949
 950   if (active) {
 951     GST_QUEUE_MUTEX_LOCK (queue);
 952     queue->srcresult = GST_FLOW_OK;
 953     GST_QUEUE_MUTEX_UNLOCK (queue);
 954   } else {
 955     /* step 1, unblock chain function */
 956     GST_QUEUE_MUTEX_LOCK (queue);
 957     queue->srcresult = GST_FLOW_WRONG_STATE;
 958     gst_queue_locked_flush (queue);
 959     GST_QUEUE_MUTEX_UNLOCK (queue);
 960
 961     /* and make sure the chain function finishes */
 962     GST_STREAM_LOCK (pad);
 963     GST_STREAM_UNLOCK (pad);
 964   }
 965
 966   gst_object_unref (queue);
 967
 968   return result;
 969 }
 970
 971 static gboolean
 972 gst_queue_src_activate_push (GstPad * pad, gboolean active)
 973 {
 974   gboolean result = FALSE;
 975   GstQueue *queue;
 976
 977   queue = GST_QUEUE (gst_pad_get_parent (pad));
 978
 979   if (active) {
 980     GST_QUEUE_MUTEX_LOCK (queue);
 981     queue->srcresult = GST_FLOW_OK;
 982     /* we do not start the task yet if the pad is not connected */
 983     if (gst_pad_is_linked (pad))
 984       result = gst_pad_start_task (pad, (GstTaskFunction) gst_queue_loop, pad);
 985     else {
 986       GST_DEBUG_OBJECT (queue, "not starting task as pad is not linked");
 987       result = TRUE;
 988     }
 989     GST_QUEUE_MUTEX_UNLOCK (queue);
 990   } else {
 991     /* step 1, unblock loop function */
 992     GST_QUEUE_MUTEX_LOCK (queue);
 993     queue->srcresult = GST_FLOW_WRONG_STATE;
 994     /* the item add signal will unblock */
 995     g_cond_signal (queue->item_add);
 996     GST_QUEUE_MUTEX_UNLOCK (queue);
 997
 998     /* step 2, make sure streaming finishes */
 999     result = gst_pad_stop_task (pad);
1000   }
1001
1002   gst_object_unref (queue);
1003
1004   return result;
1005 }
1006
1007 static GstStateChangeReturn
1008 gst_queue_change_state (GstElement * element, GstStateChange transition)
1009 {
1010   GstQueue *queue;
1011   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1012
1013   queue = GST_QUEUE (element);
1014
1015   GST_CAT_LOG_OBJECT (GST_CAT_STATES, element, "starting state change");
1016
1017   switch (transition) {
1018     case GST_STATE_CHANGE_NULL_TO_READY:
1019       break;
1020     case GST_STATE_CHANGE_READY_TO_PAUSED:
1021       break;
1022     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1023       break;
1024     default:
1025       break;
1026   }
1027
1028   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1029
1030   switch (transition) {
1031     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1032       break;
1033     case GST_STATE_CHANGE_PAUSED_TO_READY:
1034       break;
1035     case GST_STATE_CHANGE_READY_TO_NULL:
1036       break;
1037     default:
1038       break;
1039   }
1040
1041   return ret;
1042 }
1043
1044 static void
1045 gst_queue_set_property (GObject * object,
1046     guint prop_id, const GValue * value, GParamSpec * pspec)
1047 {
1048   GstQueue *queue = GST_QUEUE (object);
1049
1050   /* someone could change levels here, and since this
1051    * affects the get/put funcs, we need to lock for safety. */
1052   GST_QUEUE_MUTEX_LOCK (queue);
1053
1054   switch (prop_id) {
1055     case ARG_MAX_SIZE_BYTES:
1056       queue->max_size.bytes = g_value_get_uint (value);
1057       break;
1058     case ARG_MAX_SIZE_BUFFERS:
1059       queue->max_size.buffers = g_value_get_uint (value);
1060       break;
1061     case ARG_MAX_SIZE_TIME:
1062       queue->max_size.time = g_value_get_uint64 (value);
1063       break;
1064     case ARG_MIN_THRESHOLD_BYTES:
1065       queue->min_threshold.bytes = g_value_get_uint (value);
1066       break;
1067     case ARG_MIN_THRESHOLD_BUFFERS:
1068       queue->min_threshold.buffers = g_value_get_uint (value);
1069       break;
1070     case ARG_MIN_THRESHOLD_TIME:
1071       queue->min_threshold.time = g_value_get_uint64 (value);
1072       break;
1073     case ARG_LEAKY:
1074       queue->leaky = g_value_get_enum (value);
1075       break;
1076     case ARG_MAY_DEADLOCK:
1077       queue->may_deadlock = g_value_get_boolean (value);
1078       break;
1079     case ARG_BLOCK_TIMEOUT:
1080       queue->block_timeout = g_value_get_uint64 (value);
1081       break;
1082     default:
1083       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1084       break;
1085   }
1086
1087   GST_QUEUE_MUTEX_UNLOCK (queue);
1088 }
1089
1090 static void
1091 gst_queue_get_property (GObject * object,
1092     guint prop_id, GValue * value, GParamSpec * pspec)
1093 {
1094   GstQueue *queue = GST_QUEUE (object);
1095
1096   GST_QUEUE_MUTEX_LOCK (queue);
1097
1098   switch (prop_id) {
1099     case ARG_CUR_LEVEL_BYTES:
1100       g_value_set_uint (value, queue->cur_level.bytes);
1101       break;
1102     case ARG_CUR_LEVEL_BUFFERS:
1103       g_value_set_uint (value, queue->cur_level.buffers);
1104       break;
1105     case ARG_CUR_LEVEL_TIME:
1106       g_value_set_uint64 (value, queue->cur_level.time);
1107       break;
1108     case ARG_MAX_SIZE_BYTES:
1109       g_value_set_uint (value, queue->max_size.bytes);
1110       break;
1111     case ARG_MAX_SIZE_BUFFERS:
1112       g_value_set_uint (value, queue->max_size.buffers);
1113       break;
1114     case ARG_MAX_SIZE_TIME:
1115       g_value_set_uint64 (value, queue->max_size.time);
1116       break;
1117     case ARG_MIN_THRESHOLD_BYTES:
1118       g_value_set_uint (value, queue->min_threshold.bytes);
1119       break;
1120     case ARG_MIN_THRESHOLD_BUFFERS:
1121       g_value_set_uint (value, queue->min_threshold.buffers);
1122       break;
1123     case ARG_MIN_THRESHOLD_TIME:
1124       g_value_set_uint64 (value, queue->min_threshold.time);
1125       break;
1126     case ARG_LEAKY:
1127       g_value_set_enum (value, queue->leaky);
1128       break;
1129     case ARG_MAY_DEADLOCK:
1130       g_value_set_boolean (value, queue->may_deadlock);
1131       break;
1132     case ARG_BLOCK_TIMEOUT:
1133       g_value_set_uint64 (value, queue->block_timeout);
1134       break;
1135     default:
1136       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1137       break;
1138   }
1139
1140   GST_QUEUE_MUTEX_UNLOCK (queue);
1141 }