1 files changed, 63 insertions, 23 deletions

diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index ff4fdc70aa..2aa0a27197 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -69,6 +69,13 @@ static void chk_task_queues(Port *pp, ErtsPortTask *execq, int processing_busy_q
 #define  DTRACE_DRIVER(PROBE_NAME, PP) do {} while(0)
 #endif
 
+#define ERTS_SMP_LC_VERIFY_RQ(RQ, PP)					\
+    do {								\
+	ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq));		\
+	ERTS_SMP_LC_ASSERT((RQ) == ((ErtsRunQueue *)			\
+				    erts_smp_atomic_read_nob(&(PP)->run_queue))); \
+    } while (0)
+
 erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
 
 #define ERTS_PT_STATE_SCHEDULED		0
@@ -809,12 +816,13 @@ schedule_port_task_handle_list_free(ErtsPortTaskHandleList *pthlp)
 static ERTS_INLINE void
 abort_nosuspend_task(Port *pp,
 		     ErtsPortTaskType type,
-		     ErtsPortTaskTypeData *tdp)
+		     ErtsPortTaskTypeData *tdp,
+		     int bpq_data)
 {
 
     ASSERT(type == ERTS_PORT_TASK_PROC_SIG);
 
-    if (!pp->sched.taskq.bpq)
+    if (!bpq_data)
 	tdp->psig.callback(NULL,
 			   ERTS_PORT_SFLG_INVALID,
 			   ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND,
@@ -888,6 +896,11 @@ enqueue_port(ErtsRunQueue *runq, Port *pp)
     ASSERT(runq->ports.start && runq->ports.end);
 
     erts_smp_inc_runq_len(runq, &runq->ports.info, ERTS_PORT_PRIO_LEVEL);
+
+#ifdef ERTS_SMP
+    if (runq->halt_in_progress)
+	erts_non_empty_runq(runq);
+#endif
 }
 
 static ERTS_INLINE Port *
@@ -997,6 +1010,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;
@@ -1013,6 +1027,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;
 
@@ -1024,12 +1042,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;
 }
 
@@ -1351,7 +1381,7 @@ erts_port_task_abort_nosuspend_tasks(Port *pp)
 #endif
 	schedule_port_task_handle_list_free(pthlp);
 
-	abort_nosuspend_task(pp, type, &td);
+	abort_nosuspend_task(pp, type, &td, pp->sched.taskq.bpq != NULL);
     }
 }
 
@@ -1375,6 +1405,7 @@ erts_port_task_schedule(Eterm id,
     Port *pp;
     ErtsPortTask *ptp = NULL;
     erts_aint32_t act, add_flags;
+    unsigned int prof_runnable_ports;
 
     ERTS_LC_ASSERT(!pthp || !erts_port_task_is_scheduled(pthp));
 
@@ -1460,6 +1491,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;
 
@@ -1484,6 +1519,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);
@@ -1492,8 +1534,10 @@ erts_port_task_schedule(Eterm id,
 
 #ifdef ERTS_SMP
     xrunq = erts_check_emigration_need(runq, ERTS_PORT_PRIO_LEVEL);
+    ERTS_SMP_LC_ASSERT(runq != xrunq);
+    ERTS_SMP_LC_VERIFY_RQ(runq, pp);
     if (xrunq) {
-	/* Port emigrated ... */
+	/* Emigrate port ... */
 	erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq);
 	erts_smp_runq_unlock(runq);
 	runq = erts_port_runq(pp);
@@ -1503,10 +1547,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);
 
@@ -1514,6 +1554,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);
@@ -1528,7 +1571,7 @@ abort_nosuspend:
 	erts_port_dec_refc(pp);
 #endif
 
-    abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td);
+    abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td, 0);
 
     ASSERT(ns_pthlp);
     erts_free(ERTS_ALC_T_PT_HNDL_LIST, ns_pthlp);
@@ -1612,6 +1655,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	goto done;
     }
 
+    ERTS_SMP_LC_VERIFY_RQ(runq, pp);
+
     erts_smp_runq_unlock(runq);
 
     *curr_port_pp = pp;
@@ -1683,7 +1728,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	    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 */
+	    /* NOTE some windows/ose 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);
@@ -1771,10 +1817,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)
@@ -1797,11 +1839,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
@@ -1810,6 +1848,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 
 #ifdef ERTS_SMP
 	xrunq = erts_check_emigration_need(runq, ERTS_PORT_PRIO_LEVEL);
+	ERTS_SMP_LC_ASSERT(runq != xrunq);
+	ERTS_SMP_LC_VERIFY_RQ(runq, pp);
 	if (!xrunq) {
 #endif
 	    enqueue_port(runq, pp);
@@ -1817,7 +1857,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 #ifdef ERTS_SMP
 	}
 	else {
-	    /* Port emigrated ... */
+	    /* Emigrate port... */
 	    erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq);
 	    erts_smp_runq_unlock(runq);
 
@@ -2041,9 +2081,9 @@ begin_port_cleanup(Port *pp, ErtsPortTask **execqp, int *processing_busy_q_p)
 	    DTRACE_CHARBUF(pid_str, 16);
 	    ErtsProcList* plp2 = plp;
 
-	    erts_snprintf(port_str, sizeof(port_str), "%T", pp->common.id);
+	    erts_snprintf(port_str, sizeof(DTRACE_CHARBUF_NAME(port_str)), "%T", pp->common.id);
 	    while (plp2 != NULL) {
-		erts_snprintf(pid_str, sizeof(pid_str), "%T", plp2->pid);
+		erts_snprintf(pid_str, sizeof(DTRACE_CHARBUF_NAME(pid_str)), "%T", plp2->pid);
 		DTRACE2(process_port_unblocked, pid_str, port_str);
 	    }
 	}