1 files changed, 34 insertions, 19 deletions

diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 133ab485d9..b961c639f5 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -375,7 +375,6 @@ static Port *create_port(char *name,
     prt->control_flags = 0;
     prt->bytes_in = 0;
     prt->bytes_out = 0;
-    prt->dist_entry = NULL;
     ERTS_PORT_INIT_CONNECTED(prt, pid);
     prt->common.u.alive.reg = NULL;
     ERTS_PTMR_INIT(prt);
@@ -3613,12 +3612,12 @@ terminate_port(Port *prt)
 
     erts_cleanup_port_data(prt);
 
+    ASSERT(erts_prtsd_get(prt, ERTS_PRTSD_DIST_ENTRY) == NULL);
+
     psd = (ErtsPrtSD *) erts_atomic_read_nob(&prt->psd);
     if (psd)
 	erts_free(ERTS_ALC_T_PRTSD, psd);
 
-    ASSERT(prt->dist_entry == NULL);
-
     kill_port(prt);
 
     /*
@@ -3645,20 +3644,22 @@ typedef struct {
     Eterm reason;
 } ErtsPortExitContext;
 
-static void link_port_exit(ErtsLink *lnk, void *vpectxt)
+static int link_port_exit(ErtsLink *lnk, void *vpectxt, Sint reds)
 {
     ErtsPortExitContext *pectxt = vpectxt;
     erts_proc_sig_send_link_exit(NULL, pectxt->port_id,
                                  lnk, pectxt->reason, NIL);
+    return 1;
 }
 
-static void monitor_port_exit(ErtsMonitor *mon, void *vpectxt)
+static int monitor_port_exit(ErtsMonitor *mon, void *vpectxt, Sint reds)
 {
     ErtsPortExitContext *pectxt = vpectxt;
     if (erts_monitor_is_target(mon))
         erts_proc_sig_send_monitor_down(mon, pectxt->reason);
     else
         erts_proc_sig_send_demonitor(mon);
+    return 1;
 }
 
 /* 'from' is sending 'this_port' an exit signal, (this_port must be internal).
@@ -3759,10 +3760,12 @@ erts_deliver_port_exit(Port *prt, Eterm from, Eterm reason, int send_closed,
        DRV_MONITOR_UNLOCK_PDL(prt);
    }
 
-   if ((state & ERTS_PORT_SFLG_DISTRIBUTION) && prt->dist_entry) {
-       erts_do_net_exits(prt->dist_entry, modified_reason);
-       erts_deref_dist_entry(prt->dist_entry);
-       prt->dist_entry = NULL;
+   if (state & ERTS_PORT_SFLG_DISTRIBUTION) {
+       DistEntry *dep = (DistEntry*) erts_prtsd_get(prt, ERTS_PRTSD_DIST_ENTRY);
+       ASSERT(dep);
+       erts_do_net_exits(dep, modified_reason);
+       erts_deref_dist_entry(dep);
+       erts_prtsd_set(prt, ERTS_PRTSD_DIST_ENTRY, NULL);
        erts_atomic32_read_band_relb(&prt->state,
 				    ~ERTS_PORT_SFLG_DISTRIBUTION);
    }
@@ -4072,7 +4075,7 @@ done:
  * to the caller.
  */
 int
-erl_drv_port_control(Eterm port_num, char cmd, char* buff, ErlDrvSizeT size)
+erl_drv_port_control(Eterm port_num, unsigned int cmd, char* buff, ErlDrvSizeT size)
 {
     ErtsProc2PortSigData *sigdp = erts_port_task_alloc_p2p_sig_data();
 
@@ -4835,7 +4838,7 @@ typedef struct {
     void *arg;
 } prt_one_lnk_data;
 
-static void prt_one_monitor(ErtsMonitor *mon, void *vprtd)
+static int prt_one_monitor(ErtsMonitor *mon, void *vprtd, Sint reds)
 {
     ErtsMonitorData *mdp = erts_monitor_to_data(mon);
     prt_one_lnk_data *prtd = (prt_one_lnk_data *) vprtd;
@@ -4843,12 +4846,14 @@ static void prt_one_monitor(ErtsMonitor *mon, void *vprtd)
         erts_print(prtd->to, prtd->arg, "(%p,%T)", mon->other.ptr, mdp->ref);
     else
         erts_print(prtd->to, prtd->arg, "(%T,%T)", mon->other.item, mdp->ref);
+    return 1;
 }
 
-static void prt_one_lnk(ErtsLink *lnk, void *vprtd)
+static int prt_one_lnk(ErtsLink *lnk, void *vprtd, Sint reds)
 {
     prt_one_lnk_data *prtd = (prt_one_lnk_data *) vprtd;
     erts_print(prtd->to, prtd->arg, "%T", lnk->other.item);
+    return 1;
 }
 
 static void dump_port_state(fmtfn_t to, void *arg, erts_aint32_t state)
@@ -5050,7 +5055,7 @@ set_busy_port(ErlDrvPort dprt, int on)
             DTRACE1(port_not_busy, port_str);
         }
 #endif
-	if (prt->dist_entry) {
+	if (erts_prtsd_get(prt, ERTS_PRTSD_DIST_ENTRY) != NULL) {
 	    /*
 	     * Processes suspended on distribution ports are
 	     * normally queued on the dist entry.
@@ -5099,7 +5104,7 @@ erts_port_resume_procs(Port *prt)
 
 	    erts_snprintf(port_str, sizeof(DTRACE_CHARBUF_NAME(port_str)), "%T", prt->common.id);
 	    while (plp2 != NULL) {
-		erts_snprintf(pid_str, sizeof(DTRACE_CHARBUF_NAME(pid_str)), "%T", plp2->pid);
+		erts_snprintf(pid_str, sizeof(DTRACE_CHARBUF_NAME(pid_str)), "%T", plp2->u.pid);
 		DTRACE2(process_port_unblocked, pid_str, port_str);
 	    }
 	}
@@ -6169,10 +6174,14 @@ int driver_output_binary(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
     else
 	erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + len));
     if (state & ERTS_PORT_SFLG_DISTRIBUTION) {
-        erts_atomic64_inc_nob(&prt->dist_entry->in);
+        DistEntry* dep = (DistEntry*) erts_prtsd_get(prt, ERTS_PRTSD_DIST_ENTRY);
+        Uint32 conn_id = (Uint32)(UWord) erts_prtsd_get(prt, ERTS_PRTSD_CONN_ID);
+        erts_atomic64_inc_nob(&dep->in);
 	return erts_net_message(prt,
-				prt->dist_entry,
+				dep,
+                                conn_id,
 				(byte*) hbuf, hlen,
+                                ErlDrvBinary2Binary(bin),
 				(byte*) (bin->orig_bytes+offs), len);
     }
     else
@@ -6210,16 +6219,22 @@ int driver_output2(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
     else
 	erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + len));
     if (state & ERTS_PORT_SFLG_DISTRIBUTION) {
-        erts_atomic64_inc_nob(&prt->dist_entry->in);
+        DistEntry *dep = (DistEntry*) erts_prtsd_get(prt, ERTS_PRTSD_DIST_ENTRY);
+        Uint32 conn_id = (Uint32)(UWord) erts_prtsd_get(prt, ERTS_PRTSD_CONN_ID);
+        erts_atomic64_inc_nob(&dep->in);
 	if (len == 0)
 	    return erts_net_message(prt,
-				    prt->dist_entry,
+				    dep,
+                                    conn_id,
 				    NULL, 0,
+                                    NULL,
 				    (byte*) hbuf, hlen);
 	else
 	    return erts_net_message(prt,
-				    prt->dist_entry,
+				    dep,
+                                    conn_id,
 				    (byte*) hbuf, hlen,
+                                    NULL,
 				    (byte*) buf, len);
     }
     else if (state & ERTS_PORT_SFLG_LINEBUF_IO)