1 /* GLIB - Library of useful routines for C programming
2 * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald
4 * GAsyncQueue: asynchronous queue implementation, based on GQueue.
5 * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
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.1 of the License, or (at your option) any later version.
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.
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/>.
27 #include "gasyncqueue.h"
28 #include "gasyncqueueprivate.h"
33 #include "gtestutils.h"
36 #include "deprecated/gthread.h"
40 * SECTION:async_queues
41 * @title: Asynchronous Queues
42 * @short_description: asynchronous communication between threads
43 * @see_also: #GThreadPool
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.
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.
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
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.
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.
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).
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
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.
100 GDestroyNotify item_free_func
;
101 guint waiting_threads
;
107 GCompareDataFunc func
;
114 * Creates a new asynchronous queue.
116 * Returns: a new #GAsyncQueue. Free with g_async_queue_unref()
119 g_async_queue_new (void)
121 return g_async_queue_new_full (NULL
);
125 * g_async_queue_new_full:
126 * @item_free_func: function to free queue elements
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.
132 * Returns: a new #GAsyncQueue. Free with g_async_queue_unref()
137 g_async_queue_new_full (GDestroyNotify item_free_func
)
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
;
154 * @queue: a #GAsyncQueue
156 * Increases the reference count of the asynchronous @queue by 1.
157 * You do not need to hold the lock to call this function.
159 * Returns: the @queue that was passed in (since 2.6)
162 g_async_queue_ref (GAsyncQueue
*queue
)
164 g_return_val_if_fail (queue
, NULL
);
166 g_atomic_int_inc (&queue
->ref_count
);
172 * g_async_queue_ref_unlocked:
173 * @queue: a #GAsyncQueue
175 * Increases the reference count of the asynchronous @queue by 1.
177 * Deprecated: 2.8: Reference counting is done atomically.
178 * so g_async_queue_ref() can be used regardless of the @queue's
182 g_async_queue_ref_unlocked (GAsyncQueue
*queue
)
184 g_return_if_fail (queue
);
186 g_atomic_int_inc (&queue
->ref_count
);
190 * g_async_queue_unref_and_unlock:
191 * @queue: a #GAsyncQueue
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.
198 * Deprecated: 2.8: Reference counting is done atomically.
199 * so g_async_queue_unref() can be used regardless of the @queue's
203 g_async_queue_unref_and_unlock (GAsyncQueue
*queue
)
205 g_return_if_fail (queue
);
207 g_mutex_unlock (&queue
->mutex
);
208 g_async_queue_unref (queue
);
212 * g_async_queue_unref:
213 * @queue: a #GAsyncQueue.
215 * Decreases the reference count of the asynchronous @queue by 1.
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.
223 g_async_queue_unref (GAsyncQueue
*queue
)
225 g_return_if_fail (queue
);
227 if (g_atomic_int_dec_and_test (&queue
->ref_count
))
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
);
240 * g_async_queue_lock:
241 * @queue: a #GAsyncQueue
243 * Acquires the @queue's lock. If another thread is already
244 * holding the lock, this call will block until the lock
247 * Call g_async_queue_unlock() to drop the lock again.
249 * While holding the lock, you can only call the
250 * g_async_queue_*_unlocked() functions on @queue. Otherwise,
251 * deadlock may occur.
254 g_async_queue_lock (GAsyncQueue
*queue
)
256 g_return_if_fail (queue
);
258 g_mutex_lock (&queue
->mutex
);
262 * g_async_queue_unlock:
263 * @queue: a #GAsyncQueue
265 * Releases the queue's lock.
267 * Calling this function when you have not acquired
268 * the with g_async_queue_lock() leads to undefined
272 g_async_queue_unlock (GAsyncQueue
*queue
)
274 g_return_if_fail (queue
);
276 g_mutex_unlock (&queue
->mutex
);
280 * g_async_queue_push:
281 * @queue: a #GAsyncQueue
282 * @data: @data to push into the @queue
284 * Pushes the @data into the @queue. @data must not be %NULL.
287 g_async_queue_push (GAsyncQueue
*queue
,
290 g_return_if_fail (queue
);
291 g_return_if_fail (data
);
293 g_mutex_lock (&queue
->mutex
);
294 g_async_queue_push_unlocked (queue
, data
);
295 g_mutex_unlock (&queue
->mutex
);
299 * g_async_queue_push_unlocked:
300 * @queue: a #GAsyncQueue
301 * @data: @data to push into the @queue
303 * Pushes the @data into the @queue. @data must not be %NULL.
305 * This function must be called while holding the @queue's lock.
308 g_async_queue_push_unlocked (GAsyncQueue
*queue
,
311 g_return_if_fail (queue
);
312 g_return_if_fail (data
);
314 g_queue_push_head (&queue
->queue
, data
);
315 if (queue
->waiting_threads
> 0)
316 g_cond_signal (&queue
->cond
);
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.
326 * Inserts @data into @queue using @func to determine the new
329 * This function requires that the @queue is sorted before pushing on
330 * new elements, see g_async_queue_sort().
332 * This function will lock @queue before it sorts the queue and unlock
333 * it when it is finished.
335 * For an example of @func see g_async_queue_sort().
340 g_async_queue_push_sorted (GAsyncQueue
*queue
,
342 GCompareDataFunc func
,
345 g_return_if_fail (queue
!= NULL
);
347 g_mutex_lock (&queue
->mutex
);
348 g_async_queue_push_sorted_unlocked (queue
, data
, func
, user_data
);
349 g_mutex_unlock (&queue
->mutex
);
353 g_async_queue_invert_compare (gpointer v1
,
357 return -sd
->func (v1
, v2
, sd
->user_data
);
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.
367 * Inserts @data into @queue using @func to determine the new
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
376 * This function requires that the @queue is sorted before pushing on
377 * new elements, see g_async_queue_sort().
379 * This function must be called while holding the @queue's lock.
381 * For an example of @func see g_async_queue_sort().
386 g_async_queue_push_sorted_unlocked (GAsyncQueue
*queue
,
388 GCompareDataFunc func
,
393 g_return_if_fail (queue
!= NULL
);
396 sd
.user_data
= user_data
;
398 g_queue_insert_sorted (&queue
->queue
,
400 (GCompareDataFunc
)g_async_queue_invert_compare
,
402 if (queue
->waiting_threads
> 0)
403 g_cond_signal (&queue
->cond
);
407 g_async_queue_pop_intern_unlocked (GAsyncQueue
*queue
,
413 if (!g_queue_peek_tail_link (&queue
->queue
) && wait
)
415 queue
->waiting_threads
++;
416 while (!g_queue_peek_tail_link (&queue
->queue
))
419 g_cond_wait (&queue
->cond
, &queue
->mutex
);
422 if (!g_cond_wait_until (&queue
->cond
, &queue
->mutex
, end_time
))
426 queue
->waiting_threads
--;
429 retval
= g_queue_pop_tail (&queue
->queue
);
431 g_assert (retval
|| !wait
|| end_time
> 0);
438 * @queue: a #GAsyncQueue
440 * Pops data from the @queue. If @queue is empty, this function
441 * blocks until data becomes available.
443 * Returns: data from the queue
446 g_async_queue_pop (GAsyncQueue
*queue
)
450 g_return_val_if_fail (queue
, NULL
);
452 g_mutex_lock (&queue
->mutex
);
453 retval
= g_async_queue_pop_intern_unlocked (queue
, TRUE
, -1);
454 g_mutex_unlock (&queue
->mutex
);
460 * g_async_queue_pop_unlocked:
461 * @queue: a #GAsyncQueue
463 * Pops data from the @queue. If @queue is empty, this function
464 * blocks until data becomes available.
466 * This function must be called while holding the @queue's lock.
468 * Returns: data from the queue.
471 g_async_queue_pop_unlocked (GAsyncQueue
*queue
)
473 g_return_val_if_fail (queue
, NULL
);
475 return g_async_queue_pop_intern_unlocked (queue
, TRUE
, -1);
479 * g_async_queue_try_pop:
480 * @queue: a #GAsyncQueue
482 * Tries to pop data from the @queue. If no data is available,
485 * Returns: data from the queue or %NULL, when no data is
486 * available immediately.
489 g_async_queue_try_pop (GAsyncQueue
*queue
)
493 g_return_val_if_fail (queue
, NULL
);
495 g_mutex_lock (&queue
->mutex
);
496 retval
= g_async_queue_pop_intern_unlocked (queue
, FALSE
, -1);
497 g_mutex_unlock (&queue
->mutex
);
503 * g_async_queue_try_pop_unlocked:
504 * @queue: a #GAsyncQueue
506 * Tries to pop data from the @queue. If no data is available,
509 * This function must be called while holding the @queue's lock.
511 * Returns: data from the queue or %NULL, when no data is
512 * available immediately.
515 g_async_queue_try_pop_unlocked (GAsyncQueue
*queue
)
517 g_return_val_if_fail (queue
, NULL
);
519 return g_async_queue_pop_intern_unlocked (queue
, FALSE
, -1);
523 * g_async_queue_timeout_pop:
524 * @queue: a #GAsyncQueue
525 * @timeout: the number of microseconds to wait
527 * Pops data from the @queue. If the queue is empty, blocks for
528 * @timeout microseconds, or until data becomes available.
530 * If no data is received before the timeout, %NULL is returned.
532 * Returns: data from the queue or %NULL, when no data is
533 * received before the timeout.
536 g_async_queue_timeout_pop (GAsyncQueue
*queue
,
539 gint64 end_time
= g_get_monotonic_time () + timeout
;
542 g_mutex_lock (&queue
->mutex
);
543 retval
= g_async_queue_pop_intern_unlocked (queue
, TRUE
, end_time
);
544 g_mutex_unlock (&queue
->mutex
);
550 * g_async_queue_timeout_pop_unlocked:
551 * @queue: a #GAsyncQueue
552 * @timeout: the number of microseconds to wait
554 * Pops data from the @queue. If the queue is empty, blocks for
555 * @timeout microseconds, or until data becomes available.
557 * If no data is received before the timeout, %NULL is returned.
559 * This function must be called while holding the @queue's lock.
561 * Returns: data from the queue or %NULL, when no data is
562 * received before the timeout.
565 g_async_queue_timeout_pop_unlocked (GAsyncQueue
*queue
,
568 gint64 end_time
= g_get_monotonic_time () + timeout
;
570 return g_async_queue_pop_intern_unlocked (queue
, TRUE
, end_time
);
574 * g_async_queue_timed_pop:
575 * @queue: a #GAsyncQueue
576 * @end_time: a #GTimeVal, determining the final time
578 * Pops data from the @queue. If the queue is empty, blocks until
579 * @end_time or until data becomes available.
581 * If no data is received before @end_time, %NULL is returned.
583 * To easily calculate @end_time, a combination of g_get_current_time()
584 * and g_time_val_add() can be used.
586 * Returns: data from the queue or %NULL, when no data is
587 * received before @end_time.
589 * Deprecated: use g_async_queue_timeout_pop().
592 g_async_queue_timed_pop (GAsyncQueue
*queue
,
598 g_return_val_if_fail (queue
, NULL
);
600 if (end_time
!= NULL
)
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 ());
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
);
616 * g_async_queue_timed_pop_unlocked:
617 * @queue: a #GAsyncQueue
618 * @end_time: a #GTimeVal, determining the final time
620 * Pops data from the @queue. If the queue is empty, blocks until
621 * @end_time or until data becomes available.
623 * If no data is received before @end_time, %NULL is returned.
625 * To easily calculate @end_time, a combination of g_get_current_time()
626 * and g_time_val_add() can be used.
628 * This function must be called while holding the @queue's lock.
630 * Returns: data from the queue or %NULL, when no data is
631 * received before @end_time.
633 * Deprecated: use g_async_queue_timeout_pop_unlocked().
636 g_async_queue_timed_pop_unlocked (GAsyncQueue
*queue
,
641 g_return_val_if_fail (queue
, NULL
);
643 if (end_time
!= NULL
)
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 ());
651 return g_async_queue_pop_intern_unlocked (queue
, TRUE
, m_end_time
);
655 * g_async_queue_length:
656 * @queue: a #GAsyncQueue.
658 * Returns the length of the queue.
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.
667 * Returns: the length of the @queue
670 g_async_queue_length (GAsyncQueue
*queue
)
674 g_return_val_if_fail (queue
, 0);
676 g_mutex_lock (&queue
->mutex
);
677 retval
= queue
->queue
.length
- queue
->waiting_threads
;
678 g_mutex_unlock (&queue
->mutex
);
684 * g_async_queue_length_unlocked:
685 * @queue: a #GAsyncQueue
687 * Returns the length of the queue.
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.
696 * This function must be called while holding the @queue's lock.
698 * Returns: the length of the @queue.
701 g_async_queue_length_unlocked (GAsyncQueue
*queue
)
703 g_return_val_if_fail (queue
, 0);
705 return queue
->queue
.length
- queue
->waiting_threads
;
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
714 * Sorts @queue using @func.
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
722 * This function will lock @queue before it sorts the queue and unlock
723 * it when it is finished.
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" -->
731 * id1 = GPOINTER_TO_INT (element1);
732 * id2 = GPOINTER_TO_INT (element2);
734 * return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
740 g_async_queue_sort (GAsyncQueue
*queue
,
741 GCompareDataFunc func
,
744 g_return_if_fail (queue
!= NULL
);
745 g_return_if_fail (func
!= NULL
);
747 g_mutex_lock (&queue
->mutex
);
748 g_async_queue_sort_unlocked (queue
, func
, user_data
);
749 g_mutex_unlock (&queue
->mutex
);
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
758 * Sorts @queue using @func.
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
766 * This function must be called while holding the @queue's lock.
771 g_async_queue_sort_unlocked (GAsyncQueue
*queue
,
772 GCompareDataFunc func
,
777 g_return_if_fail (queue
!= NULL
);
778 g_return_if_fail (func
!= NULL
);
781 sd
.user_data
= user_data
;
783 g_queue_sort (&queue
->queue
,
784 (GCompareDataFunc
)g_async_queue_invert_compare
,
789 * g_async_queue_remove:
790 * @queue: a #GAsyncQueue
791 * @item: the data to remove from the @queue
793 * Remove an item from the queue.
795 * Returns: %TRUE if the item was removed
800 g_async_queue_remove (GAsyncQueue
*queue
,
805 g_return_val_if_fail (queue
!= NULL
, FALSE
);
806 g_return_val_if_fail (item
!= NULL
, FALSE
);
808 g_mutex_lock (&queue
->mutex
);
809 ret
= g_async_queue_remove_unlocked (queue
, item
);
810 g_mutex_unlock (&queue
->mutex
);
816 * g_async_queue_remove_unlocked:
817 * @queue: a #GAsyncQueue
818 * @item: the data to remove from the @queue
820 * Remove an item from the queue.
822 * This function must be called while holding the @queue's lock.
824 * Returns: %TRUE if the item was removed
829 g_async_queue_remove_unlocked (GAsyncQueue
*queue
,
832 g_return_val_if_fail (queue
!= NULL
, FALSE
);
833 g_return_val_if_fail (item
!= NULL
, FALSE
);
835 return g_queue_remove (&queue
->queue
, item
);
839 * g_async_queue_push_front:
840 * @queue: a #GAsyncQueue
841 * @item: data to push into the @queue
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.
851 g_async_queue_push_front (GAsyncQueue
*queue
,
854 g_return_if_fail (queue
!= NULL
);
855 g_return_if_fail (item
!= NULL
);
857 g_mutex_lock (&queue
->mutex
);
858 g_async_queue_push_front_unlocked (queue
, item
);
859 g_mutex_unlock (&queue
->mutex
);
863 * g_async_queue_push_front_unlocked:
864 * @queue: a #GAsyncQueue
865 * @item: data to push into the @queue
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.
872 * This function must be called while holding the @queue's lock.
877 g_async_queue_push_front_unlocked (GAsyncQueue
*queue
,
880 g_return_if_fail (queue
!= NULL
);
881 g_return_if_fail (item
!= NULL
);
883 g_queue_push_tail (&queue
->queue
, item
);
884 if (queue
->waiting_threads
> 0)
885 g_cond_signal (&queue
->cond
);
893 _g_async_queue_get_mutex (GAsyncQueue
*queue
)
895 g_return_val_if_fail (queue
, NULL
);
897 return &queue
->mutex
;