1 files changed, 56 insertions, 63 deletions

diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index e6b55c45e4..3dc7c14faf 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 2006-2011. All Rights Reserved.
+ * Copyright Ericsson AB 2006-2012. All Rights Reserved.
  *
  * The contents of this file are subject to the Erlang Public License,
  * Version 1.1, (the "License"); you may not use this file except in
@@ -32,6 +32,7 @@
 #include "global.h"
 #include "erl_port_task.h"
 #include "dist.h"
+#include "dtrace-wrapper.h"
 
 #if defined(DEBUG) && 0
 #define HARD_DEBUG
@@ -61,6 +62,20 @@ do {					\
     (P)->sched.next = NULL;		\
 } while (0)
 
+#ifdef USE_VM_PROBES
+#define DTRACE_DRIVER(PROBE_NAME, PP)                              \
+    if (DTRACE_ENABLED(driver_ready_input)) {                      \
+        DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE);         \
+        DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE);            \
+                                                                   \
+        dtrace_pid_str(PP->connected, process_str);                \
+        dtrace_port_str(PP, port_str);                             \
+        DTRACE3(PROBE_NAME, process_str, port_str, PP->name);      \
+    }
+#else
+#define  DTRACE_DRIVER(PROBE_NAME, PP) do {} while(0)
+#endif
+
 erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
 
 struct ErtsPortTaskQueue_ {
@@ -121,7 +136,7 @@ ERTS_SCHED_PREF_QUICK_ALLOC_IMPL(port_taskq,
 static ERTS_INLINE ErtsPortTask *
 handle2task(ErtsPortTaskHandle *pthp)
 {
-    return (ErtsPortTask *) erts_smp_atomic_read(pthp);
+    return (ErtsPortTask *) erts_smp_atomic_read_nob(pthp);
 }
 
 static ERTS_INLINE void
@@ -129,7 +144,7 @@ reset_handle(ErtsPortTask *ptp)
 {
     if (ptp->handle) {
 	ASSERT(ptp == handle2task(ptp->handle));
-	erts_smp_atomic_set(ptp->handle, (erts_aint_t) NULL);
+	erts_smp_atomic_set_nob(ptp->handle, (erts_aint_t) NULL);
     }
 }
 
@@ -138,7 +153,7 @@ set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp)
 {
     ptp->handle = pthp;
     if (pthp) {
-	erts_smp_atomic_set(pthp, (erts_aint_t) ptp);
+	erts_smp_atomic_set_nob(pthp, (erts_aint_t) ptp);
 	ASSERT(ptp == handle2task(ptp->handle));
     }
 }
@@ -479,8 +494,8 @@ erts_port_task_abort(Eterm id, ErtsPortTaskHandle *pthp)
     case ERTS_PORT_TASK_INPUT:
     case ERTS_PORT_TASK_OUTPUT:
     case ERTS_PORT_TASK_EVENT:
-	ASSERT(erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) > 0);
-	erts_smp_atomic_dec(&erts_port_task_outstanding_io_tasks);
+	ASSERT(erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) > 0);
+	erts_smp_atomic_dec_relb(&erts_port_task_outstanding_io_tasks);
 	break;
     default:
 	break;
@@ -558,26 +573,26 @@ erts_port_task_schedule(Eterm id,
 
     ERTS_PT_CHK_PRES_PORTQ(runq, pp);
 
-    if (!pp->sched.taskq) {
-	pp->sched.taskq = port_taskq_init(port_taskq_alloc(), pp);
-	enq_port = !pp->sched.exe_taskq;
-    }
-
+    if (!pp->sched.taskq && !pp->sched.exe_taskq) {
 #ifdef ERTS_SMP
-    if (enq_port) {
 	ErtsRunQueue *xrunq = erts_check_emigration_need(runq, ERTS_PORT_PRIO_LEVEL);
 	if (xrunq) {
 	    /* Port emigrated ... */
-	    erts_smp_atomic_set(&pp->run_queue, (erts_aint_t) xrunq);
+	    erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq);
 	    erts_smp_runq_unlock(runq);
 	    runq = xrunq;
 	}
-    }
+	enq_port = !pp->sched.taskq && !pp->sched.exe_taskq;
+#else
+	enq_port = 1;
 #endif
+    }
 
     ASSERT(!enq_port || !(runq->flags & ERTS_RUNQ_FLG_SUSPENDED));
 
-    ASSERT(pp->sched.taskq);
+    if (!pp->sched.taskq)
+	pp->sched.taskq = port_taskq_init(port_taskq_alloc(), pp);
+
     ASSERT(ptp);
 
     ptp->type = type;
@@ -594,7 +609,7 @@ erts_port_task_schedule(Eterm id,
     case ERTS_PORT_TASK_INPUT:
     case ERTS_PORT_TASK_OUTPUT:
     case ERTS_PORT_TASK_EVENT:
-	erts_smp_atomic_inc(&erts_port_task_outstanding_io_tasks);
+	erts_smp_atomic_inc_relb(&erts_port_task_outstanding_io_tasks);
 	/* Fall through... */
     default:
 	enqueue_task(pp->sched.taskq, ptp);
@@ -662,7 +677,7 @@ erts_port_task_free_port(Port *pp)
 	pp->status |= ERTS_PORT_SFLG_FREE_SCHEDULED;
 	erts_may_save_closed_port(pp);
 	erts_smp_port_state_unlock(pp);
-	ERTS_SMP_LC_ASSERT(erts_smp_atomic_read(&pp->refc) > 1);
+	ERTS_SMP_LC_ASSERT(erts_smp_atomic_read_nob(&pp->refc) > 1);
 	ptp->type = ERTS_PORT_TASK_FREE;
 	ptp->event = (ErlDrvEvent) -1;
 	ptp->event_data = NULL;
@@ -684,9 +699,9 @@ erts_port_task_free_port(Port *pp)
 	erts_may_save_closed_port(pp);
 	erts_smp_port_state_unlock(pp);
 #ifdef ERTS_SMP
-	erts_smp_atomic_dec(&pp->refc); /* Not alive */
+	erts_smp_atomic_dec_nob(&pp->refc); /* Not alive */
 #endif
-	ERTS_SMP_LC_ASSERT(erts_smp_atomic_read(&pp->refc) > 0); /* Lock */
+	ERTS_SMP_LC_ASSERT(erts_smp_atomic_read_nob(&pp->refc) > 0); /* Lock */
 	handle_remaining_tasks(runq, pp); /* May release runq lock */
 	ASSERT(!pp->sched.exe_taskq && (!ptqp || !ptqp->first));
 	pp->sched.taskq = NULL;
@@ -711,23 +726,6 @@ typedef struct {
     int *resp;
 } ErtsPortTaskExeBlockData;
 
-static void
-prepare_for_block(void *vd)
-{
-    ErtsPortTaskExeBlockData *d = (ErtsPortTaskExeBlockData *) vd;
-    erts_smp_runq_unlock(d->runq);
-}
-
-static void
-resume_after_block(void *vd)
-{
-    ErtsPortTaskExeBlockData *d = (ErtsPortTaskExeBlockData *) vd;
-    erts_smp_runq_lock(d->runq);
-    if (d->resp)
-	*d->resp = (erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks)
-		    != (erts_aint_t) 0);
-}
-
 /*
  * Run all scheduled tasks for the first port in run queue. If
  * new tasks appear while running reschedule port (free task is
@@ -748,15 +746,9 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
     int reds = ERTS_PORT_REDS_EXECUTE;
     erts_aint_t io_tasks_executed = 0;
     int fpe_was_unmasked;
-    ErtsPortTaskExeBlockData blk_data = {runq, NULL};
 
     ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq));
 
-    erts_smp_activity_begin(ERTS_ACTIVITY_IO,
-			    prepare_for_block,
-			    resume_after_block,
-			    (void *) &blk_data);
-
     ERTS_PT_CHK_PORTQ(runq);
 
     pp = pop_port(runq);
@@ -832,8 +824,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	    ASSERT(!ptqp->first
 		   && (!pp->sched.taskq || !pp->sched.taskq->first));
 #ifdef ERTS_SMP
-	    erts_smp_atomic_dec(&pp->refc); /* Not alive */
-	    ERTS_SMP_LC_ASSERT(erts_smp_atomic_read(&pp->refc) > 0); /* Lock */
+	    erts_smp_atomic_dec_nob(&pp->refc); /* Not alive */
+	    ERTS_SMP_LC_ASSERT(erts_smp_atomic_read_nob(&pp->refc) > 0); /* Lock */
 #else
 	    erts_port_status_bor_set(pp, ERTS_PORT_SFLG_FREE);	    
 #endif
@@ -846,12 +838,15 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	    goto tasks_done;
 	case ERTS_PORT_TASK_TIMEOUT:
 	    reds += ERTS_PORT_REDS_TIMEOUT;
-	    if (!(pp->status & ERTS_PORT_SFLGS_DEAD))
+	    if (!(pp->status & 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;
 	    ASSERT((pp->status & ERTS_PORT_SFLGS_DEAD) == 0);
+            DTRACE_DRIVER(driver_ready_input, pp);
 	    /* NOTE some windows drivers use ->ready_input for input and output */
 	    (*pp->drv_ptr->ready_input)((ErlDrvData) pp->drv_data, ptp->event);
 	    io_tasks_executed++;
@@ -859,12 +854,14 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	case ERTS_PORT_TASK_OUTPUT:
 	    reds += ERTS_PORT_REDS_OUTPUT;
 	    ASSERT((pp->status & ERTS_PORT_SFLGS_DEAD) == 0);
+            DTRACE_DRIVER(driver_ready_output, pp);
 	    (*pp->drv_ptr->ready_output)((ErlDrvData) pp->drv_data, ptp->event);
 	    io_tasks_executed++;
 	    break;
 	case ERTS_PORT_TASK_EVENT:
 	    reds += ERTS_PORT_REDS_EVENT;
 	    ASSERT((pp->status & ERTS_PORT_SFLGS_DEAD) == 0);
+            DTRACE_DRIVER(driver_event, pp);
 	    (*pp->drv_ptr->event)((ErlDrvData) pp->drv_data, ptp->event, ptp->event_data);
 	    io_tasks_executed++;
 	    break;
@@ -906,14 +903,16 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
     erts_unblock_fpe(fpe_was_unmasked);
 
     if (io_tasks_executed) {
-	ASSERT(erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) >= io_tasks_executed);
-	erts_smp_atomic_add(&erts_port_task_outstanding_io_tasks, -1*io_tasks_executed);
+	ASSERT(erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks)
+	       >= io_tasks_executed);
+	erts_smp_atomic_add_relb(&erts_port_task_outstanding_io_tasks,
+				 -1*io_tasks_executed);
     }
 
     *curr_port_pp = NULL;
 
 #ifdef ERTS_SMP
-    ASSERT(runq == (ErtsRunQueue *) erts_smp_atomic_read(&pp->run_queue));
+    ASSERT(runq == (ErtsRunQueue *) erts_smp_atomic_read_nob(&pp->run_queue));
 #endif
 
     if (!pp->sched.taskq) {
@@ -940,7 +939,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	}
 	else {
 	    /* Port emigrated ... */
-	    erts_smp_atomic_set(&pp->run_queue, (erts_aint_t) xrunq);
+	    erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq);
 	    enqueue_port(xrunq, pp);
 	    ASSERT(pp->sched.exe_taskq);
 	    pp->sched.exe_taskq = NULL;
@@ -951,7 +950,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	port_was_enqueued = 1;
     }
 
-    res = (erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks)
+    res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks)
 	   != (erts_aint_t) 0);
 
     ERTS_PT_CHK_PRES_PORTQ(runq, pp);
@@ -972,27 +971,22 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
     {
 	erts_aint_t refc;
 	erts_smp_mtx_unlock(pp->lock);
-	refc = erts_smp_atomic_dectest(&pp->refc);
+	refc = erts_smp_atomic_dec_read_nob(&pp->refc);
 	ASSERT(refc >= 0);
 	if (refc == 0) {
 	    erts_smp_runq_unlock(runq);
 	    erts_port_cleanup(pp); /* Might aquire runq lock */
 	    erts_smp_runq_lock(runq);
-	    res = (erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks)
+	    res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks)
 		   != (erts_aint_t) 0);
 	}
     }
 #endif
 
  done:
-    blk_data.resp = &res;
-    erts_smp_activity_end(ERTS_ACTIVITY_IO,
-			  prepare_for_block,
-			  resume_after_block,
-			  (void *) &blk_data);
+    runq->scheduler->reductions += reds;
 
     ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq));
-
     ERTS_PORT_REDUCTIONS_EXECUTED(runq, reds);
 
     return res;
@@ -1072,8 +1066,6 @@ erts_port_migrate(Port *prt, int *prt_locked,
     ERTS_SMP_LC_CHK_RUNQ_LOCK(from_rq, *from_locked);
     ERTS_SMP_LC_CHK_RUNQ_LOCK(to_rq, *to_locked);
 
-    ASSERT(!erts_common_run_queue);
-
     if (!*from_locked || !*to_locked) {
 	if (from_rq < to_rq) {
 	    if (!*to_locked) {
@@ -1107,12 +1099,12 @@ erts_port_migrate(Port *prt, int *prt_locked,
     /* Refuse to migrate to a suspended run queue */
     if (to_rq->flags & ERTS_RUNQ_FLG_SUSPENDED)
 	return ERTS_MIGRATE_FAILED_RUNQ_SUSPENDED;
-    if (from_rq != (ErtsRunQueue *) erts_smp_atomic_read(&prt->run_queue))
+    if (from_rq != (ErtsRunQueue *) erts_smp_atomic_read_nob(&prt->run_queue))
 	return ERTS_MIGRATE_FAILED_RUNQ_CHANGED;
     if (!ERTS_PORT_IS_IN_RUNQ(from_rq, prt))
 	return ERTS_MIGRATE_FAILED_NOT_IN_RUNQ;
     dequeue_port(from_rq, prt);
-    erts_smp_atomic_set(&prt->run_queue, (erts_aint_t) to_rq);
+    erts_smp_atomic_set_nob(&prt->run_queue, (erts_aint_t) to_rq);
     enqueue_port(to_rq, prt);
     return ERTS_MIGRATE_SUCCESS;
 }
@@ -1125,7 +1117,8 @@ erts_port_migrate(Port *prt, int *prt_locked,
 void
 erts_port_task_init(void)
 {
-    erts_smp_atomic_init(&erts_port_task_outstanding_io_tasks, (erts_aint_t) 0);
+    erts_smp_atomic_init_nob(&erts_port_task_outstanding_io_tasks,
+			     (erts_aint_t) 0);
     init_port_task_alloc();
     init_port_taskq_alloc();
 }