Ensure "runnable port" trace messages are not sent out of order

4 files changed, 86 insertions, 31 deletions

diff --git a/erts/emulator/beam/erl_lock_check.c b/erts/emulator/beam/erl_lock_check.c
index c665aa51a2..b105ece6f1 100644
--- a/erts/emulator/beam/erl_lock_check.c
+++ b/erts/emulator/beam/erl_lock_check.c
@@ -139,7 +139,6 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"drv_tsd",				NULL			},
     {	"async_enq_mtx",			NULL			},
 #ifdef ERTS_SMP
-    {	"sys_msg_q", 				NULL			},
     {	"atom_tab",				NULL			},
     {	"make_ref",				NULL			},
     {	"misc_op_list_pre_alloc_lock",		"address"		},
@@ -148,6 +147,7 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"btm_pre_alloc_lock",			NULL,			},
     {	"dist_entry_out_queue",			"address"		},
     {	"port_sched_lock",			"port_id"		},
+    {	"sys_msg_q", 				NULL			},
     {   "port_table",                           NULL                    },
 #endif
     {	"mtrace_op",				NULL			},
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index 4103d1192a..682f6f8f4b 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -999,6 +999,7 @@ static ERTS_INLINE int
 finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q)
 {
     erts_aint32_t act;
+    unsigned int prof_runnable_ports;
 
     if (!processing_busy_q)
 	pp->sched.taskq.local.first = *execq;
@@ -1015,6 +1016,10 @@ finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q)
     if (act & ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q)
 	act = check_unset_busy_port_q(pp, act, pp->sched.taskq.bpq);
 
+    prof_runnable_ports = erts_system_profile_flags.runnable_ports;
+    if (prof_runnable_ports)
+	erts_port_task_sched_lock(&pp->sched);
+
     while (1) {
 	erts_aint32_t new, exp;
 
@@ -1026,12 +1031,24 @@ finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q)
 
 	act = erts_smp_atomic32_cmpxchg_relb(&pp->sched.flags, new, exp);
 
-	ASSERT(!(act & ERTS_PTS_FLG_IN_RUNQ));
+	ERTS_LC_ASSERT(!(act & ERTS_PTS_FLG_IN_RUNQ));
+	ERTS_LC_ASSERT(!(act & ERTS_PTS_FLG_EXEC_IMM));
 
 	if (exp == act)
 	    break;
     }
 
+    if (prof_runnable_ports | IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS)) {
+	/* trace port scheduling, out */
+	if (IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS))
+	    trace_sched_ports(pp, am_out);
+	if (prof_runnable_ports) {
+	    if (!(act & (ERTS_PTS_FLG_EXEC_IMM|ERTS_PTS_FLG_HAVE_TASKS)))
+		profile_runnable_port(pp, am_inactive);
+	    erts_port_task_sched_unlock(&pp->sched);
+	}
+    }
+
     return (act & ERTS_PTS_FLG_HAVE_TASKS) != 0;
 }
 
@@ -1377,6 +1394,7 @@ erts_port_task_schedule(Eterm id,
     Port *pp;
     ErtsPortTask *ptp = NULL;
     erts_aint32_t act, add_flags;
+    unsigned int prof_runnable_ports;
 
     if (pthp && erts_port_task_is_scheduled(pthp)) {
 	ASSERT(0);
@@ -1465,6 +1483,10 @@ erts_port_task_schedule(Eterm id,
     if (ns_pthlp)
 	add_flags |= ERTS_PTS_FLG_HAVE_NS_TASKS;
 
+    prof_runnable_ports = erts_system_profile_flags.runnable_ports;
+    if (prof_runnable_ports)
+	erts_port_task_sched_lock(&pp->sched);
+
     while (1) {
 	erts_aint32_t new, exp;
 
@@ -1489,6 +1511,13 @@ erts_port_task_schedule(Eterm id,
 	    goto done; /* Died after our task insert... */
     }
 
+    if (prof_runnable_ports) {
+	if (!(act & ERTS_PTS_FLG_EXEC_IMM))
+	    profile_runnable_port(pp, am_active);
+	erts_port_task_sched_unlock(&pp->sched);
+	prof_runnable_ports = 0;
+    }
+
     /* Enqueue port on run-queue */
 
     runq = erts_port_runq(pp);
@@ -1510,10 +1539,6 @@ erts_port_task_schedule(Eterm id,
 #endif
 
     enqueue_port(runq, pp);
-	    
-    if (erts_system_profile_flags.runnable_ports) {
-	profile_runnable_port(pp, am_active);
-    }
 
     erts_smp_runq_unlock(runq);
 
@@ -1521,6 +1546,9 @@ erts_port_task_schedule(Eterm id,
 
 done:
 
+    if (prof_runnable_ports)
+	erts_port_task_sched_unlock(&pp->sched);
+
 #ifdef ERTS_SMP
     if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED)
 	erts_port_dec_refc(pp);
@@ -1777,10 +1805,6 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 
     erts_unblock_fpe(fpe_was_unmasked);
 
-    /* trace port scheduling, out */
-    if (IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS)) {
-    	trace_sched_ports(pp, am_out);
-    }
 
     if (io_tasks_executed) {
 	ASSERT(erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks)
@@ -1803,11 +1827,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 
     erts_smp_runq_lock(runq);
  
-    if (!active) {
-	if (erts_system_profile_flags.runnable_ports)
-	    profile_runnable_port(pp, am_inactive);
-    }
-    else {
+    if (active) {
 #ifdef ERTS_SMP
 	ErtsRunQueue *xrunq;
 #endif
diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index 1d30465ec9..9ef0cfcedc 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -78,6 +78,7 @@ extern erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
 #define ERTS_PTS_FLG_PARALLELISM		(((erts_aint32_t) 1) <<  9)
 #define ERTS_PTS_FLG_FORCE_SCHED		(((erts_aint32_t) 1) << 10)
 #define ERTS_PTS_FLG_EXITING			(((erts_aint32_t) 1) << 11)
+#define ERTS_PTS_FLG_EXEC_IMM			(((erts_aint32_t) 1) << 12)
 
 #define ERTS_PTS_FLGS_BUSY \
     (ERTS_PTS_FLG_BUSY_PORT | ERTS_PTS_FLG_BUSY_PORT_Q)
@@ -87,6 +88,7 @@ extern erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
      | ERTS_PTS_FLG_HAVE_BUSY_TASKS		\
      | ERTS_PTS_FLG_HAVE_TASKS			\
      | ERTS_PTS_FLG_EXEC			\
+     | ERTS_PTS_FLG_EXEC_IMM			\
      | ERTS_PTS_FLG_FORCE_SCHED			\
      | ERTS_PTS_FLG_EXITING)
 
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 50e08de8a4..ae053fc191 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -1218,9 +1218,10 @@ typedef struct {
 static ERTS_INLINE ErtsTryImmDrvCallResult
 try_imm_drv_call(ErtsTryImmDrvCallState *sp)
 {
+    unsigned int prof_runnable_ports;
     ErtsTryImmDrvCallResult res;
     int reds_left_in;
-    erts_aint32_t invalid_state, invalid_sched_flags;
+    erts_aint32_t act, exp, invalid_state, invalid_sched_flags;
     Port *prt = sp->port;
     Process *c_p = sp->c_p;
 
@@ -1247,11 +1248,26 @@ try_imm_drv_call(ErtsTryImmDrvCallState *sp)
 	goto locked_fail;
     }
 
-    sp->sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
-    if (sp->sched_flags & invalid_sched_flags) {
-	res = ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS;
-	goto locked_fail;
-    }
+    prof_runnable_ports = erts_system_profile_flags.runnable_ports;
+    if (prof_runnable_ports)
+	erts_port_task_sched_lock(&prt->sched);
+
+    act = erts_smp_atomic32_read_nob(&prt->sched.flags);
+
+    do {
+	erts_aint32_t new;
+	
+	if (act & invalid_sched_flags) {
+	    res = ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS;
+	    sp->sched_flags = act;
+	    goto locked_fail;
+	}
+	exp = act;
+	new = act | ERTS_PTS_FLG_EXEC_IMM;
+	act = erts_smp_atomic32_cmpxchg_mb(&prt->sched.flags, new, exp);
+    } while (act != exp);
+    
+    sp->sched_flags = act;
 
     if (!c_p)
 	reds_left_in = CONTEXT_REDS/10;
@@ -1279,11 +1295,14 @@ try_imm_drv_call(ErtsTryImmDrvCallState *sp)
 
     ERTS_SMP_CHK_NO_PROC_LOCKS;
 
-    if (IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS))
-	trace_sched_ports_where(prt, am_in, sp->port_op);
-    if (erts_system_profile_flags.runnable_ports
-	&& !erts_port_is_scheduled(prt))
-    	profile_runnable_port(prt, am_active);
+    if (prof_runnable_ports | IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS)) {
+	if (prof_runnable_ports && !(act & (ERTS_PTS_FLG_IN_RUNQ|ERTS_PTS_FLG_EXEC)))
+	    profile_runnable_port(prt, am_active);
+	if (IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS))
+	    trace_sched_ports_where(prt, am_in, sp->port_op);
+	if (prof_runnable_ports)
+	    erts_port_task_sched_unlock(&prt->sched);
+    }
 
     sp->fpe_was_unmasked = erts_block_fpe();
 
@@ -1300,17 +1319,31 @@ finalize_imm_drv_call(ErtsTryImmDrvCallState *sp)
     int reds;
     Port *prt = sp->port;
     Process *c_p = sp->c_p;
+    erts_aint32_t act;
+    unsigned int prof_runnable_ports;
 
     reds = prt->reds;
     reds += erts_port_driver_callback_epilogue(prt, NULL);
 
     erts_unblock_fpe(sp->fpe_was_unmasked);
 
-    if (IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS))
-	trace_sched_ports_where(prt, am_out, sp->port_op);
-    if (erts_system_profile_flags.runnable_ports
-	&& !erts_port_is_scheduled(prt))
-    	profile_runnable_port(prt, am_inactive);
+    prof_runnable_ports = erts_system_profile_flags.runnable_ports;
+    if (prof_runnable_ports)
+	erts_port_task_sched_lock(&prt->sched);
+
+    act = erts_smp_atomic32_read_band_mb(&prt->sched.flags,
+					 ~ERTS_PTS_FLG_EXEC_IMM);
+    ERTS_SMP_LC_ASSERT(act & ERTS_PTS_FLG_EXEC_IMM);
+
+    if (prof_runnable_ports | IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS)) {
+	if (IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS))
+	    trace_sched_ports_where(prt, am_out, sp->port_op);
+	if (prof_runnable_ports) {
+	    if (!(act & (ERTS_PTS_FLG_IN_RUNQ|ERTS_PTS_FLG_EXEC)))
+		profile_runnable_port(prt, am_inactive);
+	    erts_port_task_sched_unlock(&prt->sched);
+	}
+    }
 
     erts_port_release(prt);