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"
33 #include "gtestutils.h"
38 * SECTION: async_queues
39 * @title: Asynchronous Queues
40 * @short_description: asynchronous communication between threads
42 * Often you need to communicate between different threads. In general
43 * it's safer not to do this by shared memory, but by explicit message
44 * passing. These messages only make sense asynchronously for
45 * multi-threaded applications though, as a synchronous operation could
46 * as well be done in the same thread.
48 * Asynchronous queues are an exception from most other GLib data
49 * structures, as they can be used simultaneously from multiple threads
50 * without explicit locking and they bring their own builtin reference
51 * counting. This is because the nature of an asynchronous queue is that
52 * it will always be used by at least 2 concurrent threads.
54 * For using an asynchronous queue you first have to create one with
55 * g_async_queue_new(). A newly-created queue will get the reference
56 * count 1. Whenever another thread is creating a new reference of (that
57 * is, pointer to) the queue, it has to increase the reference count
58 * (using g_async_queue_ref()). Also, before removing this reference,
59 * the reference count has to be decreased (using g_async_queue_unref()).
60 * After that the queue might no longer exist so you must not access
61 * it after that point.
63 * A thread, which wants to send a message to that queue simply calls
64 * g_async_queue_push() to push the message to the queue.
66 * A thread, which is expecting messages from an asynchronous queue
67 * simply calls g_async_queue_pop() for that queue. If no message is
68 * available in the queue at that point, the thread is now put to sleep
69 * until a message arrives. The message will be removed from the queue
70 * and returned. The functions g_async_queue_try_pop() and
71 * g_async_queue_timed_pop() can be used to only check for the presence
72 * of messages or to only wait a certain time for messages respectively.
74 * For almost every function there exist two variants, one that locks
75 * the queue and one that doesn't. That way you can hold the queue lock
76 * (acquire it with g_async_queue_lock() and release it with
77 * g_async_queue_unlock()) over multiple queue accessing instructions.
78 * This can be necessary to ensure the integrity of the queue, but should
79 * only be used when really necessary, as it can make your life harder
80 * if used unwisely. Normally you should only use the locking function
81 * variants (those without the suffix _unlocked)
87 * The GAsyncQueue struct is an opaque data structure, which represents
88 * an asynchronous queue. It should only be accessed through the
89 * <function>g_async_queue_*</function> functions.
96 GDestroyNotify item_free_func;
97 guint waiting_threads;
102 GCompareDataFunc func;
109 * Creates a new asynchronous queue with the initial reference count of 1.
111 * Return value: the new #GAsyncQueue.
114 g_async_queue_new (void)
116 GAsyncQueue* retval = g_new (GAsyncQueue, 1);
117 retval->mutex = g_mutex_new ();
119 g_queue_init (&retval->queue);
120 retval->waiting_threads = 0;
121 retval->ref_count = 1;
122 retval->item_free_func = NULL;
127 * g_async_queue_new_full:
128 * @item_free_func: function to free queue elements
130 * Creates a new asynchronous queue with an initial reference count of 1 and
131 * sets up a destroy notify function that is used to free any remaining
132 * queue items when the queue is destroyed after the final unref.
134 * Return value: the new #GAsyncQueue.
139 g_async_queue_new_full (GDestroyNotify item_free_func)
141 GAsyncQueue *async_queue = g_async_queue_new ();
142 async_queue->item_free_func = item_free_func;
148 * @queue: a #GAsyncQueue.
150 * Increases the reference count of the asynchronous @queue by 1. You
151 * do not need to hold the lock to call this function.
153 * Returns: the @queue that was passed in (since 2.6)
156 g_async_queue_ref (GAsyncQueue *queue)
158 g_return_val_if_fail (queue, NULL);
159 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
161 g_atomic_int_inc (&queue->ref_count);
167 * g_async_queue_ref_unlocked:
168 * @queue: a #GAsyncQueue.
170 * Increases the reference count of the asynchronous @queue by 1.
172 * @Deprecated: Since 2.8, reference counting is done atomically
173 * so g_async_queue_ref() can be used regardless of the @queue's
177 g_async_queue_ref_unlocked (GAsyncQueue *queue)
179 g_return_if_fail (queue);
180 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
182 g_atomic_int_inc (&queue->ref_count);
186 * g_async_queue_unref_and_unlock:
187 * @queue: a #GAsyncQueue.
189 * Decreases the reference count of the asynchronous @queue by 1 and
190 * releases the lock. This function must be called while holding the
191 * @queue's lock. If the reference count went to 0, the @queue will be
192 * destroyed and the memory allocated will be freed.
194 * @Deprecated: Since 2.8, reference counting is done atomically
195 * so g_async_queue_unref() can be used regardless of the @queue's
199 g_async_queue_unref_and_unlock (GAsyncQueue *queue)
201 g_return_if_fail (queue);
202 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
204 g_mutex_unlock (queue->mutex);
205 g_async_queue_unref (queue);
209 * g_async_queue_unref:
210 * @queue: a #GAsyncQueue.
212 * Decreases the reference count of the asynchronous @queue by 1. If
213 * the reference count went to 0, the @queue will be destroyed and the
214 * memory allocated will be freed. So you are not allowed to use the
215 * @queue afterwards, as it might have disappeared. You do not need to
216 * hold the lock to call this function.
219 g_async_queue_unref (GAsyncQueue *queue)
221 g_return_if_fail (queue);
222 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
224 if (g_atomic_int_dec_and_test (&queue->ref_count))
226 g_return_if_fail (queue->waiting_threads == 0);
227 g_mutex_free (queue->mutex);
229 g_cond_free (queue->cond);
230 if (queue->item_free_func)
231 g_queue_foreach (&queue->queue, (GFunc) queue->item_free_func, NULL);
232 g_queue_clear (&queue->queue);
238 * g_async_queue_lock:
239 * @queue: a #GAsyncQueue.
241 * Acquires the @queue's lock. After that you can only call the
242 * <function>g_async_queue_*_unlocked()</function> function variants on that
243 * @queue. Otherwise it will deadlock.
246 g_async_queue_lock (GAsyncQueue *queue)
248 g_return_if_fail (queue);
249 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
251 g_mutex_lock (queue->mutex);
255 * g_async_queue_unlock:
256 * @queue: a #GAsyncQueue.
258 * Releases the queue's lock.
261 g_async_queue_unlock (GAsyncQueue *queue)
263 g_return_if_fail (queue);
264 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
266 g_mutex_unlock (queue->mutex);
270 * g_async_queue_push:
271 * @queue: a #GAsyncQueue.
272 * @data: @data to push into the @queue.
274 * Pushes the @data into the @queue. @data must not be %NULL.
277 g_async_queue_push (GAsyncQueue* queue, gpointer data)
279 g_return_if_fail (queue);
280 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
281 g_return_if_fail (data);
283 g_mutex_lock (queue->mutex);
284 g_async_queue_push_unlocked (queue, data);
285 g_mutex_unlock (queue->mutex);
289 * g_async_queue_push_unlocked:
290 * @queue: a #GAsyncQueue.
291 * @data: @data to push into the @queue.
293 * Pushes the @data into the @queue. @data must not be %NULL. This
294 * function must be called while holding the @queue's lock.
297 g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)
299 g_return_if_fail (queue);
300 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
301 g_return_if_fail (data);
303 g_queue_push_head (&queue->queue, data);
304 if (queue->waiting_threads > 0)
305 g_cond_signal (queue->cond);
309 * g_async_queue_push_sorted:
310 * @queue: a #GAsyncQueue
311 * @data: the @data to push into the @queue
312 * @func: the #GCompareDataFunc is used to sort @queue. This function
313 * is passed two elements of the @queue. The function should return
314 * 0 if they are equal, a negative value if the first element
315 * should be higher in the @queue or a positive value if the first
316 * element should be lower in the @queue than the second element.
317 * @user_data: user data passed to @func.
319 * Inserts @data into @queue using @func to determine the new
322 * This function requires that the @queue is sorted before pushing on
325 * This function will lock @queue before it sorts the queue and unlock
326 * it when it is finished.
328 * For an example of @func see g_async_queue_sort().
333 g_async_queue_push_sorted (GAsyncQueue *queue,
335 GCompareDataFunc func,
338 g_return_if_fail (queue != NULL);
340 g_mutex_lock (queue->mutex);
341 g_async_queue_push_sorted_unlocked (queue, data, func, user_data);
342 g_mutex_unlock (queue->mutex);
346 g_async_queue_invert_compare (gpointer v1,
350 return -sd->func (v1, v2, sd->user_data);
354 * g_async_queue_push_sorted_unlocked:
355 * @queue: a #GAsyncQueue
356 * @data: the @data to push into the @queue
357 * @func: the #GCompareDataFunc is used to sort @queue. This function
358 * is passed two elements of the @queue. The function should return
359 * 0 if they are equal, a negative value if the first element
360 * should be higher in the @queue or a positive value if the first
361 * element should be lower in the @queue than the second element.
362 * @user_data: user data passed to @func.
364 * Inserts @data into @queue using @func to determine the new
367 * This function requires that the @queue is sorted before pushing on
370 * This function is called while holding the @queue's lock.
372 * For an example of @func see g_async_queue_sort().
377 g_async_queue_push_sorted_unlocked (GAsyncQueue *queue,
379 GCompareDataFunc func,
384 g_return_if_fail (queue != NULL);
387 sd.user_data = user_data;
389 g_queue_insert_sorted (&queue->queue,
391 (GCompareDataFunc)g_async_queue_invert_compare,
393 if (queue->waiting_threads > 0)
394 g_cond_signal (queue->cond);
398 g_async_queue_pop_intern_unlocked (GAsyncQueue *queue,
404 if (!g_queue_peek_tail_link (&queue->queue))
410 queue->cond = g_cond_new ();
414 queue->waiting_threads++;
415 while (!g_queue_peek_tail_link (&queue->queue))
416 g_cond_wait (queue->cond, queue->mutex);
417 queue->waiting_threads--;
421 queue->waiting_threads++;
422 while (!g_queue_peek_tail_link (&queue->queue))
423 if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
425 queue->waiting_threads--;
426 if (!g_queue_peek_tail_link (&queue->queue))
431 retval = g_queue_pop_tail (&queue->queue);
440 * @queue: a #GAsyncQueue.
442 * Pops data from the @queue. This function blocks until data become
445 * Return value: data from the queue.
448 g_async_queue_pop (GAsyncQueue* queue)
452 g_return_val_if_fail (queue, NULL);
453 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
455 g_mutex_lock (queue->mutex);
456 retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
457 g_mutex_unlock (queue->mutex);
463 * g_async_queue_pop_unlocked:
464 * @queue: a #GAsyncQueue.
466 * Pops data from the @queue. This function blocks until data become
467 * available. This function must be called while holding the @queue's
470 * Return value: data from the queue.
473 g_async_queue_pop_unlocked (GAsyncQueue* queue)
475 g_return_val_if_fail (queue, NULL);
476 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
478 return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
482 * g_async_queue_try_pop:
483 * @queue: a #GAsyncQueue.
485 * Tries to pop data from the @queue. If no data is available, %NULL is
488 * Return value: data from the queue or %NULL, when no data is
489 * available immediately.
492 g_async_queue_try_pop (GAsyncQueue* queue)
496 g_return_val_if_fail (queue, NULL);
497 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
499 g_mutex_lock (queue->mutex);
500 retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
501 g_mutex_unlock (queue->mutex);
507 * g_async_queue_try_pop_unlocked:
508 * @queue: a #GAsyncQueue.
510 * Tries to pop data from the @queue. If no data is available, %NULL is
511 * returned. This function must be called while holding the @queue's
514 * Return value: data from the queue or %NULL, when no data is
515 * available immediately.
518 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
520 g_return_val_if_fail (queue, NULL);
521 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
523 return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
527 * g_async_queue_timed_pop:
528 * @queue: a #GAsyncQueue.
529 * @end_time: a #GTimeVal, determining the final time.
531 * Pops data from the @queue. If no data is received before @end_time,
534 * To easily calculate @end_time a combination of g_get_current_time()
535 * and g_time_val_add() can be used.
537 * Return value: data from the queue or %NULL, when no data is
538 * received before @end_time.
541 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
545 g_return_val_if_fail (queue, NULL);
546 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
548 g_mutex_lock (queue->mutex);
549 retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
550 g_mutex_unlock (queue->mutex);
556 * g_async_queue_timed_pop_unlocked:
557 * @queue: a #GAsyncQueue.
558 * @end_time: a #GTimeVal, determining the final time.
560 * Pops data from the @queue. If no data is received before @end_time,
561 * %NULL is returned. This function must be called while holding the
564 * To easily calculate @end_time a combination of g_get_current_time()
565 * and g_time_val_add() can be used.
567 * Return value: data from the queue or %NULL, when no data is
568 * received before @end_time.
571 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
573 g_return_val_if_fail (queue, NULL);
574 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
576 return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
580 * g_async_queue_length:
581 * @queue: a #GAsyncQueue.
583 * Returns the length of the queue, negative values mean waiting
584 * threads, positive values mean available entries in the
585 * @queue. Actually this function returns the number of data items in
586 * the queue minus the number of waiting threads. Thus a return value
587 * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
588 * That can happen due to locking of the queue or due to
591 * Return value: the length of the @queue.
594 g_async_queue_length (GAsyncQueue* queue)
598 g_return_val_if_fail (queue, 0);
599 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
601 g_mutex_lock (queue->mutex);
602 retval = queue->queue.length - queue->waiting_threads;
603 g_mutex_unlock (queue->mutex);
609 * g_async_queue_length_unlocked:
610 * @queue: a #GAsyncQueue.
612 * Returns the length of the queue, negative values mean waiting
613 * threads, positive values mean available entries in the
614 * @queue. Actually this function returns the number of data items in
615 * the queue minus the number of waiting threads. Thus a return value
616 * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
617 * That can happen due to locking of the queue or due to
618 * scheduling. This function must be called while holding the @queue's
621 * Return value: the length of the @queue.
624 g_async_queue_length_unlocked (GAsyncQueue* queue)
626 g_return_val_if_fail (queue, 0);
627 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
629 return queue->queue.length - queue->waiting_threads;
633 * g_async_queue_sort:
634 * @queue: a #GAsyncQueue
635 * @func: the #GCompareDataFunc is used to sort @queue. This
636 * function is passed two elements of the @queue. The function
637 * should return 0 if they are equal, a negative value if the
638 * first element should be higher in the @queue or a positive
639 * value if the first element should be lower in the @queue than
640 * the second element.
641 * @user_data: user data passed to @func
643 * Sorts @queue using @func.
645 * This function will lock @queue before it sorts the queue and unlock
646 * it when it is finished.
648 * If you were sorting a list of priority numbers to make sure the
649 * lowest priority would be at the top of the queue, you could use:
654 * id1 = GPOINTER_TO_INT (element1);
655 * id2 = GPOINTER_TO_INT (element2);
657 * return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
663 g_async_queue_sort (GAsyncQueue *queue,
664 GCompareDataFunc func,
667 g_return_if_fail (queue != NULL);
668 g_return_if_fail (func != NULL);
670 g_mutex_lock (queue->mutex);
671 g_async_queue_sort_unlocked (queue, func, user_data);
672 g_mutex_unlock (queue->mutex);
676 * g_async_queue_sort_unlocked:
677 * @queue: a #GAsyncQueue
678 * @func: the #GCompareDataFunc is used to sort @queue. This
679 * function is passed two elements of the @queue. The function
680 * should return 0 if they are equal, a negative value if the
681 * first element should be higher in the @queue or a positive
682 * value if the first element should be lower in the @queue than
683 * the second element.
684 * @user_data: user data passed to @func
686 * Sorts @queue using @func.
688 * This function is called while holding the @queue's lock.
693 g_async_queue_sort_unlocked (GAsyncQueue *queue,
694 GCompareDataFunc func,
699 g_return_if_fail (queue != NULL);
700 g_return_if_fail (func != NULL);
703 sd.user_data = user_data;
705 g_queue_sort (&queue->queue,
706 (GCompareDataFunc)g_async_queue_invert_compare,
715 _g_async_queue_get_mutex (GAsyncQueue* queue)
717 g_return_val_if_fail (queue, NULL);
718 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);