erts: Implement fragmentation of distrubution messages

23 files changed, 1377 insertions, 607 deletions

diff --git a/erts/doc/src/erl_dist_protocol.xml b/erts/doc/src/erl_dist_protocol.xml
index ce5acfbd55..185c75fe84 100644
--- a/erts/doc/src/erl_dist_protocol.xml
+++ b/erts/doc/src/erl_dist_protocol.xml
@@ -899,7 +899,7 @@ DiB == gen_digest(ChA, ICA)?
       <item>
         <p>
           <seealso marker="erl_ext_dist#distribution_header">Distribution header
-          describing the atom cache.
+          describing the atom cache and fragmented distribution messages.
           </seealso>
         </p>
       </item>
diff --git a/erts/doc/src/erl_ext_dist.xml b/erts/doc/src/erl_ext_dist.xml
index b7090d0472..29c6694a6b 100644
--- a/erts/doc/src/erl_ext_dist.xml
+++ b/erts/doc/src/erl_ext_dist.xml
@@ -136,16 +136,9 @@
     </note>
   </section>
 
-  <section>  
+  <section>
     <title>Distribution Header</title>
     <p>
-      <marker id="distribution_header"/>
-      As from ERTS 5.7.2 the old atom cache protocol was
-      dropped and a new one was introduced. This protocol
-      introduced the distribution header. Nodes with an ERTS version
-      earlier than 5.7.2 can still communicate with new nodes,
-      but no distribution header and no atom cache are used.</p>
-    <p>
       The distribution header only contains an atom cache
       reference section, but can in the future contain more
       information. The distribution header precedes one or more Erlang
diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c
index ff7db0e742..a71907505c 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -225,6 +225,7 @@ BIF_RETTYPE link_1(BIF_ALIST_1)
             code = erts_dsig_send_link(&dsd, BIF_P->common.id, BIF_ARG_1);
             if (code == ERTS_DSIG_SEND_YIELD)
                 ERTS_BIF_YIELD_RETURN(BIF_P, am_true);
+            ASSERT(code == ERTS_DSIG_SEND_OK);
             BIF_RET(am_true);
             break;
         }
@@ -2094,6 +2095,7 @@ BIF_RETTYPE send_3(BIF_ALIST_3)
     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;
+    ctx->dss.from = BIF_P->common.id;
 
     while (is_list(l)) {
 	if (CAR(list_val(l)) == am_noconnect) {
@@ -2246,6 +2248,7 @@ Eterm erl_send(Process *p, Eterm to, Eterm msg)
     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;
+    ctx->dss.from = p->common.id;
 
     result = do_send(p, to, msg, &ref, ctx);
 
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);
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index 81a09bc69c..034b75df38 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -48,6 +48,7 @@
 #define DFLAG_BIG_SEQTRACE_LABELS 0x100000
 #define DFLAG_NO_MAGIC            0x200000 /* internal for pending connection */
 #define DFLAG_EXIT_PAYLOAD        0x400000
+#define DFLAG_FRAGMENTS           0x800000
 
 /* Mandatory flags for distribution */
 #define DFLAG_DIST_MANDATORY (DFLAG_EXTENDED_REFERENCES         \
@@ -77,7 +78,8 @@
                             | DFLAG_BIG_CREATION              \
                             | DFLAG_SEND_SENDER               \
                             | DFLAG_BIG_SEQTRACE_LABELS       \
-                            | DFLAG_EXIT_PAYLOAD)
+                            | DFLAG_EXIT_PAYLOAD              \
+                            | DFLAG_FRAGMENTS)
 
 /* Flags addable by local distr implementations */
 #define DFLAG_DIST_ADDABLE    DFLAG_DIST_DEFAULT
@@ -101,7 +103,7 @@
                                | DFLAG_BIG_CREATION)
 
 /* opcodes used in distribution messages */
-enum {
+enum dop {
     DOP_LINK                = 1,
     DOP_SEND                = 2,
     DOP_EXIT                = 3,
@@ -150,6 +152,16 @@ typedef struct {
     Uint32 flags;
 } ErtsDSigData;
 
+/* Must be larger or equal to 16 */
+#ifdef DEBUG
+#define ERTS_DIST_FRAGMENT_SIZE 16
+#else
+/* This should be made configurable */
+#define ERTS_DIST_FRAGMENT_SIZE (64 * 1024)
+#endif
+
+#define ERTS_DIST_FRAGMENT_HEADER_SIZE (1 + 1 + 8 + 8) /* magic, header, seq id, frag id*/
+
 #define ERTS_DE_BUSY_LIMIT (1024*1024)
 extern int erts_dist_buf_busy_limit;
 extern int erts_is_alive;
@@ -347,7 +359,8 @@ enum erts_dsig_send_phase {
     ERTS_DSIG_SEND_PHASE_MSG_SIZE,
     ERTS_DSIG_SEND_PHASE_ALLOC,
     ERTS_DSIG_SEND_PHASE_MSG_ENCODE,
-    ERTS_DSIG_SEND_PHASE_FIN
+    ERTS_DSIG_SEND_PHASE_FIN,
+    ERTS_DSIG_SEND_PHASE_SEND
 };
 
 struct erts_dsig_send_context {
@@ -356,12 +369,14 @@ struct erts_dsig_send_context {
 
     Eterm ctl;
     Eterm msg;
+    Eterm from;
     int force_busy;
     int force_encode;
     Uint32 max_finalize_prepend;
     Uint data_size, dhdr_ext_size;
     ErtsAtomCacheMap *acmp;
     ErtsDistOutputBuf *obuf;
+    Uint fragments;
     Uint32 flags;
     Process *c_p;
     union {
@@ -383,6 +398,7 @@ typedef struct {
     Eterm return_term;
 }ErtsSendContext;
 
+typedef struct dist_sequences DistSeqNode;
 
 /*
  * erts_dsig_send_* return values.
@@ -398,11 +414,11 @@ 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(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_exit(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_exit2(ErtsDSigData *, Eterm, Eterm, Eterm);
 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_m_exit(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm, Eterm);
 
 extern int erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx);
 extern int erts_dsend_context_dtor(Binary*);
@@ -416,4 +432,8 @@ extern Uint erts_dist_cache_size(void);
 
 extern Sint erts_abort_connection_rwunlock(DistEntry *dep);
 
+extern void erts_dist_seq_tree_foreach(
+    DistEntry *dep,
+    int (*func)(ErtsDistExternal *, void*, Sint), void *args);
+
 #endif
diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 31f19d1b88..490a033b8a 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -276,6 +276,7 @@ type	PF3_ARGS	SHORT_LIVED	PROCESSES	process_flag_3_arguments
 type	SETUP_CONN_ARG	SHORT_LIVED	PROCESSES	setup_connection_argument
 type    LIST_TRAP       SHORT_LIVED     PROCESSES       list_bif_trap_state
 type    CONT_EXIT_TRAP  SHORT_LIVED     PROCESSES       continue_exit_trap_state
+type    SEQ_YIELD_STATE SHORT_LIVED     SYSTEM          dist_seq_yield_state
 
 type	ENVIRONMENT	SYSTEM		SYSTEM		environment
 
diff --git a/erts/emulator/beam/erl_message.c b/erts/emulator/beam/erl_message.c
index cd7bce8bf4..e12a51a0ae 100644
--- a/erts/emulator/beam/erl_message.c
+++ b/erts/emulator/beam/erl_message.c
@@ -219,6 +219,10 @@ erts_cleanup_message(ErtsMessage *mp)
     if (ERTS_SIG_IS_MSG(mp) && mp->data.attached != ERTS_MSG_COMBINED_HFRAG) {
         bp = mp->data.heap_frag;
     } else {
+        /* All non msg signals are combined HFRAG messages,
+           but we overwrite the mp->data field with the
+           nm_signal queue ptr so have to fix that here
+           before freeing it. */
         mp->data.attached = ERTS_MSG_COMBINED_HFRAG;
         bp = mp->hfrag.next;
         erts_cleanup_offheap(&mp->hfrag.off_heap);
@@ -234,7 +238,7 @@ erts_cleanup_messages(ErtsMessage *msgp)
     ErtsMessage *mp = msgp;
     while (mp) {
 	ErtsMessage *fmp;
-        erts_cleanup_message(mp);
+	erts_cleanup_message(mp);
 	fmp = mp;
 	mp = mp->next;
 	erts_free_message(fmp);
@@ -266,6 +270,7 @@ void
 erts_queue_dist_message(Process *rcvr,
 			ErtsProcLocks rcvr_locks,
 			ErtsDistExternal *dist_ext,
+                        ErlHeapFragment *hfrag,
 			Eterm token,
                         Eterm from)
 {
@@ -274,8 +279,26 @@ erts_queue_dist_message(Process *rcvr,
 
     ERTS_LC_ASSERT(rcvr_locks == erts_proc_lc_my_proc_locks(rcvr));
 
-    mp = erts_alloc_message(0, NULL);
-    mp->data.dist_ext = dist_ext;
+    if (hfrag) {
+        /* Fragmented message, allocate a message reference */
+        mp = erts_alloc_message(0, NULL);
+        mp->data.heap_frag = hfrag;
+    } else {
+        /* Un-fragmented message, allocate space for
+           token and dist_ext in message. */
+        Uint dist_ext_sz = erts_dist_ext_size(dist_ext) / sizeof(Eterm);
+        Uint token_sz = size_object(token);
+        Uint sz = token_sz + dist_ext_sz;
+        Eterm *hp;
+
+        mp = erts_alloc_message(sz, &hp);
+        mp->data.heap_frag = &mp->hfrag;
+        mp->hfrag.used_size = token_sz;
+
+        erts_make_dist_ext_copy(dist_ext, erts_get_dist_ext(mp->data.heap_frag));
+
+        token = copy_struct(token, token_sz, &hp, &mp->data.heap_frag->off_heap);
+    }
 
     ERL_MESSAGE_FROM(mp) = dist_ext->dep->sysname;
     ERL_MESSAGE_TERM(mp) = THE_NON_VALUE;
@@ -499,25 +522,27 @@ Uint
 erts_msg_attached_data_size_aux(ErtsMessage *msg)
 {
     Sint sz;
-    ASSERT(is_non_value(ERL_MESSAGE_TERM(msg)));
-    ASSERT(msg->data.dist_ext);
-    ASSERT(msg->data.dist_ext->heap_size < 0);
-
-    sz = erts_decode_dist_ext_size(msg->data.dist_ext);
-    if (sz < 0) {
-	/* Bad external
-	 * We leave the message intact in this case as it's not worth the trouble
-	 * to make all callers remove it from queue. It will be detected again
-	 * and removed from message queue later anyway.
-	 */
-	return 0;
-    }
+    ErtsDistExternal *edep = erts_get_dist_ext(msg->data.heap_frag);
+    ASSERT(ERTS_SIG_IS_EXTERNAL_MSG(msg));
+
+    if (edep->heap_size < 0) {
 
-    msg->data.dist_ext->heap_size = sz;
-    if (is_not_nil(msg->m[1])) {
-	ErlHeapFragment *heap_frag;
-	heap_frag = erts_dist_ext_trailer(msg->data.dist_ext);
-	sz += heap_frag->used_size;
+        sz = erts_decode_dist_ext_size(edep, 1);
+        if (sz < 0) {
+            /* Bad external
+             * We leave the message intact in this case as it's not worth the trouble
+             * to make all callers remove it from queue. It will be detected again
+             * and removed from message queue later anyway.
+             */
+            return 0;
+        }
+
+        edep->heap_size = sz;
+    } else {
+        sz = edep->heap_size;
+    }
+    if (is_not_nil(ERL_MESSAGE_TOKEN(msg))) {
+	sz += msg->data.heap_frag->used_size;
     }
     return sz;
 }
@@ -1165,7 +1190,7 @@ erts_factory_message_create(ErtsHeapFactory* factory,
     int on_heap;
     erts_aint32_t state;
 
-    state = proc ? erts_atomic32_read_nob(&proc->state) : 0;
+    state = proc ? erts_atomic32_read_nob(&proc->state) : ERTS_PSFLG_OFF_HEAP_MSGQ;
 
     if (state & ERTS_PSFLG_OFF_HEAP_MSGQ) {
 	msgp = erts_alloc_message(sz, &hp);
@@ -1398,8 +1423,8 @@ void erts_factory_close(ErtsHeapFactory* factory)
 	    else
 		factory->message->data.heap_frag = factory->heap_frags;
 
-	    /* Fall through */
-	case FACTORY_HEAP_FRAGS:
+    /* Fall through */
+    case FACTORY_HEAP_FRAGS:
 	    bp = factory->heap_frags;
 	}
 
diff --git a/erts/emulator/beam/erl_message.h b/erts/emulator/beam/erl_message.h
index 58294648b4..4c2674394e 100644
--- a/erts/emulator/beam/erl_message.h
+++ b/erts/emulator/beam/erl_message.h
@@ -138,7 +138,7 @@ typedef struct erl_heap_fragment ErlHeapFragment;
 struct erl_heap_fragment {
     ErlHeapFragment* next;	/* Next heap fragment */
     ErlOffHeap off_heap;	/* Offset heap data. */
-    Uint alloc_size;		/* Size in (half)words of mem */
+    Uint alloc_size;		/* Size in words of mem */
     Uint used_size;		/* With terms to be moved to heap by GC */
     Eterm mem[1];		/* Data */
 };
@@ -167,7 +167,6 @@ struct erl_heap_fragment {
 #define ERL_MESSAGE_REF_FIELDS__			\
     ErtsMessage *next;	/* Next message */		\
     union {						\
-	ErtsDistExternal *dist_ext;			\
 	ErlHeapFragment *heap_frag;			\
 	void *attached;					\
     } data;						\
@@ -438,7 +437,8 @@ ErlHeapFragment* new_message_buffer(Uint);
 ErlHeapFragment* erts_resize_message_buffer(ErlHeapFragment *, Uint,
 					    Eterm *, Uint);
 void free_message_buffer(ErlHeapFragment *);
-void erts_queue_dist_message(Process*, ErtsProcLocks, ErtsDistExternal *, Eterm, Eterm);
+void erts_queue_dist_message(Process*, ErtsProcLocks, ErtsDistExternal *,
+                             ErlHeapFragment *, Eterm, Eterm);
 void erts_queue_message(Process*, ErtsProcLocks,ErtsMessage*, Eterm, Eterm);
 void erts_queue_proc_message(Process* from,Process* to, ErtsProcLocks,ErtsMessage*, Eterm);
 void erts_queue_proc_messages(Process* from, Process* to, ErtsProcLocks,
@@ -583,22 +583,11 @@ ERTS_GLB_INLINE Uint erts_used_frag_sz(const ErlHeapFragment* bp)
 ERTS_GLB_INLINE Uint erts_msg_attached_data_size(ErtsMessage *msg)
 {
     ASSERT(msg->data.attached);
-    if (is_value(ERL_MESSAGE_TERM(msg))) {
-	ErlHeapFragment *bp;
-        bp = erts_message_to_heap_frag(msg);
-	return erts_used_frag_sz(bp);
-    }
-    else if (msg->data.dist_ext->heap_size < 0)
-	return erts_msg_attached_data_size_aux(msg);
-    else {
-	Uint sz = msg->data.dist_ext->heap_size;
-	if (is_not_nil(ERL_MESSAGE_TOKEN(msg))) {
-	    ErlHeapFragment *heap_frag;
-	    heap_frag = erts_dist_ext_trailer(msg->data.dist_ext);
-	    sz += heap_frag->used_size;
-	}
-	return sz;
-    }
+
+    if (ERTS_SIG_IS_INTERNAL_MSG(msg))
+	return erts_used_frag_sz(erts_message_to_heap_frag(msg));
+
+    return erts_msg_attached_data_size_aux(msg);
 }
 
 #endif
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index b7dbe625a2..ede9f092e9 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -201,6 +201,7 @@ dist_table_alloc(void *dep_tmpl)
     dep->send				= NULL;
     dep->cache				= NULL;
     dep->transcode_ctx                  = NULL;
+    dep->sequences                      = NULL;
 
     /* Link in */
 
@@ -1177,6 +1178,7 @@ static Eterm AM_system;
 static Eterm AM_timer;
 static Eterm AM_delayed_delete_timer;
 static Eterm AM_thread_progress_delete_timer;
+static Eterm AM_sequence;
 static Eterm AM_signal;
 
 static void setup_reference_table(void);
@@ -1218,6 +1220,7 @@ typedef struct dist_referrer_ {
     int ctrl_ref;
     int system_ref;
     int signal_ref;
+    int sequence_ref;
     Eterm id;
     Uint creation;
     Uint id_heap[ID_HEAP_SIZE];
@@ -1272,6 +1275,7 @@ erts_get_node_and_dist_references(struct process *proc)
 	INIT_AM(delayed_delete_timer);
         INIT_AM(thread_progress_delete_timer);
 	INIT_AM(signal);
+	INIT_AM(sequence);
 	references_atoms_need_init = 0;
     }
 
@@ -1309,8 +1313,9 @@ erts_get_node_and_dist_references(struct process *proc)
 #define TIMER_REF 8
 #define SYSTEM_REF 9
 #define SIGNAL_REF 10
+#define SEQUENCE_REF 11
 
-#define INC_TAB_SZ 10
+#define INC_TAB_SZ 11
 
 static void
 insert_dist_referrer(ReferredDist *referred_dist,
@@ -1344,6 +1349,7 @@ insert_dist_referrer(ReferredDist *referred_dist,
 	drp->ctrl_ref = 0;
 	drp->system_ref = 0;
         drp->signal_ref = 0;
+        drp->sequence_ref = 0;
     }
 
     switch (type) {
@@ -1353,6 +1359,7 @@ insert_dist_referrer(ReferredDist *referred_dist,
     case ETS_REF:	drp->ets_ref++;	        break;
     case SYSTEM_REF:	drp->system_ref++;	break;
     case SIGNAL_REF:	drp->signal_ref++;	break;
+    case SEQUENCE_REF:	drp->sequence_ref++;	break;
     default:		ASSERT(0);
     }
 }
@@ -1583,6 +1590,20 @@ insert_dist_monitors(DistEntry *dep)
     }
 }
 
+
+static int
+insert_sequence(ErtsDistExternal *edep, void *arg, Sint reds)
+{
+    insert_dist_entry(edep->dep, SEQUENCE_REF, *(Eterm*)arg, 0);
+    return 1;
+}
+
+static void
+insert_dist_sequences(DistEntry *dep)
+{
+    erts_dist_seq_tree_foreach(dep, insert_sequence, (void *) &dep->sysname);
+}
+
 static void
 clear_visited_p_monitors(ErtsPTabElementCommon *p)
 {
@@ -1774,11 +1795,9 @@ insert_message(ErtsMessage *msg, int type, Process *proc)
         else if (ERTS_SIG_IS_INTERNAL_MSG(msg))
             heap_frag = msg->data.heap_frag;
         else {
-            if (msg->data.dist_ext->dep)
-                insert_dist_entry(msg->data.dist_ext->dep,
-                                  type, proc->common.id, 0);
-            if (is_not_nil(ERL_MESSAGE_TOKEN(msg)))
-                heap_frag = erts_dist_ext_trailer(msg->data.dist_ext);
+            heap_frag = msg->data.heap_frag;
+            insert_dist_entry(erts_get_dist_ext(heap_frag)->dep,
+                              type, proc->common.id, 0);
         }
     }
     while (heap_frag) {
@@ -1819,6 +1838,13 @@ insert_sig_link(ErtsLink *lnk, void *arg, Sint reds)
 }
 
 static void
+insert_sig_ext(ErtsDistExternal *edep, void *arg)
+{
+    Process *proc = arg;
+    insert_dist_entry(edep->dep, SIGNAL_REF, proc->common.id, 0);
+}
+
+static void
 setup_reference_table(void)
 {
     ErlHeapFragment *hfp;
@@ -1898,6 +1924,7 @@ setup_reference_table(void)
                                             insert_sig_offheap,
                                             insert_sig_monitor,
                                             insert_sig_link,
+                                            insert_sig_ext,
                                             (void *) proc);
 
 	    /* Insert links */
@@ -1989,16 +2016,19 @@ setup_reference_table(void)
     for(dep = erts_visible_dist_entries; dep; dep = dep->next) {
         insert_dist_links(dep);
         insert_dist_monitors(dep);
+        insert_dist_sequences(dep);
     }
 
     for(dep = erts_hidden_dist_entries; dep; dep = dep->next) {
         insert_dist_links(dep);
         insert_dist_monitors(dep);
+        insert_dist_sequences(dep);
     }
 
     for(dep = erts_pending_dist_entries; dep; dep = dep->next) {
         insert_dist_links(dep);
         insert_dist_monitors(dep);
+        insert_dist_sequences(dep);
     }
 
     /* Not connected dist entries should not have any links,
@@ -2006,6 +2036,7 @@ setup_reference_table(void)
     for(dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
         insert_dist_links(dep);
         insert_dist_monitors(dep);
+        insert_dist_sequences(dep);
     }
 
     /* Insert all ets tables */
@@ -2179,6 +2210,10 @@ reference_table_term(Uint **hpp, ErlOffHeap *ohp, Uint *szp)
 		tup = MK_2TUP(AM_system, MK_UINT(drp->system_ref));
 		drl = MK_CONS(tup, drl);
 	    }
+            if(drp->sequence_ref) {
+		tup = MK_2TUP(AM_sequence, MK_UINT(drp->sequence_ref));
+		drl = MK_CONS(tup, drl);
+	    }
 	    if(drp->signal_ref) {
 		tup = MK_2TUP(AM_signal, MK_UINT(drp->signal_ref));
 		drl = MK_CONS(tup, drl);
@@ -2253,6 +2288,12 @@ static void noop_sig_offheap(ErlOffHeap *oh, void *arg)
 
 }
 
+static void noop_sig_ext(ErtsDistExternal *ext, void *arg)
+{
+
+}
+
+
 static void
 delete_reference_table(void)
 {
@@ -2303,6 +2344,7 @@ delete_reference_table(void)
                                             noop_sig_offheap,
                                             clear_visited_monitor,
                                             clear_visited_link,
+                                            noop_sig_ext,
                                             (void *) proc);
         }
     }
diff --git a/erts/emulator/beam/erl_node_tables.h b/erts/emulator/beam/erl_node_tables.h
index c44f1f8991..cd5d69a302 100644
--- a/erts/emulator/beam/erl_node_tables.h
+++ b/erts/emulator/beam/erl_node_tables.h
@@ -98,11 +98,22 @@ struct ErtsDistOutputBuf_ {
     byte *alloc_endp;
 #endif
     ErtsDistOutputBuf *next;
-    Uint hopefull_flags;
+    Binary *bin;
+    /* Pointers to the distribution header,
+       if NULL the distr header is in the extp */
+    byte *hdrp;
+    byte *hdr_endp;
+    /* Pointers to the ctl + payload */
     byte *extp;
     byte *ext_endp;
+    /* Start of payload and hopefull_flags, used by transcode */
+    Uint hopefull_flags;
     byte *msg_start;
-    byte data[1];
+    /* start of the ext buffer, this is not always the same as extp
+       as the atom cache handling can use less then the allotted buffer.
+       This value is needed to calculate the size of this output buffer.*/
+    byte *ext_start;
+
 };
 
 typedef struct {
@@ -161,6 +172,8 @@ struct dist_entry_ {
     ErtsThrPrgrLaterOp later_op;
 
     struct transcode_context* transcode_ctx;
+
+    struct dist_sequences *sequences; /* Ongoing distribution sequences */
 };
 
 typedef struct erl_node_ {
diff --git a/erts/emulator/beam/erl_proc_sig_queue.c b/erts/emulator/beam/erl_proc_sig_queue.c
index 1bd219b6d7..8113abebde 100644
--- a/erts/emulator/beam/erl_proc_sig_queue.c
+++ b/erts/emulator/beam/erl_proc_sig_queue.c
@@ -36,6 +36,7 @@
 #include "erl_port_task.h"
 #include "erl_trace.h"
 #include "beam_bp.h"
+#include "erl_binary.h"
 #include "big.h"
 #include "erl_gc.h"
 #include "bif.h"
@@ -80,7 +81,10 @@
 #define ERTS_SIG_Q_TYPE_ADJUST_TRACE_INFO \
     ERTS_SIG_Q_TYPE_MAX
 
-#define ERTS_SIG_IS_EXTERNAL(sig) is_non_value(get_exit_signal_data(sig)->reason)
+#define ERTS_SIG_IS_GEN_EXIT(sig)                                       \
+    (ERTS_PROC_SIG_TYPE(((ErtsSignal *) sig)->common.tag) == ERTS_SIG_Q_TYPE_GEN_EXIT)
+#define ERTS_SIG_IS_GEN_EXIT_EXTERNAL(sig)                              \
+    (ASSERT(ERTS_SIG_IS_GEN_EXIT(sig)),is_non_value(get_exit_signal_data(sig)->reason))
 
 Process *ERTS_WRITE_UNLIKELY(erts_dirty_process_signal_handler);
 Process *ERTS_WRITE_UNLIKELY(erts_dirty_process_signal_handler_high);
@@ -114,8 +118,6 @@ typedef struct {
     Eterm message;
     Eterm from;
     Eterm reason;
-    struct erl_dist_external *dist_ext;
-    ErlHeapFragment *heap_frag;
     union {
         Eterm ref;
         int normal_kills;
@@ -945,13 +947,15 @@ erts_proc_sig_get_external(ErtsMessage *msgp)
 {
     if (ERTS_SIG_IS_EXTERNAL_MSG(msgp)) {
         return erts_get_dist_ext(msgp->data.heap_frag);
-    } else if (ERTS_SIG_IS_NON_MSG(msgp) && ERTS_SIG_IS_EXTERNAL(msgp)) {
+    } else if (ERTS_SIG_IS_NON_MSG(msgp) &&
+               ERTS_SIG_IS_GEN_EXIT(msgp) &&
+               ERTS_SIG_IS_GEN_EXIT_EXTERNAL(msgp)) {
         ErtsDistExternal *edep;
         ErtsExitSignalData *xsigd = get_exit_signal_data(msgp);
         ASSERT(ERTS_PROC_SIG_TYPE(((ErtsSignal *) msgp)->common.tag) == ERTS_SIG_Q_TYPE_GEN_EXIT);
         ASSERT(is_non_value(xsigd->reason));
         if (msgp->hfrag.next == NULL)
-            edep = (ErtsDistExternal*)((void*)xsigd) + sizeof(ErtsExitSignalData);
+            edep = (ErtsDistExternal*)(xsigd + 1);
         else
             edep = erts_get_dist_ext(msgp->hfrag.next);
         return edep;
@@ -965,6 +969,7 @@ static void
 send_gen_exit_signal(Process *c_p, Eterm from_tag,
                      Eterm from, Eterm to,
                      Sint16 op, Eterm reason, ErtsDistExternal *dist_ext,
+                     ErlHeapFragment *dist_ext_hfrag,
                      Eterm ref, Eterm token, int normal_kills)
 {
     ErtsExitSignalData *xsigd;
@@ -972,7 +977,7 @@ send_gen_exit_signal(Process *c_p, Eterm from_tag,
     ErtsMessage *mp;
     ErlHeapFragment *hfrag;
     ErlOffHeap *ohp;
-    Uint hsz, from_sz, reason_sz, ref_sz, token_sz;
+    Uint hsz, from_sz, reason_sz, ref_sz, token_sz, dist_ext_sz;
     int seq_trace;
 #ifdef USE_VM_PROBES
     Eterm s_utag, utag;
@@ -984,7 +989,7 @@ send_gen_exit_signal(Process *c_p, Eterm from_tag,
 
     ASSERT(is_immed(from_tag));
 
-    hsz = sizeof(ErtsExitSignalData)/sizeof(Uint);
+    hsz = sizeof(ErtsExitSignalData)/sizeof(Eterm);
 
     seq_trace = c_p && have_seqtrace(token);
     if (seq_trace)
@@ -1012,6 +1017,8 @@ send_gen_exit_signal(Process *c_p, Eterm from_tag,
     ref_sz = size_object(ref);
     hsz += ref_sz;
 
+    /* The reason was part of the control message,
+       just use copy it into the xsigd */
     if (is_value(reason)) {
         reason_sz = size_object(reason);
         hsz += reason_sz;
@@ -1032,6 +1039,11 @@ send_gen_exit_signal(Process *c_p, Eterm from_tag,
             ERTS_INTERNAL_ERROR("Invalid exit signal op");
             break;
         }
+    } else if (dist_ext != NULL && dist_ext_hfrag == NULL) {
+        /* The message was not fragmented so we need to create space
+           for a single dist_ext element */
+        dist_ext_sz = erts_dist_ext_size(dist_ext) / sizeof(Eterm);
+        hsz += dist_ext_sz;
     }
 
     /*
@@ -1087,13 +1099,12 @@ send_gen_exit_signal(Process *c_p, Eterm from_tag,
 
     hfrag->used_size = hp - start_hp;
 
-    xsigd = (ErtsExitSignalData *) (char *) hp;
+    xsigd = (ErtsExitSignalData *) hp;
 
     xsigd->message = s_message;
     xsigd->from = s_from;
     xsigd->reason = s_reason;
-    xsigd->dist_ext = dist_ext;
-    xsigd->heap_frag = NULL;
+    hfrag->next = dist_ext_hfrag;
 
     if (is_nil(s_ref))
         xsigd->u.normal_kills = normal_kills;
@@ -1102,6 +1113,15 @@ send_gen_exit_signal(Process *c_p, Eterm from_tag,
         xsigd->u.ref = s_ref;
     }
 
+    hp += sizeof(ErtsExitSignalData)/sizeof(Eterm);
+
+    if (dist_ext != NULL && dist_ext_hfrag == NULL && is_non_value(reason)) {
+        erts_make_dist_ext_copy(dist_ext, (ErtsDistExternal *) hp);
+        hp += dist_ext_sz;
+    }
+
+    ASSERT(hp == mp->hfrag.mem + mp->hfrag.alloc_size);
+
     if (seq_trace)
         do_seq_trace_output(to, s_token, s_message);
 
@@ -1234,17 +1254,18 @@ erts_proc_sig_send_exit(Process *c_p, Eterm from, Eterm to,
         from_tag = dep->sysname;
     }
     send_gen_exit_signal(c_p, from_tag, from, to, ERTS_SIG_Q_OP_EXIT,
-                         reason, NULL, NIL, token, normal_kills);
+                         reason, NULL, NULL, NIL, token, normal_kills);
 }
 
 void
 erts_proc_sig_send_dist_exit(DistEntry *dep,
                              Eterm from, Eterm to,
                              ErtsDistExternal *dist_ext,
+                             ErlHeapFragment *hfrag,
                              Eterm reason, Eterm token)
 {
     send_gen_exit_signal(NULL, dep->sysname, from, to, ERTS_SIG_Q_OP_EXIT,
-                         reason, dist_ext, NIL, token, 0);
+                         reason, dist_ext, hfrag, NIL, token, 0);
 
 }
 
@@ -1259,7 +1280,7 @@ erts_proc_sig_send_link_exit(Process *c_p, Eterm from, ErtsLink *lnk,
     if (is_not_immed(reason) || is_not_nil(token)) {
         ASSERT(is_internal_pid(from) || is_internal_port(from));
         send_gen_exit_signal(c_p, from, from, to, ERTS_SIG_Q_OP_EXIT_LINKED,
-                             reason, NULL, NIL, token, 0);
+                             reason, NULL, NULL, NIL, token, 0);
     }
     else {
         /* Pass signal using old link structure... */
@@ -1315,10 +1336,11 @@ void
 erts_proc_sig_send_dist_link_exit(DistEntry *dep,
                                   Eterm from, Eterm to,
                                   ErtsDistExternal *dist_ext,
+                                  ErlHeapFragment *hfrag,
                                   Eterm reason, Eterm token)
 {
     send_gen_exit_signal(NULL, dep->sysname, from, to, ERTS_SIG_Q_OP_EXIT_LINKED,
-                         reason, dist_ext, NIL, token, 0);
+                         reason, dist_ext, hfrag, NIL, token, 0);
 
 }
 
@@ -1342,6 +1364,7 @@ void
 erts_proc_sig_send_dist_monitor_down(DistEntry *dep, Eterm ref,
                                      Eterm from, Eterm to,
                                      ErtsDistExternal *dist_ext,
+                                     ErlHeapFragment *hfrag,
                                      Eterm reason)
 {
     Eterm monitored, heap[3];
@@ -1351,7 +1374,7 @@ erts_proc_sig_send_dist_monitor_down(DistEntry *dep, Eterm ref,
         monitored = from;
     send_gen_exit_signal(NULL, dep->sysname, monitored,
                          to, ERTS_SIG_Q_OP_MONITOR_DOWN,
-                         reason, dist_ext, ref, NIL, 0);
+                         reason, dist_ext, hfrag, ref, NIL, 0);
 }
 
 void
@@ -1422,7 +1445,7 @@ erts_proc_sig_send_monitor_down(ErtsMonitor *mon, Eterm reason)
         }
         send_gen_exit_signal(NULL, from_tag, monitored,
                              to, ERTS_SIG_Q_OP_MONITOR_DOWN,
-                             reason, NULL, mdp->ref, NIL, 0);
+                             reason, NULL, NULL, mdp->ref, NIL, 0);
     }
     erts_monitor_release(mon);
 }
@@ -2105,9 +2128,9 @@ handle_exit_signal(Process *c_p, ErtsSigRecvTracing *tracing,
         }
 
         /* This GEN_EXIT was received from another node, decode the exit reason */
-        if (ERTS_SIG_IS_EXTERNAL(sig))
+        if (ERTS_SIG_IS_GEN_EXIT_EXTERNAL(sig))
             erts_proc_sig_decode_dist(c_p, ERTS_PROC_LOCK_MAIN, sig, 1);
-        
+
         reason = xsigd->reason;
 
         if (is_non_value(reason)) {
@@ -2115,17 +2138,14 @@ handle_exit_signal(Process *c_p, ErtsSigRecvTracing *tracing,
             ignore = !0;
             destroy = !0;
         }
-        
+
         if (!ignore) {
 
             if ((op != ERTS_SIG_Q_OP_EXIT || reason != am_kill)
                 && (c_p->flags & F_TRAP_EXIT)) {
                 convert_prepared_sig_to_msg(c_p, sig,
                                             xsigd->message, next_nm_sig);
-                ASSERT(sig->hfrag.next == NULL);
-                sig->hfrag.next = xsigd->heap_frag;
                 conv_msg = sig;
-                
             }
             else if (reason == am_normal && !xsigd->u.normal_kills) {
                 /* Ignore it... */
@@ -2991,50 +3011,29 @@ erts_proc_sig_decode_dist(Process *proc, ErtsProcLocks proc_locks,
                           ErtsMessage *msgp, int force_off_heap)
 {
     ErtsHeapFactory factory;
+    ErlHeapFragment *hfrag;
     Eterm msg;
-    ErlHeapFragment *bp;
     Sint need;
-    int decode_in_heap_frag;
-    ErtsDistExternal *dist_ext;
+    ErtsDistExternal *edep;
     ErtsExitSignalData *xsigd = NULL;
 
-    decode_in_heap_frag = (force_off_heap
-			   || !(proc_locks & ERTS_PROC_LOCK_MAIN)
-			   || (proc->flags & F_OFF_HEAP_MSGQ));
-
-    if (ERTS_SIG_IS_EXTERNAL_MSG(msgp))
-        dist_ext = msgp->data.dist_ext;
-    else {
+    edep = erts_proc_sig_get_external(msgp);
+    if (!ERTS_SIG_IS_EXTERNAL_MSG(msgp))
         xsigd = get_exit_signal_data(msgp);
-        ASSERT(ERTS_PROC_SIG_TYPE(((ErtsSignal *) msgp)->common.tag) == ERTS_SIG_Q_TYPE_GEN_EXIT);
-        ASSERT(is_non_value(xsigd->reason));
-        dist_ext = xsigd->dist_ext;
-    }
 
-    if (dist_ext->heap_size >= 0)
-	need = dist_ext->heap_size;
+    if (edep->heap_size >= 0)
+	need = edep->heap_size;
     else {
-	need = erts_decode_dist_ext_size(dist_ext);
+	need = erts_decode_dist_ext_size(edep, 1);
 	if (need < 0) {
-	    /* bad msg; remove it... */
-	    if (is_not_immed(ERL_MESSAGE_TOKEN(msgp))) {
-		bp = erts_dist_ext_trailer(dist_ext);
-		erts_cleanup_offheap(&bp->off_heap);
-	    }
-	    erts_free_dist_ext_copy(dist_ext);
-	    dist_ext = NULL;
+	    /* bad signal; remove it... */
 	    return 0;
 	}
 
-	dist_ext->heap_size = need;
+	edep->heap_size = need;
     }
 
-    if (is_not_immed(ERL_MESSAGE_TOKEN(msgp))) {
-	bp = erts_dist_ext_trailer(dist_ext);
-	need += bp->used_size;
-    }
-
-    if (xsigd) {
+    if (ERTS_SIG_IS_NON_MSG(msgp)) {
         switch (ERTS_PROC_SIG_OP(ERL_MESSAGE_TERM(msgp))) {
         case ERTS_SIG_Q_OP_EXIT:
         case ERTS_SIG_Q_OP_EXIT_LINKED:
@@ -3051,26 +3050,22 @@ erts_proc_sig_decode_dist(Process *proc, ErtsProcLocks proc_locks,
         }
     }
 
-    if (decode_in_heap_frag)
-	erts_factory_heap_frag_init(&factory, new_message_buffer(need));
-    else
-	erts_factory_proc_prealloc_init(&factory, proc, need);
-
-    ASSERT(dist_ext->heap_size >= 0);
-    if (is_not_immed(ERL_MESSAGE_TOKEN(msgp))) {
-	ErlHeapFragment *heap_frag;
-	heap_frag = erts_dist_ext_trailer(dist_ext);
-	ERL_MESSAGE_TOKEN(msgp) = copy_struct(ERL_MESSAGE_TOKEN(msgp),
-					      heap_frag->used_size,
-					      &factory.hp,
-					      factory.off_heap);
-	erts_cleanup_offheap(&heap_frag->off_heap);
+    hfrag = new_message_buffer(need);
+    erts_factory_heap_frag_init(&factory, hfrag);
+
+    ASSERT(edep->heap_size >= 0);
+
+    msg = erts_decode_dist_ext(&factory, edep, 1);
+
+    if (is_non_value(msg)) {
+        erts_factory_undo(&factory);
+        return 0;
     }
 
-    msg = erts_decode_dist_ext(&factory, dist_ext);
-    if (!xsigd) {
+    if (ERTS_SIG_IS_MSG(msgp)) {
         ERL_MESSAGE_TERM(msgp) = msg;
-        msgp->data.attached = NULL;
+        if (msgp->data.heap_frag == &msgp->hfrag)
+            msgp->data.heap_frag = ERTS_MSG_COMBINED_HFRAG;
     } else {
         switch (ERTS_PROC_SIG_OP(ERL_MESSAGE_TERM(msgp))) {
         case ERTS_SIG_Q_OP_EXIT:
@@ -3089,22 +3084,21 @@ erts_proc_sig_decode_dist(Process *proc, ErtsProcLocks proc_locks,
         }
         xsigd->reason = msg;
     }
-    erts_free_dist_ext_copy(dist_ext);
 
-    if (is_non_value(msg)) {
-	erts_factory_undo(&factory);
-	return 0;
-    }
+    erts_free_dist_ext_copy(edep);
 
     erts_factory_close(&factory);
 
-    ASSERT(!msgp->data.heap_frag || xsigd);
+    hfrag = factory.heap_frags;
+    while (hfrag->next)
+        hfrag = hfrag->next;
 
-    if (decode_in_heap_frag) {
-        if (!xsigd)
-            msgp->data.heap_frag = factory.heap_frags;
-        else
-            xsigd->heap_frag = factory.heap_frags;
+    if (ERTS_SIG_IS_MSG(msgp) && msgp->data.heap_frag != ERTS_MSG_COMBINED_HFRAG) {
+        hfrag->next = msgp->data.heap_frag;
+        msgp->data.heap_frag = factory.heap_frags;
+    } else {
+        hfrag->next = msgp->hfrag.next;
+        msgp->hfrag.next = factory.heap_frags;
     }
 
     return 1;
@@ -3274,8 +3268,9 @@ erts_proc_sig_handle_incoming(Process *c_p, erts_aint32_t *statep,
                 xsigd = get_exit_signal_data(sig);
 
                 /* This GEN_EXIT was received from another node, decode the exit reason */
-                if (ERTS_SIG_IS_EXTERNAL(sig))
-                    erts_proc_sig_decode_dist(c_p, ERTS_PROC_LOCK_MAIN, sig, 1);
+                if (ERTS_SIG_IS_GEN_EXIT_EXTERNAL(sig))
+                    if (!erts_proc_sig_decode_dist(c_p, ERTS_PROC_LOCK_MAIN, sig, 1))
+                        break; /* Decode failed, just remove signal */
 
                 omon = erts_monitor_tree_lookup(ERTS_P_MONITORS(c_p),
                                                 xsigd->u.ref);
@@ -4375,11 +4370,13 @@ handle_msg_tracing(Process *c_p, ErtsSigRecvTracing *tracing,
             return -1; /* Yield... */
         }
         if (ERTS_SIG_IS_EXTERNAL_MSG(sig)) {
-            cnt++;
+            cnt += 50; /* Decode is expensive... */
             if (!erts_proc_sig_decode_dist(c_p, ERTS_PROC_LOCK_MAIN,
                                            sig, 0)) {
                 /* Bad dist message; remove it... */
                 remove_mq_m_sig(c_p, sig, next_sig, next_nm_sig);
+                sig->next = NULL;
+                erts_cleanup_messages(sig);
                 sig = *next_sig;
                 continue;
             }
@@ -4451,18 +4448,12 @@ erts_proc_sig_prep_msgq_for_inspection(Process *c_p,
 
 	if (ERTS_SIG_IS_EXTERNAL_MSG(mp)) {
 	    /* decode it... */
-	    if (mp->data.attached)
-		erts_proc_sig_decode_dist(rp, rp_locks, mp, !0);
-
-	    msg = ERL_MESSAGE_TERM(mp);
-
-	    if (is_non_value(msg)) {
+            if (!erts_proc_sig_decode_dist(rp, rp_locks, mp, !0)) {
 		ErtsMessage *bad_mp = mp;
 		/*
 		 * Bad distribution message; remove
 		 * it from the queue...
 		 */
-		ASSERT(!mp->data.attached);
 
 		ASSERT(*mpp == bad_mp);
 
@@ -4474,6 +4465,8 @@ erts_proc_sig_prep_msgq_for_inspection(Process *c_p,
 		erts_cleanup_messages(bad_mp);
 		continue;
 	    }
+
+	    msg = ERL_MESSAGE_TERM(mp);
 	}
 
 	ASSERT(is_value(msg));
@@ -4632,12 +4625,21 @@ debug_foreach_sig_fake_oh(Eterm term,
 
 }
 
+static void
+debug_foreach_sig_external(ErtsMessage *msgp,
+                           void (*ext_func)(ErtsDistExternal *, void *),
+                           void *arg)
+{
+    ext_func(erts_proc_sig_get_external(msgp), arg);
+}
+
 void
 erts_proc_sig_debug_foreach_sig(Process *c_p,
                                 void (*msg_func)(ErtsMessage *, void *),
                                 void (*oh_func)(ErlOffHeap *, void *),
                                 ErtsMonitorFunc mon_func,
                                 ErtsLinkFunc lnk_func,
+                                void (*ext_func)(ErtsDistExternal *, void *),
                                 void *arg)
 {
     ErtsMessage *queue[] = {c_p->sig_qs.first, c_p->sig_qs.cont, c_p->sig_inq.first};
@@ -4646,10 +4648,10 @@ erts_proc_sig_debug_foreach_sig(Process *c_p,
     for (qix = 0; qix < sizeof(queue)/sizeof(queue[0]); qix++) {
         ErtsMessage *sig;
         for (sig = queue[qix]; sig; sig = sig->next) {
-            
-            if (ERTS_SIG_IS_MSG(sig))
+
+            if (ERTS_SIG_IS_MSG(sig)) {
                 msg_func(sig, arg);
-            else {
+            } else {
                 Eterm tag;
                 Uint16 type;
                 int op;
@@ -4668,7 +4670,10 @@ erts_proc_sig_debug_foreach_sig(Process *c_p,
                 case ERTS_SIG_Q_OP_MONITOR_DOWN:
                     switch (type) {
                     case ERTS_SIG_Q_TYPE_GEN_EXIT:
-                        debug_foreach_sig_heap_frags(&sig->hfrag, oh_func, arg);
+                        if (ERTS_SIG_IS_GEN_EXIT_EXTERNAL(sig))
+                            debug_foreach_sig_external(sig, ext_func, arg);
+                        else
+                            debug_foreach_sig_heap_frags(&sig->hfrag, oh_func, arg);
                         break;
                     case ERTS_LNK_TYPE_PORT:
                     case ERTS_LNK_TYPE_PROC:
diff --git a/erts/emulator/beam/erl_proc_sig_queue.h b/erts/emulator/beam/erl_proc_sig_queue.h
index 39cee6f230..2b055e73bc 100644
--- a/erts/emulator/beam/erl_proc_sig_queue.h
+++ b/erts/emulator/beam/erl_proc_sig_queue.h
@@ -228,6 +228,9 @@ erts_proc_sig_send_exit(Process *c_p, Eterm from, Eterm to,
  *
  * @param[in]     dist_ext      The exit reason in external term format
  *
+ * @param[in]     hfrag         Heap frag with trace token and dist_ext
+ *                              iff available, otherwise NULL.
+ *
  * @param[in]     reason        Exit reason.
  *
  * @param[in]     token         Seq trace token.
@@ -237,6 +240,7 @@ void
 erts_proc_sig_send_dist_exit(DistEntry *dep,
                              Eterm from, Eterm to,
                              ErtsDistExternal *dist_ext,
+                             ErlHeapFragment *hfrag,
                              Eterm reason, Eterm token);
 
 /**
@@ -322,6 +326,9 @@ erts_proc_sig_send_unlink(Process *c_p, ErtsLink *lnk);
  *
  * @param[in]     dist_ext      The exit reason in external term format
  *
+ * @param[in]     hfrag         Heap frag with trace token and dist_ext
+ *                              iff available, otherwise NULL.
+ *
  * @param[in]     reason        Exit reason.
  *
  * @param[in]     token         Seq trace token.
@@ -331,6 +338,7 @@ void
 erts_proc_sig_send_dist_link_exit(struct dist_entry_ *dep,
                                   Eterm from, Eterm to,
                                   ErtsDistExternal *dist_ext,
+                                  ErlHeapFragment *hfrag,
                                   Eterm reason, Eterm token);
 
 /**
@@ -426,6 +434,9 @@ erts_proc_sig_send_monitor(ErtsMonitor *mon, Eterm to);
  *
  * @param[in]     dist_ext      The exit reason in external term format
  *
+ * @param[in]     hfrag         Heap frag with trace token and dist_ext
+ *                              iff available, otherwise NULL.
+ *
  * @param[in]     reason        Exit reason.
  *
  */
@@ -433,6 +444,7 @@ void
 erts_proc_sig_send_dist_monitor_down(DistEntry *dep, Eterm ref,
                                      Eterm from, Eterm to,
                                      ErtsDistExternal *dist_ext,
+                                     ErlHeapFragment *hfrag,
                                      Eterm reason);
 
 /**
@@ -1035,6 +1047,7 @@ erts_proc_sig_debug_foreach_sig(Process *c_p,
                                 void (*oh_func)(ErlOffHeap *, void *),
                                 ErtsMonitorFunc mon_func,
                                 ErtsLinkFunc lnk_func,
+                                void (*ext_func)(ErtsDistExternal *, void *),
                                 void *arg);
 
 extern Process *erts_dirty_process_signal_handler;
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 66af9574b8..837b3f4ace 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -12111,6 +12111,7 @@ erts_proc_exit_handle_monitor(ErtsMonitor *mon, void *vctxt, Sint reds)
             case ERTS_DSIG_PREP_PENDING:
                 if (dist->connection_id == dsd.connection_id) {
                     code = erts_dsig_send_m_exit(&dsd,
+                                                 c_p->common.id,
                                                  watcher,
                                                  watched,
                                                  mdp->ref,
diff --git a/erts/emulator/beam/erl_process_dump.c b/erts/emulator/beam/erl_process_dump.c
index 46e460582c..a164ed543e 100644
--- a/erts/emulator/beam/erl_process_dump.c
+++ b/erts/emulator/beam/erl_process_dump.c
@@ -191,11 +191,11 @@ static ERTS_INLINE void
 dump_msg(fmtfn_t to, void *to_arg, ErtsMessage *mp)
 {
     if (ERTS_SIG_IS_MSG((ErtsSignal *) mp)) {
-        Eterm mesg = ERL_MESSAGE_TERM(mp);
-        if (is_value(mesg))
-            dump_element(to, to_arg, mesg);
+        Eterm mesg;
+        if (ERTS_SIG_IS_INTERNAL_MSG(mp))
+            dump_element(to, to_arg, ERL_MESSAGE_TERM(mp));
         else
-            dump_dist_ext(to, to_arg, mp->data.dist_ext);
+            dump_dist_ext(to, to_arg, erts_get_dist_ext(mp->data.heap_frag));
         mesg = ERL_MESSAGE_TOKEN(mp);
         erts_print(to, to_arg, ":");
         dump_element(to, to_arg, mesg);
@@ -267,6 +267,7 @@ dump_dist_ext(fmtfn_t to, void *to_arg, ErtsDistExternal *edep)
     else {
 	byte *e;
 	size_t sz;
+        int i;
 
 	if (!(edep->flags & ERTS_DIST_EXT_ATOM_TRANS_TAB))
 	    erts_print(to, to_arg, "D0:");
@@ -276,8 +277,8 @@ dump_dist_ext(fmtfn_t to, void *to_arg, ErtsDistExternal *edep)
 	    for (i = 0; i < edep->attab.size; i++)
 		dump_element(to, to_arg, edep->attab.atom[i]);
 	}
-	sz = edep->ext_endp - edep->extp;
-	e = edep->extp;
+	sz = edep->data->ext_endp - edep->data->extp;
+	e = edep->data->extp;
 	if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) {
 	    ASSERT(*e != VERSION_MAGIC);
 	    sz++;
@@ -288,15 +289,19 @@ dump_dist_ext(fmtfn_t to, void *to_arg, ErtsDistExternal *edep)
 	erts_print(to, to_arg, "E%X:", sz);
         if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) {
             byte sbuf[3];
-            int i = 0;
+
+            i = 0;
 
             sbuf[i++] = VERSION_MAGIC;
-            while (i < sizeof(sbuf) && e < edep->ext_endp) {
+            while (i < sizeof(sbuf) && e < edep->data->ext_endp) {
                 sbuf[i++] = *e++;
             }
             erts_print_base64(to, to_arg, sbuf, i);
         }
-        erts_print_base64(to, to_arg, e, edep->ext_endp - e);
+        erts_print_base64(to, to_arg, e, edep->data->ext_endp - e);
+        for (i = 1; i < edep->data->frag_id; i++)
+            erts_print_base64(to, to_arg, edep->data[i].extp,
+                              edep->data[i].ext_endp - edep->data[i].extp);
     }
 }
 
diff --git a/erts/emulator/beam/erl_ptab.h b/erts/emulator/beam/erl_ptab.h
index 94f0247492..c30a684002 100644
--- a/erts/emulator/beam/erl_ptab.h
+++ b/erts/emulator/beam/erl_ptab.h
@@ -68,8 +68,11 @@ typedef struct {
 	    Uint64 started_interval;
 	    struct reg_proc *reg;
 	    ErtsLink *links;
-	    ErtsMonitor *monitors;
+             /* Local target monitors, double linked list
+                contains the remote part of local monitors  */
             ErtsMonitor *lt_monitors;
+             /* other monitors, rb tree */
+	    ErtsMonitor *monitors;
 	} alive;
 
 	/* --- While being released --- */
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index 640ca338d9..84e6e0d6fd 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -262,19 +262,12 @@ erts_finalize_atom_cache_map(ErtsAtomCacheMap *acmp, Uint32 dflags)
     if (acmp) {
 	int long_atoms = 0; /* !0 if one or more atoms are longer than 255. */
 	int i;
-	int sz;
-	int fix_sz
-	    = 1 /* VERSION_MAGIC */
-	    + 1 /* DIST_HEADER */
-	    + 1 /* dist header flags */
-	    + 1 /* number of internal cache entries */
-	    ;
+	int sz = 0;
 	int min_sz;
         ASSERT(dflags & DFLAG_UTF8_ATOMS);
 	ASSERT(acmp->hdr_sz < 0);
 	/* Make sure cache update instructions fit */
-	min_sz = fix_sz+(2+4)*acmp->sz;
-	sz = fix_sz;
+	min_sz = (2+4)*acmp->sz;
 	for (i = 0; i < acmp->sz; i++) {
 	    Atom *a;
 	    Eterm atom;
@@ -302,17 +295,28 @@ erts_finalize_atom_cache_map(ErtsAtomCacheMap *acmp, Uint32 dflags)
 }
 
 Uint
-erts_encode_ext_dist_header_size(ErtsAtomCacheMap *acmp)
+erts_encode_ext_dist_header_size(ErtsAtomCacheMap *acmp, Uint fragments)
 {
     if (!acmp)
 	return 0;
     else {
+        int fix_sz
+            = 1 /* VERSION_MAGIC */
+            + 1 /* DIST_HEADER */
+            + 1 /* dist header flags */
+            + 1 /* number of internal cache entries */
+            ;
 	ASSERT(acmp->hdr_sz >= 0);
-	return acmp->hdr_sz;
+        if (fragments > 1)
+            fix_sz += 8 /* sequence id */
+                + 8 /* number of fragments */
+                ;
+        return fix_sz + acmp->hdr_sz;
     }
 }
 
-byte *erts_encode_ext_dist_header_setup(byte *ctl_ext, ErtsAtomCacheMap *acmp)
+byte *erts_encode_ext_dist_header_setup(byte *ctl_ext, ErtsAtomCacheMap *acmp,
+                                        Uint fragments, Eterm from)
 {
     /* Maximum number of atom must be less than the maximum of a 32 bits
        unsigned integer. Check is done in erl_init.c, erl_start function. */
@@ -346,12 +350,37 @@ byte *erts_encode_ext_dist_header_setup(byte *ctl_ext, ErtsAtomCacheMap *acmp)
 	put_int8(acmp->sz, ep);
 	--ep;
 	put_int8(dist_hdr_flags, ep);
-	*--ep = DIST_HEADER;
-	*--ep = VERSION_MAGIC;
+        if (fragments > 1) {
+            ASSERT(is_pid(from));
+            ep -= 8;
+            put_int64(fragments, ep);
+            ep -= 8;
+            put_int64(from, ep);
+            *--ep = DIST_FRAG_HEADER;
+        } else {
+            *--ep = DIST_HEADER;
+        }
+        *--ep = VERSION_MAGIC;
 	return ep;
     }
 }
 
+byte *erts_encode_ext_dist_header_fragment(byte **hdrpp,
+                                           Uint fragment,
+                                           Eterm from)
+{
+    byte *ep = *hdrpp, *start = ep;
+    ASSERT(is_pid(from));
+    *ep++ = VERSION_MAGIC;
+    *ep++ = DIST_FRAG_CONT;
+    put_int64(from, ep);
+    ep += 8;
+    put_int64(fragment, ep);
+    ep += 8;
+    *hdrpp = ep;
+    return start;
+}
+
 
 #define PASS_THROUGH 'p'
 
@@ -365,7 +394,8 @@ Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf* ob,
     int ci, sz;
     byte dist_hdr_flags;
     int long_atoms;
-    register byte *ep = ob->extp;
+    Uint64 seq_id = 0, frag_id = 0;
+    register byte *ep = ob->hdrp ? ob->hdrp : ob->extp;
     ASSERT(dflags & DFLAG_UTF8_ATOMS);
 
     /*
@@ -416,7 +446,7 @@ Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf* ob,
         }
         goto done;
     }
-    else if (ep[1] != DIST_HEADER) {
+    else if (ep[1] != DIST_HEADER && ep[1] != DIST_FRAG_HEADER && ep[1] != DIST_FRAG_CONT) {
         ASSERT(ep[1] == SMALL_TUPLE_EXT || ep[1] == LARGE_TUPLE_EXT);
         ASSERT(!(dflags & DFLAG_DIST_HDR_ATOM_CACHE));
         /* Node without atom cache, 'pass through' needed */
@@ -424,6 +454,17 @@ Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf* ob,
         goto done;
     }
 
+    if (ep[1] == DIST_FRAG_CONT) {
+        ep = ob->extp;
+        goto done;
+    } else if (ep[1] == DIST_FRAG_HEADER) {
+        /* skip the seq id and frag id */
+        seq_id = get_int64(&ep[2]);
+        ep += 8;
+        frag_id = get_int64(&ep[2]);
+        ep += 8;
+    }
+
     dist_hdr_flags = ep[2];
     long_atoms = ERTS_DIST_HDR_LONG_ATOMS_FLG & ((int) dist_hdr_flags);
 
@@ -546,11 +587,19 @@ Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf* ob,
     }
     --ep;
     put_int8(ci, ep);
-    *--ep = DIST_HEADER;
+    if (seq_id) {
+        ep -= 8;
+        put_int64(frag_id, ep);
+        ep -= 8;
+        put_int64(seq_id, ep);
+        *--ep = DIST_FRAG_HEADER;
+    } else {
+        *--ep = DIST_HEADER;
+    }
     *--ep = VERSION_MAGIC;
 done:
     ob->extp = ep;
-    ASSERT(&ob->data[0] <= ob->extp && ob->extp < ob->ext_endp);
+    ASSERT((byte*)ob->bin->orig_bytes <= ob->extp && ob->extp < ob->ext_endp);
     return reds < 0 ? 0 : reds;
 }
 
@@ -635,72 +684,90 @@ byte* erts_encode_ext_ets(Eterm term, byte *ep, struct erl_off_heap_header** off
 		    off_heap);
 }
 
-ErtsDistExternal *
-erts_make_dist_ext_copy(ErtsDistExternal *edep, Eterm *token)
+
+static Uint
+dist_ext_size(ErtsDistExternal *edep)
 {
-    size_t align_sz;
-    size_t dist_ext_sz;
-    size_t ext_sz = 0;
-    size_t token_sz = 0;
-    Eterm token_size;
-    byte *ep;
-    ErtsDistExternal *new_edep;
+    Uint sz = sizeof(ErtsDistExternal);
 
-    dist_ext_sz = ERTS_DIST_EXT_SIZE(edep);
-    ASSERT(edep->ext_endp && edep->extp);
-    ASSERT(edep->ext_endp >= edep->extp);
-    if (edep->binp == NULL)
-        ext_sz = edep->ext_endp - edep->extp;
+    ASSERT(edep->data->ext_endp && edep->data->extp);
+    ASSERT(edep->data->ext_endp >= edep->data->extp);
 
-    align_sz = ERTS_EXTRA_DATA_ALIGN_SZ(dist_ext_sz + ext_sz);
+    if (edep->flags & ERTS_DIST_EXT_ATOM_TRANS_TAB) {
+        ASSERT(0 <= edep->attab.size \
+               && edep->attab.size <= ERTS_ATOM_CACHE_SIZE);
+        sz -= sizeof(Eterm)*(ERTS_ATOM_CACHE_SIZE - edep->attab.size);
+    } else {
+        sz -= sizeof(ErtsAtomTranslationTable);
+    }
+    return sz;
+}
 
-    token_size = size_object(*token);
-    if (token_size)
-        token_sz = ERTS_HEAP_FRAG_SIZE(token_size);
+Uint
+erts_dist_ext_size(ErtsDistExternal *edep)
+{
+    Uint sz = dist_ext_size(edep);
+    sz += edep->data[0].frag_id * sizeof(ErtsDistExternalData);
+    return sz + ERTS_EXTRA_DATA_ALIGN_SZ(sz);
+}
 
-    new_edep = erts_alloc(ERTS_ALC_T_EXT_TERM_DATA,
-			  dist_ext_sz + ext_sz + align_sz + token_sz);
+Uint
+erts_dist_ext_data_size(ErtsDistExternal *edep)
+{
+    Uint sz = 0, i;
+    for (i = 0; i < edep->data->frag_id; i++)
+        sz += edep->data[i].ext_endp - edep->data[i].extp;
+    return sz;
+}
 
-    ep = (byte *) new_edep;
-    sys_memcpy((void *) ep, (void *) edep, dist_ext_sz);
-    if (new_edep->dep)
-        erts_ref_dist_entry(new_edep->dep);
-    new_edep->heap_size = -1;
-
-    if (ext_sz) {
-        ep += dist_ext_sz;
-        new_edep->extp = ep;
-        new_edep->ext_endp = ep + ext_sz;
-        sys_memcpy((void *) ep, (void *) edep->extp, ext_sz);
+void
+erts_dist_ext_frag(ErtsDistExternalData *ede_datap, ErtsDistExternal *edep)
+{
+    ErtsDistExternalData *new_ede_datap = &edep->data[edep->data->frag_id - ede_datap->frag_id];
+    sys_memcpy(new_ede_datap, ede_datap, sizeof(ErtsDistExternalData));
+
+    /* If the data is not backed by a binary, we create one here to keep
+       things simple. Only custom distribution drivers should use lists. */
+    if (new_ede_datap->binp == NULL) {
+        size_t ext_sz = ede_datap->ext_endp - ede_datap->extp;
+        new_ede_datap->binp = erts_bin_nrml_alloc(ext_sz);
+        sys_memcpy(new_ede_datap->binp->orig_bytes, (void *) ede_datap->extp, ext_sz);
+        new_ede_datap->extp = (byte*)new_ede_datap->binp->orig_bytes;
+        new_ede_datap->ext_endp = (byte*)new_ede_datap->binp->orig_bytes + ext_sz;
     } else {
-        erts_refc_inc(&new_edep->binp->intern.refc, 2);
+        erts_refc_inc(&new_ede_datap->binp->intern.refc, 2);
     }
+}
 
-    /* Copy the seq_trace token */
-    if (is_not_nil(*token)) {
-        ErlHeapFragment *heap_frag;
-        ErlOffHeap *ohp;
-        Eterm *hp;
-        heap_frag = erts_dist_ext_trailer(new_edep);
-        ERTS_INIT_HEAP_FRAG(heap_frag, token_size, token_size);
-        hp = heap_frag->mem;
-        ohp = &heap_frag->off_heap;
-        *token = copy_struct(*token, token_size, &hp, ohp);
-    }
-    return new_edep;
+void
+erts_make_dist_ext_copy(ErtsDistExternal *edep, ErtsDistExternal *new_edep)
+{
+    size_t dist_ext_sz = dist_ext_size(edep);
+    byte *ep;
+
+    ep = (byte *) new_edep;
+    sys_memcpy((void *) ep, (void *) edep, dist_ext_sz);
+    erts_ref_dist_entry(new_edep->dep);
+
+    ep += dist_ext_sz;
+
+    new_edep->data = (ErtsDistExternalData*)ep;
+    sys_memzero(new_edep->data, sizeof(ErtsDistExternalData) * edep->data->frag_id);
+    new_edep->data->frag_id = edep->data->frag_id;
+    erts_dist_ext_frag(edep->data, new_edep);
 }
 
 void
 erts_free_dist_ext_copy(ErtsDistExternal *edep)
 {
-    if (edep->dep)
-	erts_deref_dist_entry(edep->dep);
-    if (edep->binp)
-        erts_bin_release(edep->binp);
-    erts_free(ERTS_ALC_T_EXT_TERM_DATA, edep);
+    int i;
+    erts_deref_dist_entry(edep->dep);
+    for (i = 0; i < edep->data->frag_id; i++)
+        if (edep->data[i].binp)
+            erts_bin_release(edep->data[i].binp);
 }
 
-int
+ErtsPrepDistExtRes
 erts_prepare_dist_ext(ErtsDistExternal *edep,
 		      byte *ext,
 		      Uint size,
@@ -711,10 +778,6 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 {
     register byte *ep;
 
-    edep->heap_size = -1;
-    edep->flags = 0;
-    edep->dep = dep;
-
     ASSERT(dep);
     erts_de_rlock(dep);
 
@@ -734,10 +797,6 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
         size--;
     }
 
-    edep->heap_size = -1;
-    edep->ext_endp = ext + size;
-    edep->binp = binp;
-
     ep = ext;
 
     if (size < 2)
@@ -753,16 +812,33 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 	goto fail;
     }
 
+    edep->heap_size = -1;
+    edep->flags = 0;
+    edep->dep = dep;
+    edep->connection_id = conn_id;
+    edep->data->ext_endp = ext+size;
+    edep->data->binp = binp;
+    edep->data->seq_id = 0;
+    edep->data->frag_id = 1;
+
     if (dep->flags & DFLAG_DIST_HDR_ATOM_CACHE)
         edep->flags |= ERTS_DIST_EXT_DFLAG_HDR;
 
-    edep->connection_id = dep->connection_id;
-
-    if (ep[1] != DIST_HEADER) {
+    if (ep[1] != DIST_HEADER && ep[1] != DIST_FRAG_HEADER && ep[1] != DIST_FRAG_CONT) {
 	if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR)
 	    goto bad_hdr;
 	edep->attab.size = 0;
-	edep->extp = ext;
+	edep->data->extp = ext;
+    }
+    else if (ep[1] == DIST_FRAG_CONT) {
+        if (!(dep->flags & DFLAG_FRAGMENTS))
+            goto bad_hdr;
+        edep->attab.size = 0;
+	edep->data->extp = ext + 1 + 1 + 8 + 8;
+        edep->data->seq_id = get_int64(&ep[2]);
+        edep->data->frag_id = get_int64(&ep[2+8]);
+        erts_de_runlock(dep);
+        return ERTS_PREP_DIST_EXT_FRAG_CONT;
     }
     else {
 	int tix;
@@ -771,9 +847,17 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 	if (!(edep->flags & ERTS_DIST_EXT_DFLAG_HDR))
 	    goto bad_hdr;
 
+        if (ep[1] == DIST_FRAG_HEADER) {
+            if (!(dep->flags & DFLAG_FRAGMENTS))
+                goto bad_hdr;
+            edep->data->seq_id = get_int64(&ep[2]);
+            edep->data->frag_id = get_int64(&ep[2+8]);
+            ep += 16;
+        }
+
 #undef CHKSIZE
 #define CHKSIZE(SZ) \
-	do { if ((SZ) > edep->ext_endp - ep) goto bad_hdr; } while(0)
+	do { if ((SZ) > edep->data->ext_endp - ep) goto bad_hdr; } while(0)
 
 	CHKSIZE(1+1+1);
 	ep += 2;
@@ -903,7 +987,7 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 #endif
 	    }
 	}
-	edep->extp = ep;
+	edep->data->extp = ep;
 #ifdef ERTS_DEBUG_USE_DIST_SEP
 	if (*ep != VERSION_MAGIC)
 	    goto bad_hdr;
@@ -928,7 +1012,7 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 		      erts_this_node->sysname,
 		      edep->dep->sysname,
 		      dist_entry_channel_no(edep->dep));
-	for (ep = ext; ep < edep->ext_endp; ep++)
+	for (ep = ext; ep < edep->data->ext_endp; ep++)
 	    erts_dsprintf(dsbufp, ep != ext ? ",%b8u" : "<<%b8u", *ep);
 	erts_dsprintf(dsbufp, ">>");
 	erts_send_warning_to_logger_nogl(dsbufp);
@@ -953,9 +1037,9 @@ bad_dist_ext(ErtsDistExternal *edep)
 		      erts_this_node->sysname,
 		      dep->sysname,
 		      dist_entry_channel_no(dep));
-	for (ep = edep->extp; ep < edep->ext_endp; ep++)
+	for (ep = edep->data->extp; ep < edep->data->ext_endp; ep++)
 	    erts_dsprintf(dsbufp,
-			  ep != edep->extp ? ",%b8u" : "<<...,%b8u",
+			  ep != edep->data->extp ? ",%b8u" : "<<...,%b8u",
 			  *ep);
 	erts_dsprintf(dsbufp, ">>\n");
 	erts_dsprintf(dsbufp, "ATOM_CACHE_REF translations: ");
@@ -973,30 +1057,32 @@ bad_dist_ext(ErtsDistExternal *edep)
 }
 
 Sint
-erts_decode_dist_ext_size(ErtsDistExternal *edep)
+erts_decode_dist_ext_size(ErtsDistExternal *edep, int kill_connection)
 {
     Sint res;
     byte *ep;
-    if (edep->extp >= edep->ext_endp)
+
+    if (edep->data->extp >= edep->data->ext_endp)
 	goto fail;
 #ifndef ERTS_DEBUG_USE_DIST_SEP
     if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) {
-	if (*edep->extp == VERSION_MAGIC)
+	if (*edep->data->extp == VERSION_MAGIC)
 	    goto fail;
-	ep = edep->extp;
+	ep = edep->data->extp;
     }
     else
 #endif
     {
-	if (*edep->extp != VERSION_MAGIC)
+	if (*edep->data->extp != VERSION_MAGIC)
 	    goto fail;
-	ep = edep->extp+1;
+	ep = edep->data->extp+1;
     }
-    res = decoded_size(ep, edep->ext_endp, 0, NULL);
+    res = decoded_size(ep, edep->data->ext_endp, 0, NULL);
     if (res >= 0)
 	return res;
  fail:
-    bad_dist_ext(edep);
+    if (kill_connection)
+        bad_dist_ext(edep);
     return -1;
 }
 
@@ -1022,12 +1108,15 @@ Sint erts_decode_ext_size_ets(byte *ext, Uint size)
 */
 Eterm
 erts_decode_dist_ext(ErtsHeapFactory* factory,
-		     ErtsDistExternal *edep)
+		     ErtsDistExternal *edep,
+                     int kill_connection)
 {
     Eterm obj;
-    byte* ep = edep->extp;
+    byte* ep;
+
+    ep = edep->data->extp;
 
-    if (ep >= edep->ext_endp)
+    if (ep >= edep->data->ext_endp)
 	goto error;
 #ifndef ERTS_DEBUG_USE_DIST_SEP
     if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) {
@@ -1045,14 +1134,15 @@ erts_decode_dist_ext(ErtsHeapFactory* factory,
     if (!ep)
 	goto error;
 
-    edep->extp = ep;
+    edep->data->extp = ep;
 
     return obj;
 
  error:
     erts_factory_undo(factory);
 
-    bad_dist_ext(edep);
+    if (kill_connection)
+        bad_dist_ext(edep);
 
     return THE_NON_VALUE;
 }
@@ -1097,6 +1187,7 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2)
     Eterm res;
     Sint hsz;
     ErtsDistExternal ede;
+    ErtsDistExternalData ede_data;
     Eterm *tp;
     Eterm real_bin;
     Uint offset;
@@ -1109,7 +1200,8 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2)
     ede.flags = ERTS_DIST_EXT_ATOM_TRANS_TAB;
     ede.dep = NULL;
     ede.heap_size = -1;
-    
+    ede.data = &ede_data;
+
     if (is_not_tuple(BIF_ARG_1))
 	goto badarg;
     tp = tuple_val(BIF_ARG_1);
@@ -1134,15 +1226,15 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2)
     if (bitsize != 0)
 	goto badarg;
 
-    ede.extp = binary_bytes(real_bin)+offset;
-    ede.ext_endp = ede.extp + size;
+    ede.data->extp = binary_bytes(real_bin)+offset;
+    ede.data->ext_endp = ede.data->extp + size;
 
-    hsz = erts_decode_dist_ext_size(&ede);
+    hsz = erts_decode_dist_ext_size(&ede, 1);
     if (hsz < 0)
 	goto badarg;
 
     erts_factory_proc_prealloc_init(&factory, BIF_P, hsz);
-    res = erts_decode_dist_ext(&factory, &ede);
+    res = erts_decode_dist_ext(&factory, &ede, 1);
     erts_factory_close(&factory);
 
     if (is_value(res))
diff --git a/erts/emulator/beam/external.h b/erts/emulator/beam/external.h
index 91c028c9bd..396cd9f802 100644
--- a/erts/emulator/beam/external.h
+++ b/erts/emulator/beam/external.h
@@ -58,6 +58,8 @@
 #define SMALL_ATOM_UTF8_EXT 'w'
 
 #define DIST_HEADER       'D'
+#define DIST_FRAG_HEADER  'E'
+#define DIST_FRAG_CONT    'F'
 #define ATOM_CACHE_REF    'R'
 #define ATOM_INTERNAL_REF2 'I'
 #define ATOM_INTERNAL_REF3 'K'
@@ -123,15 +125,22 @@ typedef struct {
 #define ERTS_DIST_CON_ID_MASK ((Uint32) 0x00ffffff) /* also in net_kernel.erl */
 
 struct binary;
+typedef struct erl_dist_external_data ErtsDistExternalData;
 
-typedef struct erl_dist_external {
-    DistEntry *dep;
+struct erl_dist_external_data {
+    Uint64 seq_id;
+    Uint64 frag_id;
     byte *extp;
     byte *ext_endp;
     struct binary *binp;
+};
+
+typedef struct erl_dist_external {
     Sint heap_size;
-    Uint32 connection_id;
+    DistEntry *dep;
     Uint32 flags;
+    Uint32 connection_id;
+    ErtsDistExternalData *data;
     ErtsAtomTranslationTable attab;
 } ErtsDistExternal;
 
@@ -158,8 +167,9 @@ void erts_reset_atom_cache_map(ErtsAtomCacheMap *);
 void erts_destroy_atom_cache_map(ErtsAtomCacheMap *);
 void erts_finalize_atom_cache_map(ErtsAtomCacheMap *, Uint32);
 
-Uint erts_encode_ext_dist_header_size(ErtsAtomCacheMap *);
-byte *erts_encode_ext_dist_header_setup(byte *, ErtsAtomCacheMap *);
+Uint erts_encode_ext_dist_header_size(ErtsAtomCacheMap *, Uint);
+byte *erts_encode_ext_dist_header_setup(byte *, ErtsAtomCacheMap *, Uint, Eterm);
+byte *erts_encode_ext_dist_header_fragment(byte **, Uint, Eterm);
 Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf*, DistEntry *, Uint32 dflags, Sint reds);
 struct erts_dsig_send_context;
 int erts_encode_dist_ext_size(Eterm, Uint32, ErtsAtomCacheMap*, Uint* szp);
@@ -174,20 +184,24 @@ Uint erts_encode_ext_size_ets(Eterm);
 void erts_encode_ext(Eterm, byte **);
 byte* erts_encode_ext_ets(Eterm, byte *, struct erl_off_heap_header** ext_off_heap);
 
+Uint erts_dist_ext_size(ErtsDistExternal *);
+Uint erts_dist_ext_data_size(ErtsDistExternal *);
 void erts_free_dist_ext_copy(ErtsDistExternal *);
-ERTS_GLB_INLINE void *erts_dist_ext_trailer(ErtsDistExternal *);
-ErtsDistExternal *erts_make_dist_ext_copy(ErtsDistExternal *, Eterm *);
-void *erts_dist_ext_trailer(ErtsDistExternal *);
-void erts_destroy_dist_ext_copy(ErtsDistExternal *);
-
-#define ERTS_PREP_DIST_EXT_FAILED       (-1)
-#define ERTS_PREP_DIST_EXT_SUCCESS      (0)
-#define ERTS_PREP_DIST_EXT_CLOSED       (1)
-
-int erts_prepare_dist_ext(ErtsDistExternal *, byte *, Uint, struct binary *,
-			  DistEntry *, Uint32, ErtsAtomCache *);
-Sint erts_decode_dist_ext_size(ErtsDistExternal *);
-Eterm erts_decode_dist_ext(ErtsHeapFactory* factory, ErtsDistExternal *);
+void erts_make_dist_ext_copy(ErtsDistExternal *, ErtsDistExternal *);
+void erts_dist_ext_frag(ErtsDistExternalData *, ErtsDistExternal *);
+#define erts_get_dist_ext(HFRAG) ((ErtsDistExternal*)((HFRAG)->mem + (HFRAG)->used_size))
+
+typedef enum {
+    ERTS_PREP_DIST_EXT_FAILED,
+    ERTS_PREP_DIST_EXT_SUCCESS,
+    ERTS_PREP_DIST_EXT_FRAG_CONT,
+    ERTS_PREP_DIST_EXT_CLOSED
+} ErtsPrepDistExtRes;
+
+ErtsPrepDistExtRes erts_prepare_dist_ext(ErtsDistExternal *, byte *, Uint, struct binary *,
+                                         DistEntry *, Uint32, ErtsAtomCache *);
+Sint erts_decode_dist_ext_size(ErtsDistExternal *, int);
+Eterm erts_decode_dist_ext(ErtsHeapFactory*, ErtsDistExternal *, int);
 
 Sint erts_decode_ext_size(byte*, Uint);
 Sint erts_decode_ext_size_ets(byte*, Uint);
@@ -203,17 +217,4 @@ int erts_debug_max_atom_out_cache_index(void);
 int erts_debug_atom_to_out_cache_index(Eterm);
 void transcode_free_ctx(DistEntry* dep);
 
-#if ERTS_GLB_INLINE_INCL_FUNC_DEF
-
-ERTS_GLB_INLINE void *
-erts_dist_ext_trailer(ErtsDistExternal *edep)
-{
-    void *res = (void *) (edep->ext_endp
-			  + ERTS_EXTRA_DATA_ALIGN_SZ(edep->ext_endp));
-    ASSERT((((UWord) res) % sizeof(Uint)) == 0);
-    return res;
-}
-
-#endif
-
 #endif /* ERL_EXTERNAL_H__ */
diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c
index d413659f81..8e8198b0b0 100644
--- a/erts/emulator/sys/common/erl_check_io.c
+++ b/erts/emulator/sys/common/erl_check_io.c
@@ -44,7 +44,7 @@
 #include "erl_time.h"
 
 #if 0
-#define DEBUG_PRINT(FMT, ...) erts_printf(FMT "\r\n", ##__VA_ARGS__)
+#define DEBUG_PRINT(FMT, ...) do { erts_printf(FMT "\r\n", ##__VA_ARGS__); fflush(stdout); } while(0)
 #define DEBUG_PRINT_FD(FMT, STATE, ...)                                 \
     DEBUG_PRINT("%d: " FMT " (ev=%s, ac=%s, flg=%s)",                   \
                 (STATE) ? (STATE)->fd : (ErtsSysFdType)-1, ##__VA_ARGS__, \
diff --git a/erts/emulator/test/distribution_SUITE.erl b/erts/emulator/test/distribution_SUITE.erl
index 885c66331c..5e268c748f 100644
--- a/erts/emulator/test/distribution_SUITE.erl
+++ b/erts/emulator/test/distribution_SUITE.erl
@@ -62,7 +62,8 @@
          bad_dist_ext_control/1,
          bad_dist_ext_connection_id/1,
          bad_dist_ext_size/1,
-	 start_epmd_false/1, epmd_module/1]).
+	 start_epmd_false/1, epmd_module/1,
+         bad_dist_fragments/1]).
 
 %% Internal exports.
 -export([sender/3, receiver2/2, dummy_waiter/0, dead_process/0,
@@ -90,7 +91,7 @@ all() ->
      dist_parallel_send, atom_roundtrip, unicode_atom_roundtrip,
      atom_roundtrip_r16b,
      contended_atom_cache_entry, contended_unicode_atom_cache_entry,
-     bad_dist_structure, {group, bad_dist_ext},
+     {group, bad_dist}, {group, bad_dist_ext},
      start_epmd_false, epmd_module].
 
 groups() ->
@@ -100,6 +101,8 @@ groups() ->
      {trap_bif, [], [trap_bif_1, trap_bif_2, trap_bif_3]},
      {dist_auto_connect, [],
       [dist_auto_connect_never, dist_auto_connect_once]},
+     {bad_dist, [],
+      [bad_dist_structure, bad_dist_fragments]},
      {bad_dist_ext, [],
       [bad_dist_ext_receive, bad_dist_ext_process_info,
        bad_dist_ext_size,
@@ -1382,6 +1385,15 @@ get_conflicting_unicode_atoms(CIX, N) ->
 -define(DOP_DEMONITOR_P,	20).
 -define(DOP_MONITOR_P_EXIT,	21).
 
+-define(DOP_SEND_SENDER, 22).
+-define(DOP_SEND_SENDER_TT, 23).
+
+-define(DOP_PAYLOAD_EXIT, 24).
+-define(DOP_PAYLOAD_EXIT_TT, 25).
+-define(DOP_PAYLOAD_EXIT2, 26).
+-define(DOP_PAYLOAD_EXIT2_TT, 27).
+-define(DOP_PAYLOAD_MONITOR_P_EXIT, 28).
+
 start_monitor(Offender,P) ->
     Parent = self(),
     Q = spawn(Offender,
@@ -1515,7 +1527,145 @@ bad_dist_structure(Config) when is_list(Config) ->
     stop_node(Victim),
     ok.
 
+%% Test various dist fragmentation errors
+bad_dist_fragments(Config) when is_list(Config) ->
+    ct:timetrap({seconds, 15}),
+
+    {ok, Offender} = start_node(bad_dist_fragment_offender),
+    {ok, Victim} = start_node(bad_dist_fragment_victim),
 
+    Msg = iolist_to_binary(dmsg_ext(lists:duplicate(255,255))),
+
+    start_node_monitors([Offender,Victim]),
+    Parent = self(),
+    P = spawn(Victim,
+              fun () ->
+                      process_flag(trap_exit,true),
+                      Parent ! {self(), started},
+                      receive check_msgs -> ok end,
+                      bad_dist_struct_check_msgs([one,
+                                                  two]),
+                      Parent ! {self(), messages_checked},
+                      receive done -> ok end
+              end),
+    receive {P, started} -> ok end,
+    pong = rpc:call(Victim, net_adm, ping, [Offender]),
+    verify_up(Offender, Victim),
+    true = lists:member(Offender, rpc:call(Victim, erlang, nodes, [])),
+    start_monitor(Offender,P),
+    P ! one,
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{frg, 1, binary:part(Msg, 10,byte_size(Msg)-10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{hdr, 3, binary:part(Msg, 0,10)},
+                       {frg, 1, binary:part(Msg, 10,byte_size(Msg)-10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{hdr, 3, binary:part(Msg, 0,10)},
+                       {hdr, 3, binary:part(Msg, 0,10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P,broken},3,
+                      [{hdr, 1, binary:part(Msg, 10,byte_size(Msg)-10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{hdr, 3, binary:part(Msg, 10,byte_size(Msg)-10)},
+                       close]),
+
+    start_monitor(Offender,P),
+    ExitVictim = spawn(Victim, fun() -> receive ok -> ok end end),
+    send_bad_fragments(Offender, Victim, P,{?DOP_PAYLOAD_EXIT,P,ExitVictim},2,
+                      [{hdr, 1, [131]}]),
+
+    start_monitor(Offender,P),
+    Exit2Victim = spawn(Victim, fun() -> receive ok -> ok end end),
+    send_bad_fragments(Offender, Victim, P,{?DOP_PAYLOAD_EXIT2,P,ExitVictim},2,
+                      [{hdr, 1, [132]}]),
+
+    start_monitor(Offender,P),
+    DownVictim = spawn(Victim, fun() -> receive ok -> ok end end),
+    DownRef = erlang:monitor(process, DownVictim),
+    send_bad_fragments(Offender, Victim, P,{?DOP_PAYLOAD_MONITOR_P_EXIT,P,DownVictim,DownRef},2,
+                      [{hdr, 1, [133]}]),
+
+    P ! two,
+    P ! check_msgs,
+    receive
+        {P, messages_checked} -> ok
+    after 5000 ->
+              exit(victim_is_dead)
+    end,
+
+    {message_queue_len, 0}
+    = rpc:call(Victim, erlang, process_info, [P, message_queue_len]),
+
+    unlink(P),
+    P ! done,
+    stop_node(Offender),
+    stop_node(Victim),
+    ok.
+
+dmsg_frag_hdr(Frag) ->
+    dmsg_frag_hdr(erlang:phash2(self()), Frag).
+dmsg_frag_hdr(Seq, Frag) ->
+    [131, $E, uint64_be(Seq), uint64_be(Frag), 0].
+
+dmsg_frag(Frag) ->
+    dmsg_frag(erlang:phash2(self()), Frag).
+dmsg_frag(Seq, Frag) ->
+    [131, $F, uint64_be(Seq), uint64_be(Frag)].
+
+send_bad_fragments(Offender,VictimNode,Victim,Ctrl,WhereToPutSelf,Fragments) ->
+    Parent = self(),
+    Done = make_ref(),
+    ct:pal("Send: ~p",[Fragments]),
+    spawn_link(Offender,
+          fun () ->
+                  Node = node(Victim),
+                  pong = net_adm:ping(Node),
+                  erlang:monitor_node(Node, true),
+                  DCtrl = dctrl(Node),
+                  Ctrl1 = case WhereToPutSelf of
+                             0 ->
+                                 Ctrl;
+                             N when N > 0 ->
+                                 setelement(N,Ctrl,self())
+                         end,
+
+                  FragData = [case Type of
+                                  hdr ->
+                                      [dmsg_frag_hdr(FragId),
+                                       dmsg_ext(Ctrl1), FragPayload];
+                                  frg ->
+                                      [dmsg_frag(FragId), FragPayload]
+                              end || {Type, FragId, FragPayload} <- Fragments],
+
+                  receive {nodedown, Node} -> exit("premature nodedown")
+                  after 10 -> ok
+                  end,
+
+                  [ dctrl_send(DCtrl, D) || D <- FragData ],
+                  [ erlang:port_close(DCtrl) || close <- Fragments],
+
+                  receive {nodedown, Node} -> ok
+                  after 5000 -> exit("missing nodedown")
+                  end,
+                  Parent ! {FragData,Done}
+          end),
+    receive
+        {WhatSent,Done} ->
+            io:format("Offender sent ~p~n",[WhatSent]),
+            verify_nc(VictimNode),
+            ok
+    after 7000 ->
+              exit(unable_to_send)
+    end.
 
 bad_dist_ext_receive(Config) when is_list(Config) ->
     {ok, Offender} = start_node(bad_dist_ext_receive_offender),
@@ -2124,8 +2274,25 @@ start_node(Config, Args, Rel) when is_list(Config), is_list(Rel) ->
     start_node(Name, Args, Rel).
 
 stop_node(Node) ->
+    verify_nc(Node),
     test_server:stop_node(Node).
 
+verify_nc(Node) ->
+    P = self(),
+    Ref = make_ref(),
+    spawn(Node,
+          fun() ->
+                  R = erts_test_utils:check_node_dist(fun(E) -> E end),
+                  P ! {Ref, R}
+          end),
+    receive
+        {Ref, ok} ->
+            ok;
+        {Ref, Error} ->
+            ct:log("~s",[Error]),
+            ct:fail(failed_nc_refc_check)
+    end.
+
 freeze_node(Node, MS) ->
     Own = 300,
     DoingIt = make_ref(),
@@ -2485,6 +2652,17 @@ mk_ref({NodeNameExt, Creation}, Numbers) when is_integer(Creation),
             exit({unexpected_binary_to_term_result, Other})
     end.
 
+uint64_be(Uint) when is_integer(Uint), 0 =< Uint, Uint < 1 bsl 64 ->
+    [(Uint bsr 56) band 16#ff,
+     (Uint bsr 48) band 16#ff,
+     (Uint bsr 40) band 16#ff,
+     (Uint bsr 32) band 16#ff,
+     (Uint bsr 24) band 16#ff,
+     (Uint bsr 16) band 16#ff,
+     (Uint bsr 8) band 16#ff,
+     Uint band 16#ff];
+uint64_be(Uint) ->
+    exit({badarg, uint64_be, [Uint]}).
 
 uint32_be(Uint) when is_integer(Uint), 0 =< Uint, Uint < 1 bsl 32 ->
     [(Uint bsr 24) band 16#ff,
diff --git a/erts/emulator/test/erts_test_utils.erl b/erts/emulator/test/erts_test_utils.erl
index ac2f2435be..0c3ef3e0fc 100644
--- a/erts/emulator/test/erts_test_utils.erl
+++ b/erts/emulator/test/erts_test_utils.erl
@@ -27,6 +27,7 @@
 -export([mk_ext_pid/3,
          mk_ext_port/2,
          mk_ext_ref/2,
+         available_internal_state/1,
          check_node_dist/0, check_node_dist/1, check_node_dist/3]).
 
 
@@ -157,6 +158,21 @@ mk_ext_ref({NodeName, Creation}, Numbers) when is_list(NodeName),
     end.
 
 
+available_internal_state(Bool) when Bool == true; Bool == false ->
+    case {Bool,
+          (catch erts_debug:get_internal_state(available_internal_state))} of
+        {true, true} ->
+            true;
+        {false, true} ->
+            erts_debug:set_internal_state(available_internal_state, false),
+            true;
+        {true, _} ->
+            erts_debug:set_internal_state(available_internal_state, true),
+            false;
+        {false, _} ->
+            false
+    end.
+
 
 %%
 %% Check reference counters for node- and dist entries.
@@ -168,16 +184,21 @@ check_node_dist() ->
                     end).
 
 check_node_dist(Fail) ->
+    AIS = available_internal_state(true),
+    [erlang:garbage_collect(P) || P <- erlang:processes()],
     {{node_references, NodeRefs},
      {dist_references, DistRefs}} =
         erts_debug:get_internal_state(node_and_dist_references),
-    check_node_dist(Fail, NodeRefs, DistRefs).
-
-
+    R = check_node_dist(Fail, NodeRefs, DistRefs),
+    available_internal_state(AIS),
+    R.
 
 check_node_dist(Fail, NodeRefs, DistRefs) ->
-    check_nd_refc({node(),erlang:system_info(creation)},                  
-                  NodeRefs, DistRefs, Fail).
+    AIS = available_internal_state(true),
+    R = check_nd_refc({node(),erlang:system_info(creation)},
+                  NodeRefs, DistRefs, Fail),
+    available_internal_state(AIS),
+    R.
 
 
 check_nd_refc({ThisNodeName, ThisCreation}, NodeRefs, DistRefs, Fail) ->
diff --git a/erts/emulator/test/node_container_SUITE.erl b/erts/emulator/test/node_container_SUITE.erl
index b3d8f9584d..ca5c93f535 100644
--- a/erts/emulator/test/node_container_SUITE.erl
+++ b/erts/emulator/test/node_container_SUITE.erl
@@ -71,25 +71,10 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     erts_debug:set_internal_state(available_internal_state, true),
     erts_debug:set_internal_state(node_tab_delayed_delete, -1), %% restore original value
-    available_internal_state(false).
-
-available_internal_state(Bool) when Bool == true; Bool == false ->
-    case {Bool,
-          (catch erts_debug:get_internal_state(available_internal_state))} of
-        {true, true} ->
-            true;
-        {false, true} ->
-            erts_debug:set_internal_state(available_internal_state, false),
-            true;
-        {true, _} ->
-            erts_debug:set_internal_state(available_internal_state, true),
-            false;
-        {false, _} ->
-            false
-    end.
+    erts_test_util:available_internal_state(false).
 
 init_per_testcase(_Case, Config) when is_list(Config) ->
-    available_internal_state(true),
+    erts_test_util:available_internal_state(true),
     Config.
 
 end_per_testcase(_Case, Config) when is_list(Config) ->
@@ -928,9 +913,9 @@ id(X) ->
 -define(ND_REFS, erts_debug:get_internal_state(node_and_dist_references)).
 
 node_container_refc_check(Node) when is_atom(Node) ->
-    AIS = available_internal_state(true),
+    AIS = erts_test_util:available_internal_state(true),
     nc_refc_check(Node),
-    available_internal_state(AIS).
+    erts_test_util:available_internal_state(AIS).
 
 nc_refc_check(Node) when is_atom(Node) ->
     Ref = make_ref(),
diff --git a/lib/kernel/include/dist.hrl b/lib/kernel/include/dist.hrl
index 48701e12e4..f06fc328d7 100644
--- a/lib/kernel/include/dist.hrl
+++ b/lib/kernel/include/dist.hrl
@@ -44,6 +44,7 @@
 -define(DFLAG_BIG_SEQTRACE_LABELS, 16#100000).
 %% -define(DFLAG_NO_MAGIC, 16#200000). %% Used internally only
 -define(DFLAG_EXIT_PAYLOAD, 16#400000).
+-define(DFLAG_FRAGMENTS, 16#800000).
 
 %% Also update dflag2str() in ../src/dist_util.erl
 %% when adding flags...
diff --git a/lib/kernel/src/dist_util.erl b/lib/kernel/src/dist_util.erl
index 1d70a0593c..09ed31f10c 100644
--- a/lib/kernel/src/dist_util.erl
+++ b/lib/kernel/src/dist_util.erl
@@ -118,6 +118,8 @@ dflag2str(?DFLAG_BIG_SEQTRACE_LABELS) ->
     "BIG_SEQTRACE_LABELS";
 dflag2str(?DFLAG_EXIT_PAYLOAD) ->
     "EXIT_PAYLOAD";
+dflag2str(?DFLAG_FRAGMENTS) ->
+    "FRAGMENTS";
 dflag2str(_) ->
     "UNKNOWN".