erts: Implement fragmentation of distrubution messages

3 files changed, 210 insertions, 26 deletions

diff --git a/erts/emulator/test/distribution_SUITE.erl b/erts/emulator/test/distribution_SUITE.erl
index 885c66331c..5e268c748f 100644
--- a/erts/emulator/test/distribution_SUITE.erl
+++ b/erts/emulator/test/distribution_SUITE.erl
@@ -62,7 +62,8 @@
          bad_dist_ext_control/1,
          bad_dist_ext_connection_id/1,
          bad_dist_ext_size/1,
-	 start_epmd_false/1, epmd_module/1]).
+	 start_epmd_false/1, epmd_module/1,
+         bad_dist_fragments/1]).
 
 %% Internal exports.
 -export([sender/3, receiver2/2, dummy_waiter/0, dead_process/0,
@@ -90,7 +91,7 @@ all() ->
      dist_parallel_send, atom_roundtrip, unicode_atom_roundtrip,
      atom_roundtrip_r16b,
      contended_atom_cache_entry, contended_unicode_atom_cache_entry,
-     bad_dist_structure, {group, bad_dist_ext},
+     {group, bad_dist}, {group, bad_dist_ext},
      start_epmd_false, epmd_module].
 
 groups() ->
@@ -100,6 +101,8 @@ groups() ->
      {trap_bif, [], [trap_bif_1, trap_bif_2, trap_bif_3]},
      {dist_auto_connect, [],
       [dist_auto_connect_never, dist_auto_connect_once]},
+     {bad_dist, [],
+      [bad_dist_structure, bad_dist_fragments]},
      {bad_dist_ext, [],
       [bad_dist_ext_receive, bad_dist_ext_process_info,
        bad_dist_ext_size,
@@ -1382,6 +1385,15 @@ get_conflicting_unicode_atoms(CIX, N) ->
 -define(DOP_DEMONITOR_P,	20).
 -define(DOP_MONITOR_P_EXIT,	21).
 
+-define(DOP_SEND_SENDER, 22).
+-define(DOP_SEND_SENDER_TT, 23).
+
+-define(DOP_PAYLOAD_EXIT, 24).
+-define(DOP_PAYLOAD_EXIT_TT, 25).
+-define(DOP_PAYLOAD_EXIT2, 26).
+-define(DOP_PAYLOAD_EXIT2_TT, 27).
+-define(DOP_PAYLOAD_MONITOR_P_EXIT, 28).
+
 start_monitor(Offender,P) ->
     Parent = self(),
     Q = spawn(Offender,
@@ -1515,7 +1527,145 @@ bad_dist_structure(Config) when is_list(Config) ->
     stop_node(Victim),
     ok.
 
+%% Test various dist fragmentation errors
+bad_dist_fragments(Config) when is_list(Config) ->
+    ct:timetrap({seconds, 15}),
+
+    {ok, Offender} = start_node(bad_dist_fragment_offender),
+    {ok, Victim} = start_node(bad_dist_fragment_victim),
 
+    Msg = iolist_to_binary(dmsg_ext(lists:duplicate(255,255))),
+
+    start_node_monitors([Offender,Victim]),
+    Parent = self(),
+    P = spawn(Victim,
+              fun () ->
+                      process_flag(trap_exit,true),
+                      Parent ! {self(), started},
+                      receive check_msgs -> ok end,
+                      bad_dist_struct_check_msgs([one,
+                                                  two]),
+                      Parent ! {self(), messages_checked},
+                      receive done -> ok end
+              end),
+    receive {P, started} -> ok end,
+    pong = rpc:call(Victim, net_adm, ping, [Offender]),
+    verify_up(Offender, Victim),
+    true = lists:member(Offender, rpc:call(Victim, erlang, nodes, [])),
+    start_monitor(Offender,P),
+    P ! one,
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{frg, 1, binary:part(Msg, 10,byte_size(Msg)-10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{hdr, 3, binary:part(Msg, 0,10)},
+                       {frg, 1, binary:part(Msg, 10,byte_size(Msg)-10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{hdr, 3, binary:part(Msg, 0,10)},
+                       {hdr, 3, binary:part(Msg, 0,10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P,broken},3,
+                      [{hdr, 1, binary:part(Msg, 10,byte_size(Msg)-10)}]),
+
+    start_monitor(Offender,P),
+    send_bad_fragments(Offender, Victim, P,{?DOP_SEND,?COOKIE,P},3,
+                      [{hdr, 3, binary:part(Msg, 10,byte_size(Msg)-10)},
+                       close]),
+
+    start_monitor(Offender,P),
+    ExitVictim = spawn(Victim, fun() -> receive ok -> ok end end),
+    send_bad_fragments(Offender, Victim, P,{?DOP_PAYLOAD_EXIT,P,ExitVictim},2,
+                      [{hdr, 1, [131]}]),
+
+    start_monitor(Offender,P),
+    Exit2Victim = spawn(Victim, fun() -> receive ok -> ok end end),
+    send_bad_fragments(Offender, Victim, P,{?DOP_PAYLOAD_EXIT2,P,ExitVictim},2,
+                      [{hdr, 1, [132]}]),
+
+    start_monitor(Offender,P),
+    DownVictim = spawn(Victim, fun() -> receive ok -> ok end end),
+    DownRef = erlang:monitor(process, DownVictim),
+    send_bad_fragments(Offender, Victim, P,{?DOP_PAYLOAD_MONITOR_P_EXIT,P,DownVictim,DownRef},2,
+                      [{hdr, 1, [133]}]),
+
+    P ! two,
+    P ! check_msgs,
+    receive
+        {P, messages_checked} -> ok
+    after 5000 ->
+              exit(victim_is_dead)
+    end,
+
+    {message_queue_len, 0}
+    = rpc:call(Victim, erlang, process_info, [P, message_queue_len]),
+
+    unlink(P),
+    P ! done,
+    stop_node(Offender),
+    stop_node(Victim),
+    ok.
+
+dmsg_frag_hdr(Frag) ->
+    dmsg_frag_hdr(erlang:phash2(self()), Frag).
+dmsg_frag_hdr(Seq, Frag) ->
+    [131, $E, uint64_be(Seq), uint64_be(Frag), 0].
+
+dmsg_frag(Frag) ->
+    dmsg_frag(erlang:phash2(self()), Frag).
+dmsg_frag(Seq, Frag) ->
+    [131, $F, uint64_be(Seq), uint64_be(Frag)].
+
+send_bad_fragments(Offender,VictimNode,Victim,Ctrl,WhereToPutSelf,Fragments) ->
+    Parent = self(),
+    Done = make_ref(),
+    ct:pal("Send: ~p",[Fragments]),
+    spawn_link(Offender,
+          fun () ->
+                  Node = node(Victim),
+                  pong = net_adm:ping(Node),
+                  erlang:monitor_node(Node, true),
+                  DCtrl = dctrl(Node),
+                  Ctrl1 = case WhereToPutSelf of
+                             0 ->
+                                 Ctrl;
+                             N when N > 0 ->
+                                 setelement(N,Ctrl,self())
+                         end,
+
+                  FragData = [case Type of
+                                  hdr ->
+                                      [dmsg_frag_hdr(FragId),
+                                       dmsg_ext(Ctrl1), FragPayload];
+                                  frg ->
+                                      [dmsg_frag(FragId), FragPayload]
+                              end || {Type, FragId, FragPayload} <- Fragments],
+
+                  receive {nodedown, Node} -> exit("premature nodedown")
+                  after 10 -> ok
+                  end,
+
+                  [ dctrl_send(DCtrl, D) || D <- FragData ],
+                  [ erlang:port_close(DCtrl) || close <- Fragments],
+
+                  receive {nodedown, Node} -> ok
+                  after 5000 -> exit("missing nodedown")
+                  end,
+                  Parent ! {FragData,Done}
+          end),
+    receive
+        {WhatSent,Done} ->
+            io:format("Offender sent ~p~n",[WhatSent]),
+            verify_nc(VictimNode),
+            ok
+    after 7000 ->
+              exit(unable_to_send)
+    end.
 
 bad_dist_ext_receive(Config) when is_list(Config) ->
     {ok, Offender} = start_node(bad_dist_ext_receive_offender),
@@ -2124,8 +2274,25 @@ start_node(Config, Args, Rel) when is_list(Config), is_list(Rel) ->
     start_node(Name, Args, Rel).
 
 stop_node(Node) ->
+    verify_nc(Node),
     test_server:stop_node(Node).
 
+verify_nc(Node) ->
+    P = self(),
+    Ref = make_ref(),
+    spawn(Node,
+          fun() ->
+                  R = erts_test_utils:check_node_dist(fun(E) -> E end),
+                  P ! {Ref, R}
+          end),
+    receive
+        {Ref, ok} ->
+            ok;
+        {Ref, Error} ->
+            ct:log("~s",[Error]),
+            ct:fail(failed_nc_refc_check)
+    end.
+
 freeze_node(Node, MS) ->
     Own = 300,
     DoingIt = make_ref(),
@@ -2485,6 +2652,17 @@ mk_ref({NodeNameExt, Creation}, Numbers) when is_integer(Creation),
             exit({unexpected_binary_to_term_result, Other})
     end.
 
+uint64_be(Uint) when is_integer(Uint), 0 =< Uint, Uint < 1 bsl 64 ->
+    [(Uint bsr 56) band 16#ff,
+     (Uint bsr 48) band 16#ff,
+     (Uint bsr 40) band 16#ff,
+     (Uint bsr 32) band 16#ff,
+     (Uint bsr 24) band 16#ff,
+     (Uint bsr 16) band 16#ff,
+     (Uint bsr 8) band 16#ff,
+     Uint band 16#ff];
+uint64_be(Uint) ->
+    exit({badarg, uint64_be, [Uint]}).
 
 uint32_be(Uint) when is_integer(Uint), 0 =< Uint, Uint < 1 bsl 32 ->
     [(Uint bsr 24) band 16#ff,
diff --git a/erts/emulator/test/erts_test_utils.erl b/erts/emulator/test/erts_test_utils.erl
index ac2f2435be..0c3ef3e0fc 100644
--- a/erts/emulator/test/erts_test_utils.erl
+++ b/erts/emulator/test/erts_test_utils.erl
@@ -27,6 +27,7 @@
 -export([mk_ext_pid/3,
          mk_ext_port/2,
          mk_ext_ref/2,
+         available_internal_state/1,
          check_node_dist/0, check_node_dist/1, check_node_dist/3]).
 
 
@@ -157,6 +158,21 @@ mk_ext_ref({NodeName, Creation}, Numbers) when is_list(NodeName),
     end.
 
 
+available_internal_state(Bool) when Bool == true; Bool == false ->
+    case {Bool,
+          (catch erts_debug:get_internal_state(available_internal_state))} of
+        {true, true} ->
+            true;
+        {false, true} ->
+            erts_debug:set_internal_state(available_internal_state, false),
+            true;
+        {true, _} ->
+            erts_debug:set_internal_state(available_internal_state, true),
+            false;
+        {false, _} ->
+            false
+    end.
+
 
 %%
 %% Check reference counters for node- and dist entries.
@@ -168,16 +184,21 @@ check_node_dist() ->
                     end).
 
 check_node_dist(Fail) ->
+    AIS = available_internal_state(true),
+    [erlang:garbage_collect(P) || P <- erlang:processes()],
     {{node_references, NodeRefs},
      {dist_references, DistRefs}} =
         erts_debug:get_internal_state(node_and_dist_references),
-    check_node_dist(Fail, NodeRefs, DistRefs).
-
-
+    R = check_node_dist(Fail, NodeRefs, DistRefs),
+    available_internal_state(AIS),
+    R.
 
 check_node_dist(Fail, NodeRefs, DistRefs) ->
-    check_nd_refc({node(),erlang:system_info(creation)},                  
-                  NodeRefs, DistRefs, Fail).
+    AIS = available_internal_state(true),
+    R = check_nd_refc({node(),erlang:system_info(creation)},
+                  NodeRefs, DistRefs, Fail),
+    available_internal_state(AIS),
+    R.
 
 
 check_nd_refc({ThisNodeName, ThisCreation}, NodeRefs, DistRefs, Fail) ->
diff --git a/erts/emulator/test/node_container_SUITE.erl b/erts/emulator/test/node_container_SUITE.erl
index b3d8f9584d..ca5c93f535 100644
--- a/erts/emulator/test/node_container_SUITE.erl
+++ b/erts/emulator/test/node_container_SUITE.erl
@@ -71,25 +71,10 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     erts_debug:set_internal_state(available_internal_state, true),
     erts_debug:set_internal_state(node_tab_delayed_delete, -1), %% restore original value
-    available_internal_state(false).
-
-available_internal_state(Bool) when Bool == true; Bool == false ->
-    case {Bool,
-          (catch erts_debug:get_internal_state(available_internal_state))} of
-        {true, true} ->
-            true;
-        {false, true} ->
-            erts_debug:set_internal_state(available_internal_state, false),
-            true;
-        {true, _} ->
-            erts_debug:set_internal_state(available_internal_state, true),
-            false;
-        {false, _} ->
-            false
-    end.
+    erts_test_util:available_internal_state(false).
 
 init_per_testcase(_Case, Config) when is_list(Config) ->
-    available_internal_state(true),
+    erts_test_util:available_internal_state(true),
     Config.
 
 end_per_testcase(_Case, Config) when is_list(Config) ->
@@ -928,9 +913,9 @@ id(X) ->
 -define(ND_REFS, erts_debug:get_internal_state(node_and_dist_references)).
 
 node_container_refc_check(Node) when is_atom(Node) ->
-    AIS = available_internal_state(true),
+    AIS = erts_test_util:available_internal_state(true),
     nc_refc_check(Node),
-    available_internal_state(AIS).
+    erts_test_util:available_internal_state(AIS).
 
 nc_refc_check(Node) when is_atom(Node) ->
     Ref = make_ref(),