Implement erl_drv_consume_timeslice()

6 files changed, 89 insertions, 21 deletions

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
@@ -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