From 953a4bd91e471126370bf5a70956ad233fda189a Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Fri, 8 Feb 2013 14:39:31 +0100 Subject: Implement erl_drv_consume_timeslice() --- erts/emulator/beam/bif.c | 17 +++++++++-- erts/emulator/beam/bif.h | 7 +++++ erts/emulator/beam/erl_driver.h | 5 ++++ erts/emulator/beam/erl_port.h | 1 + erts/emulator/beam/erl_port_task.c | 58 +++++++++++++++++++++++++++----------- erts/emulator/beam/io.c | 22 +++++++++++++-- 6 files changed, 89 insertions(+), 21 deletions(-) (limited to 'erts/emulator/beam') diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c index 46f76624a5..3c6cffb21e 100644 --- a/erts/emulator/beam/bif.c +++ b/erts/emulator/beam/bif.c @@ -2069,11 +2069,16 @@ BIF_RETTYPE send_3(BIF_ALIST_3) result = do_send(p, to, msg, suspend, &ref); if (result > 0) { ERTS_VBUMP_REDS(p, result); + if (ERTS_IS_PROC_OUT_OF_REDS(p)) + goto yield_return; BIF_RET(am_ok); } switch (result) { case 0: + /* May need to yield even though we do not bump reds here... */ + if (ERTS_IS_PROC_OUT_OF_REDS(p)) + goto yield_return; BIF_RET(am_ok); break; case SEND_TRAP: @@ -2091,10 +2096,10 @@ BIF_RETTYPE send_3(BIF_ALIST_3) } break; case SEND_YIELD_RETURN: - if (suspend) - ERTS_BIF_YIELD_RETURN(p, am_ok); - else + if (!suspend) BIF_RET(am_nosuspend); + yield_return: + ERTS_BIF_YIELD_RETURN(p, am_ok); case SEND_AWAIT_RESULT: ASSERT(is_internal_ref(ref)); BIF_TRAP3(await_port_send_result_trap, p, ref, am_nosuspend, am_ok); @@ -2133,11 +2138,16 @@ Eterm erl_send(Process *p, Eterm to, Eterm msg) if (result > 0) { ERTS_VBUMP_REDS(p, result); + if (ERTS_IS_PROC_OUT_OF_REDS(p)) + goto yield_return; BIF_RET(msg); } switch (result) { case 0: + /* May need to yield even though we do not bump reds here... */ + if (ERTS_IS_PROC_OUT_OF_REDS(p)) + goto yield_return; BIF_RET(msg); break; case SEND_TRAP: @@ -2147,6 +2157,7 @@ Eterm erl_send(Process *p, Eterm to, Eterm msg) ERTS_BIF_YIELD2(bif_export[BIF_send_2], p, to, msg); break; case SEND_YIELD_RETURN: + yield_return: ERTS_BIF_YIELD_RETURN(p, msg); case SEND_AWAIT_RESULT: ASSERT(is_internal_ref(ref)); diff --git a/erts/emulator/beam/bif.h b/erts/emulator/beam/bif.h index ceaf747875..51b77a95ed 100644 --- a/erts/emulator/beam/bif.h +++ b/erts/emulator/beam/bif.h @@ -35,6 +35,13 @@ extern Export* erts_format_cpu_topology_trap; #define BIF_ARG_2 (BIF__ARGS[1]) #define BIF_ARG_3 (BIF__ARGS[2]) +#define ERTS_IS_PROC_OUT_OF_REDS(p) \ + ((p)->fcalls > 0 \ + ? 0 \ + : (!ERTS_PROC_GET_SAVED_CALLS_BUF((p)) \ + ? (p)->fcalls == 0 \ + : ((p)->fcalls == -CONTEXT_REDS))) + #define BUMP_ALL_REDS(p) do { \ if (!ERTS_PROC_GET_SAVED_CALLS_BUF((p))) \ (p)->fcalls = 0; \ diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h index a9a50a10bf..e280563de1 100644 --- a/erts/emulator/beam/erl_driver.h +++ b/erts/emulator/beam/erl_driver.h @@ -407,6 +407,11 @@ EXTERN int driver_set_timer(ErlDrvPort port, unsigned long time); EXTERN int driver_cancel_timer(ErlDrvPort port); EXTERN int driver_read_timer(ErlDrvPort port, unsigned long *time_left); +/* + * Inform runtime system about lengthy work. + */ +EXTERN int erl_drv_consume_timeslice(ErlDrvPort port, int percent); + /* * Get plain-text error message from within a driver */ diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h index a971685d7b..ac4f7af5a7 100644 --- a/erts/emulator/beam/erl_port.h +++ b/erts/emulator/beam/erl_port.h @@ -175,6 +175,7 @@ struct _erl_drv_port { ErlDrvPDL port_data_lock; ErtsPrtSD *psd; /* Port specific data */ + int reds; /* Only used while executing driver callbacks */ }; #define ERTS_PORT_GET_CONNECTED(PRT) \ diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c index 9687db4780..ce045ec94e 100644 --- a/erts/emulator/beam/erl_port_task.c +++ b/erts/emulator/beam/erl_port_task.c @@ -1159,6 +1159,27 @@ select_task_for_exec(Port *pp, } } +/* + * Cut time slice + */ + +int +erl_drv_consume_timeslice(ErlDrvPort dprt, int percent) +{ + Port *pp = erts_drvport2port(dprt); + if (pp == ERTS_INVALID_ERL_DRV_PORT) + return -1; + if (percent < 1) + percent = 1; + else if (100 < percent) + percent = 100; + pp->reds += percent*((CONTEXT_REDS+99)/100); + if (pp->reds < CONTEXT_REDS) + return 0; + pp->reds = CONTEXT_REDS; + return 1; +} + /* * Abort a scheduled task. */ @@ -1550,7 +1571,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) int processing_busy_q; int res = 0; int vreds = 0; - int reds = ERTS_PORT_REDS_EXECUTE; + int reds = 0; erts_aint_t io_tasks_executed = 0; int fpe_was_unmasked; erts_aint32_t state; @@ -1595,6 +1616,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) fpe_was_unmasked = erts_block_fpe(); state = erts_atomic32_read_nob(&pp->state); + pp->reds = ERTS_PORT_REDS_EXECUTE; goto begin_handle_tasks; while (1) { @@ -1621,14 +1643,14 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) switch (ptp->type) { case ERTS_PORT_TASK_TIMEOUT: - reds += ERTS_PORT_REDS_TIMEOUT; + reds = ERTS_PORT_REDS_TIMEOUT; if (!(state & ERTS_PORT_SFLGS_DEAD)) { DTRACE_DRIVER(driver_timeout, pp); (*pp->drv_ptr->timeout)((ErlDrvData) pp->drv_data); } break; case ERTS_PORT_TASK_INPUT: - reds += ERTS_PORT_REDS_INPUT; + reds = ERTS_PORT_REDS_INPUT; ASSERT((state & ERTS_PORT_SFLGS_DEAD) == 0); DTRACE_DRIVER(driver_ready_input, pp); /* NOTE some windows drivers use ->ready_input for input and output */ @@ -1637,7 +1659,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) io_tasks_executed++; break; case ERTS_PORT_TASK_OUTPUT: - reds += ERTS_PORT_REDS_OUTPUT; + reds = ERTS_PORT_REDS_OUTPUT; ASSERT((state & ERTS_PORT_SFLGS_DEAD) == 0); DTRACE_DRIVER(driver_ready_output, pp); (*pp->drv_ptr->ready_output)((ErlDrvData) pp->drv_data, @@ -1645,7 +1667,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) io_tasks_executed++; break; case ERTS_PORT_TASK_EVENT: - reds += ERTS_PORT_REDS_EVENT; + reds = ERTS_PORT_REDS_EVENT; ASSERT((state & ERTS_PORT_SFLGS_DEAD) == 0); DTRACE_DRIVER(driver_event, pp); (*pp->drv_ptr->event)((ErlDrvData) pp->drv_data, @@ -1657,22 +1679,22 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) ErtsProc2PortSigData *sigdp = &ptp->u.alive.td.psig.data; ASSERT((state & ERTS_PORT_SFLGS_DEAD) == 0); if (!pp->sched.taskq.bpq) - reds += ptp->u.alive.td.psig.callback(pp, - state, - ERTS_PROC2PORT_SIG_EXEC, - sigdp); + reds = ptp->u.alive.td.psig.callback(pp, + state, + ERTS_PROC2PORT_SIG_EXEC, + sigdp); else { ErlDrvSizeT size = erts_proc2port_sig_command_data_size(sigdp); - reds += ptp->u.alive.td.psig.callback(pp, - state, - ERTS_PROC2PORT_SIG_EXEC, - sigdp); + reds = ptp->u.alive.td.psig.callback(pp, + state, + ERTS_PROC2PORT_SIG_EXEC, + sigdp); dequeued_proc2port_data(pp, size); } break; } case ERTS_PORT_TASK_DIST_CMD: - reds += erts_dist_command(pp, CONTEXT_REDS-reds); + reds = erts_dist_command(pp, CONTEXT_REDS - pp->reds); break; default: erl_exit(ERTS_ABORT_EXIT, @@ -1697,7 +1719,10 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) vreds += ERTS_PORT_CALLBACK_VREDS; reds += ERTS_PORT_CALLBACK_VREDS; - if (reds >= CONTEXT_REDS) + pp->reds += reds; + reds = 0; + + if (pp->reds >= CONTEXT_REDS) break; } @@ -1721,6 +1746,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) active = finalize_exec(pp, &execq, processing_busy_q); + reds = pp->reds - vreds; + erts_port_release(pp); *curr_port_pp = NULL; @@ -1766,7 +1793,6 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) != (erts_aint_t) 0); - reds -= vreds; runq->scheduler->reductions += reds; ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq)); diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index 99766b7997..b73c883658 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -1199,6 +1199,7 @@ typedef struct { int async; /* Asynchronous operation */ int pre_chk_sched_flags; /* Check sched flags before lock? */ int fpe_was_unmasked; + int reds_left_in; } ErtsTryImmDrvCallState; #define ERTS_INIT_TRY_IMM_DRV_CALL_STATE(C_P, PRT, SFLGS, PTS_FLGS, A, PRT_OP) \ @@ -1213,6 +1214,7 @@ static ERTS_INLINE ErtsTryImmDrvCallResult try_imm_drv_call(ErtsTryImmDrvCallState *sp) { ErtsTryImmDrvCallResult res; + int reds_left_in; erts_aint32_t invalid_state, invalid_sched_flags; Port *prt = sp->port; Process *c_p = sp->c_p; @@ -1246,16 +1248,24 @@ try_imm_drv_call(ErtsTryImmDrvCallState *sp) goto locked_fail; } - if (c_p) { + + if (!c_p) + reds_left_in = CONTEXT_REDS/10; + else { if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) trace_virtual_sched(c_p, am_out); if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) profile_runnable_proc(c_p, am_inactive); + reds_left_in = ERTS_BIF_REDS_LEFT(c_p); erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_MAIN); } + ASSERT(0 <= reds_left_in && reds_left_in <= CONTEXT_REDS); + sp->reds_left_in = reds_left_in; + prt->reds = CONTEXT_REDS - reds_left_in; + ERTS_SMP_CHK_NO_PROC_LOCKS; if (IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS)) @@ -1276,10 +1286,12 @@ locked_fail: static ERTS_INLINE void finalize_imm_drv_call(ErtsTryImmDrvCallState *sp) { + int reds; Port *prt = sp->port; Process *c_p = sp->c_p; - erts_port_driver_callback_epilogue(prt, NULL); + reds = prt->reds; + reds += erts_port_driver_callback_epilogue(prt, NULL); erts_unblock_fpe(sp->fpe_was_unmasked); @@ -1294,6 +1306,12 @@ finalize_imm_drv_call(ErtsTryImmDrvCallState *sp) if (c_p) { erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_MAIN); + if (reds != (CONTEXT_REDS - sp->reds_left_in)) { + int bump_reds = reds - (CONTEXT_REDS - sp->reds_left_in); + ASSERT(bump_reds > 0); + BUMP_REDS(c_p, bump_reds); + } + if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) trace_virtual_sched(c_p, am_in); if (erts_system_profile_flags.runnable_procs -- cgit v1.2.3