%% ==========================================================================================================
%% Syn - A global Process Registry and Process Group manager.
%%
%% The MIT License (MIT)
%%
%% Copyright (c) 2019-2021 Roberto Ostinelli <roberto@ostinelli.net> and Neato Robotics, Inc.
%%
%% Permission is hereby granted, free of charge, to any person obtaining a copy
%% of this software and associated documentation files (the "Software"), to deal
%% in the Software without restriction, including without limitation the rights
%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%% copies of the Software, and to permit persons to whom the Software is
%% furnished to do so, subject to the following conditions:
%%
%% The above copyright notice and this permission notice shall be included in
%% all copies or substantial portions of the Software.
%%
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
%% THE SOFTWARE.
%% ==========================================================================================================
-module(syn_benchmark).
%% API
-export([
start/0,
start_processes/1,
process_loop/0,
register_on_node/4,
unregister_on_node/4,
join_on_node/3,
leave_on_node/3,
wait_registry_propagation/1,
wait_pg_propagation/1
]).
-export([
start_profiling/1,
stop_profiling/1,
start_profiling_on_node/0,
stop_profiling_on_node/0
]).
%% macros
-define(TEST_SCOPE, scope_test).
-define(TEST_GROUP_NAME, <<"test-group">>).
%% ===================================================================
%% API
%% ===================================================================
%% example run: `PROCESS_COUNT=100000 WORKERS_PER_NODE=100 NODES_COUNT=2 make bench`
start() ->
%% init
ProcessCount = list_to_integer(os:getenv("PROCESS_COUNT", "100000")),
WorkersPerNode = list_to_integer(os:getenv("WORKERS_PER_NODE", "100")),
SlavesCount = list_to_integer(os:getenv("NODES_COUNT", "2")),
SkipRegistry = case os:getenv("SKIP_REGISTRY") of false -> false; _ -> true end,
SkipPG = case os:getenv("SKIP_PG") of false -> false; _ -> true end,
ProcessesPerNode = round(ProcessCount / SlavesCount),
io:format("====> Starting benchmark~n"),
io:format(" --> Nodes: ~w / ~w slave(s)~n", [SlavesCount + 1, SlavesCount]),
io:format(" --> Total processes: ~w (~w / slave node)~n", [ProcessCount, ProcessesPerNode]),
io:format(" --> Workers per node: ~w~n~n", [WorkersPerNode]),
%% start nodes
NodesInfo = lists:foldl(fun(I, Acc) ->
%% start slave
CountBin = integer_to_binary(I),
NodeShortName = list_to_atom(binary_to_list(<<"slave_", CountBin/binary>>)),
{ok, Node} = ct_slave:start(NodeShortName, [
{boot_timeout, 10},
{monitor_master, true}
]),
%% add code path
CodePath = code:get_path(),
true = rpc:call(Node, code, set_path, [CodePath]),
%% start syn
ok = rpc:call(Node, syn, start, []),
ok = rpc:call(Node, syn, add_node_to_scopes, [[?TEST_SCOPE]]),
%% gather data
FromName = (I - 1) * ProcessesPerNode + 1,
ToName = FromName + ProcessesPerNode - 1,
%% fold
[{Node, FromName, ToName} | Acc]
end, [], lists:seq(1, SlavesCount)),
%% start syn locally
ok = syn:start(),
ok = syn:add_node_to_scopes([?TEST_SCOPE]),
timer:sleep(1000),
CollectorPid = self(),
case SkipRegistry of
false ->
io:format("~n====> Starting REGISTRY benchmark~n~n"),
%% start processes
RegistryPidsMap = lists:foldl(fun({Node, _FromName, _ToName}, Acc) ->
Pids = rpc:call(Node, ?MODULE, start_processes, [ProcessesPerNode]),
maps:put(Node, Pids, Acc)
end, #{}, NodesInfo),
%% start registration
lists:foreach(fun({Node, FromName, _ToName}) ->
Pids = maps:get(Node, RegistryPidsMap),
rpc:cast(Node, ?MODULE, register_on_node, [CollectorPid, WorkersPerNode, FromName, Pids])
end, NodesInfo),
%% wait
RegRemoteNodesTimes = wait_from_all_remote_nodes(nodes(), []),
io:format("----> Remote registration times:~n"),
io:format(" --> MIN: ~p secs.~n", [lists:min(RegRemoteNodesTimes)]),
io:format(" --> MAX: ~p secs.~n", [lists:max(RegRemoteNodesTimes)]),
{RegPropagationTimeMs, _} = timer:tc(?MODULE, wait_registry_propagation, [ProcessCount]),
RegPropagationTime = RegPropagationTimeMs / 1000000,
io:format("----> Eventual additional time to propagate all to master: ~p secs.~n", [RegPropagationTime]),
%% sum
RegTakenTime = (lists:max(RegRemoteNodesTimes) + RegPropagationTime),
RegistrationRate = ProcessCount / RegTakenTime,
io:format("====> Registeration rate (with propagation): ~p/sec.~n~n", [RegistrationRate]),
%% start unregistration
lists:foreach(fun({Node, FromName, ToName}) ->
rpc:cast(Node, ?MODULE, unregister_on_node, [CollectorPid, WorkersPerNode, FromName, ToName])
end, NodesInfo),
%% wait
UnregRemoteNodesTimes = wait_from_all_remote_nodes(nodes(), []),
io:format("----> Remote unregistration times:~n"),
io:format(" --> MIN: ~p secs.~n", [lists:min(UnregRemoteNodesTimes)]),
io:format(" --> MAX: ~p secs.~n", [lists:max(UnregRemoteNodesTimes)]),
{UnregPropagationTimeMs, _} = timer:tc(?MODULE, wait_registry_propagation, [0]),
UnregPropagationTime = UnregPropagationTimeMs / 1000000,
io:format("----> Eventual additional time to propagate all to master: ~p secs.~n", [UnregPropagationTime]),
%% sum
UnregTakenTime = (lists:max(UnregRemoteNodesTimes) + UnregPropagationTime),
UnregistrationRate = ProcessCount / UnregTakenTime,
io:format("====> Unregisteration rate (with propagation): ~p/sec.~n~n", [UnregistrationRate]),
%% start re-registration
lists:foreach(fun({Node, FromName, _ToName}) ->
Pids = maps:get(Node, RegistryPidsMap),
rpc:cast(Node, ?MODULE, register_on_node, [CollectorPid, WorkersPerNode, FromName, Pids])
end, NodesInfo),
%% wait
ReRegRemoteNodesTimes = wait_from_all_remote_nodes(nodes(), []),
io:format("----> Remote re-registration times:~n"),
io:format(" --> MIN: ~p secs.~n", [lists:min(ReRegRemoteNodesTimes)]),
io:format(" --> MAX: ~p secs.~n", [lists:max(ReRegRemoteNodesTimes)]),
{ReRegPropagationTimeMs, _} = timer:tc(?MODULE, wait_registry_propagation, [ProcessCount]),
ReRegPropagationTime = ReRegPropagationTimeMs / 1000000,
io:format("----> Eventual additional time to propagate all to master: ~p secs.~n", [ReRegPropagationTime]),
%% sum
ReRegTakenTime = (lists:max(ReRegRemoteNodesTimes) + ReRegPropagationTime),
ReRegistrationRate = ProcessCount / ReRegTakenTime,
io:format("====> Re-registeration rate (with propagation): ~p/sec.~n~n", [ReRegistrationRate]),
%% kill all processes
maps:foreach(fun(_Node, Pids) ->
lists:foreach(fun(Pid) -> exit(Pid, kill) end, Pids)
end, RegistryPidsMap),
%% wait all unregistered
{RegKillPropagationTimeMs, _} = timer:tc(?MODULE, wait_registry_propagation, [0]),
RegKillPropagationTime = RegKillPropagationTimeMs / 1000000,
io:format("----> Time to propagate killed process to to master: ~p secs.~n", [RegKillPropagationTime]),
RegKillRate = ProcessCount / RegKillPropagationTime,
io:format("====> Unregistered after kill rate (with propagation): ~p/sec.~n~n", [RegKillRate]);
true ->
io:format("~n====> Skipping REGISTRY.~n~n")
end,
case SkipPG of
false ->
io:format("~n====> Starting PG benchmark~n~n"),
%% start processes
PgPidsMap = lists:foldl(fun({Node, _FromName, _ToName}, Acc) ->
Pids = rpc:call(Node, ?MODULE, start_processes, [ProcessesPerNode]),
maps:put(Node, Pids, Acc)
end, #{}, NodesInfo),
%% start joining
lists:foreach(fun({Node, _FromName, _ToName}) ->
Pids = maps:get(Node, PgPidsMap),
rpc:cast(Node, ?MODULE, join_on_node, [CollectorPid, WorkersPerNode, Pids])
end, NodesInfo),
%% wait
JoinRemoteNodesTimes = wait_from_all_remote_nodes(nodes(), []),
io:format("----> Remote join times:~n"),
io:format(" --> MIN: ~p secs.~n", [lists:min(JoinRemoteNodesTimes)]),
io:format(" --> MAX: ~p secs.~n", [lists:max(JoinRemoteNodesTimes)]),
{JoinPropagationTimeMs, _} = timer:tc(?MODULE, wait_pg_propagation, [ProcessCount]),
JoinPropagationTime = JoinPropagationTimeMs / 1000000,
io:format("----> Eventual additional time to propagate all to master: ~p secs.~n", [JoinPropagationTime]),
%% sum
JoinTakenTime = (lists:max(JoinRemoteNodesTimes) + JoinPropagationTime),
JoinRate = ProcessCount / JoinTakenTime,
io:format("====> Join rate (with propagation): ~p/sec.~n~n", [JoinRate]),
%% start leaving
lists:foreach(fun({Node, _FromName, _ToName}) ->
Pids = maps:get(Node, PgPidsMap),
rpc:cast(Node, ?MODULE, leave_on_node, [CollectorPid, WorkersPerNode, Pids])
end, NodesInfo),
%% wait
LeaveRemoteNodesTimes = wait_from_all_remote_nodes(nodes(), []),
io:format("----> Remote leave times:~n"),
io:format(" --> MIN: ~p secs.~n", [lists:min(LeaveRemoteNodesTimes)]),
io:format(" --> MAX: ~p secs.~n", [lists:max(LeaveRemoteNodesTimes)]),
{LeavePropagationTimeMs, _} = timer:tc(?MODULE, wait_pg_propagation, [0]),
LeavePropagationTime = LeavePropagationTimeMs / 1000000,
io:format("----> Eventual additional time to propagate all to master: ~p secs.~n", [LeavePropagationTime]),
%% sum
LeaveTakenTime = (lists:max(LeaveRemoteNodesTimes) + LeavePropagationTime),
LeaveRate = ProcessCount / LeaveTakenTime,
io:format("====> Leave rate (with propagation): ~p/sec.~n~n", [LeaveRate]),
%% start re-joining
lists:foreach(fun({Node, _FromName, _ToName}) ->
Pids = maps:get(Node, PgPidsMap),
rpc:cast(Node, ?MODULE, join_on_node, [CollectorPid, WorkersPerNode, Pids])
end, NodesInfo),
%% wait
ReJoinRemoteNodesTimes = wait_from_all_remote_nodes(nodes(), []),
io:format("----> Remote join times:~n"),
io:format(" --> MIN: ~p secs.~n", [lists:min(ReJoinRemoteNodesTimes)]),
io:format(" --> MAX: ~p secs.~n", [lists:max(ReJoinRemoteNodesTimes)]),
{ReJoinPropagationTimeMs, _} = timer:tc(?MODULE, wait_pg_propagation, [ProcessCount]),
ReJoinPropagationTime = ReJoinPropagationTimeMs / 1000000,
io:format("----> Eventual additional time to propagate all to master: ~p secs.~n", [ReJoinPropagationTime]),
%% sum
ReJoinTakenTime = (lists:max(ReJoinRemoteNodesTimes) + ReJoinPropagationTime),
ReJoinRate = ProcessCount / ReJoinTakenTime,
io:format("====> Re-join rate (with propagation): ~p/sec.~n~n", [ReJoinRate]),
%% kill all processes
maps:foreach(fun(_Node, Pids) ->
lists:foreach(fun(Pid) -> exit(Pid, kill) end, Pids)
end, PgPidsMap),
%% wait all unregistered
{PgKillPropagationTimeMs, _} = timer:tc(?MODULE, wait_pg_propagation, [0]),
PgKillPropagationTime = PgKillPropagationTimeMs / 1000000,
io:format("----> Time to propagate killed process to to master: ~p secs.~n", [PgKillPropagationTime]),
PgKillRate = ProcessCount / PgKillPropagationTime,
io:format("====> Left after kill rate (with propagation): ~p/sec.~n~n", [PgKillRate]);
true ->
io:format("~n====> Skipping PG.~n")
end,
%% stop node
init:stop().
register_on_node(CollectorPid, WorkersPerNode, FromName, Pids) ->
%% split pids in workers
PidsPerNode = round(length(Pids) / WorkersPerNode),
{WorkerInfo, []} = lists:foldl(fun(I, {WInfo, RPids}) ->
{WorkerPids, RestOfPids} = case I of
WorkersPerNode ->
%% last in the loop, get remaining pids
{RPids, []};
_ ->
%% get portion of pids
lists:split(PidsPerNode, RPids)
end,
WorkerFromName = FromName + (PidsPerNode * (I - 1)),
{[{WorkerFromName, WorkerPids} | WInfo], RestOfPids}
end, {[], Pids}, lists:seq(1, WorkersPerNode)),
%% spawn workers
ReplyPid = self(),
lists:foreach(fun({WorkerFromName, WorkerPids}) ->
spawn(fun() ->
StartAt = os:system_time(millisecond),
worker_register_on_node(WorkerFromName, WorkerPids),
Time = (os:system_time(millisecond) - StartAt) / 1000,
ReplyPid ! {done, Time}
end)
end, WorkerInfo),
%% wait
Time = wait_done_on_node(CollectorPid, 0, WorkersPerNode),
io:format("----> Registered on node ~p on ~p secs.~n", [node(), Time]).
worker_register_on_node(_Name, []) -> ok;
worker_register_on_node(Name, [Pid | PidsTail]) ->
ok = syn:register(?TEST_SCOPE, Name, Pid),
worker_register_on_node(Name + 1, PidsTail).
unregister_on_node(CollectorPid, WorkersPerNode, FromName, ToName) ->
%% split pids in workers
ProcessesPerNode = ToName - FromName + 1,
ProcessesPerWorker = round(ProcessesPerNode / WorkersPerNode),
WorkerInfo = lists:foldl(fun(I, Acc) ->
{WorkerFromName, WorkerToName} = case I of
WorkersPerNode ->
%% last in the loop
{FromName + (I - 1) * ProcessesPerWorker, ToName};
_ ->
{FromName + (I - 1) * ProcessesPerWorker, FromName + I * ProcessesPerWorker - 1}
end,
[{WorkerFromName, WorkerToName} | Acc]
end, [], lists:seq(1, WorkersPerNode)),
%% spawn workers
ReplyPid = self(),
lists:foreach(fun({WorkerFromName, WorkerToName}) ->
spawn(fun() ->
StartAt = os:system_time(millisecond),
worker_unregister_on_node(WorkerFromName, WorkerToName),
Time = (os:system_time(millisecond) - StartAt) / 1000,
ReplyPid ! {done, Time}
end)
end, WorkerInfo),
%% wait
Time = wait_done_on_node(CollectorPid, 0, WorkersPerNode),
io:format("----> Unregistered on node ~p on ~p secs.~n", [node(), Time]).
worker_unregister_on_node(FromName, ToName) when FromName > ToName -> ok;
worker_unregister_on_node(Name, ToName) ->
ok = syn:unregister(?TEST_SCOPE, Name),
worker_unregister_on_node(Name + 1, ToName).
join_on_node(CollectorPid, WorkersPerNode, Pids) ->
%% split pids in workers
PidsPerNode = round(length(Pids) / WorkersPerNode),
{PidsPerWorker, []} = lists:foldl(fun(I, {P, RPids}) ->
{WPids, RestOfPids} = case I of
WorkersPerNode ->
%% last in the loop, get remaining pids
{RPids, []};
_ ->
%% get portion of pids
lists:split(PidsPerNode, RPids)
end,
{[WPids | P], RestOfPids}
end, {[], Pids}, lists:seq(1, WorkersPerNode)),
%% spawn workers
ReplyPid = self(),
lists:foreach(fun(WorkerPids) ->
spawn(fun() ->
StartAt = os:system_time(millisecond),
worker_join_on_node(WorkerPids),
Time = (os:system_time(millisecond) - StartAt) / 1000,
ReplyPid ! {done, Time}
end)
end, PidsPerWorker),
%% wait
Time = wait_done_on_node(CollectorPid, 0, WorkersPerNode),
io:format("----> Joined on node ~p on ~p secs.~n", [node(), Time]).
worker_join_on_node([]) -> ok;
worker_join_on_node([Pid | PidsTail]) ->
ok = syn:join(?TEST_SCOPE, ?TEST_GROUP_NAME, Pid),
worker_join_on_node(PidsTail).
leave_on_node(CollectorPid, WorkersPerNode, Pids) ->
%% split pids in workers
PidsPerNode = round(length(Pids) / WorkersPerNode),
{PidsPerWorker, []} = lists:foldl(fun(I, {P, RPids}) ->
{WPids, RestOfPids} = case I of
WorkersPerNode ->
%% last in the loop, get remaining pids
{RPids, []};
_ ->
%% get portion of pids
lists:split(PidsPerNode, RPids)
end,
{[WPids | P], RestOfPids}
end, {[], Pids}, lists:seq(1, WorkersPerNode)),
%% spawn workers
ReplyPid = self(),
lists:foreach(fun(WorkerPids) ->
spawn(fun() ->
StartAt = os:system_time(millisecond),
worker_leave_on_node(WorkerPids),
Time = (os:system_time(millisecond) - StartAt) / 1000,
ReplyPid ! {done, Time}
end)
end, PidsPerWorker),
%% wait
Time = wait_done_on_node(CollectorPid, 0, WorkersPerNode),
io:format("----> Left on node ~p on ~p secs.~n", [node(), Time]).
worker_leave_on_node([]) -> ok;
worker_leave_on_node([Pid | PidsTail]) ->
ok = syn:leave(?TEST_SCOPE, ?TEST_GROUP_NAME, Pid),
worker_leave_on_node(PidsTail).
wait_done_on_node(CollectorPid, Time, 0) ->
CollectorPid ! {done, node(), Time},
Time;
wait_done_on_node(CollectorPid, Time, WorkersRemainingCount) ->
receive
{done, WorkerTime} ->
Time1 = lists:max([WorkerTime, Time]),
wait_done_on_node(CollectorPid, Time1, WorkersRemainingCount - 1)
end.
start_processes(Count) ->
start_processes(Count, []).
start_processes(0, Pids) ->
Pids;
start_processes(Count, Pids) ->
Pid = spawn(fun process_loop/0),
start_processes(Count - 1, [Pid | Pids]).
process_loop() ->
receive
_ -> ok
end.
wait_from_all_remote_nodes([], Times) -> Times;
wait_from_all_remote_nodes([RemoteNode | Tail], Times) ->
receive
{done, RemoteNode, Time} ->
wait_from_all_remote_nodes(Tail, [Time | Times])
end.
wait_registry_propagation(DesiredCount) ->
case syn:registry_count(?TEST_SCOPE) of
DesiredCount ->
ok;
_ ->
timer:sleep(50),
wait_registry_propagation(DesiredCount)
end.
wait_pg_propagation(DesiredCount) ->
case length(syn:members(?TEST_SCOPE, ?TEST_GROUP_NAME)) of
DesiredCount ->
ok;
_ ->
timer:sleep(50),
wait_pg_propagation(DesiredCount)
end.
start_profiling(NodesInfo) ->
{Node, _FromName, _ToName} = hd(NodesInfo),
ok = rpc:call(Node, ?MODULE, start_profiling_on_node, []).
stop_profiling(NodesInfo) ->
{Node, _FromName, _ToName} = hd(NodesInfo),
ok = rpc:call(Node, ?MODULE, stop_profiling_on_node, []).
start_profiling_on_node() ->
{ok, P} = eprof:start(),
eprof:start_profiling(erlang:processes() -- [P]),
ok.
stop_profiling_on_node() ->
eprof:stop_profiling(),
eprof:analyze(total),
ok.