erts: Implement fragmentation of distrubution messages

1 files changed, 653 insertions, 276 deletions

diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 91a9481d84..78674848a1 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -55,15 +55,21 @@
  */
 #if 0
 #define ERTS_DIST_MSG_DBG
+FILE *dbg_file;
 #endif
 #if 0
+/* Enable this to print the dist debug messages to a file instead */
+#define ERTS_DIST_MSG_DBG_FILE "/tmp/dist_dbg.%d"
+#endif
+#if 0
+/* Enable this to print the raw bytes sent and received */
 #define ERTS_RAW_DIST_MSG_DBG
 #endif
 
 #if defined(ERTS_DIST_MSG_DBG) || defined(ERTS_RAW_DIST_MSG_DBG)
 static void bw(byte *buf, ErlDrvSizeT sz)
 {
-    bin_write(ERTS_PRINT_STDERR, NULL, buf, sz);
+    bin_write(ERTS_PRINT_FILE, dbg_file, buf, sz);
 }
 #endif
 
@@ -71,37 +77,93 @@ static void bw(byte *buf, ErlDrvSizeT sz)
 static void
 dist_msg_dbg(ErtsDistExternal *edep, char *what, byte *buf, int sz)
 {
-    ErtsHeapFactory factory;
-    byte *extp = edep->extp;
+    byte *extp = edep->data->extp;
     Eterm msg;
     Sint ctl_len;
-    Sint size = ctl_len = erts_decode_dist_ext_size(edep);
+    Sint size = ctl_len = erts_decode_dist_ext_size(edep, 0);
     if (size < 0) {
-	erts_fprintf(stderr,
+	erts_fprintf(dbg_file,
 		     "DIST MSG DEBUG: erts_decode_dist_ext_size(%s) failed:\n",
 		     what);
 	bw(buf, sz);
     }
     else {
-	ErlHeapFragment *mbuf = new_message_buffer(size);
-	erts_factory_static_init(&factory, mbuf->mem, ctl_len, &mbuf->off_heap);
-	msg = erts_decode_dist_ext(&factory, edep);
+        ErtsHeapFactory factory;
+	ErtsMessage *mbuf = erts_factory_message_create(&factory, NULL, 0, ctl_len);
+        /* Set mbuf msg to NIL as erts_factory_undo will fail otherwise */
+        ERL_MESSAGE_TERM(mbuf) = NIL;
+        msg = erts_decode_dist_ext(&factory, edep, 0);
 	if (is_value(msg))
-	    erts_fprintf(stderr, "    %s: %.80T\n", what, msg);
+	    erts_fprintf(dbg_file, "    %s: %.80T\n", what, msg);
 	else {
-	    erts_fprintf(stderr,
+	    erts_fprintf(dbg_file,
 			 "DIST MSG DEBUG: erts_decode_dist_ext(%s) failed:\n",
 			 what);
 	    bw(buf, sz);
 	}
-	free_message_buffer(mbuf);
-	edep->extp = extp;
+        erts_factory_undo(&factory);
+	edep->data->extp = extp;
     }
 }
 
+static char *erts_dop_to_string(enum dop dop) {
+    if (dop == DOP_LINK)
+        return "LINK";
+    if (dop == DOP_SEND)
+        return "SEND";
+    if (dop == DOP_EXIT)
+        return "EXIT";
+    if (dop == DOP_UNLINK)
+        return "UNLINK";
+    if (dop == DOP_REG_SEND)
+        return "REG_SEND";
+    if (dop == DOP_GROUP_LEADER)
+        return "GROUP_LEADER";
+    if (dop == DOP_EXIT2)
+        return "EXIT2";
+    if (dop == DOP_SEND_TT)
+        return "SEND_TT";
+    if (dop == DOP_EXIT_TT)
+        return "EXIT_TT";
+    if (dop == DOP_REG_SEND_TT)
+        return "REG_SEND_TT";
+    if (dop == DOP_EXIT2_TT)
+        return "EXIT2_TT";
+    if (dop == DOP_MONITOR_P)
+        return "MONITOR_P";
+    if (dop == DOP_DEMONITOR_P)
+        return "DEMONITOR_P";
+    if (dop == DOP_MONITOR_P_EXIT)
+        return "MONITOR_P_EXIT";
+    if (dop == DOP_SEND_SENDER)
+        return "SEND_SENDER";
+    if (dop == DOP_SEND_SENDER_TT)
+        return "SEND_SENDER_TT";
+    if (dop == DOP_PAYLOAD_EXIT)
+        return "PAYLOAD_EXIT";
+    if (dop == DOP_PAYLOAD_EXIT_TT)
+        return "PAYLOAD_EXIT_TT";
+    if (dop == DOP_PAYLOAD_EXIT2)
+        return "PAYLOAD_EXIT2";
+    if (dop == DOP_PAYLOAD_EXIT2_TT)
+        return "PAYLOAD_EXIT2_TT";
+    if (dop == DOP_PAYLOAD_MONITOR_P_EXIT)
+        return "PAYLOAD_MONITOR_P_EXIT";
+    ASSERT(0);
+    return "UNKNOWN";
+}
+
 #endif
 
+#if defined(VALGRIND)
+#include <valgrind/valgrind.h>
+#include <valgrind/memcheck.h>
 
+#  define PURIFY_MSG(msg)                                                    \
+    VALGRIND_PRINTF("%s, line %d: %s", __FILE__, __LINE__, msg)
+#else
+#  define PURIFY_MSG(msg)
+#endif
 
 int erts_is_alive; /* System must be blocked on change */
 int erts_dist_buf_busy_limit;
@@ -115,13 +177,16 @@ static Export *dist_ctrl_put_data_trap;
 
 /* forward declarations */
 
-static int dsig_send_exit(ErtsDSigData* dsdp, Eterm ctl, Eterm msg, int force_busy);
+static int dsig_send_exit(ErtsDSigData* dsdp, Eterm ctl, Eterm msg, Eterm from, int force_busy);
 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);
 static Sint abort_connection(DistEntry* dep, Uint32 conn_id);
 static ErtsDistOutputBuf* clear_de_out_queues(DistEntry*);
 static void free_de_out_queues(DistEntry*, ErtsDistOutputBuf*);
+int erts_dist_seq_tree_foreach_delete_yielding(DistSeqNode **root,
+                                               void **vyspp,
+                                               Sint limit);
 
 static erts_atomic_t no_caches;
 static erts_atomic_t no_nodes;
@@ -142,7 +207,6 @@ delete_cache(ErtsAtomCache *cache)
     }
 }
 
-
 static void
 create_cache(DistEntry *dep)
 {
@@ -207,12 +271,14 @@ typedef enum {
     ERTS_CML_CLEANUP_STATE_LINKS,
     ERTS_CML_CLEANUP_STATE_MONITORS,
     ERTS_CML_CLEANUP_STATE_ONAME_MONITORS,
+    ERTS_CML_CLEANUP_STATE_SEQUENCES,
     ERTS_CML_CLEANUP_STATE_NODE_MONITORS
-} ErtsConMonLnkCleaupState;
+} ErtsConMonLnkSeqCleanupState;
 
 typedef struct {
-    ErtsConMonLnkCleaupState state;
+    ErtsConMonLnkSeqCleanupState state;
     ErtsMonLnkDist *dist;
+    DistSeqNode *seq;
     void *yield_state;
     int trigger_node_monitors;
     Eterm nodename;
@@ -220,12 +286,12 @@ typedef struct {
     Eterm reason;
     ErlOffHeap oh;
     Eterm heap[1];
-} ErtsConMonLnkCleanup;
+} ErtsConMonLnkSeqCleanup;
 
 static void
-con_monitor_link_cleanup(void *vcmlcp)
+con_monitor_link_seq_cleanup(void *vcmlcp)
 {
-    ErtsConMonLnkCleanup *cmlcp = vcmlcp;
+    ErtsConMonLnkSeqCleanup *cmlcp = vcmlcp;
     ErtsMonLnkDist *dist = cmlcp->dist;
     ErtsSchedulerData *esdp;
     int reds = CONTEXT_REDS;
@@ -263,6 +329,15 @@ con_monitor_link_cleanup(void *vcmlcp)
         erts_mon_link_dist_dec_refc(dist);
 
         ASSERT(!cmlcp->yield_state);
+        cmlcp->state = ERTS_CML_CLEANUP_STATE_SEQUENCES;
+    case ERTS_CML_CLEANUP_STATE_SEQUENCES:
+        reds = erts_dist_seq_tree_foreach_delete_yielding(&cmlcp->seq,
+                                                          &cmlcp->yield_state,
+                                                          reds);
+        if (reds <= 0)
+            break;
+
+        ASSERT(!cmlcp->yield_state);
         cmlcp->state = ERTS_CML_CLEANUP_STATE_NODE_MONITORS;
     case ERTS_CML_CLEANUP_STATE_NODE_MONITORS:
         if (cmlcp->trigger_node_monitors) {
@@ -282,22 +357,23 @@ con_monitor_link_cleanup(void *vcmlcp)
     esdp = erts_get_scheduler_data();
     ASSERT(esdp && esdp->type == ERTS_SCHED_NORMAL);
     erts_schedule_misc_aux_work((int) esdp->no,
-                                con_monitor_link_cleanup,
+                                con_monitor_link_seq_cleanup,
                                 (void *) cmlcp);
 }
 
 static void
-schedule_con_monitor_link_cleanup(ErtsMonLnkDist *dist,
-                                  Eterm nodename,
-                                  Eterm visability,
-                                  Eterm reason)
+schedule_con_monitor_link_seq_cleanup(ErtsMonLnkDist *dist,
+                                      DistSeqNode *seq,
+                                      Eterm nodename,
+                                      Eterm visability,
+                                      Eterm reason)
 {
-    if (dist || is_value(nodename)) {
+    if (dist || is_value(nodename) || seq) {
         ErtsSchedulerData *esdp;
-        ErtsConMonLnkCleanup *cmlcp;
+        ErtsConMonLnkSeqCleanup *cmlcp;
         Uint rsz, size;
 
-        size = sizeof(ErtsConMonLnkCleanup);
+        size = sizeof(ErtsConMonLnkSeqCleanup);
 
         if (is_non_value(reason) || is_immed(reason)) {
             rsz = 0;
@@ -324,6 +400,8 @@ schedule_con_monitor_link_cleanup(ErtsMonLnkDist *dist,
             erts_mtx_unlock(&dist->mtx);
         }
 
+        cmlcp->seq = seq;
+
         cmlcp->trigger_node_monitors = is_value(nodename);
         cmlcp->nodename = nodename;
         cmlcp->visability = visability;
@@ -337,13 +415,13 @@ schedule_con_monitor_link_cleanup(ErtsMonLnkDist *dist,
         esdp = erts_get_scheduler_data();
         ASSERT(esdp && esdp->type == ERTS_SCHED_NORMAL);
         erts_schedule_misc_aux_work((int) esdp->no,
-                                    con_monitor_link_cleanup,
+                                    con_monitor_link_seq_cleanup,
                                     (void *) cmlcp);
     }
 }
 
 /*
-** A full node name constists of a "n@h"
+** A full node name consists of a "n@h"
 **
 ** n must be a valid node name: string of ([a-z][A-Z][0-9]_-)+
 ** 
@@ -560,6 +638,7 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
     }
     else { /* Call from distribution controller (port/process) */
         ErtsMonLnkDist *mld;
+        DistSeqNode *sequences;
         ErtsAtomCache *cache;
         ErtsProcList *suspendees;
         ErtsDistOutputBuf *obuf;
@@ -589,6 +668,9 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
         mld = dep->mld;
         dep->mld = NULL;
 
+        sequences = dep->sequences;
+        dep->sequences = NULL;
+
 	nodename = dep->sysname;
 	flags = dep->flags;
 
@@ -612,14 +694,15 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
 
 	erts_de_rwunlock(dep);
 
-        schedule_con_monitor_link_cleanup(mld,
-                                          nodename,
-                                          (flags & DFLAG_PUBLISHED
-                                           ? am_visible
-                                           : am_hidden),
-                                          (reason == am_normal
-                                           ? am_connection_closed
-                                           : reason));
+        schedule_con_monitor_link_seq_cleanup(mld,
+                                              sequences,
+                                              nodename,
+                                              (flags & DFLAG_PUBLISHED
+                                               ? am_visible
+                                               : am_hidden),
+                                              (reason == am_normal
+                                               ? am_connection_closed
+                                               : reason));
 
         erts_resume_processes(suspendees);
 
@@ -653,6 +736,16 @@ void init_dist(void)
 {
     init_nodes_monitors();
 
+#ifdef ERTS_DIST_MSG_DBG_FILE
+    {
+        char buff[255];
+        sprintf(buff, ERTS_DIST_MSG_DBG_FILE, getpid());
+        dbg_file = fopen(buff,"w+");
+    }
+#elif defined (ERTS_DIST_MSG_DBG)
+    dbg_file = stderr;
+#endif
+
     nodedown.reason = NIL;
     nodedown.bp = NULL;
 
@@ -676,21 +769,27 @@ void init_dist(void)
     }
 }
 
-#define ErtsDistOutputBuf2Binary(OB) \
-  ((Binary *) (((char *) (OB)) - offsetof(Binary, orig_bytes)))
+#define ErtsDistOutputBuf2Binary(OB) OB->bin
 
 static ERTS_INLINE ErtsDistOutputBuf *
-alloc_dist_obuf(Uint size)
+alloc_dist_obuf(Uint size, Uint headers)
 {
+    int i;
     ErtsDistOutputBuf *obuf;
-    Uint obuf_size = sizeof(ErtsDistOutputBuf)+sizeof(byte)*(size-1);
+    Uint obuf_size = sizeof(ErtsDistOutputBuf)*(headers) +
+        sizeof(byte)*size;
     Binary *bin = erts_bin_drv_alloc(obuf_size);
-    obuf = (ErtsDistOutputBuf *) &bin->orig_bytes[0];
+    obuf = (ErtsDistOutputBuf *) &bin->orig_bytes[size];
+    erts_refc_add(&bin->intern.refc, headers - 1, 1);
+    for (i = 0; i < headers; i++) {
+        obuf[i].bin = bin;
+        obuf[i].extp = (byte *)&bin->orig_bytes[0];
 #ifdef DEBUG
-    obuf->dbg_pattern = ERTS_DIST_OUTPUT_BUF_DBG_PATTERN;
-    obuf->alloc_endp = obuf->data + size;
-    ASSERT(bin == ErtsDistOutputBuf2Binary(obuf));
+        obuf[i].dbg_pattern = ERTS_DIST_OUTPUT_BUF_DBG_PATTERN;
+        obuf[i].alloc_endp = obuf->extp + size;
+        ASSERT(bin == ErtsDistOutputBuf2Binary(obuf));
 #endif
+    }
     return obuf;
 }
 
@@ -705,8 +804,8 @@ free_dist_obuf(ErtsDistOutputBuf *obuf)
 static ERTS_INLINE Sint
 size_obuf(ErtsDistOutputBuf *obuf)
 {
-    Binary *bin = ErtsDistOutputBuf2Binary(obuf);
-    return bin->orig_size;
+    return sizeof(ErtsDistOutputBuf) + (obuf->ext_endp - obuf->ext_start)
+        + (obuf->hdr_endp - obuf->hdrp);
 }
 
 static ErtsDistOutputBuf* clear_de_out_queues(DistEntry* dep)
@@ -771,7 +870,9 @@ int erts_dsend_context_dtor(Binary* ctx_bin)
     default:;
     }
     if (ctx->dss.phase >= ERTS_DSIG_SEND_PHASE_ALLOC && ctx->dss.obuf) {
-	free_dist_obuf(ctx->dss.obuf);
+        int i;
+        for (i = 0; i < ctx->dss.fragments; i++)
+            free_dist_obuf(&ctx->dss.obuf[i]);
     }
     if (ctx->deref_dep)
 	erts_deref_dist_entry(ctx->dep);
@@ -845,8 +946,8 @@ erts_dsig_send_unlink(ErtsDSigData *dsdp, Eterm local, Eterm remote)
 /* A local process that's being monitored by a remote one exits. We send:
    {DOP_MONITOR_P_EXIT, Local pid or name, Remote pid, ref, reason} */
 int
-erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm watcher, Eterm watched, 
-			  Eterm ref, Eterm reason)
+erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm from, Eterm watcher, Eterm watched, 
+                      Eterm ref, Eterm reason)
 {
     Eterm ctl, msg;
     DeclareTmpHeapNoproc(ctl_heap,6);
@@ -871,7 +972,7 @@ erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
         msg = THE_NON_VALUE;
     }
 
-    res = dsig_send_exit(dsdp, ctl, msg, 1);
+    res = dsig_send_exit(dsdp, ctl, msg, from, 1);
     UnUseTmpHeapNoproc(6);
     return res;
 }
@@ -1031,8 +1132,7 @@ erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
 }
 
 int
-erts_dsig_send_reg_msg(Eterm remote_name, Eterm message,
-		       ErtsSendContext* ctx)
+erts_dsig_send_reg_msg(Eterm remote_name, Eterm message, ErtsSendContext* ctx)
 {
     Eterm ctl;
     Eterm token = NIL;
@@ -1149,13 +1249,13 @@ erts_dsig_send_exit_tt(ErtsDSigData *dsdp, Eterm local, Eterm remote,
 #endif
     DTRACE7(process_exit_signal_remote, sender_name, node_name,
             remote_name, reason_str, tok_label, tok_lastcnt, tok_serial);
-    res = dsig_send_exit(dsdp, ctl, msg, 1);
+    res = dsig_send_exit(dsdp, ctl, msg, local, 1);
     UnUseTmpHeapNoproc(6);
     return res;
 }
 
 int
-erts_dsig_send_exit(ErtsDSigData *dsdp, Eterm local, Eterm remote, Eterm reason)
+erts_dsig_send_exit(ErtsDSigData *dsdp, Eterm from, Eterm local, Eterm remote, Eterm reason)
 {
     DeclareTmpHeapNoproc(ctl_heap,5);
     int res;
@@ -1170,7 +1270,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);
         msg = THE_NON_VALUE;
     }
-    res = dsig_send_exit(dsdp, ctl, msg, 1);
+    res = dsig_send_exit(dsdp, ctl, msg, from, 1);
     UnUseTmpHeapNoproc(5);
     return res;
 }
@@ -1193,7 +1293,7 @@ erts_dsig_send_exit2(ErtsDSigData *dsdp, Eterm local, Eterm remote, Eterm reason
         msg = THE_NON_VALUE;
     }
 
-    res = dsig_send_exit(dsdp, ctl, msg, 0);
+    res = dsig_send_exit(dsdp, ctl, msg, local, 0);
     UnUseTmpHeapNoproc(5);
     return res;
 }
@@ -1215,17 +1315,123 @@ erts_dsig_send_group_leader(ErtsDSigData *dsdp, Eterm leader, Eterm remote)
     return res;
 }
 
-#if defined(PURIFY)
-#  define PURIFY_MSG(msg) \
-    purify_printf("%s, line %d: %s", __FILE__, __LINE__, msg)
-#elif defined(VALGRIND)
-#include <valgrind/valgrind.h>
-#include <valgrind/memcheck.h>
-#  define PURIFY_MSG(msg)                                                    \
-    VALGRIND_PRINTF("%s, line %d: %s", __FILE__, __LINE__, msg)
-#else
-#  define PURIFY_MSG(msg)
-#endif
+struct dist_sequences {
+    ErlHeapFragment hfrag;
+    struct dist_sequences *parent;
+    struct dist_sequences *left;
+    struct dist_sequences *right;
+    char is_red;
+
+    Uint64 seq_id;
+    int cnt;
+    Sint ctl_len;
+};
+
+#define ERTS_RBT_PREFIX dist_seq
+#define ERTS_RBT_T DistSeqNode
+#define ERTS_RBT_KEY_T Uint
+#define ERTS_RBT_FLAGS_T int
+#define ERTS_RBT_INIT_EMPTY_TNODE(T)            \
+    do {                                        \
+        (T)->parent = NULL;                     \
+        (T)->left = NULL;                       \
+        (T)->right = NULL;                      \
+        (T)->is_red = 0;                        \
+    } while(0)
+#define ERTS_RBT_IS_RED(T) ((T)->is_red)
+#define ERTS_RBT_SET_RED(T) ((T)->is_red = 1)
+#define ERTS_RBT_IS_BLACK(T) (!ERTS_RBT_IS_RED(T))
+#define ERTS_RBT_SET_BLACK(T) ((T)->is_red = 0)
+#define ERTS_RBT_GET_FLAGS(T) ((T)->is_red)
+#define ERTS_RBT_SET_FLAGS(T, F) ((T)->is_red = F)
+#define ERTS_RBT_GET_PARENT(T) ((T)->parent)
+#define ERTS_RBT_SET_PARENT(T, P) ((T)->parent = P)
+#define ERTS_RBT_GET_RIGHT(T) ((T)->right)
+#define ERTS_RBT_SET_RIGHT(T, R) ((T)->right = (R))
+#define ERTS_RBT_GET_LEFT(T) ((T)->left)
+#define ERTS_RBT_SET_LEFT(T, L) ((T)->left = (L))
+#define ERTS_RBT_GET_KEY(T) ((T)->seq_id)
+#define ERTS_RBT_IS_LT(KX, KY) (KX < KY)
+#define ERTS_RBT_IS_EQ(KX, KY) (KX == KY)
+#define ERTS_RBT_WANT_DELETE
+#define ERTS_RBT_WANT_LOOKUP_INSERT
+#define ERTS_RBT_WANT_LOOKUP
+#define ERTS_RBT_WANT_FOREACH
+#define ERTS_RBT_WANT_FOREACH_DESTROY_YIELDING
+
+#include "erl_rbtree.h"
+
+struct erts_dist_seq_tree_foreach_iter_arg {
+    int (*func)(ErtsDistExternal *, void *, Sint);
+    void *arg;
+};
+
+static int
+erts_dist_seq_tree_foreach_iter(DistSeqNode *seq, void *arg, Sint reds)
+{
+    struct erts_dist_seq_tree_foreach_iter_arg *state = arg;
+    return state->func(erts_get_dist_ext(&seq->hfrag), state->arg, reds);
+}
+
+void
+erts_dist_seq_tree_foreach(DistEntry *dep, int (*func)(ErtsDistExternal *, void *, Sint), void *arg)
+{
+    struct erts_dist_seq_tree_foreach_iter_arg state;
+    state.func = func;
+    state.arg = arg;
+    dist_seq_rbt_foreach(dep->sequences, erts_dist_seq_tree_foreach_iter, &state);
+}
+
+static int dist_seq_cleanup(DistSeqNode *seq, void *unused, Sint reds)
+{
+    erts_free_dist_ext_copy(erts_get_dist_ext(&seq->hfrag));
+    free_message_buffer(&seq->hfrag);
+    return 1;
+}
+
+typedef struct {
+    DistSeqNode *root;
+    dist_seq_rbt_yield_state_t rbt_ystate;
+} DistSeqNodeYieldState;
+
+int
+erts_dist_seq_tree_foreach_delete_yielding(DistSeqNode **root,
+                                           void **vyspp,
+                                           Sint limit)
+{
+    DistSeqNodeYieldState ys = {*root, ERTS_RBT_YIELD_STAT_INITER};
+    DistSeqNodeYieldState *ysp;
+    int res;
+
+    ysp = (DistSeqNodeYieldState *) *vyspp;
+    if (!ysp) {
+        *root = NULL;
+	ysp = &ys;
+    }
+    res = dist_seq_rbt_foreach_destroy_yielding(&ysp->root,
+                                               dist_seq_cleanup,
+                                               NULL,
+                                               &ysp->rbt_ystate,
+                                               limit);
+    if (res > 0) {
+	if (ysp != &ys)
+	    erts_free(ERTS_ALC_T_ML_YIELD_STATE, ysp);
+	*vyspp = NULL;
+    }
+    else {
+
+	if (ysp == &ys) {
+	    ysp = erts_alloc(ERTS_ALC_T_SEQ_YIELD_STATE,
+			     sizeof(DistSeqNodeYieldState));
+	    sys_memcpy((void *) ysp, (void *) &ys,
+		       sizeof(DistSeqNodeYieldState));
+	}
+
+	*vyspp = (void *) ysp;
+    }
+
+    return res;
+}
 
 /*
 ** Input from distribution port.
@@ -1246,7 +1452,9 @@ int erts_net_message(Port *prt,
 		     byte *buf,
 		     ErlDrvSizeT len)
 {
-    ErtsDistExternal ede;
+    ErtsDistExternal ede, *edep = &ede;
+    ErtsDistExternalData ede_data;
+    ErlHeapFragment *ede_hfrag = NULL;
     Sint ctl_len;
     Eterm arg;
     Eterm from, to;
@@ -1285,10 +1493,12 @@ int erts_net_message(Port *prt,
     }
 
 #ifdef ERTS_RAW_DIST_MSG_DBG
-    erts_fprintf(stderr, "<< ");
+    erts_fprintf(dbg_file, "RECV: ");
     bw(buf, len);
 #endif
 
+    ede.data = &ede_data;
+
     res = erts_prepare_dist_ext(&ede, buf, len, bin, dep, conn_id, dep->cache);
 
     switch (res) {
@@ -1296,51 +1506,164 @@ int erts_net_message(Port *prt,
         return 0; /* Connection not alive; ignore signal... */
     case ERTS_PREP_DIST_EXT_FAILED:
 #ifdef ERTS_DIST_MSG_DBG
-	erts_fprintf(stderr, "DIST MSG DEBUG: erts_prepare_dist_ext() failed:\n");
+	erts_fprintf(dbg_file, "DIST MSG DEBUG: erts_prepare_dist_ext() failed:\n");
 	bw(buf, orig_len);
 #endif
         goto data_error;
     case ERTS_PREP_DIST_EXT_SUCCESS:
-	ctl_len = erts_decode_dist_ext_size(&ede);
+	ctl_len = erts_decode_dist_ext_size(&ede, 1);
         if (ctl_len < 0) {
 #ifdef ERTS_DIST_MSG_DBG
-            erts_fprintf(stderr, "DIST MSG DEBUG: erts_decode_dist_ext_size(CTL) failed:\n");
+            erts_fprintf(dbg_file, "DIST MSG DEBUG: erts_decode_dist_ext_size(CTL) failed:\n");
             bw(buf, orig_len);
 #endif
             PURIFY_MSG("data error");
             goto data_error;
         }
+
+        /* A non-fragmented message */
+        if (!ede.data->seq_id) {
+            if (ctl_len > DIST_CTL_DEFAULT_SIZE) {
+                ctl = erts_alloc(ERTS_ALC_T_DCTRL_BUF, ctl_len * sizeof(Eterm));
+            }
+
+            erts_factory_tmp_init(&factory, ctl, ctl_len, ERTS_ALC_T_DCTRL_BUF);
+            break;
+        } else {
+            DistSeqNode *seq;
+            Uint sz = erts_dist_ext_size(&ede);
+            Uint used_sz = ctl_len * sizeof(Eterm);
+
+            /* We calculate the size of the heap fragment to be allocated.
+               The used_size part has to be larger that the ctl data and the
+               DistSeqNode. */
+            if (used_sz + (sizeof(ErlHeapFragment) - sizeof(Eterm)) < sizeof(DistSeqNode))
+                used_sz = sizeof(DistSeqNode) - (sizeof(ErlHeapFragment) - sizeof(Eterm));
+
+            seq = (DistSeqNode *)new_message_buffer((sz + used_sz) / sizeof(Eterm));
+            seq->hfrag.used_size = used_sz / sizeof(Eterm);
+
+            seq->ctl_len = ctl_len;
+            seq->seq_id = ede.data->seq_id;
+            seq->cnt = ede.data->frag_id;
+            if (dist_seq_rbt_lookup_insert(&dep->sequences, seq) != NULL) {
+                free_message_buffer(&seq->hfrag);
+                goto data_error;
+            }
+
+            erts_make_dist_ext_copy(&ede, erts_get_dist_ext(&seq->hfrag));
+
+            if (ede.data->frag_id > 1) {
+                seq->cnt--;
+                return 0;
+            }
+        }
+
+        /* fall through, the first fragment in the sequence was the last fragment */
+    case ERTS_PREP_DIST_EXT_FRAG_CONT: {
+        DistSeqNode *seq = dist_seq_rbt_lookup(dep->sequences, ede.data->seq_id);
+
+        if (!seq)
+            goto data_error;
+
+        /* If we did a fall-though we already did this */
+        if (res == ERTS_PREP_DIST_EXT_FRAG_CONT)
+            erts_dist_ext_frag(&ede_data, erts_get_dist_ext(&seq->hfrag));
+
+        /* Verify that the fragments have arrived in the correct order */
+        if (seq->cnt != ede.data->frag_id)
+            goto data_error;
+
+        seq->cnt--;
+
+        /* Check if this was the last fragment */
+        if (ede.data->frag_id > 1)
+            return 0;
+
+        /* Last fragment arrived, time to dispatch the signal */
+        dist_seq_rbt_delete(&dep->sequences, seq);
+        ctl_len = seq->ctl_len;
+
+        /* Now that we no longer need the DistSeqNode we re-use the heapfragment
+           to decode the ctl msg into. We don't need the ctl message to be in
+           the heapfragment, but we decode into the heapfragment speculatively
+           in case there is a trace token that we need. */
+        erts_factory_heap_frag_init(&factory, &seq->hfrag);
+        edep = erts_get_dist_ext(&seq->hfrag);
+        ede_hfrag = &seq->hfrag;
+
+        /* If the sequence consisted of more than 1 fragment we create one large
+           binary out of all of the fragments. This because erts_decode_ext
+           cannot handle a segmented buffer.
+           TODO: Move this copy to as late as possible, preferably in in the
+           erts_decode_dist_ext in the receiving process.
+        */
+        if (edep->data->frag_id > 1) {
+            Uint sz = 0;
+            Binary *bin;
+            int i;
+            byte *ep;
+
+            for (i = 0; i < edep->data->frag_id; i++)
+                sz += edep->data[i].ext_endp - edep->data[i].extp;
+
+            bin = erts_bin_nrml_alloc(sz);
+            ep = (byte*)bin->orig_bytes;
+
+            for (i = 0; i < edep->data->frag_id; i++) {
+                sys_memcpy(ep, edep->data[i].extp, edep->data[i].ext_endp - edep->data[i].extp);
+                ep += edep->data[i].ext_endp - edep->data[i].extp;
+                erts_bin_release(edep->data[i].binp);
+                edep->data[i].binp = NULL;
+                edep->data[i].extp = NULL;
+                edep->data[i].ext_endp = NULL;
+            }
+
+            edep->data->frag_id = 1;
+            edep->data->extp = (byte*)bin->orig_bytes;
+            edep->data->ext_endp = ep;
+            edep->data->binp = bin;
+        }
+
         break;
+    }
     default:
         ERTS_INTERNAL_ERROR("Unexpected result from erts_prepare_dist_ext()");
         break;
     }
 
-    if (ctl_len > DIST_CTL_DEFAULT_SIZE) {
-	ctl = erts_alloc(ERTS_ALC_T_DCTRL_BUF, ctl_len * sizeof(Eterm));
-    }
-
-    erts_factory_tmp_init(&factory, ctl, ctl_len, ERTS_ALC_T_DCTRL_BUF);
-    arg = erts_decode_dist_ext(&factory, &ede);
+    arg = erts_decode_dist_ext(&factory, edep, 1);
     if (is_non_value(arg)) {
 #ifdef ERTS_DIST_MSG_DBG
-	erts_fprintf(stderr, "DIST MSG DEBUG: erts_decode_dist_ext(CTL) failed:\n");
+	erts_fprintf(dbg_file, "DIST MSG DEBUG: erts_decode_dist_ext(CTL) failed:\n");
 	bw(buf, orig_len);
 #endif
 	PURIFY_MSG("data error");
 	goto decode_error;
     }
 
-#ifdef ERTS_DIST_MSG_DBG
-    erts_fprintf(stderr, "<< CTL: %.80T\n", arg);
-#endif
+    /* Fill the unused part of the hfrag with a bignum header */
+    if (ede_hfrag && ede_hfrag->mem + ede_hfrag->used_size > factory.hp) {
+        Uint slot = factory.hp - ede_hfrag->mem;
+        ede_hfrag->mem[slot] = make_pos_bignum_header(ede_hfrag->used_size - slot - 1);
+    }
 
     if (is_not_tuple(arg) || 
 	(tuple = tuple_val(arg), (tuple_arity = arityval(*tuple)) < 1) ||
 	is_not_small(tuple[1])) {
+#ifdef ERTS_DIST_MSG_DBG
+        if (is_tuple(arg) && arityval(*tuple) > 1)
+            erts_fprintf(dbg_file, "RECV: CTL: %s: %.80T\n",
+                         erts_dop_to_string(unsigned_val(tuple[1])), arg);
+#endif
  	goto invalid_message;
     }
 
+#ifdef ERTS_DIST_MSG_DBG
+    erts_fprintf(dbg_file, "RECV: CTL: %s: %.80T\n",
+                 erts_dop_to_string(unsigned_val(tuple[1])), arg);
+#endif
+
     token = NIL;
 
     switch (type = unsigned_val(tuple[1])) {
@@ -1385,7 +1708,7 @@ int erts_net_message(Port *prt,
 
         code = erts_dsig_prepare(&dsd, dep, NULL, 0, ERTS_DSP_NO_LOCK, 0, 0);
         if (code == ERTS_DSIG_PREP_CONNECTED) {
-            code = erts_dsig_send_exit(&dsd, to, from, am_noproc);
+            code = erts_dsig_send_exit(&dsd, to, to, from, am_noproc);
             ASSERT(code == ERTS_DSIG_SEND_OK);
         }
 
@@ -1475,7 +1798,7 @@ int erts_net_message(Port *prt,
 
         code = erts_dsig_prepare(&dsd, dep, NULL, 0, ERTS_DSP_NO_LOCK, 0, 0);
         if (code == ERTS_DSIG_PREP_CONNECTED) {
-            code = erts_dsig_send_m_exit(&dsd, watcher, watched, ref,
+            code = erts_dsig_send_m_exit(&dsd, pid, watcher, watched, ref,
                                          am_noproc);
             ASSERT(code == ERTS_DSIG_SEND_OK);
         }
@@ -1549,7 +1872,7 @@ int erts_net_message(Port *prt,
 	}
 
 #ifdef ERTS_DIST_MSG_DBG
-	dist_msg_dbg(&ede, "MSG", buf, orig_len);
+	dist_msg_dbg(edep, "MSG", buf, orig_len);
 #endif
 
 	from = tuple[2];
@@ -1560,7 +1883,6 @@ int erts_net_message(Port *prt,
 	rp = erts_whereis_process(NULL, 0, to, 0, 0);
 	if (rp) {
 	    ErtsProcLocks locks = 0;
-	    ErtsDistExternal *ede_copy;
 
 	    if (type == DOP_REG_SEND) {
 		token = NIL;
@@ -1568,13 +1890,14 @@ int erts_net_message(Port *prt,
 		token = tuple[5];
 	    }
 
-	    ede_copy = erts_make_dist_ext_copy(&ede, &token);
-
-            erts_queue_dist_message(rp, locks, ede_copy, token, from);
+            erts_queue_dist_message(rp, locks, edep, ede_hfrag, token, from);
 
 	    if (locks)
 		erts_proc_unlock(rp, locks);
-	}
+	} else if (ede_hfrag) {
+            erts_free_dist_ext_copy(erts_get_dist_ext(ede_hfrag));
+            free_message_buffer(ede_hfrag);
+        }
 	break;
 
     case DOP_SEND_SENDER_TT: {
@@ -1606,7 +1929,7 @@ int erts_net_message(Port *prt,
                : tuple[2] == am_Empty);
 
 #ifdef ERTS_DIST_MSG_DBG
-	dist_msg_dbg(&ede, "MSG", buf, orig_len);
+	dist_msg_dbg(edep, "MSG", buf, orig_len);
 #endif
 	to = tuple[3];
 	if (is_not_pid(to)) {
@@ -1615,20 +1938,19 @@ int erts_net_message(Port *prt,
 	rp = erts_proc_lookup(to);
 	if (rp) {
 	    ErtsProcLocks locks = 0;
-	    ErtsDistExternal *ede_copy;
-
-	    ede_copy = erts_make_dist_ext_copy(&ede, &token);
 
-	    erts_queue_dist_message(rp, locks, ede_copy, token, am_Empty);
+	    erts_queue_dist_message(rp, locks, edep, ede_hfrag, token, am_Empty);
 	    if (locks)
 		erts_proc_unlock(rp, locks);
-	}
+	} else if (ede_hfrag) {
+            erts_free_dist_ext_copy(erts_get_dist_ext(ede_hfrag));
+            free_message_buffer(ede_hfrag);
+        }
 	break;
     }
 
     case DOP_PAYLOAD_MONITOR_P_EXIT:
     case DOP_MONITOR_P_EXIT: {
-        ErtsDistExternal *ede_copy = NULL;
 
 	/* We are monitoring a process on the remote node which dies, we get
 	   {DOP_MONITOR_P_EXIT, Remote pid or name, Local pid, ref, reason} */
@@ -1663,19 +1985,18 @@ int erts_net_message(Port *prt,
             goto invalid_message;
         }
 
+        if (!erts_proc_lookup(watcher)) break; /* Process not alive */
+
         if (reason == THE_NON_VALUE) {
 
 #ifdef ERTS_DIST_MSG_DBG
-            dist_msg_dbg(&ede, "MSG", buf, orig_len);
+            dist_msg_dbg(edep, "MSG", buf, orig_len);
 #endif
 
-            if (!erts_proc_lookup(watcher)) break; /* Process not alive */
-
-            ede_copy = erts_make_dist_ext_copy(&ede, &token);
         }
 
-        erts_proc_sig_send_dist_monitor_down(dep, ref, watched,
-                                             watcher, ede_copy, reason);
+        erts_proc_sig_send_dist_monitor_down(
+            dep, ref, watched, watcher, edep, ede_hfrag, reason);
 	break;
     }
 
@@ -1683,7 +2004,6 @@ int erts_net_message(Port *prt,
     case DOP_PAYLOAD_EXIT_TT:
     case DOP_EXIT_TT:
     case DOP_EXIT: {
-        ErtsDistExternal *ede_copy = NULL;
 
 	/* 'from', which 'to' is linked to, died */
         from = tuple[2];
@@ -1720,19 +2040,16 @@ int erts_net_message(Port *prt,
 	    goto invalid_message;
 	}
 
-        if (reason == THE_NON_VALUE) {
+        if (!erts_proc_lookup(to)) break; /* Process not alive */
 
+        if (reason == THE_NON_VALUE) {
 #ifdef ERTS_DIST_MSG_DBG
-            dist_msg_dbg(&ede, "MSG", buf, orig_len);
+            dist_msg_dbg(edep, "MSG", buf, orig_len);
 #endif
-
-            if (!erts_proc_lookup(to)) break; /* Process not alive */
-
-            ede_copy = erts_make_dist_ext_copy(&ede, &token);
         }
 
         erts_proc_sig_send_dist_link_exit(dep,
-                                          from, to, ede_copy,
+                                          from, to, edep, ede_hfrag,
                                           reason, token);
 	break;
     }
@@ -1740,7 +2057,6 @@ int erts_net_message(Port *prt,
     case DOP_PAYLOAD_EXIT2:
     case DOP_EXIT2_TT:
     case DOP_EXIT2: {
-        ErtsDistExternal *ede_copy = NULL;
 
 	/* 'from' is send an exit signal to 'to' */
         from = tuple[2];
@@ -1777,18 +2093,15 @@ int erts_net_message(Port *prt,
 	    goto invalid_message;
 	}
 
-        if (reason == THE_NON_VALUE) {
+        if (!erts_proc_lookup(to)) break; /* Process not alive */
 
+        if (reason == THE_NON_VALUE) {
 #ifdef ERTS_DIST_MSG_DBG
-            dist_msg_dbg(&ede, "MSG", buf, orig_len);
+            dist_msg_dbg(edep, "MSG", buf, orig_len);
 #endif
-
-            if (!erts_proc_lookup(to)) break; /* Process not alive */
-
-            ede_copy = erts_make_dist_ext_copy(&ede, &token);
         }
 
-        erts_proc_sig_send_dist_exit(dep, from, to, ede_copy, reason, token);
+        erts_proc_sig_send_dist_exit(dep, from, to, edep, ede_hfrag, reason, token);
 	break;
     }
     case DOP_GROUP_LEADER:
@@ -1804,13 +2117,15 @@ int erts_net_message(Port *prt,
         (void) erts_proc_sig_send_group_leader(NULL, to, from, NIL);
 	break;
 
-    default: 
+    default:
 	goto invalid_message;
     }
 
-    erts_factory_close(&factory);
-    if (ctl != ctl_default) {
-	erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
+    if (ede_hfrag == NULL) {
+        erts_factory_close(&factory);
+        if (ctl != ctl_default) {
+            erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
+        }
     }
     UnUseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
     ERTS_CHK_NO_PROC_LOCKS;
@@ -1823,9 +2138,14 @@ int erts_net_message(Port *prt,
     }
 decode_error:
     PURIFY_MSG("data error");
-    erts_factory_close(&factory);
-    if (ctl != ctl_default) {
-	erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
+    if (ede_hfrag == NULL) {
+        erts_factory_close(&factory);
+        if (ctl != ctl_default) {
+            erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
+        }
+    } else {
+        erts_free_dist_ext_copy(erts_get_dist_ext(ede_hfrag));
+        free_message_buffer(ede_hfrag);
     }
 data_error:
     UnUseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
@@ -1834,18 +2154,17 @@ data_error:
     return -1;
 }
 
-static int dsig_send_exit(ErtsDSigData* dsdp, Eterm ctl, Eterm msg, int force_busy)
+static int dsig_send_exit(ErtsDSigData* dsdp, Eterm ctl, Eterm msg, Eterm from, int force_busy)
 {
     struct erts_dsig_send_context ctx;
     int ret;
     ctx.ctl = ctl;
     ctx.msg = msg;
+    ctx.from = from;
     ctx.force_busy = force_busy;
     ctx.force_encode = 1;
     ctx.phase = ERTS_DSIG_SEND_PHASE_INIT;
-#ifdef DEBUG
     ctx.reds = 1; /* provoke assert below (no reduction count with force_encode) */
-#endif
     ret = erts_dsig_send(dsdp, &ctx);
     ASSERT(ret != ERTS_DSIG_SEND_CONTINUE);
     return ret;
@@ -1857,12 +2176,11 @@ static int dsig_send_ctl(ErtsDSigData* dsdp, Eterm ctl, int force_busy)
     int ret;
     ctx.ctl = ctl;
     ctx.msg = THE_NON_VALUE;
+    ctx.from = THE_NON_VALUE;
     ctx.force_busy = force_busy;
     ctx.force_encode = 1;
     ctx.phase = ERTS_DSIG_SEND_PHASE_INIT;
-#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;
@@ -1926,9 +2244,11 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 	    }
 
     #ifdef ERTS_DIST_MSG_DBG
-            erts_fprintf(stderr, ">> CTL: %.80T\n", ctx->ctl);
+            erts_fprintf(dbg_file, "SEND: CTL: %s: %.80T\n",
+                         erts_dop_to_string(unsigned_val(tuple_val(ctx->ctl)[1])),
+                         ctx->ctl);
             if (is_value(ctx->msg))
-                erts_fprintf(stderr, "    MSG: %.80T\n", ctx->msg);
+                erts_fprintf(dbg_file, "    MSG: %.160T\n", ctx->msg);
     #endif
 
 	    ctx->data_size = ctx->max_finalize_prepend;
@@ -1958,16 +2278,32 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 	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;
+            if (ctx->flags & DFLAG_FRAGMENTS && is_value(ctx->msg) && is_not_immed(ctx->msg)) {
+                /* Calculate the max number of fragments that are needed */
+                ASSERT(is_pid(ctx->from) &&
+                       "from has to be a pid because it is used as sequence id");
+                ctx->fragments = ctx->data_size / ERTS_DIST_FRAGMENT_SIZE + 1;
+            } else
+                ctx->fragments = 1;
 
-	    ctx->obuf = alloc_dist_obuf(ctx->data_size);
-	    ctx->obuf->ext_endp = &ctx->obuf->data[0] + ctx->max_finalize_prepend + ctx->dhdr_ext_size;
+	    ctx->dhdr_ext_size = erts_encode_ext_dist_header_size(ctx->acmp, ctx->fragments);
+
+	    ctx->obuf = alloc_dist_obuf(
+                ctx->dhdr_ext_size + ctx->data_size +
+                (ctx->fragments-1) * ERTS_DIST_FRAGMENT_HEADER_SIZE,
+                ctx->fragments);
+            ctx->obuf->ext_start = &ctx->obuf->extp[0];
+	    ctx->obuf->ext_endp = &ctx->obuf->extp[0] + ctx->max_finalize_prepend + 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);
+	    ctx->obuf->extp = erts_encode_ext_dist_header_setup(
+                ctx->obuf->ext_endp, ctx->acmp, ctx->fragments, ctx->from);
 	    /* Encode control message */
 	    erts_encode_dist_ext(ctx->ctl, &ctx->obuf->ext_endp, ctx->flags, ctx->acmp, NULL, NULL);
+
+            ctx->obuf->hdrp = NULL;
+            ctx->obuf->hdr_endp = NULL;
+
 	    if (is_non_value(ctx->msg)) {
                 ctx->obuf->msg_start = NULL;
                 ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
@@ -1988,33 +2324,91 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                     goto done;
                 }
             } else {
-                erts_encode_dist_ext(ctx->msg, &ctx->obuf->ext_endp, ctx->flags, ctx->acmp, NULL, NULL);
+                erts_encode_dist_ext(ctx->msg, &ctx->obuf->ext_endp, ctx->flags,
+                                     ctx->acmp, NULL, NULL);
             }
 
-	    ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
+            ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
 	case ERTS_DSIG_SEND_PHASE_FIN: {
-	    DistEntry *dep = dsdp->dep;
-	    int suspended = 0;
-	    int resume = 0;
 
 	    ASSERT(ctx->obuf->extp < ctx->obuf->ext_endp);
-	    ASSERT(&ctx->obuf->data[0] <= ctx->obuf->extp - ctx->max_finalize_prepend);
-	    ASSERT(ctx->obuf->ext_endp <= &ctx->obuf->data[0] + ctx->data_size);
+	    ASSERT(((byte*)&ctx->obuf->bin->orig_bytes[0]+obuf_list_size) <= ctx->obuf->extp - ctx->max_finalize_prepend);
+	    ASSERT(ctx->obuf->ext_endp <= ((byte*)ctx->obuf->bin->orig_bytes+obuf_list_size) + ctx->data_size + ctx->dhdr_ext_size);
 
 	    ctx->data_size = ctx->obuf->ext_endp - ctx->obuf->extp;
-	    if (ctx->data_size > (Uint) INT_MAX) {
-		free_dist_obuf(ctx->obuf);
-                ctx->obuf = NULL;
-		retval = ERTS_DSIG_SEND_TOO_LRG;
-		goto done;
-	    }
 
             ctx->obuf->hopefull_flags = ctx->u.ec.hopefull_flags;
-	    /*
+
+            if (ctx->fragments > 1) {
+                int fin_fragments;
+                int i;
+                byte *msg = ctx->obuf->msg_start,
+                    *msg_end = ctx->obuf->ext_endp,
+                    *hdrp = msg_end;
+
+                ASSERT((ctx->obuf->hopefull_flags & ctx->flags) == ctx->obuf->hopefull_flags);
+                ASSERT(get_int64(ctx->obuf->extp + 1 + 1 + 8) == ctx->fragments);
+
+                /* Now that encoding is done we know how large the term will
+                   be so we adjust the number of fragments to send. Note that
+                   this can mean that only 1 fragment is sent. */
+                fin_fragments = (ctx->obuf->ext_endp - ctx->obuf->msg_start + ERTS_DIST_FRAGMENT_SIZE-1) /
+                    ERTS_DIST_FRAGMENT_SIZE - 1;
+
+                /* Update the frag_id in the DIST_FRAG_HEADER */
+                put_int64(fin_fragments+1, ctx->obuf->extp + 1 + 1 + 8);
+
+                if (fin_fragments > 0)
+                    msg += ERTS_DIST_FRAGMENT_SIZE;
+                else
+                    msg = msg_end;
+                ctx->obuf->next = &ctx->obuf[1];
+                ctx->obuf->ext_endp = msg;
+
+                /* Loop through all fragments, updating the output buffers
+                   to be correct and also writing the DIST_FRAG_CONT header. */
+                for (i = 1; i < fin_fragments + 1; i++) {
+                    ctx->obuf[i].hopefull_flags = 0;
+                    ctx->obuf[i].extp = msg;
+                    ctx->obuf[i].ext_start = msg;
+                    if (msg + ERTS_DIST_FRAGMENT_SIZE > msg_end)
+                        ctx->obuf[i].ext_endp = msg_end;
+                    else {
+                        msg += ERTS_DIST_FRAGMENT_SIZE;
+                        ctx->obuf[i].ext_endp = msg;
+                    }
+                    ASSERT(ctx->obuf[i].ext_endp > ctx->obuf[i].extp);
+                    ctx->obuf[i].hdrp = erts_encode_ext_dist_header_fragment(
+                        &hdrp, fin_fragments - i + 1, ctx->from);
+                    ctx->obuf[i].hdr_endp = hdrp;
+                    ctx->obuf[i].next = &ctx->obuf[i+1];
+                }
+                /* If the initial fragment calculation was incorrect we free the
+                   remaining output buffers. */
+                for (; i < ctx->fragments; i++) {
+                    free_dist_obuf(&ctx->obuf[i]);
+                }
+                if (!ctx->force_encode && !ctx->force_busy)
+                    ctx->reds -= ctx->fragments;
+                ctx->fragments = fin_fragments + 1;
+            }
+
+            ctx->phase = ERTS_DSIG_SEND_PHASE_SEND;
+
+            if (ctx->reds <= 0) {
+                retval = ERTS_DSIG_SEND_CONTINUE;
+                goto done;
+            }
+        }
+        case ERTS_DSIG_SEND_PHASE_SEND: {
+            /*
 	     * Signal encoded; now verify that the connection still exists,
 	     * and if so enqueue the signal and schedule it for send.
 	     */
-	    ctx->obuf->next = NULL;
+            DistEntry *dep = dsdp->dep;
+	    int suspended = 0;
+	    int resume = 0;
+            int i;
 	    erts_de_rlock(dep);
 	    cid = dep->cid;
 	    if (dep->state == ERTS_DE_STATE_EXITING
@@ -2022,14 +2416,38 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                 || dep->connection_id != dsdp->connection_id) {
 		/* Not the same connection as when we started; drop message... */
 		erts_de_runlock(dep);
-		free_dist_obuf(ctx->obuf);
+                for (i = 0; i < ctx->fragments; i++)
+                    free_dist_obuf(&ctx->obuf[i]);
+                ctx->fragments = 0;
 	    }
 	    else {
-                Sint qsize;
+                Sint qsize = erts_atomic_read_nob(&dep->qsize);
                 erts_aint32_t qflgs;
 		ErtsProcList *plp = NULL;
                 Eterm notify_proc = NIL;
-                Sint obsz = size_obuf(ctx->obuf);
+                Sint obsz;
+                int fragments;
+
+                /* Calculate how many fragments to send. This depends on
+                   the available space in the distr queue and the amount
+                   of remaining reductions. */
+                for (fragments = 0, obsz = 0;
+                     fragments < ctx->fragments &&
+                         ((ctx->reds > 0 && (qsize + obsz) < erts_dist_buf_busy_limit) ||
+                          ctx->force_encode || ctx->force_busy);
+                     fragments++) {
+#ifdef DEBUG
+                    int reds = 100;
+#else
+                    int reds = 10;
+#endif
+                    if (!ctx->force_encode && !ctx->force_busy)
+                        ctx->reds -= reds;
+                    obsz += size_obuf(&ctx->obuf[fragments]);
+                }
+
+                ASSERT(fragments == ctx->fragments ||
+                       (!ctx->force_encode && !ctx->force_busy));
 
 		erts_mtx_lock(&dep->qlock);
 		qsize = erts_atomic_add_read_nob(&dep->qsize, (erts_aint_t) obsz);
@@ -2059,12 +2477,25 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 		    erts_mtx_lock(&dep->qlock);
 		}
 
-		/* 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 (fragments > 1) {
+                    if (!ctx->obuf->hdrp) {
+                        ASSERT(get_int64(ctx->obuf->extp + 10) == ctx->fragments);
+                    } else {
+                        ASSERT(get_int64(ctx->obuf->hdrp + 10) == ctx->fragments);
+                    }
+                }
+
+                if (fragments) {
+                    ctx->obuf[fragments-1].next = NULL;
+                    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[fragments-1];
+
+                    ctx->fragments -= fragments;
+                    ctx->obuf = &ctx->obuf[fragments];
+                }
 
 		if (!ctx->force_busy) {
                     qflgs = erts_atomic32_read_nob(&dep->qflgs);
@@ -2113,6 +2544,13 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 		     * erroneously scheduled when it shouldn't be.
 		     */
 		}
+                /* More fragments left to be sent, yield and re-schedule */
+                if (ctx->fragments) {
+                    retval = ERTS_DSIG_SEND_CONTINUE;
+                    if (!resume && erts_system_monitor_flags.busy_dist_port)
+                        monitor_generic(ctx->c_p, am_busy_dist_port, cid);
+                    goto done;
+                }
 	    }
 	    ctx->obuf = NULL;
 
@@ -2197,9 +2635,9 @@ static Uint
 dist_port_commandv(Port *prt, ErtsDistOutputBuf *obuf)
 {
     int fpe_was_unmasked;
-    ErlDrvSizeT size;
-    SysIOVec iov[2];
-    ErlDrvBinary* bv[2];
+    ErlDrvSizeT size = 0;
+    SysIOVec iov[3];
+    ErlDrvBinary* bv[3];
     ErlIOVec eiov;
 
     ERTS_CHK_NO_PROC_LOCKS;
@@ -2214,12 +2652,31 @@ dist_port_commandv(Port *prt, ErtsDistOutputBuf *obuf)
         eiov.vsize = 1;
     }
     else {
-        size = obuf->ext_endp - obuf->extp;
+        int i = 1;
         eiov.vsize = 2;
 
-        iov[1].iov_base = obuf->extp;
-        iov[1].iov_len = size;
-        bv[1] = Binary2ErlDrvBinary(ErtsDistOutputBuf2Binary(obuf));
+        if (obuf->hdrp) {
+            eiov.vsize = 3;
+            iov[i].iov_base = obuf->hdrp;
+            iov[i].iov_len = obuf->hdr_endp - obuf->hdrp;
+            size += iov[i].iov_len;
+            bv[i] = Binary2ErlDrvBinary(ErtsDistOutputBuf2Binary(obuf));
+#ifdef ERTS_RAW_DIST_MSG_DBG
+            erts_fprintf(dbg_file, "SEND: ");
+            bw(iov[i].iov_base, iov[i].iov_len);
+#endif
+            i++;
+
+        }
+
+        iov[i].iov_base = obuf->extp;
+        iov[i].iov_len = obuf->ext_endp - obuf->extp;
+#ifdef ERTS_RAW_DIST_MSG_DBG
+            erts_fprintf(dbg_file, "SEND: ");
+            bw(iov[i].iov_base, iov[i].iov_len);
+#endif
+        size += iov[i].iov_len;
+        bv[i] = Binary2ErlDrvBinary(ErtsDistOutputBuf2Binary(obuf));
     }
 
     eiov.size = size;
@@ -2326,6 +2783,23 @@ erts_dist_command(Port *prt, int initial_reds)
     dep->finalized_out_queue.first = NULL;
     dep->finalized_out_queue.last = NULL;
 
+#ifdef DEBUG
+    {
+        Uint sz = 0;
+        ErtsDistOutputBuf *curr = oq.first;
+        while (curr) {
+            sz += size_obuf(curr);
+            curr = curr->next;
+        }
+        curr = foq.first;
+        while (curr) {
+            sz += size_obuf(curr);
+            curr = curr->next;
+        }
+        ASSERT(sz <= erts_atomic_read_nob(&dep->qsize));
+    }
+#endif
+
     sched_flags = erts_atomic32_read_nob(&prt->sched.flags);
 
     if (reds < 0)
@@ -2339,10 +2813,6 @@ erts_dist_command(Port *prt, int initial_reds)
             size = (*send)(prt, foq.first);
             erts_atomic64_inc_nob(&dep->out);
             esdp->io.out += (Uint64) size;
-#ifdef ERTS_RAW_DIST_MSG_DBG
-            erts_fprintf(stderr, ">> ");
-            bw(foq.first->extp, size);
-#endif
 	    reds -= ERTS_PORT_REDS_DIST_CMD_DATA(size);
 	    fob = foq.first;
 	    obufsize += size_obuf(fob);
@@ -2411,15 +2881,11 @@ erts_dist_command(Port *prt, int initial_reds)
                 preempt = 1;
                 break;
             }
-	    ASSERT(&oq.first->data[0] <= oq.first->extp
-		   && oq.first->extp <= oq.first->ext_endp);
+	    ASSERT(oq.first->bin->orig_bytes <= (char*)oq.first->extp
+                   && oq.first->extp <= oq.first->ext_endp);
 	    size = (*send)(prt, oq.first);
             erts_atomic64_inc_nob(&dep->out);
 	    esdp->io.out += (Uint64) size;
-#ifdef ERTS_RAW_DIST_MSG_DBG
-            erts_fprintf(stderr, ">> ");
-            bw(oq.first->extp, size);
-#endif
 	    reds -= ERTS_PORT_REDS_DIST_CMD_DATA(size);
 	    fob = oq.first;
 	    obufsize += size_obuf(fob);
@@ -2556,100 +3022,6 @@ erts_dist_command(Port *prt, int initial_reds)
     goto done;
 }
 
-#if 0
-
-int
-dist_data_finalize(Process *c_p, int reds_limit)
-{
-    int reds = 5;
-    DistEntry *dep = ;
-    ErtsDistOutputQueue oq, foq;
-    ErtsDistOutputBuf *ob;
-    int preempt;
-
-
-    erts_mtx_lock(&dep->qlock);
-    flags = dep->flags;
-    oq.first = dep->out_queue.first;
-    oq.last = dep->out_queue.last;
-    dep->out_queue.first = NULL;
-    dep->out_queue.last = NULL;
-    erts_mtx_unlock(&dep->qlock);
-
-    if (!oq.first) {
-        ASSERT(!oq.last);
-        oq.first = dep->tmp_out_queue.first;
-        oq.last = dep->tmp_out_queue.last;
-    }
-    else {
-        ErtsDistOutputBuf *f, *l;
-        ASSERT(oq.last);
-        if (dep->tmp_out_queue.last) {
-            dep->tmp_out_queue.last->next = oq.first;
-            oq.first = dep->tmp_out_queue.first;
-        }
-    }
-
-    if (!oq.first) {
-        /* Nothing to do... */
-        ASSERT(!oq.last);
-        return reds;
-    }
-
-    foq.first = dep->finalized_out_queue.first;
-    foq.last = dep->finalized_out_queue.last;
-
-    preempt = 0;
-    ob = oq.first;
-    ASSERT(ob);
-
-    do {
-        ob->extp = erts_encode_ext_dist_header_finalize(ob->extp,
-                                                        dep->cache,
-                                                        flags);
-        if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
-            *--ob->extp = PASS_THROUGH; /* Old node; 'pass through'
-                                           needed */
-        ASSERT(&ob->data[0] <= ob->extp && ob->extp < ob->ext_endp);
-        reds += ERTS_PORT_REDS_DIST_CMD_FINALIZE;
-        preempt = reds > reds_limit;
-        if (preempt)
-            break;
-        ob = ob->next;
-    } while (ob);
-    /*
-     * At least one buffer was finalized; if we got preempted,
-     * ob points to the last buffer that we finalized.
-     */
-    if (foq.last)
-        foq.last->next = oq.first;
-    else
-        foq.first = oq.first;
-    if (!preempt) {
-        /* All buffers finalized */
-        foq.last = oq.last;
-        oq.first = oq.last = NULL;
-    }
-    else {
-        /* Not all buffers finalized; split oq. */
-        foq.last = ob;
-        oq.first = ob->next;
-        if (oq.first)
-            ob->next = NULL;
-        else
-            oq.last = NULL;
-    }
-
-    dep->finalized_out_queue.first = foq.first;
-    dep->finalized_out_queue.last = foq.last;
-    dep->tmp_out_queue.first = oq.first;
-    dep->tmp_out_queue.last = oq.last;
-
-    return reds;
-}
-
-#endif
-
 BIF_RETTYPE
 dist_ctrl_get_data_notification_1(BIF_ALIST_1)
 {
@@ -2998,6 +3370,10 @@ static void kill_connection(DistEntry *dep)
 void
 erts_kill_dist_connection(DistEntry *dep, Uint32 conn_id)
 {
+#ifdef ERTS_DIST_MSG_DBG
+    erts_fprintf(dbg_file, "INTR: kill dist conn to %T:%u\n",
+                 dep->sysname, conn_id);
+#endif
     erts_de_rwlock(dep);
     if (conn_id == dep->connection_id
         && dep->state == ERTS_DE_STATE_CONNECTED) {
@@ -3659,8 +4035,9 @@ Sint erts_abort_connection_rwunlock(DistEntry* dep)
         erts_set_dist_entry_not_connected(dep);
 	erts_de_rwunlock(dep);
 
-        schedule_con_monitor_link_cleanup(mld, THE_NON_VALUE,
-                                          THE_NON_VALUE, THE_NON_VALUE);
+        schedule_con_monitor_link_seq_cleanup(
+            mld, NULL, THE_NON_VALUE,
+            THE_NON_VALUE, THE_NON_VALUE);
 
         if (resume_procs) {
             int resumed = erts_resume_processes(resume_procs);