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  *
   6  * gstqueue.c:
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  21  * Boston, MA 02111-1307, USA.
  22  */
  23
  24
  25 #include "gst_private.h"
  26
  27 #include "gstqueue.h"
  28 #include "gstscheduler.h"
  29 #include "gstevent.h"
  30 #include "gstinfo.h"
  31
  32 static GstElementDetails gst_queue_details = GST_ELEMENT_DETAILS (
  33   "Queue",
  34   "Generic",
  35   "Simple data queue",
  36   "Erik Walthinsen <omega@cse.ogi.edu>"
  37 );
  38
  39
  40 /* Queue signals and args */
  41 enum {
  42   SIGNAL_UNDERRUN,
  43   SIGNAL_RUNNING,
  44   SIGNAL_OVERRUN,
  45   LAST_SIGNAL
  46 };
  47
  48 enum {
  49   ARG_0,
  50   /* FIXME: don't we have another way of doing this
  51    * "Gstreamer format" (frame/byte/time) queries? */
  52   ARG_CUR_LEVEL_BUFFERS,
  53   ARG_CUR_LEVEL_BYTES,
  54   ARG_CUR_LEVEL_TIME,
  55   ARG_MAX_SIZE_BUFFERS,
  56   ARG_MAX_SIZE_BYTES,
  57   ARG_MAX_SIZE_TIME,
  58   ARG_MIN_TRESHOLD_BUFFERS,
  59   ARG_MIN_TRESHOLD_BYTES,
  60   ARG_MIN_TRESHOLD_TIME,
  61   ARG_LEAKY,
  62   ARG_MAY_DEADLOCK,
  63   ARG_BLOCK_TIMEOUT
  64   /* FILL ME */
  65 };
  66
  67 typedef struct _GstQueueEventResponse {
  68   GstEvent *event;
  69   gboolean ret, handled;
  70 } GstQueueEventResponse;
  71
  72 static void     gst_queue_base_init             (GstQueueClass *klass);
  73 static void     gst_queue_class_init            (GstQueueClass *klass);
  74 static void     gst_queue_init                  (GstQueue      *queue);
  75 static void     gst_queue_dispose               (GObject       *object);
  76
  77 static void     gst_queue_set_property          (GObject       *object,
  78                                                  guint          prop_id,
  79                                                  const GValue  *value,
  80                                                  GParamSpec    *pspec);
  81 static void     gst_queue_get_property          (GObject       *object,
  82                                                  guint          prop_id,
  83                                                  GValue        *value,
  84                                                  GParamSpec    *pspec);
  85
  86 static GstCaps *gst_queue_getcaps               (GstPad        *pad,
  87                                                  GstCaps       *caps);
  88 static GstPadLinkReturn
  89                 gst_queue_link                  (GstPad        *pad,
  90                                                  GstCaps       *caps);
  91 static void     gst_queue_chain                 (GstPad        *pad,
  92                                                  GstData       *data);
  93 static GstData *gst_queue_get                   (GstPad        *pad);
  94 static GstBufferPool *
  95                 gst_queue_get_bufferpool        (GstPad        *pad);
  96
  97 static gboolean gst_queue_handle_src_event      (GstPad        *pad,
  98                                                  GstEvent      *event);
  99
 100 static void     gst_queue_locked_flush          (GstQueue      *queue);
 101
 102 static GstElementStateReturn
 103                 gst_queue_change_state          (GstElement    *element);
 104 static gboolean gst_queue_release_locks         (GstElement    *element);
 105
 106
 107 #define GST_TYPE_QUEUE_LEAKY (queue_leaky_get_type ())
 108
 109 static GType
 110 queue_leaky_get_type (void)
 111 {
 112   static GType queue_leaky_type = 0;
 113   static GEnumValue queue_leaky[] = {
 114     { GST_QUEUE_NO_LEAK,                "0", "Not Leaky" },
 115     { GST_QUEUE_LEAK_UPSTREAM,          "1", "Leaky on Upstream" },
 116     { GST_QUEUE_LEAK_DOWNSTREAM,        "2", "Leaky on Downstream" },
 117     { 0, NULL, NULL },
 118   };
 119   if (!queue_leaky_type) {
 120     queue_leaky_type = g_enum_register_static("GstQueueLeaky", queue_leaky);
 121   }
 122   return queue_leaky_type;
 123 }
 124
 125 static GstElementClass *parent_class = NULL;
 126 static guint gst_queue_signals[LAST_SIGNAL] = { 0 };
 127
 128 GType
 129 gst_queue_get_type (void)
 130 {
 131   static GType queue_type = 0;
 132
 133   if (!queue_type) {
 134     static const GTypeInfo queue_info = {
 135       sizeof (GstQueueClass),
 136       (GBaseInitFunc) gst_queue_base_init,
 137       NULL,
 138       (GClassInitFunc) gst_queue_class_init,
 139       NULL,
 140       NULL,
 141       sizeof (GstQueue),
 142       4,
 143       (GInstanceInitFunc) gst_queue_init,
 144       NULL
 145     };
 146
 147     queue_type = g_type_register_static (GST_TYPE_ELEMENT,
 148                                          "GstQueue", &queue_info, 0);
 149   }
 150
 151   return queue_type;
 152 }
 153
 154 static void
 155 gst_queue_base_init (GstQueueClass *klass)
 156 {
 157   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 158
 159   gst_element_class_set_details (gstelement_class, &gst_queue_details);
 160 }
 161
 162 static void
 163 gst_queue_class_init (GstQueueClass *klass)
 164 {
 165   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 166   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 167
 168   parent_class = g_type_class_peek_parent (klass);
 169
 170   /* signals */
 171   gst_queue_signals[SIGNAL_UNDERRUN] =
 172     g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 173                   G_STRUCT_OFFSET (GstQueueClass, underrun), NULL, NULL,
 174                   g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 175   gst_queue_signals[SIGNAL_RUNNING] =
 176     g_signal_new ("running", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 177                   G_STRUCT_OFFSET (GstQueueClass, running), NULL, NULL,
 178                   g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 179   gst_queue_signals[SIGNAL_OVERRUN] =
 180     g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 181                   G_STRUCT_OFFSET (GstQueueClass, overrun), NULL, NULL,
 182                   g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 183
 184   /* properties */
 185   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BYTES,
 186     g_param_spec_uint ("current-level-bytes", "Current level (kB)",
 187                        "Current amount of data in the queue (bytes)",
 188                        0, G_MAXUINT, 0, G_PARAM_READABLE));
 189   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BUFFERS,
 190     g_param_spec_uint ("current-level-buffers", "Current level (buffers)",
 191                        "Current number of buffers in the queue",
 192                        0, G_MAXUINT, 0, G_PARAM_READABLE));
 193   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_TIME,
 194     g_param_spec_uint64 ("current-level-time", "Current level (ns)",
 195                          "Current amount of data in the queue (in ns)",
 196                          0, G_MAXUINT64, 0, G_PARAM_READABLE));
 197
 198   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,
 199     g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
 200                        "Max. amount of data in the queue (bytes, 0=disable)",
 201                        0, G_MAXUINT, 0, G_PARAM_READWRITE));
 202   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,
 203     g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
 204                        "Max. number of buffers in the queue (0=disable)",
 205                        0, G_MAXUINT, 0, G_PARAM_READWRITE));
 206   g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,
 207     g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 208                          "Max. amount of data in the queue (in ns, 0=disable)",
 209                          0, G_MAXUINT64, 0, G_PARAM_READWRITE));
 210
 211   g_object_class_install_property (gobject_class, ARG_MIN_TRESHOLD_BYTES,
 212     g_param_spec_uint ("min-treshold-bytes", "Min. treshold (kB)",
 213                        "Min. amount of data in the queue to allow reading (bytes, 0=disable)",
 214                        0, G_MAXUINT, 0, G_PARAM_READWRITE));
 215   g_object_class_install_property (gobject_class, ARG_MIN_TRESHOLD_BUFFERS,
 216     g_param_spec_uint ("min-treshold-buffers", "Min. treshold (buffers)",
 217                        "Min. number of buffers in the queue to allow reading (0=disable)",
 218                        0, G_MAXUINT, 0, G_PARAM_READWRITE));
 219   g_object_class_install_property (gobject_class, ARG_MIN_TRESHOLD_TIME,
 220     g_param_spec_uint64 ("min-treshold-time", "Min. treshold (ns)",
 221                          "Min. amount of data in the queue to allow reading (in ns, 0=disable)",
 222                          0, G_MAXUINT64, 0, G_PARAM_READWRITE));
 223
 224   g_object_class_install_property (gobject_class, ARG_LEAKY,
 225     g_param_spec_enum ("leaky", "Leaky",
 226                        "Where the queue leaks, if at all",
 227                        GST_TYPE_QUEUE_LEAKY, GST_QUEUE_NO_LEAK, G_PARAM_READWRITE));
 228   g_object_class_install_property (gobject_class, ARG_MAY_DEADLOCK,
 229     g_param_spec_boolean ("may_deadlock", "May Deadlock",
 230                           "The queue may deadlock if it's full and not PLAYING",
 231                           TRUE, G_PARAM_READWRITE));
 232   g_object_class_install_property (gobject_class, ARG_BLOCK_TIMEOUT,
 233     g_param_spec_uint64 ("block_timeout", "Timeout for Block",
 234                          "Nanoseconds until blocked queue times out and returns filler event. "
 235                          "Value of -1 disables timeout",
 236                          0, G_MAXUINT64, -1, G_PARAM_READWRITE));
 237
 238   /* set several parent class virtual functions */
 239   gobject_class->dispose        = GST_DEBUG_FUNCPTR (gst_queue_dispose);
 240   gobject_class->set_property   = GST_DEBUG_FUNCPTR (gst_queue_set_property);
 241   gobject_class->get_property   = GST_DEBUG_FUNCPTR (gst_queue_get_property);
 242
 243   gstelement_class->change_state  = GST_DEBUG_FUNCPTR (gst_queue_change_state);
 244   gstelement_class->release_locks = GST_DEBUG_FUNCPTR (gst_queue_release_locks);
 245 }
 246
 247 static void
 248 gst_queue_init (GstQueue *queue)
 249 {
 250   /* scheduling on this kind of element is, well, interesting */
 251   GST_FLAG_SET (queue, GST_ELEMENT_DECOUPLED);
 252   GST_FLAG_SET (queue, GST_ELEMENT_EVENT_AWARE);
 253
 254   queue->sinkpad = gst_pad_new ("sink", GST_PAD_SINK);
 255   gst_pad_set_chain_function (queue->sinkpad, GST_DEBUG_FUNCPTR (gst_queue_chain));
 256   gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
 257   gst_pad_set_bufferpool_function (queue->sinkpad, GST_DEBUG_FUNCPTR (gst_queue_get_bufferpool));
 258   gst_pad_set_link_function (queue->sinkpad, GST_DEBUG_FUNCPTR (gst_queue_link));
 259   gst_pad_set_getcaps_function (queue->sinkpad, GST_DEBUG_FUNCPTR (gst_queue_getcaps));
 260   gst_pad_set_active (queue->sinkpad, TRUE);
 261
 262   queue->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 263   gst_pad_set_get_function (queue->srcpad, GST_DEBUG_FUNCPTR (gst_queue_get));
 264   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 265   gst_pad_set_link_function (queue->srcpad, GST_DEBUG_FUNCPTR (gst_queue_link));
 266   gst_pad_set_getcaps_function (queue->srcpad, GST_DEBUG_FUNCPTR (gst_queue_getcaps));
 267   gst_pad_set_event_function (queue->srcpad, GST_DEBUG_FUNCPTR (gst_queue_handle_src_event));
 268   gst_pad_set_active (queue->srcpad, TRUE);
 269
 270   queue->cur_level.buffers      = 0; /* no content */
 271   queue->cur_level.bytes        = 0; /* no content */
 272   queue->cur_level.time         = 0; /* no content */
 273   queue->max_size.buffers       = 100;          /* max. 100 buffers */
 274   queue->max_size.bytes         = 1024 * 1024;  /* max. 1 MB */
 275   queue->max_size.time          = GST_SECOND;   /* max. 1 sec. */
 276   queue->min_treshold.buffers   = 0; /* no treshold */
 277   queue->min_treshold.bytes     = 0; /* no treshold */
 278   queue->min_treshold.time      = 0; /* no treshold */
 279
 280   queue->leaky = GST_QUEUE_NO_LEAK;
 281   queue->may_deadlock = TRUE;
 282   queue->block_timeout = GST_CLOCK_TIME_NONE;
 283   queue->interrupt = FALSE;
 284   queue->flush = FALSE;
 285
 286   queue->qlock = g_mutex_new ();
 287   queue->item_add = g_cond_new ();
 288   queue->item_del = g_cond_new ();
 289   queue->event_done = g_cond_new ();
 290   queue->events = g_queue_new ();
 291   queue->queue = g_queue_new ();
 292
 293   GST_CAT_DEBUG_OBJECT (GST_CAT_THREAD, queue,
 294                         "initialized queue's not_empty & not_full conditions");
 295 }
 296
 297 static void
 298 gst_queue_dispose (GObject *object)
 299 {
 300   GstQueue *queue = GST_QUEUE (object);
 301
 302   gst_element_set_state (GST_ELEMENT (queue), GST_STATE_NULL);
 303
 304   while (!g_queue_is_empty (queue->queue)) {
 305     GstData *data = g_queue_pop_head (queue->queue);
 306     gst_data_unref (data);
 307   }
 308   g_queue_free (queue->queue);
 309   g_mutex_free (queue->qlock);
 310   g_cond_free (queue->item_add);
 311   g_cond_free (queue->item_del);
 312   g_cond_free (queue->event_done);
 313   while (!g_queue_is_empty (queue->events)) {
 314     GstEvent *event = g_queue_pop_head (queue->events);
 315     gst_event_unref (event);
 316   }
 317
 318   if (G_OBJECT_CLASS (parent_class)->dispose)
 319     G_OBJECT_CLASS (parent_class)->dispose (object);
 320 }
 321
 322 static GstPad *
 323 gst_queue_otherpad (GstPad *pad)
 324 {
 325   GstQueue *queue = GST_QUEUE (gst_pad_get_parent (pad));
 326   GstPad *otherpad;
 327
 328   if (pad == queue->srcpad)
 329     otherpad = queue->sinkpad;
 330   else
 331     otherpad = queue->srcpad;
 332
 333   return otherpad;
 334 }
 335
 336 static GstPadLinkReturn
 337 gst_queue_link (GstPad  *pad,
 338                 GstCaps *caps)
 339 {
 340   return gst_pad_proxy_link (gst_queue_otherpad (pad), caps);
 341 }
 342
 343 static GstCaps *
 344 gst_queue_getcaps (GstPad  *pad,
 345                    GstCaps *caps)
 346 {
 347   GstPad *otherpad = GST_PAD_PEER (gst_queue_otherpad (pad));
 348
 349   if (otherpad)
 350     return gst_pad_get_caps (otherpad);
 351
 352   return NULL;
 353 }
 354
 355 static GstBufferPool *
 356 gst_queue_get_bufferpool (GstPad *pad)
 357 {
 358   return gst_pad_get_bufferpool (gst_queue_otherpad (pad));
 359 }
 360
 361 static void
 362 gst_queue_locked_flush (GstQueue *queue)
 363 {
 364   while (!g_queue_is_empty (queue->queue)) {
 365     GstData *data = g_queue_pop_head (queue->queue);
 366     gst_data_unref (data);
 367   }
 368   queue->timeval = NULL;
 369   queue->cur_level.buffers = 0;
 370   queue->cur_level.bytes = 0;
 371   queue->cur_level.time = 0;
 372
 373   /* make sure any pending buffers to be added are flushed too */
 374   queue->flush = TRUE;
 375
 376   /* we deleted something... */
 377   g_cond_signal (queue->item_del);
 378 }
 379
 380 #define STATUS(queue, msg) \
 381   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, \
 382                       "(%s:%s) " msg ": %u of %u-%u buffers, %u of %u-%u " \
 383                       "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
 384                       "-%" G_GUINT64_FORMAT " ns, %u elements", \
 385                       GST_DEBUG_PAD_NAME (pad), \
 386                       queue->cur_level.buffers, \
 387                       queue->min_treshold.buffers, \
 388                       queue->max_size.buffers, \
 389                       queue->cur_level.bytes, \
 390                       queue->min_treshold.bytes, \
 391                       queue->max_size.bytes, \
 392                       queue->cur_level.time, \
 393                       queue->min_treshold.time, \
 394                       queue->max_size.time, \
 395                       queue->queue->length)
 396
 397 static void
 398 gst_queue_chain (GstPad  *pad,
 399                  GstData *data)
 400 {
 401   GstQueue *queue;
 402
 403   g_return_if_fail (pad != NULL);
 404   g_return_if_fail (GST_IS_PAD (pad));
 405   g_return_if_fail (data != NULL);
 406
 407   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 408
 409 restart:
 410   /* we have to lock the queue since we span threads */
 411   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "locking t:%p", g_thread_self ());
 412   g_mutex_lock (queue->qlock);
 413   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "locked t:%p", g_thread_self ());
 414
 415   /* check for events to send upstream */
 416   while (!g_queue_is_empty (queue->events)){
 417     GstQueueEventResponse *er = g_queue_pop_head (queue->events);
 418     GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "sending event upstream");
 419     er->ret = gst_pad_event_default (GST_PAD_PEER (pad), er->event);
 420     er->handled = TRUE;
 421     g_cond_signal (queue->event_done);
 422     GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "event sent");
 423   }
 424
 425   /* assume don't need to flush this buffer when the queue is filled */
 426   queue->flush = FALSE;
 427
 428   if (GST_IS_EVENT (data)) {
 429     switch (GST_EVENT_TYPE (data)) {
 430       case GST_EVENT_FLUSH:
 431         STATUS (queue, "received flush event");
 432         gst_queue_locked_flush (queue);
 433         break;
 434       case GST_EVENT_EOS:
 435         STATUS (queue, "received EOS");
 436         break;
 437       default:
 438         /* we put the event in the queue, we don't have to act ourselves */
 439         GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue,
 440                             "adding event %p of type %d",
 441                             data, GST_EVENT_TYPE (data));
 442         break;
 443     }
 444   }
 445
 446   if (GST_IS_BUFFER (data))
 447     GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue,
 448                         "adding buffer %p of size %d",
 449                         data, GST_BUFFER_SIZE (data));
 450
 451   /* We make space available if we're "full" according to whatever
 452    * the user defined as "full". Note that this only applies to buffers.
 453    * We always handle events and they don't count in our statistics. */
 454   if (GST_IS_BUFFER (data) &&
 455       ((queue->max_size.buffers > 0 &&
 456         queue->cur_level.buffers >= queue->max_size.buffers) ||
 457        (queue->max_size.bytes > 0 &&
 458         queue->cur_level.bytes >= queue->max_size.bytes) ||
 459        (queue->max_size.time > 0 &&
 460         queue->cur_level.time >= queue->max_size.time))) {
 461     g_mutex_unlock (queue->qlock);
 462     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_OVERRUN], 0);
 463     g_mutex_lock (queue->qlock);
 464
 465     /* how are we going to make space for this buffer? */
 466     switch (queue->leaky) {
 467       /* leak current buffer */
 468       case GST_QUEUE_LEAK_UPSTREAM:
 469         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue,
 470                               "queue is full, leaking buffer on upstream end");
 471         /* now we can clean up and exit right away */
 472         g_mutex_unlock (queue->qlock);
 473         goto out_unref;
 474
 475       /* leak first buffer in the queue */
 476       case GST_QUEUE_LEAK_DOWNSTREAM: {
 477         /* this is a bit hacky. We'll manually iterate the list
 478          * and find the first buffer from the head on. We'll
 479          * unref that and "fix up" the GQueue object... */
 480         GList *item;
 481         GstData *leak = NULL;
 482
 483         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue,
 484                               "queue is full, leaking buffer on downstream end");
 485
 486         for (item = queue->queue->head; item != NULL; item = item->next) {
 487           if (GST_IS_BUFFER (item->data)) {
 488             leak = item->data;
 489             break;
 490           }
 491         }
 492
 493         /* if we didn't find anything, it means we have no buffers
 494          * in here. That cannot happen, since we had >= 1 bufs */
 495         g_assert (leak);
 496
 497         /* Now remove it from the list, fixing up the GQueue
 498          * CHECKME: is a queue->head the first or the last item? */
 499         item = g_list_delete_link (queue->queue->head, item);
 500         queue->queue->head = g_list_first (item);
 501         queue->queue->tail = g_list_last (item);
 502         queue->queue->length--;
 503
 504         /* and unref the data at the end. Twice, because we keep a ref
 505          * to make things read-only. Also keep our list uptodate. */
 506         queue->cur_level.bytes -= GST_BUFFER_SIZE (data);
 507         queue->cur_level.buffers --;
 508         if (GST_BUFFER_DURATION (data) != GST_CLOCK_TIME_NONE)
 509           queue->cur_level.time -= GST_BUFFER_DURATION (data);
 510
 511         gst_data_unref (data);
 512         gst_data_unref (data);
 513         break;
 514       }
 515
 516       default:
 517         g_warning ("Unknown leaky type, using default");
 518         /* fall-through */
 519
 520       /* don't leak. Instead, wait for space to be available */
 521       case GST_QUEUE_NO_LEAK:
 522         STATUS (queue, "pre-full wait");
 523
 524         while ((queue->max_size.buffers > 0 &&
 525                 queue->cur_level.buffers >= queue->max_size.buffers) ||
 526                (queue->max_size.bytes > 0 &&
 527                 queue->cur_level.bytes >= queue->max_size.bytes) ||
 528                (queue->max_size.time > 0 &&
 529                 queue->cur_level.time >= queue->max_size.time)) {
 530           /* if there's a pending state change for this queue
 531            * or its manager, switch back to iterator so bottom
 532            * half of state change executes */
 533           if (queue->interrupt) {
 534             GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "interrupted");
 535             g_mutex_unlock (queue->qlock);
 536             if (gst_scheduler_interrupt (gst_pad_get_scheduler (queue->sinkpad),
 537                                          GST_ELEMENT (queue))) {
 538               goto out_unref;
 539             }
 540             /* if we got here because we were unlocked after a
 541              * flush, we don't need to add the buffer to the
 542              * queue again */
 543             if (queue->flush) {
 544               GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue,
 545                                     "not adding pending buffer after flush");
 546               goto out_unref;
 547             }
 548             GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue,
 549                                   "adding pending buffer after interrupt");
 550             goto restart;
 551           }
 552
 553           if (GST_STATE (queue) != GST_STATE_PLAYING) {
 554             /* this means the other end is shut down. Try to
 555              * signal to resolve the error */
 556             if (!queue->may_deadlock) {
 557               g_mutex_unlock (queue->qlock);
 558               gst_data_unref (data);
 559               gst_element_error (GST_ELEMENT (queue),
 560                                  "deadlock found, source pad elements are shut down");
 561               /* we don't go to out_unref here, since we want to
 562                * unref the buffer *before* calling gst_element_error */
 563               return;
 564             } else {
 565               GST_CAT_WARNING_OBJECT (GST_CAT_DATAFLOW, queue,
 566                                       "%s: waiting for the app to restart "
 567                                       "source pad elements",
 568                                       GST_ELEMENT_NAME (queue));
 569             }
 570           }
 571
 572           STATUS (queue, "waiting for item_del signal");
 573           g_cond_wait (queue->item_del, queue->qlock);
 574           STATUS (queue, "received item_del signal");
 575         }
 576
 577         STATUS (queue, "post-full wait");
 578         g_mutex_unlock (queue->qlock);
 579         g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_RUNNING], 0);
 580         g_mutex_lock (queue->qlock);
 581         break;
 582     }
 583   }
 584
 585   /* put the buffer on the tail of the list. We keep a reference,
 586    * so that the data is read-only while in here. There's a good
 587    * reason to do so: we have a size and time counter, and any
 588    * modification to the content could change any of the two. */
 589   gst_data_ref (data);
 590   g_queue_push_tail (queue->queue, data);
 591
 592   /* Note that we only add buffers (not events) to the statistics */
 593   if (GST_IS_BUFFER (data)) {
 594     queue->cur_level.buffers++;
 595     queue->cur_level.bytes += GST_BUFFER_SIZE (data);
 596     if (GST_BUFFER_DURATION (data) != GST_CLOCK_TIME_NONE)
 597       queue->cur_level.time += GST_BUFFER_DURATION (data);
 598   }
 599
 600   STATUS (queue, "+ level");
 601
 602   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "signalling item_add");
 603   g_cond_signal (queue->item_add);
 604   g_mutex_unlock (queue->qlock);
 605
 606   return;
 607
 608 out_unref:
 609   gst_data_unref (data);
 610   return;
 611 }
 612
 613 static GstData *
 614 gst_queue_get (GstPad *pad)
 615 {
 616   GstQueue *queue;
 617   GstData *data;
 618
 619   g_return_val_if_fail (pad != NULL, NULL);
 620   g_return_val_if_fail (GST_IS_PAD (pad), NULL);
 621
 622   queue = GST_QUEUE (gst_pad_get_parent (pad));
 623
 624 restart:
 625   /* have to lock for thread-safety */
 626   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue,
 627                       "locking t:%p", g_thread_self ());
 628   g_mutex_lock (queue->qlock);
 629   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue,
 630                       "locked t:%p", g_thread_self ());
 631
 632   if (queue->queue->length == 0 ||
 633       (queue->min_treshold.buffers > 0 &&
 634        queue->cur_level.buffers < queue->min_treshold.buffers) ||
 635       (queue->min_treshold.bytes > 0 &&
 636        queue->cur_level.bytes < queue->min_treshold.bytes) ||
 637       (queue->min_treshold.time > 0 &&
 638        queue->cur_level.time < queue->min_treshold.time)) {
 639     g_mutex_unlock (queue->qlock);
 640     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_UNDERRUN], 0);
 641     g_mutex_lock (queue->qlock);
 642
 643     STATUS (queue, "pre-empty wait");
 644     while (queue->queue->length == 0 ||
 645            (queue->min_treshold.buffers > 0 &&
 646             queue->cur_level.buffers < queue->min_treshold.buffers) ||
 647            (queue->min_treshold.bytes > 0 &&
 648             queue->cur_level.bytes < queue->min_treshold.bytes) ||
 649            (queue->min_treshold.time > 0 &&
 650             queue->cur_level.time < queue->min_treshold.time)) {
 651       /* if there's a pending state change for this queue or its
 652        * manager, switch back to iterator so bottom half of state
 653        * change executes. */
 654       if (queue->interrupt) {
 655         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "interrupted");
 656         g_mutex_unlock (queue->qlock);
 657         if (gst_scheduler_interrupt (gst_pad_get_scheduler (queue->srcpad),
 658                                      GST_ELEMENT (queue)))
 659           return GST_DATA (gst_event_new (GST_EVENT_INTERRUPT));
 660         goto restart;
 661       }
 662       if (GST_STATE (queue) != GST_STATE_PLAYING) {
 663         /* this means the other end is shut down */
 664         if (!queue->may_deadlock) {
 665           g_mutex_unlock (queue->qlock);
 666           gst_element_error (GST_ELEMENT (queue),
 667                              "deadlock found, sink pad elements are shut down");
 668           goto restart;
 669         } else {
 670           GST_CAT_WARNING_OBJECT (GST_CAT_DATAFLOW, queue,
 671                                   "%s: waiting for the app to restart "
 672                                   "source pad elements",
 673                                   GST_ELEMENT_NAME (queue));
 674         }
 675       }
 676
 677       STATUS (queue, "waiting for item_add");
 678
 679       if (queue->block_timeout != GST_CLOCK_TIME_NONE) {
 680         GTimeVal timeout;
 681         g_get_current_time (&timeout);
 682         g_time_val_add (&timeout, queue->block_timeout / 1000);
 683         if (!g_cond_timed_wait (queue->item_add, queue->qlock, &timeout)){
 684           g_mutex_unlock (queue->qlock);
 685           GST_CAT_WARNING_OBJECT (GST_CAT_DATAFLOW, queue,
 686                                   "Sending filler event");
 687           return GST_DATA (gst_event_new_filler ());
 688         }
 689       } else {
 690         g_cond_wait (queue->item_add, queue->qlock);
 691       }
 692       STATUS (queue, "got item_add signal");
 693     }
 694
 695     STATUS (queue, "post-empty wait");
 696     g_mutex_unlock (queue->qlock);
 697     g_signal_emit (G_OBJECT (queue), gst_queue_signals[SIGNAL_RUNNING], 0);
 698     g_mutex_lock (queue->qlock);
 699   }
 700
 701   /* There's something in the list now, whatever it is */
 702   data = g_queue_pop_head (queue->queue);
 703   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue,
 704                       "retrieved data %p from queue", data);
 705
 706   if (GST_IS_BUFFER (data)) {
 707     /* Update statistics */
 708     queue->cur_level.buffers--;
 709     queue->cur_level.bytes -= GST_BUFFER_SIZE (data);
 710     if (GST_BUFFER_DURATION (data) != GST_CLOCK_TIME_NONE)
 711       queue->cur_level.time -= GST_BUFFER_DURATION (data);
 712   }
 713
 714   /* Now that we're done, we can lose our own reference to
 715    * the item, since we're no longer in danger. */
 716   gst_data_unref (data);
 717
 718   STATUS (queue, "after _get()");
 719
 720   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "signalling item_del");
 721   g_cond_signal (queue->item_del);
 722
 723   g_mutex_unlock (queue->qlock);
 724
 725   /* FIXME: I suppose this needs to be locked, since the EOS
 726    * bit affects the pipeline state. However, that bit is
 727    * locked too so it'd cause a deadlock. */
 728   if (GST_IS_EVENT (data)) {
 729     GstEvent *event = GST_EVENT (data);
 730     switch (GST_EVENT_TYPE (event)) {
 731       case GST_EVENT_EOS:
 732         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue,
 733                               "queue \"%s\" eos",
 734                               GST_ELEMENT_NAME (queue));
 735         gst_element_set_eos (GST_ELEMENT (queue));
 736         break;
 737       default:
 738         break;
 739     }
 740   }
 741
 742   return data;
 743 }
 744
 745
 746 static gboolean
 747 gst_queue_handle_src_event (GstPad   *pad,
 748                             GstEvent *event)
 749 {
 750   GstQueue *queue = GST_QUEUE (gst_pad_get_parent (pad));
 751   gboolean res;
 752
 753   g_mutex_lock (queue->qlock);
 754
 755   if (gst_element_get_state (GST_ELEMENT (queue)) == GST_STATE_PLAYING) {
 756     GstQueueEventResponse er;
 757
 758     /* push the event to the queue and wait for upstream consumption */
 759     er.event = event;
 760     er.handled = FALSE;
 761     g_queue_push_tail (queue->events, &er);
 762     while (!er.handled) {
 763       g_cond_wait (queue->event_done, queue->qlock);
 764     }
 765     res = er.ret;
 766   } else {
 767     res = gst_pad_event_default (pad, event);
 768
 769     switch (GST_EVENT_TYPE (event)) {
 770       case GST_EVENT_FLUSH:
 771         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue,
 772                               "FLUSH event, flushing queue\n");
 773         gst_queue_locked_flush (queue);
 774         break;
 775       case GST_EVENT_SEEK:
 776         if (GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH) {
 777           gst_queue_locked_flush (queue);
 778         }
 779       default:
 780         break;
 781     }
 782   }
 783
 784   g_mutex_unlock (queue->qlock);
 785
 786   /* we have to claim success, but we don't really know */
 787   return res;
 788 }
 789
 790 static gboolean
 791 gst_queue_release_locks (GstElement *element)
 792 {
 793   GstQueue *queue;
 794
 795   queue = GST_QUEUE (element);
 796
 797   g_mutex_lock (queue->qlock);
 798   queue->interrupt = TRUE;
 799   g_cond_signal (queue->item_add);
 800   g_cond_signal (queue->item_del);
 801   g_mutex_unlock (queue->qlock);
 802
 803   return TRUE;
 804 }
 805
 806 static GstElementStateReturn
 807 gst_queue_change_state (GstElement *element)
 808 {
 809   GstQueue *queue;
 810   GstElementStateReturn ret = GST_STATE_SUCCESS;
 811
 812   queue = GST_QUEUE (element);
 813
 814   GST_CAT_LOG_OBJECT (GST_CAT_STATES, element, "starting state change");
 815
 816   /* lock the queue so another thread (not in sync with this thread's state)
 817    * can't call this queue's _get (or whatever)
 818    */
 819   g_mutex_lock (queue->qlock);
 820
 821   switch (GST_STATE_TRANSITION (element)) {
 822     case GST_STATE_NULL_TO_READY:
 823       gst_queue_locked_flush (queue);
 824       break;
 825     case GST_STATE_PAUSED_TO_PLAYING:
 826       if (!GST_PAD_IS_LINKED (queue->sinkpad)) {
 827         GST_CAT_DEBUG_OBJECT (GST_CAT_STATES, queue,
 828                               "queue %s is not linked",
 829                               GST_ELEMENT_NAME (queue));
 830         /* FIXME can this be? */
 831         g_cond_signal (queue->item_add);
 832
 833         ret = GST_STATE_FAILURE;
 834         goto error;
 835       } else {
 836         GstScheduler *src_sched, *sink_sched;
 837
 838         src_sched = gst_pad_get_scheduler (GST_PAD (queue->srcpad));
 839         sink_sched = gst_pad_get_scheduler (GST_PAD (queue->sinkpad));
 840
 841         if (src_sched == sink_sched) {
 842           GST_CAT_DEBUG_OBJECT (GST_CAT_STATES, queue,
 843                                 "queue %s does not connect different schedulers",
 844                                 GST_ELEMENT_NAME (queue));
 845
 846           g_warning ("queue %s does not connect different schedulers",
 847                      GST_ELEMENT_NAME (queue));
 848
 849           ret = GST_STATE_FAILURE;
 850           goto error;
 851         }
 852       }
 853       queue->interrupt = FALSE;
 854       break;
 855     case GST_STATE_PAUSED_TO_READY:
 856       gst_queue_locked_flush (queue);
 857       break;
 858     default:
 859       break;
 860   }
 861
 862   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 863     ret = GST_ELEMENT_CLASS (parent_class)->change_state (element);
 864
 865   /* this is an ugly hack to make sure our pads are always active.
 866    * Reason for this is that pad activation for the queue element
 867    * depends on 2 schedulers (ugh) */
 868   gst_pad_set_active (queue->sinkpad, TRUE);
 869   gst_pad_set_active (queue->srcpad, TRUE);
 870
 871 error:
 872   g_mutex_unlock (queue->qlock);
 873
 874   GST_CAT_LOG_OBJECT (GST_CAT_STATES, element, "done with state change");
 875
 876   return ret;
 877 }
 878
 879
 880 static void
 881 gst_queue_set_property (GObject      *object,
 882                         guint         prop_id,
 883                         const GValue *value,
 884                         GParamSpec   *pspec)
 885 {
 886   GstQueue *queue = GST_QUEUE (object);
 887
 888   /* someone could change levels here, and since this
 889    * affects the get/put funcs, we need to lock for safety. */
 890   g_mutex_lock (queue->qlock);
 891
 892   switch (prop_id) {
 893     case ARG_MAX_SIZE_BYTES:
 894       queue->max_size.bytes = g_value_get_uint (value);
 895       break;
 896     case ARG_MAX_SIZE_BUFFERS:
 897       queue->max_size.buffers = g_value_get_uint (value);
 898       break;
 899     case ARG_MAX_SIZE_TIME:
 900       queue->max_size.time = g_value_get_uint64 (value);
 901       break;
 902     case ARG_MIN_TRESHOLD_BYTES:
 903       queue->max_size.bytes = g_value_get_uint (value);
 904       break;
 905     case ARG_MIN_TRESHOLD_BUFFERS:
 906       queue->max_size.buffers = g_value_get_uint (value);
 907       break;
 908     case ARG_MIN_TRESHOLD_TIME:
 909       queue->max_size.time = g_value_get_uint64 (value);
 910       break;
 911     case ARG_LEAKY:
 912       queue->leaky = g_value_get_enum (value);
 913       break;
 914     case ARG_MAY_DEADLOCK:
 915       queue->may_deadlock = g_value_get_boolean (value);
 916       break;
 917     case ARG_BLOCK_TIMEOUT:
 918       queue->block_timeout = g_value_get_uint64 (value);
 919       break;
 920     default:
 921       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 922       break;
 923   }
 924
 925   g_mutex_unlock (queue->qlock);
 926 }
 927
 928 static void
 929 gst_queue_get_property (GObject    *object,
 930                         guint       prop_id,
 931                         GValue     *value,
 932                         GParamSpec *pspec)
 933 {
 934   GstQueue *queue = GST_QUEUE (object);
 935
 936   switch (prop_id) {
 937     case ARG_CUR_LEVEL_BYTES:
 938       g_value_set_uint (value, queue->cur_level.bytes);
 939       break;
 940     case ARG_CUR_LEVEL_BUFFERS:
 941       g_value_set_uint (value, queue->cur_level.buffers);
 942       break;
 943     case ARG_CUR_LEVEL_TIME:
 944       g_value_set_uint64 (value, queue->cur_level.time);
 945       break;
 946     case ARG_MAX_SIZE_BYTES:
 947       g_value_set_uint (value, queue->max_size.bytes);
 948       break;
 949     case ARG_MAX_SIZE_BUFFERS:
 950       g_value_set_uint (value, queue->max_size.buffers);
 951       break;
 952     case ARG_MAX_SIZE_TIME:
 953       g_value_set_uint64 (value, queue->max_size.time);
 954       break;
 955     case ARG_MIN_TRESHOLD_BYTES:
 956       g_value_set_uint (value, queue->min_treshold.bytes);
 957       break;
 958     case ARG_MIN_TRESHOLD_BUFFERS:
 959       g_value_set_uint (value, queue->min_treshold.buffers);
 960       break;
 961     case ARG_MIN_TRESHOLD_TIME:
 962       g_value_set_uint64 (value, queue->min_treshold.time);
 963       break;
 964     case ARG_LEAKY:
 965       g_value_set_enum (value, queue->leaky);
 966       break;
 967     case ARG_MAY_DEADLOCK:
 968       g_value_set_boolean (value, queue->may_deadlock);
 969       break;
 970     case ARG_BLOCK_TIMEOUT:
 971       g_value_set_uint64 (value, queue->block_timeout);
 972       break;
 973     default:
 974       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 975       break;
 976   }
 977 }