X-Git-Url: http://review.tizen.org/git/?a=blobdiff_plain;f=gst%2Fgstatomicqueue.c;h=a80089c649a53507bf2a6e04f84e9b265556cfdc;hb=5688773b91c3ea1094996a573f119230727a0978;hp=2e8b6b8c3a29d684454867dacabb0c1806430094;hpb=cc5edeefcd7772ac645fa0b8b91cfe7be201d78a;p=platform%2Fupstream%2Fgstreamer.git diff --git a/gst/gstatomicqueue.c b/gst/gstatomicqueue.c index 2e8b6b8..a80089c 100644 --- a/gst/gstatomicqueue.c +++ b/gst/gstatomicqueue.c @@ -16,8 +16,8 @@ * * You should have received a copy of the GNU Library General Public * License along with this library; if not, write to the - * Free Software Foundation, Inc., 59 Temple Place - Suite 330, - * Boston, MA 02111-1307, USA. + * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, + * Boston, MA 02110-1301, USA. */ #include "gst_private.h" @@ -26,6 +26,7 @@ #include #include "gstatomicqueue.h" +#include "glib-compat-private.h" /** * SECTION:gstatomicqueue @@ -34,10 +35,12 @@ * * The #GstAtomicQueue object implements a queue that can be used from multiple * threads without performing any blocking operations. - * - * Since: 0.10.33 */ +G_DEFINE_BOXED_TYPE (GstAtomicQueue, gst_atomic_queue, + (GBoxedCopyFunc) gst_atomic_queue_ref, + (GBoxedFreeFunc) gst_atomic_queue_unref); + /* By default the queue uses 2 * sizeof(gpointer) * clp2 (max_items) of * memory. clp2(x) is the next power of two >= than x. * @@ -55,7 +58,8 @@ struct _GstAQueueMem gint size; gpointer *array; volatile gint head; - volatile gint tail; + volatile gint tail_write; + volatile gint tail_read; GstAQueueMem *next; GstAQueueMem *free; }; @@ -82,7 +86,8 @@ new_queue_mem (guint size, gint pos) mem->size = clp2 (MAX (size, 16)) - 1; mem->array = g_new0 (gpointer, mem->size + 1); mem->head = pos; - mem->tail = pos; + mem->tail_write = pos; + mem->tail_read = pos; mem->next = NULL; mem->free = NULL; @@ -112,8 +117,8 @@ add_to_free_list (GstAtomicQueue * queue, GstAQueueMem * mem) { do { mem->free = g_atomic_pointer_get (&queue->free_list); - } while (!g_atomic_pointer_compare_and_exchange ((gpointer *) & - queue->free_list, mem->free, mem)); + } while (!g_atomic_pointer_compare_and_exchange (&queue->free_list, + mem->free, mem)); } static void @@ -126,8 +131,8 @@ clear_free_list (GstAtomicQueue * queue) free_list = g_atomic_pointer_get (&queue->free_list); if (free_list == NULL) return; - } while (!g_atomic_pointer_compare_and_exchange ((gpointer *) & - queue->free_list, free_list, NULL)); + } while (!g_atomic_pointer_compare_and_exchange (&queue->free_list, free_list, + NULL)); while (free_list) { GstAQueueMem *next = free_list->free; @@ -146,8 +151,6 @@ clear_free_list (GstAtomicQueue * queue) * nearest power of 2 and used as the initial size of the queue. * * Returns: a new #GstAtomicQueue - * - * Since: 0.10.33 */ GstAtomicQueue * gst_atomic_queue_new (guint initial_size) @@ -211,7 +214,8 @@ gst_atomic_queue_unref (GstAtomicQueue * queue) * * Peek the head element of the queue without removing it from the queue. * - * Returns: the head element of @queue or NULL when the queue is empty. + * Returns: (transfer none) (nullable): the head element of @queue or + * %NULL when the queue is empty. */ gpointer gst_atomic_queue_peek (GstAtomicQueue * queue) @@ -227,7 +231,7 @@ gst_atomic_queue_peek (GstAtomicQueue * queue) head_mem = g_atomic_pointer_get (&queue->head_mem); head = g_atomic_int_get (&head_mem->head); - tail = g_atomic_int_get (&head_mem->tail); + tail = g_atomic_int_get (&head_mem->tail_read); size = head_mem->size; /* when we are not empty, we can continue */ @@ -241,8 +245,8 @@ gst_atomic_queue_peek (GstAtomicQueue * queue) /* now we try to move the next array as the head memory. If we fail to do that, * some other reader managed to do it first and we retry */ - if (!g_atomic_pointer_compare_and_exchange ((gpointer *) & - queue->head_mem, head_mem, next)) + if (!g_atomic_pointer_compare_and_exchange (&queue->head_mem, head_mem, + next)) continue; /* when we managed to swing the head pointer the old head is now @@ -260,7 +264,8 @@ gst_atomic_queue_peek (GstAtomicQueue * queue) * * Get the head element of the queue. * - * Returns: the head element of @queue or NULL when the queue is empty. + * Returns: (transfer full): the head element of @queue or %NULL when + * the queue is empty. */ gpointer gst_atomic_queue_pop (GstAtomicQueue * queue) @@ -282,12 +287,13 @@ gst_atomic_queue_pop (GstAtomicQueue * queue) head_mem = g_atomic_pointer_get (&queue->head_mem); head = g_atomic_int_get (&head_mem->head); - tail = g_atomic_int_get (&head_mem->tail); + tail = g_atomic_int_get (&head_mem->tail_read); size = head_mem->size; /* when we are not empty, we can continue */ - if (G_LIKELY (head != tail)) - break; + if G_LIKELY + (head != tail) + break; /* else array empty, try to take next */ next = g_atomic_pointer_get (&head_mem->next); @@ -296,9 +302,10 @@ gst_atomic_queue_pop (GstAtomicQueue * queue) /* now we try to move the next array as the head memory. If we fail to do that, * some other reader managed to do it first and we retry */ - if (!g_atomic_pointer_compare_and_exchange ((gpointer *) & - queue->head_mem, head_mem, next)) - continue; + if G_UNLIKELY + (!g_atomic_pointer_compare_and_exchange (&queue->head_mem, head_mem, + next)) + continue; /* when we managed to swing the head pointer the old head is now * useless and we add it to the freelist. We can't free the memory yet @@ -307,8 +314,8 @@ gst_atomic_queue_pop (GstAtomicQueue * queue) } ret = head_mem->array[head & size]; - } while (!g_atomic_int_compare_and_exchange (&head_mem->head, head, - head + 1)); + } while G_UNLIKELY + (!g_atomic_int_compare_and_exchange (&head_mem->head, head, head + 1)); #ifdef LOW_MEM /* decrement number of readers, when we reach 0 readers we can be sure that @@ -341,34 +348,44 @@ gst_atomic_queue_push (GstAtomicQueue * queue, gpointer data) tail_mem = g_atomic_pointer_get (&queue->tail_mem); head = g_atomic_int_get (&tail_mem->head); - tail = g_atomic_int_get (&tail_mem->tail); + tail = g_atomic_int_get (&tail_mem->tail_write); size = tail_mem->size; /* we're not full, continue */ - if (tail - head <= size) - break; + if G_LIKELY + (tail - head <= size) + break; /* else we need to grow the array, we store a mask so we have to add 1 */ mem = new_queue_mem ((size << 1) + 1, tail); /* try to make our new array visible to other writers */ - if (!g_atomic_pointer_compare_and_exchange ((gpointer *) & - queue->tail_mem, tail_mem, mem)) { + if G_UNLIKELY + (!g_atomic_pointer_compare_and_exchange (&queue->tail_mem, tail_mem, + mem)) { /* we tried to swap the new writer array but something changed. This is * because some other writer beat us to it, we free our memory and try * again */ free_queue_mem (mem); continue; - } + } /* make sure that readers can find our new array as well. The one who * manages to swap the pointer is the only one who can set the next * pointer to the new array */ g_atomic_pointer_set (&tail_mem->next, mem); } - } while (!g_atomic_int_compare_and_exchange (&tail_mem->tail, tail, - tail + 1)); + } while G_UNLIKELY + (!g_atomic_int_compare_and_exchange (&tail_mem->tail_write, tail, tail + 1)); tail_mem->array[tail & size] = data; + + /* now wait until all writers have completed their write before we move the + * tail_read to this new item. It is possible that other writers are still + * updating the previous array slots and we don't want to reveal their changes + * before they are done. FIXME, it would be nice if we didn't have to busy + * wait here. */ + while G_UNLIKELY + (!g_atomic_int_compare_and_exchange (&tail_mem->tail_read, tail, tail + 1)); } /** @@ -395,7 +412,7 @@ gst_atomic_queue_length (GstAtomicQueue * queue) head = g_atomic_int_get (&head_mem->head); tail_mem = g_atomic_pointer_get (&queue->tail_mem); - tail = g_atomic_int_get (&tail_mem->tail); + tail = g_atomic_int_get (&tail_mem->tail_read); #ifdef LOW_MEM if (g_atomic_int_dec_and_test (&queue->num_readers))