%% Demonstration of the snowball effect triggered by a transient
%% overload and selective receive on a long message queue.

%% Adjust ?PERIOD so that CPU load is about 10 % during the first 10 s.

-module(syncsnowball).
-compile(export_all).

-define(PERIOD, 1000).
-define(N_CLIENTS, 1000).

run() ->
    {ok,_} = gen_server:start_link({local,backend}, ?MODULE, 0, []),
    Server = spawn_link(?MODULE, server, [arg]),
    spawn_link(?MODULE, monitor, [Server]),
    register(server, Server),
    spawn_clients(?N_CLIENTS),
    io:format("~w clients spawned.~n", [?N_CLIENTS]),
    receive after 10000 -> ok end,
    ok = gen_server:call(backend, pause),
    receive after 60000 -> ok end,
    erlang:halt().

monitor(Server) ->
    receive after 1000 -> ok end,
    {_, MQL} = process_info(Server, message_queue_len),
    Count = gen_server:call(backend, read),
    io:format("queue_len = ~w  throughput = ~w msg/s~n", [MQL, Count]),
    monitor(Server).

spawn_clients(0) ->
    ok;
spawn_clients(N) ->
    receive after 1 -> ok end,
    spawn_link(?MODULE, client, [mtime()]),
    spawn_clients(N-1).

mtime() -> {MS,S,US} = now(), (MS*1000+S)*1000 + US div 1000.

%%%% Client.

client(Next) ->
    Wait = lists:max([0, Next-mtime()]),
    receive after Wait -> ok end,
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    server ! {self(),{req1,req2}},
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    receive ack -> ok end,
    client(Next+?PERIOD).

%%%% Server.

server(State) ->
    receive
	{Client, Msg} ->
	    State1 = gen_server:call(backend, {State,Msg}),
	    Client ! ack,
	    server(State1)
    end.

%%%% Backend.

init(State) ->
    {ok, State}.
handle_call(read, _From, State) ->
    {reply, State, 0};
handle_call(pause, _From, State) ->
    io:format("~nBackend pausing for 1 second.~n~n"),
    receive after 1000 -> ok end,
    {reply, ok, State};
handle_call({RS,_RM}, _From, State) ->
    {reply, RS, State+1}.
