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