Introduce support for eager check I/O scheduling

9 files changed, 103 insertions, 29 deletions

diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index bb5eba80be..0bbd3e2dd5 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -393,6 +393,7 @@ type	DRV_EV_STATE	LONG_LIVED	SYSTEM		driver_event_state
 type	DRV_EV_D_STATE	FIXED_SIZE	SYSTEM		driver_event_data_state
 type	DRV_SEL_D_STATE	FIXED_SIZE	SYSTEM		driver_select_data_state
 type	FD_LIST		SHORT_LIVED	SYSTEM		fd_list
+type	ACTIVE_FD_ARR	SHORT_LIVED	SYSTEM		active_fd_array
 type	POLLSET		LONG_LIVED	SYSTEM		pollset
 type	POLLSET_UPDREQ	SHORT_LIVED	SYSTEM		pollset_update_req
 type	POLL_FDS	LONG_LIVED	SYSTEM		poll_fds
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index d7f1e2d971..8f4095f236 100755
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -2643,6 +2643,9 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1)
     else if (ERTS_IS_ATOM_STR("ets_limit",BIF_ARG_1)) {
         BIF_RET(make_small(erts_db_get_max_tabs()));
     }
+    else if (ERTS_IS_ATOM_STR("eager_check_io",BIF_ARG_1)) {
+	BIF_RET(erts_eager_check_io ? am_true : am_false);
+    }
 
     BIF_ERROR(BIF_P, BADARG);
 }
@@ -3232,17 +3235,38 @@ BIF_RETTYPE erts_debug_get_internal_state_1(BIF_ALIST_1)
 	    BIF_RET(make_small((Uint) words));
 	}
 	else if (ERTS_IS_ATOM_STR("check_io_debug", BIF_ARG_1)) {
-	    /* Used by (emulator) */
-	    int res;
+	    /* Used by driver_SUITE (emulator) */
+	    Uint sz, *szp;
+	    Eterm res, *hp, **hpp;
+	    int no_errors;
+	    ErtsCheckIoDebugInfo ciodi = {0};
 #ifdef HAVE_ERTS_CHECK_IO_DEBUG
 	    erts_smp_proc_unlock(BIF_P,ERTS_PROC_LOCK_MAIN);
-	    res = erts_check_io_debug();
+	    no_errors = erts_check_io_debug(&ciodi);
 	    erts_smp_proc_lock(BIF_P,ERTS_PROC_LOCK_MAIN);
 #else
-	    res = 0;
-#endif
-	    ASSERT(res >= 0);
-	    BIF_RET(erts_make_integer((Uint) res, BIF_P));
+	    no_errors = 0;
+#endif
+	    sz = 0;
+	    szp = &sz;
+	    hpp = NULL;
+	    while (1) {
+		res = erts_bld_tuple(hpp, szp, 4,
+				     erts_bld_uint(hpp, szp,
+						   (Uint) no_errors),
+				     erts_bld_uint(hpp, szp,
+						   (Uint) ciodi.no_used_fds),
+				     erts_bld_uint(hpp, szp,
+						   (Uint) ciodi.no_driver_select_structs),
+				     erts_bld_uint(hpp, szp,
+						   (Uint) ciodi.no_driver_event_structs));
+		if (hpp)
+		    break;
+		hp = HAlloc(BIF_P, sz);
+		szp = NULL;
+		hpp = &hp;
+	    }
+	    BIF_RET(res);
 	}
 	else if (ERTS_IS_ATOM_STR("process_info_args", BIF_ARG_1)) {
 	    /* Used by process_SUITE (emulator) */
diff --git a/erts/emulator/beam/erl_init.c b/erts/emulator/beam/erl_init.c
index 8c4fffa75b..c3be23c84a 100644
--- a/erts/emulator/beam/erl_init.c
+++ b/erts/emulator/beam/erl_init.c
@@ -537,6 +537,8 @@ void erts_usage(void)
     erts_fprintf(stderr, "            see the erl(1) documentation for more info.\n");
     erts_fprintf(stderr, "-sct cput   set cpu topology,\n");
     erts_fprintf(stderr, "            see the erl(1) documentation for more info.\n");
+    erts_fprintf(stderr, "-secio bool enable/disable eager check I/O scheduling,\n");
+    erts_fprintf(stderr, "            see the erl(1) documentation for more info.\n");
     erts_fprintf(stderr, "-sws val    set scheduler wakeup strategy, valid values are:\n");
     erts_fprintf(stderr, "            default|legacy.\n");
     erts_fprintf(stderr, "-swct val   set scheduler wake cleanup threshold, valid values are:\n");
@@ -1487,6 +1489,19 @@ erl_start(int argc, char **argv)
 		    erts_usage();
 		}
 	    }
+	    else if (has_prefix("ecio", sub_param)) {
+		arg = get_arg(sub_param+4, argv[i+1], &i);
+		if (sys_strcmp("true", arg) == 0)
+		    erts_eager_check_io = 1;
+		else if (sys_strcmp("false", arg) == 0)
+		    erts_eager_check_io = 0;
+		else {
+		    erts_fprintf(stderr,
+				 "bad schedule eager check I/O value '%s'\n",
+				 arg);
+		    erts_usage();
+		}
+	    }
 	    else if (has_prefix("pp", sub_param)) {
 		arg = get_arg(sub_param+2, argv[i+1], &i);
 		if (sys_strcmp(arg, "true") == 0)
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index 547a42beb2..ff4fdc70aa 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>
 
@@ -543,6 +544,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;
@@ -1365,10 +1376,7 @@ erts_port_task_schedule(Eterm id,
     ErtsPortTask *ptp = NULL;
     erts_aint32_t act, add_flags;
 
-    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));
 
@@ -1654,8 +1662,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();
 	}
@@ -1666,6 +1672,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);
@@ -1679,6 +1686,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	    /* NOTE some windows drivers use ->ready_input 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:
@@ -1687,6 +1695,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:
@@ -1696,10 +1705,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,
@@ -1717,6 +1728,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:
diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index 123253a057..e1e06b565e 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -154,7 +154,7 @@ erts_port_task_handle_init(ErtsPortTaskHandle *pthp)
 ERTS_GLB_INLINE int
 erts_port_task_is_scheduled(ErtsPortTaskHandle *pthp)
 {
-    return ((void *) erts_smp_atomic_read_nob(pthp)) != NULL;
+    return ((void *) erts_smp_atomic_read_acqb(pthp)) != NULL;
 }
 
 ERTS_GLB_INLINE void erts_port_task_pre_init_sched(ErtsPortTaskSched *ptsp,
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 1efd070afd..4f3ea92cf5 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -143,6 +143,7 @@ extern BeamInstr beam_apply[];
 extern BeamInstr beam_exit[];
 extern BeamInstr beam_continue_exit[];
 
+int erts_eager_check_io = 0;
 int erts_sched_compact_load;
 Uint erts_no_schedulers;
 
@@ -1977,19 +1978,28 @@ try_set_sys_scheduling(void)
 #endif
 
 static ERTS_INLINE int
-prepare_for_sys_schedule(void)
+prepare_for_sys_schedule(int non_blocking)
 {
+    if (non_blocking && erts_eager_check_io) {
 #ifdef ERTS_SMP
-    while (!erts_port_task_have_outstanding_io_tasks()
-	   && try_set_sys_scheduling()) {
-	if (!erts_port_task_have_outstanding_io_tasks())
-	    return 1;
-	clear_sys_scheduling();
+	return try_set_sys_scheduling();
+#else
+	return 1;
+#endif
     }
-    return 0;
+    else {
+#ifdef ERTS_SMP
+	while (!erts_port_task_have_outstanding_io_tasks()
+	       && try_set_sys_scheduling()) {
+	    if (!erts_port_task_have_outstanding_io_tasks())
+		return 1;
+	    clear_sys_scheduling();
+	}
+	return 0;
 #else
-    return !erts_port_task_have_outstanding_io_tasks();
+	return !erts_port_task_have_outstanding_io_tasks();
 #endif
+    }
 }
 
 #ifdef ERTS_SMP
@@ -2307,7 +2317,7 @@ scheduler_wait(int *fcalls, ErtsSchedulerData *esdp, ErtsRunQueue *rq)
      * be waiting in erl_sys_schedule()
      */
 
-    if (!prepare_for_sys_schedule()) {
+    if (!prepare_for_sys_schedule(0)) {
 
 	sched_waiting(esdp->no, rq);
 
@@ -2459,7 +2469,7 @@ scheduler_wait(int *fcalls, ErtsSchedulerData *esdp, ErtsRunQueue *rq)
 		 * Got to check that we still got I/O tasks; otherwise
 		 * we have to continue checking for I/O...
 		 */
-		if (!prepare_for_sys_schedule()) {
+		if (!prepare_for_sys_schedule(0)) {
 		    spincount *= ERTS_SCHED_TSE_SLEEP_SPINCOUNT_FACT;
 		    goto tse_wait;
 		}
@@ -2481,7 +2491,7 @@ scheduler_wait(int *fcalls, ErtsSchedulerData *esdp, ErtsRunQueue *rq)
 	     * Got to check that we still got I/O tasks; otherwise
 	     * we have to wait in erl_sys_schedule() after all...
 	     */
-	    if (!prepare_for_sys_schedule()) {
+	    if (!prepare_for_sys_schedule(0)) {
 		/*
 		 * Not allowed to wait in erl_sys_schedule;
 		 * do tse wait instead...
@@ -7022,7 +7032,7 @@ Process *schedule(Process *p, int calls)
 
 	    goto check_activities_to_run;
 	}
-	else if (fcalls > input_reductions && prepare_for_sys_schedule()) {
+	else if (fcalls > input_reductions && prepare_for_sys_schedule(!0)) {
 	    /*
 	     * Schedule system-level activities.
 	     */
@@ -7030,8 +7040,6 @@ Process *schedule(Process *p, int calls)
 	    erts_smp_atomic32_set_relb(&function_calls, 0);
 	    fcalls = 0;
 
-	    ASSERT(!erts_port_task_have_outstanding_io_tasks());
-
 #if 0 /* Not needed since we wont wait in sys schedule */
 	    erts_sys_schedule_interrupt(0);
 #endif
@@ -7063,7 +7071,7 @@ Process *schedule(Process *p, int calls)
 	if (RUNQ_READ_LEN(&rq->ports.info.len)) {
 	    int have_outstanding_io;
 	    have_outstanding_io = erts_port_task_execute(rq, &esdp->current_port);
-	    if ((have_outstanding_io && fcalls > 2*input_reductions)
+	    if ((!erts_eager_check_io && have_outstanding_io && fcalls > 2*input_reductions)
 		|| rq->halt_in_progress) {
 		/*
 		 * If we have performed more than 2*INPUT_REDUCTIONS since
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 8d136f6e8b..41936c1bb0 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -97,6 +97,7 @@ struct saved_calls {
 };
 
 extern Export exp_send, exp_receive, exp_timeout;
+extern int erts_eager_check_io;
 extern int erts_sched_compact_load;
 extern Uint erts_no_schedulers;
 extern Uint erts_no_run_queues;
diff --git a/erts/emulator/beam/sys.h b/erts/emulator/beam/sys.h
index 31252ed78f..e691d5c55c 100644
--- a/erts/emulator/beam/sys.h
+++ b/erts/emulator/beam/sys.h
@@ -64,8 +64,12 @@
  */
 
 #ifndef ERTS_SYS_FD_TYPE
+#define ERTS_SYS_FD_INVALID ((ErtsSysFdType) -1)
 typedef int ErtsSysFdType;
 #else
+#ifndef ERTS_SYS_FD_INVALID
+# error missing ERTS_SYS_FD_INVALID
+#endif
 typedef ERTS_SYS_FD_TYPE ErtsSysFdType;
 #endif
 
@@ -731,6 +735,14 @@ void init_getenv_state(GETENV_STATE *);
 char * getenv_string(GETENV_STATE *);
 void fini_getenv_state(GETENV_STATE *);
 
+#define HAVE_ERTS_CHECK_IO_DEBUG
+typedef struct {
+    int no_used_fds;
+    int no_driver_select_structs;
+    int no_driver_event_structs;
+} ErtsCheckIoDebugInfo;
+int erts_check_io_debug(ErtsCheckIoDebugInfo *ip);
+
 /* xxxP */
 #define SYS_DEFAULT_FLOAT_DECIMALS 20
 void init_sys_float(void);
diff --git a/erts/emulator/beam/utils.c b/erts/emulator/beam/utils.c
index 0d75bbcc77..426e43d5d8 100644
--- a/erts/emulator/beam/utils.c
+++ b/erts/emulator/beam/utils.c
@@ -47,6 +47,7 @@
 #include "erl_sched_spec_pre_alloc.h"
 #include "beam_bp.h"
 #include "erl_ptab.h"
+#include "erl_check_io.h"
 
 #undef M_TRIM_THRESHOLD
 #undef M_TOP_PAD