The R13B03 release.OTP_R13B03

1 files changed, 3256 insertions, 0 deletions

diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
new file mode 100644
index 0000000000..e3094404e2
--- /dev/null
+++ b/erts/emulator/beam/dist.c
@@ -0,0 +1,3256 @@
+/*
+ * %CopyrightBegin%
+ * 
+ * Copyright Ericsson AB 1996-2009. 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
+ * compliance with the License. You should have received a copy of the
+ * Erlang Public License along with this software. If not, it can be
+ * retrieved online at http://www.erlang.org/.
+ * 
+ * Software distributed under the License is distributed on an "AS IS"
+ * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+ * the License for the specific language governing rights and limitations
+ * under the License.
+ * 
+ * %CopyrightEnd%
+ */
+
+/*
+ * distribution of erlang messages to other nodes.
+ */
+
+
+/* define this to get a lot of debug output */
+/* #define ERTS_DIST_MSG_DBG */
+
+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+
+#define ERTS_WANT_EXTERNAL_TAGS
+
+#include <stddef.h>
+#include "sys.h"
+#include "erl_vm.h"
+#include "global.h"
+#include "erl_process.h"
+#include "error.h"
+#include "dist.h"
+#include "bif.h"
+#include "external.h"
+#include "erl_binary.h"
+
+/* Turn this on to get printouts of all distribution messages
+ * which go on the line
+ */
+#if 0
+#define ERTS_DIST_MSG_DBG
+#endif
+#if 0
+#define ERTS_RAW_DIST_MSG_DBG
+#endif
+
+#if defined(ERTS_DIST_MSG_DBG) || defined(ERTS_RAW_DIST_MSG_DBG)
+static void bw(byte *buf, int sz)
+{
+    bin_write(ERTS_PRINT_STDERR,NULL,buf,sz);
+}
+#endif
+
+#ifdef ERTS_DIST_MSG_DBG
+static void
+dist_msg_dbg(ErtsDistExternal *edep, char *what, byte *buf, int sz)
+{
+    byte *extp = edep->extp;
+    Eterm msg;
+    Sint size = erts_decode_dist_ext_size(edep, 0);
+    if (size < 0) {
+	erts_fprintf(stderr,
+		     "DIST MSG DEBUG: erts_decode_dist_ext_size(%s) failed:\n",
+		     what);
+	bw(buf, sz);
+    }
+    else {
+	Eterm *hp;
+	ErlHeapFragment *mbuf = new_message_buffer(size);
+	hp = mbuf->mem;
+	msg = erts_decode_dist_ext(&hp, &mbuf->off_heap, edep);
+	if (is_value(msg))
+	    erts_fprintf(stderr, "    %s: %T\n", what, msg);
+	else {
+	    erts_fprintf(stderr,
+			 "DIST MSG DEBUG: erts_decode_dist_ext(%s) failed:\n",
+			 what);
+	    bw(buf, sz);
+	}
+	free_message_buffer(mbuf);
+	edep->extp = extp;
+    }
+}
+
+#endif
+
+
+
+#define PASS_THROUGH 'p'        /* This code should go */
+
+int erts_is_alive; /* System must be blocked on change */
+
+/* distribution trap functions */
+Export* dsend2_trap = NULL;
+Export* dsend3_trap = NULL;
+/*Export* dsend_nosuspend_trap = NULL;*/
+Export* dlink_trap = NULL;
+Export* dunlink_trap = NULL;
+Export* dmonitor_node_trap = NULL;
+Export* dgroup_leader_trap = NULL;
+Export* dexit_trap = NULL;
+Export* dmonitor_p_trap = NULL;
+
+/* local variables */
+
+
+/* forward declarations */
+
+static void clear_dist_entry(DistEntry*);
+static int dsig_send(ErtsDSigData *, Eterm, Eterm, int);
+static void send_nodes_mon_msgs(Process *, Eterm, Eterm, Eterm, Eterm);
+static void init_nodes_monitors(void);
+
+static erts_smp_atomic_t no_caches;
+
+static void
+delete_cache(ErtsAtomCache *cache)
+{
+    if (cache) {
+	erts_free(ERTS_ALC_T_DCACHE, (void *) cache);
+	ASSERT(erts_smp_atomic_read(&no_caches) > 0);
+	erts_smp_atomic_dec(&no_caches);
+    }
+}
+
+
+static void
+create_cache(DistEntry *dep)
+{
+    int i;
+    ErtsAtomCache *cp;
+
+    ERTS_SMP_LC_ASSERT(
+	is_internal_port(dep->cid)
+	&& erts_lc_is_port_locked(&erts_port[internal_port_index(dep->cid)]));
+    ASSERT(!dep->cache);
+
+    dep->cache = cp = (ErtsAtomCache*) erts_alloc(ERTS_ALC_T_DCACHE,
+						  sizeof(ErtsAtomCache));
+    erts_smp_atomic_inc(&no_caches);
+    for (i = 0; i < sizeof(cp->in_arr)/sizeof(cp->in_arr[0]); i++) {
+	cp->in_arr[i] = THE_NON_VALUE;
+	cp->out_arr[i] = THE_NON_VALUE;
+    }
+}
+
+Uint erts_dist_cache_size(void)
+{
+    return (Uint) erts_smp_atomic_read(&no_caches)*sizeof(ErtsAtomCache);
+}
+
+static ErtsProcList *
+get_suspended_on_de(DistEntry *dep, Uint32 unset_qflgs)
+{
+    ERTS_SMP_LC_ASSERT(erts_smp_lc_spinlock_is_locked(&dep->qlock));
+    dep->qflgs &= ~unset_qflgs;
+    if (dep->qflgs & ERTS_DE_QFLG_EXIT) {
+	/* No resume when exit has been scheduled */
+	return NULL;
+    }
+    else {
+	ErtsProcList *plp;
+	plp = dep->suspended.first;
+	dep->suspended.first = NULL;
+	dep->suspended.last = NULL;
+	return plp;
+    }
+}
+
+/*
+** A full node name constists of a "n@h"
+**
+** n must be a valid node name: string of ([a-z][A-Z][0-9]_-)+
+** 
+** h is not checked at all, we assume that we have a properly
+** configured machine where the networking is ok for the OS
+**
+** We do check that there is not a second @ in the string, since
+** many distributed operations are guaranteed not to work then.
+*/
+
+
+static int is_node_name(char *ptr, int len)
+{
+   int c = '\0';		/* suppress use-before-set warning */
+   int pos = 0;
+
+   while (pos < len) {
+      c = ptr[pos++];
+      if (! ((c == '-') || (c == '_') ||
+	     ((c >= 'a') && (c <= 'z')) ||
+	     ((c >= 'A') && (c <= 'Z')) ||
+	     ((c >= '0') && (c <= '9'))))
+	 break;
+   }
+
+   /* Scanned past the host name: now we want to see a '@', and there
+      should be text both before and after it. */
+   if (c != '@' || pos < 2 || pos == len)
+      return 0;
+
+   while (pos < len) {
+      c = ptr[pos++];
+      if (c == '@')
+	 return 0;
+   }
+
+   return 1;
+}
+
+int is_node_name_atom(Eterm a)
+{
+    int i;
+    if(is_not_atom(a))
+	return 0;
+    i = atom_val(a);
+    ASSERT((i > 0) && (i < atom_table_size()) &&  (atom_tab(i) != NULL));
+    return is_node_name((char*)atom_tab(i)->name, atom_tab(i)->len);
+}
+
+typedef struct {
+    DistEntry *dep;
+} NetExitsContext;
+
+/* 
+** This function is called when a distribution 
+** port or process terminates
+*/
+static void doit_monitor_net_exits(ErtsMonitor *mon, void *vnecp)
+{
+    Process *rp;
+    ErtsMonitor *rmon;
+    DistEntry *dep = ((NetExitsContext *) vnecp)->dep;
+    ErtsProcLocks rp_locks = ERTS_PROC_LOCK_LINK;
+
+    rp = erts_pid2proc(NULL, 0, mon->pid, rp_locks);
+    if (!rp)
+	goto done;
+
+    if (mon->type == MON_ORIGIN) {
+	/* local pid is beeing monitored */
+	rmon = erts_remove_monitor(&(rp->monitors),mon->ref);
+	/* ASSERT(rmon != NULL); nope, can happen during process exit */
+	if (rmon != NULL) {
+	    erts_destroy_monitor(rmon);
+	}
+    } else {
+	Eterm lhp[3];
+	Eterm watched;
+	ASSERT(mon->type == MON_TARGET);
+	rmon = erts_remove_monitor(&(rp->monitors),mon->ref);
+	/* ASSERT(rmon != NULL); can happen during process exit */
+	if (rmon != NULL) {
+	    ASSERT(is_atom(rmon->name) || is_nil(rmon->name));
+	    watched = (is_atom(rmon->name)
+		       ? TUPLE2(lhp, rmon->name, dep->sysname)
+		       : rmon->pid);
+#ifdef ERTS_SMP
+	    rp_locks |= ERTS_PROC_LOCKS_MSG_SEND;
+	    erts_smp_proc_lock(rp, ERTS_PROC_LOCKS_MSG_SEND);
+#endif
+	    erts_queue_monitor_message(rp, &rp_locks, mon->ref, am_process, 
+				       watched, am_noconnection);
+	    erts_destroy_monitor(rmon);
+	}
+    }
+    erts_smp_proc_unlock(rp, rp_locks);
+ done:
+    erts_destroy_monitor(mon);
+}
+	
+typedef struct {
+    NetExitsContext *necp;
+    ErtsLink *lnk;
+} LinkNetExitsContext;
+
+/* 
+** This is the function actually doing the job of sending exit messages
+** for links in a dist entry upon net_exit (the node goes down), NB,
+** only process links, not node monitors are handled here, 
+** they reside in a separate tree....
+*/
+static void doit_link_net_exits_sub(ErtsLink *sublnk, void *vlnecp)
+{
+    ErtsLink *lnk = ((LinkNetExitsContext *) vlnecp)->lnk; /* the local pid */
+    ErtsLink *rlnk;
+    Process *rp;
+
+    ASSERT(lnk->type == LINK_PID);
+    if (is_internal_pid(lnk->pid)) {
+	int xres;
+	ErtsProcLocks rp_locks = ERTS_PROC_LOCK_LINK|ERTS_PROC_LOCKS_XSIG_SEND;
+
+	rp = erts_pid2proc(NULL, 0, lnk->pid, rp_locks);
+	if (!rp) {
+	    goto done;
+	}
+
+	rlnk = erts_remove_link(&(rp->nlinks), sublnk->pid);
+	xres = erts_send_exit_signal(NULL,
+				     sublnk->pid,
+				     rp,
+				     &rp_locks,
+				     am_noconnection,
+				     NIL,
+				     NULL,
+				     0);
+
+	if (rlnk) {
+	    erts_destroy_link(rlnk);
+	    if (xres >= 0 && IS_TRACED_FL(rp, F_TRACE_PROCS)) {
+		/* We didn't exit the process and it is traced */
+		trace_proc(NULL, rp, am_getting_unlinked, sublnk->pid);
+	    }
+	}
+	erts_smp_proc_unlock(rp, rp_locks);
+    }
+ done:
+    erts_destroy_link(sublnk);
+
+}
+    
+
+
+
+
+/* 
+** This function is called when a distribution 
+** port or process terminates, once for each link on the high level, 
+** it in turn traverses the link subtree for the specific link node...
+*/
+static void doit_link_net_exits(ErtsLink *lnk, void *vnecp)
+{
+    LinkNetExitsContext lnec = {(NetExitsContext *) vnecp, lnk};
+    ASSERT(lnk->type == LINK_PID)
+    erts_sweep_links(ERTS_LINK_ROOT(lnk), &doit_link_net_exits_sub, (void *) &lnec);
+#ifdef DEBUG
+    ERTS_LINK_ROOT(lnk) = NULL;
+#endif
+    erts_destroy_link(lnk);
+}
+
+
+static void doit_node_link_net_exits(ErtsLink *lnk, void *vnecp)
+{
+    DistEntry *dep = ((NetExitsContext *) vnecp)->dep;
+    Eterm name = dep->sysname;
+    Process *rp;
+    ErtsLink *rlnk;
+    Uint i,n;
+    ASSERT(lnk->type == LINK_NODE)
+    if (is_internal_pid(lnk->pid)) {
+	ErtsProcLocks rp_locks = ERTS_PROC_LOCK_LINK;
+	rp = erts_pid2proc(NULL, 0, lnk->pid, rp_locks);
+	if (!rp) {
+	    goto done;
+	}
+	rlnk = erts_remove_link(&(rp->nlinks), name);
+	if (rlnk != NULL) {
+	    ASSERT(is_atom(rlnk->pid) && (rlnk->type == LINK_NODE));
+	    erts_destroy_link(rlnk);
+	}
+	n = ERTS_LINK_REFC(lnk);
+	for (i = 0; i < n; ++i) {
+	    ErlHeapFragment* bp;
+	    ErlOffHeap *ohp;
+	    Eterm tup;
+	    Eterm *hp = erts_alloc_message_heap(3,&bp,&ohp,rp,&rp_locks);
+	    tup = TUPLE2(hp, am_nodedown, name);
+	    erts_queue_message(rp, &rp_locks, bp, tup, NIL);
+	}
+	erts_smp_proc_unlock(rp, rp_locks);
+    }
+ done:
+    erts_destroy_link(lnk);
+}
+
+	
+/*
+ * proc is currently running or exiting process.
+ */
+int erts_do_net_exits(DistEntry *dep, Eterm reason)
+{
+    Eterm nodename;
+
+    if (dep == erts_this_dist_entry) {  /* Net kernel has died (clean up!!) */
+	Eterm nd_reason = (reason == am_no_network
+			   ? am_no_network
+			   : am_net_kernel_terminated);
+	erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+
+	/* KILL all port controllers */
+	while(erts_visible_dist_entries || erts_hidden_dist_entries) {
+	    DistEntry *tdep;
+	    Eterm prt_id;
+	    Port *prt;
+	    if(erts_hidden_dist_entries)
+		tdep = erts_hidden_dist_entries;
+	    else
+		tdep = erts_visible_dist_entries;
+	    prt_id = tdep->cid;
+	    ASSERT(is_internal_port(prt_id));
+	    erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+
+	    prt = erts_id2port(prt_id, NULL, 0);
+	    if (prt) {
+		ASSERT(prt->status & ERTS_PORT_SFLG_DISTRIBUTION);
+		ASSERT(prt->dist_entry);
+		/* will call do_net_exists !!! */
+		erts_do_exit_port(prt, prt_id, nd_reason);
+		erts_port_release(prt);
+	    }
+
+	    erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+	}
+
+	erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+
+	nodename = erts_this_dist_entry->sysname;
+	erts_smp_block_system(ERTS_BS_FLG_ALLOW_GC);
+	erts_set_this_node(am_Noname, 0);
+	erts_is_alive = 0;
+	send_nodes_mon_msgs(NULL, am_nodedown, nodename, am_visible, nd_reason);
+	erts_smp_release_system();
+
+    }
+    else { /* recursive call via erts_do_exit_port() will end up here */
+	NetExitsContext nec = {dep};
+	ErtsLink *nlinks;
+	ErtsLink *node_links;
+	ErtsMonitor *monitors;
+	Uint32 flags;
+
+	erts_smp_atomic_set(&dep->dist_cmd_scheduled, 1);
+	erts_smp_de_rwlock(dep);
+
+	ERTS_SMP_LC_ASSERT(is_internal_port(dep->cid)
+			   && erts_lc_is_port_locked(&erts_port[internal_port_index(dep->cid)]));
+
+	if (erts_port_task_is_scheduled(&dep->dist_cmd))
+	    erts_port_task_abort(dep->cid, &dep->dist_cmd);
+
+	if (dep->status & ERTS_DE_SFLG_EXITING) {
+#ifdef DEBUG
+	    erts_smp_spin_lock(&dep->qlock);
+	    ASSERT(dep->qflgs & ERTS_DE_QFLG_EXIT);
+	    erts_smp_spin_unlock(&dep->qlock);
+#endif
+	}
+	else {
+	    dep->status |= ERTS_DE_SFLG_EXITING;
+	    erts_smp_spin_lock(&dep->qlock);
+	    ASSERT(!(dep->qflgs & ERTS_DE_QFLG_EXIT));
+	    dep->qflgs |= ERTS_DE_QFLG_EXIT;
+	    erts_smp_spin_unlock(&dep->qlock);
+	}
+
+	erts_smp_de_links_lock(dep);
+	monitors	= dep->monitors;
+        nlinks		= dep->nlinks;
+	node_links	= dep->node_links;
+	dep->monitors	= NULL;
+        dep->nlinks	= NULL;
+	dep->node_links	= NULL;
+	erts_smp_de_links_unlock(dep);
+
+	nodename = dep->sysname;
+	flags = dep->flags;
+
+	erts_set_dist_entry_not_connected(dep);
+
+	erts_smp_de_rwunlock(dep);
+
+	erts_sweep_monitors(monitors, &doit_monitor_net_exits, (void *) &nec);
+	erts_sweep_links(nlinks, &doit_link_net_exits, (void *) &nec);
+	erts_sweep_links(node_links, &doit_node_link_net_exits, (void *) &nec);
+
+	send_nodes_mon_msgs(NULL,
+			    am_nodedown,
+			    nodename,
+			    flags & DFLAG_PUBLISHED ? am_visible : am_hidden,
+			    reason == am_normal ? am_connection_closed : reason);
+
+	clear_dist_entry(dep);
+
+    }
+    return 1;
+}
+
+static Export*
+trap_function(Eterm func, int arity)
+{
+    return erts_export_put(am_erlang, func, arity);
+}
+
+void init_dist(void)
+{
+    init_nodes_monitors();
+
+    erts_smp_atomic_init(&no_caches, 0);
+
+    /* Lookup/Install all references to trap functions */
+    dsend2_trap = trap_function(am_dsend,2);
+    dsend3_trap = trap_function(am_dsend,3);
+    /*    dsend_nosuspend_trap = trap_function(am_dsend_nosuspend,2);*/
+    dlink_trap = trap_function(am_dlink,1);
+    dunlink_trap = trap_function(am_dunlink,1);
+    dmonitor_node_trap = trap_function(am_dmonitor_node,3);
+    dgroup_leader_trap = trap_function(am_dgroup_leader,2);
+    dexit_trap = trap_function(am_dexit, 2);
+    dmonitor_p_trap = trap_function(am_dmonitor_p, 2);
+}
+
+#define ErtsDistOutputBuf2Binary(OB) \
+  ((Binary *) (((char *) (OB)) - offsetof(Binary, orig_bytes)))
+
+static ERTS_INLINE ErtsDistOutputBuf *
+alloc_dist_obuf(Uint size)
+{
+    ErtsDistOutputBuf *obuf;
+    Uint obuf_size = sizeof(ErtsDistOutputBuf)+sizeof(byte)*(size-1);
+    Binary *bin = erts_bin_drv_alloc(obuf_size);
+    bin->flags = BIN_FLAG_DRV;
+    erts_refc_init(&bin->refc, 1);
+    bin->orig_size = (long) obuf_size;
+    obuf = (ErtsDistOutputBuf *) &bin->orig_bytes[0];
+#ifdef DEBUG
+    obuf->dbg_pattern = ERTS_DIST_OUTPUT_BUF_DBG_PATTERN;
+    ASSERT(bin == ErtsDistOutputBuf2Binary(obuf));
+#endif
+    return obuf;
+}
+
+static ERTS_INLINE void
+free_dist_obuf(ErtsDistOutputBuf *obuf)
+{
+    Binary *bin = ErtsDistOutputBuf2Binary(obuf);
+    ASSERT(obuf->dbg_pattern == ERTS_DIST_OUTPUT_BUF_DBG_PATTERN);
+    if (erts_refc_dectest(&bin->refc, 0) == 0)
+	erts_bin_free(bin);
+}
+
+static ERTS_INLINE Sint
+size_obuf(ErtsDistOutputBuf *obuf)
+{
+    Binary *bin = ErtsDistOutputBuf2Binary(obuf);
+    return bin->orig_size;
+}
+
+static void clear_dist_entry(DistEntry *dep)
+{
+    Sint obufsize = 0;
+    ErtsAtomCache *cache;
+    ErtsProcList *suspendees;
+    ErtsDistOutputBuf *obuf;
+
+    erts_smp_de_rwlock(dep);
+    cache = dep->cache;
+    dep->cache = NULL;
+
+#ifdef DEBUG
+    erts_smp_de_links_lock(dep);
+    ASSERT(!dep->nlinks);
+    ASSERT(!dep->node_links);
+    ASSERT(!dep->monitors);
+    erts_smp_de_links_unlock(dep);
+#endif
+
+    erts_smp_spin_lock(&dep->qlock);
+
+    if (!dep->out_queue.last)
+	obuf = dep->finalized_out_queue.first;
+    else {
+	dep->out_queue.last->next = dep->finalized_out_queue.first;
+	obuf = dep->out_queue.first;
+    }
+
+    dep->out_queue.first = NULL;
+    dep->out_queue.last = NULL;
+    dep->finalized_out_queue.first = NULL;
+    dep->finalized_out_queue.last = NULL;
+    dep->status = 0;
+    suspendees = get_suspended_on_de(dep, ERTS_DE_QFLGS_ALL);
+
+    erts_smp_spin_unlock(&dep->qlock);
+    erts_smp_atomic_set(&dep->dist_cmd_scheduled, 0);
+    dep->send = NULL;
+    erts_smp_de_rwunlock(dep);
+
+    erts_resume_processes(suspendees);
+
+    delete_cache(cache);
+
+    while (obuf) {
+	ErtsDistOutputBuf *fobuf;
+	fobuf = obuf;
+	obuf = obuf->next;
+	obufsize += size_obuf(fobuf);
+	free_dist_obuf(fobuf);
+    }
+
+    if (obufsize) {
+	erts_smp_spin_lock(&dep->qlock);
+	ASSERT(dep->qsize >= obufsize);
+	dep->qsize -= obufsize;
+	erts_smp_spin_unlock(&dep->qlock);
+    }
+}
+
+/*
+ * The erts_dsig_send_*() functions implemented below, sends asynchronous
+ * distributed signals to other Erlang nodes. Before sending a distributed
+ * signal, you need to prepare the operation by calling erts_dsig_prepare()
+ * (see dist.h).
+ *
+ * Note that the distributed signal send operation is truly asynchronous,
+ * and the signal is not guaranteed to reach the receiver if the connection
+ * goes down before the signal has reached the receiver.
+ */
+
+/*
+** Send a DOP_LINK link message
+*/
+int
+erts_dsig_send_link(ErtsDSigData *dsdp, Eterm local, Eterm remote)
+{
+    Eterm ctl_heap[4];
+    Eterm ctl = TUPLE3(&ctl_heap[0], make_small(DOP_LINK), local, remote);
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+}
+
+int
+erts_dsig_send_unlink(ErtsDSigData *dsdp, Eterm local, Eterm remote)
+{
+    Eterm ctl_heap[4];
+    Eterm ctl = TUPLE3(&ctl_heap[0], make_small(DOP_UNLINK), local, remote);
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+}
+
+
+/* A local process that's beeing monitored by a remote one exits. We send:
+   {DOP_MONITOR_P_EXIT, Local pid or name, Remote pid, ref, reason},
+   which is rather sad as only the ref is needed, no pid's... */
+int
+erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm watcher, Eterm watched, 
+			  Eterm ref, Eterm reason)
+{
+    Eterm ctl;
+    Eterm ctl_heap[6];
+
+    ctl = TUPLE5(&ctl_heap[0], make_small(DOP_MONITOR_P_EXIT),
+		 watched, watcher, ref, reason);
+
+#ifdef DEBUG
+    erts_smp_de_links_lock(dsdp->dep);
+    ASSERT(!erts_lookup_monitor(dsdp->dep->monitors, ref));
+    erts_smp_de_links_unlock(dsdp->dep);
+#endif
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 1);
+}
+
+/* We want to monitor a process (named or unnamed) on another node, we send:
+   {DOP_MONITOR_P, Local pid, Remote pid or name, Ref}, which is exactly what's
+   needed on the other side... */
+int
+erts_dsig_send_monitor(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
+		       Eterm ref)
+{
+    Eterm ctl;
+    Eterm ctl_heap[5];
+
+    ctl = TUPLE4(&ctl_heap[0],
+		 make_small(DOP_MONITOR_P),
+		 watcher, watched, ref);
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+}
+
+/* A local process monitoring a remote one wants to stop monitoring, either 
+   because of a demonitor bif call or because the local process died. We send
+   {DOP_DEMONITOR_P, Local pid, Remote pid or name, ref}, which is once again
+   rather redundant as only the ref will be needed on the other side... */
+int
+erts_dsig_send_demonitor(ErtsDSigData *dsdp, Eterm watcher,
+			 Eterm watched, Eterm ref, int force)
+{
+    Eterm ctl;
+    Eterm ctl_heap[5];
+
+    ctl = TUPLE4(&ctl_heap[0],
+		 make_small(DOP_DEMONITOR_P),
+		 watcher, watched, ref);
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, force);
+}
+
+int
+erts_dsig_send_msg(ErtsDSigData *dsdp, Eterm remote, Eterm message)
+{
+    Eterm ctl;
+    Eterm ctl_heap[5];
+    Eterm token = NIL;
+    Process *sender = dsdp->proc;
+
+    if (SEQ_TRACE_TOKEN(sender) != NIL) {
+	seq_trace_update_send(sender);
+	token = SEQ_TRACE_TOKEN(sender);
+	seq_trace_output(token, message, SEQ_TRACE_SEND, remote, sender);
+    }
+
+    if (token != NIL)
+	ctl = TUPLE4(&ctl_heap[0],
+		     make_small(DOP_SEND_TT), am_Cookie, remote, token);
+    else
+	ctl = TUPLE3(&ctl_heap[0], make_small(DOP_SEND), am_Cookie, remote);
+    return dsig_send(dsdp, ctl, message, 0);
+}
+
+int
+erts_dsig_send_reg_msg(ErtsDSigData *dsdp, Eterm remote_name, Eterm message)
+{
+    Eterm ctl;
+    Eterm ctl_heap[6];
+    Eterm token = NIL;
+    Process *sender = dsdp->proc;
+
+    if (SEQ_TRACE_TOKEN(sender) != NIL) {
+	seq_trace_update_send(sender);
+	token = SEQ_TRACE_TOKEN(sender);
+	seq_trace_output(token, message, SEQ_TRACE_SEND, remote_name, sender);
+    }
+
+    if (token != NIL)
+	ctl = TUPLE5(&ctl_heap[0], make_small(DOP_REG_SEND_TT),
+		     sender->id, am_Cookie, remote_name, token);
+    else
+	ctl = TUPLE4(&ctl_heap[0], make_small(DOP_REG_SEND),
+		     sender->id, am_Cookie, remote_name);
+    return dsig_send(dsdp, ctl, message, 0);
+}
+
+/* local has died, deliver the exit signal to remote */
+int
+erts_dsig_send_exit_tt(ErtsDSigData *dsdp, Eterm local, Eterm remote, 
+		       Eterm reason, Eterm token)
+{
+    Eterm ctl;
+    Eterm ctl_heap[6];
+
+    if (token != NIL) {	
+	seq_trace_update_send(dsdp->proc);
+	seq_trace_output_exit(token, reason, SEQ_TRACE_SEND, remote, local);
+	ctl = TUPLE5(&ctl_heap[0],
+		     make_small(DOP_EXIT_TT), local, remote, token, reason);
+    } else {
+	ctl = TUPLE4(&ctl_heap[0], make_small(DOP_EXIT), local, remote, reason);
+    }
+    /* forced, i.e ignore busy */
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 1);
+}
+
+int
+erts_dsig_send_exit(ErtsDSigData *dsdp, Eterm local, Eterm remote, Eterm reason)
+{
+    Eterm ctl_heap[5];
+    Eterm ctl = TUPLE4(&ctl_heap[0],
+		       make_small(DOP_EXIT), local, remote, reason);
+    /* forced, i.e ignore busy */
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 1);
+}
+
+int
+erts_dsig_send_exit2(ErtsDSigData *dsdp, Eterm local, Eterm remote, Eterm reason)
+{
+    Eterm ctl_heap[5];
+    Eterm ctl = TUPLE4(&ctl_heap[0],
+		       make_small(DOP_EXIT2), local, remote, reason);
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+}
+
+
+int
+erts_dsig_send_group_leader(ErtsDSigData *dsdp, Eterm leader, Eterm remote)
+{
+    Eterm ctl_heap[4];
+    Eterm ctl = TUPLE3(&ctl_heap[0],
+		       make_small(DOP_GROUP_LEADER), leader, remote);
+
+    return dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+}
+
+#if defined(PURIFY)
+#  define PURIFY_MSG(msg) \
+    purify_printf("%s, line %d: %s", __FILE__, __LINE__, msg)
+#elif defined(VALGRIND)
+#include <valgrind/valgrind.h>
+#include <valgrind/memcheck.h>
+
+#  define PURIFY_MSG(msg)                                                \
+    do {								 \
+	char buf__[1]; size_t bufsz__ = sizeof(buf__);			 \
+	if (erts_sys_getenv("VALGRIND_LOG_XML", buf__, &bufsz__) >= 0) { \
+	    VALGRIND_PRINTF("<erlang_error_log>"			 \
+			    "%s, line %d: %s</erlang_error_log>\n",	 \
+			    __FILE__, __LINE__, msg);			 \
+	} else {							 \
+	    VALGRIND_PRINTF("%s, line %d: %s", __FILE__, __LINE__, msg); \
+	}								 \
+    } while (0)
+#else
+#  define PURIFY_MSG(msg)
+#endif
+
+/*
+** Input from distribution port.
+**  Input follows the distribution protocol v4.5
+**  
+**   The protocol is a 4 byte header protocol
+**   the DOP_DATA is stripped by driver_output
+**
+**   assert  hlen == 0 !!!
+*/
+int erts_net_message(Port *prt,
+		     DistEntry *dep,
+		     byte *hbuf,
+		     int hlen,
+		     byte *buf,
+		     int len)
+{
+    ErtsDistExternal ede;
+    byte *t;
+    Sint ctl_len;
+    int orig_ctl_len;
+    Eterm arg;
+    Eterm from, to;
+    Eterm watcher, watched;
+    Eterm ref;
+    Eterm *tuple;
+    Eterm reason;
+    Process* rp;
+    Eterm ctl_default[64];
+    Eterm* ctl = ctl_default;
+    ErlOffHeap off_heap;
+    Eterm* hp;
+    Sint type;
+    Eterm token;
+    Eterm token_size;
+    ErtsMonitor *mon;
+    ErtsLink *lnk;
+    int res;
+#ifdef ERTS_DIST_MSG_DBG
+    int orig_len = len;
+#endif
+
+    /* Thanks to Luke Gorrie */
+    off_heap.mso = NULL;
+#ifndef HYBRID /* FIND ME! */
+    off_heap.funs = NULL;
+#endif
+    off_heap.overhead = 0;
+    off_heap.externals = NULL;
+
+    ERTS_SMP_CHK_NO_PROC_LOCKS;
+
+    ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+
+    if (!erts_is_alive)
+	return 0;
+    if (hlen > 0)
+	goto data_error;
+    if (len == 0)  /* HANDLE TICK !!! */
+	return 0;
+
+#ifdef ERTS_RAW_DIST_MSG_DBG
+    erts_fprintf(stderr, "<< ");
+    bw(buf, len);
+#endif
+
+    if (dep->flags & DFLAG_DIST_HDR_ATOM_CACHE)
+	t = buf;
+    else {
+	/* Skip PASS_THROUGH */
+	t = buf+1;
+	len--;
+    }
+
+    if (len == 0) {
+	PURIFY_MSG("data error");
+	goto data_error;
+    }
+
+    res = erts_prepare_dist_ext(&ede, t, len, dep, dep->cache);
+
+    if (res >= 0)
+	res = ctl_len = erts_decode_dist_ext_size(&ede, 0);
+    else {
+#ifdef ERTS_DIST_MSG_DBG
+	erts_fprintf(stderr, "DIST MSG DEBUG: erts_prepare_dist_ext() failed:\n");
+	bw(buf, orig_len);
+#endif
+	ctl_len = 0;
+    }
+
+    if (res < 0) {
+#ifdef ERTS_DIST_MSG_DBG
+	erts_fprintf(stderr, "DIST MSG DEBUG: erts_decode_dist_ext_size(CTL) failed:\n");
+	bw(buf, orig_len);
+#endif
+	PURIFY_MSG("data error");
+	goto data_error;
+    }
+    orig_ctl_len = ctl_len;
+    if (ctl_len > sizeof(ctl_default)/sizeof(ctl_default[0])) {
+	ctl = erts_alloc(ERTS_ALC_T_DCTRL_BUF, ctl_len * sizeof(Eterm));
+    }
+    hp = ctl;
+
+    arg = erts_decode_dist_ext(&hp, &off_heap, &ede);
+    if (is_non_value(arg)) {
+#ifdef ERTS_DIST_MSG_DBG
+	erts_fprintf(stderr, "DIST MSG DEBUG: erts_dist_ext_size(CTL) failed:\n");
+	bw(buf, orig_len);
+#endif
+	PURIFY_MSG("data error");
+	goto data_error;
+    }
+    ctl_len = t - buf;
+
+#ifdef ERTS_DIST_MSG_DBG
+    erts_fprintf(stderr, "<<%s CTL: %T\n", len != orig_len ? "P" : " ", arg);
+#endif
+
+    if (is_not_tuple(arg) || 
+	(tuple = tuple_val(arg), arityval(*tuple) < 1) ||
+	is_not_small(tuple[1])) {
+	erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+	erts_dsprintf(dsbufp, "Invalid distribution message: %.200T", arg);
+	erts_send_error_to_logger_nogl(dsbufp);
+	goto data_error;
+    }
+
+    token_size = 0;
+
+    switch (type = unsigned_val(tuple[1])) {
+    case DOP_LINK:
+	from = tuple[2];
+	to   = tuple[3];  /* local proc to link to */
+
+	if (is_not_pid(from) || is_not_pid(to)) {
+	    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+	    PURIFY_MSG("data error");
+	    erts_dsprintf(dsbufp,
+			  "Invalid DOP_LINK distribution message: %.200T",
+			  arg);
+	    erts_send_error_to_logger_nogl(dsbufp);
+	    goto data_error;
+	}
+
+	rp = erts_pid2proc_opt(NULL, 0,
+			       to, ERTS_PROC_LOCK_LINK,
+			       ERTS_P2P_FLG_ALLOW_OTHER_X);
+	if (!rp) {
+	    /* This is tricky (we MUST force a distributed send) */
+	    ErtsDSigData dsd;
+	    int code;
+	    code = erts_dsig_prepare(&dsd, dep, NULL, ERTS_DSP_NO_LOCK, 0);
+	    if (code == ERTS_DSIG_PREP_CONNECTED) {
+		code = erts_dsig_send_exit(&dsd, to, from, am_noproc);
+		ASSERT(code == ERTS_DSIG_SEND_OK);
+	    }
+	    break;
+	}
+
+	erts_smp_de_links_lock(dep);
+	res = erts_add_link(&(rp->nlinks), LINK_PID, from);
+
+	if (res < 0) {
+	    /* It was already there! Lets skip the rest... */
+	    erts_smp_de_links_unlock(dep);
+	    erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+	    break;
+	}
+	lnk = erts_add_or_lookup_link(&(dep->nlinks), LINK_PID, rp->id);
+	erts_add_link(&(ERTS_LINK_ROOT(lnk)), LINK_PID, from);
+	erts_smp_de_links_unlock(dep);
+
+	if (IS_TRACED_FL(rp, F_TRACE_PROCS))
+	    trace_proc(NULL, rp, am_getting_linked, from);
+
+	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+	break;
+
+    case DOP_UNLINK: {
+	ErtsDistLinkData dld;
+	from = tuple[2];
+	to = tuple[3];
+	
+	rp = erts_pid2proc_opt(NULL, 0,
+			       to, ERTS_PROC_LOCK_LINK,
+			       ERTS_P2P_FLG_ALLOW_OTHER_X);
+	if (!rp)
+	    break;
+
+	lnk = erts_remove_link(&(rp->nlinks), from);
+
+	if (IS_TRACED_FL(rp, F_TRACE_PROCS) && lnk != NULL) {
+	    trace_proc(NULL, rp, am_getting_unlinked, from);
+	}
+
+	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+
+	erts_remove_dist_link(&dld, to, from, dep);
+	erts_destroy_dist_link(&dld);
+	if (lnk)
+	    erts_destroy_link(lnk);
+	break;
+    }
+    
+    case DOP_MONITOR_P: {
+	/* A remote process wants to monitor us, we get:
+	   {DOP_MONITOR_P, Remote pid, local pid or name, ref} */
+	Eterm name;
+
+	watcher = tuple[2];
+	watched = tuple[3];  /* local proc to monitor */
+	ref     = tuple[4];
+
+	if (is_atom(watched)) {
+	    name = watched;
+	    rp = erts_whereis_process(NULL, 0,
+				      watched, ERTS_PROC_LOCK_LINK,
+				      ERTS_P2P_FLG_ALLOW_OTHER_X);
+	}
+	else {
+	    name = NIL;
+	    rp = erts_pid2proc_opt(NULL, 0,
+				   watched, ERTS_PROC_LOCK_LINK,
+				   ERTS_P2P_FLG_ALLOW_OTHER_X);
+	}
+
+	if (!rp) {
+	    ErtsDSigData dsd;
+	    int code;
+	    code = erts_dsig_prepare(&dsd, dep, NULL, ERTS_DSP_NO_LOCK, 0);
+	    if (code == ERTS_DSIG_PREP_CONNECTED) {
+		code = erts_dsig_send_m_exit(&dsd, watcher, watched, ref,
+					     am_noproc);
+		ASSERT(code == ERTS_DSIG_SEND_OK);
+	    }
+	}
+	else {
+	    if (is_atom(watched))
+		watched = rp->id;
+	    erts_smp_de_links_lock(dep);
+	    erts_add_monitor(&(dep->monitors), MON_ORIGIN, ref, watched, name);
+	    erts_add_monitor(&(rp->monitors), MON_TARGET, ref, watcher, name);
+	    erts_smp_de_links_unlock(dep);
+	    erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+	}
+
+	break;
+    }
+
+    case DOP_DEMONITOR_P:
+	/* A remote node informs us that a local pid in no longer monitored
+	   We get {DOP_DEMONITOR_P, Remote pid, Local pid or name, ref},
+	   We need only the ref of course */
+
+	/* watcher = tuple[2]; */
+	/* watched = tuple[3]; May be an atom in case of monitor name */
+	ref = tuple[4];
+
+	erts_smp_de_links_lock(dep);
+	mon = erts_remove_monitor(&(dep->monitors),ref);
+	erts_smp_de_links_unlock(dep);
+	/* ASSERT(mon != NULL); can happen in case of broken dist message */
+	if (mon == NULL) {
+	    break;
+	}
+	watched = mon->pid;
+	erts_destroy_monitor(mon);
+	rp = erts_pid2proc_opt(NULL, 0,
+			       watched, ERTS_PROC_LOCK_LINK,
+			       ERTS_P2P_FLG_ALLOW_OTHER_X);
+	if (!rp) {
+	    break;
+	}
+	mon = erts_remove_monitor(&(rp->monitors),ref);
+	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+	ASSERT(mon != NULL);
+	if (mon == NULL) {
+	    break;
+	}
+	erts_destroy_monitor(mon);
+	break;
+
+    case DOP_NODE_LINK: /* XXX never sent ?? */
+	break;
+
+    case DOP_REG_SEND_TT:
+	token_size = size_object(tuple[5]);
+	/* Fall through ... */
+    case DOP_REG_SEND:
+	/* {DOP_REG_SEND, From, Cookie, ToName} -- Message */
+	/* {DOP_REG_SEND_TT, From, Cookie, ToName, TraceToken} -- Message */
+
+	/*
+	 * There is intentionally no testing of the cookie (it is always '')
+	 * from R9B and onwards.
+	 */
+#ifdef ERTS_DIST_MSG_DBG
+	dist_msg_dbg(&ede, "MSG", buf, orig_len);
+#endif
+
+	from = tuple[2];
+	to = tuple[4];
+	rp = erts_whereis_process(NULL, 0, to, 0, ERTS_P2P_FLG_SMP_INC_REFC);
+	if (rp) {
+	    Uint xsize = (type == DOP_REG_SEND
+			  ? 0
+			  : ERTS_HEAP_FRAG_SIZE(token_size));
+	    ErtsProcLocks locks = 0;
+	    ErtsDistExternal *ede_copy;
+
+	    ede_copy = erts_make_dist_ext_copy(&ede, xsize);
+	    if (type == DOP_REG_SEND) {
+		token = NIL;
+	    } else {
+		ErlHeapFragment *heap_frag;
+		ErlOffHeap *ohp;
+		ASSERT(xsize);
+		heap_frag = erts_dist_ext_trailer(ede_copy);
+		ERTS_INIT_HEAP_FRAG(heap_frag, token_size);
+		hp = heap_frag->mem;
+		ohp = &heap_frag->off_heap;
+		token = tuple[5];
+		token = copy_struct(token, token_size, &hp, ohp);
+	    }
+
+	    erts_queue_dist_message(rp, &locks, ede_copy, token);
+	    if (locks)
+		erts_smp_proc_unlock(rp, locks);
+	    erts_smp_proc_dec_refc(rp);
+	}
+	break;
+
+    case DOP_SEND_TT:
+	token_size = size_object(tuple[4]);
+	/* Fall through ... */
+    case DOP_SEND:
+	/*
+	 * There is intentionally no testing of the cookie (it is always '')
+	 * from R9B and onwards.
+	 */
+#ifdef ERTS_DIST_MSG_DBG
+	dist_msg_dbg(&ede, "MSG", buf, orig_len);
+#endif
+
+	to = tuple[3];
+	rp = erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_SMP_INC_REFC);
+	if (rp) {
+	    Uint xsize = type == DOP_SEND ? 0 : ERTS_HEAP_FRAG_SIZE(token_size);
+	    ErtsProcLocks locks = 0;
+	    ErtsDistExternal *ede_copy;
+
+	    ede_copy = erts_make_dist_ext_copy(&ede, xsize);
+	    if (type == DOP_SEND) {
+		token = NIL;
+	    } else {
+		ErlHeapFragment *heap_frag;
+		ErlOffHeap *ohp;
+		ASSERT(xsize);
+		heap_frag = erts_dist_ext_trailer(ede_copy);
+		ERTS_INIT_HEAP_FRAG(heap_frag, token_size);
+		hp = heap_frag->mem;
+		ohp = &heap_frag->off_heap;
+		token = tuple[4];
+		token = copy_struct(token, token_size, &hp, ohp);
+	    }
+
+	    erts_queue_dist_message(rp, &locks, ede_copy, token);
+	    if (locks)
+		erts_smp_proc_unlock(rp, locks);
+	    erts_smp_proc_dec_refc(rp);
+	}
+	break;
+
+    case DOP_MONITOR_P_EXIT: {
+	/* We are monitoring a process on the remote node which dies, we get
+	   {DOP_MONITOR_P_EXIT, Remote pid or name, Local pid, ref, reason} */
+	   
+
+	Eterm lhp[3];
+	Eterm sysname;
+	ErtsProcLocks rp_locks = ERTS_PROC_LOCKS_MSG_SEND|ERTS_PROC_LOCK_LINK;
+
+	/* watched = tuple[2]; */  /* remote proc which died */
+	/* watcher = tuple[3]; */
+	ref     = tuple[4];
+	reason  = tuple[5];
+
+	erts_smp_de_links_lock(dep);
+	sysname = dep->sysname;
+	mon = erts_remove_monitor(&(dep->monitors), ref);
+	/*
+	 * If demonitor was performed at the same time as the
+	 * monitored process exits, monitoring side will have
+	 * removed info about monitor. In this case, do nothing
+	 * and everything will be as it should.
+	 */
+	erts_smp_de_links_unlock(dep);
+	if (mon == NULL) {
+	    break;
+	}
+	rp = erts_pid2proc(NULL, 0, mon->pid, rp_locks);
+	if (rp == NULL) {
+	    break;
+	}
+
+	erts_destroy_monitor(mon);
+
+	mon = erts_remove_monitor(&(rp->monitors),ref);
+
+	if (mon == NULL) {
+	    erts_smp_proc_unlock(rp, rp_locks);
+	    break;
+	}
+	
+	watched = (is_not_nil(mon->name)
+		   ? TUPLE2(&lhp[0], mon->name, sysname)
+		   : mon->pid);
+	
+	erts_queue_monitor_message(rp, &rp_locks,
+				   ref, am_process, watched, reason);
+	erts_smp_proc_unlock(rp, rp_locks);
+	erts_destroy_monitor(mon);
+	break;
+    }
+
+    case DOP_EXIT_TT:
+    case DOP_EXIT: {
+	ErtsDistLinkData dld;
+	ErtsProcLocks rp_locks = ERTS_PROC_LOCK_LINK|ERTS_PROC_LOCKS_XSIG_SEND;
+	/* 'from', which 'to' is linked to, died */
+	if (type == DOP_EXIT) {
+	   from = tuple[2];
+	   to = tuple[3];
+	   reason = tuple[4];
+	   token = NIL;
+	} else {
+	   from = tuple[2];
+	   to = tuple[3];
+	   token = tuple[4];
+	   reason = tuple[5];
+	}
+	if (is_not_internal_pid(to)) {
+	    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+	    PURIFY_MSG("data error");
+	    erts_dsprintf(dsbufp,
+			  "Invalid DOP_EXIT distribution message: %.200T",
+			  arg);
+	    erts_send_error_to_logger_nogl(dsbufp);
+	    goto data_error;
+	}
+
+	rp = erts_pid2proc(NULL, 0, to, rp_locks);
+	if (!rp)
+	    lnk = NULL;
+	else {
+	    lnk = erts_remove_link(&(rp->nlinks), from);
+
+	    /* If lnk == NULL, we have unlinked on this side, i.e.
+	     * ignore exit.
+	     */
+	    if (lnk) {
+		int xres;
+#if 0
+		/* Arndt: Maybe it should never be 'kill', but it can be,
+		   namely when a linked process does exit(kill). Until we know
+		   whether that is incorrect and what should happen instead,
+		   we leave the assertion out. */
+		ASSERT(reason != am_kill); /* should never be kill (killed) */
+#endif
+		xres = erts_send_exit_signal(NULL,
+					     from,
+					     rp,
+					     &rp_locks, 
+					     reason,
+					     token,
+					     NULL,
+					     ERTS_XSIG_FLG_IGN_KILL);
+		if (xres >= 0 && IS_TRACED_FL(rp, F_TRACE_PROCS)) {
+		    /* We didn't exit the process and it is traced */
+		    trace_proc(NULL, rp, am_getting_unlinked, from);
+		}
+	    }
+	    erts_smp_proc_unlock(rp, rp_locks);
+	}
+	erts_remove_dist_link(&dld, to, from, dep);
+	if (lnk)
+	    erts_destroy_link(lnk);
+	erts_destroy_dist_link(&dld);
+	break;
+    }
+    case DOP_EXIT2_TT:
+    case DOP_EXIT2: {
+	ErtsProcLocks rp_locks = ERTS_PROC_LOCKS_XSIG_SEND;
+	/* 'from' is send an exit signal to 'to' */
+	if (type == DOP_EXIT2) {
+	   from = tuple[2];
+	   to = tuple[3];
+	   reason = tuple[4];
+	   token = NIL;
+	} else {
+	   from = tuple[2];
+	   to = tuple[3];
+	   token = tuple[4];
+	   reason = tuple[5];
+	}
+	rp = erts_pid2proc_opt(NULL, 0, to, rp_locks,
+			       ERTS_P2P_FLG_SMP_INC_REFC);
+	if (rp) {
+	    (void) erts_send_exit_signal(NULL,
+					 from,
+					 rp,
+					 &rp_locks,
+					 reason,
+					 token,
+					 NULL,
+					 0);
+	    erts_smp_proc_unlock(rp, rp_locks);
+	    erts_smp_proc_dec_refc(rp);
+	}
+	break;
+    }
+    case DOP_GROUP_LEADER:
+	from = tuple[2];   /* Group leader  */
+	to = tuple[3];     /* new member */
+	if (is_not_pid(from))
+	    break;
+
+	rp = erts_pid2proc(NULL, 0, to, ERTS_PROC_LOCK_MAIN);
+	if (!rp)
+	    break;
+	rp->group_leader = STORE_NC_IN_PROC(rp, from);
+	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_MAIN);
+	break;
+
+    default: {
+	    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+	    erts_dsprintf(dsbufp,
+			  "Illegal value in distribution dispatch switch: "
+			  "%.200T",
+			  arg);
+	    erts_send_error_to_logger_nogl(dsbufp);
+	    PURIFY_MSG("data error");
+	    goto data_error;
+	}
+    }
+
+    if (off_heap.mso) {
+	erts_cleanup_mso(off_heap.mso);
+    }
+    if (off_heap.externals) {
+	erts_cleanup_externals(off_heap.externals);
+    }
+#ifndef HYBRID /* FIND ME! */
+    if (off_heap.funs) {
+	erts_cleanup_funs(off_heap.funs);
+    }
+    if (ctl != ctl_default) {
+	erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
+    }
+#endif
+    ERTS_SMP_CHK_NO_PROC_LOCKS;
+    return 0;
+
+ data_error:
+    if (off_heap.mso) {
+	erts_cleanup_mso(off_heap.mso);
+    }
+    if (off_heap.externals) {
+	erts_cleanup_externals(off_heap.externals);
+    }
+#ifndef HYBRID /* FIND ME! */
+    if (off_heap.funs) {
+	erts_cleanup_funs(off_heap.funs);
+    }
+    if (ctl != ctl_default) {
+	erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
+    }
+#endif
+    erts_do_exit_port(prt, dep->cid, am_killed);
+    ERTS_SMP_CHK_NO_PROC_LOCKS;
+    return -1;
+}
+
+#define ERTS_DE_BUSY_LIMIT (128*1024)
+
+static int
+dsig_send(ErtsDSigData *dsdp, Eterm ctl, Eterm msg, int force_busy)
+{
+    Eterm cid;
+    int suspended = 0;
+    int resume = 0;
+    Uint32 pass_through_size;
+    Uint data_size, dhdr_ext_size;
+    ErtsAtomCacheMap *acmp;
+    ErtsDistOutputBuf *obuf;
+    DistEntry *dep = dsdp->dep;
+    Uint32 flags = dep->flags;
+    Process *c_p = dsdp->proc;
+
+    if (!c_p || dsdp->no_suspend)
+	force_busy = 1;
+
+    ERTS_SMP_LC_ASSERT(!c_p
+		       || (ERTS_PROC_LOCK_MAIN
+			   == erts_proc_lc_my_proc_locks(c_p)));
+
+    if (!erts_is_alive)
+	return ERTS_DSIG_SEND_OK;
+
+    if (flags & DFLAG_DIST_HDR_ATOM_CACHE) {
+	acmp = erts_get_atom_cache_map(c_p);
+	pass_through_size = 0;
+    }
+    else {
+	acmp = NULL;
+	pass_through_size = 1;
+    }
+
+#ifdef ERTS_DIST_MSG_DBG
+    erts_fprintf(stderr, ">>%s CTL: %T\n", pass_through_size ? "P" : " ", ctl);
+    if (is_value(msg))
+	erts_fprintf(stderr, "    MSG: %T\n", msg);
+#endif
+
+    data_size = pass_through_size;
+    erts_reset_atom_cache_map(acmp);
+    data_size += erts_encode_dist_ext_size(ctl, flags, acmp);
+    if (is_value(msg))
+	data_size += erts_encode_dist_ext_size(msg, flags, acmp);
+    erts_finalize_atom_cache_map(acmp);
+
+    dhdr_ext_size = erts_encode_ext_dist_header_size(acmp);
+    data_size += dhdr_ext_size;
+
+    obuf = alloc_dist_obuf(data_size);
+    obuf->ext_endp = &obuf->data[0] + pass_through_size + dhdr_ext_size;
+
+    /* Encode internal version of dist header */
+    obuf->extp = erts_encode_ext_dist_header_setup(obuf->ext_endp, acmp);
+    /* Encode control message */
+    erts_encode_dist_ext(ctl, &obuf->ext_endp, flags, acmp);
+    if (is_value(msg)) {
+	/* Encode message */
+	erts_encode_dist_ext(msg, &obuf->ext_endp, flags, acmp);
+    }
+
+    ASSERT(obuf->extp < obuf->ext_endp);
+    ASSERT(&obuf->data[0] <= obuf->extp - pass_through_size);
+    ASSERT(obuf->ext_endp <= &obuf->data[0] + data_size);
+
+    data_size = obuf->ext_endp - obuf->extp;
+
+    /*
+     * Signal encoded; now verify that the connection still exists,
+     * and if so enqueue the signal and schedule it for send.
+     */
+    obuf->next = NULL;
+    erts_smp_de_rlock(dep);
+    cid = dep->cid;
+    if (cid != dsdp->cid
+	|| dep->connection_id != dsdp->connection_id
+	|| dep->status & ERTS_DE_SFLG_EXITING) {
+	/* Not the same connection as when we started; drop message... */
+	erts_smp_de_runlock(dep);
+	free_dist_obuf(obuf);
+    }
+    else {
+	ErtsProcList *plp = NULL;
+	erts_smp_spin_lock(&dep->qlock);
+	dep->qsize += size_obuf(obuf);
+	if (dep->qsize >= ERTS_DE_BUSY_LIMIT)
+	    dep->qflgs |= ERTS_DE_QFLG_BUSY;
+	if (!force_busy && (dep->qflgs & ERTS_DE_QFLG_BUSY)) {
+	    erts_smp_spin_unlock(&dep->qlock);
+
+	    plp = erts_proclist_create(c_p);
+	    plp->next = NULL;
+	    erts_suspend(c_p, ERTS_PROC_LOCK_MAIN, NULL);
+	    suspended = 1;
+	    erts_smp_spin_lock(&dep->qlock);
+	}
+
+	/* Enqueue obuf on dist entry */
+	if (dep->out_queue.last)
+	    dep->out_queue.last->next = obuf;
+	else
+	    dep->out_queue.first = obuf;
+	dep->out_queue.last = obuf;
+
+	if (!force_busy) {
+	    if (!(dep->qflgs & ERTS_DE_QFLG_BUSY)) {
+		if (suspended)
+		    resume = 1; /* was busy when we started, but isn't now */
+	    }
+	    else {
+		/* Enqueue suspended process on dist entry */
+		ASSERT(plp);
+		if (dep->suspended.last)
+		    dep->suspended.last->next = plp;
+		else
+		    dep->suspended.first = plp;
+		dep->suspended.last = plp;
+	    }
+	}
+
+	erts_smp_spin_unlock(&dep->qlock);
+	erts_schedule_dist_command(NULL, dep);
+	erts_smp_de_runlock(dep);
+	
+	if (resume) {
+	    erts_resume(c_p, ERTS_PROC_LOCK_MAIN);
+	    erts_proclist_destroy(plp);
+	    /*
+	     * Note that the calling process still have to yield as if it
+	     * suspended. If not, the calling process could later be
+	     * erroneously scheduled when it shouldn't be.
+	     */
+	}
+    }
+
+    if (c_p) {
+	int reds;
+	/* 
+	 * Bump reductions on calling process.
+	 *
+	 * This is the reduction cost: Always a base cost of 8 reductions
+	 * plus 16 reductions per kilobyte generated external data.
+	 */
+
+	data_size >>= (10-4);
+#if defined(ARCH_64)
+	data_size &= 0x003fffffffffffff;
+#elif defined(ARCH_32)
+	data_size &= 0x003fffff;
+#else
+#       error "Ohh come on ... !?!"
+#endif
+	reds = 8 + ((int) data_size > 1000000 ? 1000000 : (int) data_size);
+	BUMP_REDS(c_p, reds);
+    }
+
+    if (suspended) {
+	if (!resume && erts_system_monitor_flags.busy_dist_port)
+	    monitor_generic(c_p, am_busy_dist_port, cid);
+	return ERTS_DSIG_SEND_YIELD;
+    }
+    return ERTS_DSIG_SEND_OK;
+}
+
+
+static Uint
+dist_port_command(Port *prt, ErtsDistOutputBuf *obuf)
+{
+    int fpe_was_unmasked;
+    Uint size = obuf->ext_endp - obuf->extp;
+
+    ERTS_SMP_CHK_NO_PROC_LOCKS;
+    ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+
+    if (size > (Uint) INT_MAX)
+	erl_exit(ERTS_ABORT_EXIT,
+		 "Absurdly large distribution output data buffer "
+		 "(%bpu bytes) passed.\n",
+		 size);
+
+    prt->caller = NIL;
+    fpe_was_unmasked = erts_block_fpe();
+    (*prt->drv_ptr->output)((ErlDrvData) prt->drv_data,
+			    (char*) obuf->extp,
+			    (int) size);
+    erts_unblock_fpe(fpe_was_unmasked);
+    return size;
+}
+
+static Uint
+dist_port_commandv(Port *prt, ErtsDistOutputBuf *obuf)
+{
+    int fpe_was_unmasked;
+    Uint size = obuf->ext_endp - obuf->extp;
+    SysIOVec iov[2];
+    ErlDrvBinary* bv[2];
+    ErlIOVec eiov;
+
+    ERTS_SMP_CHK_NO_PROC_LOCKS;
+    ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+
+    if (size > (Uint) INT_MAX)
+	erl_exit(ERTS_ABORT_EXIT,
+		 "Absurdly large distribution output data buffer "
+		 "(%bpu bytes) passed.\n",
+		 size);
+
+    iov[0].iov_base = NULL;
+    iov[0].iov_len = 0;
+    bv[0] = NULL;
+
+    iov[1].iov_base = obuf->extp;
+    iov[1].iov_len = size;
+    bv[1] = Binary2ErlDrvBinary(ErtsDistOutputBuf2Binary(obuf));
+
+    eiov.vsize = 2;
+    eiov.size = size;
+    eiov.iov = iov;
+    eiov.binv = bv;
+
+    ASSERT(prt->drv_ptr->outputv);
+
+    prt->caller = NIL;
+    fpe_was_unmasked = erts_block_fpe();
+    (*prt->drv_ptr->outputv)((ErlDrvData) prt->drv_data, &eiov);
+    erts_unblock_fpe(fpe_was_unmasked);
+
+    return size;
+}
+
+
+#if defined(ARCH_64)
+#define ERTS_PORT_REDS_MASK__ 0x003fffffffffffffL
+#elif defined(ARCH_32)
+#define ERTS_PORT_REDS_MASK__ 0x003fffff
+#else
+#  error "Ohh come on ... !?!"
+#endif
+
+#define ERTS_PORT_REDS_DIST_CMD_START 5
+#define ERTS_PORT_REDS_DIST_CMD_FINALIZE 3
+#define ERTS_PORT_REDS_DIST_CMD_EXIT 200
+#define ERTS_PORT_REDS_DIST_CMD_RESUMED 5
+#define ERTS_PORT_REDS_DIST_CMD_DATA(SZ) \
+  ((SZ) < (1 << 10) \
+   ? ((Sint) 1) \
+   : ((((Sint) (SZ)) >> 10) & ((Sint) ERTS_PORT_REDS_MASK__)))
+
+int
+erts_dist_command(Port *prt, int reds_limit)
+{
+    Sint reds = ERTS_PORT_REDS_DIST_CMD_START;
+    int prt_busy;
+    int de_busy;
+    Uint32 status;
+    Uint32 flags;
+    Uint32 qflgs;
+    Sint obufsize = 0;
+    ErtsDistOutputQueue oq, foq;
+    DistEntry *dep = prt->dist_entry;
+    Uint (*send)(Port *prt, ErtsDistOutputBuf *obuf);
+
+    ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+
+    erts_refc_inc(&dep->refc, 1); /* Otherwise dist_entry might be
+				     removed if port command fails */
+
+    erts_smp_atomic_xchg(&dep->dist_cmd_scheduled, 0);
+
+    erts_smp_de_rlock(dep);
+    flags = dep->flags;
+    status = dep->status;
+    send = dep->send;
+    erts_smp_de_runlock(dep);
+
+    if (status & ERTS_DE_SFLG_EXITING) {
+	erts_do_exit_port(prt, prt->id, am_killed);
+	erts_deref_dist_entry(dep);
+	return reds + ERTS_PORT_REDS_DIST_CMD_EXIT;
+    }
+
+    ASSERT(send);
+
+    /*
+     * We need to remove both out queues from the
+     * dist entry while passing it to port command;
+     * otherwise, port command will free the buffers
+     * in the queues on failure and we'll end up with
+     * a mess.
+     */
+
+    erts_smp_spin_lock(&dep->qlock);
+    oq.first = dep->out_queue.first;
+    oq.last = dep->out_queue.last;
+    dep->out_queue.first = NULL;
+    dep->out_queue.last = NULL;
+    qflgs = dep->qflgs;
+    erts_smp_spin_unlock(&dep->qlock);
+
+    foq.first = dep->finalized_out_queue.first;
+    foq.last = dep->finalized_out_queue.last;
+    dep->finalized_out_queue.first = NULL;
+    dep->finalized_out_queue.last = NULL;
+
+    if (reds > reds_limit)
+	goto preempted;
+
+    prt_busy = (int) (prt->status & ERTS_PORT_SFLG_PORT_BUSY);
+    de_busy = (int) (qflgs & ERTS_DE_QFLG_BUSY);
+
+    if (prt_busy) {
+	if (!de_busy) {
+	    erts_smp_spin_lock(&dep->qlock);
+	    dep->qflgs |= ERTS_DE_QFLG_BUSY;
+	    erts_smp_spin_unlock(&dep->qlock);
+	    de_busy = 1;
+	}
+    }
+    else if (foq.first) {
+	int preempt = 0;
+	do {
+	    Uint size;
+	    ErtsDistOutputBuf *fob;
+
+	    size = (*send)(prt, foq.first);
+#ifdef ERTS_RAW_DIST_MSG_DBG
+	    erts_fprintf(stderr, ">> ");
+	    bw(foq.first->extp, size);
+#endif
+	    reds += ERTS_PORT_REDS_DIST_CMD_DATA(size);
+	    fob = foq.first;
+	    obufsize += size_obuf(fob);
+	    foq.first = foq.first->next;
+	    free_dist_obuf(fob);
+	    preempt = reds > reds_limit || (prt->status & ERTS_PORT_SFLGS_DEAD);
+	    if (prt->status & ERTS_PORT_SFLG_PORT_BUSY) {
+		erts_smp_spin_lock(&dep->qlock);
+		dep->qflgs |= ERTS_DE_QFLG_BUSY;
+		erts_smp_spin_unlock(&dep->qlock);
+		de_busy = prt_busy = 1;
+		break;
+	    }
+	} while (foq.first && !preempt);
+	if (!foq.first)
+	    foq.last = NULL;
+	if (preempt)
+	    goto preempted;
+    }
+
+    if (prt_busy) {
+	if (oq.first) {
+	    ErtsDistOutputBuf *ob;
+	    int preempt;
+	finalize_only:
+	    preempt = 0;
+	    ob = oq.first;
+	    ASSERT(ob);
+	    do {
+		ob->extp = erts_encode_ext_dist_header_finalize(ob->extp,
+								dep->cache);
+		if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
+		    *--ob->extp = PASS_THROUGH; /* Old node; 'pass through'
+						   needed */
+		ASSERT(&ob->data[0] <= ob->extp && ob->extp < ob->ext_endp);
+		reds += ERTS_PORT_REDS_DIST_CMD_FINALIZE;
+		preempt = reds > reds_limit;
+		if (preempt)
+		    break;
+		ob = ob->next;
+	    } while (ob);
+	    /*
+	     * At least one buffer was finalized; if we got preempted,
+	     * ob points to the last buffer that we finalized.
+	     */
+	    if (foq.last)
+		foq.last->next = oq.first;
+	    else
+		foq.first = oq.first;
+	    if (!preempt) {
+		/* All buffers finalized */
+		foq.last = oq.last;
+		oq.first = oq.last = NULL;
+	    }
+	    else {
+		/* Not all buffers finalized; split oq. */
+		foq.last = ob;
+		oq.first = ob->next;
+		if (oq.first)
+		    ob->next = NULL;
+		else
+		    oq.last = NULL;
+	    }
+	    if (preempt)
+		goto preempted;
+	}
+    }
+    else {
+	int preempt = 0;
+	while (oq.first && !preempt) {
+	    ErtsDistOutputBuf *fob;
+	    Uint size;
+	    oq.first->extp
+		= erts_encode_ext_dist_header_finalize(oq.first->extp,
+						       dep->cache);
+	    reds += ERTS_PORT_REDS_DIST_CMD_FINALIZE;
+	    if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
+		*--oq.first->extp = PASS_THROUGH; /* Old node; 'pass through'
+						     needed */
+	    ASSERT(&oq.first->data[0] <= oq.first->extp
+		   && oq.first->extp < oq.first->ext_endp);
+	    size = (*send)(prt, oq.first);
+#ifdef ERTS_RAW_DIST_MSG_DBG
+	    erts_fprintf(stderr, ">> ");
+	    bw(oq.first->extp, size);
+#endif
+	    reds += ERTS_PORT_REDS_DIST_CMD_DATA(size);
+	    fob = oq.first;
+	    obufsize += size_obuf(fob);
+	    oq.first = oq.first->next;
+	    free_dist_obuf(fob);
+	    preempt = reds > reds_limit || (prt->status & ERTS_PORT_SFLGS_DEAD);
+	    if (prt->status & ERTS_PORT_SFLG_PORT_BUSY) {
+		erts_smp_spin_lock(&dep->qlock);
+		dep->qflgs |= ERTS_DE_QFLG_BUSY;
+		erts_smp_spin_unlock(&dep->qlock);
+		de_busy = prt_busy = 1;
+		if (oq.first && !preempt)
+		    goto finalize_only;
+	    }
+	}
+
+	ASSERT(!oq.first || preempt);
+
+	/*
+	 * Preempt if not all buffers have been handled.
+	 */
+	if (preempt && oq.first)
+	    goto preempted;
+
+#ifdef DEBUG
+	oq.last = NULL;
+#endif
+	ASSERT(!oq.first);
+	ASSERT(!foq.first && !foq.last);
+
+	/*
+	 * Everything that was buffered when we started have now been
+	 * written to the port. If port isn't busy but dist entry is
+	 * and we havn't got too muched queued on dist entry, set
+	 * dist entry in a non-busy state and resume suspended
+	 * processes.
+	 */
+	erts_smp_spin_lock(&dep->qlock);
+	ASSERT(dep->qsize >= obufsize);
+	dep->qsize -= obufsize;
+	obufsize = 0;
+	if (de_busy && !prt_busy && dep->qsize < ERTS_DE_BUSY_LIMIT) {
+	    ErtsProcList *suspendees;
+	    int resumed;
+	    suspendees = get_suspended_on_de(dep, ERTS_DE_QFLG_BUSY);
+	    erts_smp_spin_unlock(&dep->qlock);
+
+	    resumed = erts_resume_processes(suspendees);
+	    reds += resumed*ERTS_PORT_REDS_DIST_CMD_RESUMED;
+	    de_busy = 0;
+	}
+	else
+	    erts_smp_spin_unlock(&dep->qlock);
+    }
+
+    ASSERT(!oq.first && !oq.last);
+
+ done:
+
+    if (obufsize != 0) {
+	ASSERT(obufsize > 0);
+	erts_smp_spin_lock(&dep->qlock);
+	ASSERT(dep->qsize >= obufsize);
+	dep->qsize -= obufsize;
+	erts_smp_spin_unlock(&dep->qlock);
+    }
+
+    ASSERT(foq.first || !foq.last);
+    ASSERT(!foq.first || foq.last);
+    ASSERT(!dep->finalized_out_queue.first);
+    ASSERT(!dep->finalized_out_queue.last);
+
+    if (foq.first) {
+	dep->finalized_out_queue.first = foq.first;
+	dep->finalized_out_queue.last = foq.last;
+    }
+
+     /* Avoid wrapping reduction counter... */
+    if (reds > INT_MAX/2)
+	reds = INT_MAX/2;
+
+    erts_deref_dist_entry(dep);
+
+    return reds;
+
+ preempted:
+
+    ASSERT(oq.first || !oq.last);
+    ASSERT(!oq.first || oq.last);
+
+    if (prt->status & ERTS_PORT_SFLGS_DEAD) {
+	/*
+	 * Port died during port command; clean up 'oq'
+	 * and 'foq'. Things buffered in dist entry after
+	 * we begun processing the queues have already been
+	 * cleaned up when port terminated.
+	 */
+
+	if (oq.first)
+	    oq.last->next = foq.first;
+	else
+	    oq.first = foq.first;
+
+	while (oq.first) {
+	    ErtsDistOutputBuf *fob = oq.first;
+	    oq.first = oq.first->next;
+	    obufsize += size_obuf(fob);
+	    free_dist_obuf(fob);
+	}
+
+	foq.first = NULL;
+	foq.last = NULL;
+
+#ifdef DEBUG
+	erts_smp_spin_lock(&dep->qlock);
+	ASSERT(dep->qsize == obufsize);
+	erts_smp_spin_unlock(&dep->qlock);
+#endif
+    }
+    else {
+	if (oq.first) {
+	    /*
+	     * Unhandle buffers need to be put back first
+	     * in out_queue.
+	     */
+	    erts_smp_spin_lock(&dep->qlock);
+	    dep->qsize -= obufsize;
+	    obufsize = 0;
+	    oq.last->next = dep->out_queue.first;
+	    dep->out_queue.first = oq.first;
+	    if (!dep->out_queue.last)
+		dep->out_queue.last = oq.last;
+	    erts_smp_spin_unlock(&dep->qlock);
+	}
+
+	erts_schedule_dist_command(prt, NULL);
+    }
+    goto done;
+}
+
+void
+erts_dist_port_not_busy(Port *prt)
+{
+    erts_schedule_dist_command(prt, NULL);
+}
+
+void
+erts_kill_dist_connection(DistEntry *dep, Uint32 connection_id)
+{
+    erts_smp_de_rwlock(dep);
+    if (is_internal_port(dep->cid)
+	&& connection_id == dep->connection_id
+	&& !(dep->status & ERTS_DE_SFLG_EXITING)) {
+
+	dep->status |= ERTS_DE_SFLG_EXITING;
+
+	erts_smp_spin_lock(&dep->qlock);
+	ASSERT(!(dep->qflgs & ERTS_DE_QFLG_EXIT));
+	dep->qflgs |= ERTS_DE_QFLG_EXIT;
+	erts_smp_spin_unlock(&dep->qlock);
+
+	erts_schedule_dist_command(NULL, dep);
+    }
+    erts_smp_de_rwunlock(dep);
+}
+
+struct print_to_data {
+    int to;
+    void *arg;
+};
+
+static void doit_print_monitor_info(ErtsMonitor *mon, void *vptdp)
+{
+    int to = ((struct print_to_data *) vptdp)->to;
+    void *arg = ((struct print_to_data *) vptdp)->arg;
+    Process *rp;
+    ErtsMonitor *rmon;
+    rp = erts_pid2proc_unlocked(mon->pid);
+    if (!rp || (rmon = erts_lookup_monitor(rp->monitors, mon->ref)) == NULL) {
+	erts_print(to, arg, "Warning, stray monitor for: %T\n", mon->pid);
+    } else if (mon->type == MON_ORIGIN) {
+	/* Local pid is being monitored */
+	erts_print(to, arg, "Remotely monitored by: %T %T\n",
+		   mon->pid, rmon->pid);
+    } else {
+	erts_print(to, arg, "Remote monitoring: %T ", mon->pid);
+	if (is_not_atom(rmon->pid))
+	    erts_print(to, arg, "%T\n", rmon->pid);
+	else
+	    erts_print(to, arg, "{%T, %T}\n",
+		       rmon->name,
+		       rmon->pid); /* which in this case is the 
+				      remote system name... */
+    }
+}    
+
+static void print_monitor_info(int to, void *arg, ErtsMonitor *mon)
+{
+    struct print_to_data ptd = {to, arg};
+    erts_doforall_monitors(mon,&doit_print_monitor_info,&ptd);
+}
+
+typedef struct {
+    struct print_to_data *ptdp;
+    Eterm from;
+} PrintLinkContext;
+
+static void doit_print_link_info2(ErtsLink *lnk, void *vpplc)
+{
+    PrintLinkContext *pplc = (PrintLinkContext *) vpplc;
+    erts_print(pplc->ptdp->to, pplc->ptdp->arg, "Remote link: %T %T\n",
+	       pplc->from, lnk->pid);
+}
+
+static void doit_print_link_info(ErtsLink *lnk, void *vptdp)
+{
+    if (is_internal_pid(lnk->pid) && erts_pid2proc_unlocked(lnk->pid)) {
+	PrintLinkContext plc = {(struct print_to_data *) vptdp, lnk->pid};
+	erts_doforall_links(ERTS_LINK_ROOT(lnk), &doit_print_link_info2, &plc);
+    } 
+}
+
+static void print_link_info(int to, void *arg, ErtsLink *lnk)
+{
+    struct print_to_data ptd = {to, arg};
+    erts_doforall_links(lnk, &doit_print_link_info, (void *) &ptd);
+}
+
+typedef struct {
+    struct print_to_data ptd;
+    Eterm sysname;
+} PrintNodeLinkContext;
+    
+
+static void doit_print_nodelink_info(ErtsLink *lnk, void *vpcontext)
+{
+    PrintNodeLinkContext *pcontext = vpcontext;
+
+    if (is_internal_pid(lnk->pid) && erts_pid2proc_unlocked(lnk->pid))
+	erts_print(pcontext->ptd.to, pcontext->ptd.arg,
+		   "Remote monitoring: %T %T\n", lnk->pid, pcontext->sysname);
+}
+
+static void print_nodelink_info(int to, void *arg, ErtsLink *lnk, Eterm sysname)
+{
+    PrintNodeLinkContext context = {{to, arg}, sysname};
+    erts_doforall_links(lnk, &doit_print_nodelink_info, &context);
+}
+
+
+static int
+info_dist_entry(int to, void *arg, DistEntry *dep, int visible, int connected)
+{
+
+  if (visible && connected) {
+      erts_print(to, arg, "=visible_node:");
+  } else if (connected) {
+      erts_print(to, arg, "=hidden_node:");
+  } else {
+      erts_print(to, arg, "=not_connected:");
+  }
+  erts_print(to, arg, "%d\n", dist_entry_channel_no(dep));
+
+  if(connected && is_nil(dep->cid)) {
+    erts_print(to, arg,
+	       "Error: Not connected node still registered as connected:%T\n",
+	       dep->sysname);
+    return 0;
+  }
+
+  if(!connected && is_not_nil(dep->cid)) {
+    erts_print(to, arg,
+	       "Error: Connected node not registered as connected:%T\n",
+	       dep->sysname);
+    return 0;
+  }
+
+  erts_print(to, arg, "Name: %T", dep->sysname);
+#ifdef DEBUG
+  erts_print(to, arg, " (refc=%d)", erts_refc_read(&dep->refc, 1));
+#endif
+  erts_print(to, arg, "\n");
+  if (!connected && is_nil(dep->cid)) {
+    if (dep->nlinks) {
+      erts_print(to, arg, "Error: Got links to not connected node:%T\n",
+		 dep->sysname);
+    }
+    return 0;
+  }
+
+  erts_print(to, arg, "Controller: %T\n", dep->cid, to);
+
+  erts_print_node_info(to, arg, dep->sysname, NULL, NULL);
+  print_monitor_info(to, arg, dep->monitors);
+  print_link_info(to, arg, dep->nlinks);
+  print_nodelink_info(to, arg, dep->node_links, dep->sysname);
+
+  return 0;
+    
+}
+int distribution_info(int to, void *arg)	/* Called by break handler */
+{
+    DistEntry *dep;
+
+    erts_print(to, arg, "=node:%T\n", erts_this_dist_entry->sysname);
+ 
+    if (erts_this_node->sysname == am_Noname) {
+	erts_print(to, arg, "=no_distribution\n");
+	return(0);
+    }
+
+#if 0
+    if (!erts_visible_dist_entries && !erts_hidden_dist_entries) 
+      erts_print(to, arg, "Alive but not holding any connections \n");
+#endif
+
+    for(dep = erts_visible_dist_entries; dep; dep = dep->next) {
+      info_dist_entry(to, arg, dep, 1, 1);
+    }
+
+    for(dep = erts_hidden_dist_entries; dep; dep = dep->next) {
+      info_dist_entry(to, arg, dep, 0, 1);
+    }
+
+    for (dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
+	info_dist_entry(to, arg, dep, 0, 0);
+    }
+
+    return(0);
+}
+
+/****************************************************************************
+  DISTRIBUTION BIFS:
+
+            setnode/2     -- start distribution
+            setnode/3     -- set node controller
+
+            node/1        -- return objects node name
+            node/0        -- return this node name
+            nodes/0       -- return a list of all (non hidden) nodes
+            is_alive      -- return true if distribution is running else false
+	    monitor_node  -- turn on/off node monitoring
+
+            node controller only:
+            dist_exit/3       -- send exit signals from remote to local process
+            dist_link/2       -- link a remote process to a local
+            dist_unlink/2     -- unlink a remote from a local
+****************************************************************************/
+
+
+
+/**********************************************************************
+ ** Set the node name of current node fail if node already is set.
+ ** setnode(name@host, Creation)
+ ** loads functions pointer to trap_functions from module erlang.
+ **    erlang:dsend/2
+ **    erlang:dlink/1
+ **    erlang:dunlink/1
+ **    erlang:dmonitor_node/3
+ **    erlang:dgroup_leader/2
+ **    erlang:dexit/2
+ **  -- are these needed ?
+ **    dexit/1
+ ***********************************************************************/
+
+BIF_RETTYPE setnode_2(BIF_ALIST_2)
+{
+    Process *net_kernel;
+    Uint creation;
+
+    /* valid creation ? */
+    if(!term_to_Uint(BIF_ARG_2, &creation))
+	goto error;
+    if(creation > 3)
+	goto error;
+
+    /* valid node name ? */
+    if (!is_node_name_atom(BIF_ARG_1))
+	goto error;
+
+    if (BIF_ARG_1 == am_Noname) /* cant use this name !! */
+	goto error;
+    if (erts_is_alive)     /* must not be alive! */
+	goto error;
+
+    /* Check that all trap functions are defined !! */
+    if (dsend2_trap->address == NULL ||
+	dsend3_trap->address == NULL ||
+	/*	dsend_nosuspend_trap->address == NULL ||*/
+	dlink_trap->address == NULL ||
+	dunlink_trap->address == NULL ||
+	dmonitor_node_trap->address == NULL ||
+	dgroup_leader_trap->address == NULL ||
+	dmonitor_p_trap->address == NULL ||
+	dexit_trap->address == NULL) {
+	goto error;
+    }
+
+    net_kernel = erts_whereis_process(BIF_P, ERTS_PROC_LOCK_MAIN,
+				      am_net_kernel, ERTS_PROC_LOCK_MAIN, 0);
+    if (!net_kernel)
+	goto error;
+
+    /* By setting dist_entry==erts_this_dist_entry and DISTRIBUTION on
+       net_kernel do_net_exist will be called when net_kernel
+       is terminated !! */
+    (void *) ERTS_PROC_SET_DIST_ENTRY(net_kernel,
+				      ERTS_PROC_LOCK_MAIN,
+				      erts_this_dist_entry);
+    erts_refc_inc(&erts_this_dist_entry->refc, 2);
+    net_kernel->flags |= F_DISTRIBUTION;
+
+    if (net_kernel != BIF_P)
+	erts_smp_proc_unlock(net_kernel, ERTS_PROC_LOCK_MAIN);
+
+#ifdef DEBUG
+    erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+    ASSERT(!erts_visible_dist_entries && !erts_hidden_dist_entries);
+    erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+#endif
+
+    erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
+    erts_smp_block_system(ERTS_BS_FLG_ALLOW_GC);
+    erts_set_this_node(BIF_ARG_1, (Uint32) creation);
+    erts_is_alive = 1;
+    send_nodes_mon_msgs(NULL, am_nodeup, BIF_ARG_1, am_visible, NIL);
+    erts_smp_release_system();
+    erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN);
+
+    BIF_RET(am_true);
+
+ error:
+    BIF_ERROR(BIF_P, BADARG);
+}
+
+/**********************************************************************
+ ** Allocate a dist entry, set node name install the connection handler
+ ** setnode_3({name@host, Creation}, Cid, {Type, Version, Initial, IC, OC})
+ ** Type = flag field, where the flags are specified in dist.h
+ ** Version = distribution version, >= 1
+ ** IC = in_cookie (ignored)
+ ** OC = out_cookie (ignored)
+ **
+ ** Note that in distribution protocols above 1, the Initial parameter
+ ** is always NIL and the cookies are always the atom '', cookies are not
+ ** sent in the distribution messages but are only used in 
+ ** the handshake.
+ **
+ ***********************************************************************/
+
+BIF_RETTYPE setnode_3(BIF_ALIST_3)
+{
+    BIF_RETTYPE ret;
+    Uint flags;
+    unsigned long version;
+    Eterm ic, oc;
+    Eterm *tp;
+    DistEntry *dep = NULL;
+    Port *pp = NULL;
+
+    /* Prepare for success */
+    ERTS_BIF_PREP_RET(ret, am_true);
+
+    /*
+     * Check and pick out arguments
+     */
+
+    if (!is_node_name_atom(BIF_ARG_1) ||
+	is_not_internal_port(BIF_ARG_2) ||
+	(erts_this_node->sysname == am_Noname)) {
+	goto badarg;
+    }
+
+    if (!is_tuple(BIF_ARG_3))
+	goto badarg;
+    tp = tuple_val(BIF_ARG_3);
+    if (*tp++ != make_arityval(4))
+	goto badarg;
+    if (!is_small(*tp))
+	goto badarg;
+    flags = unsigned_val(*tp++);
+    if (!is_small(*tp) || (version = unsigned_val(*tp)) == 0)
+	goto badarg;
+    ic = *(++tp);
+    oc = *(++tp);
+    if (!is_atom(ic) || !is_atom(oc))
+	goto badarg;
+
+    /* DFLAG_EXTENDED_REFERENCES is compulsory from R9 and forward */
+    if (!(DFLAG_EXTENDED_REFERENCES & flags)) {
+	erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+	erts_dsprintf(dsbufp, "%T", BIF_P->id);
+	if (BIF_P->reg)
+	    erts_dsprintf(dsbufp, " (%T)", BIF_P->reg->name);
+	erts_dsprintf(dsbufp,
+		      " attempted to enable connection to node %T "
+		      "which is not able to handle extended references.\n",
+		      BIF_ARG_1);
+	erts_send_error_to_logger(BIF_P->group_leader, dsbufp);
+	goto badarg;
+    }
+
+    /*
+     * Arguments seem to be in order.
+     */
+
+    /* get dist_entry */
+    dep = erts_find_or_insert_dist_entry(BIF_ARG_1);
+    if (dep == erts_this_dist_entry)
+	goto badarg;
+    else if (!dep)
+	goto system_limit; /* Should never happen!!! */
+
+    pp = erts_id2port(BIF_ARG_2, BIF_P, ERTS_PROC_LOCK_MAIN);
+    erts_smp_de_rwlock(dep);
+
+    if (!pp || (pp->status & ERTS_PORT_SFLG_EXITING))
+	goto badarg;
+
+    if ((pp->drv_ptr->flags & ERL_DRV_FLAG_SOFT_BUSY) == 0)
+	goto badarg;
+
+    if (dep->cid == BIF_ARG_2 && pp->dist_entry == dep)
+	goto done; /* Already set */
+
+    if (dep->status & ERTS_DE_SFLG_EXITING) {
+	/* Suspend on dist entry waiting for the exit to finish */
+	ErtsProcList *plp = erts_proclist_create(BIF_P);
+	plp->next = NULL;
+	erts_suspend(BIF_P, ERTS_PROC_LOCK_MAIN, NULL);
+	erts_smp_spin_lock(&dep->qlock);
+	if (dep->suspended.last)
+	    dep->suspended.last->next = plp;
+	else
+	    dep->suspended.first = plp;
+	dep->suspended.last = plp;
+	erts_smp_spin_unlock(&dep->qlock);
+	goto yield;
+    }
+
+    ASSERT(!(dep->status & ERTS_DE_SFLG_EXITING));
+
+    if (pp->dist_entry || is_not_nil(dep->cid))
+	goto badarg;
+
+    erts_port_status_bor_set(pp, ERTS_PORT_SFLG_DISTRIBUTION);
+
+    pp->dist_entry = dep;
+
+    dep->version = version;
+    dep->creation = 0;
+
+    ASSERT(pp->drv_ptr->outputv || pp->drv_ptr->output);
+
+#if 1
+    dep->send = (pp->drv_ptr->outputv
+		 ? dist_port_commandv
+		 : dist_port_command);
+#else
+    dep->send = dist_port_command;
+#endif
+    ASSERT(dep->send);
+
+#ifdef DEBUG
+    erts_smp_spin_lock(&dep->qlock);
+    ASSERT(dep->qsize == 0);
+    erts_smp_spin_unlock(&dep->qlock);
+#endif
+
+    erts_set_dist_entry_connected(dep, BIF_ARG_2, flags);
+
+    if (flags & DFLAG_DIST_HDR_ATOM_CACHE)
+	create_cache(dep);
+
+    erts_smp_de_rwunlock(dep);
+    dep = NULL; /* inc of refc transferred to port (dist_entry field) */
+
+    send_nodes_mon_msgs(BIF_P,
+			am_nodeup,
+			BIF_ARG_1,
+			flags & DFLAG_PUBLISHED ? am_visible : am_hidden,
+			NIL);
+ done:
+
+    if (dep && dep != erts_this_dist_entry) {
+	erts_smp_de_rwunlock(dep);
+	erts_deref_dist_entry(dep);
+    }
+
+    if (pp)
+	erts_smp_port_unlock(pp);
+
+    return ret;
+
+ yield:
+    ERTS_BIF_PREP_YIELD3(ret, bif_export[BIF_setnode_3], BIF_P,
+			 BIF_ARG_1, BIF_ARG_2, BIF_ARG_3);
+    goto done;
+
+ badarg:
+    ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
+    goto done;
+
+ system_limit:
+    ERTS_BIF_PREP_ERROR(ret, BIF_P, SYSTEM_LIMIT);
+    goto done;
+}
+
+
+/**********************************************************************/
+/* dist_exit(Local, Term, Remote) -> Bool */
+
+BIF_RETTYPE dist_exit_3(BIF_ALIST_3)
+{
+    Eterm local;
+    Eterm remote;
+    DistEntry *rdep;
+
+    local = BIF_ARG_1;
+    remote = BIF_ARG_3;
+
+    /* Check that remote is a remote process */
+    if (is_not_external_pid(remote))
+	goto error;
+
+    rdep = external_dist_entry(remote);
+    
+    if(rdep == erts_this_dist_entry)
+	goto error;
+
+    /* Check that local is local */
+    if (is_internal_pid(local)) {
+	Process *lp;
+	ErtsProcLocks lp_locks;
+	if (BIF_P->id == local) {
+	    lp_locks = ERTS_PROC_LOCKS_ALL;
+	    lp = BIF_P;
+	    erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCKS_ALL_MINOR);
+	}
+	else {
+	    lp_locks = ERTS_PROC_LOCKS_XSIG_SEND;
+	    lp = erts_pid2proc_opt(BIF_P, ERTS_PROC_LOCK_MAIN,
+				   local, lp_locks,
+				   ERTS_P2P_FLG_SMP_INC_REFC);
+	    if (!lp) {
+		BIF_RET(am_true); /* ignore */
+	    }
+	}
+	
+	(void) erts_send_exit_signal(BIF_P,
+				     remote,
+				     lp,
+				     &lp_locks,
+				     BIF_ARG_2,
+				     NIL,
+				     NULL,
+				     0);
+#ifdef ERTS_SMP
+	if (lp == BIF_P)
+	    lp_locks &= ~ERTS_PROC_LOCK_MAIN;
+#endif
+	erts_smp_proc_unlock(lp, lp_locks);
+	if (lp != BIF_P)
+	    erts_smp_proc_dec_refc(lp);
+	else {
+	    /*
+	     * We may have exited current process and may have to take action.
+	     */
+	    ERTS_BIF_CHK_EXITED(BIF_P);
+	    ERTS_SMP_BIF_CHK_PENDING_EXIT(BIF_P, ERTS_PROC_LOCK_MAIN);
+	}
+    }
+    else if (is_external_pid(local)
+	     && external_dist_entry(local) == erts_this_dist_entry) {
+	BIF_RET(am_true); /* ignore */
+    }
+    else
+	goto error;
+    BIF_RET(am_true);
+
+ error:
+    BIF_ERROR(BIF_P, BADARG);
+}
+
+/**********************************************************************/
+/* node(Object) -> Node */
+
+BIF_RETTYPE node_1(BIF_ALIST_1)
+{ 
+    if (is_not_node_container(BIF_ARG_1))
+      BIF_ERROR(BIF_P, BADARG);
+    BIF_RET(node_container_node_name(BIF_ARG_1));
+}
+
+/**********************************************************************/
+/* node() -> Node */
+
+BIF_RETTYPE node_0(BIF_ALIST_0)
+{
+    BIF_RET(erts_this_dist_entry->sysname);
+}
+
+
+/**********************************************************************/
+/* nodes() -> [ Node ] */
+
+#if 0 /* Done in erlang.erl instead. */
+BIF_RETTYPE nodes_0(BIF_ALIST_0)
+{
+  return nodes_1(BIF_P, am_visible);
+}
+#endif
+
+
+BIF_RETTYPE nodes_1(BIF_ALIST_1)
+{
+    Eterm result;
+    int length;
+    Eterm* hp;
+    int not_connected = 0;
+    int visible = 0;
+    int hidden = 0;
+    int this = 0;
+    Uint  buf[2]; /* For one cons-cell */
+    DistEntry *dep;
+    Eterm arg_list = BIF_ARG_1;
+#ifdef DEBUG
+    Eterm* endp;
+#endif
+    if (is_atom(BIF_ARG_1))
+      arg_list = CONS(buf, BIF_ARG_1, NIL);
+
+    while (is_list(arg_list)) {
+      switch(CAR(list_val(arg_list))) {
+      case am_visible:   visible = 1;                                 break;
+      case am_hidden:    hidden = 1;                                  break;
+      case am_known:     visible = hidden = not_connected = this = 1; break;
+      case am_this:      this = 1;                                    break;
+      case am_connected: visible = hidden = 1;                        break;
+      default:           BIF_ERROR(BIF_P, BADARG);                    break;
+      }
+      arg_list = CDR(list_val(arg_list));
+    }
+
+    if (is_not_nil(arg_list))
+      BIF_ERROR(BIF_P, BADARG);
+
+    length = 0;
+
+    erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+
+    ASSERT(erts_no_of_not_connected_dist_entries >= 0);
+    ASSERT(erts_no_of_hidden_dist_entries >= 0);
+    ASSERT(erts_no_of_visible_dist_entries >= 0);
+    if(not_connected)
+      length += erts_no_of_not_connected_dist_entries;
+    if(hidden)
+      length += erts_no_of_hidden_dist_entries;
+    if(visible)
+      length += erts_no_of_visible_dist_entries;
+    if(this)
+      length++;
+
+    result = NIL;
+
+    if (length == 0) {
+	erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+	BIF_RET(result);
+    }
+
+    hp = HAlloc(BIF_P, 2*length);
+
+#ifdef DEBUG
+    endp = hp + length*2;
+#endif
+    if(not_connected)
+      for(dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
+	result = CONS(hp, dep->sysname, result);
+	hp += 2;
+      }
+    if(hidden)
+      for(dep = erts_hidden_dist_entries; dep; dep = dep->next) {
+	result = CONS(hp, dep->sysname, result);
+	hp += 2;
+      }
+    if(visible)
+      for(dep = erts_visible_dist_entries; dep; dep = dep->next) {
+	result = CONS(hp, dep->sysname, result);
+	hp += 2;
+      }
+    if(this) {
+	result = CONS(hp, erts_this_dist_entry->sysname, result);
+	hp += 2;
+    }
+    ASSERT(endp == hp);
+    erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+    BIF_RET(result);
+}
+
+/**********************************************************************/
+/* is_alive() -> Bool */
+
+BIF_RETTYPE is_alive_0(BIF_ALIST_0)
+{
+    Eterm res = erts_is_alive ? am_true : am_false;
+    BIF_RET(res);
+}
+
+/**********************************************************************/
+/* erlang:monitor_node(Node, Bool, Options) -> Bool */
+
+BIF_RETTYPE monitor_node_3(BIF_ALIST_3)
+{
+    DistEntry *dep;
+    ErtsLink *lnk;
+    Eterm l;
+
+    for (l = BIF_ARG_3; l != NIL && is_list(l); l = CDR(list_val(l))) {
+	Eterm t = CAR(list_val(l));
+	/* allow_passive_connect the only available option right now */
+	if (t != am_allow_passive_connect) {
+	    BIF_ERROR(BIF_P, BADARG);
+	}
+    }
+    if (l != NIL) {
+	BIF_ERROR(BIF_P, BADARG);
+    }
+
+    if (is_not_atom(BIF_ARG_1) ||
+	((BIF_ARG_2 != am_true) && (BIF_ARG_2 != am_false)) ||
+	((erts_this_node->sysname == am_Noname)
+	 && (BIF_ARG_1 != erts_this_node->sysname))) {
+	BIF_ERROR(BIF_P, BADARG);
+    }
+    dep = erts_sysname_to_connected_dist_entry(BIF_ARG_1);
+    if (!dep) {
+    do_trap:
+	BIF_TRAP3(dmonitor_node_trap, BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3);
+    }
+    if (dep == erts_this_dist_entry)
+	goto done;
+
+    erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_LINK);
+    erts_smp_de_rlock(dep);
+    if (ERTS_DE_IS_NOT_CONNECTED(dep)) {
+	erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK);
+	erts_smp_de_runlock(dep);
+	goto do_trap;
+    }
+    erts_smp_de_links_lock(dep);
+    erts_smp_de_runlock(dep);
+
+    if (BIF_ARG_2 == am_true) {
+	ASSERT(dep->cid != NIL);
+	lnk = erts_add_or_lookup_link(&(dep->node_links), LINK_NODE, 
+				      BIF_P->id);
+	++ERTS_LINK_REFC(lnk);
+	lnk = erts_add_or_lookup_link(&(BIF_P->nlinks), LINK_NODE, BIF_ARG_1);
+	++ERTS_LINK_REFC(lnk);
+    }
+    else  {
+	lnk = erts_lookup_link(dep->node_links, BIF_P->id);
+	if (lnk != NULL) {
+	    if ((--ERTS_LINK_REFC(lnk)) == 0) {
+		erts_destroy_link(erts_remove_link(&(dep->node_links), 
+						   BIF_P->id));
+	    }
+	}
+	lnk = erts_lookup_link(BIF_P->nlinks, BIF_ARG_1);
+	if (lnk != NULL) {
+	    if ((--ERTS_LINK_REFC(lnk)) == 0) {
+		erts_destroy_link(erts_remove_link(&(BIF_P->nlinks), 
+						   BIF_ARG_1));
+	    }
+	}
+    }
+
+    erts_smp_de_links_unlock(dep);
+    erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK);
+
+ done:
+    erts_deref_dist_entry(dep);
+    BIF_RET(am_true);
+}
+
+/* monitor_node(Node, Bool) -> Bool */
+
+BIF_RETTYPE monitor_node_2(BIF_ALIST_2)
+{
+    BIF_RET(monitor_node_3(BIF_P,BIF_ARG_1,BIF_ARG_2,NIL));
+}
+
+BIF_RETTYPE net_kernel_dflag_unicode_io_1(BIF_ALIST_1)
+{
+    DistEntry *de;
+    Uint32 f;
+    if (is_not_pid(BIF_ARG_1)) {
+	BIF_ERROR(BIF_P,BADARG);
+    }
+    de = pid_dist_entry(BIF_ARG_1);
+    ASSERT(de != NULL);
+    if (de == erts_this_dist_entry) {
+	BIF_RET(am_true);
+    }
+    erts_smp_de_rlock(de);
+    f = de->flags;
+    erts_smp_de_runlock(de);
+    BIF_RET(((f & DFLAG_UNICODE_IO) ? am_true : am_false));
+}
+    
+/*
+ * The major part of the implementation of net_kernel:monitor_nodes/[1,2]
+ * follows.
+ *
+ * Currently net_kernel:monitor_nodes/[1,2] calls process_flag/2 which in
+ * turn calls erts_monitor_nodes(). If the process_flag() call fails (with
+ * badarg), the code in net_kernel determines what type of error to return.
+ * This in order to simplify the task of being backward compatible.
+ */
+
+#define ERTS_NODES_MON_OPT_TYPE_VISIBLE		(((Uint16) 1) << 0)
+#define ERTS_NODES_MON_OPT_TYPE_HIDDEN		(((Uint16) 1) << 1)
+#define ERTS_NODES_MON_OPT_DOWN_REASON		(((Uint16) 1) << 2)
+
+#define ERTS_NODES_MON_OPT_TYPES \
+  (ERTS_NODES_MON_OPT_TYPE_VISIBLE|ERTS_NODES_MON_OPT_TYPE_HIDDEN)
+
+typedef struct ErtsNodesMonitor_ ErtsNodesMonitor;
+struct ErtsNodesMonitor_ {
+    ErtsNodesMonitor *prev;
+    ErtsNodesMonitor *next;
+    Process *proc;
+    Uint16 opts;
+    Uint16 no;
+};
+
+static erts_smp_mtx_t nodes_monitors_mtx;
+static ErtsNodesMonitor *nodes_monitors;
+static ErtsNodesMonitor *nodes_monitors_end;
+
+/*
+ * Nodes monitors are stored in a double linked list. 'nodes_monitors'
+ * points to the beginning of the list and 'nodes_monitors_end' points
+ * to the end of the list.
+ *
+ * There might be more than one entry per process in the list. If so,
+ * they are located in sequence. The 'nodes_monitors' field of the
+ * process struct refers to the first element in the sequence
+ * corresponding to the process in question.
+ */
+
+static void
+init_nodes_monitors(void)
+{
+    erts_smp_mtx_init(&nodes_monitors_mtx, "nodes_monitors");
+    nodes_monitors = NULL;
+    nodes_monitors_end = NULL;
+}
+
+static ERTS_INLINE Uint
+nodes_mon_msg_sz(ErtsNodesMonitor *nmp, Eterm what, Eterm reason)
+{
+    Uint sz;
+    if (!nmp->opts) {
+	sz = 3;
+    }
+    else {
+	sz = 0;
+
+	if (nmp->opts & ERTS_NODES_MON_OPT_TYPES)
+	    sz += 2 + 3;
+
+	if (what == am_nodedown
+	    && (nmp->opts & ERTS_NODES_MON_OPT_DOWN_REASON)) {
+	    if (is_not_immed(reason))
+		sz += size_object(reason);
+	    sz += 2 + 3;
+	}
+
+	sz += 4;
+    }
+    return sz;
+}
+
+static ERTS_INLINE void
+send_nodes_mon_msg(Process *rp,
+		   ErtsProcLocks *rp_locksp,
+		   ErtsNodesMonitor *nmp,
+		   Eterm node,
+		   Eterm what,
+		   Eterm type,
+		   Eterm reason,
+		   Uint sz)
+{
+    Eterm msg;
+    ErlHeapFragment* bp;
+    ErlOffHeap *ohp;
+    Eterm *hp = erts_alloc_message_heap(sz, &bp, &ohp, rp, rp_locksp);
+#ifdef DEBUG
+    Eterm *hend = hp + sz;
+#endif
+
+    if (!nmp->opts) {
+	msg = TUPLE2(hp, what, node);
+#ifdef DEBUG
+	hp += 3;
+#endif
+    }
+    else {
+	Eterm tup;
+	Eterm info = NIL;
+
+	if (nmp->opts & (ERTS_NODES_MON_OPT_TYPE_VISIBLE
+			 | ERTS_NODES_MON_OPT_TYPE_HIDDEN)) {
+
+	    tup = TUPLE2(hp, am_node_type, type);
+	    hp += 3;
+	    info = CONS(hp, tup, info);
+	    hp += 2;
+	}
+
+	if (what == am_nodedown
+	    && (nmp->opts & ERTS_NODES_MON_OPT_DOWN_REASON)) {
+	    Eterm rsn_cpy;
+	    
+	    if (is_immed(reason))
+		rsn_cpy = reason;
+	    else {
+		Eterm rsn_sz = size_object(reason);
+		rsn_cpy = copy_struct(reason, rsn_sz, &hp, ohp);
+	    }
+
+	    tup = TUPLE2(hp, am_nodedown_reason, rsn_cpy);
+	    hp += 3;
+	    info = CONS(hp, tup, info);
+	    hp += 2;
+	}
+
+	msg = TUPLE3(hp, what, node, info);
+#ifdef DEBUG
+	hp += 4;
+#endif
+    }
+
+    ASSERT(hend == hp);
+    erts_queue_message(rp, rp_locksp, bp, msg, NIL);
+}
+
+static void
+send_nodes_mon_msgs(Process *c_p, Eterm what, Eterm node, Eterm type, Eterm reason)
+{
+    ErtsNodesMonitor *nmp;
+    ErtsProcLocks rp_locks = 0; /* Init to shut up false warning */
+    Process *rp = NULL;
+
+    ASSERT(is_immed(what));
+    ASSERT(is_immed(node));
+    ASSERT(is_immed(type));
+
+    ERTS_SMP_LC_ASSERT(!c_p
+		       || (erts_proc_lc_my_proc_locks(c_p)
+			   == ERTS_PROC_LOCK_MAIN));
+    erts_smp_mtx_lock(&nodes_monitors_mtx);
+
+    for (nmp = nodes_monitors; nmp; nmp = nmp->next) {
+	int i;
+	Uint16 no;
+	Uint sz;
+
+	ASSERT(nmp->proc != NULL);
+
+	if (!nmp->opts) {
+	    if (type != am_visible)
+		continue;
+	}
+	else {
+	    switch (type) {
+	    case am_hidden:
+		if (!(nmp->opts & ERTS_NODES_MON_OPT_TYPE_HIDDEN))
+		    continue;
+		break;
+	    case am_visible:
+		if ((nmp->opts & ERTS_NODES_MON_OPT_TYPES)
+		    && !(nmp->opts & ERTS_NODES_MON_OPT_TYPE_VISIBLE))
+		    continue;
+		break;
+	    default:
+		erl_exit(ERTS_ABORT_EXIT, "Bad node type found\n");
+	    }
+	}
+
+	if (rp != nmp->proc) {
+	    if (rp) {
+		if (rp == c_p)
+		    rp_locks &= ~ERTS_PROC_LOCK_MAIN;
+		erts_smp_proc_unlock(rp, rp_locks);
+	    }
+
+	    rp = nmp->proc;
+	    rp_locks = 0;
+	    if (rp == c_p)
+		rp_locks |= ERTS_PROC_LOCK_MAIN;
+	}
+
+	ASSERT(rp);
+
+	sz = nodes_mon_msg_sz(nmp, what, reason);
+
+	for (i = 0, no = nmp->no; i < no; i++)
+	    send_nodes_mon_msg(rp,
+			       &rp_locks,
+			       nmp,
+			       node,
+			       what,
+			       type,
+			       reason,
+			       sz);
+    }
+
+    if (rp) {
+	if (rp == c_p)
+	    rp_locks &= ~ERTS_PROC_LOCK_MAIN;
+	erts_smp_proc_unlock(rp, rp_locks);
+    }
+
+    erts_smp_mtx_unlock(&nodes_monitors_mtx);
+}
+
+static Eterm
+insert_nodes_monitor(Process *c_p, Uint32 opts)
+{
+    Uint16 no = 1;
+    Eterm res = am_false;
+    ErtsNodesMonitor *xnmp, *nmp;
+
+    ERTS_SMP_LC_ASSERT(erts_smp_lc_mtx_is_locked(&nodes_monitors_mtx));
+    ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) & ERTS_PROC_LOCK_MAIN);
+
+    xnmp = c_p->nodes_monitors;
+    if (xnmp) {
+	ASSERT(!xnmp->prev || xnmp->prev->proc != c_p);
+
+	while (1) {
+	    ASSERT(xnmp->proc == c_p);
+	    if (xnmp->opts == opts)
+		break;
+	    if (!xnmp->next || xnmp->next->proc != c_p)
+		break;
+	    xnmp = xnmp->next;
+	}
+	ASSERT(xnmp);
+	ASSERT(xnmp->proc == c_p);
+	ASSERT(xnmp->opts == opts
+	       || !xnmp->next
+	       || xnmp->next->proc != c_p);
+
+	if (xnmp->opts != opts)
+	    goto alloc_new;
+	else {
+	    res = am_true;
+	    no = xnmp->no++;
+	    if (!xnmp->no) {
+		/*
+		 * 'no' wrapped; transfer all prevous monitors to new
+		 * element (which will be the next element in the list)
+		 * and set this to one...
+		 */
+		xnmp->no = 1;
+		goto alloc_new;
+	    }
+	}
+    }
+    else {
+    alloc_new:
+	nmp = erts_alloc(ERTS_ALC_T_NODES_MON, sizeof(ErtsNodesMonitor));
+	nmp->proc = c_p;
+	nmp->opts = opts;
+	nmp->no = no;
+
+	if (xnmp) {
+	    ASSERT(nodes_monitors);
+	    ASSERT(c_p->nodes_monitors);
+	    nmp->next = xnmp->next;
+	    nmp->prev = xnmp;
+	    xnmp->next = nmp;
+	    if (nmp->next) {
+		ASSERT(nodes_monitors_end != xnmp);
+		ASSERT(nmp->next->prev == xnmp);
+		nmp->next->prev = nmp;
+	    }
+	    else {
+		ASSERT(nodes_monitors_end == xnmp);
+		nodes_monitors_end = nmp;
+	    }
+	}
+	else {
+	    ASSERT(!c_p->nodes_monitors);
+	    c_p->nodes_monitors = nmp;
+	    nmp->next = NULL;
+	    nmp->prev = nodes_monitors_end;
+	    if (nodes_monitors_end) {
+		ASSERT(nodes_monitors);
+		nodes_monitors_end->next = nmp;
+	    }
+	    else {
+		ASSERT(!nodes_monitors);
+		nodes_monitors = nmp;
+	    }
+	    nodes_monitors_end = nmp;
+	}
+    }
+    return res;
+}
+
+static Eterm
+remove_nodes_monitors(Process *c_p, Uint32 opts, int all)
+{
+    Eterm res = am_false;
+    ErtsNodesMonitor *nmp;
+
+    ERTS_SMP_LC_ASSERT(erts_smp_lc_mtx_is_locked(&nodes_monitors_mtx));
+    ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) & ERTS_PROC_LOCK_MAIN);
+
+    nmp = c_p->nodes_monitors;
+    ASSERT(!nmp || !nmp->prev || nmp->prev->proc != c_p);
+
+    while (nmp && nmp->proc == c_p) {
+	if (!all && nmp->opts != opts)
+	    nmp = nmp->next;
+	else { /* if (all || nmp->opts == opts) */
+	    ErtsNodesMonitor *free_nmp;
+	    res = am_true;
+	    if (nmp->prev) {
+		ASSERT(nodes_monitors != nmp);
+		nmp->prev->next = nmp->next;
+	    }
+	    else {
+		ASSERT(nodes_monitors == nmp);
+		nodes_monitors = nmp->next;
+	    }
+	    if (nmp->next) {
+		ASSERT(nodes_monitors_end != nmp);
+		nmp->next->prev = nmp->prev;
+	    }
+	    else {
+		ASSERT(nodes_monitors_end == nmp);
+		nodes_monitors_end = nmp->prev;
+	    }
+	    free_nmp = nmp;
+	    nmp = nmp->next;
+	    if (c_p->nodes_monitors == free_nmp)
+		c_p->nodes_monitors = nmp && nmp->proc == c_p ? nmp : NULL;
+	    erts_free(ERTS_ALC_T_NODES_MON, free_nmp);
+	}
+    }
+    
+    ASSERT(!all || !c_p->nodes_monitors);
+    return res;
+}
+
+void
+erts_delete_nodes_monitors(Process *c_p, ErtsProcLocks locks)
+{
+#if defined(ERTS_ENABLE_LOCK_CHECK) && defined(ERTS_SMP)
+    if (c_p) {
+	ErtsProcLocks might_unlock = locks & ~ERTS_PROC_LOCK_MAIN;
+	if (might_unlock)
+	    erts_proc_lc_might_unlock(c_p, might_unlock);
+    }
+#endif
+    if (erts_smp_mtx_trylock(&nodes_monitors_mtx) == EBUSY) {
+	ErtsProcLocks unlock_locks = locks & ~ERTS_PROC_LOCK_MAIN;
+	if (c_p && unlock_locks)
+	    erts_smp_proc_unlock(c_p, unlock_locks);
+	erts_smp_mtx_lock(&nodes_monitors_mtx);
+	if (c_p && unlock_locks)
+	    erts_smp_proc_lock(c_p, unlock_locks);
+    }
+    remove_nodes_monitors(c_p, 0, 1);
+    erts_smp_mtx_unlock(&nodes_monitors_mtx);
+}
+
+Eterm
+erts_monitor_nodes(Process *c_p, Eterm on, Eterm olist)
+{
+    Eterm res;
+    Eterm opts_list = olist;
+    Uint16 opts = (Uint16) 0;
+
+    ASSERT(c_p);
+    ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) == ERTS_PROC_LOCK_MAIN);
+
+    if (on != am_true && on != am_false)
+	return THE_NON_VALUE;
+
+    if (is_not_nil(opts_list)) {
+	int all = 0, visible = 0, hidden = 0;
+
+	while (is_list(opts_list)) {
+	    Eterm *cp = list_val(opts_list);
+	    Eterm opt = CAR(cp);
+	    opts_list = CDR(cp);
+	    if (opt == am_nodedown_reason)
+		opts |= ERTS_NODES_MON_OPT_DOWN_REASON;
+	    else if (is_tuple(opt)) {
+		Eterm* tp = tuple_val(opt);
+		if (arityval(tp[0]) != 2)
+		    return THE_NON_VALUE;
+		switch (tp[1]) {
+		case am_node_type:
+		    switch (tp[2]) {
+		    case am_visible:
+			if (hidden || all)
+			    return THE_NON_VALUE;
+			opts |= ERTS_NODES_MON_OPT_TYPE_VISIBLE;
+			visible = 1;
+			break;
+		    case am_hidden:
+			if (visible || all)
+			    return THE_NON_VALUE;
+			opts |= ERTS_NODES_MON_OPT_TYPE_HIDDEN;
+			hidden = 1;
+			break;
+		    case am_all:
+			if (visible || hidden)
+			    return THE_NON_VALUE;
+			opts |= ERTS_NODES_MON_OPT_TYPES;
+			all = 1;
+			break;
+		    default:
+			return THE_NON_VALUE;
+		    }
+		    break;
+		default:
+		    return THE_NON_VALUE;
+		}
+	    }
+	    else {
+		return THE_NON_VALUE;
+	    }
+	}
+
+	if (is_not_nil(opts_list))
+	    return THE_NON_VALUE;
+    }
+
+    erts_smp_mtx_lock(&nodes_monitors_mtx);
+
+    if (on == am_true)
+	res = insert_nodes_monitor(c_p, opts);
+    else
+	res = remove_nodes_monitors(c_p, opts, 0);
+
+    erts_smp_mtx_unlock(&nodes_monitors_mtx);
+
+    return res;
+}
+
+/*
+ * Note, this function is only used for debuging.
+ */
+
+Eterm
+erts_processes_monitoring_nodes(Process *c_p)
+{
+    ErtsNodesMonitor *nmp;
+    Eterm res;
+    Eterm *hp;
+    Eterm **hpp;
+    Uint sz;
+    Uint *szp;
+#ifdef DEBUG
+    Eterm *hend;
+#endif
+
+    ASSERT(c_p);
+    ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) == ERTS_PROC_LOCK_MAIN);
+    erts_smp_mtx_lock(&nodes_monitors_mtx);
+
+    sz = 0;
+    szp = &sz;
+    hpp = NULL;
+
+ bld_result:
+    res = NIL;
+
+    for (nmp = nodes_monitors_end; nmp; nmp = nmp->prev) {
+	Uint16 i;
+	for (i = 0; i < nmp->no; i++) {
+	    Eterm olist = NIL;
+	    if (nmp->opts & ERTS_NODES_MON_OPT_TYPES) {
+		Eterm type;
+		switch (nmp->opts & ERTS_NODES_MON_OPT_TYPES) {
+		case ERTS_NODES_MON_OPT_TYPES:        type = am_all;     break;
+		case ERTS_NODES_MON_OPT_TYPE_VISIBLE: type = am_visible; break;
+		case ERTS_NODES_MON_OPT_TYPE_HIDDEN:  type = am_hidden;  break;
+		default: erl_exit(ERTS_ABORT_EXIT, "Bad node type found\n");
+		}
+		olist = erts_bld_cons(hpp, szp, 
+				      erts_bld_tuple(hpp, szp, 2,
+						     am_node_type,
+						     type),
+				      olist);
+	    }
+	    if (nmp->opts & ERTS_NODES_MON_OPT_DOWN_REASON)
+		olist = erts_bld_cons(hpp, szp, am_nodedown_reason, olist);
+	    res = erts_bld_cons(hpp, szp,
+				erts_bld_tuple(hpp, szp, 2,
+					       nmp->proc->id,
+					       olist),
+				res);
+	}
+    }
+
+    if (!hpp) {
+	hp = HAlloc(c_p, sz);
+#ifdef DEBUG
+	hend = hp + sz;
+#endif
+	hpp = &hp;
+	szp = NULL;
+	goto bld_result;
+    }
+
+    ASSERT(hp == hend);
+
+    erts_smp_mtx_unlock(&nodes_monitors_mtx);
+
+    return res;
+}