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, write to the
19 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
20 * Boston, MA 02111-1307, USA.
29 #include "gasyncqueue.h"
30 #include "gasyncqueueprivate.h"
34 #include "gtestutils.h"
39 * SECTION:async_queues
40 * @title: Asynchronous Queues
41 * @short_description: asynchronous communication between threads
42 * @see_also: #GThreadPool
44 * Often you need to communicate between different threads. In general
45 * it's safer not to do this by shared memory, but by explicit message
46 * passing. These messages only make sense asynchronously for
47 * multi-threaded applications though, as a synchronous operation could
48 * as well be done in the same thread.
50 * Asynchronous queues are an exception from most other GLib data
51 * structures, as they can be used simultaneously from multiple threads
52 * without explicit locking and they bring their own builtin reference
53 * counting. This is because the nature of an asynchronous queue is that
54 * it will always be used by at least 2 concurrent threads.
56 * For using an asynchronous queue you first have to create one with
57 * g_async_queue_new(). #GAsyncQueue structs are reference counted,
58 * use g_async_queue_ref() and g_async_queue_unref() to manage your
61 * A thread which wants to send a message to that queue simply calls
62 * g_async_queue_push() to push the message to the queue.
64 * A thread which is expecting messages from an asynchronous queue
65 * simply calls g_async_queue_pop() for that queue. If no message is
66 * available in the queue at that point, the thread is now put to sleep
67 * until a message arrives. The message will be removed from the queue
68 * and returned. The functions g_async_queue_try_pop() and
69 * g_async_queue_timed_pop() can be used to only check for the presence
70 * of messages or to only wait a certain time for messages respectively.
72 * For almost every function there exist two variants, one that locks
73 * the queue and one that doesn't. That way you can hold the queue lock
74 * (acquire it with g_async_queue_lock() and release it with
75 * g_async_queue_unlock()) over multiple queue accessing instructions.
76 * This can be necessary to ensure the integrity of the queue, but should
77 * only be used when really necessary, as it can make your life harder
78 * if used unwisely. Normally you should only use the locking function
79 * variants (those without the _unlocked suffix).
81 * In many cases, it may be more convenient to use #GThreadPool when
82 * you need to distribute work to a set of worker threads instead of
83 * using #GAsyncQueue manually. #GThreadPool uses a GAsyncQueue
90 * The GAsyncQueue struct is an opaque data structure which represents
91 * an asynchronous queue. It should only be accessed through the
92 * <function>g_async_queue_*</function> functions.
99 GDestroyNotify item_free_func;
100 guint waiting_threads;
106 GCompareDataFunc func;
113 * Creates a new asynchronous queue.
115 * Return value: a new #GAsyncQueue. Free with g_async_queue_unref()
118 g_async_queue_new (void)
120 return g_async_queue_new_full (NULL);
124 * g_async_queue_new_full:
125 * @item_free_func: function to free queue elements
127 * Creates a new asynchronous queue and sets up a destroy notify
128 * function that is used to free any remaining queue items when
129 * the queue is destroyed after the final unref.
131 * Return value: a new #GAsyncQueue. Free with g_async_queue_unref()
136 g_async_queue_new_full (GDestroyNotify item_free_func)
140 queue = g_new (GAsyncQueue, 1);
141 g_mutex_init (&queue->mutex);
143 g_queue_init (&queue->queue);
144 queue->waiting_threads = 0;
145 queue->ref_count = 1;
146 queue->item_free_func = NULL;
153 * @queue: a #GAsyncQueue
155 * Increases the reference count of the asynchronous @queue by 1.
156 * You do not need to hold the lock to call this function.
158 * Returns: the @queue that was passed in (since 2.6)
161 g_async_queue_ref (GAsyncQueue *queue)
163 g_return_val_if_fail (queue, NULL);
165 g_atomic_int_inc (&queue->ref_count);
171 * g_async_queue_ref_unlocked:
172 * @queue: a #GAsyncQueue
174 * Increases the reference count of the asynchronous @queue by 1.
176 * @Deprecated: Since 2.8, reference counting is done atomically
177 * so g_async_queue_ref() can be used regardless of the @queue's
181 g_async_queue_ref_unlocked (GAsyncQueue *queue)
183 g_return_if_fail (queue);
185 g_atomic_int_inc (&queue->ref_count);
189 * g_async_queue_unref_and_unlock:
190 * @queue: a #GAsyncQueue
192 * Decreases the reference count of the asynchronous @queue by 1
193 * and releases the lock. This function must be called while holding
194 * the @queue's lock. If the reference count went to 0, the @queue
195 * will be destroyed and the memory allocated will be freed.
197 * @Deprecated: Since 2.8, reference counting is done atomically
198 * so g_async_queue_unref() can be used regardless of the @queue's
202 g_async_queue_unref_and_unlock (GAsyncQueue *queue)
204 g_return_if_fail (queue);
206 g_mutex_unlock (&queue->mutex);
207 g_async_queue_unref (queue);
211 * g_async_queue_unref:
212 * @queue: a #GAsyncQueue.
214 * Decreases the reference count of the asynchronous @queue by 1.
216 * If the reference count went to 0, the @queue will be destroyed
217 * and the memory allocated will be freed. So you are not allowed
218 * to use the @queue afterwards, as it might have disappeared.
219 * You do not need to hold the lock to call this function.
222 g_async_queue_unref (GAsyncQueue *queue)
224 g_return_if_fail (queue);
226 if (g_atomic_int_dec_and_test (&queue->ref_count))
228 g_return_if_fail (queue->waiting_threads == 0);
229 g_mutex_clear (&queue->mutex);
231 g_cond_free (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 * <function>g_async_queue_*_unlocked()</function> functions
251 * on @queue. Otherwise, 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))
419 queue->cond = g_cond_new ();
423 queue->waiting_threads++;
424 while (!g_queue_peek_tail_link (&queue->queue))
425 g_cond_wait (queue->cond, &queue->mutex);
426 queue->waiting_threads--;
430 queue->waiting_threads++;
431 while (!g_queue_peek_tail_link (&queue->queue))
432 if (!g_cond_timed_wait (queue->cond, &queue->mutex, end_time))
434 queue->waiting_threads--;
435 if (!g_queue_peek_tail_link (&queue->queue))
440 retval = g_queue_pop_tail (&queue->queue);
449 * @queue: a #GAsyncQueue
451 * Pops data from the @queue. If @queue is empty, this function
452 * blocks until data becomes available.
454 * Return value: data from the queue
457 g_async_queue_pop (GAsyncQueue *queue)
461 g_return_val_if_fail (queue, NULL);
463 g_mutex_lock (&queue->mutex);
464 retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
465 g_mutex_unlock (&queue->mutex);
471 * g_async_queue_pop_unlocked:
472 * @queue: a #GAsyncQueue
474 * Pops data from the @queue. If @queue is empty, this function
475 * blocks until data becomes available.
477 * This function must be called while holding the @queue's lock.
479 * Return value: data from the queue.
482 g_async_queue_pop_unlocked (GAsyncQueue *queue)
484 g_return_val_if_fail (queue, NULL);
486 return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
490 * g_async_queue_try_pop:
491 * @queue: a #GAsyncQueue
493 * Tries to pop data from the @queue. If no data is available,
496 * Return value: data from the queue or %NULL, when no data is
497 * available immediately.
500 g_async_queue_try_pop (GAsyncQueue *queue)
504 g_return_val_if_fail (queue, NULL);
506 g_mutex_lock (&queue->mutex);
507 retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
508 g_mutex_unlock (&queue->mutex);
514 * g_async_queue_try_pop_unlocked:
515 * @queue: a #GAsyncQueue
517 * Tries to pop data from the @queue. If no data is available,
520 * This function must be called while holding the @queue's lock.
522 * Return value: data from the queue or %NULL, when no data is
523 * available immediately.
526 g_async_queue_try_pop_unlocked (GAsyncQueue *queue)
528 g_return_val_if_fail (queue, NULL);
530 return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
534 * g_async_queue_timed_pop:
535 * @queue: a #GAsyncQueue
536 * @end_time: a #GTimeVal, determining the final time
538 * Pops data from the @queue. If the queue is empty, blocks until
539 * @end_time or until data becomes available.
541 * If no data is received before @end_time, %NULL is returned.
543 * To easily calculate @end_time, a combination of g_get_current_time()
544 * and g_time_val_add() can be used.
546 * Return value: data from the queue or %NULL, when no data is
547 * received before @end_time.
550 g_async_queue_timed_pop (GAsyncQueue *queue,
555 g_return_val_if_fail (queue, NULL);
557 g_mutex_lock (&queue->mutex);
558 retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
559 g_mutex_unlock (&queue->mutex);
565 * g_async_queue_timed_pop_unlocked:
566 * @queue: a #GAsyncQueue
567 * @end_time: a #GTimeVal, determining the final time
569 * Pops data from the @queue. If the queue is empty, blocks until
570 * @end_time or until data becomes available.
572 * If no data is received before @end_time, %NULL is returned.
574 * To easily calculate @end_time, a combination of g_get_current_time()
575 * and g_time_val_add() can be used.
577 * This function must be called while holding the @queue's lock.
579 * Return value: data from the queue or %NULL, when no data is
580 * received before @end_time.
583 g_async_queue_timed_pop_unlocked (GAsyncQueue *queue,
586 g_return_val_if_fail (queue, NULL);
588 return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
592 * g_async_queue_length:
593 * @queue: a #GAsyncQueue.
595 * Returns the length of the queue.
597 * Actually this function returns the number of data items in
598 * the queue minus the number of waiting threads, so a negative
599 * value means waiting threads, and a positive value means available
600 * entries in the @queue. A return value of 0 could mean n entries
601 * in the queue and n threads waiting. This can happen due to locking
602 * of the queue or due to scheduling.
604 * Return value: the length of the @queue
607 g_async_queue_length (GAsyncQueue *queue)
611 g_return_val_if_fail (queue, 0);
613 g_mutex_lock (&queue->mutex);
614 retval = queue->queue.length - queue->waiting_threads;
615 g_mutex_unlock (&queue->mutex);
621 * g_async_queue_length_unlocked:
622 * @queue: a #GAsyncQueue
624 * Returns the length of the queue.
626 * Actually this function returns the number of data items in
627 * the queue minus the number of waiting threads, so a negative
628 * value means waiting threads, and a positive value means available
629 * entries in the @queue. A return value of 0 could mean n entries
630 * in the queue and n threads waiting. This can happen due to locking
631 * of the queue or due to scheduling.
633 * This function must be called while holding the @queue's lock.
635 * Return value: the length of the @queue.
638 g_async_queue_length_unlocked (GAsyncQueue *queue)
640 g_return_val_if_fail (queue, 0);
642 return queue->queue.length - queue->waiting_threads;
646 * g_async_queue_sort:
647 * @queue: a #GAsyncQueue
648 * @func: the #GCompareDataFunc is used to sort @queue
649 * @user_data: user data passed to @func
651 * Sorts @queue using @func.
653 * The sort function @func is passed two elements of the @queue.
654 * It should return 0 if they are equal, a negative value if the
655 * first element should be higher in the @queue or a positive value
656 * if the first element should be lower in the @queue than the second
659 * This function will lock @queue before it sorts the queue and unlock
660 * it when it is finished.
662 * If you were sorting a list of priority numbers to make sure the
663 * lowest priority would be at the top of the queue, you could use:
668 * id1 = GPOINTER_TO_INT (element1);
669 * id2 = GPOINTER_TO_INT (element2);
671 * return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
677 g_async_queue_sort (GAsyncQueue *queue,
678 GCompareDataFunc func,
681 g_return_if_fail (queue != NULL);
682 g_return_if_fail (func != NULL);
684 g_mutex_lock (&queue->mutex);
685 g_async_queue_sort_unlocked (queue, func, user_data);
686 g_mutex_unlock (&queue->mutex);
690 * g_async_queue_sort_unlocked:
691 * @queue: a #GAsyncQueue
692 * @func: the #GCompareDataFunc is used to sort @queue
693 * @user_data: user data passed to @func
695 * Sorts @queue using @func.
697 * The sort function @func is passed two elements of the @queue.
698 * It should return 0 if they are equal, a negative value if the
699 * first element should be higher in the @queue or a positive value
700 * if the first element should be lower in the @queue than the second
703 * This function must be called while holding the @queue's lock.
708 g_async_queue_sort_unlocked (GAsyncQueue *queue,
709 GCompareDataFunc func,
714 g_return_if_fail (queue != NULL);
715 g_return_if_fail (func != NULL);
718 sd.user_data = user_data;
720 g_queue_sort (&queue->queue,
721 (GCompareDataFunc)g_async_queue_invert_compare,
730 _g_async_queue_get_mutex (GAsyncQueue *queue)
732 g_return_val_if_fail (queue, NULL);
734 return &queue->mutex;