Add optimistic DFLAG_DIST_HOPEFULLY for pending connections

to avoid tuple fallbacks for export funs and bitstrings.

ToDo: Re-encode if receiver turn out to be erl_interface/jinterface.

5 files changed, 62 insertions, 8 deletions

diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 6d2e907f18..7877865ad4 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -3443,7 +3443,7 @@ BIF_RETTYPE new_connection_id_1(BIF_ALIST_1)
 	conn_id = dep->connection_id;
     else if (dep->status == 0) {
 	dep->status = ERTS_DE_SFLG_PENDING;
-	dep->flags = DFLAG_DIST_MANDATORY | DFLAG_PENDING_CONNECTION;
+	dep->flags = (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY);
 	dep->connection_id++;
 	dep->connection_id &= ERTS_DIST_CON_ID_MASK;
 	conn_id = dep->connection_id;
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index 505d510473..f9a2037687 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -45,7 +45,7 @@
 #define DFLAG_MAP_TAG             0x20000
 #define DFLAG_BIG_CREATION        0x40000
 #define DFLAG_SEND_SENDER         0x80000
-#define DFLAG_PENDING_CONNECTION  0x100000
+#define DFLAG_NO_MAGIC            0x100000
 
 /* Mandatory flags for distribution (sync with dist_util.erl) */
 #define DFLAG_DIST_MANDATORY (DFLAG_EXTENDED_REFERENCES         \
@@ -53,6 +53,11 @@
 			      | 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            \
+                              | DFLAG_BIT_BINARIES)
+
 /* 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		\
@@ -206,7 +211,7 @@ retry:
 	    Eterm msg, conn_id;
 
 	    dep->status = ERTS_DE_SFLG_PENDING;
-	    dep->flags = DFLAG_DIST_MANDATORY | DFLAG_PENDING_CONNECTION;
+	    dep->flags = (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY);
 	    dep->connection_id++;
 	    dep->connection_id &= ERTS_DIST_CON_ID_MASK;
             conn_id = make_small(dep->connection_id);
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index dd607f438e..2a33766ac8 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -628,7 +628,7 @@ erts_set_dist_entry_connected(DistEntry *dep, Eterm cid, Uint flags)
 	dep->connection_id &= ERTS_DIST_CON_ID_MASK;
     }
     dep->status |= ERTS_DE_SFLG_CONNECTED;
-    dep->flags = flags & ~DFLAG_PENDING_CONNECTION;
+    dep->flags = flags & ~DFLAG_NO_MAGIC;
     dep->cid = cid;
     erts_atomic_set_nob(&dep->input_handler,
                             (erts_aint_t) cid);
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index d8f52ceb57..bfac48580c 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -561,7 +561,7 @@ int erts_encode_dist_ext_size_int(Eterm term, struct erts_dsig_send_context* ctx
 	return -1;
     } else {
 #ifndef ERTS_DEBUG_USE_DIST_SEP
-	if (!(ctx->flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_PENDING_CONNECTION)))
+	if (!(ctx->flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_NO_MAGIC)))
 #endif
 	    sz++ /* VERSION_MAGIC */;
 
@@ -593,7 +593,7 @@ int erts_encode_dist_ext(Eterm term, byte **ext, Uint32 flags, ErtsAtomCacheMap
 {
     if (!ctx || !ctx->wstack.wstart) {
     #ifndef ERTS_DEBUG_USE_DIST_SEP
-	if (!(flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_PENDING_CONNECTION)))
+	if (!(flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_NO_MAGIC)))
     #endif
 	    *(*ext)++ = VERSION_MAGIC;
     }
diff --git a/erts/emulator/test/distribution_SUITE.erl b/erts/emulator/test/distribution_SUITE.erl
index fa683f363f..f1831b7c45 100644
--- a/erts/emulator/test/distribution_SUITE.erl
+++ b/erts/emulator/test/distribution_SUITE.erl
@@ -41,6 +41,7 @@
 -export([all/0, suite/0, groups/0,
          ping/1, bulk_send_small/1,
          group_leader/1,
+         optimistic_dflags/1,
          bulk_send_big/1, bulk_send_bigbig/1,
          local_send_small/1, local_send_big/1,
          local_send_legal/1, link_to_busy/1, exit_to_busy/1,
@@ -66,6 +67,7 @@
 %% Internal exports.
 -export([sender/3, receiver2/2, dummy_waiter/0, dead_process/0,
          group_leader_1/1,
+         optimistic_dflags_echo/0, optimistic_dflags_sender/1,
          roundtrip/1, bounce/1, do_dist_auto_connect/1, inet_rpc_server/1,
          dist_parallel_sender/3, dist_parallel_receiver/0,
          dist_evil_parallel_receiver/0]).
@@ -80,6 +82,7 @@ suite() ->
 all() ->
     [ping, {group, bulk_send}, {group, local_send},
      group_leader,
+     optimistic_dflags,
      link_to_busy, exit_to_busy, lost_exit, link_to_dead,
      link_to_dead_new_node,
      ref_port_roundtrip, nil_roundtrip, stop_dist,
@@ -166,12 +169,58 @@ group_leader_1(Node2) ->
     ?Line {ExtPid, group_leader, GL2} = receive_one(),
     ok.
 
-receive_one() ->
-    receive M -> M after 1000 -> timeout end.
+%% Test optimistic distribution flags toward pending connections (DFLAG_DIST_HOPEFULLY)
+optimistic_dflags(Config) when is_list(Config) ->
+    ?Line Sender = start_relay_node(optimistic_dflags_sender, []),
+    ?Line Echo = start_relay_node(optimistic_dflags_echo, []),
+    try
+        ?Line {ok, ok} = do_inet_rpc(Echo, ?MODULE, optimistic_dflags_echo, []),
+
+        ?Line EchoNode = inet_rpc_nodename(Echo),
+        ?Line {ok, ok} = do_inet_rpc(Sender, ?MODULE, optimistic_dflags_sender, [EchoNode])
+    after
+        ?Line stop_relay_node(Sender),
+        ?Line stop_relay_node(Echo)
+    end,
+    ok.
+
+optimistic_dflags_echo() ->
+    P = spawn(fun F() ->
+                      receive {From, Term} ->
+                              From ! {self(), Term}
+                      end,
+                      F()
+              end),
+    register(optimistic_dflags_echo, P),
+    optimistic_dflags_echo ! {self(), hello},
+    {P, hello} = receive_one(),
+    ok.
+
+optimistic_dflags_sender(EchoNode) ->
+    ?Line net_kernel:monitor_nodes(true),
 
+    optimistic_dflags_do(EchoNode, <<1:1>>),
+    optimistic_dflags_do(EchoNode, fun lists:map/2),
+    ok.
 
+optimistic_dflags_do(EchoNode, Term) ->
+    ?Line {optimistic_dflags_echo, EchoNode} ! {self(), Term},
+    ?Line {nodeup, EchoNode} = receive_one(),
+    ?Line {EchoPid, Term} = receive_one(),
+    %% repeat with pid destination
+    ?Line net_kernel:disconnect(EchoNode),
+    ?Line {nodedown, EchoNode} = receive_one(),
+    ?Line EchoPid ! {self(), Term},
+    ?Line {nodeup, EchoNode} = receive_one(),
+    ?Line {EchoPid, Term} = receive_one(),
+
+    ?Line net_kernel:disconnect(EchoNode),
+    ?Line {nodedown, EchoNode} = receive_one(),
+    ok.
 
 
+receive_one() ->
+    receive M -> M after 1000 -> timeout end.
 
 
 bulk_send_small(Config) when is_list(Config) ->