aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam')
-rw-r--r--erts/emulator/beam/bif.c17
-rw-r--r--erts/emulator/beam/bif.h7
-rw-r--r--erts/emulator/beam/erl_driver.h5
-rw-r--r--erts/emulator/beam/erl_port.h1
-rw-r--r--erts/emulator/beam/erl_port_task.c58
-rw-r--r--erts/emulator/beam/io.c22
6 files changed, 89 insertions, 21 deletions
diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c
index 5adcf6d5c7..9c438679ea 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -2070,11 +2070,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:
@@ -2092,10 +2097,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);
@@ -2134,11 +2139,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:
@@ -2148,6 +2158,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
@@ -408,6 +408,11 @@ 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
*/
EXTERN char* erl_errno_id(int error);
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
@@ -1160,6 +1160,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
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-08 22:49:22 +0000