Merge branch 'sverker/async-auto-connect/OTP-14370'

* sverker/async-auto-connect/OTP-14370: (37 commits)
  Move new|abort_connection_id to erts_internal
  Refactor erts_dsig_prepare argument dep(p)
  Cleanup net_kernel
  Improve connection aborting
  Abort all pending connections if net_kernel terminates
  erts: Put pending DistrEntry in separate list
  Refactor auto_connect into an outline function
  Remove unused ERTS_DSP_RWLOCK
  fix erlang specs and preloaded
  erts: Keep magic ref to DistEntry in net_kernel
  Allow DistEntries in ETS
  Remove faulty assert
  erts: Transcode tuple fallbacks
  erts: Ensure enc_term_int() always do progress
  erl_interface: Add tuple fallback tests
  erl_interface: Refactor ei_accept_SUITE
  Add optimistic DFLAG_DIST_HOPEFULLY for pending connections
  erts: Fix auto-connect toward erl_interface/jinterface
  erts: Let send(_,_,[noconnect]) enqueue msg on pending connection.
  Remove obsolete erlang:dgroup_leader
  ...

7 files changed, 301 insertions, 160 deletions

diff --git a/lib/erl_interface/test/ei_accept_SUITE.erl b/lib/erl_interface/test/ei_accept_SUITE.erl
index e06ee762d7..16eb0901ef 100644
--- a/lib/erl_interface/test/ei_accept_SUITE.erl
+++ b/lib/erl_interface/test/ei_accept_SUITE.erl
@@ -44,33 +44,31 @@ ei_accept(Config) when is_list(Config) ->
     io:format("Myname ~p ~n",  [Myname]),
     EINode = list_to_atom("c42@"++Myname),
     io:format("EINode ~p ~n",  [EINode]),
+
+    %% We take this opportunity to also test export-funs and bit-strings
+    %% with (ugly) tuple fallbacks.
+    %% Test both toward pending connection and established connection.
+    RealTerms = [<<1:1>>,     fun lists:map/2],
+    Fallbacks = [{<<128>>,1}, {lists,map}],
+
     Self = self(),
-    TermToSend= {call, Self, "Test"},
-    F= fun() ->
-               case waitfornode("c42",20) of
-                   true ->
-                       {any, EINode} ! TermToSend,
-                       Self ! sent_ok;
-                   false ->
-                       Self ! never_published
-               end,
-               ok
-       end,
-
-    spawn(F),
+    Funny = fun() -> hello end,
+    TermToSend = {call, Self, "Test", Funny, RealTerms},
+    TermToGet  = {call, Self, "Test", Funny, Fallbacks},
     Port = 6543,
-    {ok, Fd, _Node} = ei_accept(P, Port),
-    TermReceived= ei_receive(P, Fd),
-    io:format("Sent ~p received ~p ~n", [TermToSend, TermReceived]),
-    TermToSend= TermReceived,
-    receive
-        sent_ok ->
-            ok;
-        Unknown ->
-            io:format("~p ~n", [Unknown])
-    after 1000 ->
-              io:format("timeout ~n")
-    end,
+    {ok, ListenFd} = ei_publish(P, Port),
+    {any, EINode} ! TermToSend,
+    {ok, Fd, _Node} = ei_accept(P, ListenFd),
+    Got1 = ei_receive(P, Fd),
+
+    %% Send again, now without auto-connect
+    {any, EINode} ! TermToSend,
+    Got2 = ei_receive(P, Fd),
+
+    io:format("Sent ~p~nExp. ~p~nGot1 ~p~nGot2 ~p~n", [TermToSend, TermToGet, Got1, Got2]),
+    TermToGet = Got1,
+    TermToGet = Got2,
+
     runner:finish(P),
     ok.
 
@@ -137,8 +135,15 @@ ei_connect_init(P, Num, Cookie, Creation) ->
         {term,Int} when is_integer(Int) -> Int
     end.
 
-ei_accept(P, PortNo) ->
-    send_command(P, ei_accept, [PortNo]),
+ei_publish(P, PortNo) ->
+    send_command(P, ei_publish, [PortNo]),
+    case get_term(P) of
+        {term,{ListenFd, EpmdFd, _}} when ListenFd >= 0, EpmdFd >= 0 -> {ok, ListenFd};
+        {term,{_, _, Errno}} -> {error,Errno}
+    end.
+
+ei_accept(P, ListenFd) ->
+    send_command(P, ei_accept, [ListenFd]),
     case get_term(P) of
         {term,{Fd, _, Node}} when Fd >= 0 -> {ok, Fd, Node};
         {term,{_Fd, Errno, _Node}} -> {error,Errno}
diff --git a/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c b/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c
index 7b81ee5491..2355192cc2 100644
--- a/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c
+++ b/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c
@@ -43,6 +43,7 @@
 #include "ei_runner.h"
 
 static void cmd_ei_connect_init(char* buf, int len);
+static void cmd_ei_publish(char* buf, int len);
 static void cmd_ei_accept(char* buf, int len);
 static void cmd_ei_receive(char* buf, int len);
 static void cmd_ei_unpublish(char* buf, int len);
@@ -58,6 +59,7 @@ static struct {
     void (*func)(char* buf, int len);
 } commands[] = {
     "ei_connect_init",  3, cmd_ei_connect_init,
+    "ei_publish", 	1, cmd_ei_publish,
     "ei_accept", 	1, cmd_ei_accept,
     "ei_receive",  	1, cmd_ei_receive,
     "ei_unpublish",     0, cmd_ei_unpublish
@@ -149,11 +151,10 @@ static int my_listen(int port)
     return listen_fd;
 }
 
-static void cmd_ei_accept(char* buf, int len)
+static void cmd_ei_publish(char* buf, int len)
 {
     int index = 0;
     int listen, r;
-    ErlConnect conn;
     long port;
     ei_x_buff x;
     int i;
@@ -170,6 +171,29 @@ static void cmd_ei_accept(char* buf, int len)
 #ifdef VXWORKS
     save_fd(i);
 #endif
+    /* send listen-fd, result and errno */
+    ei_x_new_with_version(&x);
+    ei_x_encode_tuple_header(&x, 3);
+    ei_x_encode_long(&x, listen);
+    ei_x_encode_long(&x, i);
+    ei_x_encode_long(&x, erl_errno);
+    send_bin_term(&x);
+    ei_x_free(&x);
+}
+
+static void cmd_ei_accept(char* buf, int len)
+{
+    int index = 0;
+    int r;
+    ErlConnect conn;
+    long listen;
+    ei_x_buff x;
+    int i;
+
+    /* get port */
+    if (ei_decode_long(buf, &index, &listen) < 0)
+	fail("expected int (listen fd)");
+
     r = ei_accept(&ec, listen, &conn);
 #ifdef VXWORKS
     save_fd(r);
diff --git a/lib/jinterface/test/nc_SUITE.erl b/lib/jinterface/test/nc_SUITE.erl
index 81944f71d4..daa17195ee 100644
--- a/lib/jinterface/test/nc_SUITE.erl
+++ b/lib/jinterface/test/nc_SUITE.erl
@@ -651,11 +651,10 @@ do_echo(DataList, Config, OutTrans, InTrans, ExtraArgs)
 echo_loop([D|Ds], Echoer, OutTrans, InTrans, TermAcc) ->
     OutMsg = OutTrans(D),
     Echoer ! OutMsg,
-    io:format("echo_server ~p: ~p ! ~P~n", [self(),Echoer,OutMsg,10]),
+    %%io:format("echo_server ~p: ~p ! ~P~n", [self(),Echoer,OutMsg,10]),
     receive
 	Reply ->
-	    io:format("echo_server ~p: receive ~P~n",
-		      [self(),Reply,10]),
+	    %%io:format("echo_server ~p: receive ~P~n", [self(),Reply,10]),
 	    InTrans(Echoer, D, Reply)
     end,
     Term = case OutMsg of
@@ -670,11 +669,10 @@ echo_loop(Cont, Echoer, OutTrans, InTrans, TermAcc)
   when is_function(Cont, 0) ->
     check_terms(Echoer, TermAcc),
     OutMsg = Echoer ! {self(),undefined,hash_clear},
-    io:format("echo_server ~p: ~p ! ~P~n", [self(),Echoer,OutMsg,10]),
+    %%io:format("echo_server ~p: ~p ! ~P~n", [self(),Echoer,OutMsg,10]),
     receive
 	{Echoer,hash_cleared,hash_clear}=Reply ->
-	    io:format("echo_server ~p: receive ~P~n",
-		      [self(),Reply,10]),
+	    %%io:format("echo_server ~p: receive ~P~n", [self(),Reply,10]),
 	    ok;
 	Other ->
 	    io:format("echo_server_terms unexpected ~p: receive ~P~n",
@@ -686,11 +684,10 @@ echo_loop(Cont, Echoer, OutTrans, InTrans, TermAcc)
 check_terms(Echoer, [Term | Rest]) ->
     OutMsg = {self(),Term,hash_lookup},
     Echoer ! OutMsg,
-    io:format("check_terms ~p: ~p ! ~P~n", [self(),Echoer,OutMsg,10]),
+    %%io:format("check_terms ~p: ~p ! ~P~n", [self(),Echoer,OutMsg,10]),
     receive
 	{Echoer,true,hash_lookup} = ReplyMsg ->
-	    io:format("check_terms ~p: receive ~P~n",
-		      [self(),ReplyMsg,10]),
+	    %%io:format("check_terms ~p: receive ~P~n", [self(),ReplyMsg,10]),
 	    check_terms(Echoer, Rest);
 	Other ->
 	    io:format("check_terms unexpected ~p: receive ~P~n",
diff --git a/lib/kernel/src/dist_util.erl b/lib/kernel/src/dist_util.erl
index 08bd5946cd..fb9f7fd7eb 100644
--- a/lib/kernel/src/dist_util.erl
+++ b/lib/kernel/src/dist_util.erl
@@ -250,7 +250,8 @@ check_dflags(#hs_data{other_node = Node,
                       require_flags = RequiredFlags} = HSData) ->
     Mandatory = ((?DFLAG_EXTENDED_REFERENCES
                       bor ?DFLAG_EXTENDED_PIDS_PORTS
-                      bor ?DFLAG_UTF8_ATOMS)
+                      bor ?DFLAG_UTF8_ATOMS
+                      bor ?DFLAG_NEW_FUN_TAGS)
                      bor RequiredFlags),
     Missing = check_mandatory(0, ?DFLAGS_ALL, Mandatory,
                               OtherFlags, []),
diff --git a/lib/kernel/src/net_kernel.erl b/lib/kernel/src/net_kernel.erl
index f36b4f1e6a..cdb10a7b12 100644
--- a/lib/kernel/src/net_kernel.erl
+++ b/lib/kernel/src/net_kernel.erl
@@ -70,8 +70,8 @@
 	 protocol_childspecs/0,
 	 epmd_module/0]).
 
--export([connect/1, disconnect/1, hidden_connect/1, passive_cnct/1]).
--export([hidden_connect_node/1]). %% explicit connect
+-export([disconnect/1, passive_cnct/1]).
+-export([hidden_connect_node/1]).
 -export([set_net_ticktime/1, set_net_ticktime/2, get_net_ticktime/0]).
 
 -export([node_info/1, node_info/2, nodes_info/0,
@@ -122,6 +122,7 @@
 
 -record(connection, {
 		     node,          %% remote node name
+                     conn_id,       %% Connection identity
 		     state,         %% pending | up | up_pending
 		     owner,         %% owner pid
 	             pending_owner, %% possible new owner
@@ -247,14 +248,15 @@ ticktime_res(A)      when is_atom(A)             -> A.
 
 %% Called though BIF's
 
-connect(Node) ->               do_connect(Node, normal, false).
 %%% Long timeout if blocked (== barred), only affects nodes with
 %%% {dist_auto_connect, once} set.
-passive_cnct(Node) ->              do_connect(Node, normal, true).
-disconnect(Node) ->            request({disconnect, Node}).
+passive_cnct(Node) ->
+    case request({passive_cnct, Node}) of
+        ignored -> false;
+        Other -> Other
+    end.
 
-%% connect but not seen
-hidden_connect(Node) ->        do_connect(Node, hidden, false).
+disconnect(Node) ->            request({disconnect, Node}).
 
 %% Should this node publish itself on Node?
 publish_on_node(Node) when is_atom(Node) ->
@@ -272,67 +274,30 @@ connect_node(Node) when is_atom(Node) ->
 hidden_connect_node(Node) when is_atom(Node) ->
     request({connect, hidden, Node}).
 
-do_connect(Node, Type, WaitForBarred) -> %% Type = normal | hidden
-    case catch ets:lookup(sys_dist, Node) of
-	{'EXIT', _} ->
-	    ?connect_failure(Node,{table_missing, sys_dist}),
-	    false;
-	[#barred_connection{}] ->
-	    case WaitForBarred of
-		false ->
-		    false;
-		true ->
-		    Pid = spawn(?MODULE,passive_connect_monitor,[self(),Node]),
-		    receive
-			{Pid, true} ->
-			    %%io:format("Net Kernel: barred connection (~p) "
-			    %%          "connected from other end.~n",[Node]),
-			    true;
-			{Pid, false} ->
-			    ?connect_failure(Node,{barred_connection,
-						   ets:lookup(sys_dist, Node)}),
-			    %%io:format("Net Kernel: barred connection (~p) "
-			    %%      "- failure.~n",[Node]),
-			    false
-		    end
-	    end;
-	Else ->
-	    case application:get_env(kernel, dist_auto_connect) of
-		{ok, never} ->
-		    ?connect_failure(Node,{dist_auto_connect,never}),
-		    false;
-		% This might happen due to connection close
-		% not beeing propagated to user space yet.
-		% Save the day by just not connecting...
-		{ok, once} when Else =/= [],
-				(hd(Else))#connection.state =:= up ->
-		    ?connect_failure(Node,{barred_connection,
-				ets:lookup(sys_dist, Node)}),
-		    false;
-		_ ->
-		    request({connect, Type, Node})
-	    end
-    end.
 
-passive_connect_monitor(Parent, Node) ->
+passive_connect_monitor(From, Node) ->
     ok = monitor_nodes(true,[{node_type,all}]),
-    case lists:member(Node,nodes([connected])) of
-	true ->
-	    ok = monitor_nodes(false,[{node_type,all}]),
-	    Parent ! {self(),true};
-	_ ->
-	    Ref = make_ref(),
-	    Tref = erlang:send_after(connecttime(),self(),Ref),
-	    receive
-		Ref ->
-		    ok = monitor_nodes(false,[{node_type,all}]),
-		    Parent ! {self(), false};
-		{nodeup,Node,_} ->
-		    ok = monitor_nodes(false,[{node_type,all}]),
-		    _ = erlang:cancel_timer(Tref),
-		    Parent ! {self(),true}
-	    end
-    end.
+    Reply = case lists:member(Node,nodes([connected])) of
+                true ->
+                    io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]),
+                    true;
+                _ ->
+                    receive
+                        {nodeup,Node,_} ->
+                            io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]),
+                            true
+                    after connecttime() ->
+                            io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]),
+                            false
+                    end
+            end,
+    ok = monitor_nodes(false,[{node_type,all}]),
+    io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]),
+    {Pid, Tag} = From,
+    io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]),
+    erlang:send(Pid, {Tag, Reply}),
+    io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]).
+
 
 %% If the net_kernel isn't running we ignore all requests to the
 %% kernel, thus basically accepting them :-)
@@ -394,40 +359,135 @@ init({Name, LongOrShortNames, TickT, CleanHalt}) ->
     end.
 
 
+do_auto_connect(Type, Node, ConnId, WaitForBarred, From, State) ->
+    ConnLookup = ets:lookup(sys_dist, Node),
+
+    case ConnLookup of
+        [#barred_connection{}] ->
+            case WaitForBarred of
+                false ->
+                    {reply, false, State};
+                true ->
+                    spawn(?MODULE,passive_connect_monitor,[From,Node]),
+                    {noreply, State}
+            end;
+
+        [#connection{conn_id=ConnId, state = up}] ->
+            {reply, true, State};
+        [#connection{conn_id=ConnId, waiting=Waiting}=Conn] ->
+            case From of
+                noreply -> ok;
+                _ -> ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]})
+            end,
+            {noreply, State};
+
+        _ ->
+            case application:get_env(kernel, dist_auto_connect) of
+                {ok, never} ->
+                    ?connect_failure(Node,{dist_auto_connect,never}),
+                    {reply, false, State};
+
+                %% This might happen due to connection close
+                %% not beeing propagated to user space yet.
+                %% Save the day by just not connecting...
+                {ok, once} when ConnLookup =/= [],
+                                (hd(ConnLookup))#connection.state =:= up ->
+                    ?connect_failure(Node,{barred_connection,
+                                           ets:lookup(sys_dist, Node)}),
+                    {reply, false, State};
+                _ ->
+                    case setup(ConnLookup, Node,ConnId,Type,From,State) of
+                        {ok, SetupPid} ->
+                            Owners = [{SetupPid, Node} | State#state.conn_owners],
+                            {noreply,State#state{conn_owners=Owners}};
+                        _Error  ->
+                            ?connect_failure(Node, {setup_call, failed, _Error}),
+                            {reply, false, State}
+                    end
+            end
+    end.
+
+
+do_explicit_connect([#connection{conn_id = ConnId, state = up}], _, _, ConnId, _From, State) ->
+    {reply, true, State};
+do_explicit_connect([#connection{conn_id = ConnId}=Conn], _, _, ConnId, From, State)
+  when Conn#connection.state =:= pending;
+       Conn#connection.state =:= up_pending ->
+    Waiting = Conn#connection.waiting,
+    ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}),    
+    {noreply, State};
+do_explicit_connect([#barred_connection{}], Type, Node, ConnId, From , State) ->
+    %% Barred connection only affects auto_connect, ignore it.
+    do_explicit_connect([], Type, Node, ConnId, From , State);
+do_explicit_connect(ConnLookup, Type, Node, ConnId, From , State) ->
+    case setup(ConnLookup, Node,ConnId,Type,From,State) of
+        {ok, SetupPid} ->
+            Owners = [{SetupPid, Node} | State#state.conn_owners],
+            {noreply,State#state{conn_owners=Owners}};
+        _Error ->
+            ?connect_failure(Node, {setup_call, failed, _Error}),
+            {reply, false, State}
+    end.
+
+-define(ERTS_DIST_CON_ID_MASK, 16#ffffff).  % also in external.h
+
+verify_new_conn_id([], {Nr,_DHandle})
+  when (Nr band (bnot ?ERTS_DIST_CON_ID_MASK)) =:= 0 ->
+    true;
+verify_new_conn_id([#connection{conn_id = {Old,_}}], {New,_})
+  when New =:= ((Old+1) band ?ERTS_DIST_CON_ID_MASK) ->
+    true;
+verify_new_conn_id(_, _) ->
+    false.
+                                                              
+
+
 %% ------------------------------------------------------------
 %% handle_call.
 %% ------------------------------------------------------------
 
 %%
-%% Set up a connection to Node.
-%% The response is delayed until the connection is up and
-%% running.
+%% Passive auto-connect to Node.
+%% The response is delayed until the connection is up and running.
 %%
-handle_call({connect, _, Node}, From, State) when Node =:= node() ->
+handle_call({passive_cnct, Node}, From, State) when Node =:= node() ->
+    async_reply({reply, true, State}, From);
+handle_call({passive_cnct, Node}, From, State) ->
+    verbose({passive_cnct, Node}, 1, State),
+    Type = normal,
+    WaitForBarred = true,
+    R = case (catch erts_internal:new_connection(Node)) of
+            {Nr,_DHandle}=ConnId when is_integer(Nr) ->
+                do_auto_connect(Type, Node, ConnId, WaitForBarred, From, State);
+
+            _Error ->
+                error_logger:error_msg("~n** Cannot get connection id for node ~w~n",
+                                       [Node]),
+                {reply, false, State}
+        end,
+
+    return_call(R, From);
+
+%%
+%% Explicit connect
+%% The response is delayed until the connection is up and running.
+%%
+handle_call({connect, _, Node, _, _}, From, State) when Node =:= node() ->
     async_reply({reply, true, State}, From);
 handle_call({connect, Type, Node}, From, State) ->
     verbose({connect, Type, Node}, 1, State),
-    case ets:lookup(sys_dist, Node) of
-	[Conn] when Conn#connection.state =:= up ->
-	    async_reply({reply, true, State}, From);
-	[Conn] when Conn#connection.state =:= pending ->
-	    Waiting = Conn#connection.waiting,
-	    ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}),
-	    {noreply, State};
-	[Conn] when Conn#connection.state =:= up_pending ->
-	    Waiting = Conn#connection.waiting,
-	    ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}),
-	    {noreply, State};
-	_ ->
-	    case setup(Node,Type,From,State) of
-		{ok, SetupPid} ->
-		    Owners = [{SetupPid, Node} | State#state.conn_owners],
-		    {noreply,State#state{conn_owners=Owners}};
-		_Error  ->
-		    ?connect_failure(Node, {setup_call, failed, _Error}),
-		    async_reply({reply, false, State}, From)
-	    end
-    end;
+    ConnLookup = ets:lookup(sys_dist, Node),
+    R = case (catch erts_internal:new_connection(Node)) of
+            {Nr,_DHandle}=ConnId when is_integer(Nr) ->
+                do_explicit_connect(ConnLookup, Type, Node, ConnId, From, State);
+                    
+            _Error ->
+                error_logger:error_msg("~n** Cannot get connection id for node ~w~n",
+                                       [Node]),
+                {reply, false, State}                    
+        end,
+    return_call(R, From);
+            
 
 %%
 %% Close the connection to Node.
@@ -634,6 +694,26 @@ terminate(_Reason, State) ->
 %% ------------------------------------------------------------
 
 %%
+%% Asynchronous auto connect request
+%%
+handle_info({auto_connect,Node, Nr, DHandle}, State) ->
+    verbose({auto_connect, Node, Nr, DHandle}, 1, State),
+    ConnId = {Nr, DHandle},
+    NewState =
+        case do_auto_connect(normal, Node, ConnId, false, noreply, State) of
+            {noreply, S} ->           %% Pending connection
+                S;
+
+            {reply, true, S} ->  %% Already connected
+                S;
+
+            {reply, false, S} -> %% Connection refused
+                erts_internal:abort_connection(Node, ConnId),
+                S
+        end,
+    {noreply, NewState};
+
+%%
 %% accept a new connection.
 %%
 handle_info({accept,AcceptPid,Socket,Family,Proto}, State) ->
@@ -713,14 +793,23 @@ handle_info({AcceptPid, {accept_pending,MyNode,Node,Address,Type}}, State) ->
 	    AcceptPid ! {self(), {accept_pending, already_pending}},
 	    {noreply, State};
 	_ ->
-	    ets:insert(sys_dist, #connection{node = Node,
-					     state = pending,
-					     owner = AcceptPid,
-					     address = Address,
-					     type = Type}),
-	    AcceptPid ! {self(),{accept_pending,ok}},
-	    Owners = [{AcceptPid,Node} | State#state.conn_owners],
-	    {noreply, State#state{conn_owners = Owners}}
+            case (catch erts_internal:new_connection(Node)) of
+                {Nr,_DHandle}=ConnId when is_integer(Nr) ->
+                    ets:insert(sys_dist, #connection{node = Node,
+                                                     conn_id = ConnId,
+                                                     state = pending,
+                                                     owner = AcceptPid,
+                                                     address = Address,
+                                                     type = Type}),
+                    AcceptPid ! {self(),{accept_pending,ok}},
+                    Owners = [{AcceptPid,Node} | State#state.conn_owners],
+                    {noreply, State#state{conn_owners = Owners}};
+
+                _ ->
+                    error_logger:error_msg("~n** Cannot get connection id for node ~w~n",
+                                           [Node]),
+                    AcceptPid ! {self(),{accept_pending,nok_pending}}
+            end
     end;
 
 handle_info({SetupPid, {is_pending, Node}}, State) ->
@@ -906,6 +995,7 @@ pending_nodedown(Conn, Node, Type, State) ->
     % Don't bar connections that have never been alive
     %mark_sys_dist_nodedown(Node),
     % - instead just delete the node:
+    erts_internal:abort_connection(Node, Conn#connection.conn_id),
     ets:delete(sys_dist, Node),
     reply_waiting(Node,Conn#connection.waiting, false),
     case Type of
@@ -920,7 +1010,9 @@ up_pending_nodedown(Conn, Node, _Reason, _Type, State) ->
     AcceptPid = Conn#connection.pending_owner,
     Owners = State#state.conn_owners,
     Pend = lists:keydelete(AcceptPid, 1, State#state.pend_owners),
+    erts_internal:abort_connection(Node, Conn#connection.conn_id),
     Conn1 = Conn#connection { owner = AcceptPid,
+                              conn_id = erts_internal:new_connection(Node),
 			      pending_owner = undefined,
 			      state = pending },
     ets:insert(sys_dist, Conn1),
@@ -928,15 +1020,16 @@ up_pending_nodedown(Conn, Node, _Reason, _Type, State) ->
     State#state{conn_owners = [{AcceptPid,Node}|Owners], pend_owners = Pend}.
 
 
-up_nodedown(_Conn, Node, _Reason, Type, State) ->
-    mark_sys_dist_nodedown(Node),
+up_nodedown(Conn, Node, _Reason, Type, State) ->
+    mark_sys_dist_nodedown(Conn, Node),
     case Type of
 	normal -> ?nodedown(Node, State);
 	_ -> ok
     end,
     State.
 
-mark_sys_dist_nodedown(Node) ->
+mark_sys_dist_nodedown(Conn, Node) ->
+    erts_internal:abort_connection(Node, Conn#connection.conn_id),
     case application:get_env(kernel, dist_auto_connect) of
 	{ok, once} ->
 	    ets:insert(sys_dist, #barred_connection{node = Node});
@@ -1179,15 +1272,8 @@ spawn_func(_,{From,Tag},M,F,A,Gleader) ->
 %% Set up connection to a new node.
 %% -----------------------------------------------------------
 
-setup(Node,Type,From,State) ->
-    Allowed = State#state.allowed,
-    case lists:member(Node, Allowed) of
-	false when Allowed =/= [] ->
-	    error_msg("** Connection attempt with "
-		      "disallowed node ~w ** ~n", [Node]),
-	    {error, bad_node};
-	_ ->
-	    case select_mod(Node, State#state.listen) of
+setup(ConnLookup, Node,ConnId,Type,From,State) ->
+    case setup_check(ConnLookup, Node, ConnId, State) of
 		{ok, L} ->
 		    Mod = L#listen.module,
 		    LAddr = L#listen.address,
@@ -1200,18 +1286,45 @@ setup(Node,Type,From,State) ->
 		    Addr = LAddr#net_address {
 					      address = undefined,
 					      host = undefined },
+                    Waiting = case From of
+                                  noreply -> [];
+                                  _ -> [From]
+                              end,
 		    ets:insert(sys_dist, #connection{node = Node,
+                                                     conn_id = ConnId,
 						     state = pending,
 						     owner = Pid,
-						     waiting = [From],
+						     waiting = Waiting,
 						     address = Addr,
 						     type = normal}),
 		    {ok, Pid};
 		Error ->
 		    Error
-	    end
     end.
 
+setup_check(ConnLookup, Node, ConnId, State) ->
+    Allowed = State#state.allowed,    
+    case lists:member(Node, Allowed) of
+	false when Allowed =/= [] ->
+	    error_msg("** Connection attempt with "
+		      "disallowed node ~w ** ~n", [Node]),
+	    {error, bad_node};
+       _ ->
+            case verify_new_conn_id(ConnLookup, ConnId) of
+                false ->
+                    error_msg("** Connection attempt to ~w with "
+                              "bad connection id ~w ** ~n", [Node, ConnId]),
+                    {error, bad_conn_id};
+                true ->
+                    case select_mod(Node, State#state.listen) of
+                        {ok, _L}=OK -> OK;
+                        Error -> Error
+                    end
+            end
+    end.                    
+
+    
+
 %%
 %% Find a module that is willing to handle connection setup to Node
 %%
@@ -1652,6 +1765,11 @@ verbose(_, _, _) ->
 getnode(P) when is_pid(P) -> node(P);
 getnode(P) -> P.
 
+return_call({noreply, _State}=R, _From) ->
+    R;
+return_call(R, From) ->
+    async_reply(R, From).
+
 async_reply({reply, Msg, State}, From) ->
     async_gen_server_reply(From, Msg),
     {noreply, State}.
diff --git a/lib/kernel/test/erl_distribution_wb_SUITE.erl b/lib/kernel/test/erl_distribution_wb_SUITE.erl
index 258ed4f88c..1145d30e5e 100644
--- a/lib/kernel/test/erl_distribution_wb_SUITE.erl
+++ b/lib/kernel/test/erl_distribution_wb_SUITE.erl
@@ -61,9 +61,11 @@
 %% From R9 and forward extended references is compulsory
 %% From R10 and forward extended pids and ports are compulsory
 %% From R20 and forward UTF8 atoms are compulsory
+%% From R21 and forward NEW_FUN_TAGS is compulsory (no more tuple fallback {fun, ...})
 -define(COMPULSORY_DFLAGS, (?DFLAG_EXTENDED_REFERENCES bor
                             ?DFLAG_EXTENDED_PIDS_PORTS bor
-                            ?DFLAG_UTF8_ATOMS)).
+                            ?DFLAG_UTF8_ATOMS bor
+                            ?DFLAG_NEW_FUN_TAGS)).
 
 -define(PASS_THROUGH, $p).
 
@@ -682,7 +684,7 @@ recv_message(Socket) ->
 	    <<_:Siz/binary,B2/binary>> = B1,
 	    Message = case (catch binary_to_term(B2)) of
 			  {'EXIT', _} ->
-			      could_not_digest_message;
+			      {could_not_digest_message,B2};
 			  Other ->
 			      Other
 		      end,
diff --git a/lib/tools/emacs/erlang.el b/lib/tools/emacs/erlang.el
index 429188b028..d9efadf64a 100644
--- a/lib/tools/emacs/erlang.el
+++ b/lib/tools/emacs/erlang.el
@@ -895,8 +895,6 @@ resulting regexp is surrounded by \\_< and \\_>."
       "decode_packet"
       "delay_trap"
       "delete_element"
-      "dexit"
-      "dgroup_leader"
       "display"
       "display_nl"
       "display_string"
@@ -905,11 +903,8 @@ resulting regexp is surrounded by \\_< and \\_>."
       "dist_ctrl_get_data_notification"
       "dist_ctrl_input_handler"
       "dist_ctrl_put_data"
-      "dist_exit"
-      "dlink"
       "dmonitor_node"
       "dmonitor_p"
-      "dsend"
       "dt_append_vm_tag_data"
       "dt_get_tag"
       "dt_get_tag_data"
@@ -917,7 +912,6 @@ resulting regexp is surrounded by \\_< and \\_>."
       "dt_put_tag"
       "dt_restore_tag"
       "dt_spread_tag"
-      "dunlink"
       "convert_time_unit"
       "external_size"
       "finish_after_on_load"