glib/gasyncqueue.c

   1 /* GLIB - Library of useful routines for C programming
   2  * Copyright (C) 1995-1997  Peter Mattis, Spencer Kimball and Josh MacDonald
   3  *
   4  * GAsyncQueue: asynchronous queue implementation, based on Gqueue.
   5  * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 /*
  24  * MT safe
  25  */
  26
  27 #include "config.h"
  28
  29 #include "gasyncqueue.h"
  30
  31 #include "gmem.h"
  32 #include "gqueue.h"
  33 #include "gtestutils.h"
  34 #include "gthread.h"
  35
  36
  37 /**
  38  * SECTION:async_queues
  39  * @title: Asynchronous Queues
  40  * @short_description: asynchronous communication between threads
  41  *
  42  * Often you need to communicate between different threads. In general
  43  * it's safer not to do this by shared memory, but by explicit message
  44  * passing. These messages only make sense asynchronously for
  45  * multi-threaded applications though, as a synchronous operation could
  46  * as well be done in the same thread.
  47  *
  48  * Asynchronous queues are an exception from most other GLib data
  49  * structures, as they can be used simultaneously from multiple threads
  50  * without explicit locking and they bring their own builtin reference
  51  * counting. This is because the nature of an asynchronous queue is that
  52  * it will always be used by at least 2 concurrent threads.
  53  *
  54  * For using an asynchronous queue you first have to create one with
  55  * g_async_queue_new(). A newly-created queue will get the reference
  56  * count 1. Whenever another thread is creating a new reference of (that
  57  * is, pointer to) the queue, it has to increase the reference count
  58  * (using g_async_queue_ref()). Also, before removing this reference,
  59  * the reference count has to be decreased (using g_async_queue_unref()).
  60  * After that the queue might no longer exist so you must not access
  61  * it after that point.
  62  *
  63  * A thread, which wants to send a message to that queue simply calls
  64  * g_async_queue_push() to push the message to the queue.
  65  *
  66  * A thread, which is expecting messages from an asynchronous queue
  67  * simply calls g_async_queue_pop() for that queue. If no message is
  68  * available in the queue at that point, the thread is now put to sleep
  69  * until a message arrives. The message will be removed from the queue
  70  * and returned. The functions g_async_queue_try_pop() and
  71  * g_async_queue_timed_pop() can be used to only check for the presence
  72  * of messages or to only wait a certain time for messages respectively.
  73  *
  74  * For almost every function there exist two variants, one that locks
  75  * the queue and one that doesn't. That way you can hold the queue lock
  76  * (acquire it with g_async_queue_lock() and release it with
  77  * g_async_queue_unlock()) over multiple queue accessing instructions.
  78  * This can be necessary to ensure the integrity of the queue, but should
  79  * only be used when really necessary, as it can make your life harder
  80  * if used unwisely. Normally you should only use the locking function
  81  * variants (those without the suffix _unlocked)
  82  */
  83
  84 /**
  85  * GAsyncQueue:
  86  *
  87  * The GAsyncQueue struct is an opaque data structure, which represents
  88  * an asynchronous queue. It should only be accessed through the
  89  * <function>g_async_queue_*</function> functions.
  90  */
  91 struct _GAsyncQueue
  92 {
  93   GMutex *mutex;
  94   GCond *cond;
  95   GQueue queue;
  96   GDestroyNotify item_free_func;
  97   guint waiting_threads;
  98   gint ref_count;
  99 };
 100
 101 typedef struct {
 102   GCompareDataFunc func;
 103   gpointer         user_data;
 104 } SortData;
 105
 106 /**
 107  * g_async_queue_new:
 108  *
 109  * Creates a new asynchronous queue with the initial reference count of 1.
 110  *
 111  * Return value: the new #GAsyncQueue.
 112  **/
 113 GAsyncQueue*
 114 g_async_queue_new (void)
 115 {
 116   GAsyncQueue* retval = g_new (GAsyncQueue, 1);
 117   retval->mutex = g_mutex_new ();
 118   retval->cond = NULL;
 119   g_queue_init (&retval->queue);
 120   retval->waiting_threads = 0;
 121   retval->ref_count = 1;
 122   retval->item_free_func = NULL;
 123   return retval;
 124 }
 125
 126 /**
 127  * g_async_queue_new_full:
 128  * @item_free_func: function to free queue elements
 129  *
 130  * Creates a new asynchronous queue with an initial reference count of 1 and
 131  * sets up a destroy notify function that is used to free any remaining
 132  * queue items when the queue is destroyed after the final unref.
 133  *
 134  * Return value: the new #GAsyncQueue.
 135  *
 136  * Since: 2.16
 137  **/
 138 GAsyncQueue*
 139 g_async_queue_new_full (GDestroyNotify item_free_func)
 140 {
 141   GAsyncQueue *async_queue = g_async_queue_new ();
 142   async_queue->item_free_func = item_free_func;
 143   return async_queue;
 144 }
 145
 146 /**
 147  * g_async_queue_ref:
 148  * @queue: a #GAsyncQueue.
 149  *
 150  * Increases the reference count of the asynchronous @queue by 1. You
 151  * do not need to hold the lock to call this function.
 152  *
 153  * Returns: the @queue that was passed in (since 2.6)
 154  **/
 155 GAsyncQueue *
 156 g_async_queue_ref (GAsyncQueue *queue)
 157 {
 158   g_return_val_if_fail (queue, NULL);
 159
 160   g_atomic_int_inc (&queue->ref_count);
 161
 162   return queue;
 163 }
 164
 165 /**
 166  * g_async_queue_ref_unlocked:
 167  * @queue: a #GAsyncQueue.
 168  *
 169  * Increases the reference count of the asynchronous @queue by 1.
 170  *
 171  * @Deprecated: Since 2.8, reference counting is done atomically
 172  * so g_async_queue_ref() can be used regardless of the @queue's
 173  * lock.
 174  **/
 175 void
 176 g_async_queue_ref_unlocked (GAsyncQueue *queue)
 177 {
 178   g_return_if_fail (queue);
 179
 180   g_atomic_int_inc (&queue->ref_count);
 181 }
 182
 183 /**
 184  * g_async_queue_unref_and_unlock:
 185  * @queue: a #GAsyncQueue.
 186  *
 187  * Decreases the reference count of the asynchronous @queue by 1 and
 188  * releases the lock. This function must be called while holding the
 189  * @queue's lock. If the reference count went to 0, the @queue will be
 190  * destroyed and the memory allocated will be freed.
 191  *
 192  * @Deprecated: Since 2.8, reference counting is done atomically
 193  * so g_async_queue_unref() can be used regardless of the @queue's
 194  * lock.
 195  **/
 196 void
 197 g_async_queue_unref_and_unlock (GAsyncQueue *queue)
 198 {
 199   g_return_if_fail (queue);
 200
 201   g_mutex_unlock (queue->mutex);
 202   g_async_queue_unref (queue);
 203 }
 204
 205 /**
 206  * g_async_queue_unref:
 207  * @queue: a #GAsyncQueue.
 208  *
 209  * Decreases the reference count of the asynchronous @queue by 1. If
 210  * the reference count went to 0, the @queue will be destroyed and the
 211  * memory allocated will be freed. So you are not allowed to use the
 212  * @queue afterwards, as it might have disappeared. You do not need to
 213  * hold the lock to call this function.
 214  **/
 215 void
 216 g_async_queue_unref (GAsyncQueue *queue)
 217 {
 218   g_return_if_fail (queue);
 219
 220   if (g_atomic_int_dec_and_test (&queue->ref_count))
 221     {
 222       g_return_if_fail (queue->waiting_threads == 0);
 223       g_mutex_free (queue->mutex);
 224       if (queue->cond)
 225         g_cond_free (queue->cond);
 226       if (queue->item_free_func)
 227         g_queue_foreach (&queue->queue, (GFunc) queue->item_free_func, NULL);
 228       g_queue_clear (&queue->queue);
 229       g_free (queue);
 230     }
 231 }
 232
 233 /**
 234  * g_async_queue_lock:
 235  * @queue: a #GAsyncQueue.
 236  *
 237  * Acquires the @queue's lock. After that you can only call the
 238  * <function>g_async_queue_*_unlocked()</function> function variants on that
 239  * @queue. Otherwise it will deadlock.
 240  **/
 241 void
 242 g_async_queue_lock (GAsyncQueue *queue)
 243 {
 244   g_return_if_fail (queue);
 245
 246   g_mutex_lock (queue->mutex);
 247 }
 248
 249 /**
 250  * g_async_queue_unlock:
 251  * @queue: a #GAsyncQueue.
 252  *
 253  * Releases the queue's lock.
 254  **/
 255 void
 256 g_async_queue_unlock (GAsyncQueue *queue)
 257 {
 258   g_return_if_fail (queue);
 259
 260   g_mutex_unlock (queue->mutex);
 261 }
 262
 263 /**
 264  * g_async_queue_push:
 265  * @queue: a #GAsyncQueue.
 266  * @data: @data to push into the @queue.
 267  *
 268  * Pushes the @data into the @queue. @data must not be %NULL.
 269  **/
 270 void
 271 g_async_queue_push (GAsyncQueue* queue, gpointer data)
 272 {
 273   g_return_if_fail (queue);
 274   g_return_if_fail (data);
 275
 276   g_mutex_lock (queue->mutex);
 277   g_async_queue_push_unlocked (queue, data);
 278   g_mutex_unlock (queue->mutex);
 279 }
 280
 281 /**
 282  * g_async_queue_push_unlocked:
 283  * @queue: a #GAsyncQueue.
 284  * @data: @data to push into the @queue.
 285  *
 286  * Pushes the @data into the @queue. @data must not be %NULL. This
 287  * function must be called while holding the @queue's lock.
 288  **/
 289 void
 290 g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)
 291 {
 292   g_return_if_fail (queue);
 293   g_return_if_fail (data);
 294
 295   g_queue_push_head (&queue->queue, data);
 296   if (queue->waiting_threads > 0)
 297     g_cond_signal (queue->cond);
 298 }
 299
 300 /**
 301  * g_async_queue_push_sorted:
 302  * @queue: a #GAsyncQueue
 303  * @data: the @data to push into the @queue
 304  * @func: the #GCompareDataFunc is used to sort @queue. This function
 305  *     is passed two elements of the @queue. The function should return
 306  *     0 if they are equal, a negative value if the first element
 307  *     should be higher in the @queue or a positive value if the first
 308  *     element should be lower in the @queue than the second element.
 309  * @user_data: user data passed to @func.
 310  *
 311  * Inserts @data into @queue using @func to determine the new
 312  * position.
 313  *
 314  * This function requires that the @queue is sorted before pushing on
 315  * new elements.
 316  *
 317  * This function will lock @queue before it sorts the queue and unlock
 318  * it when it is finished.
 319  *
 320  * For an example of @func see g_async_queue_sort().
 321  *
 322  * Since: 2.10
 323  **/
 324 void
 325 g_async_queue_push_sorted (GAsyncQueue      *queue,
 326                            gpointer          data,
 327                            GCompareDataFunc  func,
 328                            gpointer          user_data)
 329 {
 330   g_return_if_fail (queue != NULL);
 331
 332   g_mutex_lock (queue->mutex);
 333   g_async_queue_push_sorted_unlocked (queue, data, func, user_data);
 334   g_mutex_unlock (queue->mutex);
 335 }
 336
 337 static gint
 338 g_async_queue_invert_compare (gpointer  v1,
 339                               gpointer  v2,
 340                               SortData *sd)
 341 {
 342   return -sd->func (v1, v2, sd->user_data);
 343 }
 344
 345 /**
 346  * g_async_queue_push_sorted_unlocked:
 347  * @queue: a #GAsyncQueue
 348  * @data: the @data to push into the @queue
 349  * @func: the #GCompareDataFunc is used to sort @queue. This function
 350  *     is passed two elements of the @queue. The function should return
 351  *     0 if they are equal, a negative value if the first element
 352  *     should be higher in the @queue or a positive value if the first
 353  *     element should be lower in the @queue than the second element.
 354  * @user_data: user data passed to @func.
 355  *
 356  * Inserts @data into @queue using @func to determine the new
 357  * position.
 358  *
 359  * This function requires that the @queue is sorted before pushing on
 360  * new elements.
 361  *
 362  * This function is called while holding the @queue's lock.
 363  *
 364  * For an example of @func see g_async_queue_sort().
 365  *
 366  * Since: 2.10
 367  **/
 368 void
 369 g_async_queue_push_sorted_unlocked (GAsyncQueue      *queue,
 370                                     gpointer          data,
 371                                     GCompareDataFunc  func,
 372                                     gpointer          user_data)
 373 {
 374   SortData sd;
 375
 376   g_return_if_fail (queue != NULL);
 377
 378   sd.func = func;
 379   sd.user_data = user_data;
 380
 381   g_queue_insert_sorted (&queue->queue,
 382                          data,
 383                          (GCompareDataFunc)g_async_queue_invert_compare,
 384                          &sd);
 385   if (queue->waiting_threads > 0)
 386     g_cond_signal (queue->cond);
 387 }
 388
 389 static gpointer
 390 g_async_queue_pop_intern_unlocked (GAsyncQueue *queue,
 391                                    gboolean     try,
 392                                    GTimeVal    *end_time)
 393 {
 394   gpointer retval;
 395
 396   if (!g_queue_peek_tail_link (&queue->queue))
 397     {
 398       if (try)
 399         return NULL;
 400
 401       if (!queue->cond)
 402         queue->cond = g_cond_new ();
 403
 404       if (!end_time)
 405         {
 406           queue->waiting_threads++;
 407           while (!g_queue_peek_tail_link (&queue->queue))
 408             g_cond_wait (queue->cond, queue->mutex);
 409           queue->waiting_threads--;
 410         }
 411       else
 412         {
 413           queue->waiting_threads++;
 414           while (!g_queue_peek_tail_link (&queue->queue))
 415             if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
 416               break;
 417           queue->waiting_threads--;
 418           if (!g_queue_peek_tail_link (&queue->queue))
 419             return NULL;
 420         }
 421     }
 422
 423   retval = g_queue_pop_tail (&queue->queue);
 424
 425   g_assert (retval);
 426
 427   return retval;
 428 }
 429
 430 /**
 431  * g_async_queue_pop:
 432  * @queue: a #GAsyncQueue.
 433  *
 434  * Pops data from the @queue. This function blocks until data become
 435  * available.
 436  *
 437  * Return value: data from the queue.
 438  **/
 439 gpointer
 440 g_async_queue_pop (GAsyncQueue* queue)
 441 {
 442   gpointer retval;
 443
 444   g_return_val_if_fail (queue, NULL);
 445
 446   g_mutex_lock (queue->mutex);
 447   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 448   g_mutex_unlock (queue->mutex);
 449
 450   return retval;
 451 }
 452
 453 /**
 454  * g_async_queue_pop_unlocked:
 455  * @queue: a #GAsyncQueue.
 456  *
 457  * Pops data from the @queue. This function blocks until data become
 458  * available. This function must be called while holding the @queue's
 459  * lock.
 460  *
 461  * Return value: data from the queue.
 462  **/
 463 gpointer
 464 g_async_queue_pop_unlocked (GAsyncQueue* queue)
 465 {
 466   g_return_val_if_fail (queue, NULL);
 467
 468   return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 469 }
 470
 471 /**
 472  * g_async_queue_try_pop:
 473  * @queue: a #GAsyncQueue.
 474  *
 475  * Tries to pop data from the @queue. If no data is available, %NULL is
 476  * returned.
 477  *
 478  * Return value: data from the queue or %NULL, when no data is
 479  * available immediately.
 480  **/
 481 gpointer
 482 g_async_queue_try_pop (GAsyncQueue* queue)
 483 {
 484   gpointer retval;
 485
 486   g_return_val_if_fail (queue, NULL);
 487
 488   g_mutex_lock (queue->mutex);
 489   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 490   g_mutex_unlock (queue->mutex);
 491
 492   return retval;
 493 }
 494
 495 /**
 496  * g_async_queue_try_pop_unlocked:
 497  * @queue: a #GAsyncQueue.
 498  *
 499  * Tries to pop data from the @queue. If no data is available, %NULL is
 500  * returned. This function must be called while holding the @queue's
 501  * lock.
 502  *
 503  * Return value: data from the queue or %NULL, when no data is
 504  * available immediately.
 505  **/
 506 gpointer
 507 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
 508 {
 509   g_return_val_if_fail (queue, NULL);
 510
 511   return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 512 }
 513
 514 /**
 515  * g_async_queue_timed_pop:
 516  * @queue: a #GAsyncQueue.
 517  * @end_time: a #GTimeVal, determining the final time.
 518  *
 519  * Pops data from the @queue. If no data is received before @end_time,
 520  * %NULL is returned.
 521  *
 522  * To easily calculate @end_time a combination of g_get_current_time()
 523  * and g_time_val_add() can be used.
 524  *
 525  * Return value: data from the queue or %NULL, when no data is
 526  * received before @end_time.
 527  **/
 528 gpointer
 529 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
 530 {
 531   gpointer retval;
 532
 533   g_return_val_if_fail (queue, NULL);
 534
 535   g_mutex_lock (queue->mutex);
 536   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 537   g_mutex_unlock (queue->mutex);
 538
 539   return retval;
 540 }
 541
 542 /**
 543  * g_async_queue_timed_pop_unlocked:
 544  * @queue: a #GAsyncQueue.
 545  * @end_time: a #GTimeVal, determining the final time.
 546  *
 547  * Pops data from the @queue. If no data is received before @end_time,
 548  * %NULL is returned. This function must be called while holding the
 549  * @queue's lock.
 550  *
 551  * To easily calculate @end_time a combination of g_get_current_time()
 552  * and g_time_val_add() can be used.
 553  *
 554  * Return value: data from the queue or %NULL, when no data is
 555  * received before @end_time.
 556  **/
 557 gpointer
 558 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
 559 {
 560   g_return_val_if_fail (queue, NULL);
 561
 562   return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 563 }
 564
 565 /**
 566  * g_async_queue_length:
 567  * @queue: a #GAsyncQueue.
 568  *
 569  * Returns the length of the queue, negative values mean waiting
 570  * threads, positive values mean available entries in the
 571  * @queue. Actually this function returns the number of data items in
 572  * the queue minus the number of waiting threads. Thus a return value
 573  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 574  * That can happen due to locking of the queue or due to
 575  * scheduling.
 576  *
 577  * Return value: the length of the @queue.
 578  **/
 579 gint
 580 g_async_queue_length (GAsyncQueue* queue)
 581 {
 582   gint retval;
 583
 584   g_return_val_if_fail (queue, 0);
 585
 586   g_mutex_lock (queue->mutex);
 587   retval = queue->queue.length - queue->waiting_threads;
 588   g_mutex_unlock (queue->mutex);
 589
 590   return retval;
 591 }
 592
 593 /**
 594  * g_async_queue_length_unlocked:
 595  * @queue: a #GAsyncQueue.
 596  *
 597  * Returns the length of the queue, negative values mean waiting
 598  * threads, positive values mean available entries in the
 599  * @queue. Actually this function returns the number of data items in
 600  * the queue minus the number of waiting threads. Thus a return value
 601  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 602  * That can happen due to locking of the queue or due to
 603  * scheduling. This function must be called while holding the @queue's
 604  * lock.
 605  *
 606  * Return value: the length of the @queue.
 607  **/
 608 gint
 609 g_async_queue_length_unlocked (GAsyncQueue* queue)
 610 {
 611   g_return_val_if_fail (queue, 0);
 612
 613   return queue->queue.length - queue->waiting_threads;
 614 }
 615
 616 /**
 617  * g_async_queue_sort:
 618  * @queue: a #GAsyncQueue
 619  * @func: the #GCompareDataFunc is used to sort @queue. This
 620  *     function is passed two elements of the @queue. The function
 621  *     should return 0 if they are equal, a negative value if the
 622  *     first element should be higher in the @queue or a positive
 623  *     value if the first element should be lower in the @queue than
 624  *     the second element.
 625  * @user_data: user data passed to @func
 626  *
 627  * Sorts @queue using @func.
 628  *
 629  * This function will lock @queue before it sorts the queue and unlock
 630  * it when it is finished.
 631  *
 632  * If you were sorting a list of priority numbers to make sure the
 633  * lowest priority would be at the top of the queue, you could use:
 634  * |[
 635  *  gint32 id1;
 636  *  gint32 id2;
 637  *
 638  *  id1 = GPOINTER_TO_INT (element1);
 639  *  id2 = GPOINTER_TO_INT (element2);
 640  *
 641  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 642  * ]|
 643  *
 644  * Since: 2.10
 645  **/
 646 void
 647 g_async_queue_sort (GAsyncQueue      *queue,
 648                     GCompareDataFunc  func,
 649                     gpointer          user_data)
 650 {
 651   g_return_if_fail (queue != NULL);
 652   g_return_if_fail (func != NULL);
 653
 654   g_mutex_lock (queue->mutex);
 655   g_async_queue_sort_unlocked (queue, func, user_data);
 656   g_mutex_unlock (queue->mutex);
 657 }
 658
 659 /**
 660  * g_async_queue_sort_unlocked:
 661  * @queue: a #GAsyncQueue
 662  * @func: the #GCompareDataFunc is used to sort @queue. This
 663  *     function is passed two elements of the @queue. The function
 664  *     should return 0 if they are equal, a negative value if the
 665  *     first element should be higher in the @queue or a positive
 666  *     value if the first element should be lower in the @queue than
 667  *     the second element.
 668  * @user_data: user data passed to @func
 669  *
 670  * Sorts @queue using @func.
 671  *
 672  * This function is called while holding the @queue's lock.
 673  *
 674  * Since: 2.10
 675  **/
 676 void
 677 g_async_queue_sort_unlocked (GAsyncQueue      *queue,
 678                              GCompareDataFunc  func,
 679                              gpointer          user_data)
 680 {
 681   SortData sd;
 682
 683   g_return_if_fail (queue != NULL);
 684   g_return_if_fail (func != NULL);
 685
 686   sd.func = func;
 687   sd.user_data = user_data;
 688
 689   g_queue_sort (&queue->queue,
 690                 (GCompareDataFunc)g_async_queue_invert_compare,
 691                 &sd);
 692 }
 693
 694 /*
 695  * Private API
 696  */
 697
 698 GMutex*
 699 _g_async_queue_get_mutex (GAsyncQueue* queue)
 700 {
 701   g_return_val_if_fail (queue, NULL);
 702
 703   return queue->mutex;
 704 }