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  * SPDX-License-Identifier: LGPL-2.1-or-later
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2.1 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Lesser General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Lesser General Public
  20  * License along with this library; if not, see <http://www.gnu.org/licenses/>.
  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  * An opaque data structure which represents an asynchronous queue.
  94  *
  95  * It should only be accessed through the `g_async_queue_*` 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  * Returns: 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: (nullable): 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  * Returns: 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  * g_async_queue_*_unlocked() functions on @queue. Otherwise,
 253  * 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: (not nullable): data to push onto the @queue
 285  *
 286  * Pushes the @data into the @queue.
 287  *
 288  * The @data parameter must not be %NULL.
 289  */
 290 void
 291 g_async_queue_push (GAsyncQueue *queue,
 292                     gpointer     data)
 293 {
 294   g_return_if_fail (queue);
 295   g_return_if_fail (data);
 296
 297   g_mutex_lock (&queue->mutex);
 298   g_async_queue_push_unlocked (queue, data);
 299   g_mutex_unlock (&queue->mutex);
 300 }
 301
 302 /**
 303  * g_async_queue_push_unlocked:
 304  * @queue: a #GAsyncQueue
 305  * @data: (not nullable): data to push onto the @queue
 306  *
 307  * Pushes the @data into the @queue.
 308  *
 309  * The @data parameter must not be %NULL.
 310  *
 311  * This function must be called while holding the @queue's lock.
 312  */
 313 void
 314 g_async_queue_push_unlocked (GAsyncQueue *queue,
 315                              gpointer     data)
 316 {
 317   g_return_if_fail (queue);
 318   g_return_if_fail (data);
 319
 320   g_queue_push_head (&queue->queue, data);
 321   if (queue->waiting_threads > 0)
 322     g_cond_signal (&queue->cond);
 323 }
 324
 325 /**
 326  * g_async_queue_push_sorted:
 327  * @queue: a #GAsyncQueue
 328  * @data: (not nullable): the @data to push into the @queue
 329  * @func: the #GCompareDataFunc is used to sort @queue
 330  * @user_data: user data passed to @func.
 331  *
 332  * Inserts @data into @queue using @func to determine the new
 333  * position.
 334  *
 335  * This function requires that the @queue is sorted before pushing on
 336  * new elements, see g_async_queue_sort().
 337  *
 338  * This function will lock @queue before it sorts the queue and unlock
 339  * it when it is finished.
 340  *
 341  * For an example of @func see g_async_queue_sort().
 342  *
 343  * Since: 2.10
 344  */
 345 void
 346 g_async_queue_push_sorted (GAsyncQueue      *queue,
 347                            gpointer          data,
 348                            GCompareDataFunc  func,
 349                            gpointer          user_data)
 350 {
 351   g_return_if_fail (queue != NULL);
 352
 353   g_mutex_lock (&queue->mutex);
 354   g_async_queue_push_sorted_unlocked (queue, data, func, user_data);
 355   g_mutex_unlock (&queue->mutex);
 356 }
 357
 358 static gint
 359 g_async_queue_invert_compare (gpointer  v1,
 360                               gpointer  v2,
 361                               SortData *sd)
 362 {
 363   return -sd->func (v1, v2, sd->user_data);
 364 }
 365
 366 /**
 367  * g_async_queue_push_sorted_unlocked:
 368  * @queue: a #GAsyncQueue
 369  * @data: the data to push into the @queue
 370  * @func: the #GCompareDataFunc is used to sort @queue
 371  * @user_data: user data passed to @func.
 372  *
 373  * Inserts @data into @queue using @func to determine the new
 374  * position.
 375  *
 376  * The sort function @func is passed two elements of the @queue.
 377  * It should return 0 if they are equal, a negative value if the
 378  * first element should be higher in the @queue or a positive value
 379  * if the first element should be lower in the @queue than the second
 380  * element.
 381  *
 382  * This function requires that the @queue is sorted before pushing on
 383  * new elements, see g_async_queue_sort().
 384  *
 385  * This function must be called while holding the @queue's lock.
 386  *
 387  * For an example of @func see g_async_queue_sort().
 388  *
 389  * Since: 2.10
 390  */
 391 void
 392 g_async_queue_push_sorted_unlocked (GAsyncQueue      *queue,
 393                                     gpointer          data,
 394                                     GCompareDataFunc  func,
 395                                     gpointer          user_data)
 396 {
 397   SortData sd;
 398
 399   g_return_if_fail (queue != NULL);
 400
 401   sd.func = func;
 402   sd.user_data = user_data;
 403
 404   g_queue_insert_sorted (&queue->queue,
 405                          data,
 406                          (GCompareDataFunc)g_async_queue_invert_compare,
 407                          &sd);
 408   if (queue->waiting_threads > 0)
 409     g_cond_signal (&queue->cond);
 410 }
 411
 412 static gpointer
 413 g_async_queue_pop_intern_unlocked (GAsyncQueue *queue,
 414                                    gboolean     wait,
 415                                    gint64       end_time)
 416 {
 417   gpointer retval;
 418
 419   if (!g_queue_peek_tail_link (&queue->queue) && wait)
 420     {
 421       queue->waiting_threads++;
 422       while (!g_queue_peek_tail_link (&queue->queue))
 423         {
 424           if (end_time == -1)
 425             g_cond_wait (&queue->cond, &queue->mutex);
 426           else
 427             {
 428               if (!g_cond_wait_until (&queue->cond, &queue->mutex, end_time))
 429                 break;
 430             }
 431         }
 432       queue->waiting_threads--;
 433     }
 434
 435   retval = g_queue_pop_tail (&queue->queue);
 436
 437   g_assert (retval || !wait || end_time > 0);
 438
 439   return retval;
 440 }
 441
 442 /**
 443  * g_async_queue_pop:
 444  * @queue: a #GAsyncQueue
 445  *
 446  * Pops data from the @queue. If @queue is empty, this function
 447  * blocks until data becomes available.
 448  *
 449  * Returns: data from the queue
 450  */
 451 gpointer
 452 g_async_queue_pop (GAsyncQueue *queue)
 453 {
 454   gpointer retval;
 455
 456   g_return_val_if_fail (queue, NULL);
 457
 458   g_mutex_lock (&queue->mutex);
 459   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, -1);
 460   g_mutex_unlock (&queue->mutex);
 461
 462   return retval;
 463 }
 464
 465 /**
 466  * g_async_queue_pop_unlocked:
 467  * @queue: a #GAsyncQueue
 468  *
 469  * Pops data from the @queue. If @queue is empty, this function
 470  * blocks until data becomes available.
 471  *
 472  * This function must be called while holding the @queue's lock.
 473  *
 474  * Returns: data from the queue.
 475  */
 476 gpointer
 477 g_async_queue_pop_unlocked (GAsyncQueue *queue)
 478 {
 479   g_return_val_if_fail (queue, NULL);
 480
 481   return g_async_queue_pop_intern_unlocked (queue, TRUE, -1);
 482 }
 483
 484 /**
 485  * g_async_queue_try_pop:
 486  * @queue: a #GAsyncQueue
 487  *
 488  * Tries to pop data from the @queue. If no data is available,
 489  * %NULL is returned.
 490  *
 491  * Returns: (nullable): data from the queue or %NULL, when no data is
 492  *   available immediately.
 493  */
 494 gpointer
 495 g_async_queue_try_pop (GAsyncQueue *queue)
 496 {
 497   gpointer retval;
 498
 499   g_return_val_if_fail (queue, NULL);
 500
 501   g_mutex_lock (&queue->mutex);
 502   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, -1);
 503   g_mutex_unlock (&queue->mutex);
 504
 505   return retval;
 506 }
 507
 508 /**
 509  * g_async_queue_try_pop_unlocked:
 510  * @queue: a #GAsyncQueue
 511  *
 512  * Tries to pop data from the @queue. If no data is available,
 513  * %NULL is returned.
 514  *
 515  * This function must be called while holding the @queue's lock.
 516  *
 517  * Returns: (nullable): data from the queue or %NULL, when no data is
 518  *   available immediately.
 519  */
 520 gpointer
 521 g_async_queue_try_pop_unlocked (GAsyncQueue *queue)
 522 {
 523   g_return_val_if_fail (queue, NULL);
 524
 525   return g_async_queue_pop_intern_unlocked (queue, FALSE, -1);
 526 }
 527
 528 /**
 529  * g_async_queue_timeout_pop:
 530  * @queue: a #GAsyncQueue
 531  * @timeout: the number of microseconds to wait
 532  *
 533  * Pops data from the @queue. If the queue is empty, blocks for
 534  * @timeout microseconds, or until data becomes available.
 535  *
 536  * If no data is received before the timeout, %NULL is returned.
 537  *
 538  * Returns: (nullable): data from the queue or %NULL, when no data is
 539  *   received before the timeout.
 540  */
 541 gpointer
 542 g_async_queue_timeout_pop (GAsyncQueue *queue,
 543                            guint64      timeout)
 544 {
 545   gint64 end_time = g_get_monotonic_time () + timeout;
 546   gpointer retval;
 547
 548   g_return_val_if_fail (queue != NULL, NULL);
 549
 550   g_mutex_lock (&queue->mutex);
 551   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, end_time);
 552   g_mutex_unlock (&queue->mutex);
 553
 554   return retval;
 555 }
 556
 557 /**
 558  * g_async_queue_timeout_pop_unlocked:
 559  * @queue: a #GAsyncQueue
 560  * @timeout: the number of microseconds to wait
 561  *
 562  * Pops data from the @queue. If the queue is empty, blocks for
 563  * @timeout microseconds, or until data becomes available.
 564  *
 565  * If no data is received before the timeout, %NULL is returned.
 566  *
 567  * This function must be called while holding the @queue's lock.
 568  *
 569  * Returns: (nullable): data from the queue or %NULL, when no data is
 570  *   received before the timeout.
 571  */
 572 gpointer
 573 g_async_queue_timeout_pop_unlocked (GAsyncQueue *queue,
 574                                     guint64      timeout)
 575 {
 576   gint64 end_time = g_get_monotonic_time () + timeout;
 577
 578   g_return_val_if_fail (queue != NULL, NULL);
 579
 580   return g_async_queue_pop_intern_unlocked (queue, TRUE, end_time);
 581 }
 582
 583 /**
 584  * g_async_queue_timed_pop:
 585  * @queue: a #GAsyncQueue
 586  * @end_time: a #GTimeVal, determining the final time
 587  *
 588  * Pops data from the @queue. If the queue is empty, blocks until
 589  * @end_time or until data becomes available.
 590  *
 591  * If no data is received before @end_time, %NULL is returned.
 592  *
 593  * To easily calculate @end_time, a combination of g_get_real_time()
 594  * and g_time_val_add() can be used.
 595  *
 596  * Returns: (nullable): data from the queue or %NULL, when no data is
 597  *   received before @end_time.
 598  *
 599  * Deprecated: use g_async_queue_timeout_pop().
 600  */
 601 G_GNUC_BEGIN_IGNORE_DEPRECATIONS
 602 gpointer
 603 g_async_queue_timed_pop (GAsyncQueue *queue,
 604                          GTimeVal    *end_time)
 605 {
 606   gint64 m_end_time;
 607   gpointer retval;
 608
 609   g_return_val_if_fail (queue, NULL);
 610
 611   if (end_time != NULL)
 612     {
 613       m_end_time = g_get_monotonic_time () +
 614         ((gint64) end_time->tv_sec * G_USEC_PER_SEC + end_time->tv_usec - g_get_real_time ());
 615     }
 616   else
 617     m_end_time = -1;
 618
 619   g_mutex_lock (&queue->mutex);
 620   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, m_end_time);
 621   g_mutex_unlock (&queue->mutex);
 622
 623   return retval;
 624 }
 625 G_GNUC_END_IGNORE_DEPRECATIONS
 626
 627 /**
 628  * g_async_queue_timed_pop_unlocked:
 629  * @queue: a #GAsyncQueue
 630  * @end_time: a #GTimeVal, determining the final time
 631  *
 632  * Pops data from the @queue. If the queue is empty, blocks until
 633  * @end_time or until data becomes available.
 634  *
 635  * If no data is received before @end_time, %NULL is returned.
 636  *
 637  * To easily calculate @end_time, a combination of g_get_real_time()
 638  * and g_time_val_add() can be used.
 639  *
 640  * This function must be called while holding the @queue's lock.
 641  *
 642  * Returns: (nullable): data from the queue or %NULL, when no data is
 643  *   received before @end_time.
 644  *
 645  * Deprecated: use g_async_queue_timeout_pop_unlocked().
 646  */
 647 G_GNUC_BEGIN_IGNORE_DEPRECATIONS
 648 gpointer
 649 g_async_queue_timed_pop_unlocked (GAsyncQueue *queue,
 650                                   GTimeVal    *end_time)
 651 {
 652   gint64 m_end_time;
 653
 654   g_return_val_if_fail (queue, NULL);
 655
 656   if (end_time != NULL)
 657     {
 658       m_end_time = g_get_monotonic_time () +
 659         ((gint64) end_time->tv_sec * G_USEC_PER_SEC + end_time->tv_usec - g_get_real_time ());
 660     }
 661   else
 662     m_end_time = -1;
 663
 664   return g_async_queue_pop_intern_unlocked (queue, TRUE, m_end_time);
 665 }
 666 G_GNUC_END_IGNORE_DEPRECATIONS
 667
 668 /**
 669  * g_async_queue_length:
 670  * @queue: a #GAsyncQueue.
 671  *
 672  * Returns the length of the queue.
 673  *
 674  * Actually this function returns the number of data items in
 675  * the queue minus the number of waiting threads, so a negative
 676  * value means waiting threads, and a positive value means available
 677  * entries in the @queue. A return value of 0 could mean n entries
 678  * in the queue and n threads waiting. This can happen due to locking
 679  * of the queue or due to scheduling.
 680  *
 681  * Returns: the length of the @queue
 682  */
 683 gint
 684 g_async_queue_length (GAsyncQueue *queue)
 685 {
 686   gint retval;
 687
 688   g_return_val_if_fail (queue, 0);
 689
 690   g_mutex_lock (&queue->mutex);
 691   retval = queue->queue.length - queue->waiting_threads;
 692   g_mutex_unlock (&queue->mutex);
 693
 694   return retval;
 695 }
 696
 697 /**
 698  * g_async_queue_length_unlocked:
 699  * @queue: a #GAsyncQueue
 700  *
 701  * Returns the length of the queue.
 702  *
 703  * Actually this function returns the number of data items in
 704  * the queue minus the number of waiting threads, so a negative
 705  * value means waiting threads, and a positive value means available
 706  * entries in the @queue. A return value of 0 could mean n entries
 707  * in the queue and n threads waiting. This can happen due to locking
 708  * of the queue or due to scheduling.
 709  *
 710  * This function must be called while holding the @queue's lock.
 711  *
 712  * Returns: the length of the @queue.
 713  */
 714 gint
 715 g_async_queue_length_unlocked (GAsyncQueue *queue)
 716 {
 717   g_return_val_if_fail (queue, 0);
 718
 719   return queue->queue.length - queue->waiting_threads;
 720 }
 721
 722 /**
 723  * g_async_queue_sort:
 724  * @queue: a #GAsyncQueue
 725  * @func: the #GCompareDataFunc is used to sort @queue
 726  * @user_data: user data passed to @func
 727  *
 728  * Sorts @queue using @func.
 729  *
 730  * The sort function @func is passed two elements of the @queue.
 731  * It should return 0 if they are equal, a negative value if the
 732  * first element should be higher in the @queue or a positive value
 733  * if the first element should be lower in the @queue than the second
 734  * element.
 735  *
 736  * This function will lock @queue before it sorts the queue and unlock
 737  * it when it is finished.
 738  *
 739  * If you were sorting a list of priority numbers to make sure the
 740  * lowest priority would be at the top of the queue, you could use:
 741  * |[<!-- language="C" -->
 742  *  gint32 id1;
 743  *  gint32 id2;
 744  *
 745  *  id1 = GPOINTER_TO_INT (element1);
 746  *  id2 = GPOINTER_TO_INT (element2);
 747  *
 748  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 749  * ]|
 750  *
 751  * Since: 2.10
 752  */
 753 void
 754 g_async_queue_sort (GAsyncQueue      *queue,
 755                     GCompareDataFunc  func,
 756                     gpointer          user_data)
 757 {
 758   g_return_if_fail (queue != NULL);
 759   g_return_if_fail (func != NULL);
 760
 761   g_mutex_lock (&queue->mutex);
 762   g_async_queue_sort_unlocked (queue, func, user_data);
 763   g_mutex_unlock (&queue->mutex);
 764 }
 765
 766 /**
 767  * g_async_queue_sort_unlocked:
 768  * @queue: a #GAsyncQueue
 769  * @func: the #GCompareDataFunc is used to sort @queue
 770  * @user_data: user data passed to @func
 771  *
 772  * Sorts @queue using @func.
 773  *
 774  * The sort function @func is passed two elements of the @queue.
 775  * It should return 0 if they are equal, a negative value if the
 776  * first element should be higher in the @queue or a positive value
 777  * if the first element should be lower in the @queue than the second
 778  * element.
 779  *
 780  * This function must be called while holding the @queue's lock.
 781  *
 782  * Since: 2.10
 783  */
 784 void
 785 g_async_queue_sort_unlocked (GAsyncQueue      *queue,
 786                              GCompareDataFunc  func,
 787                              gpointer          user_data)
 788 {
 789   SortData sd;
 790
 791   g_return_if_fail (queue != NULL);
 792   g_return_if_fail (func != NULL);
 793
 794   sd.func = func;
 795   sd.user_data = user_data;
 796
 797   g_queue_sort (&queue->queue,
 798                 (GCompareDataFunc)g_async_queue_invert_compare,
 799                 &sd);
 800 }
 801
 802 /**
 803  * g_async_queue_remove:
 804  * @queue: a #GAsyncQueue
 805  * @item: (not nullable): the data to remove from the @queue
 806  *
 807  * Remove an item from the queue.
 808  *
 809  * Returns: %TRUE if the item was removed
 810  *
 811  * Since: 2.46
 812  */
 813 gboolean
 814 g_async_queue_remove (GAsyncQueue *queue,
 815                       gpointer     item)
 816 {
 817   gboolean ret;
 818
 819   g_return_val_if_fail (queue != NULL, FALSE);
 820   g_return_val_if_fail (item != NULL, FALSE);
 821
 822   g_mutex_lock (&queue->mutex);
 823   ret = g_async_queue_remove_unlocked (queue, item);
 824   g_mutex_unlock (&queue->mutex);
 825
 826   return ret;
 827 }
 828
 829 /**
 830  * g_async_queue_remove_unlocked:
 831  * @queue: a #GAsyncQueue
 832  * @item: the data to remove from the @queue
 833  *
 834  * Remove an item from the queue.
 835  *
 836  * This function must be called while holding the @queue's lock.
 837  *
 838  * Returns: %TRUE if the item was removed
 839  *
 840  * Since: 2.46
 841  */
 842 gboolean
 843 g_async_queue_remove_unlocked (GAsyncQueue *queue,
 844                                gpointer     item)
 845 {
 846   g_return_val_if_fail (queue != NULL, FALSE);
 847   g_return_val_if_fail (item != NULL, FALSE);
 848
 849   return g_queue_remove (&queue->queue, item);
 850 }
 851
 852 /**
 853  * g_async_queue_push_front:
 854  * @queue: a #GAsyncQueue
 855  * @item: (not nullable): data to push into the @queue
 856  *
 857  * Pushes the @item into the @queue. @item must not be %NULL.
 858  * In contrast to g_async_queue_push(), this function
 859  * pushes the new item ahead of the items already in the queue,
 860  * so that it will be the next one to be popped off the queue.
 861  *
 862  * Since: 2.46
 863  */
 864 void
 865 g_async_queue_push_front (GAsyncQueue *queue,
 866                           gpointer     item)
 867 {
 868   g_return_if_fail (queue != NULL);
 869   g_return_if_fail (item != NULL);
 870
 871   g_mutex_lock (&queue->mutex);
 872   g_async_queue_push_front_unlocked (queue, item);
 873   g_mutex_unlock (&queue->mutex);
 874 }
 875
 876 /**
 877  * g_async_queue_push_front_unlocked:
 878  * @queue: a #GAsyncQueue
 879  * @item: (not nullable): data to push into the @queue
 880  *
 881  * Pushes the @item into the @queue. @item must not be %NULL.
 882  * In contrast to g_async_queue_push_unlocked(), this function
 883  * pushes the new item ahead of the items already in the queue,
 884  * so that it will be the next one to be popped off the queue.
 885  *
 886  * This function must be called while holding the @queue's lock.
 887  *
 888  * Since: 2.46
 889  */
 890 void
 891 g_async_queue_push_front_unlocked (GAsyncQueue *queue,
 892                                    gpointer     item)
 893 {
 894   g_return_if_fail (queue != NULL);
 895   g_return_if_fail (item != NULL);
 896
 897   g_queue_push_tail (&queue->queue, item);
 898   if (queue->waiting_threads > 0)
 899     g_cond_signal (&queue->cond);
 900 }
 901
 902 /*
 903  * Private API
 904  */
 905
 906 GMutex *
 907 _g_async_queue_get_mutex (GAsyncQueue *queue)
 908 {
 909   g_return_val_if_fail (queue, NULL);
 910
 911   return &queue->mutex;
 912 }