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