From 2bfc6d6791274e8f170e8972c899dcb66e8881c1 Mon Sep 17 00:00:00 2001
From: Rickard Green <rickard@erlang.org>
Date: Wed, 7 Dec 2011 20:05:39 +0100
Subject: Fix cleanup of elements in lock-free queues

---
 erts/emulator/beam/erl_async.c     |  5 ++--
 erts/emulator/beam/erl_process.c   |  2 +-
 erts/emulator/beam/erl_thr_queue.c | 58 +++++++++++++++++++++++++-------------
 erts/emulator/beam/erl_thr_queue.h |  2 --
 4 files changed, 42 insertions(+), 25 deletions(-)

diff --git a/erts/emulator/beam/erl_async.c b/erts/emulator/beam/erl_async.c
index 2dc7237f7c..8bca9ae582 100644
--- a/erts/emulator/beam/erl_async.c
+++ b/erts/emulator/beam/erl_async.c
@@ -304,8 +304,9 @@ static ERTS_INLINE ErtsAsync *async_get(ErtsThrQ_t *q,
 	    switch (erts_thr_q_inspect(q, 1)) {
 	    case ERTS_THR_Q_DIRTY:
 		break;
+	    case ERTS_THR_Q_NEED_THR_PRGR:
 #ifdef ERTS_SMP
-	    case ERTS_THR_Q_NEED_THR_PRGR: {
+	    {
 		ErtsThrPrgrVal prgr = erts_thr_q_need_thr_progress(q);
 		erts_thr_progress_wakeup(NULL, prgr);
 		/*
@@ -522,8 +523,8 @@ int erts_async_ready_clean(void *varq, void *val)
     switch (cstate) {
     case ERTS_THR_Q_DIRTY:
 	return ERTS_ASYNC_READY_DIRTY;
-#ifdef ERTS_SMP
     case ERTS_THR_Q_NEED_THR_PRGR:
+#ifdef ERTS_SMP
 	*((ErtsThrPrgrVal *) val)
 	    = erts_thr_q_need_thr_progress(&arq->thr_q);
 	return ERTS_ASYNC_READY_NEED_THR_PRGR;
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 5e66c3e229..a6da4154e0 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -777,8 +777,8 @@ misc_aux_work_clean(ErtsThrQ_t *q,
     case ERTS_THR_Q_DIRTY:
 	set_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_MISC);
 	return aux_work | ERTS_SSI_AUX_WORK_MISC;
-#ifdef ERTS_SMP
     case ERTS_THR_Q_NEED_THR_PRGR:
+#ifdef ERTS_SMP
 	set_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_MISC_THR_PRGR);
 	erts_thr_progress_wakeup(awdp->esdp,
 				 erts_thr_q_need_thr_progress(q));
diff --git a/erts/emulator/beam/erl_thr_queue.c b/erts/emulator/beam/erl_thr_queue.c
index 9ac4cd4b8e..efb8c635d7 100644
--- a/erts/emulator/beam/erl_thr_queue.c
+++ b/erts/emulator/beam/erl_thr_queue.c
@@ -449,32 +449,44 @@ clean(ErtsThrQ_t *q, int max_ops, int do_notify)
 		if (inext == (erts_aint_t) &q->tail.data.marker) {
 		    q->head.head.ptr->next.ptr = &q->tail.data.marker;
 		    q->head.head.ptr = &q->tail.data.marker;
-#ifdef ERTS_SMP
-		    if (!q->head.next.thr_progress_reached)
-			return ERTS_THR_Q_NEED_THR_PRGR;
-#else
-		    if (do_notify)
-			q->head.notify(q->head.arg);
-#endif
-		    return ERTS_THR_Q_DIRTY;
+		    goto check_thr_progress;
 		}
 	    }
 	}
+
+	if (q->q.finalizing) {
+	    ilast = erts_atomic_read_nob(&q->tail.data.last);
+	    if (q->head.first == ((ErtsThrQElement_t *) ilast)
+		&& ((ErtsThrQElement_t *) ilast) == &q->tail.data.marker
+		&& q->head.first == &q->tail.data.marker) {
+		destroy(q);
+	    }
+	    else {
+		goto dirty;
+	    }
+	}
 	return ERTS_THR_Q_CLEAN;
     }
 
-    if (q->head.first != q->head.unref_end) {
-	if (do_notify)
-	    q->head.notify(q->head.arg);
-	return ERTS_THR_Q_DIRTY;
-    }
+    if (q->head.first != q->head.unref_end)
+	goto dirty;
+
+check_thr_progress:
 
 #ifdef ERTS_SMP
-    if (!q->head.next.thr_progress_reached)
-	return ERTS_THR_Q_NEED_THR_PRGR;
+    if (q->head.next.thr_progress_reached)
 #endif
+    {
+	int um_refc_ix = q->head.next.um_refc_ix;
+	if (erts_atomic_read_acqb(&q->tail.data.um_refc[um_refc_ix]) == 0) {
+	dirty:
+	    if (do_notify)
+		q->head.notify(q->head.arg);
+	    return ERTS_THR_Q_DIRTY;
+	}
+    }
 
-    return ERTS_THR_Q_CLEAN; /* Waiting for unmanaged threads to complete... */
+    return ERTS_THR_Q_NEED_THR_PRGR;
 }
 
 #endif
@@ -492,7 +504,9 @@ erts_thr_q_clean(ErtsThrQ_t *q)
 ErtsThrQCleanState_t
 erts_thr_q_inspect(ErtsThrQ_t *q, int ensure_empty)
 {
-#ifdef USE_THREADS
+#ifndef USE_THREADS
+    return ERTS_THR_Q_CLEAN;
+#else
     if (ensure_empty) {
 	erts_aint_t inext;
 	inext = erts_atomic_read_acqb(&q->head.head.ptr->next.atmc);
@@ -523,11 +537,15 @@ erts_thr_q_inspect(ErtsThrQ_t *q, int ensure_empty)
 	return ERTS_THR_Q_DIRTY;
 
 #ifdef ERTS_SMP
-    if (!q->head.next.thr_progress_reached)
-	return ERTS_THR_Q_NEED_THR_PRGR;
+    if (q->head.next.thr_progress_reached)
 #endif
+    {
+	int um_refc_ix = q->head.next.um_refc_ix;
+	if (erts_atomic_read_acqb(&q->tail.data.um_refc[um_refc_ix]) == 0)
+	    return ERTS_THR_Q_DIRTY;
+    }
+    return ERTS_THR_Q_NEED_THR_PRGR;
 #endif
-    return ERTS_THR_Q_CLEAN;
 }
 
 static void
diff --git a/erts/emulator/beam/erl_thr_queue.h b/erts/emulator/beam/erl_thr_queue.h
index 407c23f5eb..edcf2c3823 100644
--- a/erts/emulator/beam/erl_thr_queue.h
+++ b/erts/emulator/beam/erl_thr_queue.h
@@ -96,9 +96,7 @@ typedef struct {
 
 typedef enum {
     ERTS_THR_Q_CLEAN,
-#ifdef ERTS_SMP
     ERTS_THR_Q_NEED_THR_PRGR,
-#endif
     ERTS_THR_Q_DIRTY,
 } ErtsThrQCleanState_t;
 
-- 
cgit v1.2.3