Packages
ra
1.0.6
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_monitors.erl
-module(ra_monitors).
-export([
init/0,
add/3,
remove/3,
remove_all/2,
handle_down/2,
components/2
]).
%% holds static or rarely changing fields
-type component() :: machine | aux | snapshot_sender.
-opaque state() :: #{pid() => {reference(), #{component() => ok}},
node() => #{component() => ok}}.
-export_type([
state/0
]).
-spec init() -> state().
init() ->
ok = net_kernel:monitor_nodes(true),
#{}.
-spec add(pid() | node(), component(), state()) -> state().
add(Pid, Component, Monitors) when is_pid(Pid) ->
case Monitors of
#{Pid := {MRef, Components}} ->
Monitors#{Pid => {MRef, Components#{Component => ok}}};
_ ->
MRef = erlang:monitor(process, Pid),
Monitors#{Pid => {MRef, #{Component => ok}}}
end;
add(Node, Component, Monitors) when is_atom(Node) ->
case Monitors of
#{Node := Components} ->
case maps:is_key(Component, Components) of
false ->
emit_current_node_state(Node);
true ->
ok
end,
Monitors#{Node => Components#{Component => ok}};
_ ->
emit_current_node_state(Node),
Monitors#{Node => #{Component => ok}}
end.
emit_current_node_state(Node) ->
%% fake event for newly registered component
%% so that it discovers the current node state
case lists:member(Node, nodes()) of
true ->
self() ! {nodeup, Node, []},
ok;
false ->
self() ! {nodedown, Node, []},
ok
end.
-spec remove(pid() | node(), component(), state()) -> state().
remove(Target, Component, Monitors) ->
case Monitors of
#{Target := {MRef, #{Component := _} = Components0}} ->
case maps:remove(Component, Components0) of
Components when map_size(Components) == 0 ->
%% if pid, demonitor using ref
true = erlang:demonitor(MRef),
maps:remove(Target, Monitors);
Components ->
Monitors#{Target => Components}
end;
#{Target := #{Component := _} = Components0} ->
case maps:remove(Component, Components0) of
Components when map_size(Components) == 0 ->
maps:remove(Target, Monitors);
Components ->
Monitors#{Target => Components}
end;
_ ->
Monitors
end.
-spec remove_all(component(), state()) -> state().
remove_all(Component, Monitors) ->
lists:foldl(fun(T, Acc) ->
remove(T, Component, Acc)
end, Monitors, maps:keys(Monitors)).
-spec handle_down(pid() | node(), state()) ->
{[component()], state()}.
handle_down(Target, Monitors0)
when is_pid(Target) orelse is_atom(Target) ->
case maps:take(Target, Monitors0) of
{{_MRef, CompsMap}, Monitors} ->
{maps:keys(CompsMap), Monitors};
{CompsMap, _Monitors} when is_map(CompsMap) ->
%% nodes aren't removed when down
{maps:keys(CompsMap), Monitors0};
error ->
{[], Monitors0}
end.
-spec components(pid() | node(), state()) -> [component()].
components(Target, Monitors) ->
case maps:get(Target, Monitors, error) of
{_MRef, CompsMap} ->
maps:keys(CompsMap);
CompsMap when is_map(CompsMap) ->
maps:keys(CompsMap);
error ->
[]
end.
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
basics_test() ->
M0 = init(),
M1 = add(self(), machine, M0),
[machine] = components(self(), M1),
M2 = add(self(), aux, M1),
[aux, machine] = components(self(), M2),
M3 = remove(self(), machine, M2),
[aux] = components(self(), M3),
{[aux, machine], M5} = handle_down(self(), M2),
[] = components(self(), M5),
ok.
nodes_test() ->
M0 = init(),
M1 = add(fake@blah, machine, M0),
receive
{nodedown, fake@blah, _} -> ok
after 100 -> exit(nodedown_timeout)
end,
M2 = add(fake2@blah, machine, M1),
receive
{nodedown, fake2@blah, _} -> ok
after 100 -> exit(nodedown_timeout_2)
end,
M3 = add(fake2@blah, aux, M2),
receive
{nodedown, fake2@blah, _} -> ok
after 100 -> exit(nodedown_timeout_3)
end,
M4 = remove_all(machine, M3),
?assertEqual([fake2@blah], maps:keys(M4)),
?assertEqual([#{aux => ok}], maps:values(M4)),
ok.
-endif.