queuearray: add lockfree untested dataqueue

3 files changed, 190 insertions, 2 deletions

diff --git a/libs/gst/base/Makefile.am b/libs/gst/base/Makefile.am
index 0ece81944..72f88f8c5 100644
--- a/libs/gst/base/Makefile.am
+++ b/libs/gst/base/Makefile.am
@@ -13,7 +13,8 @@ libgstbase_@GST_MAJORMINOR@_la_SOURCES = \
 	gstcollectpads.c	\
 	gstpushsrc.c		\
 	gsttypefindhelper.c	\
-	gstdataqueue.c
+	gstdataqueue.c          \
+	gstqueuearray.c
 
 libgstbase_@GST_MAJORMINOR@_la_CFLAGS = $(GST_OBJ_CFLAGS)
 libgstbase_@GST_MAJORMINOR@_la_LIBADD = $(GST_OBJ_LIBS)
@@ -33,7 +34,8 @@ libgstbase_@GST_MAJORMINOR@include_HEADERS =	\
 	gstcollectpads.h	\
 	gstpushsrc.h		\
 	gsttypefindhelper.h	\
-	gstdataqueue.h
+	gstdataqueue.h          \
+	gstqueuearray.h
 
 noinst_HEADERS = \
 	gstbytereader-docs.h \
diff --git a/libs/gst/base/gstqueuearray.c b/libs/gst/base/gstqueuearray.c
new file mode 100644
index 000000000..584b638a1
--- /dev/null
+++ b/libs/gst/base/gstqueuearray.c
@@ -0,0 +1,155 @@
+/* GStreamer
+ * Copyright (C) 2009 Edward Hervey <bilboed@bilboed.com>
+ *
+ * gstqueuearray.c:
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * 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.
+ */
+
+#include <string.h>
+#include <gst/gst.h>
+#include "gstqueuearray.h"
+
+typedef struct _GstQueueArrayMem GstQueueArrayMem;
+
+struct _GstQueueArrayMem
+{
+  gint size;
+  gpointer *array;
+  volatile gint head;
+  volatile gint tail;
+  GstQueueArrayMem *next;
+};
+
+struct _GstQueueArray
+{
+  GstQueueArrayMem *head_mem;
+  GstQueueArrayMem *tail_mem;
+  volatile gint length;
+};
+
+GstQueueArray *
+gst_queue_array_new (guint initial_size)
+{
+  GstQueueArray *array;
+  GstQueueArrayMem *mem;
+
+  array = g_new (GstQueueArray, 1);
+  mem = g_new (GstQueueArrayMem, 1);
+
+  mem->size = initial_size;
+  mem->array = g_new0 (gpointer, initial_size);
+  mem->next = NULL;
+  mem->head = 0;
+  mem->tail = 0;
+
+  array->head_mem = array->tail_mem = mem;
+  array->length = 0;
+
+  return array;
+}
+
+gpointer
+gst_queue_array_pop_head (GstQueueArray * array)
+{
+  gpointer ret;
+  GstQueueArrayMem *head_mem;
+  gint head, tail;
+
+  do {
+    head_mem = array->head_mem;
+    head = g_atomic_int_get (&head_mem->head);
+    tail = g_atomic_int_get (&head_mem->tail);
+
+    while (head == tail) {
+      GstQueueArrayMem *next;
+
+      /* array empty, try to take next */
+      next = g_atomic_pointer_get (&head_mem->next);
+      if (next == NULL)
+        return NULL;
+
+      next->head = head;
+      array->head_mem = next;
+      /* old head array is now useless */
+      g_free (head_mem);
+      head_mem = next;
+
+      tail = g_atomic_int_get (&head_mem->tail);
+    }
+
+    ret = head_mem->array[head % head_mem->size];
+  } while (!g_atomic_int_compare_and_exchange (&head_mem->head, head,
+          head + 1));
+
+  g_atomic_int_exchange_and_add (&array->length, -1);
+
+  return ret;
+}
+
+void
+gst_queue_array_push_tail (GstQueueArray * array, gpointer data)
+{
+  GstQueueArrayMem *tail_mem;
+  gint head, tail;
+
+  do {
+    tail_mem = array->tail_mem;
+    head = g_atomic_int_get (&tail_mem->head);
+    tail = g_atomic_int_get (&tail_mem->tail);
+
+    while (tail - head >= tail_mem->size - 1) {
+      GstQueueArrayMem *mem;
+      gint idx;
+
+      /* need to grow the array */
+      mem = g_new (GstQueueArrayMem, 1);
+      mem->size = tail_mem->size * 2;
+      mem->array = g_new0 (gpointer, mem->size);
+      mem->next = NULL;
+      mem->tail = tail;
+      mem->head = head;
+
+      /* copy data */
+      idx = head % tail_mem->size;
+      memcpy (&mem->array[idx], &tail_mem->array[idx],
+          sizeof (gpointer) * (tail_mem->size - idx));
+      idx = tail % tail_mem->size;
+      memcpy (&mem->array[tail_mem->size], tail_mem->array,
+          sizeof (gpointer) * idx);
+
+      g_atomic_pointer_set (&tail_mem->next, mem);
+      array->tail_mem = mem;
+      tail_mem = mem;
+    }
+  } while (!g_atomic_int_compare_and_exchange (&tail_mem->tail, tail,
+          tail + 1));
+
+  tail_mem->array[tail % tail_mem->size] = data;
+  g_atomic_int_inc (&array->length);
+}
+
+gboolean
+gst_queue_array_is_empty (GstQueueArray * array)
+{
+  return g_atomic_int_get (&array->length) == 0;
+}
+
+void
+gst_queue_array_free (GstQueueArray * array)
+{
+  g_free (array);
+}
diff --git a/libs/gst/base/gstqueuearray.h b/libs/gst/base/gstqueuearray.h
new file mode 100644
index 000000000..87791836f
--- /dev/null
+++ b/libs/gst/base/gstqueuearray.h
@@ -0,0 +1,31 @@
+/* GStreamer
+ * Copyright (C) 2009-2010 Edward Hervey <bilboed@bilboed.com>
+ *
+ * gstqueuearray.h:
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * 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.
+ */
+
+#include <glib.h>
+
+typedef struct _GstQueueArray GstQueueArray;
+
+GstQueueArray *gst_queue_array_new (guint initial_size);
+gpointer gst_queue_array_pop_head (GstQueueArray* array);
+void gst_queue_array_push_tail (GstQueueArray* array, gpointer data);
+gboolean gst_queue_array_is_empty (GstQueueArray * array);
+void gst_queue_array_free (GstQueueArray * array);
+