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

9 files changed, 118 insertions, 37 deletions

diff --git a/erts/doc/src/erlang.xml b/erts/doc/src/erlang.xml
index d205c24350..daf3002ec7 100644
--- a/erts/doc/src/erlang.xml
+++ b/erts/doc/src/erlang.xml
@@ -3361,25 +3361,6 @@ RealSystem = system + MissedSystem</code>
               monitored process resides). </p></item>
         </taglist>
 
-        <p>If an attempt is made to monitor a process on an older node
-          (where remote process monitoring is not implemented or
-          where remote process monitoring by registered name is not
-          implemented), the call fails with <c>badarg</c>.</p>
-        <note>
-          <p>The format of the <c>'DOWN'</c> message changed in ERTS
-            5.2 (Erlang/OTP R9B) for monitoring
-            <em>by registered name</em>. Element <c>Object</c> of
-            the <c>'DOWN'</c> message could in earlier versions
-            sometimes be the process identifier of the monitored process and sometimes
-            be the registered name. Now element <c>Object</c> is
-            always a tuple consisting of the registered name and
-            the node name. Processes on new nodes (ERTS 5.2
-            or higher versions) always get <c>'DOWN'</c> messages on
-            the new format even if they are monitoring processes on old
-            nodes. Processes on old nodes always get <c>'DOWN'</c>
-            messages on the old format.</p>
-        </note>
-
         <taglist>
         <tag>Monitoring a <marker id="monitor_process"/><c>process</c></tag>
         <item>
@@ -3387,7 +3368,19 @@ RealSystem = system + MissedSystem</code>
             process identified by <c><anno>Item</anno></c>, which can be a
             <c>pid()</c> (local or remote), an atom <c>RegisteredName</c> or
             a tuple <c>{RegisteredName, Node}</c> for a registered process,
-            located elsewhere.</p>
+          located elsewhere.</p>
+
+        <note><p>Before ERTS 10.0 (OTP 21.0), monitoring a process could fail with
+	  <c>badarg</c> if the monitored process resided on a primitive node
+	  (such as erl_interface or jinterface), where remote process monitoring
+	  is not implemented.</p>
+	  <p>Now, such a call to <c>monitor</c> will instead succeed and a
+	  monitor is created. But the monitor will only supervise the
+	  connection. That is, a <c>{'DOWN', _, process, _, noconnection}</c> is
+	  the only message that may be received, as the primitive node have no
+	  way of reporting the status of the monitored process.</p>
+	</note>
+
         </item>
 
         <tag>Monitoring a <marker id="monitor_port"/><c>port</c></tag>
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 8593ad867a..7ff7462bf6 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -879,6 +879,13 @@ erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
     DeclareTmpHeapNoproc(ctl_heap,6);
     int res;
 
+    if (~dsdp->dep->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
+        /*
+         * Receiver does not support DOP_MONITOR_P_EXIT (see dsig_send_monitor)
+         */
+        return ERTS_DSIG_SEND_OK;
+    }
+
     UseTmpHeapNoproc(6);
 
     ctl = TUPLE5(&ctl_heap[0], make_small(DOP_MONITOR_P_EXIT),
@@ -906,6 +913,16 @@ erts_dsig_send_monitor(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
     DeclareTmpHeapNoproc(ctl_heap,5);
     int res;
 
+    if (~dsdp->dep->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
+        /*
+         * Receiver does not support DOP_MONITOR_P.
+         * Just avoid sending it and by doing that reduce this monitor
+         * to only supervise the connection. This will work for simple c-nodes
+         * with a 1-to-1 relation between "Erlang process" and OS-process.
+         */
+        return ERTS_DSIG_SEND_OK;
+    }
+
     UseTmpHeapNoproc(5);
     ctl = TUPLE4(&ctl_heap[0],
 		 make_small(DOP_MONITOR_P),
@@ -928,6 +945,13 @@ erts_dsig_send_demonitor(ErtsDSigData *dsdp, Eterm watcher,
     DeclareTmpHeapNoproc(ctl_heap,5);
     int res;
 
+    if (~dsdp->dep->flags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)) {
+        /*
+         * Receiver does not support DOP_DEMONITOR_P (see dsig_send_monitor)
+         */
+        return ERTS_DSIG_SEND_OK;
+    }
+
     UseTmpHeapNoproc(5);
     ctl = TUPLE4(&ctl_heap[0],
 		 make_small(DOP_DEMONITOR_P),
@@ -2337,10 +2361,10 @@ erts_dist_command(Port *prt, int initial_reds)
 	    ASSERT(ob);
 	    do {
 		reds = erts_encode_ext_dist_header_finalize(ob, dep, flags, reds);
-                if (reds >= 0) {
-                    last_finalized  = ob;
-                    ob = ob->next;
-                }
+                if (reds < 0)
+                    break;
+                last_finalized  = ob;
+                ob = ob->next;
 	    } while (ob);
             if (last_finalized) {
                 /*
@@ -2381,7 +2405,7 @@ erts_dist_command(Port *prt, int initial_reds)
                 break;
             }
 	    ASSERT(&oq.first->data[0] <= oq.first->extp
-		   && oq.first->extp < oq.first->ext_endp);
+		   && oq.first->extp <= oq.first->ext_endp);
 	    size = (*send)(prt, oq.first);
             erts_atomic64_inc_nob(&dep->out);
 	    esdp->io.out += (Uint64) size;
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index a96e39be89..ea4697815f 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -56,7 +56,9 @@
 /* Additional optimistic flags when encoding toward pending connection */
 #define DFLAG_DIST_HOPEFULLY (DFLAG_NO_MAGIC                    \
                               | DFLAG_EXPORT_PTR_TAG            \
-                              | DFLAG_BIT_BINARIES)
+                              | DFLAG_BIT_BINARIES              \
+                              | DFLAG_DIST_MONITOR              \
+                              | DFLAG_DIST_MONITOR_NAME)
 
 /* All flags that should be enabled when term_to_binary/1 is used. */
 #define TERM_TO_BINARY_DFLAGS (DFLAG_EXTENDED_REFERENCES	\
diff --git a/erts/emulator/beam/erl_binary.h b/erts/emulator/beam/erl_binary.h
index 05007e864e..46653a8580 100644
--- a/erts/emulator/beam/erl_binary.h
+++ b/erts/emulator/beam/erl_binary.h
@@ -363,8 +363,6 @@ erts_free_aligned_binary_bytes(byte* buf)
 #  define CHICKEN_PAD (sizeof(void*) - 1)
 #endif
 
-/* Caller must initialize 'refc'
-*/
 ERTS_GLB_INLINE Binary *
 erts_bin_drv_alloc_fnf(Uint size)
 {
@@ -383,8 +381,6 @@ erts_bin_drv_alloc_fnf(Uint size)
     return res;
 }
 
-/* Caller must initialize 'refc'
-*/
 ERTS_GLB_INLINE Binary *
 erts_bin_drv_alloc(Uint size)
 {
@@ -401,9 +397,6 @@ erts_bin_drv_alloc(Uint size)
     return res;
 }
 
-
-/* Caller must initialize 'refc'
-*/
 ERTS_GLB_INLINE Binary *
 erts_bin_nrml_alloc(Uint size)
 {
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index f2e6399ad7..9cc5e71afa 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -383,6 +383,21 @@ Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf* ob,
          */
         ASSERT(ep[0] == SMALL_TUPLE_EXT || ep[0] == LARGE_TUPLE_EXT);
 
+        if (~dflags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME)
+            && ep[0] == SMALL_TUPLE_EXT
+            && ep[1] == 4
+            && ep[2] == SMALL_INTEGER_EXT
+            && (ep[3] == DOP_MONITOR_P ||
+                ep[3] == DOP_MONITOR_P_EXIT ||
+                ep[3] == DOP_DEMONITOR_P)) {
+            /*
+             * Receiver does not support process monitoring.
+             * Suppress monitor control msg (see erts_dsig_send_monitor)
+             * by converting it to an empty (tick) packet.
+             */
+            ob->ext_endp = ob->extp;
+            return reds;
+        }
         if (~dflags & (DFLAG_BIT_BINARIES | DFLAG_EXPORT_PTR_TAG
                        | DFLAG_DIST_HDR_ATOM_CACHE)) {
             reds = transcode_dist_obuf(ob, dep, dflags, reds);
diff --git a/lib/erl_interface/test/ei_accept_SUITE.erl b/lib/erl_interface/test/ei_accept_SUITE.erl
index 16eb0901ef..6676e73e21 100644
--- a/lib/erl_interface/test/ei_accept_SUITE.erl
+++ b/lib/erl_interface/test/ei_accept_SUITE.erl
@@ -25,7 +25,8 @@
 -include("ei_accept_SUITE_data/ei_accept_test_cases.hrl").
 
 -export([all/0, suite/0,
-         ei_accept/1, ei_threaded_accept/1]).
+         ei_accept/1, ei_threaded_accept/1,
+         monitor_ei_process/1]).
 
 -import(runner, [get_term/1,send_term/2]).
 
@@ -34,7 +35,8 @@ suite() ->
      {timetrap, {seconds, 30}}].
 
 all() -> 
-    [ei_accept, ei_threaded_accept].
+    [ei_accept, ei_threaded_accept,
+     monitor_ei_process].
 
 ei_accept(Config) when is_list(Config) ->
     P = runner:start(?interpret),
@@ -58,6 +60,7 @@ ei_accept(Config) when is_list(Config) ->
     Port = 6543,
     {ok, ListenFd} = ei_publish(P, Port),
     {any, EINode} ! TermToSend,
+
     {ok, Fd, _Node} = ei_accept(P, ListenFd),
     Got1 = ei_receive(P, Fd),
 
@@ -85,6 +88,41 @@ ei_threaded_accept(Config) when is_list(Config) ->
     [receive I -> ok end || I <- lists:seq(0, N-1) ],
     ok.
 
+
+%% Test erlang:monitor toward erl_interface "processes"
+monitor_ei_process(Config) when is_list(Config) ->
+    P = runner:start(?interpret),
+    0 = ei_connect_init(P, 42, erlang:get_cookie(), 0),
+
+    Myname = hd(tl(string:tokens(atom_to_list(node()), "@"))),
+    io:format("Myname ~p ~n",  [Myname]),
+    EINode = list_to_atom("c42@"++Myname),
+    io:format("EINode ~p ~n",  [EINode]),
+
+    Self = self(),
+    Port = 6543,
+    {ok, ListenFd} = ei_publish(P, Port),
+    MRef1 = erlang:monitor(process, {any, EINode}),
+    {any, EINode} ! hello,
+
+    {ok, Fd, _Node} = ei_accept(P, ListenFd),
+    hello = ei_receive(P, Fd),
+
+    %% Again, now on an established connection.
+    MRef2 = erlang:monitor(process, {any, EINode}),
+    {any, EINode} ! hello,
+    hello = ei_receive(P, Fd),
+
+    ok = receive M -> M after 0 -> ok end,
+
+    runner:finish(P),
+
+    [{'DOWN', MRef1, process, {any, EINode}, noconnection},
+     {'DOWN', MRef2, process, {any, EINode}, noconnection}
+    ] = lists:sort(flush(2, 1000)),
+
+    ok.
+
 waitfornode(String,0) ->
     io:format("~s never published itself.~n",[String]),
     false;
@@ -156,3 +194,12 @@ ei_receive(P, Fd) ->
 
 send_command(P, Name, Args) ->
     runner:send_term(P, {Name,list_to_tuple(Args)}).
+
+flush(0, Timeout) ->
+    flush(1, Timeout div 10);
+flush(Expected, Timeout) ->
+    receive M ->
+            [M | flush(Expected-1, Timeout)]
+    after Timeout ->
+            []
+    end.
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 2355192cc2..2b62d7c3a4 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
@@ -224,7 +224,7 @@ static void cmd_ei_receive(char* buf, int len)
 	if (got == ERL_TICK)
 	    continue;
 	if (got == ERL_ERROR)
-	    fail("ei_xreceive_msg");
+	    fail1("ei_xreceive_msg, got==%d", got);
 	break;
     }
     index = 1;
diff --git a/lib/erl_interface/test/runner.erl b/lib/erl_interface/test/runner.erl
index 1084eec2a3..cbdeb3c8a9 100644
--- a/lib/erl_interface/test/runner.erl
+++ b/lib/erl_interface/test/runner.erl
@@ -67,7 +67,12 @@ start({Prog, Tc}) when is_list(Prog), is_integer(Tc) ->
 
 finish(Port) when is_port(Port) ->
     send_eot(Port),
-    recv_eot(Port).
+    ok = recv_eot(Port),
+    0 = receive
+            {Port,{exit_status,Status}} ->
+                Status
+        end,
+    ok.
 
 %% Sends an Erlang term to a C program.
 
diff --git a/lib/kernel/test/erl_distribution_wb_SUITE.erl b/lib/kernel/test/erl_distribution_wb_SUITE.erl
index 1145d30e5e..8256444bdc 100644
--- a/lib/kernel/test/erl_distribution_wb_SUITE.erl
+++ b/lib/kernel/test/erl_distribution_wb_SUITE.erl
@@ -677,6 +677,8 @@ build_rex_message(Cookie,OurName) ->
 %% Receive a distribution message    
 recv_message(Socket) ->
     case gen_tcp:recv(Socket, 0) of
+        {ok,[]} ->
+            recv_message(Socket);  %% a tick, ignore
 	{ok,Data} ->
 	    B0 = list_to_binary(Data),
 	    <<?PASS_THROUGH, B1/binary>> = B0,