Erlang Distribution Protocol Benchmark

dist_perf.erl

%
% $ erl -name [email protected]
%
% $ erl -name [email protected]
% ([email protected])2> dist_perf:go('[email protected]').
%  started
%  Sent:953MB/s, Hist:{6,0,0,12,0,0,0,0,0,0,0} RHist:{0,0,0,0,0,0,0,0,0,97,0}
%  Sent:961MB/s, Hist:{1,0,0,12,0,0,0,0,0,0,0} RHist:{0,0,0,0,0,0,0,0,0,155,0}
%  Sent:953MB/s, Hist:{6,0,0,12,0,0,0,0,0,0,0} RHist:{0,0,0,0,0,0,0,0,0,202,0}
%
%
% Adapted from original version by Lukas Larsson from
% https://groups.google.com/g/erlang-programming/c/sCrWDju0N8c
%

-module(dist_perf).

-export([go/1, go/2, go/3, stop/0, gc/0, stats/1]).

go(RemoteNode) ->
    go(RemoteNode, 100).
go(RemoteNode, N) ->
    Payload = [<<0:100000/unit:8>> || _ <- lists:seq(1, 10)],
    go(RemoteNode, N, Payload).
go(RemoteNode, N, Payload) when is_atom(RemoteNode), is_integer(N) ->
    stop(),
    StatsPid = spawn_link(?MODULE, stats, [RemoteNode]),
    Pids = [spawn(RemoteNode, fun
        F() ->
            receive {From, Msg} ->
                From ! erlang:iolist_size(Msg),
                F()
            end
        end
    ) || _<- lists:seq(1, N)],
    Pinger = fun
        F(Pid) ->
            Msg = {self(), Payload},
            SendRes = erlang:send(Pid, Msg, [nosuspend]),
            case SendRes of
                ok ->
                    receive _IOListSize -> F(Pid) end;
                nosuspend ->
                    erlang:send(Pid, Msg),
                    receive _IOListSize -> F(Pid) end
            end
    end,
    Ctx = {StatsPid, [{Pid, spawn(fun() -> Pinger(Pid) end)} || Pid <- Pids]},
    put(dist_perf, Ctx),
    started.

stop() ->
    case erase(dist_perf) of
        undefined ->
            ok;
        {StatsPid, Pids} ->
            [begin exit(Pid1, kill), exit(Pid2, kill) end || {Pid1, Pid2} <- Pids],
            unlink(StatsPid),
            exit(StatsPid, kill)
    end.

gc() ->
    [erlang:garbage_collect(Pid) || Pid<-processes()],
    ok.

stats(RemoteNode) ->
    {_, {output, T0Out}} = erlang:statistics(io),
    timer:sleep(5000),
    {_, {output, T1Out}} = erlang:statistics(io),
    SentMBs = round((T1Out - T0Out) / 1024 / 1024 / 5),
    io:format("Sent:~pMB/s, Hist:~p RHist:~p ~n",[SentMBs, hist(), rhist(RemoteNode)]),
    stats(RemoteNode).

hist() ->
    Opts = #{histogram_width => 11},
    Allocs = instrument:allocations(Opts),
    case Allocs of
        {ok, {_, _, #{tcp_inet := #{drv_binary := H}}}} -> H;
        _Other -> unavailable
    end.

rhist(RemoteNode) ->
    Opts = #{histogram_width => 11},
    Allocs = rpc:call(RemoteNode, instrument, allocations, [Opts]),
    case Allocs of
        {ok, {_, _, #{tcp_inet := #{drv_binary := H}}}} -> H;
        _Other -> unavailable
    end.