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 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 * Return value: 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 * Return value: 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 * Return value: 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 * Return value: 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 * Return value: 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 * Return value: 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 * Return value: 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 * Return value: 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 * Return value: 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 -
609 g_mutex_lock (&queue->mutex);
610 retval = g_async_queue_pop_intern_unlocked (queue, TRUE, m_end_time);
611 g_mutex_unlock (&queue->mutex);
617 * g_async_queue_timed_pop_unlocked:
618 * @queue: a #GAsyncQueue
619 * @end_time: a #GTimeVal, determining the final time
621 * Pops data from the @queue. If the queue is empty, blocks until
622 * @end_time or until data becomes available.
624 * If no data is received before @end_time, %NULL is returned.
626 * To easily calculate @end_time, a combination of g_get_current_time()
627 * and g_time_val_add() can be used.
629 * This function must be called while holding the @queue's lock.
631 * Return value: data from the queue or %NULL, when no data is
632 * received before @end_time.
634 * Deprecated: use g_async_queue_timeout_pop_unlocked().
637 g_async_queue_timed_pop_unlocked (GAsyncQueue *queue,
642 g_return_val_if_fail (queue, NULL);
644 if (end_time != NULL)
646 m_end_time = g_get_monotonic_time () +
647 (end_time->tv_sec * G_USEC_PER_SEC + end_time->tv_usec -
653 return g_async_queue_pop_intern_unlocked (queue, TRUE, m_end_time);
657 * g_async_queue_length:
658 * @queue: a #GAsyncQueue.
660 * Returns the length of the queue.
662 * Actually this function returns the number of data items in
663 * the queue minus the number of waiting threads, so a negative
664 * value means waiting threads, and a positive value means available
665 * entries in the @queue. A return value of 0 could mean n entries
666 * in the queue and n threads waiting. This can happen due to locking
667 * of the queue or due to scheduling.
669 * Return value: the length of the @queue
672 g_async_queue_length (GAsyncQueue *queue)
676 g_return_val_if_fail (queue, 0);
678 g_mutex_lock (&queue->mutex);
679 retval = queue->queue.length - queue->waiting_threads;
680 g_mutex_unlock (&queue->mutex);
686 * g_async_queue_length_unlocked:
687 * @queue: a #GAsyncQueue
689 * Returns the length of the queue.
691 * Actually this function returns the number of data items in
692 * the queue minus the number of waiting threads, so a negative
693 * value means waiting threads, and a positive value means available
694 * entries in the @queue. A return value of 0 could mean n entries
695 * in the queue and n threads waiting. This can happen due to locking
696 * of the queue or due to scheduling.
698 * This function must be called while holding the @queue's lock.
700 * Return value: the length of the @queue.
703 g_async_queue_length_unlocked (GAsyncQueue *queue)
705 g_return_val_if_fail (queue, 0);
707 return queue->queue.length - queue->waiting_threads;
711 * g_async_queue_sort:
712 * @queue: a #GAsyncQueue
713 * @func: the #GCompareDataFunc is used to sort @queue
714 * @user_data: user data passed to @func
716 * Sorts @queue using @func.
718 * The sort function @func is passed two elements of the @queue.
719 * It should return 0 if they are equal, a negative value if the
720 * first element should be higher in the @queue or a positive value
721 * if the first element should be lower in the @queue than the second
724 * This function will lock @queue before it sorts the queue and unlock
725 * it when it is finished.
727 * If you were sorting a list of priority numbers to make sure the
728 * lowest priority would be at the top of the queue, you could use:
729 * |[<!-- language="C" -->
733 * id1 = GPOINTER_TO_INT (element1);
734 * id2 = GPOINTER_TO_INT (element2);
736 * return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
742 g_async_queue_sort (GAsyncQueue *queue,
743 GCompareDataFunc func,
746 g_return_if_fail (queue != NULL);
747 g_return_if_fail (func != NULL);
749 g_mutex_lock (&queue->mutex);
750 g_async_queue_sort_unlocked (queue, func, user_data);
751 g_mutex_unlock (&queue->mutex);
755 * g_async_queue_sort_unlocked:
756 * @queue: a #GAsyncQueue
757 * @func: the #GCompareDataFunc is used to sort @queue
758 * @user_data: user data passed to @func
760 * Sorts @queue using @func.
762 * The sort function @func is passed two elements of the @queue.
763 * It should return 0 if they are equal, a negative value if the
764 * first element should be higher in the @queue or a positive value
765 * if the first element should be lower in the @queue than the second
768 * This function must be called while holding the @queue's lock.
773 g_async_queue_sort_unlocked (GAsyncQueue *queue,
774 GCompareDataFunc func,
779 g_return_if_fail (queue != NULL);
780 g_return_if_fail (func != NULL);
783 sd.user_data = user_data;
785 g_queue_sort (&queue->queue,
786 (GCompareDataFunc)g_async_queue_invert_compare,
795 _g_async_queue_get_mutex (GAsyncQueue *queue)
797 g_return_val_if_fail (queue, NULL);
799 return &queue->mutex;