diff options
Diffstat (limited to 'erts/emulator/beam/erl_port_task.c')
-rw-r--r-- | erts/emulator/beam/erl_port_task.c | 98 |
1 files changed, 72 insertions, 26 deletions
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c index 31d9a1e26e..2aa0a27197 100644 --- a/erts/emulator/beam/erl_port_task.c +++ b/erts/emulator/beam/erl_port_task.c @@ -32,6 +32,7 @@ #include "global.h" #include "erl_port_task.h" #include "dist.h" +#include "erl_check_io.h" #include "dtrace-wrapper.h" #include <stdarg.h> @@ -68,6 +69,13 @@ static void chk_task_queues(Port *pp, ErtsPortTask *execq, int processing_busy_q #define DTRACE_DRIVER(PROBE_NAME, PP) do {} while(0) #endif +#define ERTS_SMP_LC_VERIFY_RQ(RQ, PP) \ + do { \ + ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq)); \ + ERTS_SMP_LC_ASSERT((RQ) == ((ErtsRunQueue *) \ + erts_smp_atomic_read_nob(&(PP)->run_queue))); \ + } while (0) + erts_smp_atomic_t erts_port_task_outstanding_io_tasks; #define ERTS_PT_STATE_SCHEDULED 0 @@ -543,6 +551,16 @@ reset_handle(ErtsPortTask *ptp) } static ERTS_INLINE void +reset_executed_io_task_handle(ErtsPortTask *ptp) +{ + if (ptp->u.alive.handle) { + ASSERT(ptp == handle2task(ptp->u.alive.handle)); + erts_io_notify_port_task_executed(ptp->u.alive.handle); + reset_port_task_handle(ptp->u.alive.handle); + } +} + +static ERTS_INLINE void set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp) { ptp->u.alive.handle = pthp; @@ -798,12 +816,13 @@ schedule_port_task_handle_list_free(ErtsPortTaskHandleList *pthlp) static ERTS_INLINE void abort_nosuspend_task(Port *pp, ErtsPortTaskType type, - ErtsPortTaskTypeData *tdp) + ErtsPortTaskTypeData *tdp, + int bpq_data) { ASSERT(type == ERTS_PORT_TASK_PROC_SIG); - if (!pp->sched.taskq.bpq) + if (!bpq_data) tdp->psig.callback(NULL, ERTS_PORT_SFLG_INVALID, ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND, @@ -991,6 +1010,7 @@ static ERTS_INLINE int finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q) { erts_aint32_t act; + unsigned int prof_runnable_ports; if (!processing_busy_q) pp->sched.taskq.local.first = *execq; @@ -1007,6 +1027,10 @@ finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q) if (act & ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q) act = check_unset_busy_port_q(pp, act, pp->sched.taskq.bpq); + prof_runnable_ports = erts_system_profile_flags.runnable_ports; + if (prof_runnable_ports) + erts_port_task_sched_lock(&pp->sched); + while (1) { erts_aint32_t new, exp; @@ -1018,12 +1042,24 @@ finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q) act = erts_smp_atomic32_cmpxchg_relb(&pp->sched.flags, new, exp); - ASSERT(!(act & ERTS_PTS_FLG_IN_RUNQ)); + ERTS_LC_ASSERT(!(act & ERTS_PTS_FLG_IN_RUNQ)); + ERTS_LC_ASSERT(!(act & ERTS_PTS_FLG_EXEC_IMM)); if (exp == act) break; } + if (prof_runnable_ports | IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS)) { + /* trace port scheduling, out */ + if (IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS)) + trace_sched_ports(pp, am_out); + if (prof_runnable_ports) { + if (!(act & (ERTS_PTS_FLG_EXEC_IMM|ERTS_PTS_FLG_HAVE_TASKS))) + profile_runnable_port(pp, am_inactive); + erts_port_task_sched_unlock(&pp->sched); + } + } + return (act & ERTS_PTS_FLG_HAVE_TASKS) != 0; } @@ -1345,7 +1381,7 @@ erts_port_task_abort_nosuspend_tasks(Port *pp) #endif schedule_port_task_handle_list_free(pthlp); - abort_nosuspend_task(pp, type, &td); + abort_nosuspend_task(pp, type, &td, pp->sched.taskq.bpq != NULL); } } @@ -1369,11 +1405,9 @@ erts_port_task_schedule(Eterm id, Port *pp; ErtsPortTask *ptp = NULL; erts_aint32_t act, add_flags; + unsigned int prof_runnable_ports; - if (pthp && erts_port_task_is_scheduled(pthp)) { - ASSERT(0); - erts_port_task_abort(pthp); - } + ERTS_LC_ASSERT(!pthp || !erts_port_task_is_scheduled(pthp)); ASSERT(is_internal_port(id)); @@ -1457,6 +1491,10 @@ erts_port_task_schedule(Eterm id, if (ns_pthlp) add_flags |= ERTS_PTS_FLG_HAVE_NS_TASKS; + prof_runnable_ports = erts_system_profile_flags.runnable_ports; + if (prof_runnable_ports) + erts_port_task_sched_lock(&pp->sched); + while (1) { erts_aint32_t new, exp; @@ -1481,6 +1519,13 @@ erts_port_task_schedule(Eterm id, goto done; /* Died after our task insert... */ } + if (prof_runnable_ports) { + if (!(act & ERTS_PTS_FLG_EXEC_IMM)) + profile_runnable_port(pp, am_active); + erts_port_task_sched_unlock(&pp->sched); + prof_runnable_ports = 0; + } + /* Enqueue port on run-queue */ runq = erts_port_runq(pp); @@ -1489,8 +1534,10 @@ erts_port_task_schedule(Eterm id, #ifdef ERTS_SMP xrunq = erts_check_emigration_need(runq, ERTS_PORT_PRIO_LEVEL); + ERTS_SMP_LC_ASSERT(runq != xrunq); + ERTS_SMP_LC_VERIFY_RQ(runq, pp); if (xrunq) { - /* Port emigrated ... */ + /* Emigrate port ... */ erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq); erts_smp_runq_unlock(runq); runq = erts_port_runq(pp); @@ -1500,10 +1547,6 @@ erts_port_task_schedule(Eterm id, #endif enqueue_port(runq, pp); - - if (erts_system_profile_flags.runnable_ports) { - profile_runnable_port(pp, am_active); - } erts_smp_runq_unlock(runq); @@ -1511,6 +1554,9 @@ erts_port_task_schedule(Eterm id, done: + if (prof_runnable_ports) + erts_port_task_sched_unlock(&pp->sched); + #ifdef ERTS_SMP if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED) erts_port_dec_refc(pp); @@ -1525,7 +1571,7 @@ abort_nosuspend: erts_port_dec_refc(pp); #endif - abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td); + abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td, 0); ASSERT(ns_pthlp); erts_free(ERTS_ALC_T_PT_HNDL_LIST, ns_pthlp); @@ -1609,6 +1655,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) goto done; } + ERTS_SMP_LC_VERIFY_RQ(runq, pp); + erts_smp_runq_unlock(runq); *curr_port_pp = pp; @@ -1659,8 +1707,6 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) goto aborted_port_task; } - reset_handle(ptp); - if (erts_system_monitor_long_schedule != 0) { start_time = erts_timestamp_millis(); } @@ -1671,6 +1717,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) switch (ptp->type) { case ERTS_PORT_TASK_TIMEOUT: + reset_handle(ptp); reds = ERTS_PORT_REDS_TIMEOUT; if (!(state & ERTS_PORT_SFLGS_DEAD)) { DTRACE_DRIVER(driver_timeout, pp); @@ -1685,6 +1732,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) for input and output */ (*pp->drv_ptr->ready_input)((ErlDrvData) pp->drv_data, ptp->u.alive.td.io.event); + reset_executed_io_task_handle(ptp); io_tasks_executed++; break; case ERTS_PORT_TASK_OUTPUT: @@ -1693,6 +1741,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) DTRACE_DRIVER(driver_ready_output, pp); (*pp->drv_ptr->ready_output)((ErlDrvData) pp->drv_data, ptp->u.alive.td.io.event); + reset_executed_io_task_handle(ptp); io_tasks_executed++; break; case ERTS_PORT_TASK_EVENT: @@ -1702,10 +1751,12 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) (*pp->drv_ptr->event)((ErlDrvData) pp->drv_data, ptp->u.alive.td.io.event, ptp->u.alive.td.io.event_data); + reset_executed_io_task_handle(ptp); io_tasks_executed++; break; case ERTS_PORT_TASK_PROC_SIG: { ErtsProc2PortSigData *sigdp = &ptp->u.alive.td.psig.data; + reset_handle(ptp); ASSERT((state & ERTS_PORT_SFLGS_DEAD) == 0); if (!pp->sched.taskq.bpq) reds = ptp->u.alive.td.psig.callback(pp, @@ -1723,6 +1774,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) break; } case ERTS_PORT_TASK_DIST_CMD: + reset_handle(ptp); reds = erts_dist_command(pp, CONTEXT_REDS - pp->reds); break; default: @@ -1765,10 +1817,6 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) erts_unblock_fpe(fpe_was_unmasked); - /* trace port scheduling, out */ - if (IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS)) { - trace_sched_ports(pp, am_out); - } if (io_tasks_executed) { ASSERT(erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) @@ -1791,11 +1839,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) erts_smp_runq_lock(runq); - if (!active) { - if (erts_system_profile_flags.runnable_ports) - profile_runnable_port(pp, am_inactive); - } - else { + if (active) { #ifdef ERTS_SMP ErtsRunQueue *xrunq; #endif @@ -1804,6 +1848,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) #ifdef ERTS_SMP xrunq = erts_check_emigration_need(runq, ERTS_PORT_PRIO_LEVEL); + ERTS_SMP_LC_ASSERT(runq != xrunq); + ERTS_SMP_LC_VERIFY_RQ(runq, pp); if (!xrunq) { #endif enqueue_port(runq, pp); @@ -1811,7 +1857,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) #ifdef ERTS_SMP } else { - /* Port emigrated ... */ + /* Emigrate port... */ erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq); erts_smp_runq_unlock(runq); |