Implement erl_drv_consume_timeslice()

1 files changed, 42 insertions, 16 deletions

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));