Current section

Files

Jump to
ra src noop.erl
Raw

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.