2 * Copyright (C) 2009 Edward Hervey <bilboed@bilboed.com>
3 * 2011 Wim Taymans <wim.taymans@gmail.com>
7 * This library is free software; you can redistribute it and/or
8 * modify it under the terms of the GNU Library 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 * Library General Public License for more details.
17 * You should have received a copy of the GNU Library 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.
25 #include "gstatomicqueue.h"
27 /* By default the queue uses 2 * sizeof(gpointer) * clp2 (max_items) of
28 * memory. clp2(x) is the next power of two >= than x.
30 * The queue can operate in low memory mode, in which it consumes almost
31 * half the memory at the expense of extra overhead in the readers. This
32 * is disabled by default because even without LOW_MEM mode, the memory
33 * consumption is still lower than a plain GList.
37 typedef struct _GstAQueueMem GstAQueueMem;
61 new_queue_mem (guint size, gint pos)
65 mem = g_new (GstAQueueMem, 1);
67 /* we keep the size as a mask for performance */
68 mem->size = clp2 (MAX (size, 16)) - 1;
69 mem->array = g_new0 (gpointer, mem->size);
79 free_queue_mem (GstAQueueMem * mem)
85 struct _GstAtomicQueue
90 GstAQueueMem *head_mem;
91 GstAQueueMem *tail_mem;
92 GstAQueueMem *free_list;
96 add_to_free_list (GstAtomicQueue * queue, GstAQueueMem * mem)
99 mem->free = g_atomic_pointer_get (&queue->free_list);
100 } while (!g_atomic_pointer_compare_and_exchange ((gpointer *) &
101 queue->free_list, mem->free, mem));
105 clear_free_list (GstAtomicQueue * queue)
107 GstAQueueMem *free_list;
109 /* take the free list and replace with NULL */
111 free_list = g_atomic_pointer_get (&queue->free_list);
112 if (free_list == NULL)
114 } while (!g_atomic_pointer_compare_and_exchange ((gpointer *) &
115 queue->free_list, free_list, NULL));
118 GstAQueueMem *next = free_list->free;
120 free_queue_mem (free_list);
127 gst_atomic_queue_new (guint initial_size)
129 GstAtomicQueue *queue;
131 queue = g_new (GstAtomicQueue, 1);
134 queue->num_readers = 0;
136 queue->head_mem = queue->tail_mem = new_queue_mem (initial_size, 0);
137 queue->free_list = NULL;
143 gst_atomic_queue_free (GstAtomicQueue * queue)
145 free_queue_mem (queue->head_mem);
146 if (queue->head_mem != queue->tail_mem)
147 free_queue_mem (queue->tail_mem);
148 clear_free_list (queue);
153 gst_atomic_queue_peek (GstAtomicQueue * queue)
155 GstAQueueMem *head_mem;
156 gint head, tail, size;
161 head_mem = g_atomic_pointer_get (&queue->head_mem);
163 head = g_atomic_int_get (&head_mem->head);
164 tail = g_atomic_int_get (&head_mem->tail);
165 size = head_mem->size;
167 /* when we are not empty, we can continue */
168 if (G_LIKELY (head != tail))
171 /* else array empty, try to take next */
172 next = g_atomic_pointer_get (&head_mem->next);
176 /* now we try to move the next array as the head memory. If we fail to do that,
177 * some other reader managed to do it first and we retry */
178 if (!g_atomic_pointer_compare_and_exchange ((gpointer *) &
179 queue->head_mem, head_mem, next))
182 /* when we managed to swing the head pointer the old head is now
183 * useless and we add it to the freelist. We can't free the memory yet
184 * because we first need to make sure no reader is accessing it anymore. */
185 add_to_free_list (queue, head_mem);
188 return head_mem->array[head & size];
192 gst_atomic_queue_pop (GstAtomicQueue * queue)
195 GstAQueueMem *head_mem;
196 gint head, tail, size;
199 g_atomic_int_inc (&queue->num_readers);
206 head_mem = g_atomic_pointer_get (&queue->head_mem);
208 head = g_atomic_int_get (&head_mem->head);
209 tail = g_atomic_int_get (&head_mem->tail);
210 size = head_mem->size;
212 /* when we are not empty, we can continue */
213 if (G_LIKELY (head != tail))
216 /* else array empty, try to take next */
217 next = g_atomic_pointer_get (&head_mem->next);
221 /* now we try to move the next array as the head memory. If we fail to do that,
222 * some other reader managed to do it first and we retry */
223 if (!g_atomic_pointer_compare_and_exchange ((gpointer *) &
224 queue->head_mem, head_mem, next))
227 /* when we managed to swing the head pointer the old head is now
228 * useless and we add it to the freelist. We can't free the memory yet
229 * because we first need to make sure no reader is accessing it anymore. */
230 add_to_free_list (queue, head_mem);
233 ret = head_mem->array[head & size];
234 } while (!g_atomic_int_compare_and_exchange (&head_mem->head, head,
238 /* decrement number of readers, when we reach 0 readers we can be sure that
239 * none is accessing the memory in the free list and we can try to clean up */
240 if (g_atomic_int_dec_and_test (&queue->num_readers))
241 clear_free_list (queue);
248 gst_atomic_queue_push (GstAtomicQueue * queue, gpointer data)
250 GstAQueueMem *tail_mem;
251 gint head, tail, size;
257 tail_mem = g_atomic_pointer_get (&queue->tail_mem);
258 head = g_atomic_int_get (&tail_mem->head);
259 tail = g_atomic_int_get (&tail_mem->tail);
260 size = tail_mem->size;
262 /* we're not full, continue */
263 if (tail - head < size)
266 /* else we need to grow the array, we store a mask so we have to add 1 */
267 mem = new_queue_mem ((size << 1) + 1, tail);
269 /* try to make our new array visible to other writers */
270 if (!g_atomic_pointer_compare_and_exchange ((gpointer *) &
271 queue->tail_mem, tail_mem, mem)) {
272 /* we tried to swap the new writer array but something changed. This is
273 * because some other writer beat us to it, we free our memory and try
275 free_queue_mem (mem);
278 /* make sure that readers can find our new array as well. The one who
279 * manages to swap the pointer is the only one who can set the next
280 * pointer to the new array */
281 g_atomic_pointer_set (&tail_mem->next, mem);
283 } while (!g_atomic_int_compare_and_exchange (&tail_mem->tail, tail,
286 tail_mem->array[tail & size] = data;
290 gst_atomic_queue_length (GstAtomicQueue * queue)
292 GstAQueueMem *head_mem, *tail_mem;
296 g_atomic_int_inc (&queue->num_readers);
299 head_mem = g_atomic_pointer_get (&queue->head_mem);
300 head = g_atomic_int_get (&head_mem->head);
302 tail_mem = g_atomic_pointer_get (&queue->tail_mem);
303 tail = g_atomic_int_get (&tail_mem->tail);
306 if (g_atomic_int_dec_and_test (&queue->num_readers))
307 clear_free_list (queue);