Merge branch 'sverk/yielding-distr-send/OTP-12232'

* sverk/yielding-distr-send/OTP-12232:
  erts: Add constant TERM_TO_BINARY_MEMCPY_FACTOR
  erts: Optimize some repeated calls to {E,W}STACK_PUSH
  erts: Yield in term_to_binary when encoding big maps
  erts: Remove unnecessary goto for fun encoding
  erts: Yield in term_to_binary while copying large binaries
  erts: Implement yielding for distributed send of large messages

10 files changed, 705 insertions, 381 deletions

diff --git a/erts/emulator/beam/atom.names b/erts/emulator/beam/atom.names
index 721a1ff219..c097866c7e 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 12a1ecd50e..059f9f6c2c 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -48,7 +48,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;
 
@@ -1779,6 +1779,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)
 {
@@ -1797,34 +1799,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
@@ -1832,6 +1836,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;
@@ -1852,7 +1858,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;
@@ -1885,7 +1892,7 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
 #endif
 	    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);
 
@@ -1942,7 +1949,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)) {
@@ -1951,12 +1958,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;
@@ -2036,9 +2043,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 ??? */
@@ -2069,9 +2081,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;
@@ -2079,34 +2093,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) {
@@ -2114,68 +2138,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) {
@@ -2183,35 +2266,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;
 }
 
 /**********************************************************************/
@@ -4819,6 +4913,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 dcbbb857da..bfecac1612 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);
 
@@ -707,6 +707,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
@@ -729,7 +778,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;
 }
@@ -742,7 +791,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;
 }
@@ -770,7 +819,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;
 }
@@ -791,7 +840,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;
 }
@@ -813,18 +862,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;
@@ -836,8 +884,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
@@ -850,7 +897,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)),
@@ -865,26 +912,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;
@@ -896,7 +945,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 
@@ -910,7 +958,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)),
@@ -925,17 +973,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;
 }
 
@@ -992,7 +1042,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;
 }
@@ -1008,7 +1058,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;
 }
@@ -1024,7 +1074,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;
 }
@@ -1041,7 +1091,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;
 }
@@ -1691,194 +1741,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/erl_printf_term.c b/erts/emulator/beam/erl_printf_term.c
index 1a0c7a9fc9..74e38c13df 100644
--- a/erts/emulator/beam/erl_printf_term.c
+++ b/erts/emulator/beam/erl_printf_term.c
@@ -303,13 +303,9 @@ print_term(fmtfn_t fn, void* arg, Eterm obj, long *dcount,
 		    tl = CDR(cons);
 		    if (is_not_nil(tl)) {
 			if (is_list(tl)) {
-			    WSTACK_PUSH(s, tl);
-			    WSTACK_PUSH(s, PRT_ONE_CONS);
-			    WSTACK_PUSH(s, PRT_COMMA);
+			    WSTACK_PUSH3(s, tl, PRT_ONE_CONS, PRT_COMMA);
 			} else {
-			    WSTACK_PUSH(s, tl);
-			    WSTACK_PUSH(s, PRT_TERM);
-			    WSTACK_PUSH(s, PRT_BAR);
+			    WSTACK_PUSH3(s, tl, PRT_TERM, PRT_BAR);
 			}
 		    }
 		}
@@ -319,9 +315,7 @@ print_term(fmtfn_t fn, void* arg, Eterm obj, long *dcount,
 		break;
 	    default:		/* PRT_LAST_ARRAY_ELEMENT+1 and upwards */
 		obj = *popped.ptr;
-	        WSTACK_PUSH(s, (UWord) (popped.ptr + 1));
-		WSTACK_PUSH(s, val-1);
-		WSTACK_PUSH(s, PRT_COMMA);
+	        WSTACK_PUSH3(s, (UWord) (popped.ptr + 1), val-1, PRT_COMMA);
 		break;
 	    }
 	    break;
@@ -451,8 +445,7 @@ print_term(fmtfn_t fn, void* arg, Eterm obj, long *dcount,
 	    WSTACK_PUSH(s,PRT_CLOSE_TUPLE);
 	    ++nobj;
 	    if (i > 0) {
-		WSTACK_PUSH(s, (UWord) nobj);
-		WSTACK_PUSH(s, PRT_LAST_ARRAY_ELEMENT+i-1);
+		WSTACK_PUSH2(s, (UWord) nobj, PRT_LAST_ARRAY_ELEMENT+i-1);
 	    }
 	    break;
 	case FLOAT_DEF: {
@@ -574,19 +567,10 @@ print_term(fmtfn_t fn, void* arg, Eterm obj, long *dcount,
 		WSTACK_PUSH(s, PRT_CLOSE_TUPLE);
 		if (n > 0) {
 		    n--;
-		    WSTACK_PUSH(s, vs[n]);
-		    WSTACK_PUSH(s, PRT_TERM);
-		    WSTACK_PUSH(s, PRT_ASSOC);
-		    WSTACK_PUSH(s, ks[n]);
-		    WSTACK_PUSH(s, PRT_TERM);
-
+		    WSTACK_PUSH5(s, vs[n], PRT_TERM, PRT_ASSOC, ks[n], PRT_TERM);
 		    while (n--) {
-			WSTACK_PUSH(s, PRT_COMMA);
-			WSTACK_PUSH(s, vs[n]);
-			WSTACK_PUSH(s, PRT_TERM);
-			WSTACK_PUSH(s, PRT_ASSOC);
-			WSTACK_PUSH(s, ks[n]);
-			WSTACK_PUSH(s, PRT_TERM);
+			WSTACK_PUSH6(s, PRT_COMMA, vs[n], PRT_TERM, PRT_ASSOC,
+				     ks[n], PRT_TERM);
 		    }
 		}
 	    }
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index 48e717bb6f..601cbe9d7d 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -500,15 +500,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)
@@ -529,19 +551,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)
@@ -1742,54 +1761,14 @@ 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;
+#define TERM_TO_BINARY_MEMCPY_FACTOR 8
 
 static void ttb_context_destructor(Binary *context_bin)
 {
@@ -2323,8 +2302,9 @@ dec_pid(ErtsDistExternal *edep, Eterm** hpp, byte* ep, ErlOffHeap* off_heap, Ete
 #define ENC_TERM ((Eterm) 0)
 #define ENC_ONE_CONS ((Eterm) 1)
 #define ENC_PATCH_FUN_SIZE ((Eterm) 2)
-#define ENC_LAST_ARRAY_ELEMENT ((Eterm) 3)
-
+#define ENC_BIN_COPY ((Eterm) 3)
+#define ENC_MAP_PAIR ((Eterm) 4)
+#define ENC_LAST_ARRAY_ELEMENT ((Eterm) 5)
 
 static byte*
 enc_term(ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, Uint32 dflags,
@@ -2360,6 +2340,9 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 	    WSTACK_RESTORE(s, &ctx->wstack);
 	    ep = ctx->ep;
 	    obj = ctx->obj;
+	    if (is_non_value(obj)) {
+		goto outer_loop;
+	    }
 	}
     }
 
@@ -2383,8 +2366,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 
 		obj = CAR(cons);
 		tl = CDR(cons);
-		WSTACK_PUSH(s, is_list(tl) ? ENC_ONE_CONS : ENC_TERM);
-		WSTACK_PUSH(s, tl);
+		WSTACK_PUSH2(s, (is_list(tl) ? ENC_ONE_CONS : ENC_TERM),
+			     tl);
 	    }
 	    break;
 	case ENC_PATCH_FUN_SIZE:
@@ -2397,6 +2380,39 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		put_int32(ep - size_p, size_p);
 	    }
 	    goto outer_loop;
+	case ENC_BIN_COPY: {
+	    Uint bits = (Uint)obj;
+	    Uint bitoffs = WSTACK_POP(s);
+	    byte* bytes = (byte*) WSTACK_POP(s);
+	    byte* dst = (byte*) WSTACK_POP(s);
+	    if (bits > r * (TERM_TO_BINARY_MEMCPY_FACTOR * 8)) {
+		Uint n = r * TERM_TO_BINARY_MEMCPY_FACTOR;
+		WSTACK_PUSH5(s, (UWord)(dst + n), (UWord)(bytes + n), bitoffs,
+			     ENC_BIN_COPY, bits - 8*n);
+		bits = 8*n;
+		copy_binary_to_buffer(dst, 0, bytes, bitoffs, bits);
+		obj = THE_NON_VALUE;
+		r = 0; /* yield */
+		break;
+	    } else {
+		copy_binary_to_buffer(dst, 0, bytes, bitoffs, bits);
+		r -= bits / (TERM_TO_BINARY_MEMCPY_FACTOR * 8);
+		goto outer_loop;
+	    }
+	}
+	case ENC_MAP_PAIR: {
+	    Uint pairs_left = obj;
+	    Eterm *vptr = (Eterm*) WSTACK_POP(s);
+	    Eterm *kptr = (Eterm*) WSTACK_POP(s);
+
+	    obj = *kptr;
+	    if (--pairs_left > 0) {
+		WSTACK_PUSH4(s, (UWord)(kptr+1), (UWord)(vptr+1),
+			     ENC_MAP_PAIR, pairs_left);
+	    }
+	    WSTACK_PUSH2(s, ENC_TERM, *vptr);
+	    break;
+	}
 	case ENC_LAST_ARRAY_ELEMENT:
 	    /* obj is the tuple */
 	    {
@@ -2415,17 +2431,16 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 #else
 		Eterm* ptr = (Eterm *) obj;
 #endif
-		WSTACK_PUSH(s, val-1);
 		obj = *ptr++;
-		WSTACK_PUSH(s, (UWord)ptr);
+		WSTACK_PUSH2(s, val-1, (UWord)ptr);
 	    }
 	    break;
 	}
 
     L_jump_start:
 
-	if (ctx && --r == 0) {
-	    *reds = r;
+	if (ctx && --r <= 0) {
+	    *reds = 0;
 	    ctx->obj = obj;
 	    ctx->ep = ep;
 	    WSTACK_SAVE(s, &ctx->wstack);
@@ -2574,8 +2589,7 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		ep += 4;
 	    }
 	    if (i > 0) {
-		WSTACK_PUSH(s, ENC_LAST_ARRAY_ELEMENT+i-1);
-		WSTACK_PUSH(s, (UWord)ptr);
+		WSTACK_PUSH2(s, ENC_LAST_ARRAY_ELEMENT+i-1, (UWord)ptr);
 	    }
 	    break;
 
@@ -2591,18 +2605,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    Eterm *kptr = map_get_keys(mp);
 		    Eterm *vptr = map_get_values(mp);
 
-		    for (i = size-1; i >= 1; i--) {
-			WSTACK_PUSH(s, ENC_TERM);
-			WSTACK_PUSH(s, (UWord) vptr[i]);
-			WSTACK_PUSH(s, ENC_TERM);
-			WSTACK_PUSH(s, (UWord) kptr[i]);
-		    }
-
-		    WSTACK_PUSH(s, ENC_TERM);
-		    WSTACK_PUSH(s, (UWord) vptr[0]);
-
-		    obj = kptr[0];
-		    goto L_jump_start;
+		    WSTACK_PUSH4(s, (UWord)kptr, (UWord)vptr,
+				 ENC_MAP_PAIR, size);
 		}
 	    }
 	    break;
@@ -2640,6 +2644,7 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		Uint bitoffs;
 		Uint bitsize;
 		byte* bytes;
+		byte* data_dst;
 
 		ERTS_GET_BINARY_BYTES(obj, bytes, bitoffs, bitsize);
 		if (dflags & DFLAG_INTERNAL_TAGS) {
@@ -2685,7 +2690,7 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    j = binary_size(obj);
 		    put_int32(j, ep);
 		    ep += 4;
-		    copy_binary_to_buffer(ep, 0, bytes, bitoffs, 8*j);
+		    data_dst = ep;
 		    ep += j;
 		} else if (dflags & DFLAG_BIT_BINARIES) {
 		    /* Bit-level binary. */
@@ -2695,7 +2700,7 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    ep += 4;
 		    *ep++ = bitsize;
 		    ep[j] = 0;	/* Zero unused bits at end of binary */
-		    copy_binary_to_buffer(ep, 0, bytes, bitoffs, 8*j+bitsize);
+		    data_dst = ep;
 		    ep += j + 1;
 		} else {
 		    /*
@@ -2709,11 +2714,18 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    put_int32((j+1), ep);
 		    ep += 4;
 		    ep[j] = 0;	/* Zero unused bits at end of binary */
-		    copy_binary_to_buffer(ep, 0, bytes, bitoffs, 8*j+bitsize);
+		    data_dst = ep;
 		    ep += j+1;
 		    *ep++ = SMALL_INTEGER_EXT;
 		    *ep++ = bitsize;
 		}
+		if (ctx && j > r * TERM_TO_BINARY_MEMCPY_FACTOR) {
+		    WSTACK_PUSH5(s, (UWord)data_dst, (UWord)bytes, bitoffs,
+				 ENC_BIN_COPY, 8*j + bitsize);
+		} else {
+		    copy_binary_to_buffer(data_dst, 0, bytes, bitoffs,
+					  8 * j + bitsize);
+		}
 	    }
 	    break;
 	case EXPORT_DEF:
@@ -2742,13 +2754,12 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 	case FUN_DEF:
 	    {
 		ErlFunThing* funp = (ErlFunThing *) fun_val(obj);
+		int ei;
 
 		if ((dflags & DFLAG_NEW_FUN_TAGS) != 0) {
-		    int ei;
-
 		    *ep++ = NEW_FUN_EXT;
-		    WSTACK_PUSH(s, ENC_PATCH_FUN_SIZE);
-		    WSTACK_PUSH(s, (UWord) ep); /* Position for patching in size */
+		    WSTACK_PUSH2(s, ENC_PATCH_FUN_SIZE,
+				 (UWord) ep); /* Position for patching in size */
 		    ep += 4;
 		    *ep = funp->arity;
 		    ep += 1;
@@ -2762,16 +2773,6 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    ep = enc_term(acmp, make_small(funp->fe->old_index), ep, dflags, off_heap);
 		    ep = enc_term(acmp, make_small(funp->fe->old_uniq), ep, dflags, off_heap);
 		    ep = enc_pid(acmp, funp->creator, ep, dflags);
-
-		fun_env:
-		    for (ei = funp->num_free-1; ei > 0; ei--) {
-			WSTACK_PUSH(s, ENC_TERM);
-			WSTACK_PUSH(s, (UWord) funp->env[ei]);
-		    }
-		    if (funp->num_free != 0) {
-			obj = funp->env[0];
-			goto L_jump_start;
-		    }
 		} else {
 		    /*
 		     * Communicating with an obsolete erl_interface or
@@ -2803,7 +2804,13 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    *ep++ = SMALL_TUPLE_EXT;
 		    put_int8(funp->num_free, ep);
 		    ep += 1;
-		    goto fun_env;
+		}
+		for (ei = funp->num_free-1; ei > 0; ei--) {
+		    WSTACK_PUSH2(s, ENC_TERM, (UWord) funp->env[ei]);
+		}
+		if (funp->num_free != 0) {
+		    obj = funp->env[0];
+		    goto L_jump_start;
 		}
 	    }
 	    break;
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);
diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index da9f029a9f..ec8c1e3ccb 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -480,6 +480,17 @@ do {						\
     *s.sp++ = (z);				\
 } while(0)
 
+#define ESTACK_PUSH4(s, E1, E2, E3, E4)		\
+do {						\
+    if (s.sp > s.end - 4) {			\
+	erl_grow_estack(&s, ESTK_DEF_STACK(s)); \
+    }						\
+    *s.sp++ = (E1);				\
+    *s.sp++ = (E2);				\
+    *s.sp++ = (E3);				\
+    *s.sp++ = (E4);				\
+} while(0)
+
 #define ESTACK_COUNT(s) (s.sp - s.start)
 #define ESTACK_ISEMPTY(s) (s.sp == s.start)
 #define ESTACK_POP(s) (*(--s.sp))
@@ -598,6 +609,42 @@ do {						\
     *s.wsp++ = (z);				\
 } while(0)
 
+#define WSTACK_PUSH4(s, A1, A2, A3, A4)		\
+do {						\
+    if (s.wsp > s.wend - 4) {	\
+	erl_grow_wstack(&s, WSTK_DEF_STACK(s)); \
+    }						\
+    *s.wsp++ = (A1);				\
+    *s.wsp++ = (A2);				\
+    *s.wsp++ = (A3);				\
+    *s.wsp++ = (A4);				\
+} while(0)
+
+#define WSTACK_PUSH5(s, A1, A2, A3, A4, A5)     \
+do {						\
+    if (s.wsp > s.wend - 5) {	                \
+	erl_grow_wstack(&s, WSTK_DEF_STACK(s)); \
+    }						\
+    *s.wsp++ = (A1);				\
+    *s.wsp++ = (A2);				\
+    *s.wsp++ = (A3);				\
+    *s.wsp++ = (A4);				\
+    *s.wsp++ = (A5);				\
+} while(0)
+
+#define WSTACK_PUSH6(s, A1, A2, A3, A4, A5, A6) \
+do {						\
+    if (s.wsp > s.wend - 6) {	                \
+	erl_grow_wstack(&s, WSTK_DEF_STACK(s)); \
+    }						\
+    *s.wsp++ = (A1);				\
+    *s.wsp++ = (A2);				\
+    *s.wsp++ = (A3);				\
+    *s.wsp++ = (A4);				\
+    *s.wsp++ = (A5);				\
+    *s.wsp++ = (A6);				\
+} while(0)
+
 #define WSTACK_COUNT(s) (s.wsp - s.wstart)
 #define WSTACK_ISEMPTY(s) (s.wsp == s.wstart)
 #define WSTACK_POP(s) (*(--s.wsp))
diff --git a/erts/emulator/beam/utils.c b/erts/emulator/beam/utils.c
index 9b3ee5cc65..e03cd22070 100644
--- a/erts/emulator/beam/utils.c
+++ b/erts/emulator/beam/utils.c
@@ -1232,25 +1232,20 @@ make_hash2(Eterm term)
 		if (size == 0) {
 		    goto hash2_common;
 		}
-		ESTACK_PUSH(s, hash_xor_values);
-		ESTACK_PUSH(s, hash_xor_keys);
-		ESTACK_PUSH(s, hash);
-		ESTACK_PUSH(s, HASH_MAP_TAIL);
+		ESTACK_PUSH4(s, hash_xor_values, hash_xor_keys, hash, HASH_MAP_TAIL);
 		hash = 0;
 		hash_xor_keys = 0;
 		hash_xor_values = 0;
 		for (i = size - 1; i >= 0; i--) {
 		    tmp = vs[i];
-		    ESTACK_PUSH(s, HASH_MAP_VAL);
-		    ESTACK_PUSH(s, tmp);
+		    ESTACK_PUSH2(s, HASH_MAP_VAL, tmp);
 		}
 		/* We do not want to expose the tuple representation.
 		 * Do not push the keys as a tuple.
 		 */
 		for (i = size - 1; i >= 0; i--) {
 		    tmp = ks[i];
-		    ESTACK_PUSH(s, HASH_MAP_KEY);
-		    ESTACK_PUSH(s, tmp);
+		    ESTACK_PUSH2(s, HASH_MAP_KEY, tmp);
 		}
 		goto hash2_common;
 	    }
diff --git a/erts/emulator/hipe/hipe_bif_list.m4 b/erts/emulator/hipe/hipe_bif_list.m4
index 5f92b6bac4..96a849621f 100644
--- a/erts/emulator/hipe/hipe_bif_list.m4
+++ b/erts/emulator/hipe/hipe_bif_list.m4
@@ -277,7 +277,10 @@ ifelse($1,list_to_binary_1,hipe_wrapper_list_to_binary_1,
 ifelse($1,iolist_to_binary_1,hipe_wrapper_iolist_to_binary_1,
 ifelse($1,binary_list_to_bin_1,hipe_wrapper_binary_list_to_bin_1,
 ifelse($1,list_to_bitstring_1,hipe_wrapper_list_to_bitstring_1,
-$1)))))))))))')
+ifelse($1,send_2,hipe_wrapper_send_2,
+ifelse($1,send_3,hipe_wrapper_send_3,
+ifelse($1,ebif_bang_2,hipe_wrapper_ebif_bang_2,
+$1))))))))))))))')
 
 define(BIF_LIST,`standard_bif_interface_$3(nbif_$4, CFUN($4))')
 include(TARGET/`erl_bif_list.h')