Packages
ra
1.0.1
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/noop.erl
%% @hidden
-module(noop).
-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,
start/1,
spawn_client/1,
print_metrics/1,
% profile/0,
% stop_profile/0
prepare/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.
start(Nodes) ->
ServerIds = [{noop, N} || N <- Nodes],
Configs = [begin
rpc:call(N, ?MODULE, prepare, []),
Id = {noop, N},
UId = ra:new_uid(ra_lib:to_binary(noop)),
#{id => Id,
uid => UId,
cluster_name => noop,
metrics_key => "the noop",
log_init_args => #{uid => UId},
initial_members => ServerIds,
machine => {module, ?MODULE, #{}}}
end || N <- Nodes],
{ra:start_cluster(Configs), ServerIds}.
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(Leader) ->
receive
{ra_event, Leader, {applied, Applied}} ->
N = length(Applied),
send_n(Leader, N),
client_loop(Leader);
{ra_event, _, {rejected, {not_leader, NewLeader, _}}} ->
io:format("new leader ~w~n", [NewLeader]),
send_n(NewLeader, 1),
client_loop(NewLeader);
{ra_event, Leader, Evt} ->
io:format("unexpected ra_event ~w~n", [Evt]),
client_loop(Leader)
end.
spawn_client(Servers) ->
spawn_link(
fun () ->
%% first send one 1000 noop commands
%% then top up as they are applied
{ok, _, Leader} = ra:members(hd(Servers)),
send_n(Leader, 1000),
client_loop(Leader)
end).
print_metrics(undefined) ->
print_metrics(hd(ets:lookup(ra_metrics, noop)));
print_metrics({noop, _, A0, _, B0, _}) ->
timer:sleep(1000),
[{noop, _, A, _, B, _} = X] = ets:lookup(ra_metrics, noop),
io:format("metrics ~b ~b per second~n",
[A-A0, B-B0]),
print_metrics(X).
% 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.