aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam')
-rw-r--r--erts/emulator/beam/erl_process_lock.c36
-rw-r--r--erts/emulator/beam/erl_thr_progress.c285
-rw-r--r--erts/emulator/beam/erl_thr_progress.h62
3 files changed, 302 insertions, 81 deletions
diff --git a/erts/emulator/beam/erl_process_lock.c b/erts/emulator/beam/erl_process_lock.c
index b864340782..fa935af315 100644
--- a/erts/emulator/beam/erl_process_lock.c
+++ b/erts/emulator/beam/erl_process_lock.c
@@ -66,6 +66,7 @@
#endif
#include "erl_process.h"
+#include "erl_thr_progress.h"
const Process erts_proc_lock_busy = {ERTS_INVALID_PID};
@@ -671,7 +672,7 @@ erts_proc_lock_prepare_proc_lock_waiter(void)
*/
static void
-proc_safelock(int is_sched,
+proc_safelock(int is_managed,
Process *a_proc,
ErtsProcLocks a_have_locks,
ErtsProcLocks a_need_locks,
@@ -797,7 +798,7 @@ proc_safelock(int is_sched,
if (unlock_locks) {
have_locks1 &= ~unlock_locks;
need_locks1 |= unlock_locks;
- if (!is_sched && !have_locks1) {
+ if (!is_managed && !have_locks1) {
refc1 = 1;
erts_smp_proc_inc_refc(p1);
}
@@ -807,7 +808,7 @@ proc_safelock(int is_sched,
if (unlock_locks) {
have_locks2 &= ~unlock_locks;
need_locks2 |= unlock_locks;
- if (!is_sched && !have_locks2) {
+ if (!is_managed && !have_locks2) {
refc2 = 1;
erts_smp_proc_inc_refc(p2);
}
@@ -888,7 +889,7 @@ proc_safelock(int is_sched,
}
#endif
- if (!is_sched) {
+ if (!is_managed) {
if (refc1)
erts_smp_proc_dec_refc(p1);
if (refc2)
@@ -921,7 +922,7 @@ erts_pid2proc_opt(Process *c_p,
int flags)
{
Process *dec_refc_proc = NULL;
- int need_ptl;
+ ErtsThrPrgrDelayHandle dhndl;
ErtsProcLocks need_locks;
Uint pix;
Process *proc;
@@ -959,10 +960,7 @@ erts_pid2proc_opt(Process *c_p,
}
}
- need_ptl = !erts_get_scheduler_id();
-
- if (need_ptl)
- erts_smp_rwmtx_rwlock(&erts_proc_tab_rwmtx);
+ dhndl = erts_thr_progress_unmanaged_delay();
proc = (Process *) erts_smp_atomic_read_ddrb(&erts_proc.tab[pix]);
@@ -1026,6 +1024,7 @@ erts_pid2proc_opt(Process *c_p,
if (flags & ERTS_P2P_FLG_TRY_LOCK)
proc = ERTS_PROC_LOCK_BUSY;
else {
+ int managed;
if (flags & ERTS_P2P_FLG_SMP_INC_REFC)
erts_smp_proc_inc_refc(proc);
@@ -1033,14 +1032,21 @@ erts_pid2proc_opt(Process *c_p,
erts_lcnt_proc_lock_unaquire(&proc->lock, lcnt_locks);
#endif
- if (need_ptl) {
+ managed = dhndl == ERTS_THR_PRGR_DHANDLE_MANAGED;
+ if (!managed) {
erts_smp_proc_inc_refc(proc);
+ erts_thr_progress_unmanaged_continue(dhndl);
dec_refc_proc = proc;
- erts_smp_rwmtx_rwunlock(&erts_proc_tab_rwmtx);
- need_ptl = 0;
+
+ /*
+ * We don't want to call
+ * erts_thr_progress_unmanaged_continue()
+ * again.
+ */
+ dhndl = ERTS_THR_PRGR_DHANDLE_MANAGED;
}
- proc_safelock(!need_ptl,
+ proc_safelock(managed,
c_p,
c_p_have_locks,
c_p_have_locks,
@@ -1052,8 +1058,8 @@ erts_pid2proc_opt(Process *c_p,
}
}
- if (need_ptl)
- erts_smp_rwmtx_rwunlock(&erts_proc_tab_rwmtx);
+ if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED)
+ erts_thr_progress_unmanaged_continue(dhndl);
if (need_locks
&& proc
diff --git a/erts/emulator/beam/erl_thr_progress.c b/erts/emulator/beam/erl_thr_progress.c
index 88524bdd4c..3b2f95d43b 100644
--- a/erts/emulator/beam/erl_thr_progress.c
+++ b/erts/emulator/beam/erl_thr_progress.c
@@ -96,17 +96,14 @@
#define ERTS_THR_PRGR_LFLG_BLOCK (((erts_aint32_t) 1) << 31)
#define ERTS_THR_PRGR_LFLG_NO_LEADER (((erts_aint32_t) 1) << 30)
-#define ERTS_THR_PRGR_LFLG_ACTIVE_MASK (~(ERTS_THR_PRGR_LFLG_NO_LEADER \
- | ERTS_THR_PRGR_LFLG_BLOCK))
+#define ERTS_THR_PRGR_LFLG_WAITING_UM (((erts_aint32_t) 1) << 29)
+#define ERTS_THR_PRGR_LFLG_ACTIVE_MASK (~(ERTS_THR_PRGR_LFLG_NO_LEADER \
+ | ERTS_THR_PRGR_LFLG_BLOCK \
+ | ERTS_THR_PRGR_LFLG_WAITING_UM))
-#define ERTS_THR_PRGR_LFLGS_ACTIVE(LFLGS) \
+#define ERTS_THR_PRGR_LFLGS_ACTIVE(LFLGS) \
((LFLGS) & ERTS_THR_PRGR_LFLG_ACTIVE_MASK)
-#define ERTS_THR_PRGR_LFLGS_ALL_WAITING(LFLGS) \
- (((LFLGS) & (ERTS_THR_PRGR_LFLG_NO_LEADER \
- |ERTS_THR_PRGR_LFLG_ACTIVE_MASK)) \
- == ERTS_THR_PRGR_LFLG_NO_LEADER)
-
/*
* We use a 64-bit value for thread progress. By this wrapping of
* the thread progress will more or less never occur.
@@ -262,6 +259,11 @@ typedef struct {
erts_atomic32_t managed_count;
erts_atomic32_t managed_id;
erts_atomic32_t unmanaged_id;
+ int chk_next_ix;
+ struct {
+ int waiting;
+ erts_atomic32_t current;
+ } umrefc_ix;
} ErtsThrPrgrMiscData;
typedef struct {
@@ -276,12 +278,18 @@ typedef union {
char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsThrPrgrElement))];
} ErtsThrPrgrArray;
+typedef union {
+ erts_atomic_t refc;
+ char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(erts_atomic_t))];
+} ErtsThrPrgrUnmanagedRefc;
+
typedef struct {
union {
ErtsThrPrgrMiscData data;
char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(
sizeof(ErtsThrPrgrMiscData))];
} misc;
+ ErtsThrPrgrUnmanagedRefc umrefc[2];
ErtsThrPrgrArray *thr;
struct {
int no;
@@ -346,7 +354,9 @@ init_tmp_thr_prgr_data(ErtsThrPrgrData *tpd)
tpd->is_managed = 0;
tpd->is_blocking = 0;
tpd->is_temporary = 1;
-
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ tpd->is_delaying = 0;
+#endif
erts_tsd_set(erts_thr_prgr_data_key__, (void *) tpd);
}
@@ -461,6 +471,9 @@ erts_thr_progress_init(int no_schedulers, int managed, int unmanaged)
erts_atomic32_init_nob(&intrnl->misc.data.managed_count, 0);
erts_atomic32_init_nob(&intrnl->misc.data.managed_id, no_schedulers);
erts_atomic32_init_nob(&intrnl->misc.data.unmanaged_id, -1);
+ intrnl->misc.data.chk_next_ix = 0;
+ intrnl->misc.data.umrefc_ix.waiting = -1;
+ erts_atomic32_init_nob(&intrnl->misc.data.umrefc_ix.current, 0);
intrnl->thr = (ErtsThrPrgrArray *) ptr;
ptr += thr_arr_sz;
@@ -547,6 +560,9 @@ erts_thr_progress_register_unmanaged_thread(ErtsThrPrgrCallbacks *callbacks)
tpd->is_managed = 0;
tpd->is_blocking = is_blocking;
tpd->is_temporary = 0;
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ tpd->is_delaying = 0;
+#endif
ASSERT(tpd->id >= 0);
if (tpd->id >= intrnl->unmanaged.no)
erl_exit(ERTS_ABORT_EXIT,
@@ -600,6 +616,9 @@ erts_thr_progress_register_managed_thread(ErtsSchedulerData *esdp,
tpd->is_managed = 1;
tpd->is_blocking = is_blocking;
tpd->is_temporary = 0;
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ tpd->is_delaying = 1;
+#endif
init_wakeup_request_array(&tpd->wakeup_request[0]);
@@ -607,8 +626,8 @@ erts_thr_progress_register_managed_thread(ErtsSchedulerData *esdp,
tpd->leader = 0;
tpd->active = 1;
- tpd->previous.local = 0;
- tpd->previous.current = ERTS_THR_PRGR_VAL_WAITING;
+ tpd->confirmed = 0;
+ tpd->leader_state.current = ERTS_THR_PRGR_VAL_WAITING;
erts_tsd_set(erts_thr_prgr_data_key__, (void *) tpd);
erts_atomic32_inc_nob(&intrnl->misc.data.lflgs);
@@ -651,60 +670,113 @@ leader_update(ErtsThrPrgrData *tpd)
block_thread(tpd);
}
else {
+ ErtsThrPrgrVal current;
+ int ix, chk_next_ix, umrefc_ix, my_ix, no_managed, waiting_unmanaged;
erts_aint32_t lflgs;
ErtsThrPrgrVal next;
- int ix, sz, make_progress;
+ erts_aint_t refc;
- if (tpd->previous.current == ERTS_THR_PRGR_VAL_WAITING) {
- /* Took over as leader from another thread */
- tpd->previous.current = read_acqb(&erts_thr_prgr__.current);
- tpd->previous.next = tpd->previous.current;
- tpd->previous.next++;
- if (tpd->previous.next == ERTS_THR_PRGR_VAL_WAITING)
- tpd->previous.next = 0;
- }
+ my_ix = tpd->id;
- if (tpd->previous.local == tpd->previous.current) {
- ErtsThrPrgrVal val = tpd->previous.current + 1;
- if (val == ERTS_THR_PRGR_VAL_WAITING)
- val = 0;
- tpd->previous.local = val;
- set_mb(&intrnl->thr[tpd->id].data.current, val);
+ if (tpd->leader_state.current == ERTS_THR_PRGR_VAL_WAITING) {
+ /* Took over as leader from another thread */
+ tpd->leader_state.current = read_nob(&erts_thr_prgr__.current);
+ tpd->leader_state.next = tpd->leader_state.current;
+ tpd->leader_state.next++;
+ if (tpd->leader_state.next == ERTS_THR_PRGR_VAL_WAITING)
+ tpd->leader_state.next = 0;
+ tpd->leader_state.chk_next_ix = intrnl->misc.data.chk_next_ix;
+ tpd->leader_state.umrefc_ix.waiting = intrnl->misc.data.umrefc_ix.waiting;
+ tpd->leader_state.umrefc_ix.current =
+ (int) erts_atomic32_read_nob(&intrnl->misc.data.umrefc_ix.current);
+
+ if (tpd->confirmed == tpd->leader_state.current) {
+ ErtsThrPrgrVal val = tpd->leader_state.current + 1;
+ if (val == ERTS_THR_PRGR_VAL_WAITING)
+ val = 0;
+ tpd->confirmed = val;
+ set_mb(&intrnl->thr[my_ix].data.current, val);
+ }
}
- next = tpd->previous.next;
- make_progress = 1;
- sz = intrnl->managed.no;
- for (ix = 0; ix < sz; ix++) {
- ErtsThrPrgrVal tmp;
- tmp = read_nob(&intrnl->thr[ix].data.current);
- if (tmp != next && tmp != ERTS_THR_PRGR_VAL_WAITING) {
- make_progress = 0;
- ASSERT(erts_thr_progress_has_passed__(next, tmp));
- break;
+ next = tpd->leader_state.next;
+
+ waiting_unmanaged = 0;
+ umrefc_ix = -1; /* Shut up annoying warning */
+
+ chk_next_ix = tpd->leader_state.chk_next_ix;
+ no_managed = intrnl->managed.no;
+ ASSERT(0 <= chk_next_ix && chk_next_ix <= no_managed);
+ /* Check manged threads */
+ if (chk_next_ix < no_managed) {
+ for (ix = chk_next_ix; ix < no_managed; ix++) {
+ ErtsThrPrgrVal tmp;
+ if (ix == my_ix)
+ continue;
+ tmp = read_nob(&intrnl->thr[ix].data.current);
+ if (tmp != next && tmp != ERTS_THR_PRGR_VAL_WAITING) {
+ tpd->leader_state.chk_next_ix = ix;
+ ASSERT(erts_thr_progress_has_passed__(next, tmp));
+ goto done;
+ }
}
}
- if (make_progress) {
- ErtsThrPrgrVal current = next;
+ /* Check unmanged threads */
+ waiting_unmanaged = tpd->leader_state.umrefc_ix.waiting != -1;
+ umrefc_ix = (waiting_unmanaged
+ ? tpd->leader_state.umrefc_ix.waiting
+ : tpd->leader_state.umrefc_ix.current);
+ refc = erts_atomic_read_nob(&intrnl->umrefc[umrefc_ix].refc);
+ ASSERT(refc >= 0);
+ if (refc != 0) {
+ int new_umrefc_ix;
+
+ if (waiting_unmanaged)
+ goto done;
+
+ new_umrefc_ix = (umrefc_ix + 1) & 0x1;
+ tpd->leader_state.umrefc_ix.waiting = umrefc_ix;
+ tpd->leader_state.chk_next_ix = no_managed;
+ erts_atomic32_set_nob(&intrnl->misc.data.umrefc_ix.current,
+ (erts_aint32_t) new_umrefc_ix);
+ ETHR_MEMBAR(ETHR_StoreLoad);
+ refc = erts_atomic_read_nob(&intrnl->umrefc[umrefc_ix].refc);
+ ASSERT(refc >= 0);
+ waiting_unmanaged = 1;
+ if (refc != 0)
+ goto done;
+ }
+
+ /* Make progress */
+ current = next;
- next++;
- if (next == ERTS_THR_PRGR_VAL_WAITING)
- next = 0;
+ next++;
+ if (next == ERTS_THR_PRGR_VAL_WAITING)
+ next = 0;
- set_nob(&intrnl->thr[tpd->id].data.current, next);
- set_mb(&erts_thr_prgr__.current, current);
- tpd->previous.local = next;
- tpd->previous.next = next;
- tpd->previous.current = current;
+ set_nob(&intrnl->thr[my_ix].data.current, next);
+ set_mb(&erts_thr_prgr__.current, current);
+ tpd->confirmed = next;
+ tpd->leader_state.next = next;
+ tpd->leader_state.current = current;
#if ERTS_THR_PRGR_PRINT_VAL
- if (current % 1000 == 0)
- erts_fprintf(stderr, "%b64u\n", current);
+ if (current % 1000 == 0)
+ erts_fprintf(stderr, "%b64u\n", current);
#endif
- handle_wakeup_requests(current);
+ handle_wakeup_requests(current);
+
+ if (waiting_unmanaged) {
+ waiting_unmanaged = 0;
+ tpd->leader_state.umrefc_ix.waiting = -1;
+ erts_atomic32_read_band_nob(&intrnl->misc.data.lflgs,
+ ~ERTS_THR_PRGR_LFLG_WAITING_UM);
}
+ tpd->leader_state.chk_next_ix = 0;
+
+ done:
if (tpd->active) {
lflgs = erts_atomic32_read_nob(&intrnl->misc.data.lflgs);
@@ -712,20 +784,44 @@ leader_update(ErtsThrPrgrData *tpd)
(void) block_thread(tpd);
}
else {
+ int force_wakeup_check = 0;
+ erts_aint32_t set_flags = ERTS_THR_PRGR_LFLG_NO_LEADER;
tpd->leader = 0;
- tpd->previous.current = ERTS_THR_PRGR_VAL_WAITING;
+ tpd->leader_state.current = ERTS_THR_PRGR_VAL_WAITING;
#if ERTS_THR_PRGR_PRINT_LEADER
erts_fprintf(stderr, "L <- %d\n", tpd->id);
#endif
ERTS_THR_PROGRESS_STATE_DEBUG_SET_LEADER(tpd->id, 0);
+ if (waiting_unmanaged)
+ set_flags |= ERTS_THR_PRGR_LFLG_WAITING_UM;
+
lflgs = erts_atomic32_read_bor_relb(&intrnl->misc.data.lflgs,
- ERTS_THR_PRGR_LFLG_NO_LEADER);
+ set_flags);
+ lflgs |= set_flags;
if (lflgs & ERTS_THR_PRGR_LFLG_BLOCK)
lflgs = block_thread(tpd);
- if (ERTS_THR_PRGR_LFLGS_ACTIVE(lflgs) == 0 && got_sched_wakeups())
+
+ if (waiting_unmanaged) {
+ /* Need to check umrefc again */
+ ETHR_MEMBAR(ETHR_StoreLoad);
+ refc = erts_atomic_read_nob(&intrnl->umrefc[umrefc_ix].refc);
+ if (refc == 0) {
+ /* Need to force wakeup check */
+ force_wakeup_check = 1;
+ }
+ }
+
+ if ((force_wakeup_check
+ || ((lflgs & (ERTS_THR_PRGR_LFLG_NO_LEADER
+ | ERTS_THR_PRGR_LFLG_WAITING_UM
+ | ERTS_THR_PRGR_LFLG_ACTIVE_MASK))
+ == ERTS_THR_PRGR_LFLG_NO_LEADER))
+ && got_sched_wakeups()) {
+ /* Someone need to make progress */
wakeup_managed(0);
+ }
}
}
@@ -744,11 +840,11 @@ update(ErtsThrPrgrData *tpd)
erts_aint32_t lflgs;
res = 0;
val = read_acqb(&erts_thr_prgr__.current);
- if (tpd->previous.local == val) {
+ if (tpd->confirmed == val) {
val++;
if (val == ERTS_THR_PRGR_VAL_WAITING)
val = 0;
- tpd->previous.local = val;
+ tpd->confirmed = val;
set_mb(&intrnl->thr[tpd->id].data.current, val);
}
@@ -801,12 +897,19 @@ erts_thr_progress_prepare_wait(ErtsSchedulerData *esdp)
block_count_dec();
- tpd->previous.local = ERTS_THR_PRGR_VAL_WAITING;
+ tpd->confirmed = ERTS_THR_PRGR_VAL_WAITING;
set_mb(&intrnl->thr[tpd->id].data.current, ERTS_THR_PRGR_VAL_WAITING);
lflgs = erts_atomic32_read_nob(&intrnl->misc.data.lflgs);
- if (ERTS_THR_PRGR_LFLGS_ALL_WAITING(lflgs) && got_sched_wakeups())
- wakeup_managed(0); /* Someone need to make progress */
+
+ if ((lflgs & (ERTS_THR_PRGR_LFLG_NO_LEADER
+ | ERTS_THR_PRGR_LFLG_WAITING_UM
+ | ERTS_THR_PRGR_LFLG_ACTIVE_MASK))
+ == ERTS_THR_PRGR_LFLG_NO_LEADER
+ && got_sched_wakeups()) {
+ /* Someone need to make progress */
+ wakeup_managed(0);
+ }
}
void
@@ -828,7 +931,7 @@ erts_thr_progress_finalize_wait(ErtsSchedulerData *esdp)
val++;
if (val == ERTS_THR_PRGR_VAL_WAITING)
val = 0;
- tpd->previous.local = val;
+ tpd->confirmed = val;
set_mb(&intrnl->thr[tpd->id].data.current, val);
val = read_acqb(&erts_thr_prgr__.current);
if (current == val)
@@ -875,6 +978,68 @@ erts_thr_progress_active(ErtsSchedulerData *esdp, int on)
}
+static ERTS_INLINE void
+unmanaged_continue(ErtsThrPrgrDelayHandle handle)
+{
+ int umrefc_ix = (int) handle;
+ erts_aint_t refc;
+
+ ASSERT(umrefc_ix == 0 || umrefc_ix == 1);
+ refc = erts_atomic_dec_read_relb(&intrnl->umrefc[umrefc_ix].refc);
+ ASSERT(refc >= 0);
+ if (refc == 0) {
+ erts_aint_t lflgs;
+ ERTS_THR_READ_MEMORY_BARRIER;
+ lflgs = erts_atomic32_read_nob(&intrnl->misc.data.lflgs);
+ if ((lflgs & (ERTS_THR_PRGR_LFLG_NO_LEADER
+ | ERTS_THR_PRGR_LFLG_WAITING_UM
+ | ERTS_THR_PRGR_LFLG_ACTIVE_MASK))
+ == (ERTS_THR_PRGR_LFLG_NO_LEADER|ERTS_THR_PRGR_LFLG_WAITING_UM)
+ && got_sched_wakeups()) {
+ /* Others waiting for us... */
+ wakeup_managed(0);
+ }
+ }
+}
+
+void
+erts_thr_progress_unmanaged_continue__(ErtsThrPrgrDelayHandle handle)
+{
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ ErtsThrPrgrData *tpd = perhaps_thr_prgr_data(NULL);
+ ERTS_LC_ASSERT(tpd && tpd->is_delaying);
+ tpd->is_delaying = 0;
+ return_tmp_thr_prgr_data(tpd);
+#endif
+ ASSERT(!erts_thr_progress_is_managed_thread());
+
+ unmanaged_continue(handle);
+}
+
+ErtsThrPrgrDelayHandle
+erts_thr_progress_unmanaged_delay__(void)
+{
+ int umrefc_ix;
+ ASSERT(!erts_thr_progress_is_managed_thread());
+ umrefc_ix = (int) erts_atomic32_read_acqb(&intrnl->misc.data.umrefc_ix.current);
+ while (1) {
+ int tmp_ix;
+ erts_atomic_inc_acqb(&intrnl->umrefc[umrefc_ix].refc);
+ tmp_ix = (int) erts_atomic32_read_acqb(&intrnl->misc.data.umrefc_ix.current);
+ if (tmp_ix == umrefc_ix)
+ break;
+ unmanaged_continue(umrefc_ix);
+ umrefc_ix = tmp_ix;
+ }
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ {
+ ErtsThrPrgrData *tpd = tmp_thr_prgr_data(NULL);
+ tpd->is_delaying = 1;
+ }
+#endif
+ return (ErtsThrPrgrDelayHandle) umrefc_ix;
+}
+
static ERTS_INLINE int
has_reached_wakeup(ErtsThrPrgrVal wakeup)
{
@@ -931,7 +1096,7 @@ request_wakeup_managed(ErtsThrPrgrData *tpd, ErtsThrPrgrVal value)
*/
ASSERT(tpd->is_managed);
- ASSERT(tpd->previous.local != ERTS_THR_PRGR_VAL_WAITING);
+ ASSERT(tpd->confirmed != ERTS_THR_PRGR_VAL_WAITING);
if (has_reached_wakeup(value)) {
wakeup_managed(tpd->id);
@@ -946,7 +1111,7 @@ request_wakeup_managed(ErtsThrPrgrData *tpd, ErtsThrPrgrVal value)
tpd->wakeup_request[wix]));
- if (tpd->previous.local == value) {
+ if (tpd->confirmed == value) {
/*
* We have already confirmed this value. We need to request
* wakeup for a value later than our latest confirmed value in
diff --git a/erts/emulator/beam/erl_thr_progress.h b/erts/emulator/beam/erl_thr_progress.h
index e72321cf48..0e63d41fda 100644
--- a/erts/emulator/beam/erl_thr_progress.h
+++ b/erts/emulator/beam/erl_thr_progress.h
@@ -53,9 +53,22 @@ typedef Uint64 ErtsThrPrgrVal;
#define ERTS_THR_PRGR_WAKEUP_DATA_SIZE 4 /* Need to be an even power of 2. */
typedef struct {
+ ErtsThrPrgrVal next;
+ ErtsThrPrgrVal current;
+ int chk_next_ix;
+ struct {
+ int current;
+ int waiting;
+ } umrefc_ix;
+} ErtsThrPrgrLeaderState;
+
+typedef struct {
int id;
int is_managed;
int is_blocking;
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ int is_delaying; /* managed is always delaying */
+#endif
int is_temporary;
/* --- Part below only for registered threads --- */
@@ -66,11 +79,8 @@ typedef struct {
int leader; /* Needs to be first in the managed threads part */
int active;
- struct {
- ErtsThrPrgrVal local;
- ErtsThrPrgrVal next;
- ErtsThrPrgrVal current;
- } previous;
+ ErtsThrPrgrVal confirmed;
+ ErtsThrPrgrLeaderState leader_state;
} ErtsThrPrgrData;
void erts_thr_progress_fatal_error_block(SWord timeout,
@@ -121,6 +131,10 @@ typedef struct {
ERTS_THR_PRGR_ATOMIC current;
} ErtsThrPrgr;
+typedef int ErtsThrPrgrDelayHandle;
+#define ERTS_THR_PRGR_DHANDLE_MANAGED ((ErtsThrPrgrDelayHandle) -1)
+/* ERTS_THR_PRGR_DHANDLE_MANAGED implies managed thread */
+
extern ErtsThrPrgr erts_thr_prgr__;
void erts_thr_progress_pre_init(void);
@@ -136,6 +150,8 @@ int erts_thr_progress_update(ErtsSchedulerData *esdp);
int erts_thr_progress_leader_update(ErtsSchedulerData *esdp);
void erts_thr_progress_prepare_wait(ErtsSchedulerData *esdp);
void erts_thr_progress_finalize_wait(ErtsSchedulerData *esdp);
+ErtsThrPrgrDelayHandle erts_thr_progress_unmanaged_delay__(void);
+void erts_thr_progress_unmanaged_continue__(int umrefc_ix);
void erts_thr_progress_dbg_print_state(void);
@@ -148,6 +164,11 @@ ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_prgr_read_acqb__(ERTS_THR_PRGR_ATOMIC *a
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_prgr_read_mb__(ERTS_THR_PRGR_ATOMIC *atmc);
ERTS_GLB_INLINE int erts_thr_progress_is_managed_thread(void);
+ERTS_GLB_INLINE ErtsThrPrgrDelayHandle erts_thr_progress_unmanaged_delay(void);
+ERTS_GLB_INLINE void erts_thr_progress_unmanaged_continue(ErtsThrPrgrDelayHandle handle);
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ERTS_GLB_INLINE int erts_thr_progress_lc_is_delaying(void);
+#endif
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_progress_current_to_later__(ErtsThrPrgrVal val);
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_progress_later(ErtsSchedulerData *);
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_progress_current(void);
@@ -229,6 +250,35 @@ erts_thr_progress_is_managed_thread(void)
return tpd && tpd->is_managed;
}
+ERTS_GLB_INLINE ErtsThrPrgrDelayHandle
+erts_thr_progress_unmanaged_delay(void)
+{
+ if (erts_thr_progress_is_managed_thread())
+ return ERTS_THR_PRGR_DHANDLE_MANAGED; /* Nothing to do */
+ else
+ return erts_thr_progress_unmanaged_delay__();
+}
+
+ERTS_GLB_INLINE void
+erts_thr_progress_unmanaged_continue(ErtsThrPrgrDelayHandle handle)
+{
+ ASSERT(handle != ERTS_THR_PRGR_DHANDLE_MANAGED
+ || erts_thr_progress_is_managed_thread());
+ if (handle != ERTS_THR_PRGR_DHANDLE_MANAGED)
+ erts_thr_progress_unmanaged_continue__(handle);
+}
+
+#ifdef ERTS_ENABLE_LOCK_CHECK
+
+ERTS_GLB_INLINE int
+erts_thr_progress_lc_is_delaying(void)
+{
+ ErtsThrPrgrData *tpd = erts_tsd_get(erts_thr_prgr_data_key__);
+ return tpd && tpd->is_delaying;
+}
+
+#endif
+
ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_progress_current_to_later__(ErtsThrPrgrVal val)
{
@@ -248,7 +298,7 @@ erts_thr_progress_later(ErtsSchedulerData *esdp)
if (esdp) {
tpd = &esdp->thr_progress_data;
managed_thread:
- val = tpd->previous.local;
+ val = tpd->confirmed;
ERTS_THR_MEMORY_BARRIER;
}
else {