From bc5818cfdd56e19a16357f4443d80a56426aa134 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Mon, 10 Oct 2011 21:03:10 +0200 Subject: Replace system block with thread progress block The ERTS internal system block functionality has been replaced by new functionality for blocking the system. The old system block functionality had contention issues and complexity issues. The new functionality piggy-backs on thread progress tracking functionality needed by newly introduced lock-free synchronization in the runtime system. When the functionality for blocking the system isn't used there is more or less no overhead at all. This since the functionality for tracking thread progress is there and needed anyway. --- erts/emulator/beam/erl_trace.c | 113 +++++++++++++++++++++++++++-------------- 1 file changed, 75 insertions(+), 38 deletions(-) (limited to 'erts/emulator/beam/erl_trace.c') diff --git a/erts/emulator/beam/erl_trace.c b/erts/emulator/beam/erl_trace.c index 8833137112..ec99f76e53 100644 --- a/erts/emulator/beam/erl_trace.c +++ b/erts/emulator/beam/erl_trace.c @@ -36,6 +36,7 @@ #include "error.h" #include "erl_binary.h" #include "erl_bits.h" +#include "erl_thr_progress.h" #if 0 #define DEBUG_PRINTOUTS @@ -159,7 +160,7 @@ static Uint active_sched; void erts_system_profile_setup_active_schedulers(void) { - ERTS_SMP_LC_ASSERT(erts_is_system_blocked(0)); + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_blocking()); active_sched = erts_active_schedulers(); } @@ -1940,7 +1941,8 @@ trace_proc(Process *c_p, Process *t_p, Eterm what, Eterm data) Eterm* hp; int need; - ERTS_SMP_LC_ASSERT((erts_proc_lc_my_proc_locks(t_p) != 0) || erts_is_system_blocked(0)); + ERTS_SMP_LC_ASSERT((erts_proc_lc_my_proc_locks(t_p) != 0) + || erts_thr_progress_is_blocking()); if (is_internal_port(t_p->tracer_proc)) { #define LOCAL_HEAP_SIZE (5+5) DeclareTmpHeapNoproc(local_heap,LOCAL_HEAP_SIZE); @@ -2745,7 +2747,8 @@ trace_port(Port *t_p, Eterm what, Eterm data) { Eterm mess; Eterm* hp; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(t_p) || erts_is_system_blocked(0)); + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(t_p) + || erts_thr_progress_is_blocking()); if (is_internal_port(t_p->tracer_proc)) { #define LOCAL_HEAP_SIZE (5+5) @@ -3021,8 +3024,6 @@ static ErtsSysMsgQ *sys_message_queue_end; static erts_tid_t sys_msg_dispatcher_tid; static erts_cnd_t smq_cnd; -static int dispatcher_waiting; - ERTS_QUALLOC_IMPL(smq_element, ErtsSysMsgQ, 20, ERTS_ALC_T_SYS_MSG_Q) static void @@ -3066,18 +3067,6 @@ enqueue_sys_msg(enum ErtsSysMsgType type, erts_smp_mtx_unlock(&smq_mtx); } -static void -prepare_for_block(void *unused) -{ - erts_smp_mtx_unlock(&smq_mtx); -} - -static void -resume_after_block(void *unused) -{ - erts_smp_mtx_lock(&smq_mtx); -} - void erts_queue_error_logger_message(Eterm from, Eterm msg, ErlHeapFragment *bp) { @@ -3143,10 +3132,10 @@ sys_msg_disp_failure(ErtsSysMsgQ *smqp, Eterm receiver) && !erts_system_monitor_flags.busy_port && !erts_system_monitor_flags.busy_dist_port) break; /* Everything is disabled */ - erts_smp_block_system(ERTS_BS_FLG_ALLOW_GC); + erts_smp_thr_progress_block(); if (system_monitor == receiver || receiver == NIL) erts_system_monitor_clear(NULL); - erts_smp_release_system(); + erts_smp_thr_progress_unblock(); break; case SYS_MSG_TYPE_SYSPROF: if (receiver == NIL @@ -3156,11 +3145,11 @@ sys_msg_disp_failure(ErtsSysMsgQ *smqp, Eterm receiver) && !erts_system_profile_flags.scheduler) break; /* Block system to clear flags */ - erts_smp_block_system(0); + erts_smp_thr_progress_block(); if (system_profile == receiver || receiver == NIL) { erts_system_profile_clear(NULL); } - erts_smp_release_system(); + erts_smp_thr_progress_unblock(); break; case SYS_MSG_TYPE_ERRLGR: { char *no_elgger = "(no error logger present)"; @@ -3201,22 +3190,68 @@ sys_msg_disp_failure(ErtsSysMsgQ *smqp, Eterm receiver) } } +static void +sys_msg_dispatcher_wakeup(void *vwait_p) +{ + int *wait_p = (int *) vwait_p; + erts_smp_mtx_lock(&smq_mtx); + *wait_p = 0; + erts_smp_cnd_signal(&smq_cnd); + erts_smp_mtx_unlock(&smq_mtx); +} + +static void +sys_msg_dispatcher_prep_wait(void *vwait_p) +{ + int *wait_p = (int *) vwait_p; + erts_smp_mtx_lock(&smq_mtx); + *wait_p = 1; + erts_smp_mtx_unlock(&smq_mtx); +} + +static void +sys_msg_dispatcher_fin_wait(void *vwait_p) +{ + int *wait_p = (int *) vwait_p; + erts_smp_mtx_lock(&smq_mtx); + *wait_p = 0; + erts_smp_mtx_unlock(&smq_mtx); +} + +static void +sys_msg_dispatcher_wait(void *vwait_p) +{ + int *wait_p = (int *) vwait_p; + erts_smp_mtx_lock(&smq_mtx); + while (*wait_p) + erts_smp_cnd_wait(&smq_cnd, &smq_mtx); + erts_smp_mtx_unlock(&smq_mtx); +} + static void * sys_msg_dispatcher_func(void *unused) { + ErtsThrPrgrCallbacks callbacks; ErtsSysMsgQ *local_sys_message_queue = NULL; + int wait = 0; #ifdef ERTS_ENABLE_LOCK_CHECK erts_lc_set_thread_name("system message dispatcher"); #endif - erts_register_blockable_thread(); - erts_smp_activity_begin(ERTS_ACTIVITY_IO, NULL, NULL, NULL); + callbacks.arg = (void *) &wait; + callbacks.wakeup = sys_msg_dispatcher_wakeup; + callbacks.prepare_wait = sys_msg_dispatcher_prep_wait; + callbacks.wait = sys_msg_dispatcher_wait; + callbacks.finalize_wait = sys_msg_dispatcher_fin_wait; + + erts_thr_progress_register_managed_thread(NULL, &callbacks, 0); while (1) { + int end_wait = 0; ErtsSysMsgQ *smqp; - ERTS_SMP_LC_ASSERT(!ERTS_LC_IS_BLOCKING); + ERTS_SMP_LC_ASSERT(!erts_thr_progress_is_blocking()); erts_smp_mtx_lock(&smq_mtx); @@ -3228,20 +3263,16 @@ sys_msg_dispatcher_func(void *unused) } /* Fetch current trace message queue ... */ - erts_smp_activity_change(ERTS_ACTIVITY_IO, - ERTS_ACTIVITY_WAIT, - prepare_for_block, - resume_after_block, - NULL); - dispatcher_waiting = 1; + if (!sys_message_queue) { + erts_smp_mtx_unlock(&smq_mtx); + end_wait = 1; + erts_thr_progress_active(NULL, 0); + erts_thr_progress_prepare_wait(NULL); + erts_smp_mtx_lock(&smq_mtx); + } + while (!sys_message_queue) erts_smp_cnd_wait(&smq_cnd, &smq_mtx); - dispatcher_waiting = 0; - erts_smp_activity_change(ERTS_ACTIVITY_WAIT, - ERTS_ACTIVITY_IO, - prepare_for_block, - resume_after_block, - NULL); local_sys_message_queue = sys_message_queue; sys_message_queue = NULL; @@ -3249,6 +3280,11 @@ sys_msg_dispatcher_func(void *unused) erts_smp_mtx_unlock(&smq_mtx); + if (end_wait) { + erts_thr_progress_finalize_wait(NULL); + erts_thr_progress_active(NULL, 1); + } + /* Send trace messages ... */ ASSERT(local_sys_message_queue); @@ -3259,6 +3295,9 @@ sys_msg_dispatcher_func(void *unused) Process *proc = NULL; Port *port = NULL; + if (erts_thr_progress_update(NULL)) + erts_thr_progress_leader_update(NULL); + #ifdef DEBUG_PRINTOUTS print_msg_type(smqp); #endif @@ -3372,7 +3411,6 @@ sys_msg_dispatcher_func(void *unused) } } - erts_smp_activity_end(ERTS_ACTIVITY_IO, NULL, NULL, NULL); return NULL; } @@ -3422,7 +3460,6 @@ init_sys_msg_dispatcher(void) sys_message_queue_end = NULL; erts_smp_cnd_init(&smq_cnd); erts_smp_mtx_init(&smq_mtx, "sys_msg_q"); - dispatcher_waiting = 0; erts_smp_thr_create(&sys_msg_dispatcher_tid, sys_msg_dispatcher_func, NULL, -- cgit v1.2.3