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