plugins/elements/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       if (GST_EVENT_IS_SERIALIZED (event)) {
 559         /* we put the event in the queue, we don't have to act ourselves */
 560         GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 561             "adding event %p of type %d", event, GST_EVENT_TYPE (event));
 562       } else {
 563         gst_event_ref (event);
 564         gst_pad_push_event (queue->srcpad, event);
 565         gst_event_unref (event);
 566         goto done;
 567       }
 568       break;
 569   }
 570
 571   GST_QUEUE_MUTEX_LOCK (queue);
 572   g_queue_push_tail (queue->queue, event);
 573   g_cond_signal (queue->item_add);
 574   GST_QUEUE_MUTEX_UNLOCK (queue);
 575
 576 done:
 577
 578   return TRUE;
 579 }
 580
 581 static gboolean
 582 gst_queue_is_empty (GstQueue * queue)
 583 {
 584   return (queue->queue->length == 0 ||
 585       (queue->min_threshold.buffers > 0 &&
 586           queue->cur_level.buffers < queue->min_threshold.buffers) ||
 587       (queue->min_threshold.bytes > 0 &&
 588           queue->cur_level.bytes < queue->min_threshold.bytes) ||
 589       (queue->min_threshold.time > 0 &&
 590           queue->cur_level.time < queue->min_threshold.time));
 591 }
 592
 593 static gboolean
 594 gst_queue_is_filled (GstQueue * queue)
 595 {
 596   return (((queue->max_size.buffers > 0 &&
 597               queue->cur_level.buffers >= queue->max_size.buffers) ||
 598           (queue->max_size.bytes > 0 &&
 599               queue->cur_level.bytes >= queue->max_size.bytes) ||
 600           (queue->max_size.time > 0 &&
 601               queue->cur_level.time >= queue->max_size.time)));
 602 }
 603
 604
 605 static GstFlowReturn
 606 gst_queue_chain (GstPad * pad, GstBuffer * buffer)
 607 {
 608   GstQueue *queue;
 609
 610   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 611
 612   /* we have to lock the queue since we span threads */
 613   GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 614
 615   GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 616       "adding buffer %p of size %d", buffer, GST_BUFFER_SIZE (buffer));
 617
 618   /* We make space available if we're "full" according to whatever
 619    * the user defined as "full". Note that this only applies to buffers.
 620    * We always handle events and they don't count in our statistics. */
 621   while (gst_queue_is_filled (queue)) {
 622     GST_QUEUE_MUTEX_UNLOCK (queue);
 623     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_OVERRUN], 0);
 624     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 625
 626     /* how are we going to make space for this buffer? */
 627     switch (queue->leaky) {
 628         /* leak current buffer */
 629       case GST_QUEUE_LEAK_UPSTREAM:
 630         GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
 631             "queue is full, leaking buffer on upstream end");
 632         /* now we can clean up and exit right away */
 633         goto out_unref;
 634
 635         /* leak first buffer in the queue */
 636       case GST_QUEUE_LEAK_DOWNSTREAM:{
 637         /* this is a bit hacky. We'll manually iterate the list
 638          * and find the first buffer from the head on. We'll
 639          * unref that and "fix up" the GQueue object... */
 640         GList *item;
 641         GstMiniObject *leak = NULL;
 642
 643         GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
 644             "queue is full, leaking buffer on downstream end");
 645
 646         for (item = queue->queue->head; item != NULL; item = item->next) {
 647           if (GST_IS_BUFFER (item->data)) {
 648             leak = item->data;
 649             break;
 650           }
 651         }
 652
 653         /* if we didn't find anything, it means we have no buffers
 654          * in here. That cannot happen, since we had >= 1 bufs */
 655         g_assert (leak);
 656
 657         /* Now remove it from the list, fixing up the GQueue
 658          * CHECKME: is a queue->head the first or the last item? */
 659         item = g_list_delete_link (queue->queue->head, item);
 660         queue->queue->head = g_list_first (item);
 661         queue->queue->tail = g_list_last (item);
 662         queue->queue->length--;
 663
 664         /* and unref the buffer at the end. Twice, because we keep a ref
 665          * to make things read-only. Also keep our list uptodate. */
 666         queue->cur_level.bytes -= GST_BUFFER_SIZE (buffer);
 667         queue->cur_level.buffers--;
 668         if (GST_BUFFER_DURATION (buffer) != GST_CLOCK_TIME_NONE)
 669           queue->cur_level.time -= GST_BUFFER_DURATION (buffer);
 670
 671         gst_buffer_unref (buffer);
 672         gst_buffer_unref (buffer);
 673         break;
 674       }
 675
 676       default:
 677         g_warning ("Unknown leaky type, using default");
 678         /* fall-through */
 679
 680         /* don't leak. Instead, wait for space to be available */
 681       case GST_QUEUE_NO_LEAK:
 682         STATUS (queue, "pre-full wait");
 683
 684         while (gst_queue_is_filled (queue)) {
 685           STATUS (queue, "waiting for item_del signal from thread using qlock");
 686           g_cond_wait (queue->item_del, queue->qlock);
 687
 688           if (queue->srcresult != GST_FLOW_OK)
 689             goto out_flushing;
 690
 691           /* if there's a pending state change for this queue
 692            * or its manager, switch back to iterator so bottom
 693            * half of state change executes */
 694           STATUS (queue, "received item_del signal from thread using qlock");
 695         }
 696
 697         STATUS (queue, "post-full wait");
 698         GST_QUEUE_MUTEX_UNLOCK (queue);
 699         g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_RUNNING], 0);
 700         GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 701
 702         break;
 703     }
 704   }
 705
 706   g_queue_push_tail (queue->queue, buffer);
 707
 708   /* add buffer to the statistics */
 709   queue->cur_level.buffers++;
 710   queue->cur_level.bytes += GST_BUFFER_SIZE (buffer);
 711   if (GST_BUFFER_DURATION (buffer) != GST_CLOCK_TIME_NONE)
 712     queue->cur_level.time += GST_BUFFER_DURATION (buffer);
 713
 714   STATUS (queue, "+ level");
 715
 716   GST_CAT_LOG_OBJECT (queue_dataflow, queue, "signalling item_add");
 717   g_cond_signal (queue->item_add);
 718   GST_QUEUE_MUTEX_UNLOCK (queue);
 719
 720   return GST_FLOW_OK;
 721
 722   /* special conditions */
 723 out_unref:
 724   {
 725     GST_QUEUE_MUTEX_UNLOCK (queue);
 726
 727     gst_buffer_unref (buffer);
 728
 729     return GST_FLOW_OK;
 730   }
 731 out_flushing:
 732   {
 733     GstFlowReturn ret = queue->srcresult;
 734     const gchar *flowname = gst_flow_get_name (ret);
 735
 736     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 737         "exit because task paused, reason: %s", flowname);
 738     GST_QUEUE_MUTEX_UNLOCK (queue);
 739
 740     gst_buffer_unref (buffer);
 741
 742     return ret;
 743   }
 744 }
 745
 746 static void
 747 gst_queue_loop (GstPad * pad)
 748 {
 749   GstQueue *queue;
 750   GstMiniObject *data;
 751   gboolean restart = TRUE;
 752
 753   queue = GST_QUEUE (GST_PAD_PARENT (pad));
 754
 755   /* have to lock for thread-safety */
 756   GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 757
 758 restart:
 759   while (gst_queue_is_empty (queue)) {
 760     GST_QUEUE_MUTEX_UNLOCK (queue);
 761     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_UNDERRUN], 0);
 762     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 763
 764     STATUS (queue, "pre-empty wait");
 765     while (gst_queue_is_empty (queue)) {
 766       STATUS (queue, "waiting for item_add");
 767
 768       GST_LOG_OBJECT (queue, "doing g_cond_wait using qlock from thread %p",
 769           g_thread_self ());
 770       g_cond_wait (queue->item_add, queue->qlock);
 771
 772       /* we released the lock in the g_cond above so we might be
 773        * flushing now */
 774       if (queue->srcresult != GST_FLOW_OK)
 775         goto out_flushing;
 776
 777       GST_LOG_OBJECT (queue, "done g_cond_wait using qlock from thread %p",
 778           g_thread_self ());
 779       STATUS (queue, "got item_add signal");
 780     }
 781
 782     STATUS (queue, "post-empty wait");
 783     GST_QUEUE_MUTEX_UNLOCK (queue);
 784     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_RUNNING], 0);
 785     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 786   }
 787
 788   /* There's something in the list now, whatever it is */
 789   data = g_queue_pop_head (queue->queue);
 790   GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 791       "retrieved data %p from queue", data);
 792
 793   if (GST_IS_BUFFER (data)) {
 794     GstFlowReturn result;
 795
 796     /* Update statistics */
 797     queue->cur_level.buffers--;
 798     queue->cur_level.bytes -= GST_BUFFER_SIZE (data);
 799     if (GST_BUFFER_DURATION (data) != GST_CLOCK_TIME_NONE)
 800       queue->cur_level.time -= GST_BUFFER_DURATION (data);
 801
 802     GST_QUEUE_MUTEX_UNLOCK (queue);
 803     result = gst_pad_push (pad, GST_BUFFER (data));
 804     GST_QUEUE_MUTEX_LOCK (queue);
 805     /* can opt to check for srcresult here but the push should
 806      * return an error value that is more accurate */
 807     if (result != GST_FLOW_OK) {
 808       const gchar *flowname;
 809
 810       flowname = gst_flow_get_name (result);
 811
 812       queue->srcresult = result;
 813       if (GST_FLOW_IS_FATAL (result)) {
 814         GST_ELEMENT_ERROR (queue, STREAM, STOPPED,
 815             ("streaming stopped, reason %s", flowname),
 816             ("streaming stopped, reason %s", flowname));
 817         gst_pad_push_event (queue->srcpad, gst_event_new_eos ());
 818       }
 819       GST_DEBUG ("pausing queue, reason %s", flowname);
 820       gst_pad_pause_task (queue->srcpad);
 821     }
 822   } else {
 823     if (GST_EVENT_TYPE (data) == GST_EVENT_EOS) {
 824       /* all incomming data is now unexpected */
 825       queue->srcresult = GST_FLOW_UNEXPECTED;
 826       /* and we don't need to process anymore */
 827       GST_DEBUG ("pausing queue, we're EOS now");
 828       gst_pad_pause_task (queue->srcpad);
 829       restart = FALSE;
 830     }
 831     GST_QUEUE_MUTEX_UNLOCK (queue);
 832     gst_pad_push_event (queue->srcpad, GST_EVENT (data));
 833     GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
 834     if (restart == TRUE)
 835       goto restart;
 836   }
 837
 838   STATUS (queue, "after _get()");
 839
 840   GST_CAT_LOG_OBJECT (queue_dataflow, queue, "signalling item_del");
 841   g_cond_signal (queue->item_del);
 842   GST_QUEUE_MUTEX_UNLOCK (queue);
 843
 844   return;
 845
 846 out_flushing:
 847   {
 848     const gchar *flowname = gst_flow_get_name (queue->srcresult);
 849
 850     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
 851         "exit because task paused, reason:  %s", flowname);
 852     GST_QUEUE_MUTEX_UNLOCK (queue);
 853
 854     return;
 855   }
 856 }
 857
 858
 859 static gboolean
 860 gst_queue_handle_src_event (GstPad * pad, GstEvent * event)
 861 {
 862   gboolean res = TRUE;
 863   GstQueue *queue = GST_QUEUE (GST_PAD_PARENT (pad));
 864
 865 #ifndef GST_DISABLE_GST_DEBUG
 866   GST_CAT_DEBUG_OBJECT (queue_dataflow, queue, "got event %p (%d)",
 867       event, GST_EVENT_TYPE (event));
 868 #endif
 869
 870   res = gst_pad_push_event (queue->sinkpad, event);
 871
 872   return res;
 873 }
 874
 875 static gboolean
 876 gst_queue_handle_src_query (GstPad * pad, GstQuery * query)
 877 {
 878   GstQueue *queue = GST_QUEUE (GST_PAD_PARENT (pad));
 879   GstPad *peer;
 880   gboolean res;
 881
 882   if (!(peer = gst_pad_get_peer (queue->sinkpad)))
 883     return FALSE;
 884
 885   res = gst_pad_query (peer, query);
 886   gst_object_unref (peer);
 887   if (!res)
 888     return FALSE;
 889
 890   switch (GST_QUERY_TYPE (query)) {
 891     case GST_QUERY_POSITION:
 892     {
 893       gint64 peer_pos, peer_total;
 894       GstFormat format;
 895
 896       /* get peer position */
 897       gst_query_parse_position (query, &format, &peer_pos, &peer_total);
 898
 899       /* FIXME: this code assumes that there's no discont in the queue */
 900       switch (format) {
 901         case GST_FORMAT_BYTES:
 902           peer_pos -= queue->cur_level.bytes;
 903           break;
 904         case GST_FORMAT_TIME:
 905           peer_pos -= queue->cur_level.time;
 906           break;
 907         default:
 908           /* FIXME */
 909           break;
 910       }
 911       /* set updated positions */
 912       gst_query_set_position (query, format, peer_pos, peer_total);
 913       break;
 914     }
 915     default:
 916       break;
 917   }
 918
 919   return TRUE;
 920 }
 921
 922 static gboolean
 923 gst_queue_sink_activate_push (GstPad * pad, gboolean active)
 924 {
 925   gboolean result = FALSE;
 926   GstQueue *queue;
 927
 928   queue = GST_QUEUE (gst_pad_get_parent (pad));
 929
 930   if (active) {
 931     queue->srcresult = GST_FLOW_OK;
 932     result = TRUE;
 933   } else {
 934     /* step 1, unblock chain and loop functions */
 935     GST_QUEUE_MUTEX_LOCK (queue);
 936     queue->srcresult = GST_FLOW_WRONG_STATE;
 937     gst_queue_locked_flush (queue);
 938     g_cond_signal (queue->item_del);
 939     GST_QUEUE_MUTEX_UNLOCK (queue);
 940
 941     /* step 2, make sure streaming finishes */
 942     result = gst_pad_stop_task (pad);
 943   }
 944   gst_object_unref (queue);
 945
 946   return result;
 947 }
 948
 949 static gboolean
 950 gst_queue_src_activate_push (GstPad * pad, gboolean active)
 951 {
 952   gboolean result = FALSE;
 953   GstQueue *queue;
 954
 955   queue = GST_QUEUE (gst_pad_get_parent (pad));
 956
 957   if (active) {
 958     GST_QUEUE_MUTEX_LOCK (queue);
 959     queue->srcresult = GST_FLOW_OK;
 960     /* we do not start the task yet if the pad is not connected */
 961     if (gst_pad_is_linked (pad))
 962       result = gst_pad_start_task (pad, (GstTaskFunction) gst_queue_loop, pad);
 963     else {
 964       GST_DEBUG ("not starting task as pad is not linked");
 965       result = TRUE;
 966     }
 967     GST_QUEUE_MUTEX_UNLOCK (queue);
 968   } else {
 969     /* step 1, unblock chain and loop functions */
 970     GST_QUEUE_MUTEX_LOCK (queue);
 971     queue->srcresult = GST_FLOW_WRONG_STATE;
 972     g_cond_signal (queue->item_add);
 973     GST_QUEUE_MUTEX_UNLOCK (queue);
 974
 975     /* step 2, make sure streaming finishes */
 976     result = gst_pad_stop_task (pad);
 977   }
 978
 979   gst_object_unref (queue);
 980
 981   return result;
 982 }
 983
 984 static GstElementStateReturn
 985 gst_queue_change_state (GstElement * element)
 986 {
 987   GstQueue *queue;
 988   GstElementStateReturn ret = GST_STATE_SUCCESS;
 989
 990   queue = GST_QUEUE (element);
 991
 992   GST_CAT_LOG_OBJECT (GST_CAT_STATES, element, "starting state change");
 993
 994   switch (GST_STATE_TRANSITION (element)) {
 995     case GST_STATE_NULL_TO_READY:
 996       break;
 997     case GST_STATE_READY_TO_PAUSED:
 998       break;
 999     case GST_STATE_PAUSED_TO_PLAYING:
1000       break;
1001     default:
1002       break;
1003   }
1004
1005   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element);
1006
1007   switch (GST_STATE_TRANSITION (element)) {
1008     case GST_STATE_PLAYING_TO_PAUSED:
1009       break;
1010     case GST_STATE_PAUSED_TO_READY:
1011       break;
1012     case GST_STATE_READY_TO_NULL:
1013       break;
1014     default:
1015       break;
1016   }
1017
1018   return ret;
1019 }
1020
1021 static void
1022 gst_queue_set_property (GObject * object,
1023     guint prop_id, const GValue * value, GParamSpec * pspec)
1024 {
1025   GstQueue *queue = GST_QUEUE (object);
1026
1027   /* someone could change levels here, and since this
1028    * affects the get/put funcs, we need to lock for safety. */
1029   GST_QUEUE_MUTEX_LOCK (queue);
1030
1031   switch (prop_id) {
1032     case ARG_MAX_SIZE_BYTES:
1033       queue->max_size.bytes = g_value_get_uint (value);
1034       break;
1035     case ARG_MAX_SIZE_BUFFERS:
1036       queue->max_size.buffers = g_value_get_uint (value);
1037       break;
1038     case ARG_MAX_SIZE_TIME:
1039       queue->max_size.time = g_value_get_uint64 (value);
1040       break;
1041     case ARG_MIN_THRESHOLD_BYTES:
1042       queue->min_threshold.bytes = g_value_get_uint (value);
1043       break;
1044     case ARG_MIN_THRESHOLD_BUFFERS:
1045       queue->min_threshold.buffers = g_value_get_uint (value);
1046       break;
1047     case ARG_MIN_THRESHOLD_TIME:
1048       queue->min_threshold.time = g_value_get_uint64 (value);
1049       break;
1050     case ARG_LEAKY:
1051       queue->leaky = g_value_get_enum (value);
1052       break;
1053     case ARG_MAY_DEADLOCK:
1054       queue->may_deadlock = g_value_get_boolean (value);
1055       break;
1056     case ARG_BLOCK_TIMEOUT:
1057       queue->block_timeout = g_value_get_uint64 (value);
1058       break;
1059     default:
1060       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1061       break;
1062   }
1063
1064   GST_QUEUE_MUTEX_UNLOCK (queue);
1065 }
1066
1067 static void
1068 gst_queue_get_property (GObject * object,
1069     guint prop_id, GValue * value, GParamSpec * pspec)
1070 {
1071   GstQueue *queue = GST_QUEUE (object);
1072
1073   GST_QUEUE_MUTEX_LOCK (queue);
1074
1075   switch (prop_id) {
1076     case ARG_CUR_LEVEL_BYTES:
1077       g_value_set_uint (value, queue->cur_level.bytes);
1078       break;
1079     case ARG_CUR_LEVEL_BUFFERS:
1080       g_value_set_uint (value, queue->cur_level.buffers);
1081       break;
1082     case ARG_CUR_LEVEL_TIME:
1083       g_value_set_uint64 (value, queue->cur_level.time);
1084       break;
1085     case ARG_MAX_SIZE_BYTES:
1086       g_value_set_uint (value, queue->max_size.bytes);
1087       break;
1088     case ARG_MAX_SIZE_BUFFERS:
1089       g_value_set_uint (value, queue->max_size.buffers);
1090       break;
1091     case ARG_MAX_SIZE_TIME:
1092       g_value_set_uint64 (value, queue->max_size.time);
1093       break;
1094     case ARG_MIN_THRESHOLD_BYTES:
1095       g_value_set_uint (value, queue->min_threshold.bytes);
1096       break;
1097     case ARG_MIN_THRESHOLD_BUFFERS:
1098       g_value_set_uint (value, queue->min_threshold.buffers);
1099       break;
1100     case ARG_MIN_THRESHOLD_TIME:
1101       g_value_set_uint64 (value, queue->min_threshold.time);
1102       break;
1103     case ARG_LEAKY:
1104       g_value_set_enum (value, queue->leaky);
1105       break;
1106     case ARG_MAY_DEADLOCK:
1107       g_value_set_boolean (value, queue->may_deadlock);
1108       break;
1109     case ARG_BLOCK_TIMEOUT:
1110       g_value_set_uint64 (value, queue->block_timeout);
1111       break;
1112     default:
1113       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1114       break;
1115   }
1116
1117   GST_QUEUE_MUTEX_UNLOCK (queue);
1118 }