diff options
Diffstat (limited to 'erts/emulator/beam')
-rw-r--r-- | erts/emulator/beam/erl_process_lock.c | 36 | ||||
-rw-r--r-- | erts/emulator/beam/erl_thr_progress.c | 285 | ||||
-rw-r--r-- | erts/emulator/beam/erl_thr_progress.h | 62 |
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 { |