chut / src / benchmark_stress.erl

%%%===================================================================
%%% A benchmark made to stress a single node as much as possible. This
%%% benchmark aims to test only chut's core, and not the web server in
%%% front of it.
%%%
%%% The users for this test are separated into groups of many clients
%%% (with one client per user). The clients of each group will only
%%% message other clients from the same group. This is made to somehow
%%% imitate people having conversations together rather than spamming
%%% the whole server at large.
%%%===================================================================
-module(benchmark_stress).
-compile([export_all, debug_info]).

%%====================================================================
%% Base accumulating process
%%====================================================================
%%--------------------------------------------------------------------
%% Function: world(Groups, ClientsPerGroup, Delay) -> ok
%% Description: this function is the base of the benchmark. Its role
%% is to start each group, which will subsequently start users.
%% It also keeps time (Delay, in seconds) and then orders the group to
%% have its user stop messaging. It then compiles the stats for all
%% groups and displays them to the user.
%%--------------------------------------------------------------------
world(Groups, ClientsPerGroup, Delay) ->
    init_world(),
    Pids = spawn_groups(Groups, ClientsPerGroup, []),
    [receive {Pid, ready} -> ready end || Pid <- Pids],
    io:format("All groups spawned. Ordering to message.~n"),
    [Pid ! {self(),ok} || Pid <- Pids],
    timer:sleep(Delay*1000),
    io:format("Delay maxed. Compiling stats...~n"),
    {Sent, Received} = compile_stats(Pids),
    io:format("G:~p\tCPG:~p\tT:~p~n",[Groups,ClientsPerGroup,Delay]),
    io:format("Sent: ~p~nReceived: ~p~n",[Sent,Received]),
    io:format("Average sent per second: ~p~n", [Sent/Delay]),
    io:format("Average received per second: ~p~n", [Received/Delay]),
    io:format("Average sent per group: ~p~n", [Sent/Groups]),
    io:format("Average received per group: ~p~n", [Received/Groups]),
    io:format("Average sent per client: ~p~n", [Sent/(Groups*ClientsPerGroup)]),
    io:format("Average received per client: ~p~n", [Received/(Groups*ClientsPerGroup)]),
    io:format("Average sent per client per second: ~p~n", [Sent/(Groups*ClientsPerGroup)/Delay]),
    io:format("Average received per client per second: ~p~n", [Received/(Groups*ClientsPerGroup)/Delay]),
    ok.

%%--------------------------------------------------------------------
%% Function: spawn_groups(N, ClientsPerGroup, Acc) -> [pid()].
%% Description: creates N groups of users in different processes.
%%--------------------------------------------------------------------
spawn_groups(0, _, Pids) -> Pids;
spawn_groups(N, CPG, Pids) ->
    S = self(),
    Pid = proc_lib:spawn_link(fun() -> group(CPG,S) end),
    spawn_groups(N-1, CPG, [Pid|Pids]).

%%--------------------------------------------------------------------
%% Function: init_world() -> ok | {error, Reason}
%% Description: Starts the required applications to run the
%% benchmark.
%%--------------------------------------------------------------------
init_world() ->
    application:start(chut).

%%--------------------------------------------------------------------
%% Function: compile_stats(Pids) -> {SentSum, ReceivedSum}.
%% Description: requests statistics to be sent back by ever Pid in the
%% Pids list. The statistics have to be of the form {Sent,Received} in
%% order to be added for each Pid.
%%--------------------------------------------------------------------
compile_stats(Pids) ->
    [Pid ! {self(), stats} || Pid <- Pids],
    lists:foldl(fun({X,Y},{XSum,YSum}) -> {X+XSum,Y+YSum} end,
                {0,0},
                [collect_stats() || _Pid <- Pids]).

%%--------------------------------------------------------------------
%% Function: collect_stats() -> {Sent,Received}
%% Description: Waits for any message of the form {X,Y} and then
%% returns it. This sounds unsafe, but in practice, only statistics
%% should be returned under this form.
%%--------------------------------------------------------------------
collect_stats() ->
    receive
        M={_,_} -> M
    end.

%%====================================================================
%% Group Processes
%%====================================================================
%%--------------------------------------------------------------------
%% Function: group(N, Parent) -> void.
%% Description: The group function acts as a middleman process
%% between the base process and each client. Each group process is
%% responsible of N clients. Once they're all spawned, it sends an
%% acknowledgement to the base process, which will then tell it to
%% tell (!) the clients to start messaging eachother.
%%
%% At some point in time, the base process will alert the group
%% process that it's time to stop. The message will be relayed to all
%% clients who will in turn return their send/receive stats.
%% Each group compiles the stats in exactly the same manner done by the
%% base process.
%%--------------------------------------------------------------------
group(N, Parent) ->
    Ids = [make_ref() || _ <- lists:seq(1,N)],
    Pids = spawn_clients(Ids, [], []),
    [receive {Pid,ready} -> ready end || Pid <- Pids],
    Parent ! {self(), ready},
    receive
        {Parent,ok} -> ok
    end,
    [Pid ! {self(),ok} || Pid <- Pids],
    receive
        {From, stats} ->
            Ratios = compile_stats(Pids),
            From ! Ratios
    end.

%%--------------------------------------------------------------------
%% Function: spawn_clients(IdsToDo, IdsDone, Acc) -> Pids.
%% Description: Spawns a list of clients, giving each of them the
%% User Id of every other client in the group in order for them to be
%% message via chut.
%%--------------------------------------------------------------------
spawn_clients([], _, Pids) -> Pids;
spawn_clients([H|T], Done, Pids) ->
    S = self(),
    Pid = proc_lib:spawn_link(fun() -> client(H, T++Done, S) end),
    spawn_clients(T, [H|Done], [Pid|Pids]).


%%====================================================================
%% Client/User Processes
%%====================================================================
%%--------------------------------------------------------------------
%% Function: client(UserId, Peers, Parent) -> void
%% Description: Acts as a new client process, which starts a user on
%% chut and then connects to it. When it receives the 'ready' message
%% from its Parent (a Group process), it calls client/5 which will do
%% the messaging and stats work.
%%--------------------------------------------------------------------
client(UserId, Peers, Parent) ->
    chut_user:start(UserId, 60000, 10),
    {ok, HandlerId} = chut_user:subscribe(UserId),
    Parent ! {self(), ready},
    receive
        {Parent, ok} ->
            client(UserId, HandlerId, 0, 0, Peers)
    end.

%%--------------------------------------------------------------------
%% Function: client(UserId, HandlerId, Sent, Received, Peers) -> void.
%% Description: Messages a user at random, and then listens for
%% messages coming from other users (if any). For each of these
%% messages, a counter is incremented whether it was sent or received.
%%
%% When the group process contacts a client asking for stats, those
%% are messaged back to said group process and the client/user is
%% terminated.
%%--------------------------------------------------------------------
client(UserId, HandlerId, Sent, Received, Peers) ->
    Dest = random(Peers),
    chut_user:message(UserId, Dest, term_to_binary({<<"I've been mad for fucking years—absolutely years">>,os:timestamp()})),
    {NewSent, NewReceived} = count_messages({UserId, HandlerId}),
    receive
        {From, stats} ->
            From ! {Sent+NewSent, Received+NewReceived},
            chut_user:terminate(UserId)
    after 0 ->
        client(UserId, HandlerId, Sent+NewSent, Received+NewReceived, Peers)
    end.

%%--------------------------------------------------------------------
%% Function: count_messages({UserId, HandlerId}) -> {Sent, Received}.
%% Description: Counts each sent and received messages.
%%--------------------------------------------------------------------
count_messages(Id) ->
    lists:foldl(fun({Action,_,_}, {S,R}) ->
                    if Action =:= received -> {S,R+1}
                     ; Action =:= sent -> {S+1,R}
                    end
                end,
                {0,0},
                listen(Id)).

%%--------------------------------------------------------------------
%% Function: listen({UserId, HandlerId}) -> [Messages].
%% Description: catches all messages currently in the mailbox and
%% returns them. If the user's event handler crashes, the whole client
%% is also killed from here.
%%--------------------------------------------------------------------
listen(Id={_,Handler}) ->
    receive
        {Id, M = {_Action, _From, _Msg}} ->
            [M | listen(Id)];
        {gen_event_EXIT, Handler, normal} -> ok;
        {gen_event_EXIT, Handler, Reason} -> exit({client,Reason})
    after 0 -> []
    end.

%%--------------------------------------------------------------------
%% Function: random(ListOrTuple) -> Elem.
%% Description: gives a random element from a list or a tuple.
%%--------------------------------------------------------------------
random(L) when is_list(L) -> random(list_to_tuple(L));
random(T) when is_tuple(T) ->
    element(random:uniform(erlang:tuple_size(T)), T).
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.