erts: Yield correctly when iterating over distr exit messages

Before this fix the process would continue to process
more distributed down or exit messages until it
ran out of reductions instead of being suspended
immediately.

3 files changed, 43 insertions, 18 deletions

diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 2b45d2d353..76eec96372 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -12095,6 +12095,7 @@ erts_proc_exit_handle_dist_monitor(ErtsMonitor *mon, void *vctxt, Sint reds)
     Eterm watched;
     Uint watcher_sz, ref_sz;
     ErtsHeapFactory factory;
+    Sint reds_consumed = 0;
 
     ASSERT(erts_monitor_is_target(mon) && mon->type == ERTS_MON_TYPE_DIST_PROC);
 
@@ -12139,11 +12140,13 @@ erts_proc_exit_handle_dist_monitor(ErtsMonitor *mon, void *vctxt, Sint reds)
                                      watched,
                                      ref,
                                      reason);
+        reds_consumed = reds - (ctx.reds / TERM_TO_BINARY_LOOP_FACTOR);
         switch (code) {
         case ERTS_DSIG_SEND_CONTINUE:
         case ERTS_DSIG_SEND_YIELD:
             erts_set_gc_state(c_p, 0);
             ctxt->dist_state = erts_dsend_export_trap_context(c_p, &ctx);
+            reds_consumed = reds; /* force yield */
             break;
         case ERTS_DSIG_SEND_OK:
             break;
@@ -12163,7 +12166,7 @@ erts_proc_exit_handle_dist_monitor(ErtsMonitor *mon, void *vctxt, Sint reds)
         erts_monitor_release(mon);
     else
         erts_monitor_release_both(mdp);
-    return reds - (ctx.reds / TERM_TO_BINARY_LOOP_FACTOR);
+    return reds_consumed;
 }
 
 int
@@ -12350,6 +12353,7 @@ erts_proc_exit_handle_dist_link(ErtsLink *lnk, void *vctxt, Sint reds)
     ErtsLink *dlnk;
     ErtsLinkData *ldp = NULL;
     ErtsHeapFactory factory;
+    Sint reds_consumed = 0;
 
     ASSERT(lnk->type == ERTS_LNK_TYPE_DIST_PROC);
     dlnk = erts_link_to_other(lnk, &ldp);
@@ -12386,11 +12390,13 @@ erts_proc_exit_handle_dist_link(ErtsLink *lnk, void *vctxt, Sint reds)
                                       item,
                                       reason,
                                       SEQ_TRACE_TOKEN(c_p));
+        reds_consumed = reds - (ctx.reds / TERM_TO_BINARY_LOOP_FACTOR);
         switch (code) {
         case ERTS_DSIG_SEND_YIELD:
         case ERTS_DSIG_SEND_CONTINUE:
             erts_set_gc_state(c_p, 0);
             ctxt->dist_state = erts_dsend_export_trap_context(c_p, &ctx);
+            reds_consumed = reds; /* force yield */
             break;
         case ERTS_DSIG_SEND_OK:
             break;
@@ -12410,7 +12416,7 @@ erts_proc_exit_handle_dist_link(ErtsLink *lnk, void *vctxt, Sint reds)
         erts_link_release_both(ldp);
     else if (lnk)
         erts_link_release(lnk);
-    return reds - (ctx.reds / TERM_TO_BINARY_LOOP_FACTOR);
+    return reds_consumed;
 }
 
 int
diff --git a/erts/emulator/test/distribution_SUITE.erl b/erts/emulator/test/distribution_SUITE.erl
index d1d1e3840e..84e95e6d26 100644
--- a/erts/emulator/test/distribution_SUITE.erl
+++ b/erts/emulator/test/distribution_SUITE.erl
@@ -77,7 +77,7 @@
          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]).
+         dist_evil_parallel_receiver/0, make_busy/2]).
 
 %% epmd_module exports
 -export([start_link/0, register_node/2, register_node/3, port_please/2, address_please/3]).
diff --git a/erts/emulator/test/dump_SUITE.erl b/erts/emulator/test/dump_SUITE.erl
index 2440833992..9f8ac42fa9 100644
--- a/erts/emulator/test/dump_SUITE.erl
+++ b/erts/emulator/test/dump_SUITE.erl
@@ -137,24 +137,43 @@ exiting_dump(Config) when is_list(Config) ->
 free_dump(Config) when is_list(Config) ->
     Dump = filename:join(proplists:get_value(priv_dir, Config),"signal_abort.dump"),
 
-    {ok, Node} = start_node(Config),
-
-    Self = self(),
+    {ok, NodeA} = start_node(Config),
+    {ok, NodeB} = start_node(Config),
 
-    Pid = spawn_link(Node,
-                     fun() ->
-                             Self ! ready,
-                             receive
-                                 ok ->
-                                     spawn(fun() -> timer:sleep(5), erlang:halt("dump") end),
-                                     exit(lists:duplicate(1000,1000))
-                             end
-                     end),
 
-    true = rpc:call(Node, os, putenv, ["ERL_CRASH_DUMP",Dump]),
+    Self = self(),
 
-    [erlang:monitor(process, Pid) || _ <- lists:seq(1,10000)],
-    receive ready -> unlink(Pid), Pid ! ok end,
+    PidA = spawn_link(
+             NodeA,
+             fun() ->
+                     Self ! ready,
+                     receive
+                         ok ->
+                             spawn(fun() ->
+                                           erlang:system_monitor(self(), [busy_dist_port]),
+                                           timer:sleep(5),
+                                           receive
+                                               M ->
+                                                   io:format("~p",[M]),
+                                                   erlang:halt("dump")
+                                           end
+                                   end),
+                             exit(lists:duplicate(1000000,100))
+                     end
+             end),
+
+    spawn_link(NodeB,
+               fun() ->
+                       [erlang:monitor(process, PidA) || _ <- lists:seq(1,10000)],
+                       Self ! done,
+                       receive _ -> ok end
+               end),
+
+    receive done -> ok end,
+    true = rpc:call(NodeA, os, putenv, ["ERL_CRASH_DUMP",Dump]),
+    ct:pal("~p",[rpc:call(NodeA, distribution_SUITE, make_busy, [NodeB, 1000])]),
+
+    receive ready -> unlink(PidA), PidA ! ok end,
 
     {ok, Bin} = get_dump_when_done(Dump),