1 files changed, 225 insertions, 191 deletions

diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 749760a1b3..7856d8003c 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -177,9 +177,9 @@ static Export *dist_ctrl_put_data_trap;
 
 /* forward declarations */
 
-static int dsig_send_exit(ErtsDSigData* dsdp, Eterm ctl, Eterm msg, Eterm from, int force_busy);
-static int dsig_send_exit_ctx(ErtsSendContext *ctx, Eterm ctl, Eterm msg);
-static int dsig_send_ctl(ErtsDSigData* dsdp, Eterm ctl, int force_busy);
+static void erts_schedule_dist_command(Port *, DistEntry *);
+static int dsig_send_exit(ErtsDSigSendContext *ctx, Eterm ctl, Eterm msg);
+static int dsig_send_ctl(ErtsDSigSendContext *ctx, Eterm ctl);
 static void send_nodes_mon_msgs(Process *, Eterm, Eterm, Eterm, Eterm);
 static void init_nodes_monitors(void);
 static Sint abort_connection(DistEntry* dep, Uint32 conn_id);
@@ -860,20 +860,20 @@ static void free_de_out_queues(DistEntry* dep, ErtsDistOutputBuf *obuf)
 
 int erts_dsend_context_dtor(Binary* ctx_bin)
 {
-    ErtsSendContext* ctx = ERTS_MAGIC_BIN_DATA(ctx_bin);
-    switch (ctx->dss.phase) {
+    ErtsDSigSendContext* ctx = ERTS_MAGIC_BIN_DATA(ctx_bin);
+    switch (ctx->phase) {
     case ERTS_DSIG_SEND_PHASE_MSG_SIZE:
-	DESTROY_SAVED_WSTACK(&ctx->dss.u.sc.wstack);
+	DESTROY_SAVED_WSTACK(&ctx->u.sc.wstack);
 	break;
     case ERTS_DSIG_SEND_PHASE_MSG_ENCODE:
-	DESTROY_SAVED_WSTACK(&ctx->dss.u.ec.wstack);
+	DESTROY_SAVED_WSTACK(&ctx->u.ec.wstack);
 	break;
     default:;
     }
-    if (ctx->dss.phase >= ERTS_DSIG_SEND_PHASE_ALLOC && ctx->dss.obuf) {
+    if (ctx->phase >= ERTS_DSIG_SEND_PHASE_ALLOC && ctx->obuf) {
         int i;
-        for (i = 0; i < ctx->dss.fragments; i++)
-            free_dist_obuf(&ctx->dss.obuf[i]);
+        for (i = 0; i < ctx->fragments; i++)
+            free_dist_obuf(&ctx->obuf[i]);
     }
     if (ctx->deref_dep)
 	erts_deref_dist_entry(ctx->dep);
@@ -881,10 +881,10 @@ int erts_dsend_context_dtor(Binary* ctx_bin)
     return 1;
 }
 
-Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* ctx)
+Eterm erts_dsend_export_trap_context(Process* p, ErtsDSigSendContext* ctx)
 {
     struct exported_ctx {
-	ErtsSendContext ctx;
+	ErtsDSigSendContext ctx;
 	ErtsAtomCacheMap acm;
     };
     Binary* ctx_bin = erts_create_magic_binary(sizeof(struct exported_ctx),
@@ -892,12 +892,12 @@ Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* ctx)
     struct exported_ctx* dst = ERTS_MAGIC_BIN_DATA(ctx_bin);
     Eterm* hp = HAlloc(p, ERTS_MAGIC_REF_THING_SIZE);
 
-    sys_memcpy(&dst->ctx, ctx, sizeof(ErtsSendContext));
-    ASSERT(ctx->dss.ctl == make_tuple(ctx->ctl_heap));
-    dst->ctx.dss.ctl = make_tuple(dst->ctx.ctl_heap);
-    if (ctx->dss.acmp) {
-	sys_memcpy(&dst->acm, ctx->dss.acmp, sizeof(ErtsAtomCacheMap));
-	dst->ctx.dss.acmp = &dst->acm;
+    sys_memcpy(&dst->ctx, ctx, sizeof(ErtsDSigSendContext));
+    ASSERT(ctx->ctl == make_tuple(ctx->ctl_heap));
+    dst->ctx.ctl = make_tuple(dst->ctx.ctl_heap);
+    if (ctx->acmp) {
+	sys_memcpy(&dst->acm, ctx->acmp, sizeof(ErtsAtomCacheMap));
+	dst->ctx.acmp = &dst->acm;
     }
     return erts_mk_magic_ref(&hp, &MSO(p), ctx_bin);
 }
@@ -918,78 +918,58 @@ Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* ctx)
 ** Send a DOP_LINK link message
 */
 int
-erts_dsig_send_link(ErtsDSigData *dsdp, Eterm local, Eterm remote)
+erts_dsig_send_link(ErtsDSigSendContext *ctx, Eterm local, Eterm remote)
 {
-    DeclareTmpHeapNoproc(ctl_heap,4);
-    Eterm ctl = TUPLE3(&ctl_heap[0], make_small(DOP_LINK), local, remote);
-    int res;
-    UseTmpHeapNoproc(4);
-
-    res = dsig_send_ctl(dsdp, ctl, 0);
-    UnUseTmpHeapNoproc(4);
-    return res;
+    Eterm ctl = TUPLE3(&ctx->ctl_heap[0], make_small(DOP_LINK), local, remote);
+    return dsig_send_ctl(ctx, ctl);
 }
 
 int
-erts_dsig_send_unlink(ErtsDSigData *dsdp, Eterm local, Eterm remote)
+erts_dsig_send_unlink(ErtsDSigSendContext *ctx, Eterm local, Eterm remote)
 {
-    DeclareTmpHeapNoproc(ctl_heap,4);
-    Eterm ctl = TUPLE3(&ctl_heap[0], make_small(DOP_UNLINK), local, remote);
-    int res;
-
-    UseTmpHeapNoproc(4);
-    res = dsig_send_ctl(dsdp, ctl, 0);
-    UnUseTmpHeapNoproc(4);
-    return res;
+    Eterm ctl = TUPLE3(&ctx->ctl_heap[0], make_small(DOP_UNLINK), local, remote);
+    return dsig_send_ctl(ctx, ctl);
 }
 
 
 /* A local process that's being monitored by a remote one exits. We send:
    {DOP_MONITOR_P_EXIT, Local pid or name, Remote pid, ref, reason} */
 int
-erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm from, Eterm watcher, Eterm watched, 
+erts_dsig_send_m_exit(ErtsDSigSendContext *ctx, Eterm watcher, Eterm watched,
                       Eterm ref, Eterm reason)
 {
     Eterm ctl, msg;
-    DeclareTmpHeapNoproc(ctl_heap,6);
-    int res;
 
-    if (~dsdp->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
+    if (~ctx->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
         /*
          * Receiver does not support DOP_MONITOR_P_EXIT (see dsig_send_monitor)
          */
         return ERTS_DSIG_SEND_OK;
     }
 
-    UseTmpHeapNoproc(6);
-
-    if (dsdp->dep->flags & DFLAG_EXIT_PAYLOAD) {
-        ctl = TUPLE4(&ctl_heap[0], make_small(DOP_PAYLOAD_MONITOR_P_EXIT),
+    if (ctx->dep->flags & DFLAG_EXIT_PAYLOAD) {
+        ctl = TUPLE4(&ctx->ctl_heap[0], make_small(DOP_PAYLOAD_MONITOR_P_EXIT),
                      watched, watcher, ref);
         msg = reason;
     } else {
-        ctl = TUPLE5(&ctl_heap[0], make_small(DOP_MONITOR_P_EXIT),
+        ctl = TUPLE5(&ctx->ctl_heap[0], make_small(DOP_MONITOR_P_EXIT),
                      watched, watcher, ref, reason);
         msg = THE_NON_VALUE;
     }
 
-    res = dsig_send_exit(dsdp, ctl, msg, from, 1);
-    UnUseTmpHeapNoproc(6);
-    return res;
+    return dsig_send_exit(ctx, ctl, msg);
 }
 
 /* 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,
+erts_dsig_send_monitor(ErtsDSigSendContext *ctx, Eterm watcher, Eterm watched,
 		       Eterm ref)
 {
     Eterm ctl;
-    DeclareTmpHeapNoproc(ctl_heap,5);
-    int res;
 
-    if (~dsdp->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
+    if (~ctx->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
         /*
          * Receiver does not support DOP_MONITOR_P.
          * Just avoid sending it and by doing that reduce this monitor
@@ -999,48 +979,40 @@ erts_dsig_send_monitor(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
         return ERTS_DSIG_SEND_OK;
     }
 
-    UseTmpHeapNoproc(5);
-    ctl = TUPLE4(&ctl_heap[0],
+    ctl = TUPLE4(&ctx->ctl_heap[0],
 		 make_small(DOP_MONITOR_P),
 		 watcher, watched, ref);
 
-    res = dsig_send_ctl(dsdp, ctl, 0);
-    UnUseTmpHeapNoproc(5);
-    return res;
+    return dsig_send_ctl(ctx, ctl);
 }
 
 /* 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} */
 int
-erts_dsig_send_demonitor(ErtsDSigData *dsdp, Eterm watcher,
-			 Eterm watched, Eterm ref, int force)
+erts_dsig_send_demonitor(ErtsDSigSendContext *ctx, Eterm watcher,
+			 Eterm watched, Eterm ref)
 {
     Eterm ctl;
-    DeclareTmpHeapNoproc(ctl_heap,5);
-    int res;
 
-    if (~dsdp->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
+    if (~ctx->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
         /*
          * Receiver does not support DOP_DEMONITOR_P (see dsig_send_monitor)
          */
         return ERTS_DSIG_SEND_OK;
     }
 
-    UseTmpHeapNoproc(5);
-    ctl = TUPLE4(&ctl_heap[0],
+    ctl = TUPLE4(&ctx->ctl_heap[0],
 		 make_small(DOP_DEMONITOR_P),
 		 watcher, watched, ref);
 
-    res = dsig_send_ctl(dsdp, ctl, force);
-    UnUseTmpHeapNoproc(5);
-    return res;
+    return dsig_send_ctl(ctx, ctl);
 }
 
-static int can_send_seqtrace_token(ErtsSendContext* ctx, Eterm token) {
+static int can_send_seqtrace_token(ErtsDSigSendContext* ctx, Eterm token) {
     Eterm label;
 
-    if (ctx->dsd.flags & DFLAG_BIG_SEQTRACE_LABELS) {
+    if (ctx->flags & DFLAG_BIG_SEQTRACE_LABELS) {
         /* The other end is capable of handling arbitrary seq_trace labels. */
         return 1;
     }
@@ -1056,11 +1028,11 @@ static int can_send_seqtrace_token(ErtsSendContext* ctx, Eterm token) {
 }
 
 int
-erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
+erts_dsig_send_msg(ErtsDSigSendContext* ctx, Eterm remote, Eterm message)
 {
     Eterm ctl;
     Eterm token = NIL;
-    Process *sender = ctx->dsd.proc;
+    Process *sender = ctx->c_p;
     int res;
 #ifdef USE_VM_PROBES
     Sint tok_label = 0;
@@ -1081,7 +1053,7 @@ erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
     *node_name = *sender_name = *receiver_name = '\0';
     if (DTRACE_ENABLED(message_send) || DTRACE_ENABLED(message_send_remote)) {
         erts_snprintf(node_name, sizeof(DTRACE_CHARBUF_NAME(node_name)),
-                      "%T", ctx->dsd.dep->sysname);
+                      "%T", ctx->dep->sysname);
         erts_snprintf(sender_name, sizeof(DTRACE_CHARBUF_NAME(sender_name)),
                       "%T", sender->common.id);
         erts_snprintf(receiver_name, sizeof(DTRACE_CHARBUF_NAME(receiver_name)),
@@ -1101,7 +1073,7 @@ erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
 
         send_token = (token != NIL && can_send_seqtrace_token(ctx, token));
 
-        if (ctx->dsd.flags & DFLAG_SEND_SENDER) {
+        if (ctx->flags & DFLAG_SEND_SENDER) {
             dist_op = make_small(send_token ?
                                  DOP_SEND_SENDER_TT :
                                  DOP_SEND_SENDER);
@@ -1124,21 +1096,18 @@ erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
             msize, tok_label, tok_lastcnt, tok_serial);
     DTRACE7(message_send_remote, sender_name, node_name, receiver_name,
             msize, tok_label, tok_lastcnt, tok_serial);
-    ctx->dss.ctl = ctl;
-    ctx->dss.msg = message;
-    ctx->dss.force_busy = 0;
-    ctx->dss.force_encode = 0;
-    res = erts_dsig_send(&ctx->dsd, &ctx->dss);
+    ctx->ctl = ctl;
+    ctx->msg = message;
+    res = erts_dsig_send(ctx);
     return res;
 }
 
 int
-erts_dsig_send_reg_msg(Eterm remote_name, Eterm message, ErtsSendContext* ctx)
+erts_dsig_send_reg_msg(ErtsDSigSendContext* ctx, Eterm remote_name, Eterm message)
 {
     Eterm ctl;
     Eterm token = NIL;
-    Process *sender = ctx->dsd.proc;
-    int res;
+    Process *sender = ctx->c_p;
 #ifdef USE_VM_PROBES
     Sint tok_label = 0;
     Sint tok_lastcnt = 0;
@@ -1158,7 +1127,7 @@ erts_dsig_send_reg_msg(Eterm remote_name, Eterm message, ErtsSendContext* ctx)
     *node_name = *sender_name = *receiver_name = '\0';
     if (DTRACE_ENABLED(message_send) || DTRACE_ENABLED(message_send_remote)) {
         erts_snprintf(node_name, sizeof(DTRACE_CHARBUF_NAME(node_name)),
-                      "%T", ctx->dsd.dep->sysname);
+                      "%T", ctx->dep->sysname);
         erts_snprintf(sender_name, sizeof(DTRACE_CHARBUF_NAME(sender_name)),
                       "%T", sender->common.id);
         erts_snprintf(receiver_name, sizeof(DTRACE_CHARBUF_NAME(receiver_name)),
@@ -1183,24 +1152,19 @@ erts_dsig_send_reg_msg(Eterm remote_name, Eterm message, ErtsSendContext* ctx)
             msize, tok_label, tok_lastcnt, tok_serial);
     DTRACE7(message_send_remote, sender_name, node_name, receiver_name,
             msize, tok_label, tok_lastcnt, tok_serial);
-    ctx->dss.ctl = ctl;
-    ctx->dss.msg = message;
-    ctx->dss.force_busy = 0;
-    ctx->dss.force_encode = 0;
-    res = erts_dsig_send(&ctx->dsd, &ctx->dss);
-    return res;
+    ctx->ctl = ctl;
+    ctx->msg = message;
+    return erts_dsig_send(ctx);
 }
 
 /* local has died, deliver the exit signal to remote */
 int
-erts_dsig_send_exit_tt(ErtsDSigData *dsdp, Eterm local, Eterm remote, 
+erts_dsig_send_exit_tt(ErtsDSigSendContext *ctx, Eterm local, Eterm remote, 
 		       Eterm reason, Eterm token)
 {
     Eterm ctl, msg = THE_NON_VALUE;
-    DeclareTmpHeapNoproc(ctl_heap,6);
-    int res;
 #ifdef USE_VM_PROBES
-    Process *sender = dsdp->proc;
+    Process *sender = ctx->c_p;
     Sint tok_label = 0;
     Sint tok_lastcnt = 0;
     Sint tok_serial = 0;
@@ -1210,31 +1174,29 @@ erts_dsig_send_exit_tt(ErtsDSigData *dsdp, Eterm local, Eterm remote,
     DTRACE_CHARBUF(reason_str, 128);
 #endif
 
-    UseTmpHeapNoproc(6);
-
-    if (dsdp->dep->flags & DFLAG_EXIT_PAYLOAD)
+    if (ctx->dep->flags & DFLAG_EXIT_PAYLOAD)
         msg = reason;
 
     if (have_seqtrace(token)) {
-	seq_trace_update_send(dsdp->proc);
+	seq_trace_update_send(ctx->c_p);
 	seq_trace_output_exit(token, reason, SEQ_TRACE_SEND, remote, local);
-        if (dsdp->dep->flags & DFLAG_EXIT_PAYLOAD) {
-            ctl = TUPLE4(&ctl_heap[0],
+        if (ctx->dep->flags & DFLAG_EXIT_PAYLOAD) {
+            ctl = TUPLE4(&ctx->ctl_heap[0],
                          make_small(DOP_PAYLOAD_EXIT_TT), local, remote, token);
         } else
-            ctl = TUPLE5(&ctl_heap[0],
+            ctl = TUPLE5(&ctx->ctl_heap[0],
                          make_small(DOP_EXIT_TT), local, remote, token, reason);
     } else {
-        if (dsdp->dep->flags & DFLAG_EXIT_PAYLOAD)
-            ctl = TUPLE3(&ctl_heap[0], make_small(DOP_PAYLOAD_EXIT), local, remote);
+        if (ctx->dep->flags & DFLAG_EXIT_PAYLOAD)
+            ctl = TUPLE3(&ctx->ctl_heap[0], make_small(DOP_PAYLOAD_EXIT), local, remote);
         else
-            ctl = TUPLE4(&ctl_heap[0], make_small(DOP_EXIT), local, remote, reason);
+            ctl = TUPLE4(&ctx->ctl_heap[0], make_small(DOP_EXIT), local, remote, reason);
     }
 #ifdef USE_VM_PROBES
     *node_name = *sender_name = *remote_name = '\0';
     if (DTRACE_ENABLED(process_exit_signal_remote)) {
         erts_snprintf(node_name, sizeof(DTRACE_CHARBUF_NAME(node_name)),
-                      "%T", dsdp->dep->sysname);
+                      "%T", ctx->dep->sysname);
         erts_snprintf(sender_name, sizeof(DTRACE_CHARBUF_NAME(sender_name)),
                       "%T", sender->common.id);
         erts_snprintf(remote_name, sizeof(DTRACE_CHARBUF_NAME(remote_name)),
@@ -1250,39 +1212,30 @@ erts_dsig_send_exit_tt(ErtsDSigData *dsdp, Eterm local, Eterm remote,
 #endif
     DTRACE7(process_exit_signal_remote, sender_name, node_name,
             remote_name, reason_str, tok_label, tok_lastcnt, tok_serial);
-    res = dsig_send_exit(dsdp, ctl, msg, local, 1);
-    UnUseTmpHeapNoproc(6);
-    return res;
+    return dsig_send_exit(ctx, ctl, msg);
 }
 
 int
-erts_dsig_send_exit(ErtsDSigData *dsdp, Eterm from, Eterm local, Eterm remote, Eterm reason)
+erts_dsig_send_exit(ErtsDSigSendContext *ctx, Eterm local, Eterm remote, Eterm reason)
 {
-    DeclareTmpHeapNoproc(ctl_heap,5);
-    int res;
-    Eterm ctl, msg = dsdp->dep->flags & DFLAG_EXIT_PAYLOAD ? reason : THE_NON_VALUE;
-
-    UseTmpHeapNoproc(5);
+    Eterm ctl, msg = ctx->dep->flags & DFLAG_EXIT_PAYLOAD ? reason : THE_NON_VALUE;
 
-    if (dsdp->dep->flags & DFLAG_EXIT_PAYLOAD) {
-        ctl = TUPLE3(&ctl_heap[0], make_small(DOP_PAYLOAD_EXIT), local, remote);
+    if (ctx->dep->flags & DFLAG_EXIT_PAYLOAD) {
+        ctl = TUPLE3(&ctx->ctl_heap[0], make_small(DOP_PAYLOAD_EXIT), local, remote);
         msg = reason;
     } else {
-        ctl = TUPLE4(&ctl_heap[0], make_small(DOP_EXIT), local, remote, reason);
+        ctl = TUPLE4(&ctx->ctl_heap[0], make_small(DOP_EXIT), local, remote, reason);
         msg = THE_NON_VALUE;
     }
-    res = dsig_send_exit(dsdp, ctl, msg, from, 1);
-    UnUseTmpHeapNoproc(5);
-    return res;
+    return dsig_send_exit(ctx, ctl, msg);
 }
 
 int
-erts_dsig_send_exit2(ErtsSendContext *ctx, Eterm local, Eterm remote, Eterm reason)
+erts_dsig_send_exit2(ErtsDSigSendContext *ctx, Eterm local, Eterm remote, Eterm reason)
 {
-    int res;
     Eterm ctl, msg;
 
-    if (ctx->dsd.dep->flags & DFLAG_EXIT_PAYLOAD) {
+    if (ctx->dep->flags & DFLAG_EXIT_PAYLOAD) {
         ctl = TUPLE3(&ctx->ctl_heap[0],
                      make_small(DOP_PAYLOAD_EXIT2), local, remote);
         msg = reason;
@@ -1292,25 +1245,19 @@ erts_dsig_send_exit2(ErtsSendContext *ctx, Eterm local, Eterm remote, Eterm reas
         msg = THE_NON_VALUE;
     }
 
-    res = dsig_send_exit_ctx(ctx, ctl, msg);
-    return res;
+    return dsig_send_exit(ctx, ctl, msg);
 }
 
 
 int
-erts_dsig_send_group_leader(ErtsDSigData *dsdp, Eterm leader, Eterm remote)
+erts_dsig_send_group_leader(ErtsDSigSendContext *ctx, Eterm leader, Eterm remote)
 {
-    DeclareTmpHeapNoproc(ctl_heap,4);
-    int res;
     Eterm ctl;
 
-    UseTmpHeapNoproc(4);
-    ctl = TUPLE3(&ctl_heap[0],
+    ctl = TUPLE3(&ctx->ctl_heap[0],
 		 make_small(DOP_GROUP_LEADER), leader, remote);
 
-    res = dsig_send_ctl(dsdp, ctl, 0);
-    UnUseTmpHeapNoproc(4);
-    return res;
+    return dsig_send_ctl(ctx, ctl);
 }
 
 struct dist_sequences {
@@ -1666,7 +1613,7 @@ int erts_net_message(Port *prt,
 
     switch (type = unsigned_val(tuple[1])) {
     case DOP_LINK: {
-        ErtsDSigData dsd;
+        ErtsDSigSendContext ctx;
         int code;
 
 	if (tuple_arity != 3) {
@@ -1704,9 +1651,9 @@ int erts_net_message(Port *prt,
             erts_link_release_both(ldp);
         }
 
-        code = erts_dsig_prepare(&dsd, dep, NULL, 0, ERTS_DSP_NO_LOCK, 0, 0);
+        code = erts_dsig_prepare(&ctx, dep, NULL, 0, ERTS_DSP_NO_LOCK, 1, 1, 0);
         if (code == ERTS_DSIG_PREP_CONNECTED) {
-            code = erts_dsig_send_exit(&dsd, to, to, from, am_noproc);
+            code = erts_dsig_send_exit(&ctx, to, from, am_noproc);
             ASSERT(code == ERTS_DSIG_SEND_OK);
         }
 
@@ -1739,7 +1686,7 @@ int erts_net_message(Port *prt,
 	/* A remote process wants to monitor us, we get:
 	   {DOP_MONITOR_P, Remote pid, local pid or name, ref} */
 	Eterm pid, name;
-        ErtsDSigData dsd;
+        ErtsDSigSendContext ctx;
         int code;
 
 	if (tuple_arity != 4) {
@@ -1794,10 +1741,9 @@ int erts_net_message(Port *prt,
 
         }
 
-        code = erts_dsig_prepare(&dsd, dep, NULL, 0, ERTS_DSP_NO_LOCK, 0, 0);
+        code = erts_dsig_prepare(&ctx, dep, NULL, 0, ERTS_DSP_NO_LOCK, 1, 1, 0);
         if (code == ERTS_DSIG_PREP_CONNECTED) {
-            code = erts_dsig_send_m_exit(&dsd, pid, watcher, watched, ref,
-                                         am_noproc);
+            code = erts_dsig_send_m_exit(&ctx, watcher, watched, ref, am_noproc);
             ASSERT(code == ERTS_DSIG_SEND_OK);
         }
 
@@ -2152,45 +2098,21 @@ data_error:
     return -1;
 }
 
-static int dsig_send_exit(ErtsDSigData* dsdp, Eterm ctl, Eterm msg, Eterm from, int force_busy)
+static int dsig_send_exit(ErtsDSigSendContext *ctx, Eterm ctl, Eterm msg)
 {
-    struct erts_dsig_send_context ctx;
-    int ret;
-    ctx.ctl = ctl;
-    ctx.msg = msg;
-    ctx.from = from;
-    ctx.force_busy = force_busy;
-    ctx.force_encode = force_busy;
-    ctx.phase = ERTS_DSIG_SEND_PHASE_INIT;
-    ctx.reds = 1; /* provoke assert below (no reduction count with force_encode) */
-    ret = erts_dsig_send(dsdp, &ctx);
-    ASSERT(ret != ERTS_DSIG_SEND_CONTINUE);
-    return ret;
+    ctx->ctl = ctl;
+    ctx->msg = msg;
+    return erts_dsig_send(ctx);
 }
 
-static int dsig_send_exit_ctx(ErtsSendContext *ctx, Eterm ctl, Eterm msg)
+static int dsig_send_ctl(ErtsDSigSendContext *ctx, Eterm ctl)
 {
     int ret;
-    ctx->dss.ctl = ctl;
-    ctx->dss.msg = msg;
-    ctx->dss.force_busy = 0;
-    ctx->dss.force_encode = 0;
-    ret = erts_dsig_send(&ctx->dsd, &ctx->dss);
-    return ret;
-}
-
-static int dsig_send_ctl(ErtsDSigData* dsdp, Eterm ctl, int force_busy)
-{
-    struct erts_dsig_send_context ctx;
-    int ret;
-    ctx.ctl = ctl;
-    ctx.msg = THE_NON_VALUE;
-    ctx.from = THE_NON_VALUE;
-    ctx.force_busy = force_busy;
-    ctx.force_encode = 1;
-    ctx.phase = ERTS_DSIG_SEND_PHASE_INIT;
-    ctx.reds = 1; /* provoke assert below (no reduction count without msg) */
-    ret = erts_dsig_send(dsdp, &ctx);
+    ctx->ctl = ctl;
+    ctx->msg = THE_NON_VALUE;
+    ctx->from = THE_NON_VALUE;
+    ctx->reds = 1; /* provoke assert below (no reduction count without msg) */
+    ret = erts_dsig_send(ctx);
     ASSERT(ret != ERTS_DSIG_SEND_CONTINUE);
     return ret;
 }
@@ -2221,7 +2143,117 @@ notify_dist_data(Process *c_p, Eterm pid)
 }
 
 int
-erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
+erts_dsig_prepare(ErtsDSigSendContext *ctx,
+		  DistEntry *dep,
+		  Process *proc,
+                  ErtsProcLocks proc_locks,
+		  ErtsDSigPrepLock dspl,
+		  int no_suspend,
+                  int no_trap,
+		  int connect)
+{
+    int res;
+
+    if (!erts_is_alive)
+	return ERTS_DSIG_PREP_NOT_ALIVE;
+    if (!dep) {
+        ASSERT(!connect);
+        return ERTS_DSIG_PREP_NOT_CONNECTED;
+    }
+
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    if (connect) {
+        erts_proc_lc_might_unlock(proc, proc_locks);
+    }
+#endif
+
+retry:
+    erts_de_rlock(dep);
+
+    if (dep->state == ERTS_DE_STATE_CONNECTED) {
+	res = ERTS_DSIG_PREP_CONNECTED;
+    }
+    else if (dep->state == ERTS_DE_STATE_PENDING) {
+	res = ERTS_DSIG_PREP_PENDING;
+    }
+    else if (dep->state == ERTS_DE_STATE_EXITING) {
+	res = ERTS_DSIG_PREP_NOT_CONNECTED;
+	goto fail;
+    }
+    else if (connect) {
+        ASSERT(dep->state == ERTS_DE_STATE_IDLE);
+        erts_de_runlock(dep);
+        if (!erts_auto_connect(dep, proc, proc_locks)) {
+            return ERTS_DSIG_PREP_NOT_ALIVE;
+        }
+	goto retry;
+    }
+    else {
+        ASSERT(dep->state == ERTS_DE_STATE_IDLE);
+	res = ERTS_DSIG_PREP_NOT_CONNECTED;
+	goto fail;
+    }
+
+    if (no_suspend) {
+	if (erts_atomic32_read_acqb(&dep->qflgs) & ERTS_DE_QFLG_BUSY) {
+	    res = ERTS_DSIG_PREP_WOULD_SUSPEND;
+	    goto fail;
+	}
+    }
+
+    ctx->c_p = proc;
+    ctx->dep = dep;
+    ctx->deref_dep = 0;
+    ctx->cid = dep->cid;
+    ctx->connection_id = dep->connection_id;
+    ctx->no_suspend = no_suspend;
+    ctx->no_trap = no_trap;
+    ctx->flags = dep->flags;
+    ctx->return_term = am_true;
+    ctx->phase = ERTS_DSIG_SEND_PHASE_INIT;
+    ctx->from = proc ? proc->common.id : am_undefined;
+    ctx->reds = no_trap ? 1 : (Sint) (ERTS_BIF_REDS_LEFT(proc) * TERM_TO_BINARY_LOOP_FACTOR);
+    if (dspl == ERTS_DSP_NO_LOCK)
+	erts_de_runlock(dep);
+    return res;
+
+ fail:
+    erts_de_runlock(dep);
+    return res;
+}
+
+static
+void erts_schedule_dist_command(Port *prt, DistEntry *dist_entry)
+{
+    DistEntry *dep;
+    Eterm id;
+
+    if (prt) {
+	ERTS_LC_ASSERT(erts_lc_is_port_locked(prt));
+	ASSERT((erts_atomic32_read_nob(&prt->state)
+		& ERTS_PORT_SFLGS_DEAD) == 0);
+
+        dep = (DistEntry*) erts_prtsd_get(prt, ERTS_PRTSD_DIST_ENTRY);
+        ASSERT(dep);
+	id = prt->common.id;
+    }
+    else {
+	ASSERT(dist_entry);
+	ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&dist_entry->rwmtx)
+			   || erts_lc_rwmtx_is_rwlocked(&dist_entry->rwmtx));
+	ASSERT(is_internal_port(dist_entry->cid));
+
+ 	dep = dist_entry;
+	id = dep->cid;
+    }
+
+    if (!erts_atomic_xchg_mb(&dep->dist_cmd_scheduled, 1))
+	erts_port_task_schedule(id, &dep->dist_cmd, ERTS_PORT_TASK_DIST_CMD);
+}
+
+
+int
+erts_dsig_send(ErtsDSigSendContext *ctx)
 {
     int retval;
     Sint initial_reds = ctx->reds;
@@ -2230,11 +2262,13 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
     while (1) {
 	switch (ctx->phase) {
 	case ERTS_DSIG_SEND_PHASE_INIT:
-	    ctx->flags = dsdp->flags;
-	    ctx->c_p = dsdp->proc;
+	    ctx->flags = ctx->flags;
+	    ctx->c_p = ctx->c_p;
 
-	    if (!ctx->c_p || dsdp->no_suspend)
-		ctx->force_busy = 1;
+	    if (!ctx->c_p) {
+		ctx->no_trap = 1;
+		ctx->no_suspend = 1;
+            }
 
 	    ERTS_LC_ASSERT(!ctx->c_p
 			       || (ERTS_PROC_LOCK_MAIN
@@ -2274,7 +2308,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 
             ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_SIZE;
 	case ERTS_DSIG_SEND_PHASE_MSG_SIZE:
-            if (!ctx->force_encode) {
+            if (!ctx->no_trap) {
                 if (erts_encode_dist_ext_size_int(ctx->msg, ctx, &ctx->data_size)) {
                     retval = ERTS_DSIG_SEND_CONTINUE;
                     goto done;
@@ -2326,7 +2360,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 
             ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_ENCODE;
         case ERTS_DSIG_SEND_PHASE_MSG_ENCODE:
-            if (!ctx->force_encode) {
+            if (!ctx->no_trap) {
                 if (erts_encode_dist_ext(ctx->msg, &ctx->obuf->ext_endp, ctx->flags,
                                          ctx->acmp, &ctx->u.ec, &ctx->reds)) {
                     retval = ERTS_DSIG_SEND_CONTINUE;
@@ -2397,7 +2431,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                 for (; i < ctx->fragments; i++) {
                     free_dist_obuf(&ctx->obuf[i]);
                 }
-                if (!ctx->force_encode && !ctx->force_busy)
+                if (!ctx->no_trap && !ctx->no_suspend)
                     ctx->reds -= ctx->fragments;
                 ctx->fragments = fin_fragments + 1;
             }
@@ -2414,7 +2448,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 	     * Signal encoded; now verify that the connection still exists,
 	     * and if so enqueue the signal and schedule it for send.
 	     */
-            DistEntry *dep = dsdp->dep;
+            DistEntry *dep = ctx->dep;
 	    int suspended = 0;
 	    int resume = 0;
             int i;
@@ -2422,7 +2456,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 	    cid = dep->cid;
 	    if (dep->state == ERTS_DE_STATE_EXITING
                 || dep->state == ERTS_DE_STATE_IDLE
-                || dep->connection_id != dsdp->connection_id) {
+                || dep->connection_id != ctx->connection_id) {
 		/* Not the same connection as when we started; drop message... */
 		erts_de_runlock(dep);
                 for (i = 0; i < ctx->fragments; i++)
@@ -2443,20 +2477,20 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                 for (fragments = 0, obsz = 0;
                      fragments < ctx->fragments &&
                          ((ctx->reds > 0 && (qsize + obsz) < erts_dist_buf_busy_limit) ||
-                          ctx->force_encode || ctx->force_busy);
+                          ctx->no_trap || ctx->no_suspend);
                      fragments++) {
 #ifdef DEBUG
                     int reds = 100;
 #else
                     int reds = 10;
 #endif
-                    if (!ctx->force_encode && !ctx->force_busy)
+                    if (!ctx->no_trap && !ctx->no_suspend)
                         ctx->reds -= reds;
                     obsz += size_obuf(&ctx->obuf[fragments]);
                 }
 
                 ASSERT(fragments == ctx->fragments ||
-                       (!ctx->force_encode && !ctx->force_busy));
+                       (!ctx->no_trap && !ctx->no_suspend));
 
 		erts_mtx_lock(&dep->qlock);
 		qsize = erts_atomic_add_read_nob(&dep->qsize, (erts_aint_t) obsz);
@@ -2477,7 +2511,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                     /* else: requester will send itself the message... */
                     qflgs &= ~ERTS_DE_QFLG_REQ_INFO;
                 }
-		if (!ctx->force_busy && (qflgs & ERTS_DE_QFLG_BUSY)) {
+		if (!ctx->no_suspend && (qflgs & ERTS_DE_QFLG_BUSY)) {
 		    erts_mtx_unlock(&dep->qlock);
 
 		    plp = erts_proclist_create(ctx->c_p);
@@ -2506,7 +2540,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                     ctx->obuf = &ctx->obuf[fragments];
                 }
 
-		if (!ctx->force_busy) {
+		if (!ctx->no_suspend) {
                     qflgs = erts_atomic32_read_nob(&dep->qflgs);
 		    if (!(qflgs & ERTS_DE_QFLG_BUSY)) {
 			if (suspended)
@@ -4352,16 +4386,16 @@ monitor_node(Process* p, Eterm Node, Eterm Bool, Eterm Options)
     }
 
     case am_true: {
-        ErtsDSigData dsd;
-        dsd.node = Node;
+        ErtsDSigSendContext ctx;
+        ctx.node = Node;
 
         dep = erts_find_or_insert_dist_entry(Node);
         if (dep == erts_this_dist_entry)
             break;
 
-        switch (erts_dsig_prepare(&dsd, dep, p,
+        switch (erts_dsig_prepare(&ctx, dep, p,
                                   ERTS_PROC_LOCK_MAIN,
-                                  ERTS_DSP_RLOCK, 0, async_connect)) {
+                                  ERTS_DSP_RLOCK, 0, 0, async_connect)) {
         case ERTS_DSIG_PREP_NOT_ALIVE:
 	case ERTS_DSIG_PREP_NOT_CONNECTED:
 	    /* Trap to either send 'nodedown' or do passive connection attempt */