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, see <http://www.gnu.org/licenses/>.
  19  */
  20
  21 /*
  22  * MT safe
  23  */
  24
  25 #include "config.h"
  26
  27 #include "gasyncqueue.h"
  28 #include "gasyncqueueprivate.h"
  29
  30 #include "gmain.h"
  31 #include "gmem.h"
  32 #include "gqueue.h"
  33 #include "gtestutils.h"
  34 #include "gtimer.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_timeout_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  * g_async_queue_* 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  * Returns: 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  * Returns: 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: 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: 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  * g_async_queue_*_unlocked() functions on @queue. Otherwise,
 251  * 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                                    gint64       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 (end_time == -1)
 419             g_cond_wait (&queue->cond, &queue->mutex);
 420           else
 421             {
 422               if (!g_cond_wait_until (&queue->cond, &queue->mutex, end_time))
 423                 break;
 424             }
 425         }
 426       queue->waiting_threads--;
 427     }
 428
 429   retval = g_queue_pop_tail (&queue->queue);
 430
 431   g_assert (retval || !wait || end_time > 0);
 432
 433   return retval;
 434 }
 435
 436 /**
 437  * g_async_queue_pop:
 438  * @queue: a #GAsyncQueue
 439  *
 440  * Pops data from the @queue. If @queue is empty, this function
 441  * blocks until data becomes available.
 442  *
 443  * Returns: data from the queue
 444  */
 445 gpointer
 446 g_async_queue_pop (GAsyncQueue *queue)
 447 {
 448   gpointer retval;
 449
 450   g_return_val_if_fail (queue, NULL);
 451
 452   g_mutex_lock (&queue->mutex);
 453   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, -1);
 454   g_mutex_unlock (&queue->mutex);
 455
 456   return retval;
 457 }
 458
 459 /**
 460  * g_async_queue_pop_unlocked:
 461  * @queue: a #GAsyncQueue
 462  *
 463  * Pops data from the @queue. If @queue is empty, this function
 464  * blocks until data becomes available.
 465  *
 466  * This function must be called while holding the @queue's lock.
 467  *
 468  * Returns: data from the queue.
 469  */
 470 gpointer
 471 g_async_queue_pop_unlocked (GAsyncQueue *queue)
 472 {
 473   g_return_val_if_fail (queue, NULL);
 474
 475   return g_async_queue_pop_intern_unlocked (queue, TRUE, -1);
 476 }
 477
 478 /**
 479  * g_async_queue_try_pop:
 480  * @queue: a #GAsyncQueue
 481  *
 482  * Tries to pop data from the @queue. If no data is available,
 483  * %NULL is returned.
 484  *
 485  * Returns: data from the queue or %NULL, when no data is
 486  *     available immediately.
 487  */
 488 gpointer
 489 g_async_queue_try_pop (GAsyncQueue *queue)
 490 {
 491   gpointer retval;
 492
 493   g_return_val_if_fail (queue, NULL);
 494
 495   g_mutex_lock (&queue->mutex);
 496   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, -1);
 497   g_mutex_unlock (&queue->mutex);
 498
 499   return retval;
 500 }
 501
 502 /**
 503  * g_async_queue_try_pop_unlocked:
 504  * @queue: a #GAsyncQueue
 505  *
 506  * Tries to pop data from the @queue. If no data is available,
 507  * %NULL is returned.
 508  *
 509  * This function must be called while holding the @queue's lock.
 510  *
 511  * Returns: data from the queue or %NULL, when no data is
 512  *     available immediately.
 513  */
 514 gpointer
 515 g_async_queue_try_pop_unlocked (GAsyncQueue *queue)
 516 {
 517   g_return_val_if_fail (queue, NULL);
 518
 519   return g_async_queue_pop_intern_unlocked (queue, FALSE, -1);
 520 }
 521
 522 /**
 523  * g_async_queue_timeout_pop:
 524  * @queue: a #GAsyncQueue
 525  * @timeout: the number of microseconds to wait
 526  *
 527  * Pops data from the @queue. If the queue is empty, blocks for
 528  * @timeout microseconds, or until data becomes available.
 529  *
 530  * If no data is received before the timeout, %NULL is returned.
 531  *
 532  * Returns: data from the queue or %NULL, when no data is
 533  *     received before the timeout.
 534  */
 535 gpointer
 536 g_async_queue_timeout_pop (GAsyncQueue *queue,
 537                            guint64      timeout)
 538 {
 539   gint64 end_time = g_get_monotonic_time () + timeout;
 540   gpointer retval;
 541
 542   g_mutex_lock (&queue->mutex);
 543   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, end_time);
 544   g_mutex_unlock (&queue->mutex);
 545
 546   return retval;
 547 }
 548
 549 /**
 550  * g_async_queue_timeout_pop_unlocked:
 551  * @queue: a #GAsyncQueue
 552  * @timeout: the number of microseconds to wait
 553  *
 554  * Pops data from the @queue. If the queue is empty, blocks for
 555  * @timeout microseconds, or until data becomes available.
 556  *
 557  * If no data is received before the timeout, %NULL is returned.
 558  *
 559  * This function must be called while holding the @queue's lock.
 560  *
 561  * Returns: data from the queue or %NULL, when no data is
 562  *     received before the timeout.
 563  */
 564 gpointer
 565 g_async_queue_timeout_pop_unlocked (GAsyncQueue *queue,
 566                                     guint64      timeout)
 567 {
 568   gint64 end_time = g_get_monotonic_time () + timeout;
 569
 570   return g_async_queue_pop_intern_unlocked (queue, TRUE, end_time);
 571 }
 572
 573 /**
 574  * g_async_queue_timed_pop:
 575  * @queue: a #GAsyncQueue
 576  * @end_time: a #GTimeVal, determining the final time
 577  *
 578  * Pops data from the @queue. If the queue is empty, blocks until
 579  * @end_time or until data becomes available.
 580  *
 581  * If no data is received before @end_time, %NULL is returned.
 582  *
 583  * To easily calculate @end_time, a combination of g_get_current_time()
 584  * and g_time_val_add() can be used.
 585  *
 586  * Returns: data from the queue or %NULL, when no data is
 587  *     received before @end_time.
 588  *
 589  * Deprecated: use g_async_queue_timeout_pop().
 590  */
 591 gpointer
 592 g_async_queue_timed_pop (GAsyncQueue *queue,
 593                          GTimeVal    *end_time)
 594 {
 595   gint64 m_end_time;
 596   gpointer retval;
 597
 598   g_return_val_if_fail (queue, NULL);
 599
 600   if (end_time != NULL)
 601     {
 602       m_end_time = g_get_monotonic_time () +
 603         ((gint64) end_time->tv_sec * G_USEC_PER_SEC + end_time->tv_usec - g_get_real_time ());
 604     }
 605   else
 606     m_end_time = -1;
 607
 608   g_mutex_lock (&queue->mutex);
 609   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, m_end_time);
 610   g_mutex_unlock (&queue->mutex);
 611
 612   return retval;
 613 }
 614
 615 /**
 616  * g_async_queue_timed_pop_unlocked:
 617  * @queue: a #GAsyncQueue
 618  * @end_time: a #GTimeVal, determining the final time
 619  *
 620  * Pops data from the @queue. If the queue is empty, blocks until
 621  * @end_time or until data becomes available.
 622  *
 623  * If no data is received before @end_time, %NULL is returned.
 624  *
 625  * To easily calculate @end_time, a combination of g_get_current_time()
 626  * and g_time_val_add() can be used.
 627  *
 628  * This function must be called while holding the @queue's lock.
 629  *
 630  * Returns: data from the queue or %NULL, when no data is
 631  *     received before @end_time.
 632  *
 633  * Deprecated: use g_async_queue_timeout_pop_unlocked().
 634  */
 635 gpointer
 636 g_async_queue_timed_pop_unlocked (GAsyncQueue *queue,
 637                                   GTimeVal    *end_time)
 638 {
 639   gint64 m_end_time;
 640
 641   g_return_val_if_fail (queue, NULL);
 642
 643   if (end_time != NULL)
 644     {
 645       m_end_time = g_get_monotonic_time () +
 646         ((gint64) end_time->tv_sec * G_USEC_PER_SEC + end_time->tv_usec - g_get_real_time ());
 647     }
 648   else
 649     m_end_time = -1;
 650
 651   return g_async_queue_pop_intern_unlocked (queue, TRUE, m_end_time);
 652 }
 653
 654 /**
 655  * g_async_queue_length:
 656  * @queue: a #GAsyncQueue.
 657  *
 658  * Returns the length of the queue.
 659  *
 660  * Actually this function returns the number of data items in
 661  * the queue minus the number of waiting threads, so a negative
 662  * value means waiting threads, and a positive value means available
 663  * entries in the @queue. A return value of 0 could mean n entries
 664  * in the queue and n threads waiting. This can happen due to locking
 665  * of the queue or due to scheduling.
 666  *
 667  * Returns: the length of the @queue
 668  */
 669 gint
 670 g_async_queue_length (GAsyncQueue *queue)
 671 {
 672   gint retval;
 673
 674   g_return_val_if_fail (queue, 0);
 675
 676   g_mutex_lock (&queue->mutex);
 677   retval = queue->queue.length - queue->waiting_threads;
 678   g_mutex_unlock (&queue->mutex);
 679
 680   return retval;
 681 }
 682
 683 /**
 684  * g_async_queue_length_unlocked:
 685  * @queue: a #GAsyncQueue
 686  *
 687  * Returns the length of the queue.
 688  *
 689  * Actually this function returns the number of data items in
 690  * the queue minus the number of waiting threads, so a negative
 691  * value means waiting threads, and a positive value means available
 692  * entries in the @queue. A return value of 0 could mean n entries
 693  * in the queue and n threads waiting. This can happen due to locking
 694  * of the queue or due to scheduling.
 695  *
 696  * This function must be called while holding the @queue's lock.
 697  *
 698  * Returns: the length of the @queue.
 699  */
 700 gint
 701 g_async_queue_length_unlocked (GAsyncQueue *queue)
 702 {
 703   g_return_val_if_fail (queue, 0);
 704
 705   return queue->queue.length - queue->waiting_threads;
 706 }
 707
 708 /**
 709  * g_async_queue_sort:
 710  * @queue: a #GAsyncQueue
 711  * @func: the #GCompareDataFunc is used to sort @queue
 712  * @user_data: user data passed to @func
 713  *
 714  * Sorts @queue using @func.
 715  *
 716  * The sort function @func is passed two elements of the @queue.
 717  * It should return 0 if they are equal, a negative value if the
 718  * first element should be higher in the @queue or a positive value
 719  * if the first element should be lower in the @queue than the second
 720  * element.
 721  *
 722  * This function will lock @queue before it sorts the queue and unlock
 723  * it when it is finished.
 724  *
 725  * If you were sorting a list of priority numbers to make sure the
 726  * lowest priority would be at the top of the queue, you could use:
 727  * |[<!-- language="C" -->
 728  *  gint32 id1;
 729  *  gint32 id2;
 730  *
 731  *  id1 = GPOINTER_TO_INT (element1);
 732  *  id2 = GPOINTER_TO_INT (element2);
 733  *
 734  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 735  * ]|
 736  *
 737  * Since: 2.10
 738  */
 739 void
 740 g_async_queue_sort (GAsyncQueue      *queue,
 741                     GCompareDataFunc  func,
 742                     gpointer          user_data)
 743 {
 744   g_return_if_fail (queue != NULL);
 745   g_return_if_fail (func != NULL);
 746
 747   g_mutex_lock (&queue->mutex);
 748   g_async_queue_sort_unlocked (queue, func, user_data);
 749   g_mutex_unlock (&queue->mutex);
 750 }
 751
 752 /**
 753  * g_async_queue_sort_unlocked:
 754  * @queue: a #GAsyncQueue
 755  * @func: the #GCompareDataFunc is used to sort @queue
 756  * @user_data: user data passed to @func
 757  *
 758  * Sorts @queue using @func.
 759  *
 760  * The sort function @func is passed two elements of the @queue.
 761  * It should return 0 if they are equal, a negative value if the
 762  * first element should be higher in the @queue or a positive value
 763  * if the first element should be lower in the @queue than the second
 764  * element.
 765  *
 766  * This function must be called while holding the @queue's lock.
 767  *
 768  * Since: 2.10
 769  */
 770 void
 771 g_async_queue_sort_unlocked (GAsyncQueue      *queue,
 772                              GCompareDataFunc  func,
 773                              gpointer          user_data)
 774 {
 775   SortData sd;
 776
 777   g_return_if_fail (queue != NULL);
 778   g_return_if_fail (func != NULL);
 779
 780   sd.func = func;
 781   sd.user_data = user_data;
 782
 783   g_queue_sort (&queue->queue,
 784                 (GCompareDataFunc)g_async_queue_invert_compare,
 785                 &sd);
 786 }
 787
 788 /**
 789  * g_async_queue_remove:
 790  * @queue: a #GAsyncQueue
 791  * @item: the data to remove from the @queue
 792  *
 793  * Remove an item from the queue.
 794  *
 795  * Returns: %TRUE if the item was removed
 796  *
 797  * Since: 2.46
 798  */
 799 gboolean
 800 g_async_queue_remove (GAsyncQueue *queue,
 801                       gpointer     item)
 802 {
 803   gboolean ret;
 804
 805   g_return_val_if_fail (queue != NULL, FALSE);
 806   g_return_val_if_fail (item != NULL, FALSE);
 807
 808   g_mutex_lock (&queue->mutex);
 809   ret = g_async_queue_remove_unlocked (queue, item);
 810   g_mutex_unlock (&queue->mutex);
 811
 812   return ret;
 813 }
 814
 815 /**
 816  * g_async_queue_remove_unlocked:
 817  * @queue: a #GAsyncQueue
 818  * @item: the data to remove from the @queue
 819  *
 820  * Remove an item from the queue.
 821  *
 822  * This function must be called while holding the @queue's lock.
 823  *
 824  * Returns: %TRUE if the item was removed
 825  *
 826  * Since: 2.46
 827  */
 828 gboolean
 829 g_async_queue_remove_unlocked (GAsyncQueue *queue,
 830                                gpointer     item)
 831 {
 832   g_return_val_if_fail (queue != NULL, FALSE);
 833   g_return_val_if_fail (item != NULL, FALSE);
 834
 835   return g_queue_remove (&queue->queue, item);
 836 }
 837
 838 /**
 839  * g_async_queue_push_front:
 840  * @queue: a #GAsyncQueue
 841  * @item: data to push into the @queue
 842  *
 843  * Pushes the @item into the @queue. @item must not be %NULL.
 844  * In contrast to g_async_queue_push(), this function
 845  * pushes the new item ahead of the items already in the queue,
 846  * so that it will be the next one to be popped off the queue.
 847  *
 848  * Since: 2.46
 849  */
 850 void
 851 g_async_queue_push_front (GAsyncQueue *queue,
 852                           gpointer     item)
 853 {
 854   g_return_if_fail (queue != NULL);
 855   g_return_if_fail (item != NULL);
 856
 857   g_mutex_lock (&queue->mutex);
 858   g_async_queue_push_front_unlocked (queue, item);
 859   g_mutex_unlock (&queue->mutex);
 860 }
 861
 862 /**
 863  * g_async_queue_push_front_unlocked:
 864  * @queue: a #GAsyncQueue
 865  * @item: data to push into the @queue
 866  *
 867  * Pushes the @item into the @queue. @item must not be %NULL.
 868  * In contrast to g_async_queue_push_unlocked(), this function
 869  * pushes the new item ahead of the items already in the queue,
 870  * so that it will be the next one to be popped off the queue.
 871  *
 872  * This function must be called while holding the @queue's lock.
 873  *
 874  * Since: 2.46
 875  */
 876 void
 877 g_async_queue_push_front_unlocked (GAsyncQueue *queue,
 878                                    gpointer     item)
 879 {
 880   g_return_if_fail (queue != NULL);
 881   g_return_if_fail (item != NULL);
 882
 883   g_queue_push_tail (&queue->queue, item);
 884   if (queue->waiting_threads > 0)
 885     g_cond_signal (&queue->cond);
 886 }
 887
 888 /*
 889  * Private API
 890  */
 891
 892 GMutex *
 893 _g_async_queue_get_mutex (GAsyncQueue *queue)
 894 {
 895   g_return_val_if_fail (queue, NULL);
 896
 897   return &queue->mutex;
 898 }