An implementation of lightweight unbounded queues

2 files changed, 108 insertions, 1 deletions

diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index 594c0ccf94..20f100b427 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -380,7 +380,7 @@ extern int bif_reductions;      /* reductions + fcalls (when doing call_bif) */
 extern int stackdump_on_exit;
 
 /*
- * Here is an implementation of a lightweiht stack.
+ * Here is an implementation of a lightweight stack.
  *
  * Use it like this:
  *
@@ -845,6 +845,88 @@ do {\
 } while(0)
 
 
+/*
+ *  An implementation of lightweight unbounded queues,
+ *  using a circular dynamic array.
+ *  It does not include support for change_allocator.
+ *
+ *  Use it like this:
+ *
+ *  DECLARE_EQUEUE(Queue)	(At the start of a block)
+ *  ...
+ *  EQUEUE_PUT(Queue, Term)
+ *  ...
+ *  if (EQUEUE_ISEMPTY(Queue)) {
+ *     Queue is empty
+ *  } else {
+ *     Term = EQUEUE_GET(Stack);
+ *     Process popped Term here
+ *  }
+ *  ...
+ *  DESTROY_EQUEUE(Queue)
+ */
+
+typedef struct {
+    Eterm* start;
+    Eterm* front;
+    Eterm* back;
+    int possibly_empty;
+    Eterm* end;
+    ErtsAlcType_t alloc_type;
+}ErtsEQueue;
+
+#define DEF_EQUEUE_SIZE (16)
+
+void erl_grow_equeue(ErtsEQueue*, Eterm* def_queue);
+#define EQUE_CONCAT(a,b) a##b
+#define EQUE_DEF_QUEUE(q) EQUE_CONCAT(q,_default_equeue)
+
+#define DECLARE_EQUEUE(q)				\
+    UWord EQUE_DEF_QUEUE(q)[DEF_EQUEUE_SIZE];     	\
+    ErtsEQueue q = {					\
+        EQUE_DEF_QUEUE(q), /* start */			\
+        EQUE_DEF_QUEUE(q), /* front */			\
+        EQUE_DEF_QUEUE(q), /* back */			\
+        1,                 /* possibly_empty */		\
+        EQUE_DEF_QUEUE(q) + DEF_EQUEUE_SIZE, /* end */	\
+        ERTS_ALC_T_ESTACK  /* alloc_type */		\
+    }
+
+#define DESTROY_EQUEUE(q)				\
+do {							\
+    if (q.start != EQUE_DEF_QUEUE(q)) {			\
+      erts_free(q.alloc_type, q.start);			\
+    }							\
+} while(0)
+
+#define EQUEUE_PUT_UNCHECKED(q, x)			\
+do {							\
+    q.possibly_empty = 0;				\
+    *(q.back) = (x);                    		\
+    if (++(q.back) == q.end) {				\
+	q.back = q.start;				\
+    }							\
+} while(0)
+
+#define EQUEUE_PUT(q, x)				\
+do {							\
+    if (q.back == q.front && !q.possibly_empty) {	\
+        erl_grow_equeue(&q, EQUE_DEF_QUEUE(q));		\
+    }							\
+    EQUEUE_PUT_UNCHECKED(q, x);				\
+} while(0)
+
+#define EQUEUE_ISEMPTY(q) (q.back == q.front && q.possibly_empty)
+
+#define EQUEUE_GET(q) ({				\
+    q.possibly_empty = 1;				\
+    UWord x = *(q.front);				\
+    if (++(q.front) == q.end) {				\
+        q.front = q.start;				\
+    }							\
+    x;							\
+})
+
 /* binary.c */
 
 void erts_emasculate_writable_binary(ProcBin* pb);
diff --git a/erts/emulator/beam/utils.c b/erts/emulator/beam/utils.c
index c3735683bb..184477c36b 100644
--- a/erts/emulator/beam/utils.c
+++ b/erts/emulator/beam/utils.c
@@ -258,6 +258,31 @@ erl_grow_pstack(ErtsPStack* s, void* default_pstack, unsigned need_bytes)
     s->psp = s->pstart + sp_offs;
 }
 
+/*
+ * Helper function for the EQUEUE macros defined in global.h.
+ */
+
+void
+erl_grow_equeue(ErtsEQueue* q, Eterm* default_equeue)
+{
+    Uint old_size = (q->end - q->start);
+    Uint new_size = old_size * 2;
+    Uint first_part = (q->end - q->front);
+    Uint second_part = (q->back - q->start);
+    Eterm* new_ptr = erts_alloc(q->alloc_type, new_size*sizeof(Eterm));
+    ASSERT(q->back == q->front);   // of course the queue is full now!
+    if (first_part > 0)
+      sys_memcpy(new_ptr, q->front, first_part*sizeof(Eterm));
+    if (second_part > 0)
+      sys_memcpy(new_ptr+first_part, q->start, second_part*sizeof(Eterm));
+    if (q->start != default_equeue)
+      erts_free(q->alloc_type, q->start);
+    q->start = new_ptr;
+    q->end = q->start + new_size;
+    q->front = q->start;
+    q->back = q->start + old_size;
+}
+
 /* CTYPE macros */
 
 #define LATIN1