aboutsummaryrefslogtreecommitdiffstats
path: root/erts
diff options
context:
space:
mode:
authorRickard Green <[email protected]>2011-12-07 20:05:39 +0100
committerRickard Green <[email protected]>2011-12-07 20:21:24 +0100
commit2bfc6d6791274e8f170e8972c899dcb66e8881c1 (patch)
treef6699766d360dd0cb9991886dab9c2206730a15f /erts
parent69969c2cd7b2aa3efbf35df8a908a17377a895a4 (diff)
downloadotp-2bfc6d6791274e8f170e8972c899dcb66e8881c1.tar.gz
otp-2bfc6d6791274e8f170e8972c899dcb66e8881c1.tar.bz2
otp-2bfc6d6791274e8f170e8972c899dcb66e8881c1.zip
Fix cleanup of elements in lock-free queues
Diffstat (limited to 'erts')
-rw-r--r--erts/emulator/beam/erl_async.c5
-rw-r--r--erts/emulator/beam/erl_process.c2
-rw-r--r--erts/emulator/beam/erl_thr_queue.c58
-rw-r--r--erts/emulator/beam/erl_thr_queue.h2
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;