Merge branch 'sverker/dist-flags-consolidate'

12 files changed, 170 insertions, 85 deletions

diff --git a/erts/doc/src/alt_dist.xml b/erts/doc/src/alt_dist.xml
index d3731a5391..92d40d8558 100644
--- a/erts/doc/src/alt_dist.xml
+++ b/erts/doc/src/alt_dist.xml
@@ -782,10 +782,9 @@
 	  <taglist>
 	    <tag><c>DFLAG_DIST_HDR_ATOM_CACHE</c></tag>
 	    <item>Do not use atom cache over this connection.</item>
-	    <tag><c>DFLAGS_STRICT_ORDER_DELIVERY</c></tag>
-	    <item>Do not use any features that require strict
-	    order delivery.</item>
 	  </taglist>
+	  <p>Use function <c>dist_util:strict_order_flags/0</c> to get all flags
+	    for features that require strict order delivery.</p>
 	  <p>
 	    This flag field is optional.
 	  </p>
@@ -819,9 +818,9 @@
       <p>
 	The data delivery order can be relaxed by disabling
 	features that require strict ordering. This is done by
-	passing the <c>?DFLAGS_STRICT_ORDER_DELIVERY</c>
-	<seealso marker="erl_dist_protocol#dflags">distribution
-	flags</seealso> in the
+	passing the
+	<seealso marker="erl_dist_protocol#dflags">distribution	flags</seealso>
+	returned by <c>dist_util:strict_order_flags/0</c> in the
 	<seealso marker="alt_dist#hs_data_reject_flags"><c>reject_flags</c></seealso>
 	field of the <seealso marker="#hs_data_record"><c>#hs_data{}</c></seealso>
 	record used when setting up the connection. When relaxed
diff --git a/erts/doc/src/erl_dist_protocol.xml b/erts/doc/src/erl_dist_protocol.xml
index a78b13aaa4..98a9a76b60 100644
--- a/erts/doc/src/erl_dist_protocol.xml
+++ b/erts/doc/src/erl_dist_protocol.xml
@@ -849,10 +849,9 @@ DiB == gen_digest(ChA, ICA)?
         </item>
       </taglist>
       <p>
-	There are also a collection of <c>DFLAG</c>s bitwise or:ed
-	together in the <c>DFLAGS_STRICT_ORDER_DELIVERY</c> macro.
-	These flags corresponds to features that require strict
-	ordering of data over distribution channels.
+	There is also function <c>dist_util:strict_order_flags/0</c>
+	returning all flags (bitwise or:ed together) corresponding to features
+	that require strict ordering of data over distribution channels.
       </p>
     </section>
   </section>
diff --git a/erts/doc/src/erlang.xml b/erts/doc/src/erlang.xml
index 56eaa47af4..8250ca5aff 100644
--- a/erts/doc/src/erlang.xml
+++ b/erts/doc/src/erlang.xml
@@ -5700,11 +5700,17 @@ true</pre>
       <type name="dst"/>
       <desc>
         <p>Sends a message and returns <c><anno>Msg</anno></c>. This
-          is the same as <c><anno>Dest</anno> ! <anno>Msg</anno></c>.</p>
+        is the same as using the <seealso marker="doc/reference_manual:expressions#send">
+	send operator</seealso>:
+	<c><anno>Dest</anno> ! <anno>Msg</anno></c>.</p>
         <p><c><anno>Dest</anno></c> can be a remote or local process identifier,
           a (local) port, a locally registered name, or a tuple
           <c>{<anno>RegName</anno>, <anno>Node</anno>}</c>
-          for a registered name at another node.</p>
+        for a registered name at another node.</p>
+	<p>The function fails with a <c>badarg</c> run-time error if
+	<c><anno>Dest</anno></c> is an atom name, but this name is not
+	registered. This is the only case when <c>send</c> fails for an
+	unreachable destination <c><anno>Dest</anno></c> (of correct type).</p>
       </desc>
     </func>
 
diff --git a/erts/emulator/beam/atom.names b/erts/emulator/beam/atom.names
index b06237e4f6..38b5f0c5e3 100644
--- a/erts/emulator/beam/atom.names
+++ b/erts/emulator/beam/atom.names
@@ -243,6 +243,7 @@ atom error_handler
 atom error_logger
 atom erts_code_purger
 atom erts_debug
+atom erts_dflags
 atom erts_internal
 atom ets
 atom ETS_TRANSFER='ETS-TRANSFER'
diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index 0d1166f6ed..be653ee2a0 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -690,6 +690,7 @@ bif erlang:iolist_to_iovec/1
 # New in 21.0
 #
 
+bif erts_internal:get_dflags/0
 bif erts_internal:new_connection/1
 bif erts_internal:abort_connection/2
 bif erts_internal:map_next/3
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 30390cdb5e..cd799e04b8 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -588,13 +588,13 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
                 erts_port_task_abort(&dep->dist_cmd);
         }
 
-	if (dep->status & ERTS_DE_SFLG_EXITING) {
+	if (dep->state == ERTS_DE_STATE_EXITING) {
 #ifdef DEBUG
 	    ASSERT(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_EXIT);
 #endif
 	}
 	else {
-	    dep->status |= ERTS_DE_SFLG_EXITING;
+	    dep->state = ERTS_DE_STATE_EXITING;
 	    erts_mtx_lock(&dep->qlock);
 	    ASSERT(!(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_EXIT));
 	    erts_atomic32_read_bor_relb(&dep->qflgs, ERTS_DE_QFLG_EXIT);
@@ -641,6 +641,14 @@ trap_function(Eterm func, int arity)
     return erts_export_put(am_erlang, func, arity);
 }
 
+/*
+ * Sync with dist_util.erl:
+ *
+ * -record(erts_dflags,
+ *         {default, mandatory, addable, rejectable, strict_order}).
+ */
+static Eterm erts_dflags_record;
+
 void init_dist(void)
 {
     init_nodes_monitors();
@@ -657,6 +665,16 @@ void init_dist(void)
     dist_ctrl_put_data_trap = erts_export_put(am_erts_internal,
                                               am_dist_ctrl_put_data,
                                               2);
+    {
+        Eterm* hp = erts_alloc(ERTS_ALC_T_LITERAL, (1+6)*sizeof(Eterm));
+        erts_dflags_record = TUPLE6(hp, am_erts_dflags,
+                                    make_small(DFLAG_DIST_DEFAULT),
+                                    make_small(DFLAG_DIST_MANDATORY),
+                                    make_small(DFLAG_DIST_ADDABLE),
+                                    make_small(DFLAG_DIST_REJECTABLE),
+                                    make_small(DFLAG_DIST_STRICT_ORDER));
+        erts_set_literal_tag(&erts_dflags_record, hp, (1+6));
+    }
 }
 
 #define ErtsDistOutputBuf2Binary(OB) \
@@ -767,7 +785,7 @@ static void clear_dist_entry(DistEntry *dep)
     erts_atomic64_set_nob(&dep->out, 0);
 
     obuf = clear_de_out_queues(dep);
-    dep->status = 0;
+    dep->state = ERTS_DE_STATE_IDLE;
     suspendees = get_suspended_on_de(dep, ERTS_DE_QFLGS_ALL);
 
     erts_mtx_unlock(&dep->qlock);
@@ -1982,6 +2000,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
                 break;
             }
             ctx->u.ec.flags = ctx->flags;
+            ctx->u.ec.hopefull_flags = 0;
             ctx->u.ec.level = 0;
             ctx->u.ec.wstack.wstart = NULL;
             ctx->obuf->msg_start = ctx->obuf->ext_endp;
@@ -2005,6 +2024,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 
 	    ctx->data_size = ctx->obuf->ext_endp - ctx->obuf->extp;
 
+            ctx->obuf->hopefull_flags = ctx->u.ec.hopefull_flags;
 	    /*
 	     * Signal encoded; now verify that the connection still exists,
 	     * and if so enqueue the signal and schedule it for send.
@@ -2012,8 +2032,8 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 	    ctx->obuf->next = NULL;
 	    erts_de_rlock(dep);
 	    cid = dep->cid;
-	    if (!(dep->status & (ERTS_DE_SFLG_PENDING | ERTS_DE_SFLG_CONNECTED))
-                || dep->status & ERTS_DE_SFLG_EXITING
+	    if (dep->state == ERTS_DE_STATE_EXITING
+                || dep->state == ERTS_DE_STATE_IDLE
                 || dep->connection_id != dsdp->connection_id) {
 		/* Not the same connection as when we started; drop message... */
 		erts_de_runlock(dep);
@@ -2088,7 +2108,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 		}
 
 		erts_mtx_unlock(&dep->qlock);
-		if (!(dep->status & ERTS_DE_SFLG_PENDING)) {
+		if (dep->state != ERTS_DE_STATE_PENDING) {
                     if (is_internal_port(dep->cid))
                         erts_schedule_dist_command(NULL, dep);
                 }
@@ -2270,7 +2290,7 @@ int
 erts_dist_command(Port *prt, int initial_reds)
 {
     Sint reds = initial_reds - ERTS_PORT_REDS_DIST_CMD_START;
-    Uint32 status;
+    enum dist_entry_state state;
     Uint32 flags;
     Sint qsize, obufsize = 0;
     ErtsDistOutputQueue oq, foq;
@@ -2285,17 +2305,17 @@ erts_dist_command(Port *prt, int initial_reds)
 
     erts_de_rlock(dep);
     flags = dep->flags;
-    status = dep->status;
+    state = dep->state;
     send = dep->send;
     erts_de_runlock(dep);
 
-    if (status & ERTS_DE_SFLG_EXITING) {
+    if (state == ERTS_DE_STATE_EXITING) {
 	erts_deliver_port_exit(prt, prt->common.id, am_killed, 0, 1);
         reds -= ERTS_PORT_REDS_DIST_CMD_EXIT;
 	return initial_reds - reds;
     }
 
-    ASSERT(!(status & ERTS_DE_SFLG_PENDING));
+    ASSERT(state != ERTS_DE_STATE_PENDING);
 
     ASSERT(send);
 
@@ -2831,7 +2851,7 @@ dist_ctrl_get_data_1(BIF_ALIST_1)
 
     erts_de_rlock(dep);
 
-    if (dep->status & ERTS_DE_SFLG_EXITING)
+    if (dep->state == ERTS_DE_STATE_EXITING)
         goto return_none;
 
     ASSERT(dep->cid == BIF_P->common.id);
@@ -2934,9 +2954,9 @@ erts_dist_port_not_busy(Port *prt)
 static void kill_connection(DistEntry *dep)
 {
     ERTS_LC_ASSERT(erts_lc_is_de_rwlocked(dep));
-    ASSERT(dep->status == ERTS_DE_SFLG_CONNECTED);
+    ASSERT(dep->state == ERTS_DE_STATE_CONNECTED);
 
-    dep->status |= ERTS_DE_SFLG_EXITING;
+    dep->state = ERTS_DE_STATE_EXITING;
     erts_mtx_lock(&dep->qlock);
     ASSERT(!(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_EXIT));
     erts_atomic32_read_bor_nob(&dep->qflgs, ERTS_DE_QFLG_EXIT);
@@ -2953,7 +2973,7 @@ erts_kill_dist_connection(DistEntry *dep, Uint32 connection_id)
 {
     erts_de_rwlock(dep);
     if (connection_id == dep->connection_id
-        && dep->status == ERTS_DE_SFLG_CONNECTED) {
+        && dep->state == ERTS_DE_STATE_CONNECTED) {
 
         kill_connection(dep);
     }
@@ -3341,7 +3361,7 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
             goto badarg;
         }
 
-        if (dep->status & ERTS_DE_SFLG_EXITING) {
+        if (dep->state == ERTS_DE_STATE_EXITING) {
             /* Suspend on dist entry waiting for the exit to finish */
             ErtsProcList *plp = erts_proclist_create(BIF_P);
             plp->next = NULL;
@@ -3351,8 +3371,8 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
             erts_mtx_unlock(&dep->qlock);
             goto yield;
         }
-        if (dep->status != ERTS_DE_SFLG_PENDING) {
-            if (dep->status == 0)
+        if (dep->state != ERTS_DE_STATE_PENDING) {
+            if (dep->state == ERTS_DE_STATE_IDLE)
                 erts_set_dist_entry_pending(dep);
             else
                 goto badarg;
@@ -3390,7 +3410,7 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
             goto done; /* Already set */
         }
 
-        if (dep->status & ERTS_DE_SFLG_EXITING) {
+        if (dep->state == ERTS_DE_STATE_EXITING) {
             /* Suspend on dist entry waiting for the exit to finish */
             ErtsProcList *plp = erts_proclist_create(BIF_P);
             plp->next = NULL;
@@ -3400,8 +3420,8 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
             erts_mtx_unlock(&dep->qlock);
             goto yield;
         }
-        if (dep->status != ERTS_DE_SFLG_PENDING) {
-            if (dep->status == 0)
+        if (dep->state != ERTS_DE_STATE_PENDING) {
+            if (dep->state == ERTS_DE_STATE_IDLE)
                 erts_set_dist_entry_pending(dep);
             else
                 goto badarg;
@@ -3437,7 +3457,7 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
 
 #ifdef DEBUG
     ASSERT(erts_atomic_read_nob(&dep->qsize) == 0 
-           || (dep->status & ERTS_DE_SFLG_PENDING));
+           || (dep->state == ERTS_DE_STATE_PENDING));
 #endif
 
     if (flags & DFLAG_DIST_HDR_ATOM_CACHE)
@@ -3506,6 +3526,11 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
     goto done;
 }
 
+BIF_RETTYPE erts_internal_get_dflags_0(BIF_ALIST_0)
+{
+    return erts_dflags_record;
+}
+
 BIF_RETTYPE erts_internal_new_connection_1(BIF_ALIST_1)
 {
     DistEntry* dep;
@@ -3525,15 +3550,20 @@ BIF_RETTYPE erts_internal_new_connection_1(BIF_ALIST_1)
 
     erts_de_rwlock(dep);
 
-    if (ERTS_DE_IS_CONNECTED(dep) || dep->status & ERTS_DE_SFLG_PENDING)
+    switch (dep->state) {
+    case ERTS_DE_STATE_PENDING:
+    case ERTS_DE_STATE_CONNECTED:
 	conn_id = dep->connection_id;
-    else if (dep->status == 0) {
+        break;
+    case ERTS_DE_STATE_IDLE:
         erts_set_dist_entry_pending(dep);
 	conn_id = dep->connection_id;
-    }
-    else {
-        ASSERT(dep->status & ERTS_DE_SFLG_EXITING);
+        break;
+    case ERTS_DE_STATE_EXITING:
         conn_id = (dep->connection_id + 1) & ERTS_DIST_CON_ID_MASK;
+        break;
+    default:
+        erts_exit(ERTS_ABORT_EXIT, "Invalid dep->state (%d)\n", dep->state);
     }
     erts_de_rwunlock(dep);
     hp = HAlloc(BIF_P, 3 + ERTS_MAGIC_REF_THING_SIZE);
@@ -3548,10 +3578,10 @@ static Sint abort_connection(DistEntry* dep, Uint32 conn_id)
 
     if (dep->connection_id != conn_id)
         ;
-    else if (dep->status == ERTS_DE_SFLG_CONNECTED) {
+    else if (dep->state == ERTS_DE_STATE_CONNECTED) {
         kill_connection(dep);
     }
-    else if (dep->status == ERTS_DE_SFLG_PENDING) {
+    else if (dep->state == ERTS_DE_STATE_PENDING) {
         NetExitsContext nec = {dep};
         ErtsLink *nlinks;
         ErtsLink *node_links;
@@ -3600,6 +3630,17 @@ static Sint abort_connection(DistEntry* dep, Uint32 conn_id)
 
 	delete_cache(cache);
 	free_de_out_queues(dep, obuf);
+
+        /*
+         * We wait to make DistEntry idle and accept new connection attempts
+         * until all is cleared and deallocated. This to get some back pressure
+         * against repeated failing connection attempts saturating all CPUs
+         * with cleanup jobs.
+         */
+        erts_de_rwlock(dep);
+        ASSERT(dep->state == ERTS_DE_STATE_EXITING);
+        dep->state = ERTS_DE_STATE_IDLE;
+        erts_de_rwunlock(dep);
         return reds;
     }
     erts_de_rwunlock(dep);
@@ -3631,7 +3672,7 @@ BIF_RETTYPE erts_internal_abort_connection_2(BIF_ALIST_2)
 int erts_auto_connect(DistEntry* dep, Process *proc, ErtsProcLocks proc_locks)
 {
     erts_de_rwlock(dep);
-    if (dep->status != 0) {
+    if (dep->state != ERTS_DE_STATE_IDLE) {
         erts_de_rwunlock(dep);
     }
     else {
@@ -3907,7 +3948,8 @@ monitor_node(Process* p, Eterm Node, Eterm Bool, Eterm Options)
 
         erts_proc_lock(p, ERTS_PROC_LOCK_LINK);
         erts_de_rlock(dep);
-        if (!(dep->status & (ERTS_DE_SFLG_PENDING | ERTS_DE_SFLG_CONNECTED))) {
+        if (dep->state == ERTS_DE_STATE_IDLE) {
+            ASSERT(!dep->node_links);
             erts_proc_unlock(p, ERTS_PROC_LOCK_LINK);
             erts_de_runlock(dep);
             goto done;
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index ea4697815f..b1b7ce9c78 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -40,26 +40,52 @@
 #define DFLAG_UNICODE_IO          0x1000
 #define DFLAG_DIST_HDR_ATOM_CACHE 0x2000
 #define DFLAG_SMALL_ATOM_TAGS     0x4000
-#define DFLAG_INTERNAL_TAGS       0x8000
+#define DFLAG_INTERNAL_TAGS       0x8000   /* used by ETS 'compressed' option */
 #define DFLAG_UTF8_ATOMS          0x10000
 #define DFLAG_MAP_TAG             0x20000
 #define DFLAG_BIG_CREATION        0x40000
 #define DFLAG_SEND_SENDER         0x80000
-#define DFLAG_NO_MAGIC            0x100000
+#define DFLAG_NO_MAGIC            0x100000 /* internal for pending connection */
 
-/* Mandatory flags for distribution (sync with dist_util.erl) */
+/* Mandatory flags for distribution */
 #define DFLAG_DIST_MANDATORY (DFLAG_EXTENDED_REFERENCES         \
                               | DFLAG_EXTENDED_PIDS_PORTS       \
 			      | DFLAG_UTF8_ATOMS                \
 			      | DFLAG_NEW_FUN_TAGS)
 
-/* Additional optimistic flags when encoding toward pending connection */
-#define DFLAG_DIST_HOPEFULLY (DFLAG_NO_MAGIC                    \
-                              | DFLAG_EXPORT_PTR_TAG            \
+/*
+ * Additional optimistic flags when encoding toward pending connection.
+ * If remote node (erl_interface) does not supporting these then we may need
+ * to transcode messages enqueued before connection setup was finished.
+ */
+#define DFLAG_DIST_HOPEFULLY (DFLAG_EXPORT_PTR_TAG              \
                               | DFLAG_BIT_BINARIES              \
                               | DFLAG_DIST_MONITOR              \
                               | DFLAG_DIST_MONITOR_NAME)
 
+/* Our preferred set of flags. Used for connection setup handshake */
+#define DFLAG_DIST_DEFAULT (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY \
+                            | DFLAG_FUN_TAGS                  \
+                            | DFLAG_NEW_FLOATS                \
+                            | DFLAG_UNICODE_IO                \
+                            | DFLAG_DIST_HDR_ATOM_CACHE       \
+                            | DFLAG_SMALL_ATOM_TAGS           \
+                            | DFLAG_UTF8_ATOMS                \
+                            | DFLAG_MAP_TAG                   \
+                            | DFLAG_BIG_CREATION              \
+                            | DFLAG_SEND_SENDER)
+
+/* Flags addable by local distr implementations */
+#define DFLAG_DIST_ADDABLE    DFLAG_DIST_DEFAULT
+
+/* Flags rejectable by local distr implementation */
+#define DFLAG_DIST_REJECTABLE (DFLAG_DIST_HDR_ATOM_CACHE         \
+                               | DFLAG_HIDDEN_ATOM_CACHE         \
+                               | DFLAG_ATOM_CACHE)
+
+/* Flags for all features needing strict order delivery */
+#define DFLAG_DIST_STRICT_ORDER DFLAG_DIST_HDR_ATOM_CACHE
+
 /* All flags that should be enabled when term_to_binary/1 is used. */
 #define TERM_TO_BINARY_DFLAGS (DFLAG_EXTENDED_REFERENCES	\
 			       | DFLAG_NEW_FUN_TAGS		\
@@ -111,14 +137,6 @@ typedef struct {
     int no_suspend;
 } ErtsDSigData;
 
-#define ERTS_DE_IS_NOT_CONNECTED(DEP) \
-  (ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&(DEP)->rwmtx) \
-		      || erts_lc_rwmtx_is_rwlocked(&(DEP)->rwmtx)), \
-   (is_nil((DEP)->cid) || ((DEP)->status & ERTS_DE_SFLG_EXITING)))
-
-#define ERTS_DE_IS_CONNECTED(DEP) \
-  (!ERTS_DE_IS_NOT_CONNECTED((DEP)))
-
 #define ERTS_DE_BUSY_LIMIT (1024*1024)
 extern int erts_dist_buf_busy_limit;
 extern int erts_is_alive;
@@ -181,18 +199,18 @@ erts_dsig_prepare(ErtsDSigData *dsdp,
 retry:
     erts_de_rlock(dep);
 
-    if (ERTS_DE_IS_CONNECTED(dep)) {
+    if (dep->state == ERTS_DE_STATE_CONNECTED) {
 	res = ERTS_DSIG_PREP_CONNECTED;
     }
-    else if (dep->status & ERTS_DE_SFLG_PENDING) {
+    else if (dep->state == ERTS_DE_STATE_PENDING) {
 	res = ERTS_DSIG_PREP_PENDING;
     }
-    else if (dep->status & ERTS_DE_SFLG_EXITING) {
+    else if (dep->state == ERTS_DE_STATE_EXITING) {
 	res = ERTS_DSIG_PREP_NOT_CONNECTED;
 	goto fail;
     }
     else if (connect) {
-        ASSERT(dep->status == 0);
+        ASSERT(dep->state == ERTS_DE_STATE_IDLE);
         erts_de_runlock(dep);
         if (!erts_auto_connect(dep, proc, proc_locks)) {
             return ERTS_DSIG_PREP_NOT_ALIVE;
@@ -200,7 +218,7 @@ retry:
 	goto retry;
     }
     else {
-        ASSERT(dep->status == 0);
+        ASSERT(dep->state == ERTS_DE_STATE_IDLE);
 	res = ERTS_DSIG_PREP_NOT_CONNECTED;
 	goto fail;
     }
@@ -328,6 +346,7 @@ typedef struct TTBSizeContext_ {
 
 typedef struct TTBEncodeContext_ {
     Uint flags;
+    Uint hopefull_flags;
     int level;
     byte* ep;
     Eterm obj;
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index eaf133f5c0..e8901a652f 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -170,7 +170,7 @@ dist_table_alloc(void *dep_tmpl)
     dep->cid				= NIL;
     erts_atomic_init_nob(&dep->input_handler, (erts_aint_t) NIL);
     dep->connection_id			= 0;
-    dep->status				= 0;
+    dep->state				= ERTS_DE_STATE_IDLE;
     dep->flags				= 0;
     dep->version			= 0;
 
@@ -223,7 +223,7 @@ dist_table_free(void *vdep)
     DistEntry *dep = (DistEntry *) vdep;
 
     ASSERT(de_refc_read(dep, -1) == -1);
-    ASSERT(dep->status == 0);
+    ASSERT(dep->state == ERTS_DE_STATE_IDLE);
     ASSERT(is_nil(dep->cid));
     ASSERT(dep->nlinks == NULL);
     ASSERT(dep->node_links == NULL);
@@ -556,14 +556,14 @@ erts_set_dist_entry_not_connected(DistEntry *dep)
 
     ASSERT(dep != erts_this_dist_entry);
 
-    if (dep->status & ERTS_DE_SFLG_PENDING) {
+    if (dep->state == ERTS_DE_STATE_PENDING) {
         ASSERT(is_nil(dep->cid));
         ASSERT(erts_no_of_pending_dist_entries > 0);
         erts_no_of_pending_dist_entries--;
         head = &erts_pending_dist_entries;
     }
     else {
-        ASSERT(dep->status != 0);
+        ASSERT(dep->state != ERTS_DE_STATE_IDLE);
         ASSERT(is_internal_port(dep->cid) || is_internal_pid(dep->cid));
         if (dep->flags & DFLAG_PUBLISHED) {
             ASSERT(erts_no_of_visible_dist_entries > 0);
@@ -588,7 +588,7 @@ erts_set_dist_entry_not_connected(DistEntry *dep)
     if(dep->next)
 	dep->next->prev = dep->prev;
 
-    dep->status &= ~(ERTS_DE_SFLG_PENDING | ERTS_DE_SFLG_CONNECTED);
+    dep->state = ERTS_DE_STATE_EXITING;
     dep->flags = 0;
     dep->prev = NULL;
     dep->cid = NIL;
@@ -610,7 +610,7 @@ erts_set_dist_entry_pending(DistEntry *dep)
     erts_rwmtx_rwlock(&erts_dist_table_rwmtx);
 
     ASSERT(dep != erts_this_dist_entry);
-    ASSERT(dep->status == 0);
+    ASSERT(dep->state == ERTS_DE_STATE_IDLE);
     ASSERT(is_nil(dep->cid));
 
     if(dep->prev) {
@@ -627,8 +627,8 @@ erts_set_dist_entry_pending(DistEntry *dep)
 
     erts_no_of_not_connected_dist_entries--;
 
-    dep->status = ERTS_DE_SFLG_PENDING;
-    dep->flags = (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY);
+    dep->state = ERTS_DE_STATE_PENDING;
+    dep->flags = (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY | DFLAG_NO_MAGIC);
     dep->connection_id = (dep->connection_id + 1) & ERTS_DIST_CON_ID_MASK;
 
     dep->prev = NULL;
@@ -652,7 +652,7 @@ erts_set_dist_entry_connected(DistEntry *dep, Eterm cid, Uint flags)
 
     ASSERT(dep != erts_this_dist_entry);
     ASSERT(is_nil(dep->cid));
-    ASSERT(dep->status & ERTS_DE_SFLG_PENDING);
+    ASSERT(dep->state == ERTS_DE_STATE_PENDING);
     ASSERT(is_internal_port(cid) || is_internal_pid(cid));
 
     if(dep->prev) {
@@ -670,8 +670,7 @@ erts_set_dist_entry_connected(DistEntry *dep, Eterm cid, Uint flags)
     ASSERT(erts_no_of_pending_dist_entries > 0);
     erts_no_of_pending_dist_entries--;
 
-    dep->status &= ~ERTS_DE_SFLG_PENDING;
-    dep->status |= ERTS_DE_SFLG_CONNECTED;
+    dep->state = ERTS_DE_STATE_CONNECTED;
     dep->flags = flags & ~DFLAG_NO_MAGIC;
     dep->cid = cid;
     erts_atomic_set_nob(&dep->input_handler,
diff --git a/erts/emulator/beam/erl_node_tables.h b/erts/emulator/beam/erl_node_tables.h
index 8d29c83e15..58279017c8 100644
--- a/erts/emulator/beam/erl_node_tables.h
+++ b/erts/emulator/beam/erl_node_tables.h
@@ -57,9 +57,12 @@
  
 #define ERST_INTERNAL_CHANNEL_NO 0
 
-#define ERTS_DE_SFLG_PENDING			(((Uint32) 1) <<  0)
-#define ERTS_DE_SFLG_CONNECTED			(((Uint32) 1) <<  1)
-#define ERTS_DE_SFLG_EXITING			(((Uint32) 1) <<  2)
+enum dist_entry_state {
+    ERTS_DE_STATE_IDLE,
+    ERTS_DE_STATE_PENDING,
+    ERTS_DE_STATE_CONNECTED,
+    ERTS_DE_STATE_EXITING
+};
 
 #define ERTS_DE_QFLG_BUSY			(((erts_aint32_t) 1) <<  0)
 #define ERTS_DE_QFLG_EXIT			(((erts_aint32_t) 1) <<  1)
@@ -86,6 +89,7 @@ struct ErtsDistOutputBuf_ {
     byte *alloc_endp;
 #endif
     ErtsDistOutputBuf *next;
+    Uint hopefull_flags;
     byte *extp;
     byte *ext_endp;
     byte *msg_start;
@@ -121,7 +125,7 @@ typedef struct dist_entry_ {
     Eterm cid;			/* connection handler (pid or port),
                                    NIL == free */
     Uint32 connection_id;	/* Connection id incremented on connect */
-    Uint32 status;		/* Slot status, like exiting reserved etc */
+    enum dist_entry_state state;
     Uint32 flags;		/* Distribution flags, like hidden, 
 				   atom cache etc. */
     unsigned long version;	/* Protocol version */
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index a5d752a3d0..fb42969a8f 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -711,8 +711,8 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 
     erts_de_rlock(dep);
 
-    if (dep->status != ERTS_DE_SFLG_CONNECTED &&
-	dep->status != ERTS_DE_SFLG_PENDING) {
+    if (dep->state != ERTS_DE_STATE_CONNECTED &&
+	dep->state != ERTS_DE_STATE_PENDING) {
         erts_de_runlock(dep);
         return ERTS_PREP_DIST_EXT_CLOSED;
     }
@@ -2873,6 +2873,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    ep[j] = 0;	/* Zero unused bits at end of binary */
 		    data_dst = ep;
 		    ep += j + 1;
+                    if (ctx)
+                        ctx->hopefull_flags |= DFLAG_BIT_BINARIES;
 		} else {
 		    /*
 		     * Bit-level binary, but the receiver doesn't support it.
@@ -2908,6 +2910,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep,
 		    ep = enc_atom(acmp, exp->info.mfa.function, ep, dflags);
 		    ep = enc_term(acmp, make_small(exp->info.mfa.arity),
                                   ep, dflags, off_heap);
+                    if (ctx)
+                        ctx->hopefull_flags |= DFLAG_EXPORT_PTR_TAG;
 		} else {
 		    /* Tag, arity */
 		    *ep++ = SMALL_TUPLE_EXT;
@@ -4729,11 +4733,13 @@ Sint transcode_dist_obuf(ErtsDistOutputBuf* ob,
     struct transcode_context* ctx = dep->transcode_ctx;
 
     if (!ctx) { /* first call for 'ob' */
-
-        if (~dflags & (DFLAG_BIT_BINARIES | DFLAG_EXPORT_PTR_TAG)) {
+	ASSERT(!(ob->hopefull_flags & ~(Uint)(DFLAG_BIT_BINARIES |
+					      DFLAG_EXPORT_PTR_TAG)));
+        if (~dflags & ob->hopefull_flags) {
             /*
-             * Receiver does not support bitstrings and/or export funs.
-             * We need to transcode control and message terms to use tuple fallbacks.
+             * Receiver does not support bitstrings and/or export funs
+             * and output buffer contains such message tags (hopefull_flags).
+             * Must transcode control and message terms to use tuple fallbacks.
              */
             ctx = erts_alloc(ERTS_ALC_T_DIST_TRANSCODE, sizeof(struct transcode_context));
             dep->transcode_ctx = ctx;
@@ -4760,7 +4766,7 @@ Sint transcode_dist_obuf(ErtsDistOutputBuf* ob,
                 ctx->state = TRANSCODE_ENC_CTL;
             }
         }
-        else {
+        else if (!(dflags & DFLAG_DIST_HDR_ATOM_CACHE)) {
             /*
              * No need for full transcoding, but primitive receiver (erl_/jinterface)
              * expects VERSION_MAGIC before both control and message terms.
@@ -4777,8 +4783,10 @@ Sint transcode_dist_obuf(ErtsDistOutputBuf* ob,
             *--(ob->extp) = VERSION_MAGIC;
             goto done;
         }
+        else
+            goto done;
     }
-    else {
+    else {  /* continue after yield */
         ASSERT(ctx->dbg_ob == ob);
     }
     ctx->b2t.reds = reds * B2T_BYTES_PER_REDUCTION;
@@ -4831,6 +4839,7 @@ Sint transcode_dist_obuf(ErtsDistOutputBuf* ob,
         ob->msg_start = ob->ext_endp;
         ctx->ttb.wstack.wstart = NULL;
         ctx->ttb.flags = dflags;
+        ctx->ttb.hopefull_flags = 0;
         ctx->ttb.level = 0;
 
         ctx->state = TRANSCODE_ENC_MSG;
@@ -4843,7 +4852,7 @@ Sint transcode_dist_obuf(ErtsDistOutputBuf* ob,
         reds /= TERM_TO_BINARY_LOOP_FACTOR;
 
         ASSERT(ob->ext_endp <= ob->alloc_endp);
-
+        ASSERT(!ctx->ttb.hopefull_flags);
     }
     transcode_free_ctx(dep);
 
diff --git a/erts/preloaded/ebin/erts_internal.beam b/erts/preloaded/ebin/erts_internal.beam
index 7d2edd9845..b79f734a6d 100644
--- a/erts/preloaded/ebin/erts_internal.beam
+++ b/erts/preloaded/ebin/erts_internal.beam
diff --git a/erts/preloaded/src/erts_internal.erl b/erts/preloaded/src/erts_internal.erl
index a083e9ac2f..a51c0c4c0e 100644
--- a/erts/preloaded/src/erts_internal.erl
+++ b/erts/preloaded/src/erts_internal.erl
@@ -63,6 +63,7 @@
 
 -export([dist_ctrl_put_data/2]).
 
+-export([get_dflags/0]).
 -export([new_connection/1]).
 -export([abort_connection/2]).
 
@@ -510,6 +511,11 @@ dist_ctrl_put_data(DHandle, IoList) ->
     end.
 
 
+-spec erts_internal:get_dflags() -> {erts_dflags, integer(), integer(),
+                                     integer(), integer(), integer()}.
+get_dflags() ->
+    erlang:nif_error(undefined).
+
 -spec erts_internal:new_connection(Node) -> ConnId when
       Node :: atom(),
       ConnId :: {integer(), erlang:dist_handle()}.