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   gint32 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   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 160
 161   g_atomic_int_inc (&queue->ref_count);
 162
 163   return queue;
 164 }
 165
 166 /**
 167  * g_async_queue_ref_unlocked:
 168  * @queue: a #GAsyncQueue.
 169  *
 170  * Increases the reference count of the asynchronous @queue by 1.
 171  *
 172  * @Deprecated: Since 2.8, reference counting is done atomically
 173  * so g_async_queue_ref() can be used regardless of the @queue's
 174  * lock.
 175  **/
 176 void
 177 g_async_queue_ref_unlocked (GAsyncQueue *queue)
 178 {
 179   g_return_if_fail (queue);
 180   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 181
 182   g_atomic_int_inc (&queue->ref_count);
 183 }
 184
 185 /**
 186  * g_async_queue_unref_and_unlock:
 187  * @queue: a #GAsyncQueue.
 188  *
 189  * Decreases the reference count of the asynchronous @queue by 1 and
 190  * releases the lock. This function must be called while holding the
 191  * @queue's lock. If the reference count went to 0, the @queue will be
 192  * destroyed and the memory allocated will be freed.
 193  *
 194  * @Deprecated: Since 2.8, reference counting is done atomically
 195  * so g_async_queue_unref() can be used regardless of the @queue's
 196  * lock.
 197  **/
 198 void
 199 g_async_queue_unref_and_unlock (GAsyncQueue *queue)
 200 {
 201   g_return_if_fail (queue);
 202   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 203
 204   g_mutex_unlock (queue->mutex);
 205   g_async_queue_unref (queue);
 206 }
 207
 208 /**
 209  * g_async_queue_unref:
 210  * @queue: a #GAsyncQueue.
 211  *
 212  * Decreases the reference count of the asynchronous @queue by 1. If
 213  * the reference count went to 0, the @queue will be destroyed and the
 214  * memory allocated will be freed. So you are not allowed to use the
 215  * @queue afterwards, as it might have disappeared. You do not need to
 216  * hold the lock to call this function.
 217  **/
 218 void
 219 g_async_queue_unref (GAsyncQueue *queue)
 220 {
 221   g_return_if_fail (queue);
 222   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 223
 224   if (g_atomic_int_dec_and_test (&queue->ref_count))
 225     {
 226       g_return_if_fail (queue->waiting_threads == 0);
 227       g_mutex_free (queue->mutex);
 228       if (queue->cond)
 229         g_cond_free (queue->cond);
 230       if (queue->item_free_func)
 231         g_queue_foreach (&queue->queue, (GFunc) queue->item_free_func, NULL);
 232       g_queue_clear (&queue->queue);
 233       g_free (queue);
 234     }
 235 }
 236
 237 /**
 238  * g_async_queue_lock:
 239  * @queue: a #GAsyncQueue.
 240  *
 241  * Acquires the @queue's lock. After that you can only call the
 242  * <function>g_async_queue_*_unlocked()</function> function variants on that
 243  * @queue. Otherwise it will deadlock.
 244  **/
 245 void
 246 g_async_queue_lock (GAsyncQueue *queue)
 247 {
 248   g_return_if_fail (queue);
 249   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 250
 251   g_mutex_lock (queue->mutex);
 252 }
 253
 254 /**
 255  * g_async_queue_unlock:
 256  * @queue: a #GAsyncQueue.
 257  *
 258  * Releases the queue's lock.
 259  **/
 260 void
 261 g_async_queue_unlock (GAsyncQueue *queue)
 262 {
 263   g_return_if_fail (queue);
 264   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 265
 266   g_mutex_unlock (queue->mutex);
 267 }
 268
 269 /**
 270  * g_async_queue_push:
 271  * @queue: a #GAsyncQueue.
 272  * @data: @data to push into the @queue.
 273  *
 274  * Pushes the @data into the @queue. @data must not be %NULL.
 275  **/
 276 void
 277 g_async_queue_push (GAsyncQueue* queue, gpointer data)
 278 {
 279   g_return_if_fail (queue);
 280   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 281   g_return_if_fail (data);
 282
 283   g_mutex_lock (queue->mutex);
 284   g_async_queue_push_unlocked (queue, data);
 285   g_mutex_unlock (queue->mutex);
 286 }
 287
 288 /**
 289  * g_async_queue_push_unlocked:
 290  * @queue: a #GAsyncQueue.
 291  * @data: @data to push into the @queue.
 292  *
 293  * Pushes the @data into the @queue. @data must not be %NULL. This
 294  * function must be called while holding the @queue's lock.
 295  **/
 296 void
 297 g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)
 298 {
 299   g_return_if_fail (queue);
 300   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 301   g_return_if_fail (data);
 302
 303   g_queue_push_head (&queue->queue, data);
 304   if (queue->waiting_threads > 0)
 305     g_cond_signal (queue->cond);
 306 }
 307
 308 /**
 309  * g_async_queue_push_sorted:
 310  * @queue: a #GAsyncQueue
 311  * @data: the @data to push into the @queue
 312  * @func: the #GCompareDataFunc is used to sort @queue. This function
 313  *     is passed two elements of the @queue. The function should return
 314  *     0 if they are equal, a negative value if the first element
 315  *     should be higher in the @queue or a positive value if the first
 316  *     element should be lower in the @queue than the second element.
 317  * @user_data: user data passed to @func.
 318  *
 319  * Inserts @data into @queue using @func to determine the new
 320  * position.
 321  *
 322  * This function requires that the @queue is sorted before pushing on
 323  * new elements.
 324  *
 325  * This function will lock @queue before it sorts the queue and unlock
 326  * it when it is finished.
 327  *
 328  * For an example of @func see g_async_queue_sort().
 329  *
 330  * Since: 2.10
 331  **/
 332 void
 333 g_async_queue_push_sorted (GAsyncQueue      *queue,
 334                            gpointer          data,
 335                            GCompareDataFunc  func,
 336                            gpointer          user_data)
 337 {
 338   g_return_if_fail (queue != NULL);
 339
 340   g_mutex_lock (queue->mutex);
 341   g_async_queue_push_sorted_unlocked (queue, data, func, user_data);
 342   g_mutex_unlock (queue->mutex);
 343 }
 344
 345 static gint
 346 g_async_queue_invert_compare (gpointer  v1,
 347                               gpointer  v2,
 348                               SortData *sd)
 349 {
 350   return -sd->func (v1, v2, sd->user_data);
 351 }
 352
 353 /**
 354  * g_async_queue_push_sorted_unlocked:
 355  * @queue: a #GAsyncQueue
 356  * @data: the @data to push into the @queue
 357  * @func: the #GCompareDataFunc is used to sort @queue. This function
 358  *     is passed two elements of the @queue. The function should return
 359  *     0 if they are equal, a negative value if the first element
 360  *     should be higher in the @queue or a positive value if the first
 361  *     element should be lower in the @queue than the second element.
 362  * @user_data: user data passed to @func.
 363  *
 364  * Inserts @data into @queue using @func to determine the new
 365  * position.
 366  *
 367  * This function requires that the @queue is sorted before pushing on
 368  * new elements.
 369  *
 370  * This function is called while holding the @queue's lock.
 371  *
 372  * For an example of @func see g_async_queue_sort().
 373  *
 374  * Since: 2.10
 375  **/
 376 void
 377 g_async_queue_push_sorted_unlocked (GAsyncQueue      *queue,
 378                                     gpointer          data,
 379                                     GCompareDataFunc  func,
 380                                     gpointer          user_data)
 381 {
 382   SortData sd;
 383
 384   g_return_if_fail (queue != NULL);
 385
 386   sd.func = func;
 387   sd.user_data = user_data;
 388
 389   g_queue_insert_sorted (&queue->queue,
 390                          data,
 391                          (GCompareDataFunc)g_async_queue_invert_compare,
 392                          &sd);
 393   if (queue->waiting_threads > 0)
 394     g_cond_signal (queue->cond);
 395 }
 396
 397 static gpointer
 398 g_async_queue_pop_intern_unlocked (GAsyncQueue *queue,
 399                                    gboolean     try,
 400                                    GTimeVal    *end_time)
 401 {
 402   gpointer retval;
 403
 404   if (!g_queue_peek_tail_link (&queue->queue))
 405     {
 406       if (try)
 407         return NULL;
 408
 409       if (!queue->cond)
 410         queue->cond = g_cond_new ();
 411
 412       if (!end_time)
 413         {
 414           queue->waiting_threads++;
 415           while (!g_queue_peek_tail_link (&queue->queue))
 416             g_cond_wait (queue->cond, queue->mutex);
 417           queue->waiting_threads--;
 418         }
 419       else
 420         {
 421           queue->waiting_threads++;
 422           while (!g_queue_peek_tail_link (&queue->queue))
 423             if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
 424               break;
 425           queue->waiting_threads--;
 426           if (!g_queue_peek_tail_link (&queue->queue))
 427             return NULL;
 428         }
 429     }
 430
 431   retval = g_queue_pop_tail (&queue->queue);
 432
 433   g_assert (retval);
 434
 435   return retval;
 436 }
 437
 438 /**
 439  * g_async_queue_pop:
 440  * @queue: a #GAsyncQueue.
 441  *
 442  * Pops data from the @queue. This function blocks until data become
 443  * available.
 444  *
 445  * Return value: data from the queue.
 446  **/
 447 gpointer
 448 g_async_queue_pop (GAsyncQueue* queue)
 449 {
 450   gpointer retval;
 451
 452   g_return_val_if_fail (queue, NULL);
 453   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 454
 455   g_mutex_lock (queue->mutex);
 456   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 457   g_mutex_unlock (queue->mutex);
 458
 459   return retval;
 460 }
 461
 462 /**
 463  * g_async_queue_pop_unlocked:
 464  * @queue: a #GAsyncQueue.
 465  *
 466  * Pops data from the @queue. This function blocks until data become
 467  * available. This function must be called while holding the @queue's
 468  * lock.
 469  *
 470  * Return value: data from the queue.
 471  **/
 472 gpointer
 473 g_async_queue_pop_unlocked (GAsyncQueue* queue)
 474 {
 475   g_return_val_if_fail (queue, NULL);
 476   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 477
 478   return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 479 }
 480
 481 /**
 482  * g_async_queue_try_pop:
 483  * @queue: a #GAsyncQueue.
 484  *
 485  * Tries to pop data from the @queue. If no data is available, %NULL is
 486  * returned.
 487  *
 488  * Return value: data from the queue or %NULL, when no data is
 489  * available immediately.
 490  **/
 491 gpointer
 492 g_async_queue_try_pop (GAsyncQueue* queue)
 493 {
 494   gpointer retval;
 495
 496   g_return_val_if_fail (queue, NULL);
 497   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 498
 499   g_mutex_lock (queue->mutex);
 500   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 501   g_mutex_unlock (queue->mutex);
 502
 503   return retval;
 504 }
 505
 506 /**
 507  * g_async_queue_try_pop_unlocked:
 508  * @queue: a #GAsyncQueue.
 509  *
 510  * Tries to pop data from the @queue. If no data is available, %NULL is
 511  * returned. This function must be called while holding the @queue's
 512  * lock.
 513  *
 514  * Return value: data from the queue or %NULL, when no data is
 515  * available immediately.
 516  **/
 517 gpointer
 518 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
 519 {
 520   g_return_val_if_fail (queue, NULL);
 521   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 522
 523   return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 524 }
 525
 526 /**
 527  * g_async_queue_timed_pop:
 528  * @queue: a #GAsyncQueue.
 529  * @end_time: a #GTimeVal, determining the final time.
 530  *
 531  * Pops data from the @queue. If no data is received before @end_time,
 532  * %NULL is returned.
 533  *
 534  * To easily calculate @end_time a combination of g_get_current_time()
 535  * and g_time_val_add() can be used.
 536  *
 537  * Return value: data from the queue or %NULL, when no data is
 538  * received before @end_time.
 539  **/
 540 gpointer
 541 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
 542 {
 543   gpointer retval;
 544
 545   g_return_val_if_fail (queue, NULL);
 546   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 547
 548   g_mutex_lock (queue->mutex);
 549   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 550   g_mutex_unlock (queue->mutex);
 551
 552   return retval;
 553 }
 554
 555 /**
 556  * g_async_queue_timed_pop_unlocked:
 557  * @queue: a #GAsyncQueue.
 558  * @end_time: a #GTimeVal, determining the final time.
 559  *
 560  * Pops data from the @queue. If no data is received before @end_time,
 561  * %NULL is returned. This function must be called while holding the
 562  * @queue's lock.
 563  *
 564  * To easily calculate @end_time a combination of g_get_current_time()
 565  * and g_time_val_add() can be used.
 566  *
 567  * Return value: data from the queue or %NULL, when no data is
 568  * received before @end_time.
 569  **/
 570 gpointer
 571 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
 572 {
 573   g_return_val_if_fail (queue, NULL);
 574   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 575
 576   return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 577 }
 578
 579 /**
 580  * g_async_queue_length:
 581  * @queue: a #GAsyncQueue.
 582  *
 583  * Returns the length of the queue, negative values mean waiting
 584  * threads, positive values mean available entries in the
 585  * @queue. Actually this function returns the number of data items in
 586  * the queue minus the number of waiting threads. Thus a return value
 587  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 588  * That can happen due to locking of the queue or due to
 589  * scheduling.
 590  *
 591  * Return value: the length of the @queue.
 592  **/
 593 gint
 594 g_async_queue_length (GAsyncQueue* queue)
 595 {
 596   gint retval;
 597
 598   g_return_val_if_fail (queue, 0);
 599   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 600
 601   g_mutex_lock (queue->mutex);
 602   retval = queue->queue.length - queue->waiting_threads;
 603   g_mutex_unlock (queue->mutex);
 604
 605   return retval;
 606 }
 607
 608 /**
 609  * g_async_queue_length_unlocked:
 610  * @queue: a #GAsyncQueue.
 611  *
 612  * Returns the length of the queue, negative values mean waiting
 613  * threads, positive values mean available entries in the
 614  * @queue. Actually this function returns the number of data items in
 615  * the queue minus the number of waiting threads. Thus a return value
 616  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 617  * That can happen due to locking of the queue or due to
 618  * scheduling. This function must be called while holding the @queue's
 619  * lock.
 620  *
 621  * Return value: the length of the @queue.
 622  **/
 623 gint
 624 g_async_queue_length_unlocked (GAsyncQueue* queue)
 625 {
 626   g_return_val_if_fail (queue, 0);
 627   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 628
 629   return queue->queue.length - queue->waiting_threads;
 630 }
 631
 632 /**
 633  * g_async_queue_sort:
 634  * @queue: a #GAsyncQueue
 635  * @func: the #GCompareDataFunc is used to sort @queue. This
 636  *     function is passed two elements of the @queue. The function
 637  *     should return 0 if they are equal, a negative value if the
 638  *     first element should be higher in the @queue or a positive
 639  *     value if the first element should be lower in the @queue than
 640  *     the second element.
 641  * @user_data: user data passed to @func
 642  *
 643  * Sorts @queue using @func.
 644  *
 645  * This function will lock @queue before it sorts the queue and unlock
 646  * it when it is finished.
 647  *
 648  * If you were sorting a list of priority numbers to make sure the
 649  * lowest priority would be at the top of the queue, you could use:
 650  * |[
 651  *  gint32 id1;
 652  *  gint32 id2;
 653  *
 654  *  id1 = GPOINTER_TO_INT (element1);
 655  *  id2 = GPOINTER_TO_INT (element2);
 656  *
 657  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 658  * ]|
 659  *
 660  * Since: 2.10
 661  **/
 662 void
 663 g_async_queue_sort (GAsyncQueue      *queue,
 664                     GCompareDataFunc  func,
 665                     gpointer          user_data)
 666 {
 667   g_return_if_fail (queue != NULL);
 668   g_return_if_fail (func != NULL);
 669
 670   g_mutex_lock (queue->mutex);
 671   g_async_queue_sort_unlocked (queue, func, user_data);
 672   g_mutex_unlock (queue->mutex);
 673 }
 674
 675 /**
 676  * g_async_queue_sort_unlocked:
 677  * @queue: a #GAsyncQueue
 678  * @func: the #GCompareDataFunc is used to sort @queue. This
 679  *     function is passed two elements of the @queue. The function
 680  *     should return 0 if they are equal, a negative value if the
 681  *     first element should be higher in the @queue or a positive
 682  *     value if the first element should be lower in the @queue than
 683  *     the second element.
 684  * @user_data: user data passed to @func
 685  *
 686  * Sorts @queue using @func.
 687  *
 688  * This function is called while holding the @queue's lock.
 689  *
 690  * Since: 2.10
 691  **/
 692 void
 693 g_async_queue_sort_unlocked (GAsyncQueue      *queue,
 694                              GCompareDataFunc  func,
 695                              gpointer          user_data)
 696 {
 697   SortData sd;
 698
 699   g_return_if_fail (queue != NULL);
 700   g_return_if_fail (func != NULL);
 701
 702   sd.func = func;
 703   sd.user_data = user_data;
 704
 705   g_queue_sort (&queue->queue,
 706                 (GCompareDataFunc)g_async_queue_invert_compare,
 707                 &sd);
 708 }
 709
 710 /*
 711  * Private API
 712  */
 713
 714 GMutex*
 715 _g_async_queue_get_mutex (GAsyncQueue* queue)
 716 {
 717   g_return_val_if_fail (queue, NULL);
 718   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 719
 720   return queue->mutex;
 721 }