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