Only verify not busy for erlang:send(Port, Msg, [nosuspend]) until scheduled

7 files changed, 104 insertions, 20 deletions

diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index 4095b88369..ad3f104a68 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -811,6 +811,7 @@ struct binary;
 #define ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT	ERTS_P2P_SIG_DATA_FLG(4)
 #define ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK	ERTS_P2P_SIG_DATA_FLG(5)
 #define ERTS_P2P_SIG_DATA_FLG_SCHED		ERTS_P2P_SIG_DATA_FLG(6)
+#define ERTS_P2P_SIG_DATA_FLG_ASYNC		ERTS_P2P_SIG_DATA_FLG(7)
 
 struct ErtsProc2PortSigData_ {
     int flags;
@@ -916,6 +917,7 @@ erts_schedule_proc2port_signal(Process *,
 			       Eterm *,
 			       ErtsProc2PortSigData *,
 			       int,
+			       ErtsPortTaskHandle *,
 			       ErtsProc2PortSigCallback);
 
 int erts_deliver_port_exit(Port *, Eterm, Eterm, int);
@@ -929,6 +931,7 @@ int erts_deliver_port_exit(Port *, Eterm, Eterm, int);
 #define ERTS_PORT_SIG_FLG_BROKEN_LINK		ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK
 #define ERTS_PORT_SIG_FLG_BAD_OUTPUT		ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT
 #define ERTS_PORT_SIG_FLG_FORCE_SCHED		ERTS_P2P_SIG_DATA_FLG_SCHED
+#define ERTS_PORT_SIG_FLG_ASYNC			ERTS_P2P_SIG_DATA_FLG_ASYNC
 /* ERTS_PORT_SIG_FLG_FORCE_IMM_CALL only when crash dumping... */
 #define ERTS_PORT_SIG_FLG_FORCE_IMM_CALL	ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT
 
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index 0ed08bee01..53cb01a8c6 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -552,6 +552,19 @@ set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp)
     }
 }
 
+static ERTS_INLINE void
+set_tmp_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp)
+{
+    ptp->u.alive.handle = NULL;
+    if (pthp) {
+	/*
+	 * IMPORTANT! Task either need to be aborted, or task handle
+	 * need to be detached before thread progress has been made.
+	 */
+	erts_smp_atomic_set_relb(pthp, (erts_aint_t) ptp);
+    }
+}
+
 
 /*
  * Busy port queue management
@@ -1182,6 +1195,13 @@ erl_drv_consume_timeslice(ErlDrvPort dprt, int percent)
     return 1;
 }
 
+void
+erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *pthp)
+{
+    ERTS_SMP_LC_ASSERT(erts_thr_progress_lc_is_delaying());
+    reset_port_task_handle(pthp);
+}
+
 /*
  * Abort a scheduled task.
  */
@@ -1206,7 +1226,7 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
 	ERTS_SMP_READ_MEMORY_BARRIER;
 	old_state = erts_smp_atomic32_read_nob(&ptp->state);
 	if (old_state == ERTS_PT_STATE_SCHEDULED) {
-	    ASSERT(saved_pthp == pthp);
+	    ASSERT(!saved_pthp || saved_pthp == pthp);
 	}
 #endif
 
@@ -1227,9 +1247,6 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
 			   &erts_port_task_outstanding_io_tasks) > 0);
 		erts_smp_atomic_dec_relb(&erts_port_task_outstanding_io_tasks);
 		break;
-	    case ERTS_PORT_TASK_PROC_SIG:
-		ERTS_INTERNAL_ERROR("Aborted process to port signal");
-		break;
 	    default:
 		break;
 	    }
@@ -1404,7 +1421,6 @@ erts_port_task_schedule(Eterm id,
     }
     case ERTS_PORT_TASK_PROC_SIG: {
 	va_list argp;
-	ASSERT(!pthp);
 	va_start(argp, type);
 	sigdp = va_arg(argp, ErtsProc2PortSigData *);
 	ptp = p2p_sig_data_to_task(sigdp);
@@ -1412,7 +1428,7 @@ erts_port_task_schedule(Eterm id,
  	ptp->u.alive.flags |= va_arg(argp, int);
 	va_end(argp);
 	if (!(ptp->u.alive.flags & ERTS_PT_FLG_NOSUSPEND))
-	    set_handle(ptp, pthp);
+	    set_tmp_handle(ptp, pthp);
 	else {
 	    ns_pthlp = erts_alloc(ERTS_ALC_T_PT_HNDL_LIST,
 				  sizeof(ErtsPortTaskHandleList));
diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index ae6cd69ae2..d35a15c27b 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -243,7 +243,9 @@ int erts_port_task_execute(ErtsRunQueue *, Port **);
 void erts_port_task_init(void);
 #endif
 
+void erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *);
 int erts_port_task_abort(ErtsPortTaskHandle *);
+
 void erts_port_task_abort_nosuspend_tasks(Port *);
 
 int erts_port_task_schedule(Eterm,
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index f9ef4eac65..d6061fb137 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -4670,6 +4670,8 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online)
 	esdp->reductions = 0;
 
 	init_sched_wall_time(&esdp->sched_wall_time);
+
+	erts_port_task_handle_init(&esdp->nosuspend_port_task_handle);
     }
 
     init_misc_aux_work();
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 3d3579fa7e..232ab55e7f 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -507,6 +507,7 @@ struct ErtsSchedulerData_ {
 
     Uint64 reductions;
     ErtsSchedWallTime sched_wall_time;
+    ErtsPortTaskHandle nosuspend_port_task_handle;
 
 #ifdef ERTS_DO_VERIFY_UNUSED_TEMP_ALLOC
     erts_alloc_verify_func_t verify_unused_temp_alloc;
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index d32e7b5d21..dba78fc770 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -1441,6 +1441,7 @@ erts_schedule_proc2port_signal(Process *c_p,
 			       Eterm *refp,
 			       ErtsProc2PortSigData *sigdp,
 			       int task_flags,
+			       ErtsPortTaskHandle *pthp,
 			       ErtsProc2PortSigCallback callback)
 {
     int sched_res;
@@ -1490,7 +1491,7 @@ erts_schedule_proc2port_signal(Process *c_p,
 
     /* Schedule port close call for later execution... */
     sched_res = erts_port_task_schedule(prt->common.id,
-					NULL,
+					pthp,
 					ERTS_PORT_TASK_PROC_SIG,
 					sigdp,
 					callback,
@@ -1628,6 +1629,7 @@ bad_port_signal(Process *c_p,
 					  refp,
 					  sigdp,
 					  0,
+					  NULL,
 					  port_badsig);
 }
 
@@ -1837,8 +1839,11 @@ erts_port_output(Process *c_p,
     ErlIOVec *evp = NULL;
     char *buf = NULL;
     int force_immediate_call = (flags & ERTS_PORT_SIG_FLG_FORCE_IMM_CALL);
+    int async_nosuspend;
+    ErtsPortTaskHandle *ns_pthp;
 
     ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP
+		      | ERTS_PORT_SIG_FLG_ASYNC
 		      | ERTS_PORT_SIG_FLG_NOSUSPEND
 		      | ERTS_PORT_SIG_FLG_FORCE
 		      | ERTS_PORT_SIG_FLG_FORCE_IMM_CALL)) == 0);
@@ -1860,6 +1865,12 @@ erts_port_output(Process *c_p,
 		? ERTS_PORT_OP_DROPPED
 		: ERTS_PORT_OP_BUSY);
 
+    async_nosuspend = ((flags & (ERTS_PORT_SIG_FLG_ASYNC
+				 | ERTS_PORT_SIG_FLG_NOSUSPEND
+				 | ERTS_PORT_SIG_FLG_FORCE))
+		      == (ERTS_PORT_SIG_FLG_ASYNC
+			  | ERTS_PORT_SIG_FLG_NOSUSPEND));
+
     try_call = (force_immediate_call /* crash dumping */
 		|| !(sched_flags & (invalid_flags
 				    | ERTS_PTS_FLGS_FORCE_SCHEDULE_OP)));
@@ -1994,6 +2005,15 @@ erts_port_output(Process *c_p,
 		return ERTS_PORT_OP_DONE;
 	    case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS:
 		sched_flags = try_call_state.sched_flags;
+		if (async_nosuspend
+		    && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) {
+		    driver_free_binary(cbin);
+		    if (evp != &ev)
+			erts_free(ERTS_ALC_T_TMP, evp);
+		    return ((sched_flags & ERTS_PTS_FLG_EXIT)
+			    ? ERTS_PORT_OP_DROPPED
+			    : ERTS_PORT_OP_BUSY);
+		}
 	    case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK:
 		/* Schedule outputv() call instead... */
 		break;
@@ -2141,6 +2161,13 @@ erts_port_output(Process *c_p,
 		return ERTS_PORT_OP_DONE;
 	    case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS:
 		sched_flags = try_call_state.sched_flags;
+		if (async_nosuspend
+		    && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) {
+		    erts_free(ERTS_ALC_T_TMP, buf);
+		    return ((sched_flags & ERTS_PTS_FLG_EXIT)
+			    ? ERTS_PORT_OP_DROPPED
+			    : ERTS_PORT_OP_BUSY);
+		}
 	    case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK:
 		/* Schedule outputv() call instead... */
 		break;
@@ -2162,20 +2189,33 @@ erts_port_output(Process *c_p,
 
     task_flags = ERTS_PT_FLG_WAIT_BUSY;
     sigdp->flags |= flags;
+    ns_pthp = NULL;
     if (flags & (ERTS_P2P_SIG_DATA_FLG_FORCE|ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)) {
 	task_flags = 0;
 	if (flags & ERTS_P2P_SIG_DATA_FLG_FORCE)
 	    sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND;
+	else if (async_nosuspend) {
+	    ErtsSchedulerData *esdp = (c_p
+				       ? ERTS_PROC_GET_SCHDATA(c_p)
+				       : erts_get_scheduler_data());
+	    ASSERT(esdp);
+	    ns_pthp = &esdp->nosuspend_port_task_handle;
+	    sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND;
+	}
 	else if (flags & ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)
 	    task_flags = ERTS_PT_FLG_NOSUSPEND;
     }
 
+    ASSERT(ns_pthp || !async_nosuspend);
+    ASSERT(async_nosuspend || !ns_pthp);
+
     res = erts_schedule_proc2port_signal(c_p,
 					 prt,
 					 c_p ? c_p->common.id : ERTS_INVALID_PID,
 					 refp,
 					 sigdp,
 					 task_flags,
+					 ns_pthp,
 					 port_sig_callback);
 
     if (res != ERTS_PORT_OP_SCHEDULED) {
@@ -2186,9 +2226,23 @@ erts_port_output(Process *c_p,
 	return res;
     }
 
-    if (!(sched_flags & ERTS_PTS_FLG_EXIT) && (sched_flags & busy_flgs))
-	return ERTS_PORT_OP_BUSY_SCHEDULED;
-
+    if (!(flags & ERTS_PORT_SIG_FLG_FORCE)) {
+	sched_flags = erts_smp_atomic32_read_acqb(&prt->sched.flags);
+	if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT)) {
+	    if (async_nosuspend)
+		erts_port_task_tmp_handle_detach(ns_pthp);
+	}
+	else {
+	    if (!async_nosuspend)
+		return ERTS_PORT_OP_BUSY_SCHEDULED;
+	    else {
+		if (erts_port_task_abort(ns_pthp) == 0)
+		    return ERTS_PORT_OP_BUSY;
+		else
+		    erts_port_task_tmp_handle_detach(ns_pthp);
+	    }
+	}
+    }
     return res;
 
 bad_value:
@@ -2284,6 +2338,7 @@ erts_port_exit(Process *c_p,
     ErlHeapFragment *bp = NULL;
 
     ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP
+		      | ERTS_PORT_SIG_FLG_ASYNC
 		      | ERTS_PORT_SIG_FLG_BROKEN_LINK
 		      | ERTS_PORT_SIG_FLG_FORCE_SCHED)) == 0);
 
@@ -2344,6 +2399,7 @@ erts_port_exit(Process *c_p,
 					 refp,
 					 sigdp,
 					 0,
+					 NULL,
 					 port_sig_exit);
 
     if (res == ERTS_PORT_OP_DROPPED) {
@@ -2459,7 +2515,8 @@ erts_port_connect(Process *c_p,
 					   !refp,
 					   am_connect);
 
-    ASSERT((flags & ~ERTS_PORT_SIG_FLG_BANG_OP) == 0);
+    ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP
+		      | ERTS_PORT_SIG_FLG_ASYNC)) == 0);
 
     if (is_not_internal_pid(connect))
 	connect_id = NIL; /* Fail in op (for signal order) */
@@ -2498,6 +2555,7 @@ erts_port_connect(Process *c_p,
 					  refp,
 					  sigdp,
 					  0,
+					  NULL,
 					  port_sig_connect);
 }
 
@@ -2554,6 +2612,7 @@ erts_port_unlink(Process *c_p, Port *prt, Eterm from, Eterm *refp)
 					  refp,
 					  sigdp,
 					  0,
+					  NULL,
 					  port_sig_unlink);
 }
 
@@ -2643,6 +2702,7 @@ erts_port_link(Process *c_p, Port *prt, Eterm to, Eterm *refp)
 					  refp,
 					  sigdp,
 					  0,
+					  NULL,
 					  port_sig_link);
 }
 
@@ -3621,6 +3681,10 @@ erts_port_command(Process *c_p,
     ASSERT(port);
 
     flags |= ERTS_PORT_SIG_FLG_BANG_OP;
+    if (!erts_port_synchronous_ops) {
+	flags |= ERTS_PORT_SIG_FLG_ASYNC;
+	refp = NULL;
+    }
 
     if (is_tuple_arity(command, 2)) {
 	Eterm cntd;
@@ -3628,21 +3692,14 @@ erts_port_command(Process *c_p,
 	cntd = tp[1];
 	if (is_internal_pid(cntd)) {
 	    if (tp[2] == am_close) {
-		if (!erts_port_synchronous_ops)
-		    refp = NULL;
 		flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
 		return erts_port_exit(c_p, flags, port, cntd, am_normal, refp);
 	    } else if (is_tuple_arity(tp[2], 2)) {
 		tp = tuple_val(tp[2]);
 		if (tp[1] == am_command) {
-		    if (!(flags & ERTS_PORT_SIG_FLG_NOSUSPEND)
-			&& !erts_port_synchronous_ops)
-			refp = NULL;
 		    return erts_port_output(c_p, flags, port, cntd, tp[2], refp);
 		}
 		else if (tp[1] == am_connect) {
-		    if (!erts_port_synchronous_ops)
-			refp = NULL;
 		    flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
 		    return erts_port_connect(c_p, flags, port, cntd, tp[2], refp);
 		}
@@ -3651,8 +3708,6 @@ erts_port_command(Process *c_p,
     }
 
     /* badsig */
-    if (!erts_port_synchronous_ops)
-	refp = NULL;
     flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
     return bad_port_signal(c_p, flags, port, c_p->common.id, refp, am_command);
 }
@@ -4049,6 +4104,7 @@ erts_port_control(Process* c_p,
 					 retvalp,
 					 sigdp,
 					 0,
+					 NULL,
 					 port_sig_control);
     if (res != ERTS_PORT_OP_SCHEDULED) {
 	cleanup_scheduled_control(binp, bufp);
@@ -4329,6 +4385,7 @@ erts_port_call(Process* c_p,
 					 retvalp,
 					 sigdp,
 					 0,
+					 NULL,
 					 port_sig_call);
     if (res != ERTS_PORT_OP_SCHEDULED) {
 	cleanup_scheduled_call(bufp);
@@ -4495,6 +4552,7 @@ erts_port_info(Process* c_p,
 					  retvalp,
 					  sigdp,
 					  0,
+					  NULL,
 					  port_sig_info);
 }
 
diff --git a/erts/emulator/test/busy_port_SUITE.erl b/erts/emulator/test/busy_port_SUITE.erl
index 2c63296b83..4b4af0babe 100644
--- a/erts/emulator/test/busy_port_SUITE.erl
+++ b/erts/emulator/test/busy_port_SUITE.erl
@@ -170,6 +170,7 @@ send_3(Config) when is_list(Config) ->
     ?line {Owner,Slave} = get_slave(),
     ?line ok = erlang:send(Slave, {Owner,{command,"set busy"}}, 
 			   [nosuspend]),
+    receive after 100 -> ok end, % ensure command reached port
     ?line nosuspend = erlang:send(Slave, {Owner,{command,"busy"}},
 				  [nosuspend]),
     ?line unlock_slave(),
@@ -563,6 +564,7 @@ scheduling_delay_busy_nosuspend(Config) ->
 	 {2,{call,[{var,1},open_port]}},
 	 {0,{cast,[{var,1},{command,1,100}]}},
 	 {0,{cast,[{var,1},{busy,2}]}},
+	 {0,{timer,sleep,[200]}}, % ensure reached port
 	 {10,{call,[{var,1},{command,3,[nosuspend]}]}},
 	 {0,{timer,sleep,[200]}},
 	 {0,{erlang,port_command,[{var,2},<<$N>>,[force]]}},