From d827ac847517ed43339a6a86493c7c017be782a5 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Thu, 30 Aug 2012 23:25:34 +0200 Subject: Implement functionality for delaying thread progress from unmanaged threads --- erts/emulator/beam/erl_thr_progress.c | 285 +++++++++++++++++++++++++++------- 1 file changed, 225 insertions(+), 60 deletions(-) (limited to 'erts/emulator/beam/erl_thr_progress.c') 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 -- cgit v1.2.3