Implement generic lock-free queue

The implementation of an ERTS internal, generic, many to one, lock-free
queue for communication between threads. The many to one scenario is
very common in ERTS, so it can be used in a lot of places in the future.
Changing to this queue from a lock based queue, however, often requires
some redesigning. This since we have often used the lock of the queue
to protect other information too.

1 files changed, 211 insertions, 0 deletions

diff --git a/erts/emulator/beam/erl_thr_queue.h b/erts/emulator/beam/erl_thr_queue.h
new file mode 100644
index 0000000000..407c23f5eb
--- /dev/null
+++ b/erts/emulator/beam/erl_thr_queue.h
@@ -0,0 +1,211 @@
+/*
+ * %CopyrightBegin%
+ *
+ * Copyright Ericsson AB 2011. All Rights Reserved.
+ *
+ * The contents of this file are subject to the Erlang Public License,
+ * Version 1.1, (the "License"); you may not use this file except in
+ * compliance with the License. You should have received a copy of the
+ * Erlang Public License along with this software. If not, it can be
+ * retrieved online at http://www.erlang.org/.
+ *
+ * Software distributed under the License is distributed on an "AS IS"
+ * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+ * the License for the specific language governing rights and limitations
+ * under the License.
+ *
+ * %CopyrightEnd%
+ */
+
+/*
+ * Description: Lock-free queue for communication between threads.
+ *
+ *              Currently only a many-to-one version has been,
+ *              implemented, i.e., many threads can enqueue but
+ *              only one thread can dequeue at a time. It doesn't
+ *              have to be the same thread dequeuing every time, but
+ *              synchronization so that only one thread dequeues
+ *              at a time has to be provided by other means.
+ *
+ *              When/If the need for a many-to-many queue arises,
+ *              this implementation can relatively easy be extended
+ *              to support that too.
+ *
+ *              Usage instructions can be found in erts_thr_queue.c
+ *
+ * Author: 	Rickard Green
+ */
+
+#ifndef ERL_THR_QUEUE_H__
+#define ERL_THR_QUEUE_H__
+
+#include "sys.h"
+#include "erl_threads.h"
+#include "erl_alloc.h"
+#include "erl_thr_progress.h"
+
+typedef enum {
+    ERTS_THR_Q_LIVE_UNDEF,
+    ERTS_THR_Q_LIVE_SHORT,
+    ERTS_THR_Q_LIVE_LONG
+} ErtsThrQLive_t;
+
+#define ERTS_THR_Q_INIT_DEFAULT						\
+{									\
+    {									\
+	ERTS_THR_Q_LIVE_UNDEF,						\
+	ERTS_THR_Q_LIVE_SHORT						\
+    },									\
+    NULL,								\
+    NULL,								\
+    1									\
+}
+
+typedef struct ErtsThrQ_t_ ErtsThrQ_t;
+
+typedef struct {
+    struct {
+	ErtsThrQLive_t queue;
+	ErtsThrQLive_t objects;
+    } live;
+    void *arg;
+    void (*notify)(void *);
+    int auto_finalize_dequeue;
+} ErtsThrQInit_t;
+
+typedef struct ErtsThrQElement_t_ ErtsThrQElement_t;
+typedef struct ErtsThrQElement_t ErtsThrQPrepEnQ_t;
+
+typedef union {
+    erts_atomic_t atmc;
+    ErtsThrQElement_t *ptr;
+} ErtsThrQPtr_t;
+
+struct ErtsThrQElement_t_ {
+    ErtsThrQPtr_t next;
+    union {
+	erts_atomic_t atmc;
+	void *ptr;
+    } data;
+};
+
+typedef struct {
+    ErtsThrQElement_t *start;
+    ErtsThrQElement_t *end;
+} ErtsThrQFinDeQ_t;
+
+typedef enum {
+    ERTS_THR_Q_CLEAN,
+#ifdef ERTS_SMP
+    ERTS_THR_Q_NEED_THR_PRGR,
+#endif
+    ERTS_THR_Q_DIRTY,
+} ErtsThrQCleanState_t;
+
+#ifdef USE_THREADS
+
+typedef struct {
+    ErtsThrQElement_t marker;
+    erts_atomic_t last;
+    erts_atomic_t um_refc[2];
+    erts_atomic32_t um_refc_ix;
+    ErtsThrQLive_t live;
+#ifdef ERTS_SMP
+    erts_atomic32_t thr_prgr_clean_scheduled;
+#endif
+    void *arg;
+    void (*notify)(void *);
+} ErtsThrQTail_t;
+
+struct ErtsThrQ_t_ {
+    /*
+     * This structure needs to be cache line aligned for best
+     * performance.
+     */
+    union {
+	/* Modified by threads enqueuing */
+	ErtsThrQTail_t data;
+	char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsThrQTail_t))];
+    } tail;
+    /*
+     * Everything below this point is *only* accessed by the
+     * thread dequeuing.
+     */
+    struct {
+	ErtsThrQPtr_t head;
+	ErtsThrQLive_t live;
+	ErtsThrQElement_t *first;
+	ErtsThrQElement_t *unref_end;
+	int clean_reached_head_count;
+	struct {
+	    int automatic;
+	    ErtsThrQElement_t *start;
+	    ErtsThrQElement_t *end;
+	} deq_fini;
+	struct {
+#ifdef ERTS_SMP
+	    ErtsThrPrgrVal thr_progress;
+	    int thr_progress_reached;
+#endif
+	    int um_refc_ix;
+	    ErtsThrQElement_t *unref_end;
+	} next;
+	int used_marker;
+	void *arg;
+	void (*notify)(void *);
+    } head;
+    struct {
+	int finalizing;
+	ErtsThrQLive_t live;
+	void *blk;
+    } q;
+};
+
+#else /* !USE_THREADS */
+
+struct ErtsThrQ_t_ {
+    ErtsThrQInit_t init;
+    ErtsThrQElement_t *first;
+    ErtsThrQElement_t *last;
+    struct {
+	void *blk;
+    } q;
+};
+
+#endif
+
+void erts_thr_q_init(void);
+void erts_thr_q_initialize(ErtsThrQ_t *, ErtsThrQInit_t *);
+ErtsThrQCleanState_t erts_thr_q_finalize(ErtsThrQ_t *);
+ErtsThrQ_t *erts_thr_q_create(ErtsThrQInit_t *);
+ErtsThrQCleanState_t erts_thr_q_destroy(ErtsThrQ_t *);
+ErtsThrQCleanState_t erts_thr_q_clean(ErtsThrQ_t *);
+ErtsThrQCleanState_t erts_thr_q_inspect(ErtsThrQ_t *, int);
+ErtsThrQPrepEnQ_t *erts_thr_q_prepare_enqueue(ErtsThrQ_t *);
+void erts_thr_q_enqueue_prepared(ErtsThrQ_t *, void *, ErtsThrQPrepEnQ_t *);
+void erts_thr_q_enqueue(ErtsThrQ_t *, void *);
+void * erts_thr_q_dequeue(ErtsThrQ_t *);
+int erts_thr_q_get_finalize_dequeue_data(ErtsThrQ_t *,
+					 ErtsThrQFinDeQ_t *);
+void erts_thr_q_append_finalize_dequeue_data(ErtsThrQFinDeQ_t *,
+					     ErtsThrQFinDeQ_t *);
+int erts_thr_q_finalize_dequeue(ErtsThrQFinDeQ_t *);
+void erts_thr_q_finalize_dequeue_state_init(ErtsThrQFinDeQ_t *);
+
+#ifdef ERTS_SMP
+ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_q_need_thr_progress(ErtsThrQ_t *q);
+#endif
+
+#if ERTS_GLB_INLINE_INCL_FUNC_DEF
+
+#ifdef ERTS_SMP
+ERTS_GLB_INLINE ErtsThrPrgrVal
+erts_thr_q_need_thr_progress(ErtsThrQ_t *q)
+{
+    return q->head.next.thr_progress;
+}
+#endif
+
+#endif /* ERTS_GLB_INLINE_INCL_FUNC_DEF */
+
+#endif /* ERL_THR_QUEUE_H__ */