libs/gst/base/gstdataqueue.c

   1 /* GStreamer
   2  * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>
   3  *
   4  * gstdataqueue.c:
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 /**
  23  * SECTION:gstdataqueue
  24  * @short_description: Threadsafe queueing object
  25  *
  26  * #GstDataQueue is an object that handles threadsafe queueing of objects. It
  27  * also provides size-related functionality. This object should be used for
  28  * any #GstElement that wishes to provide some sort of queueing functionality.
  29  */
  30
  31 #include <gst/gst.h>
  32 #include "gstdataqueue.h"
  33
  34 GST_DEBUG_CATEGORY_STATIC (data_queue_debug);
  35 #define GST_CAT_DEFAULT (data_queue_debug)
  36 GST_DEBUG_CATEGORY_STATIC (data_queue_dataflow);
  37
  38
  39 /* Queue signals and args */
  40 enum
  41 {
  42   SIGNAL_EMPTY,
  43   SIGNAL_FULL,
  44   LAST_SIGNAL
  45 };
  46
  47 enum
  48 {
  49   ARG_0,
  50   ARG_CUR_LEVEL_VISIBLE,
  51   ARG_CUR_LEVEL_BYTES,
  52   ARG_CUR_LEVEL_TIME
  53       /* FILL ME */
  54 };
  55
  56 #define GST_DATA_QUEUE_MUTEX_LOCK(q) G_STMT_START {                     \
  57     GST_CAT_LOG (data_queue_dataflow,                                   \
  58       "locking qlock from thread %p",                                   \
  59       g_thread_self ());                                                \
  60   g_mutex_lock (q->qlock);                                              \
  61   GST_CAT_LOG (data_queue_dataflow,                                     \
  62       "locked qlock from thread %p",                                    \
  63       g_thread_self ());                                                \
  64 } G_STMT_END
  65
  66 #define GST_DATA_QUEUE_MUTEX_LOCK_CHECK(q, label) G_STMT_START {        \
  67     GST_DATA_QUEUE_MUTEX_LOCK (q);                                      \
  68     if (q->flushing)                                                    \
  69       goto label;                                                       \
  70   } G_STMT_END
  71
  72 #define GST_DATA_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                   \
  73     GST_CAT_LOG (data_queue_dataflow,                                   \
  74       "unlocking qlock from thread %p",                                 \
  75       g_thread_self ());                                                \
  76   g_mutex_unlock (q->qlock);                                            \
  77 } G_STMT_END
  78
  79 #define STATUS(q, msg)                                                  \
  80   GST_CAT_LOG (data_queue_dataflow,                                     \
  81                "queue:%p " msg ": %u visible items, %u "                \
  82                "bytes, %"G_GUINT64_FORMAT                               \
  83                " ns, %u elements",                                      \
  84                queue,                                                   \
  85                q->cur_level.visible,                                    \
  86                q->cur_level.bytes,                                      \
  87                q->cur_level.time,                                       \
  88                q->queue->length)
  89
  90 static void gst_data_queue_finalize (GObject * object);
  91
  92 static void gst_data_queue_set_property (GObject * object,
  93     guint prop_id, const GValue * value, GParamSpec * pspec);
  94 static void gst_data_queue_get_property (GObject * object,
  95     guint prop_id, GValue * value, GParamSpec * pspec);
  96
  97 static GObjectClass *parent_class = NULL;
  98 static guint gst_data_queue_signals[LAST_SIGNAL] = { 0 };
  99
 100 #define _do_init \
 101 { \
 102   GST_DEBUG_CATEGORY_INIT (data_queue_debug, "dataqueue", 0, \
 103       "data queue object"); \
 104   GST_DEBUG_CATEGORY_INIT (data_queue_dataflow, "data_queue_dataflow", 0, \
 105       "dataflow inside the data queue object"); \
 106 }
 107
 108
 109 G_DEFINE_TYPE_WITH_CODE (GstDataQueue, gst_data_queue, G_TYPE_OBJECT, _do_init);
 110
 111 static void
 112 gst_data_queue_class_init (GstDataQueueClass * klass)
 113 {
 114   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 115
 116   parent_class = g_type_class_peek_parent (klass);
 117
 118   gobject_class->set_property = gst_data_queue_set_property;
 119   gobject_class->get_property = gst_data_queue_get_property;
 120
 121   /* signals */
 122   /**
 123    * GstDataQueue::empty:
 124    * @queue: the queue instance
 125    *
 126    * Reports that the queue became empty (empty).
 127    * A queue is empty if the total amount of visible items inside it (num-visible, time,
 128    * size) is lower than the boundary values which can be set through the GObject
 129    * properties.
 130    */
 131   gst_data_queue_signals[SIGNAL_EMPTY] =
 132       g_signal_new ("empty", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 133       G_STRUCT_OFFSET (GstDataQueueClass, empty), NULL, NULL,
 134       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 135
 136   /**
 137    * GstDataQueue::full:
 138    * @queue: the queue instance
 139    *
 140    * Reports that the queue became full (full).
 141    * A queue is full if the total amount of data inside it (num-visible, time,
 142    * size) is higher than the boundary values which can be set through the GObject
 143    * properties.
 144    */
 145   gst_data_queue_signals[SIGNAL_FULL] =
 146       g_signal_new ("full", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
 147       G_STRUCT_OFFSET (GstDataQueueClass, full), NULL, NULL,
 148       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 149
 150   /* properties */
 151   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BYTES,
 152       g_param_spec_uint ("current-level-bytes", "Current level (kB)",
 153           "Current amount of data in the queue (bytes)",
 154           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 155   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_VISIBLE,
 156       g_param_spec_uint ("current-level-visible",
 157           "Current level (visible items)",
 158           "Current number of visible items in the queue", 0, G_MAXUINT, 0,
 159           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 160   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_TIME,
 161       g_param_spec_uint64 ("current-level-time", "Current level (ns)",
 162           "Current amount of data in the queue (in ns)", 0, G_MAXUINT64, 0,
 163           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 164
 165   gobject_class->finalize = gst_data_queue_finalize;
 166 }
 167
 168 static void
 169 gst_data_queue_init (GstDataQueue * queue)
 170 {
 171   queue->cur_level.visible = 0; /* no content */
 172   queue->cur_level.bytes = 0;   /* no content */
 173   queue->cur_level.time = 0;    /* no content */
 174
 175   queue->checkfull = NULL;
 176
 177   queue->qlock = g_mutex_new ();
 178   queue->item_add = g_cond_new ();
 179   queue->item_del = g_cond_new ();
 180   queue->queue = g_queue_new ();
 181
 182   GST_DEBUG ("initialized queue's not_empty & not_full conditions");
 183 }
 184
 185 /**
 186  * gst_data_queue_new_full:
 187  * @checkfull: the callback used to tell if the element considers the queue full
 188  * or not.
 189  * @fullcallback: the callback which will be called when the queue is considered full.
 190  * @emptycallback: the callback which will be called when the queue is considered empty.
 191  * @checkdata: a #gpointer that will be given in the @checkfull callback.
 192  *
 193  * Creates a new #GstDataQueue. The difference with @gst_data_queue_new is that it will
 194  * not emit the 'full' and 'empty' signals, but instead calling directly @fullcallback
 195  * or @emptycallback.
 196  *
 197  * Returns: a new #GstDataQueue.
 198  *
 199  * Since: 0.10.26
 200  */
 201
 202 GstDataQueue *
 203 gst_data_queue_new_full (GstDataQueueCheckFullFunction checkfull,
 204     GstDataQueueFullCallback fullcallback,
 205     GstDataQueueEmptyCallback emptycallback, gpointer checkdata)
 206 {
 207   GstDataQueue *ret;
 208
 209   g_return_val_if_fail (checkfull != NULL, NULL);
 210
 211   ret = g_object_newv (GST_TYPE_DATA_QUEUE, 0, NULL);
 212   ret->checkfull = checkfull;
 213   ret->checkdata = checkdata;
 214   ret->fullcallback = fullcallback;
 215   ret->emptycallback = emptycallback;
 216
 217   return ret;
 218 }
 219
 220 /**
 221  * gst_data_queue_new:
 222  * @checkfull: the callback used to tell if the element considers the queue full
 223  * or not.
 224  * @checkdata: a #gpointer that will be given in the @checkfull callback.
 225  *
 226  * Returns: a new #GstDataQueue.
 227  */
 228
 229 GstDataQueue *
 230 gst_data_queue_new (GstDataQueueCheckFullFunction checkfull, gpointer checkdata)
 231 {
 232   return gst_data_queue_new_full (checkfull, NULL, NULL, checkdata);
 233 }
 234
 235 static void
 236 gst_data_queue_cleanup (GstDataQueue * queue)
 237 {
 238   while (!g_queue_is_empty (queue->queue)) {
 239     GstDataQueueItem *item = g_queue_pop_head (queue->queue);
 240
 241     /* Just call the destroy notify on the item */
 242     item->destroy (item);
 243   }
 244   queue->cur_level.visible = 0;
 245   queue->cur_level.bytes = 0;
 246   queue->cur_level.time = 0;
 247 }
 248
 249 /* called only once, as opposed to dispose */
 250 static void
 251 gst_data_queue_finalize (GObject * object)
 252 {
 253   GstDataQueue *queue = GST_DATA_QUEUE (object);
 254
 255   GST_DEBUG ("finalizing queue");
 256
 257   gst_data_queue_cleanup (queue);
 258   g_queue_free (queue->queue);
 259
 260   GST_DEBUG ("free mutex");
 261   g_mutex_free (queue->qlock);
 262   GST_DEBUG ("done free mutex");
 263
 264   g_cond_free (queue->item_add);
 265   g_cond_free (queue->item_del);
 266
 267   G_OBJECT_CLASS (parent_class)->finalize (object);
 268 }
 269
 270 static void
 271 gst_data_queue_locked_flush (GstDataQueue * queue)
 272 {
 273   STATUS (queue, "before flushing");
 274   gst_data_queue_cleanup (queue);
 275   STATUS (queue, "after flushing");
 276   /* we deleted something... */
 277   g_cond_signal (queue->item_del);
 278 }
 279
 280 static gboolean
 281 gst_data_queue_locked_is_empty (GstDataQueue * queue)
 282 {
 283   return (queue->queue->length == 0);
 284 }
 285
 286 static gboolean
 287 gst_data_queue_locked_is_full (GstDataQueue * queue)
 288 {
 289   return queue->checkfull (queue, queue->cur_level.visible,
 290       queue->cur_level.bytes, queue->cur_level.time, queue->checkdata);
 291 }
 292
 293 /**
 294  * gst_data_queue_flush:
 295  * @queue: a #GstDataQueue.
 296  *
 297  * Flushes all the contents of the @queue. Any call to #gst_data_queue_push and
 298  * #gst_data_queue_pop will be released.
 299  * MT safe.
 300  */
 301 void
 302 gst_data_queue_flush (GstDataQueue * queue)
 303 {
 304   GST_DEBUG ("queue:%p", queue);
 305   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 306   gst_data_queue_locked_flush (queue);
 307   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 308 }
 309
 310 /**
 311  * gst_data_queue_is_empty:
 312  * @queue: a #GstDataQueue.
 313  *
 314  * Queries if there are any items in the @queue.
 315  * MT safe.
 316  *
 317  * Returns: #TRUE if @queue is empty.
 318  */
 319 gboolean
 320 gst_data_queue_is_empty (GstDataQueue * queue)
 321 {
 322   gboolean res;
 323
 324   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 325   res = gst_data_queue_locked_is_empty (queue);
 326   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 327
 328   return res;
 329 }
 330
 331 /**
 332  * gst_data_queue_is_full:
 333  * @queue: a #GstDataQueue.
 334  *
 335  * Queries if @queue is full. This check will be done using the
 336  * #GstDataQueueCheckFullFunction registered with @queue.
 337  * MT safe.
 338  *
 339  * Returns: #TRUE if @queue is full.
 340  */
 341 gboolean
 342 gst_data_queue_is_full (GstDataQueue * queue)
 343 {
 344   gboolean res;
 345
 346   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 347   res = gst_data_queue_locked_is_full (queue);
 348   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 349
 350   return res;
 351 }
 352
 353 /**
 354  * gst_data_queue_set_flushing:
 355  * @queue: a #GstDataQueue.
 356  * @flushing: a #gboolean stating if the queue will be flushing or not.
 357  *
 358  * Sets the queue to flushing state if @flushing is #TRUE. If set to flushing
 359  * state, any incoming data on the @queue will be discarded. Any call currently
 360  * blocking on #gst_data_queue_push or #gst_data_queue_pop will return straight
 361  * away with a return value of #FALSE. While the @queue is in flushing state,
 362  * all calls to those two functions will return #FALSE.
 363  *
 364  * MT Safe.
 365  */
 366 void
 367 gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
 368 {
 369   GST_DEBUG ("queue:%p , flushing:%d", queue, flushing);
 370
 371   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 372   queue->flushing = flushing;
 373   if (flushing) {
 374     /* release push/pop functions */
 375     g_cond_signal (queue->item_add);
 376     g_cond_signal (queue->item_del);
 377   }
 378   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 379 }
 380
 381 /**
 382  * gst_data_queue_push:
 383  * @queue: a #GstDataQueue.
 384  * @item: a #GstDataQueueItem.
 385  *
 386  * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
 387  * on the @queue. If the @queue is full, the call will block until space is
 388  * available, OR the @queue is set to flushing state.
 389  * MT safe.
 390  *
 391  * Note that this function has slightly different semantics than gst_pad_push()
 392  * and gst_pad_push_event(): this function only takes ownership of @item and
 393  * the #GstMiniObject contained in @item if the push was successful. If FALSE
 394  * is returned, the caller is responsible for freeing @item and its contents.
 395  *
 396  * Returns: #TRUE if the @item was successfully pushed on the @queue.
 397  */
 398 gboolean
 399 gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item)
 400 {
 401   g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
 402   g_return_val_if_fail (item != NULL, FALSE);
 403
 404   GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
 405
 406   STATUS (queue, "before pushing");
 407
 408   /* We ALWAYS need to check for queue fillness */
 409   if (gst_data_queue_locked_is_full (queue)) {
 410     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 411     if (G_LIKELY (queue->fullcallback))
 412       queue->fullcallback (queue, queue->checkdata);
 413     else
 414       g_signal_emit (queue, gst_data_queue_signals[SIGNAL_FULL], 0);
 415     GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
 416
 417     /* signal might have removed some items */
 418     while (gst_data_queue_locked_is_full (queue)) {
 419       g_cond_wait (queue->item_del, queue->qlock);
 420       if (queue->flushing)
 421         goto flushing;
 422     }
 423   }
 424
 425   g_queue_push_tail (queue->queue, item);
 426
 427   if (item->visible)
 428     queue->cur_level.visible++;
 429   queue->cur_level.bytes += item->size;
 430   queue->cur_level.time += item->duration;
 431
 432   STATUS (queue, "after pushing");
 433   g_cond_signal (queue->item_add);
 434
 435   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 436
 437   return TRUE;
 438
 439   /* ERRORS */
 440 flushing:
 441   {
 442     GST_DEBUG ("queue:%p, we are flushing", queue);
 443     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 444     return FALSE;
 445   }
 446 }
 447
 448 /**
 449  * gst_data_queue_pop:
 450  * @queue: a #GstDataQueue.
 451  * @item: pointer to store the returned #GstDataQueueItem.
 452  *
 453  * Retrieves the first @item available on the @queue. If the queue is currently
 454  * empty, the call will block until at least one item is available, OR the
 455  * @queue is set to the flushing state.
 456  * MT safe.
 457  *
 458  * Returns: #TRUE if an @item was successfully retrieved from the @queue.
 459  */
 460 gboolean
 461 gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
 462 {
 463   g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
 464   g_return_val_if_fail (item != NULL, FALSE);
 465
 466   GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
 467
 468   STATUS (queue, "before popping");
 469
 470   if (gst_data_queue_locked_is_empty (queue)) {
 471     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 472     if (G_LIKELY (queue->emptycallback))
 473       queue->emptycallback (queue, queue->checkdata);
 474     else
 475       g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
 476     GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
 477
 478     while (gst_data_queue_locked_is_empty (queue)) {
 479       g_cond_wait (queue->item_add, queue->qlock);
 480       if (queue->flushing)
 481         goto flushing;
 482     }
 483   }
 484
 485   /* Get the item from the GQueue */
 486   *item = g_queue_pop_head (queue->queue);
 487
 488   /* update current level counter */
 489   if ((*item)->visible)
 490     queue->cur_level.visible--;
 491   queue->cur_level.bytes -= (*item)->size;
 492   queue->cur_level.time -= (*item)->duration;
 493
 494   STATUS (queue, "after popping");
 495   g_cond_signal (queue->item_del);
 496
 497   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 498
 499   return TRUE;
 500
 501   /* ERRORS */
 502 flushing:
 503   {
 504     GST_DEBUG ("queue:%p, we are flushing", queue);
 505     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 506     return FALSE;
 507   }
 508 }
 509
 510 /**
 511  * gst_data_queue_drop_head:
 512  * @queue: The #GstDataQueue to drop an item from.
 513  * @type: The #GType of the item to drop.
 514  *
 515  * Pop and unref the head-most #GstMiniObject with the given #GType.
 516  *
 517  * Returns: TRUE if an element was removed.
 518  */
 519 gboolean
 520 gst_data_queue_drop_head (GstDataQueue * queue, GType type)
 521 {
 522   gboolean res = FALSE;
 523   GList *item;
 524   GstDataQueueItem *leak = NULL;
 525
 526   g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
 527
 528   GST_DEBUG ("queue:%p", queue);
 529
 530   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 531   for (item = g_queue_peek_head_link (queue->queue); item; item = item->next) {
 532     GstDataQueueItem *tmp = (GstDataQueueItem *) item->data;
 533
 534     if (G_TYPE_CHECK_INSTANCE_TYPE (tmp->object, type)) {
 535       leak = tmp;
 536       break;
 537     }
 538   }
 539
 540   if (!leak)
 541     goto done;
 542
 543   g_queue_delete_link (queue->queue, item);
 544
 545   if (leak->visible)
 546     queue->cur_level.visible--;
 547   queue->cur_level.bytes -= leak->size;
 548   queue->cur_level.time -= leak->duration;
 549
 550   leak->destroy (leak);
 551
 552   res = TRUE;
 553
 554 done:
 555   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 556
 557   GST_DEBUG ("queue:%p , res:%d", queue, res);
 558
 559   return res;
 560 }
 561
 562 /**
 563  * gst_data_queue_limits_changed:
 564  * @queue: The #GstDataQueue
 565  *
 566  * Inform the queue that the limits for the fullness check have changed and that
 567  * any blocking gst_data_queue_push() should be unblocked to recheck the limts.
 568  */
 569 void
 570 gst_data_queue_limits_changed (GstDataQueue * queue)
 571 {
 572   g_return_if_fail (GST_IS_DATA_QUEUE (queue));
 573
 574   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 575   GST_DEBUG ("signal del");
 576   g_cond_signal (queue->item_del);
 577   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 578 }
 579
 580 /**
 581  * gst_data_queue_get_level:
 582  * @queue: The #GstDataQueue
 583  * @level: the location to store the result
 584  *
 585  * Get the current level of the queue.
 586  */
 587 void
 588 gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize * level)
 589 {
 590   level->visible = queue->cur_level.visible;
 591   level->bytes = queue->cur_level.bytes;
 592   level->time = queue->cur_level.time;
 593 }
 594
 595 static void
 596 gst_data_queue_set_property (GObject * object,
 597     guint prop_id, const GValue * value, GParamSpec * pspec)
 598 {
 599   switch (prop_id) {
 600     default:
 601       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 602       break;
 603   }
 604 }
 605
 606 static void
 607 gst_data_queue_get_property (GObject * object,
 608     guint prop_id, GValue * value, GParamSpec * pspec)
 609 {
 610   GstDataQueue *queue = GST_DATA_QUEUE (object);
 611
 612   GST_DATA_QUEUE_MUTEX_LOCK (queue);
 613
 614   switch (prop_id) {
 615     case ARG_CUR_LEVEL_BYTES:
 616       g_value_set_uint (value, queue->cur_level.bytes);
 617       break;
 618     case ARG_CUR_LEVEL_VISIBLE:
 619       g_value_set_uint (value, queue->cur_level.visible);
 620       break;
 621     case ARG_CUR_LEVEL_TIME:
 622       g_value_set_uint64 (value, queue->cur_level.time);
 623       break;
 624     default:
 625       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 626       break;
 627   }
 628
 629   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
 630 }