anchor.diff:  -export(
      [setup/1,
       roundtrip/1,
  +    throughput_0/1,
  +    throughput_64/1,
       throughput_1024/1,
       throughput_4096/1,
       throughput_16384/1,
  
positive.code:             setup_loop(A, B, Time + T, N - 1)
      end.
  
  
  %%----------------
  %% Roundtrip speed
  
negative.code: throughput(A, B, Prefix, HA, HB, Packets, Size) ->
      [] = ssl_apply(HA, erlang, nodes, []),
      [] = ssl_apply(HB, erlang, nodes, []),
      #{time := Time,
        dist_stats := DistStats,
        client_msacc_stats := ClientMsaccStats,
        client_prof := ClientProf,
        server_msacc_stats := ServerMsaccStats,
        server_prof := ServerProf} =
          ssl_apply(HA, fun () -> throughput_runner(A, B, Packets, Size) end),
      [B] = ssl_apply(HA, erlang, nodes, []),
      [A] = ssl_apply(HB, erlang, nodes, []),
      ClientMsaccStats =:= undefined orelse
          msacc:print(ClientMsaccStats),
      io:format("DistStats: ~p~n", [DistStats]),
      Overhead =
          50 % Distribution protocol headers (empirical) (TLS+=54)
          + byte_size(erlang:term_to_binary([0|<<>>])), % Benchmark overhead
      Bytes = Packets * (Size + Overhead),
      io:format("~w bytes, ~.4g s~n", [Bytes,Time/1000000]),
      ClientMsaccStats =:= undefined orelse
          io:format(
            "Sender core usage ratio: ~.4g ns/byte~n",
            [msacc:stats(system_runtime, ClientMsaccStats)*1000/Bytes]),
      ServerMsaccStats =:= undefined orelse
          begin
              io:format(
                "Receiver core usage ratio: ~.4g ns/byte~n",
                [msacc:stats(system_runtime, ServerMsaccStats)*1000/Bytes]),
              msacc:print(ServerMsaccStats)
          end,
      io:format("******* ClientProf:~n", []), prof_print(ClientProf),
      io:format("******* ServerProf:~n", []), prof_print(ServerProf),
      Speed = round((Bytes * 1000000) / (1024 * Time)),
      report(Prefix++" Throughput_"++integer_to_list(Size), Speed, "kB/s").
  
  %% Runs on node A and spawns a server on node B
  throughput_runner(A, B, Rounds, Size) ->
      Payload = payload(Size),
      [A] = rpc:call(B, erlang, nodes, []),
      ClientPid = self(),
      ServerPid =
          erlang:spawn(
            B,
            fun () -> throughput_server(ClientPid, Rounds) end),
      ServerMon = erlang:monitor(process, ServerPid),
      msacc:available() andalso
          begin
              msacc:stop(),
              msacc:reset(),
              msacc:start(),
              ok
          end,
      prof_start(),
      {Time,ServerMsaccStats,ServerProf} =
          throughput_client(ServerPid, ServerMon, Payload, Rounds),
      prof_stop(),
      ClientMsaccStats =
          case msacc:available() of
              true ->
                  MStats = msacc:stats(),
                  msacc:stop(),
                  MStats;
              false ->
                  undefined
          end,
      ClientProf = prof_end(),
      [{_Node,Socket}] = dig_dist_node_sockets(),
      DistStats = inet:getstat(Socket),
      #{time => microseconds(Time),
        dist_stats => DistStats,
        client_msacc_stats => ClientMsaccStats,
        client_prof => ClientProf,
        server_msacc_stats => ServerMsaccStats,
        server_prof => ServerProf}.
  
  dig_dist_node_sockets() ->
      [case DistCtrl of
           {_Node,Socket} = NodeSocket when is_port(Socket) ->
               NodeSocket;
           {Node,DistCtrlPid} when is_pid(DistCtrlPid) ->
               [{links,DistCtrlLinks}] = process_info(DistCtrlPid, [links]),
               case [S || S <- DistCtrlLinks, is_port(S)] of
                   [Socket] ->
                       {Node,Socket};
                   [] ->
                       [{monitors,[{process,DistSenderPid}]}] =
                           process_info(DistCtrlPid, [monitors]),
                       [{links,DistSenderLinks}] =
                           process_info(DistSenderPid, [links]),
                       [Socket] = [S || S <- DistSenderLinks, is_port(S)],
                       {Node,Socket}
               end
       end || DistCtrl <- erlang:system_info(dist_ctrl)].
  
  
  throughput_server(Pid, N) ->
      msacc:available() andalso
          begin
              msacc:stop(),
              msacc:reset(),
              msacc:start(),
              ok
          end,
      prof_start(),
      throughput_server_loop(Pid, N).
  
  throughput_server_loop(_Pid, 0) ->
      prof_stop(),
      MsaccStats =
          case msacc:available() of
              true ->
                  msacc:stop(),
                  MStats = msacc:stats(),
                  msacc:reset(),
                  MStats;
              false ->
                  undefined
          end,
      Prof = prof_end(),
      exit({ok,MsaccStats,Prof});
  throughput_server_loop(Pid, N) ->
      receive
          {Pid, N, _} ->
              throughput_server_loop(Pid, N-1)
      end.
  
  throughput_client(Pid, Mon, Payload, N) ->
      throughput_client_loop(Pid, Mon, Payload, N, start_time()).
  
  throughput_client_loop(_Pid, Mon, _Payload, 0, StartTime) ->
      receive
          {'DOWN', Mon, _, _, {ok,MsaccStats,Prof}} ->
              {elapsed_time(StartTime),MsaccStats,Prof};
          {'DOWN', Mon, _, _, Other} ->
              exit(Other)
      end;
  throughput_client_loop(Pid, Mon, Payload, N, StartTime) ->
      Pid ! {self(), N, Payload},
      throughput_client_loop(Pid, Mon, Payload, N - 1, StartTime).
  
  
  -define(prof, none). % none | cprof | eprof
  
  -if(?prof =:= cprof).
  prof_start() ->
      cprof:stop(),
      cprof:start(),
      ok.
  -elif(?prof =:= eprof).
  prof_start() ->
      {ok,_} = eprof:start(),
      profiling = eprof:start_profiling(processes()),
      ok.
  -elif(?prof =:= none).
  prof_start() ->
      ok.
  -endif.
  
  -if(?prof =:= cprof).
  prof_stop() ->
      cprof:pause(),
      ok.
  -elif(?prof =:= eprof).
  prof_stop() ->
      _ = eprof:stop_profiling(),
      ok.
  -elif(?prof =:= none).
  prof_stop() ->
      ok.
  -endif.
  
  -if(?prof =:= cprof).
  prof_end() ->
      Prof = cprof:analyse(),
      cprof:stop(),
      Prof.
  -elif(?prof =:= eprof).
  prof_end() ->
      eprof:dump_data().
  -elif(?prof =:= none).
  prof_end() ->
      [].
  -endif.
  
  -if(?prof =:= cprof).
  prof_print(Prof) ->
      io:format("~p.~n", [Prof]).
  -elif(?prof =:= eprof).
  prof_print(Dump) ->
      eprof:analyze(undefined, total, [], Dump).
  -elif(?prof =:= none).
  prof_print([]) ->
      ok.
  -endif.
  
  %%%-------------------------------------------------------------------
  %%% Test cases helpers
  