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