erts: Implement yielding for distributed send of large messages

Use same mechanism as term_to_binary to yield
while encoding large messages for distributed send.

6 files changed, 576 insertions, 309 deletions

diff --git a/erts/emulator/beam/atom.names b/erts/emulator/beam/atom.names
index 5d06a32941..ae2c6a3c67 100644
--- a/erts/emulator/beam/atom.names
+++ b/erts/emulator/beam/atom.names
@@ -198,6 +198,7 @@ atom dotall
 atom driver
 atom driver_options
 atom dsend
+atom dsend_continue_trap
 atom dunlink
 atom duplicate_bag
 atom dupnames
diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c
index fcbeb6cf5c..99566443aa 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 1996-2013. All Rights Reserved.
+ * Copyright Ericsson AB 1996-2014. 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
@@ -46,7 +46,7 @@ static Export* set_cpu_topology_trap = NULL;
 static Export* await_proc_exit_trap = NULL;
 static Export* await_port_send_result_trap = NULL;
 Export* erts_format_cpu_topology_trap = NULL;
-
+static Export dsend_continue_trap_export;
 static Export *await_sched_wall_time_mod_trap;
 static erts_smp_atomic32_t sched_wall_time;
 
@@ -1777,6 +1777,8 @@ BIF_RETTYPE whereis_1(BIF_ALIST_1)
  * erlang:'!'/2
  */
 
+HIPE_WRAPPER_BIF_DISABLE_GC(ebif_bang, 2)
+
 BIF_RETTYPE
 ebif_bang_2(BIF_ALIST_2)
 {
@@ -1795,34 +1797,36 @@ ebif_bang_2(BIF_ALIST_2)
 #define SEND_USER_ERROR		(-5)
 #define SEND_INTERNAL_ERROR	(-6)
 #define SEND_AWAIT_RESULT	(-7)
+#define SEND_YIELD_CONTINUE     (-8)
+
 
-Sint do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp);
+Sint do_send(Process *p, Eterm to, Eterm msg, Eterm *refp, ErtsSendContext*);
 
 static Sint remote_send(Process *p, DistEntry *dep,
-			Eterm to, Eterm full_to, Eterm msg, int suspend)
+			Eterm to, Eterm full_to, Eterm msg,
+			ErtsSendContext* ctx)
 {
     Sint res;
     int code;
-    ErtsDSigData dsd;
 
     ASSERT(is_atom(to) || is_external_pid(to));
 
-    code = erts_dsig_prepare(&dsd, dep, p, ERTS_DSP_NO_LOCK, !suspend);
+    code = erts_dsig_prepare(&ctx->dsd, dep, p, ERTS_DSP_NO_LOCK, !ctx->suspend);
     switch (code) {
     case ERTS_DSIG_PREP_NOT_ALIVE:
     case ERTS_DSIG_PREP_NOT_CONNECTED:
 	res = SEND_TRAP;
 	break;
     case ERTS_DSIG_PREP_WOULD_SUSPEND:
-	ASSERT(!suspend);
+	ASSERT(!ctx->suspend);
 	res = SEND_YIELD;
 	break;
     case ERTS_DSIG_PREP_CONNECTED: {
 
 	if (is_atom(to))
-	    code = erts_dsig_send_reg_msg(&dsd, to, msg);
+	    code = erts_dsig_send_reg_msg(to, msg, ctx);
 	else
-	    code = erts_dsig_send_msg(&dsd, to, msg);
+	    code = erts_dsig_send_msg(to, msg, ctx);
 	/*
 	 * Note that reductions have been bumped on calling
 	 * process by erts_dsig_send_reg_msg() or
@@ -1830,6 +1834,8 @@ static Sint remote_send(Process *p, DistEntry *dep,
 	 */
 	if (code == ERTS_DSIG_SEND_YIELD)
 	    res = SEND_YIELD_RETURN;
+	else if (code == ERTS_DSIG_SEND_CONTINUE)
+	    res = SEND_YIELD_CONTINUE;
 	else
 	    res = 0;
 	break;
@@ -1850,7 +1856,8 @@ static Sint remote_send(Process *p, DistEntry *dep,
 }
 
 Sint
-do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
+do_send(Process *p, Eterm to, Eterm msg, Eterm *refp, ErtsSendContext* ctx)
+{
     Eterm portid;
     Port *pt;
     Process* rp;
@@ -1881,7 +1888,7 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
 	    erts_send_error_to_logger(p->group_leader, dsbufp);
 	    return 0;
 	}
-	return remote_send(p, dep, to, to, msg, suspend);
+	return remote_send(p, dep, to, to, msg, ctx);
     } else if (is_atom(to)) {
 	Eterm id = erts_whereis_name_to_id(p, to);
 
@@ -1936,7 +1943,7 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
 	ret_val = 0;
         
 	if (pt) {
-	    int ps_flags = suspend ? 0 : ERTS_PORT_SIG_FLG_NOSUSPEND;
+	    int ps_flags = ctx->suspend ? 0 : ERTS_PORT_SIG_FLG_NOSUSPEND;
 	    *refp = NIL;
 
 	    switch (erts_port_command(p, ps_flags, pt, msg, refp)) {
@@ -1945,12 +1952,12 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
 		return SEND_USER_ERROR;
 	    case ERTS_PORT_OP_BUSY:
 		/* Nothing has been sent */
-		if (suspend)
+		if (ctx->suspend)
 		    erts_suspend(p, ERTS_PROC_LOCK_MAIN, pt);
 		return SEND_YIELD;
 	    case ERTS_PORT_OP_BUSY_SCHEDULED:
 		/* Message was sent */
-		if (suspend) {
+		if (ctx->suspend) {
 		    erts_suspend(p, ERTS_PROC_LOCK_MAIN, pt);
 		    ret_val = SEND_YIELD_RETURN;
 		    break;
@@ -2030,9 +2037,14 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
 	    return 0;
 	}
 
-	ret = remote_send(p, dep, tp[1], to, msg, suspend);
-	if (dep)
-	    erts_deref_dist_entry(dep);
+	ret = remote_send(p, dep, tp[1], to, msg, ctx);
+	if (ret != SEND_YIELD_CONTINUE) {
+	    if (dep) {
+		erts_deref_dist_entry(dep);
+	    }
+	} else {
+	    ctx->dep_to_deref = dep;
+	}
 	return ret;
     } else {
 	if (IS_TRACED(p)) /* XXX Is this really neccessary ??? */
@@ -2063,9 +2075,11 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
     }
 }
 
+HIPE_WRAPPER_BIF_DISABLE_GC(send, 3)
 
 BIF_RETTYPE send_3(BIF_ALIST_3)
 {
+    BIF_RETTYPE retval;
     Eterm ref;
     Process *p = BIF_P;
     Eterm to = BIF_ARG_1;
@@ -2073,34 +2087,44 @@ BIF_RETTYPE send_3(BIF_ALIST_3)
     Eterm opts = BIF_ARG_3;
 
     int connect = !0;
-    int suspend = !0;
     Eterm l = opts;
     Sint result;
-    
+    DeclareTypedTmpHeap(ErtsSendContext, ctx, BIF_P);
+    UseTmpHeap(sizeof(ErtsSendContext)/sizeof(Eterm), BIF_P);
+
+    ctx->suspend = !0;
+    ctx->dep_to_deref = NULL;
+    ctx->return_term = am_ok;
+    ctx->dss.reds = (Sint) (ERTS_BIF_REDS_LEFT(p) * TERM_TO_BINARY_LOOP_FACTOR);
+    ctx->dss.phase = ERTS_DSIG_SEND_PHASE_INIT;
+
     while (is_list(l)) {
 	if (CAR(list_val(l)) == am_noconnect) {
 	    connect = 0;
 	} else if (CAR(list_val(l)) == am_nosuspend) {
-	    suspend = 0;
+	    ctx->suspend = 0;
 	} else {
-	    BIF_ERROR(p, BADARG);
+	    ERTS_BIF_PREP_ERROR(retval, p, BADARG);
+	    goto done;
 	}
 	l = CDR(list_val(l));
     }
     if(!is_nil(l)) {
-	BIF_ERROR(p, BADARG);
+	ERTS_BIF_PREP_ERROR(retval, p, BADARG);
+	goto done;
     }
 
 #ifdef DEBUG
     ref = NIL;
 #endif
 
-    result = do_send(p, to, msg, suspend, &ref);
+    result = do_send(p, to, msg, &ref, ctx);
     if (result > 0) {
 	ERTS_VBUMP_REDS(p, result);
 	if (ERTS_IS_PROC_OUT_OF_REDS(p))
 	    goto yield_return;
-	BIF_RET(am_ok);
+	ERTS_BIF_PREP_RET(retval, am_ok);
+	goto done;
     }
 
     switch (result) {
@@ -2108,68 +2132,127 @@ BIF_RETTYPE send_3(BIF_ALIST_3)
 	/* May need to yield even though we do not bump reds here... */
 	if (ERTS_IS_PROC_OUT_OF_REDS(p))
 	    goto yield_return;
-	BIF_RET(am_ok); 
+	ERTS_BIF_PREP_RET(retval, am_ok);
 	break;
     case SEND_TRAP:
 	if (connect) {
-	    BIF_TRAP3(dsend3_trap, p, to, msg, opts); 
+	    ERTS_BIF_PREP_TRAP3(retval, dsend3_trap, p, to, msg, opts);
 	} else {
-	    BIF_RET(am_noconnect);
+	    ERTS_BIF_PREP_RET(retval, am_noconnect);
 	}
 	break;
     case SEND_YIELD:
-	if (suspend) {
-	    ERTS_BIF_YIELD3(bif_export[BIF_send_3], p, to, msg, opts);
+	if (ctx->suspend) {
+	    ERTS_BIF_PREP_YIELD3(retval,
+				 bif_export[BIF_send_3], p, to, msg, opts);
 	} else {
-	    BIF_RET(am_nosuspend);
+	    ERTS_BIF_PREP_RET(retval, am_nosuspend);
 	}
 	break;
     case SEND_YIELD_RETURN:
-	if (!suspend)
-	    BIF_RET(am_nosuspend);
+	if (!ctx->suspend) {
+	    ERTS_BIF_PREP_RET(retval, am_nosuspend);
+	    break;
+	}
     yield_return:
-	ERTS_BIF_YIELD_RETURN(p, am_ok);
+	ERTS_BIF_PREP_YIELD_RETURN(retval, p, am_ok);
+        break;
     case SEND_AWAIT_RESULT:
 	ASSERT(is_internal_ref(ref));
-	BIF_TRAP3(await_port_send_result_trap, p, ref, am_nosuspend, am_ok);
+	ERTS_BIF_PREP_TRAP3(retval, await_port_send_result_trap, p, ref, am_nosuspend, am_ok);
+	break;
     case SEND_BADARG:
-	BIF_ERROR(p, BADARG); 
+	ERTS_BIF_PREP_ERROR(retval, p, BADARG);
 	break;
     case SEND_USER_ERROR:
-	BIF_ERROR(p, EXC_ERROR); 
+	ERTS_BIF_PREP_ERROR(retval, p, EXC_ERROR);
 	break;
     case SEND_INTERNAL_ERROR:
-	BIF_ERROR(p, EXC_INTERNAL_ERROR);
+	ERTS_BIF_PREP_ERROR(retval, p, EXC_INTERNAL_ERROR);
+	break;
+    case SEND_YIELD_CONTINUE:
+	BUMP_ALL_REDS(p);
+	erts_set_gc_state(p, 0);
+	ERTS_BIF_PREP_TRAP1(retval, &dsend_continue_trap_export, p,
+			    erts_dsend_export_trap_context(p, ctx));
 	break;
     default:
-	ASSERT(! "Illegal send result"); 
+	erl_exit(ERTS_ABORT_EXIT, "send_3 invalid result %d\n", (int)result);
 	break;
     }
-    ASSERT(! "Can not arrive here");
-    BIF_ERROR(p, BADARG);
+
+done:
+    UnUseTmpHeap(sizeof(ErtsSendContext)/sizeof(Eterm), BIF_P);
+    return retval;
 }
 
+HIPE_WRAPPER_BIF_DISABLE_GC(send, 2)
+
 BIF_RETTYPE send_2(BIF_ALIST_2)
 {
     return erl_send(BIF_P, BIF_ARG_1, BIF_ARG_2);
 }
 
+static BIF_RETTYPE dsend_continue_trap_1(BIF_ALIST_1)
+{
+    Binary* bin = ((ProcBin*) binary_val(BIF_ARG_1))->val;
+    ErtsSendContext* ctx = (ErtsSendContext*) ERTS_MAGIC_BIN_DATA(bin);
+    Sint initial_reds = (Sint) (ERTS_BIF_REDS_LEFT(BIF_P) * TERM_TO_BINARY_LOOP_FACTOR);
+    int result;
+
+    ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == erts_dsend_context_dtor);
+
+    ctx->dss.reds = initial_reds;
+    result = erts_dsig_send(&ctx->dsd, &ctx->dss);
+
+    switch (result) {
+    case ERTS_DSIG_SEND_OK:
+	erts_set_gc_state(BIF_P, 1);
+	BIF_RET(ctx->return_term);
+	break;
+    case ERTS_DSIG_SEND_YIELD: /*SEND_YIELD_RETURN*/
+	erts_set_gc_state(BIF_P, 1);
+	if (!ctx->suspend)
+	    BIF_RET(am_nosuspend);
+	ERTS_BIF_YIELD_RETURN(BIF_P, ctx->return_term);
+
+    case ERTS_DSIG_SEND_CONTINUE: { /*SEND_YIELD_CONTINUE*/
+	BUMP_ALL_REDS(BIF_P);
+	BIF_TRAP1(&dsend_continue_trap_export, BIF_P, BIF_ARG_1);
+    }
+    default:
+	erl_exit(ERTS_ABORT_EXIT, "dsend_continue_trap invalid result %d\n", (int)result);
+	break;
+    }
+    ASSERT(! "Can not arrive here");
+    BIF_ERROR(BIF_P, BADARG);
+}
+
 Eterm erl_send(Process *p, Eterm to, Eterm msg)
 {
+    Eterm retval;
     Eterm ref;
     Sint result;
+    DeclareTypedTmpHeap(ErtsSendContext, ctx, p);
+    UseTmpHeap(sizeof(ErtsSendContext)/sizeof(Eterm), p);
 
 #ifdef DEBUG
     ref = NIL;
 #endif
+    ctx->suspend = !0;
+    ctx->dep_to_deref = NULL;
+    ctx->return_term = msg;
+    ctx->dss.reds = (Sint) (ERTS_BIF_REDS_LEFT(p) * TERM_TO_BINARY_LOOP_FACTOR);
+    ctx->dss.phase = ERTS_DSIG_SEND_PHASE_INIT;
 
-    result = do_send(p, to, msg, !0, &ref);
+    result = do_send(p, to, msg, &ref, ctx);
     
     if (result > 0) {
 	ERTS_VBUMP_REDS(p, result);
 	if (ERTS_IS_PROC_OUT_OF_REDS(p))
 	    goto yield_return;
-	BIF_RET(msg);
+	ERTS_BIF_PREP_RET(retval, msg);
+	goto done;
     }
 
     switch (result) {
@@ -2177,35 +2260,46 @@ Eterm erl_send(Process *p, Eterm to, Eterm msg)
 	/* May need to yield even though we do not bump reds here... */
 	if (ERTS_IS_PROC_OUT_OF_REDS(p))
 	    goto yield_return;
-	BIF_RET(msg); 
+	ERTS_BIF_PREP_RET(retval, msg);
 	break;
     case SEND_TRAP:
-	BIF_TRAP2(dsend2_trap, p, to, msg); 
+	ERTS_BIF_PREP_TRAP2(retval, dsend2_trap, p, to, msg);
 	break;
     case SEND_YIELD:
-	ERTS_BIF_YIELD2(bif_export[BIF_send_2], p, to, msg);
+	ERTS_BIF_PREP_YIELD2(retval, bif_export[BIF_send_2], p, to, msg);
 	break;
     case SEND_YIELD_RETURN:
     yield_return:
-	ERTS_BIF_YIELD_RETURN(p, msg);
+	ERTS_BIF_PREP_YIELD_RETURN(retval, p, msg);
+        break;
     case SEND_AWAIT_RESULT:
 	ASSERT(is_internal_ref(ref));
-	BIF_TRAP3(await_port_send_result_trap, p, ref, msg, msg);
+	ERTS_BIF_PREP_TRAP3(retval,
+			    await_port_send_result_trap, p, ref, msg, msg);
+	break;
     case SEND_BADARG:
-	BIF_ERROR(p, BADARG); 
+	ERTS_BIF_PREP_ERROR(retval, p, BADARG);
 	break;
     case SEND_USER_ERROR:
-	BIF_ERROR(p, EXC_ERROR); 
+	ERTS_BIF_PREP_ERROR(retval, p, EXC_ERROR);
 	break;
     case SEND_INTERNAL_ERROR:
-	BIF_ERROR(p, EXC_INTERNAL_ERROR);
+	ERTS_BIF_PREP_ERROR(retval, p, EXC_INTERNAL_ERROR);
+	break;
+    case SEND_YIELD_CONTINUE:
+	BUMP_ALL_REDS(p);
+	erts_set_gc_state(p, 0);
+	ERTS_BIF_PREP_TRAP1(retval, &dsend_continue_trap_export, p,
+			    erts_dsend_export_trap_context(p, ctx));
 	break;
     default:
-	ASSERT(! "Illegal send result"); 
+	erl_exit(ERTS_ABORT_EXIT, "invalid send result %d\n", (int)result);
 	break;
     }
-    ASSERT(! "Can not arrive here");
-    BIF_ERROR(p, BADARG);
+
+done:
+    UnUseTmpHeap(sizeof(ErtsSendContext)/sizeof(Eterm), p);
+    return retval;
 }
 
 /**********************************************************************/
@@ -4809,6 +4903,10 @@ void erts_init_bif(void)
 #endif
 		     , &bif_return_trap);
 
+    erts_init_trap_export(&dsend_continue_trap_export,
+			  am_erts_internal, am_dsend_continue_trap, 1,
+			  dsend_continue_trap_1);
+
     flush_monitor_message_trap = erts_export_put(am_erlang,
 						 am_flush_monitor_message,
 						 2);
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index ec07ddcd9c..f73374ba71 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 1996-2013. All Rights Reserved.
+ * Copyright Ericsson AB 1996-2014. 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
@@ -119,7 +119,7 @@ Export* dmonitor_p_trap = NULL;
 /* forward declarations */
 
 static void clear_dist_entry(DistEntry*);
-static int dsig_send(ErtsDSigData *, Eterm, Eterm, int);
+static int dsig_send_ctl(ErtsDSigData* dsdp, Eterm ctl, int force_busy);
 static void send_nodes_mon_msgs(Process *, Eterm, Eterm, Eterm, Eterm);
 static void init_nodes_monitors(void);
 
@@ -709,6 +709,55 @@ static void clear_dist_entry(DistEntry *dep)
     }
 }
 
+void erts_dsend_context_dtor(Binary* ctx_bin)
+{
+    ErtsSendContext* ctx = ERTS_MAGIC_BIN_DATA(ctx_bin);
+    switch (ctx->dss.phase) {
+    case ERTS_DSIG_SEND_PHASE_MSG_SIZE:
+	DESTROY_SAVED_ESTACK(&ctx->dss.u.sc.estack);
+	break;
+    case ERTS_DSIG_SEND_PHASE_MSG_ENCODE:
+	DESTROY_SAVED_WSTACK(&ctx->dss.u.ec.wstack);
+	break;
+    default:;
+    }
+    if (ctx->dss.phase >= ERTS_DSIG_SEND_PHASE_ALLOC && ctx->dss.obuf) {
+	free_dist_obuf(ctx->dss.obuf);
+    }
+    if (ctx->dep_to_deref)
+	erts_deref_dist_entry(ctx->dep_to_deref);
+}
+
+Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* ctx)
+{
+    struct exported_ctx {
+	ErtsSendContext ctx;
+	ErtsAtomCacheMap acm;
+    };
+    Binary* ctx_bin = erts_create_magic_binary(sizeof(struct exported_ctx),
+					       erts_dsend_context_dtor);
+    struct exported_ctx* dst = ERTS_MAGIC_BIN_DATA(ctx_bin);
+    Uint ctl_size = !HALFWORD_HEAP ? 0 : (arityval(ctx->ctl_heap[0]) + 1);
+    Eterm* hp = HAlloc(p, ctl_size + PROC_BIN_SIZE);
+
+    sys_memcpy(&dst->ctx, ctx, sizeof(ErtsSendContext));
+    ASSERT(ctx->dss.ctl == make_tuple(ctx->ctl_heap));
+#if !HALFWORD_HEAP
+    dst->ctx.dss.ctl = make_tuple(dst->ctx.ctl_heap);
+#else
+    /* Must put control tuple in low mem */
+    sys_memcpy(hp, ctx->ctl_heap,  ctl_size*sizeof(Eterm));
+    dst->ctx.dss.ctl = make_tuple(hp);
+    hp += ctl_size;
+#endif
+    if (ctx->dss.acmp) {
+	sys_memcpy(&dst->acm, ctx->dss.acmp, sizeof(ErtsAtomCacheMap));
+	dst->ctx.dss.acmp = &dst->acm;
+    }
+    return erts_mk_magic_binary_term(&hp, &MSO(p), ctx_bin);
+}
+
+
 /*
  * The erts_dsig_send_*() functions implemented below, sends asynchronous
  * distributed signals to other Erlang nodes. Before sending a distributed
@@ -731,7 +780,7 @@ erts_dsig_send_link(ErtsDSigData *dsdp, Eterm local, Eterm remote)
     int res;
     UseTmpHeapNoproc(4);
 
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+    res = dsig_send_ctl(dsdp, ctl, 0);
     UnUseTmpHeapNoproc(4);
     return res;
 }
@@ -744,7 +793,7 @@ erts_dsig_send_unlink(ErtsDSigData *dsdp, Eterm local, Eterm remote)
     int res;
 
     UseTmpHeapNoproc(4);
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+    res = dsig_send_ctl(dsdp, ctl, 0);
     UnUseTmpHeapNoproc(4);
     return res;
 }
@@ -772,7 +821,7 @@ erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
     erts_smp_de_links_unlock(dsdp->dep);
 #endif
 
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 1);
+    res = dsig_send_ctl(dsdp, ctl, 1);
     UnUseTmpHeapNoproc(6);
     return res;
 }
@@ -793,7 +842,7 @@ erts_dsig_send_monitor(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
 		 make_small(DOP_MONITOR_P),
 		 watcher, watched, ref);
 
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+    res = dsig_send_ctl(dsdp, ctl, 0);
     UnUseTmpHeapNoproc(5);
     return res;
 }
@@ -815,18 +864,17 @@ erts_dsig_send_demonitor(ErtsDSigData *dsdp, Eterm watcher,
 		 make_small(DOP_DEMONITOR_P),
 		 watcher, watched, ref);
 
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, force);
+    res = dsig_send_ctl(dsdp, ctl, force);
     UnUseTmpHeapNoproc(5);
     return res;
 }
 
 int
-erts_dsig_send_msg(ErtsDSigData *dsdp, Eterm remote, Eterm message)
+erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
 {
     Eterm ctl;
-    DeclareTmpHeapNoproc(ctl_heap,5);
     Eterm token = NIL;
-    Process *sender = dsdp->proc;
+    Process *sender = ctx->dsd.proc;
     int res;
 #ifdef USE_VM_PROBES
     Sint tok_label = 0;
@@ -838,8 +886,7 @@ erts_dsig_send_msg(ErtsDSigData *dsdp, Eterm remote, Eterm message)
     DTRACE_CHARBUF(receiver_name, 64);
 #endif
 
-    UseTmpHeapNoproc(5);
-    if (SEQ_TRACE_TOKEN(sender) != NIL 
+    if (SEQ_TRACE_TOKEN(sender) != NIL
 #ifdef USE_VM_PROBES
 	&& SEQ_TRACE_TOKEN(sender) != am_have_dt_utag 
 #endif
@@ -852,7 +899,7 @@ erts_dsig_send_msg(ErtsDSigData *dsdp, Eterm remote, Eterm message)
     *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", dsdp->dep->sysname);
+                      "%T", ctx->dsd.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)),
@@ -867,26 +914,28 @@ erts_dsig_send_msg(ErtsDSigData *dsdp, Eterm remote, Eterm message)
 #endif
 
     if (token != NIL)
-	ctl = TUPLE4(&ctl_heap[0],
+	ctl = TUPLE4(&ctx->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);
+	ctl = TUPLE3(&ctx->ctl_heap[0], make_small(DOP_SEND), am_Cookie, remote);
     DTRACE6(message_send, sender_name, receiver_name,
             msize, tok_label, tok_lastcnt, tok_serial);
     DTRACE7(message_send_remote, sender_name, node_name, receiver_name,
             msize, tok_label, tok_lastcnt, tok_serial);
-    res = dsig_send(dsdp, ctl, message, 0);
-    UnUseTmpHeapNoproc(5);
+    ctx->dss.ctl = ctl;
+    ctx->dss.msg = message;
+    ctx->dss.force_busy = 0;
+    res = erts_dsig_send(&ctx->dsd, &ctx->dss);
     return res;
 }
 
 int
-erts_dsig_send_reg_msg(ErtsDSigData *dsdp, Eterm remote_name, Eterm message)
+erts_dsig_send_reg_msg(Eterm remote_name, Eterm message,
+		       ErtsSendContext* ctx)
 {
     Eterm ctl;
-    DeclareTmpHeapNoproc(ctl_heap,6);
     Eterm token = NIL;
-    Process *sender = dsdp->proc;
+    Process *sender = ctx->dsd.proc;
     int res;
 #ifdef USE_VM_PROBES
     Sint tok_label = 0;
@@ -898,7 +947,6 @@ erts_dsig_send_reg_msg(ErtsDSigData *dsdp, Eterm remote_name, Eterm message)
     DTRACE_CHARBUF(receiver_name, 128);
 #endif
 
-    UseTmpHeapNoproc(6);
     if (SEQ_TRACE_TOKEN(sender) != NIL
 #ifdef USE_VM_PROBES
 	&& SEQ_TRACE_TOKEN(sender) != am_have_dt_utag 
@@ -912,7 +960,7 @@ erts_dsig_send_reg_msg(ErtsDSigData *dsdp, Eterm remote_name, Eterm message)
     *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", dsdp->dep->sysname);
+                      "%T", ctx->dsd.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)),
@@ -927,17 +975,19 @@ erts_dsig_send_reg_msg(ErtsDSigData *dsdp, Eterm remote_name, Eterm message)
 #endif
 
     if (token != NIL)
-	ctl = TUPLE5(&ctl_heap[0], make_small(DOP_REG_SEND_TT),
+	ctl = TUPLE5(&ctx->ctl_heap[0], make_small(DOP_REG_SEND_TT),
 		     sender->common.id, am_Cookie, remote_name, token);
     else
-	ctl = TUPLE4(&ctl_heap[0], make_small(DOP_REG_SEND),
+	ctl = TUPLE4(&ctx->ctl_heap[0], make_small(DOP_REG_SEND),
 		     sender->common.id, am_Cookie, remote_name);
     DTRACE6(message_send, sender_name, receiver_name,
             msize, tok_label, tok_lastcnt, tok_serial);
     DTRACE7(message_send_remote, sender_name, node_name, receiver_name,
             msize, tok_label, tok_lastcnt, tok_serial);
-    res = dsig_send(dsdp, ctl, message, 0);
-    UnUseTmpHeapNoproc(6);
+    ctx->dss.ctl = ctl;
+    ctx->dss.msg = message;
+    ctx->dss.force_busy = 0;
+    res = erts_dsig_send(&ctx->dsd, &ctx->dss);
     return res;
 }
 
@@ -994,7 +1044,7 @@ erts_dsig_send_exit_tt(ErtsDSigData *dsdp, Eterm local, Eterm remote,
     DTRACE7(process_exit_signal_remote, sender_name, node_name,
             remote_name, reason_str, tok_label, tok_lastcnt, tok_serial);
     /* forced, i.e ignore busy */
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 1);
+    res = dsig_send_ctl(dsdp, ctl, 1);
     UnUseTmpHeapNoproc(6);
     return res;
 }
@@ -1010,7 +1060,7 @@ erts_dsig_send_exit(ErtsDSigData *dsdp, Eterm local, Eterm remote, Eterm reason)
     ctl = TUPLE4(&ctl_heap[0],
 		 make_small(DOP_EXIT), local, remote, reason);
     /* forced, i.e ignore busy */
-    res =  dsig_send(dsdp, ctl, THE_NON_VALUE, 1);
+    res =  dsig_send_ctl(dsdp, ctl, 1);
     UnUseTmpHeapNoproc(5);
     return res;
 }
@@ -1026,7 +1076,7 @@ erts_dsig_send_exit2(ErtsDSigData *dsdp, Eterm local, Eterm remote, Eterm reason
     ctl = TUPLE4(&ctl_heap[0],
 		 make_small(DOP_EXIT2), local, remote, reason);
 
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+    res = dsig_send_ctl(dsdp, ctl, 0);
     UnUseTmpHeapNoproc(5);
     return res;
 }
@@ -1043,7 +1093,7 @@ erts_dsig_send_group_leader(ErtsDSigData *dsdp, Eterm leader, Eterm remote)
     ctl = TUPLE3(&ctl_heap[0],
 		 make_small(DOP_GROUP_LEADER), leader, remote);
 
-    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
+    res = dsig_send_ctl(dsdp, ctl, 0);
     UnUseTmpHeapNoproc(4);
     return res;
 }
@@ -1693,194 +1743,235 @@ int erts_net_message(Port *prt,
     return -1;
 }
 
-static int
-dsig_send(ErtsDSigData *dsdp, Eterm ctl, Eterm msg, int force_busy)
+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.force_busy = force_busy;
+    ctx.phase = ERTS_DSIG_SEND_PHASE_INIT;
+#ifdef DEBUG
+    ctx.reds = 1; /* provoke assert below (no reduction count without msg) */
+#endif
+    ret = erts_dsig_send(dsdp, &ctx);
+    ASSERT(ret != ERTS_DSIG_SEND_CONTINUE);
+    return ret;
+}
+
+int
+erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
+{
+    int retval;
+    Sint initial_reds = ctx->reds;
     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;
+    while (1) {
+	switch (ctx->phase) {
+	case ERTS_DSIG_SEND_PHASE_INIT:
+	    ctx->flags = dsdp->dep->flags;
+	    ctx->c_p = dsdp->proc;
 
-    ERTS_SMP_LC_ASSERT(!c_p
-		       || (ERTS_PROC_LOCK_MAIN
-			   == erts_proc_lc_my_proc_locks(c_p)));
+	    if (!ctx->c_p || dsdp->no_suspend)
+		ctx->force_busy = 1;
 
-    if (!erts_is_alive)
-	return ERTS_DSIG_SEND_OK;
+	    ERTS_SMP_LC_ASSERT(!ctx->c_p
+			       || (ERTS_PROC_LOCK_MAIN
+				   == erts_proc_lc_my_proc_locks(ctx->c_p)));
 
-    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;
-    }
+	    if (!erts_is_alive)
+		return ERTS_DSIG_SEND_OK;
 
-#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
+	    if (ctx->flags & DFLAG_DIST_HDR_ATOM_CACHE) {
+		ctx->acmp = erts_get_atom_cache_map(ctx->c_p);
+		ctx->pass_through_size = 0;
+	    }
+	    else {
+		ctx->acmp = NULL;
+		ctx->pass_through_size = 1;
+	    }
 
-    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, flags);
+    #ifdef ERTS_DIST_MSG_DBG
+	    erts_fprintf(stderr, ">>%s CTL: %T\n", ctx->pass_through_size ? "P" : " ", ctx->ctl);
+	    if (is_value(msg))
+		erts_fprintf(stderr, "    MSG: %T\n", msg);
+    #endif
+
+	    ctx->data_size = ctx->pass_through_size;
+	    erts_reset_atom_cache_map(ctx->acmp);
+	    erts_encode_dist_ext_size(ctx->ctl, ctx->flags, ctx->acmp, &ctx->data_size);
+
+	    if (is_value(ctx->msg)) {
+		ctx->u.sc.estack.start = NULL;
+		ctx->u.sc.flags = ctx->flags;
+		ctx->u.sc.level = 0;
+		ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_SIZE;
+	    } else {
+		ctx->phase = ERTS_DSIG_SEND_PHASE_ALLOC;
+	    }
+	    break;
 
-    dhdr_ext_size = erts_encode_ext_dist_header_size(acmp);
-    data_size += dhdr_ext_size;
+	case ERTS_DSIG_SEND_PHASE_MSG_SIZE:
+	    if (erts_encode_dist_ext_size_int(ctx->msg, ctx, &ctx->data_size)) {
+		retval = ERTS_DSIG_SEND_CONTINUE;
+		goto done;
+	    }
 
-    obuf = alloc_dist_obuf(data_size);
-    obuf->ext_endp = &obuf->data[0] + pass_through_size + dhdr_ext_size;
+	    ctx->phase = ERTS_DSIG_SEND_PHASE_ALLOC;
+	case ERTS_DSIG_SEND_PHASE_ALLOC:
+	    erts_finalize_atom_cache_map(ctx->acmp, ctx->flags);
+
+	    ctx->dhdr_ext_size = erts_encode_ext_dist_header_size(ctx->acmp);
+	    ctx->data_size += ctx->dhdr_ext_size;
+
+	    ctx->obuf = alloc_dist_obuf(ctx->data_size);
+	    ctx->obuf->ext_endp = &ctx->obuf->data[0] + ctx->pass_through_size + ctx->dhdr_ext_size;
+
+	    /* Encode internal version of dist header */
+	    ctx->obuf->extp = erts_encode_ext_dist_header_setup(ctx->obuf->ext_endp, ctx->acmp);
+	    /* Encode control message */
+	    erts_encode_dist_ext(ctx->ctl, &ctx->obuf->ext_endp, ctx->flags, ctx->acmp, NULL, NULL);
+	    if (is_value(ctx->msg)) {
+		ctx->u.ec.flags = ctx->flags;
+		ctx->u.ec.level = 0;
+		ctx->u.ec.wstack.wstart = NULL;
+		ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_ENCODE;
+	    } else {
+		ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
+	    }
+	    break;
 
-    /* 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);
-    }
+	case ERTS_DSIG_SEND_PHASE_MSG_ENCODE:
+	    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;
+		goto done;
+	    }
 
-    ASSERT(obuf->extp < obuf->ext_endp);
-    ASSERT(&obuf->data[0] <= obuf->extp - pass_through_size);
-    ASSERT(obuf->ext_endp <= &obuf->data[0] + data_size);
+	    ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
+	case ERTS_DSIG_SEND_PHASE_FIN: {
+	    DistEntry *dep = dsdp->dep;
+	    int suspended = 0;
+	    int resume = 0;
 
-    data_size = obuf->ext_endp - obuf->extp;
+	    ASSERT(ctx->obuf->extp < ctx->obuf->ext_endp);
+	    ASSERT(&ctx->obuf->data[0] <= ctx->obuf->extp - ctx->pass_through_size);
+	    ASSERT(ctx->obuf->ext_endp <= &ctx->obuf->data[0] + ctx->data_size);
 
-    /*
-     * 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_mtx_lock(&dep->qlock);
-	dep->qsize += size_obuf(obuf);
-	if (dep->qsize >= erts_dist_buf_busy_limit)
-	    dep->qflgs |= ERTS_DE_QFLG_BUSY;
-	if (!force_busy && (dep->qflgs & ERTS_DE_QFLG_BUSY)) {
-	    erts_smp_mtx_unlock(&dep->qlock);
+	    ctx->data_size = ctx->obuf->ext_endp - ctx->obuf->extp;
 
-	    plp = erts_proclist_create(c_p);
-	    erts_suspend(c_p, ERTS_PROC_LOCK_MAIN, NULL);
-	    suspended = 1;
-	    erts_smp_mtx_lock(&dep->qlock);
-	}
+	    /*
+	     * Signal encoded; now verify that the connection still exists,
+	     * and if so enqueue the signal and schedule it for send.
+	     */
+	    ctx->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(ctx->obuf);
+	    }
+	    else {
+		ErtsProcList *plp = NULL;
+		erts_smp_mtx_lock(&dep->qlock);
+		dep->qsize += size_obuf(ctx->obuf);
+		if (dep->qsize >= erts_dist_buf_busy_limit)
+		    dep->qflgs |= ERTS_DE_QFLG_BUSY;
+		if (!ctx->force_busy && (dep->qflgs & ERTS_DE_QFLG_BUSY)) {
+		    erts_smp_mtx_unlock(&dep->qlock);
+
+		    plp = erts_proclist_create(ctx->c_p);
+		    erts_suspend(ctx->c_p, ERTS_PROC_LOCK_MAIN, NULL);
+		    suspended = 1;
+		    erts_smp_mtx_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;
+		/* Enqueue obuf on dist entry */
+		if (dep->out_queue.last)
+		    dep->out_queue.last->next = ctx->obuf;
+		else
+		    dep->out_queue.first = ctx->obuf;
+		dep->out_queue.last = ctx->obuf;
+
+		if (!ctx->force_busy) {
+		    if (!(dep->qflgs & ERTS_DE_QFLG_BUSY)) {
+			if (suspended)
+			    resume = 1; /* was busy when we started, but isn't now */
+    #ifdef USE_VM_PROBES
+			if (resume && DTRACE_ENABLED(dist_port_not_busy)) {
+			    DTRACE_CHARBUF(port_str, 64);
+			    DTRACE_CHARBUF(remote_str, 64);
+
+			    erts_snprintf(port_str, sizeof(DTRACE_CHARBUF_NAME(port_str)),
+					  "%T", cid);
+			    erts_snprintf(remote_str, sizeof(DTRACE_CHARBUF_NAME(remote_str)),
+					  "%T", dep->sysname);
+			    DTRACE3(dist_port_not_busy, erts_this_node_sysname,
+				    port_str, remote_str);
+			}
+    #endif
+		    }
+		    else {
+			/* Enqueue suspended process on dist entry */
+			ASSERT(plp);
+			erts_proclist_store_last(&dep->suspended, plp);
+		    }
+		}
 
-	if (!force_busy) {
-	    if (!(dep->qflgs & ERTS_DE_QFLG_BUSY)) {
-		if (suspended)
-		    resume = 1; /* was busy when we started, but isn't now */
-#ifdef USE_VM_PROBES
-                if (resume && DTRACE_ENABLED(dist_port_not_busy)) {
-                    DTRACE_CHARBUF(port_str, 64);
-                    DTRACE_CHARBUF(remote_str, 64);
-
-                    erts_snprintf(port_str, sizeof(DTRACE_CHARBUF_NAME(port_str)),
-                                  "%T", cid);
-                    erts_snprintf(remote_str, sizeof(DTRACE_CHARBUF_NAME(remote_str)),
-                                  "%T", dep->sysname);
-                    DTRACE3(dist_port_not_busy, erts_this_node_sysname,
-                            port_str, remote_str);
-                }
-#endif
+		erts_smp_mtx_unlock(&dep->qlock);
+		erts_schedule_dist_command(NULL, dep);
+		erts_smp_de_runlock(dep);
+
+		if (resume) {
+		    erts_resume(ctx->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.
+		     */
+		}
 	    }
-	    else {
-		/* Enqueue suspended process on dist entry */
-		ASSERT(plp);
-		erts_proclist_store_last(&dep->suspended, plp);
+	    ctx->obuf = NULL;
+
+	    if (suspended) {
+    #ifdef USE_VM_PROBES
+		if (!resume && DTRACE_ENABLED(dist_port_busy)) {
+		    DTRACE_CHARBUF(port_str, 64);
+		    DTRACE_CHARBUF(remote_str, 64);
+		    DTRACE_CHARBUF(pid_str, 16);
+
+		    erts_snprintf(port_str, sizeof(DTRACE_CHARBUF_NAME(port_str)), "%T", cid);
+		    erts_snprintf(remote_str, sizeof(DTRACE_CHARBUF_NAME(remote_str)),
+				  "%T", dep->sysname);
+		    erts_snprintf(pid_str, sizeof(DTRACE_CHARBUF_NAME(pid_str)),
+				  "%T", ctx->c_p->common.id);
+		    DTRACE4(dist_port_busy, erts_this_node_sysname,
+			    port_str, remote_str, pid_str);
+		}
+    #endif
+		if (!resume && erts_system_monitor_flags.busy_dist_port)
+		    monitor_generic(ctx->c_p, am_busy_dist_port, cid);
+		retval = ERTS_DSIG_SEND_YIELD;
+	    } else {
+		retval = ERTS_DSIG_SEND_OK;
 	    }
+	    goto done;
 	}
-
-	erts_smp_mtx_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.
-	     */
+	default:
+	    erl_exit(ERTS_ABORT_EXIT, "dsig_send invalid phase (%d)\n", (int)ctx->phase);
 	}
     }
 
-    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) && !HALFWORD_HEAP
-	data_size &= 0x003fffffffffffff;
-#elif defined(ARCH_32) || HALFWORD_HEAP
-	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) {
-#ifdef USE_VM_PROBES
-        if (!resume && DTRACE_ENABLED(dist_port_busy)) {
-            DTRACE_CHARBUF(port_str, 64);
-            DTRACE_CHARBUF(remote_str, 64);
-            DTRACE_CHARBUF(pid_str, 16);
-
-            erts_snprintf(port_str, sizeof(DTRACE_CHARBUF_NAME(port_str)), "%T", cid);
-            erts_snprintf(remote_str, sizeof(DTRACE_CHARBUF_NAME(remote_str)),
-                          "%T", dep->sysname);
-            erts_snprintf(pid_str, sizeof(DTRACE_CHARBUF_NAME(pid_str)),
-                          "%T", c_p->common.id);
-            DTRACE4(dist_port_busy, erts_this_node_sysname,
-                    port_str, remote_str, pid_str);
-        }
-#endif
-	if (!resume && erts_system_monitor_flags.busy_dist_port)
-	    monitor_generic(c_p, am_busy_dist_port, cid);
-	return ERTS_DSIG_SEND_YIELD;
+done:
+    if (ctx->msg && ctx->c_p) {
+	BUMP_REDS(ctx->c_p, (initial_reds - ctx->reds) / TERM_TO_BINARY_LOOP_FACTOR);
     }
-    return ERTS_DSIG_SEND_OK;
+    return retval;
 }
 
-
 static Uint
 dist_port_command(Port *prt, ErtsDistOutputBuf *obuf)
 {
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index f32b999198..2a2ba0c83f 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -22,6 +22,7 @@
 
 #include "erl_process.h"
 #include "erl_node_tables.h"
+#include "zlib.h"
 
 #define DFLAG_PUBLISHED           0x01
 #define DFLAG_ATOM_CACHE          0x02
@@ -264,17 +265,105 @@ erts_destroy_dist_link(ErtsDistLinkData *dldp)
 
 #endif
 
+
+
+/* Define for testing */
+/* #define EXTREME_TTB_TRAPPING 1 */
+
+#ifndef EXTREME_TTB_TRAPPING
+#define TERM_TO_BINARY_LOOP_FACTOR 32
+#else
+#define TERM_TO_BINARY_LOOP_FACTOR 1
+#endif
+
+typedef enum { TTBSize, TTBEncode, TTBCompress } TTBState;
+typedef struct TTBSizeContext_ {
+    Uint flags;
+    int level;
+    Uint result;
+    Eterm obj;
+    ErtsEStack estack;
+} TTBSizeContext;
+
+typedef struct TTBEncodeContext_ {
+    Uint flags;
+    int level;
+    byte* ep;
+    Eterm obj;
+    ErtsWStack wstack;
+    Binary *result_bin;
+} TTBEncodeContext;
+
+typedef struct {
+    Uint real_size;
+    Uint dest_len;
+    byte *dbytes;
+    Binary *result_bin;
+    Binary *destination_bin;
+    z_stream stream;
+} TTBCompressContext;
+
+typedef struct {
+    int alive;
+    TTBState state;
+    union {
+	TTBSizeContext sc;
+	TTBEncodeContext ec;
+	TTBCompressContext cc;
+    } s;
+} TTBContext;
+
+enum erts_dsig_send_phase {
+    ERTS_DSIG_SEND_PHASE_INIT,
+    ERTS_DSIG_SEND_PHASE_MSG_SIZE,
+    ERTS_DSIG_SEND_PHASE_ALLOC,
+    ERTS_DSIG_SEND_PHASE_MSG_ENCODE,
+    ERTS_DSIG_SEND_PHASE_FIN
+};
+
+struct erts_dsig_send_context {
+    enum erts_dsig_send_phase phase;
+    Sint reds;
+
+    Eterm ctl;
+    Eterm msg;
+    int force_busy;
+    Uint32 pass_through_size;
+    Uint data_size, dhdr_ext_size;
+    ErtsAtomCacheMap *acmp;
+    ErtsDistOutputBuf *obuf;
+    Uint32 flags;
+    Process *c_p;
+    union {
+	TTBSizeContext sc;
+	TTBEncodeContext ec;
+    }u;
+};
+
+typedef struct {
+    int suspend;
+
+    Eterm ctl_heap[6];
+    ErtsDSigData dsd;
+    DistEntry* dep_to_deref;
+    struct erts_dsig_send_context dss;
+
+    Eterm return_term;
+}ErtsSendContext;
+
+
 /*
  * erts_dsig_send_* return values.
  */
 #define ERTS_DSIG_SEND_OK	0
 #define ERTS_DSIG_SEND_YIELD	1
+#define ERTS_DSIG_SEND_CONTINUE 2
 
 extern int erts_dsig_send_link(ErtsDSigData *, Eterm, Eterm);
-extern int erts_dsig_send_msg(ErtsDSigData *, Eterm, Eterm);
+extern int erts_dsig_send_msg(Eterm, Eterm, ErtsSendContext*);
 extern int erts_dsig_send_exit_tt(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_unlink(ErtsDSigData *, Eterm, Eterm);
-extern int erts_dsig_send_reg_msg(ErtsDSigData *, Eterm, Eterm);
+extern int erts_dsig_send_reg_msg(Eterm, Eterm, ErtsSendContext*);
 extern int erts_dsig_send_group_leader(ErtsDSigData *, Eterm, Eterm);
 extern int erts_dsig_send_exit(ErtsDSigData *, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_exit2(ErtsDSigData *, Eterm, Eterm, Eterm);
@@ -282,6 +371,10 @@ extern int erts_dsig_send_demonitor(ErtsDSigData *, Eterm, Eterm, Eterm, int);
 extern int erts_dsig_send_monitor(ErtsDSigData *, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_m_exit(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);
 
+extern int erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx);
+extern void erts_dsend_context_dtor(Binary*);
+extern Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* ctx);
+
 extern int erts_dist_command(Port *prt, int reds);
 extern void erts_dist_port_not_busy(Port *prt);
 extern void erts_kill_dist_connection(DistEntry *dep, Uint32);
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index 8d240355b0..564657a13b 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -498,15 +498,37 @@ byte *erts_encode_ext_dist_header_finalize(byte *ext, ErtsAtomCache *cache, Uint
     return ep;
 }
 
-Uint erts_encode_dist_ext_size(Eterm term, Uint32 flags, ErtsAtomCacheMap *acmp)
+int erts_encode_dist_ext_size(Eterm term, Uint32 flags, ErtsAtomCacheMap *acmp,
+			      Uint* szp)
 {
-    Uint sz = 0;
+    Uint sz;
+    if (encode_size_struct_int(NULL, acmp, term, flags, NULL, &sz)) {
+	return -1;
+    } else {
 #ifndef ERTS_DEBUG_USE_DIST_SEP
-    if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
+	if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
 #endif
-	sz++ /* VERSION_MAGIC */;
-    sz += encode_size_struct2(acmp, term, flags);
-    return sz;
+	    sz++ /* VERSION_MAGIC */;
+
+	*szp += sz;
+	return 0;
+    }
+}
+
+int erts_encode_dist_ext_size_int(Eterm term, struct erts_dsig_send_context* ctx, Uint* szp)
+{
+    Uint sz;
+    if (encode_size_struct_int(&ctx->u.sc, ctx->acmp, term, ctx->flags, &ctx->reds, &sz)) {
+	return -1;
+    } else {
+#ifndef ERTS_DEBUG_USE_DIST_SEP
+	if (!(ctx->flags & DFLAG_DIST_HDR_ATOM_CACHE))
+#endif
+	    sz++ /* VERSION_MAGIC */;
+
+	*szp += sz;
+	return 0;
+    }
 }
 
 Uint erts_encode_ext_size(Eterm term)
@@ -527,19 +549,16 @@ Uint erts_encode_ext_size_ets(Eterm term)
 }
 
 
-void erts_encode_dist_ext(Eterm term, byte **ext, Uint32 flags, ErtsAtomCacheMap *acmp)
+int erts_encode_dist_ext(Eterm term, byte **ext, Uint32 flags, ErtsAtomCacheMap *acmp,
+			  TTBEncodeContext* ctx, Sint* reds)
 {
-    byte *ep = *ext;
-#ifndef ERTS_DEBUG_USE_DIST_SEP
-    if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
-#endif
-	*ep++ = VERSION_MAGIC;
-    ep = enc_term(acmp, term, ep, flags, NULL);
-    if (!ep)
-	erl_exit(ERTS_ABORT_EXIT,
-		 "%s:%d:erts_encode_dist_ext(): Internal data structure error\n",
-		 __FILE__, __LINE__);
-    *ext = ep;
+    if (!ctx || !ctx->wstack.wstart) {
+    #ifndef ERTS_DEBUG_USE_DIST_SEP
+	if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
+    #endif
+	    *(*ext)++ = VERSION_MAGIC;
+    }
+    return enc_term_int(ctx, acmp, term, *ext, flags, NULL, reds, ext);
 }
 
 void erts_encode_ext(Eterm term, byte **ext)
@@ -1740,55 +1759,15 @@ erts_term_to_binary(Process* p, Eterm Term, int level, Uint flags) {
     return erts_term_to_binary_simple(p, Term, size, level, flags);
 }
 
-/* Define for testing */
-/* #define EXTREME_TTB_TRAPPING 1 */
+/* Define EXTREME_TTB_TRAPPING for testing in dist.h */
 
 #ifndef EXTREME_TTB_TRAPPING
-#define TERM_TO_BINARY_LOOP_FACTOR 32
 #define TERM_TO_BINARY_COMPRESS_CHUNK (1 << 18)
 #else
-#define TERM_TO_BINARY_LOOP_FACTOR 1
 #define TERM_TO_BINARY_COMPRESS_CHUNK 10
 #endif
 
 
-typedef enum { TTBSize, TTBEncode, TTBCompress } TTBState;
-typedef struct TTBSizeContext_ {
-    Uint flags;
-    int level;
-    Uint result;
-    Eterm obj;
-    ErtsEStack estack;
-} TTBSizeContext;
-
-typedef struct TTBEncodeContext_ {
-    Uint flags;
-    int level;
-    byte* ep;
-    Eterm obj;
-    ErtsWStack wstack;
-    Binary *result_bin;
-} TTBEncodeContext;
-
-typedef struct {
-    Uint real_size;
-    Uint dest_len;
-    byte *dbytes;
-    Binary *result_bin;
-    Binary *destination_bin;
-    z_stream stream;
-} TTBCompressContext;
-
-typedef struct {
-    int alive;
-    TTBState state;
-    union {
-	TTBSizeContext sc;
-	TTBEncodeContext ec;
-	TTBCompressContext cc;
-    } s;
-} TTBContext;
-
 static void ttb_context_destructor(Binary *context_bin)
 {
     TTBContext *context = ERTS_MAGIC_BIN_DATA(context_bin);
diff --git a/erts/emulator/beam/external.h b/erts/emulator/beam/external.h
index bf00958eb1..f120e96e3b 100644
--- a/erts/emulator/beam/external.h
+++ b/erts/emulator/beam/external.h
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 1996-2013. All Rights Reserved.
+ * Copyright Ericsson AB 1996-2014. 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
@@ -150,6 +150,7 @@ typedef struct {
     Uint extsize;
 } ErtsBinary2TermState;
 
+
 /* -------------------------------------------------------------------------- */
 
 void erts_init_atom_cache_map(ErtsAtomCacheMap *);
@@ -161,8 +162,12 @@ Uint erts_encode_ext_dist_header_size(ErtsAtomCacheMap *);
 Uint erts_encode_ext_dist_header_size(ErtsAtomCacheMap *);
 byte *erts_encode_ext_dist_header_setup(byte *, ErtsAtomCacheMap *);
 byte *erts_encode_ext_dist_header_finalize(byte *, ErtsAtomCache *, Uint32);
-Uint erts_encode_dist_ext_size(Eterm, Uint32, ErtsAtomCacheMap *);
-void erts_encode_dist_ext(Eterm, byte **, Uint32, ErtsAtomCacheMap *);
+struct erts_dsig_send_context;
+int erts_encode_dist_ext_size(Eterm, Uint32, ErtsAtomCacheMap*, Uint* szp);
+int erts_encode_dist_ext_size_int(Eterm term, struct erts_dsig_send_context* ctx, Uint* szp);
+struct TTBEncodeContext_;
+int erts_encode_dist_ext(Eterm, byte **, Uint32, ErtsAtomCacheMap *,
+			  struct TTBEncodeContext_ *, Sint* reds);
 
 Uint erts_encode_ext_size(Eterm);
 Uint erts_encode_ext_size_2(Eterm, unsigned);