Packages
ra
1.0.8
3.1.9
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.2
3.0.1
3.0.0
3.0.0-beta.1
2.17.3
2.17.2
2.17.1
2.17.0
2.16.13
2.16.12
2.16.11
2.16.10
2.16.9
2.16.8
2.16.7
2.16.6
2.16.5
2.16.4
2.16.3
2.16.2
2.16.1
2.16.0
2.16.0-pre.12
2.16.0-pre.11
2.16.0-pre.10
2.16.0-pre.9
2.16.0-pre.8
2.16.0-pre.7
2.16.0-pre.6
2.16.0-pre.5
2.16.0-pre.4
2.16.0-pre.3
2.16.0-pre.2
2.16.0-pre.1
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.13.0-pre.1
2.12.0
2.11.0
2.11.0-pre.1
2.10.2-pre.2
2.10.2-pre.1
2.10.1
2.10.0
2.10.0-pre.3
2.10.0-pre.2
2.10.0-pre.1
2.9.10-pre.1
2.9.1
2.9.1-pre.2
2.9.1-pre.1
2.9.0
2.8.0
retired
2.7.3
2.7.2
2.7.1
2.7.0
2.7.0-pre.3
2.7.0-pre.2
2.7.0-pre.1
2.6.3
2.6.2
2.6.1
2.6.0-pre.1
2.5.1
2.5.1-pre.1
2.5.0
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
retired
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.13
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.9.6
0.9.5
0.9.4
0.9.2
0.3.3
retired
0.3.2
retired
0.3.1
retired
Raft library
Current section
Files
Jump to
Current section
Files
src/ra_bench.erl
%% @hidden
-module(ra_bench).
-behaviour(ra_machine).
-compile(inline_list_funcs).
-compile(inline).
-compile({no_auto_import, [apply/3]}).
-include("ra.hrl").
-include_lib("eunit/include/eunit.hrl").
-export([
init/1,
apply/3,
% profile/0,
% stop_profile/0
prepare/0,
run/1,
run/0
]).
init(#{}) ->
undefined.
% msg_ids are scoped per customer
% ra_indexes holds all raft indexes for enqueues currently on queue
apply(#{index := I}, {noop, _}, State) ->
case I rem 100000 of
0 ->
{State, ok, {release_cursor, I, State}};
_ ->
{State, ok}
end.
run() ->
run(#{name => noop,
seconds => 10,
target => 50000,
degree => 5,
nodes => [node() | nodes()]}).
run(#{name := Name,
seconds := Secs,
target := Target,
nodes := Nodes,
degree := Degree}) ->
io:format("starting nodes ~w~n", [Nodes]),
{ok, ServerIds, []} = start(Name, Nodes),
{ok,_, Leader} = ra:members(hd(ServerIds)),
TotalOps = Secs * Target,
Each = TotalOps div Degree,
Start = os:system_time(millisecond),
Pids = [spawn_client(self(), Leader, Each) || _ <- lists:seq(1, Degree)],
%% wait for each pid
Wait = ((Secs * 1000) * 4),
[begin
receive
{done, P} ->
io:format("~w is done ~n", [P]),
ok
after Wait ->
exit({timeout, P})
end
end || P <- Pids],
End = os:system_time(millisecond),
Taken = End - Start,
io:format("benchmark completed: ~b ops in ~bms rate ~b ops/sec~n",
[TotalOps, Taken, TotalOps div (Taken div 1000)]),
ra:delete_cluster(ServerIds),
print_metrics(atom_to_binary(Name, utf8)),
ok.
start(Name, Nodes) when is_atom(Name) ->
ServerIds = [{Name, N} || N <- Nodes],
Configs = [begin
rpc:call(N, ?MODULE, prepare, []),
Id = {Name, N},
UId = ra:new_uid(ra_lib:to_binary(Name)),
#{id => Id,
uid => UId,
cluster_name => Name,
metrics_key => atom_to_binary(Name, utf8),
log_init_args => #{uid => UId},
initial_members => ServerIds,
machine => {module, ?MODULE, #{}}}
end || N <- Nodes],
ra:start_cluster(Configs).
prepare() ->
_ = application:ensure_all_started(ra),
% error_logger:logfile(filename:join(ra_env:data_dir(), "log.log")),
ok.
send_n(_, 0) -> ok;
send_n(Leader, N) ->
ra:pipeline_command(Leader, {noop, <<>>}, make_ref(), normal),
send_n(Leader, N-1).
client_loop(0, 0, _Leader) ->
ok;
client_loop(Num, Sent, _Leader) ->
receive
{ra_event, Leader, {applied, Applied}} ->
N = length(Applied),
ToSend = min(Sent, N),
send_n(Leader, ToSend),
client_loop(Num - N, Sent - ToSend, Leader);
{ra_event, _, {rejected, {not_leader, NewLeader, _}}} ->
io:format("new leader ~w~n", [NewLeader]),
send_n(NewLeader, 1),
client_loop(Num, Sent, NewLeader);
{ra_event, Leader, Evt} ->
io:format("unexpected ra_event ~w~n", [Evt]),
client_loop(Num, Sent,Leader)
end.
spawn_client(Parent, Leader, Num) when Num >= 1000 ->
spawn_link(
fun () ->
%% first send one 1000 noop commands
%% then top up as they are applied
send_n(Leader, 1000),
ok = client_loop(Num, Num - 1000, Leader),
Parent ! {done, self()}
end).
print_metrics(Name) ->
io:format("metrics ~p", [ets:lookup(ra_metrics, Name)]).
% profile() ->
% GzFile = atom_to_list(node()) ++ ".gz",
% lg:trace([noop, ra_server, ra_server_proc, ra_snapshot, ra_machine,
% ra_log, ra_flru, ra_machine, ra_log_meta, ra_log_segment],
% lg_file_tracer,
% GzFile, #{running => false, mode => profile}),
% ok.
% stop_profile() ->
% lg:stop(),
% Base = atom_to_list(node()),
% GzFile = Base ++ ".gz.*",
% lg_callgrind:profile_many(GzFile, Base ++ ".out",#{}),
% ok.