%% ------------------------------------------------------------------- %% %% Copyright (c) 2016 Christopher Meiklejohn. All Rights Reserved. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file %% except in compliance with the License. You may obtain %% a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, %% software distributed under the License is distributed on an %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. %% %% ------------------------------------------------------------------- -module(partisan_hyparview_xbot_peer_service_manager). -author("Bruno Santiago Vazquez "). -behaviour(gen_server). -behaviour(partisan_peer_service_manager). -define(DEFAULT_PASSIVE_VIEW_MAINTENANCE_INTERVAL, 10000). -define(RANDOM_PROMOTION_INTERVAL, 5000). -include("partisan_logger.hrl"). -include("partisan.hrl"). %% partisan_peer_service_manager callbacks -export([cast_message/2]). -export([cast_message/3]). -export([cast_message/4]). -export([decode/1]). -export([forward_message/2]). -export([forward_message/3]). -export([forward_message/4]). -export([get_local_state/0]). -export([inject_partition/2]). -export([join/1]). -export([leave/0]). -export([leave/1]). -export([members/0]). -export([members_for_orchestration/0]). -export([myself/0]). -export([on_down/2]). -export([on_up/2]). -export([partitions/0]). -export([receive_message/2]). -export([reserve/1]). -export([resolve_partition/1]). -export([send_message/2]). -export([start_link/0]). -export([supports_capability/1]). -export([sync_join/1]). -export([update_members/1]). %% debug. -export([active/0, active/1, passive/0]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). %% temporary exceptions -export([delete_state_from_disk/0]). -type active() :: sets:set(partisan:node_spec()). -type passive() :: sets:set(partisan:node_spec()). -type reserved() :: dict:dict(atom(), partisan:node_spec()). -type tag() :: atom(). %% The epoch indicates how many times the node is restarted. -type epoch() :: non_neg_integer(). %% The epoch_count indicates how many disconnect messages are generated. -type epoch_count() :: non_neg_integer(). -type message_id() :: {epoch(), epoch_count()}. -type message_id_store() :: dict:dict(partisan:node_spec(), message_id()). -record(state, {myself :: partisan:node_spec(), active :: active(), passive :: passive(), reserved :: reserved(), out_links :: list(), tag :: tag(), max_active_size :: non_neg_integer(), min_active_size :: non_neg_integer(), max_passive_size :: non_neg_integer(), epoch :: epoch(), sent_message_map :: message_id_store(), recv_message_map :: message_id_store(), partitions :: partisan_peer_service_manager:partitions()}). -type state_t() :: #state{}. %%%=================================================================== %%% partisan_peer_service_manager callbacks %%%=================================================================== %% @doc Same as start_link([]). -spec start_link() -> {ok, pid()} | ignore | {error, term()}. start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %% @doc Return membership list. members() -> gen_server:call(?MODULE, members, infinity). %% @doc Return membership list. members_for_orchestration() -> gen_server:call(?MODULE, members_for_orchestration, infinity). %% @doc Return myself. myself() -> partisan:node_spec(). %% @doc Return local node's view of cluster membership. get_local_state() -> gen_server:call(?MODULE, get_local_state, infinity). %% @doc Register a trigger to fire when a connection drops. on_down(_Name, _Function) -> {error, not_implemented}. %% @doc Register a trigger to fire when a connection opens. on_up(_Name, _Function) -> {error, not_implemented}. %% @doc Update membership. update_members(_Nodes) -> {error, not_implemented}. %% @doc Send message to a remote manager. send_message(Name, Message) -> gen_server:call(?MODULE, {send_message, Name, Message}, infinity). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec cast_message( Term :: partisan_remote_ref:p() | partisan_remote_ref:n() | pid(), MEssage :: partisan:message()) -> ok. cast_message(Term, Message) -> FullMessage = {'$gen_cast', Message}, forward_message(Term, FullMessage, #{}). %% ----------------------------------------------------------------------------- %% @doc Cast a message to a remote gen_server. %% @end %% ----------------------------------------------------------------------------- cast_message(Node, ServerRef, Message) -> cast_message(Node, ServerRef, Message, #{}). %% ----------------------------------------------------------------------------- %% @doc Cast a message to a remote gen_server. %% @end %% ----------------------------------------------------------------------------- cast_message(Name, ServerRef, Message, Options) -> FullMessage = {'$gen_cast', Message}, forward_message(Name, ServerRef, FullMessage, Options). %% ----------------------------------------------------------------------------- %% @doc Gensym support for forwarding. %% @end %% ----------------------------------------------------------------------------- forward_message(Term, Message) -> forward_message(Term, Message, #{}). %% ----------------------------------------------------------------------------- %% @doc Gensym support for forwarding. %% @end %% ----------------------------------------------------------------------------- forward_message(Pid, Message, Opts) when is_pid(Pid) -> forward_message(partisan:node(), Pid, Message, Opts); forward_message(RemoteRef, Message, Opts) -> partisan_remote_ref:is_pid(RemoteRef) orelse partisan_remote_ref:is_name(RemoteRef) orelse error(badarg), Node = partisan_remote_ref:node(RemoteRef), Target = partisan_remote_ref:target(RemoteRef), forward_message(Node, Target, Message, Opts). %% ----------------------------------------------------------------------------- %% @doc Forward message to registered process on the remote side. %% @end %% ----------------------------------------------------------------------------- forward_message(Node, ServerRef, Message, Opts) when is_list(Opts) -> forward_message(Node, ServerRef, Message, maps:from_list(Opts)); forward_message(Node, ServerRef, Message, Opts) when is_map(Opts) -> FullMessage = {forward_message, Node, ServerRef, Message, Opts}, %% Attempt to fast-path through the memoized connection cache. case partisan_peer_connections:dispatch(FullMessage) of ok -> ok; {error, _} -> gen_server:call(?MODULE, FullMessage, infinity) end. %% @doc Receive message from a remote manager. receive_message(_Peer, Message) -> gen_server:call(?MODULE, {receive_message, Message}, infinity). %% @doc Attempt to join a remote node. join(Node) -> gen_server:call(?MODULE, {join, Node}, infinity). %% @doc Attempt to join a remote node. sync_join(_Node) -> {error, not_implemented}. %% @doc Leave the cluster. leave() -> gen_server:call(?MODULE, {leave, partisan:node()}, infinity). %% @doc Remove another node from the cluster. leave(Node) -> gen_server:call(?MODULE, {leave, Node}, infinity). %% @doc Reserve a slot for the particular tag. reserve(Tag) -> gen_server:call(?MODULE, {reserve, Tag}, infinity). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec supports_capability(Arg :: atom()) -> boolean(). supports_capability(monitoring) -> false; supports_capability(_) -> false. %% @doc Inject a partition. inject_partition(Origin, TTL) -> gen_server:call(?MODULE, {inject_partition, Origin, TTL}, infinity). %% @doc Resolve a partition. resolve_partition(Reference) -> gen_server:call(?MODULE, {resolve_partition, Reference}, infinity). %% @doc Return partitions. partitions() -> gen_server:call(?MODULE, partitions, infinity). %%%=================================================================== %%% debugging callbacks %%%=================================================================== %% @doc Debugging. active() -> gen_server:call(?MODULE, active, infinity). %% @doc Debugging. active(Tag) -> gen_server:call(?MODULE, {active, Tag}, infinity). %% @doc Debugging. passive() -> gen_server:call(?MODULE, passive, infinity). %% @doc Decode state. decode({state, Active, _Epoch}) -> sets:to_list(Active); decode(Active) -> sets:to_list(Active). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %% @private -spec init([]) -> {ok, state_t()}. init([]) -> %% Seed the random number generator. partisan_config:seed(), ok = partisan_peer_connections:init(), %% Process connection exits. process_flag(trap_exit, true), Epoch = maybe_load_epoch_from_disk(), Active = sets:add_element(myself(), sets:new()), Passive = sets:new(), Myself = myself(), SentMessageMap = dict:new(), RecvMessageMap = dict:new(), %% Partitions. Partitions = [], %% Get the default configuration. MaxActiveSize = partisan_config:get(max_active_size, 6), MinActiveSize = partisan_config:get(min_active_size, 3), MaxPassiveSize = partisan_config:get(max_passive_size, 30), %% Get tag, if set. Tag = partisan_config:get(tag, undefined), %% Reserved server slots. Reservations = partisan_config:get(reservations, []), Reserved = dict:from_list([{T, undefined} || T <- Reservations]), %% Schedule periodic maintenance of the passive view. schedule_passive_view_maintenance(), %% Schedule periodic execution of xbot algorithm (optimization) schedule_xbot_execution(), %% Schedule tree peers refresh. schedule_tree_refresh(), %% Schedule periodic random promotion when it is enabled. case partisan_config:get(random_promotion, true) of true -> schedule_random_promotion(); false -> ok end, %% Verify we don't have too many reservations. case length(Reservations) > MaxActiveSize of true -> {stop, reservation_limit_exceeded}; false -> {ok, #state{myself=Myself, active=Active, passive=Passive, reserved=Reserved, tag=Tag, out_links=[], max_active_size=MaxActiveSize, min_active_size=MinActiveSize, max_passive_size=MaxPassiveSize, epoch=Epoch + 1, sent_message_map=SentMessageMap, recv_message_map=RecvMessageMap, partitions=Partitions}} end. %% @private -spec handle_call(term(), {pid(), term()}, state_t()) -> {reply, term(), state_t()}. handle_call(partitions, _From, #state{partitions=Partitions}=State) -> {reply, {ok, Partitions}, State}; handle_call({leave, _Node}, _From, State) -> {reply, error, State}; handle_call({join, #{name := _Name} = Node}, _From, State) -> gen_server:cast(?MODULE, {join, Node}), {reply, ok, State}; handle_call({resolve_partition, Reference}, _From, State) -> Partitions = handle_partition_resolution(Reference, State), {reply, ok, State#state{partitions=Partitions}}; handle_call({inject_partition, Origin, TTL}, _From, #state{myself=Myself}=State) -> Reference = make_ref(), ?LOG_DEBUG("Injecting partition; Origin: ~p Myself: ~p TTL: ~p", [Origin, Myself, TTL]), case Origin of Myself -> Partitions = handle_partition_injection(Reference, Origin, TTL, State), {reply, {ok, Reference}, State#state{partitions=Partitions}}; _ -> Result = do_send_message( Origin, {inject_partition, Reference, Origin, TTL} ), case Result of {error, Error} -> {reply, {error, Error}, State}; ok -> {reply, {ok, Reference}, State} end end; handle_call({reserve, Tag}, _From, #state{reserved=Reserved0, max_active_size=MaxActiveSize}=State) -> Present = dict:fetch_keys(Reserved0), case length(Present) < MaxActiveSize of true -> Reserved = case lists:member(Tag, Present) of true -> Reserved0; false -> dict:store(Tag, undefined, Reserved0) end, {reply, ok, State#state{reserved=Reserved}}; false -> {reply, {error, no_available_slots}, State} end; handle_call(active, _From, #state{active=Active}=State) -> {reply, {ok, Active}, State}; handle_call({active, Tag}, _From, #state{reserved=Reserved}=State) -> Result = case dict:find(Tag, Reserved) of {ok, #{name := Peer}} -> {ok, Peer}; {ok, undefined} -> {ok, undefined}; error -> error end, {reply, Result, State}; handle_call(passive, _From, #state{passive=Passive}=State) -> {reply, {ok, Passive}, State}; handle_call({send_message, Name, Message}, _From, #state{}=State) -> Result = do_send_message(Name, Message), {reply, Result, State}; handle_call({forward_message, Name, ServerRef, Message, Options}, _From, #state{partitions=Partitions}=State) -> IsPartitioned = lists:any(fun(#{name := N}) -> case N of Name -> true; _ -> false end end, Partitions), case IsPartitioned of true -> {reply, {error, partitioned}, State}; false -> Result = do_send_message( Name, {forward_message, ServerRef, Message}, Options ), {reply, Result, State} end; handle_call({receive_message, Message}, _From, State) -> gen_server:cast(?MODULE, {receive_message, Message}), {reply, ok, State}; handle_call(members, _From, #state{myself=Myself, active=Active}=State) -> ?LOG_DEBUG("Node ~p active view: ~p", [Myself, members(Active)]), ActiveMembers = [P || #{name := P} <- members(Active)], {reply, {ok, ActiveMembers}, State}; handle_call(members_for_orchestration, _From, #state{active=Active}=State) -> {reply, {ok, members(Active)}, State}; handle_call(get_local_state, _From, #state{active=Active, epoch=Epoch}=State) -> {reply, {ok, {state, Active, Epoch}}, State}; handle_call(connections, _From, #state{myself=Myself, active=Active}=State) -> %% get a list of all the client connections to the various peers of the active view Cs = lists:map( fun(Peer) -> Pids = partisan_peer_connections:processes(Peer), ?LOG_DEBUG("peer ~p connection pids: ~p", [Peer, Pids]), {Peer, Pids} end, members(Active) -- [Myself] ), {reply, {ok, Cs}, State}; handle_call(Event, _From, State) -> ?LOG_WARNING(#{description => "Unhandled call event", event => Event}), {reply, ok, State}. %% @private -spec handle_cast(term(), state_t()) -> {noreply, state_t()}. handle_cast( {join, Peer}, #state{myself=Myself0, tag=Tag0, epoch=Epoch0}=State0) -> %% Trigger connection. ok = partisan_peer_service_manager:connect(Peer), ?LOG_DEBUG("Node ~p sends the JOIN message to ~p", [Myself0, Peer]), %% Send the JOIN message to the peer. _ = do_send_message(Peer, {join, Myself0, Tag0, Epoch0}), %% Return. {noreply, State0}; handle_cast({receive_message, Message}, State0) -> handle_message(Message, State0); %% @doc Handle disconnect messages. handle_cast({disconnect, Peer}, #state{active=Active0}=State0) -> case sets:is_element(Peer, Active0) of true -> %% If a member of the active view, remove it. Active = sets:del_element(Peer, Active0), State = add_to_passive_view(Peer, State0#state{active=Active}), ok = disconnect(Peer), {noreply, State}; false -> {noreply, State0} end; handle_cast(Event, State) -> ?LOG_WARNING(#{description => "Unhandled call event", event => Event}), {noreply, State}. %% @private -spec handle_info(term(), state_t()) -> {noreply, state_t()}. handle_info(random_promotion, #state{myself=Myself, active=Active0, passive=Passive, reserved=Reserved0, min_active_size=MinActiveSize0}=State0) -> State = case has_reached_the_limit({active, Active0, Reserved0}, MinActiveSize0) of true -> %% Do nothing if the active view reaches the MinActiveSize. State0; false -> RandomPeer = select_random(Passive, [Myself]), move_peer_from_passive_to_active(RandomPeer, State0) end, %% Schedule periodic random promotion. schedule_random_promotion(), {noreply, State}; handle_info(tree_refresh, #state{}=State) -> %% Use our tree for the root. Root = partisan:node(), %% Get lazily computed outlinks. OutLinks = try partisan_plumtree_broadcast:debug_get_peers(partisan:node(), Root) of {EagerPeers, _LazyPeers} -> ordsets:to_list(EagerPeers) catch _:_ -> [] end, %% Reschedule. schedule_tree_refresh(), {noreply, State#state{out_links=OutLinks}}; handle_info( passive_view_maintenance, #state{myself=Myself, active=Active, passive=Passive}=State0) -> Exchange0 = %% Myself. [Myself] ++ % Random members of the active list. select_random_sublist(Active, k_active()) ++ %% Random members of the passive list. select_random_sublist(Passive, k_passive()), Exchange = lists:usort(Exchange0), %% Select random member of the active list. State = case select_random(Active, [Myself]) of undefined -> State0; Random -> %% Trigger connection. ok = partisan_peer_service_manager:connect(Random), %% Forward shuffle request. _ = do_send_message( Random, {shuffle, Exchange, arwl(), Myself} ), State0 end, %% Schedule periodic maintenance of the passive view. schedule_passive_view_maintenance(), {noreply, State}; % handle optimization using xbot algorithm handle_info(xbot_execution, #state{active=Active, passive=Passive, max_active_size=MaxActiveSize, reserved=Reserved}=InitiatorState) -> % check if active view is full IsFull = is_full({active,Active,Reserved}, MaxActiveSize), case IsFull of % if full, check for candidates and try to optimize true -> Candidates = select_random_sublist(Passive, 2), send_optimization_messages(Active, Candidates, InitiatorState); % in other case, do nothing false -> ok end, %In any case, schedule periodic xbot execution algorithm (optimization) schedule_xbot_execution(), {noreply, InitiatorState}; handle_info( {'EXIT', From, Reason}, #state{myself=Myself, active=Active0, passive=Passive0}=State0) -> ?LOG_DEBUG("connection pid ~p died due to ~p", [From, Reason]), %% Prune active connections from dictionary. try partisan_peer_connections:prune(From) of {Info, _Connections} -> Peer = partisan_peer_connections:node_spec(Info), %% If it was in the passive view and our connection attempt failed, %% remove from the passive view altogether. Passive = case is_in_passive_view(Peer, Passive0) of true -> remove_from_passive_view(Peer, Passive0); false -> Passive0 end, %% If it was in the active view and our connection attempt failed, %% remove from the active view altogether. {Active, RemovedFromActive} = case is_in_active_view(Peer, Active0) of true -> {remove_from_active_view(Peer, Active0), true}; false -> {Active0, false} end, State = case RemovedFromActive of true -> RandomPeer = select_random(Passive, [Myself]), move_peer_from_passive_to_active(RandomPeer, State0#state{active=Active, passive=Passive}); false -> State0#state{active=Active, passive=Passive} end, ?LOG_DEBUG( "Node ~p active view: ~p", [Myself, members(State#state.active)] ), {noreply, State} catch error:badarg -> {noreply, State0} end; handle_info( {connected, Peer, _Channel, _Tag, _PeerEpoch, _RemoteState}, State) -> ?LOG_DEBUG("Node ~p is now connected", [Peer]), {noreply, State}; handle_info(Event, State) -> ?LOG_WARNING(#{description => "Unhandled info event", event => Event}), {noreply, State}. %% @private -spec terminate(term(), state_t()) -> term(). terminate(_Reason, #state{}) -> Fun = fun(_NodeInfo, Connections) -> lists:foreach( fun(C) -> Pid = partisan_peer_connections:pid(C), catch gen_server:stop(Pid, normal, infinity), ok end, Connections ) end, ok = partisan_peer_connections:foreach(Fun). %% @private -spec code_change(term() | {down, term()}, state_t(), term()) -> {ok, state_t()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== % Send optimization messages when apply %% @private send_optimization_messages(_, [], _) -> ok; send_optimization_messages(Active, [Candidate | RestCandidates], InitiatorState) -> % check the first candidate against every node in the active view process_candidate(sets:to_list(Active), Candidate, InitiatorState), % do same for every candidate against the active send_optimization_messages(Active, RestCandidates, InitiatorState). % check if a candidate is valid and send message to try optimization % we only send an optimization messages for one node in active view, once we have sent it we stop searching possibilities %% @private process_candidate([], _, _) -> ok; process_candidate([OldNode | RestActiveNodes], CandidateNode, #state{myself=InitiatorNode}=InitiatorState) -> #{name := InitiatorName} = InitiatorNode, #{name := CandidateName} = CandidateNode, IsBetter = is_better(?XPARAM, CandidateNode, OldNode), if IsBetter -> ok = partisan_peer_service_manager:connect(CandidateNode), % if cadidate is better that first node in active view, send optimization message _ = do_send_message( CandidateNode, {optimization, undefined, OldNode, InitiatorNode, CandidateNode, undefined} ), ?LOG_DEBUG("XBOT: Optimization message sent to Node ~p from ~p", [CandidateName, InitiatorName]); true -> % if not, continue checking against the remaining nodes in active view process_candidate(RestActiveNodes, CandidateNode, InitiatorState) end. %% @private handle_message({resolve_partition, Reference}, State) -> Partitions = handle_partition_resolution(Reference, State), {noreply, State#state{partitions=Partitions}}; %% @private handle_message({inject_partition, Reference, Origin, TTL}, State) -> Partitions = handle_partition_injection(Reference, Origin, TTL, State), {noreply, State#state{partitions=Partitions}}; %% @private handle_message({join, Peer, PeerTag, PeerEpoch}, #state{myself=Myself0, active=Active0, tag=Tag0, sent_message_map=SentMessageMap0, recv_message_map=RecvMessageMap0}=State0) -> ?LOG_DEBUG("Node ~p received the JOIN message from ~p, epoch ~p", [Myself0, Peer, PeerEpoch]), IsAddable = is_addable(PeerEpoch, Peer, SentMessageMap0), NotInActiveView = not sets:is_element(Peer, Active0), State = case IsAddable andalso NotInActiveView of true -> %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), Connected = partisan_peer_connections:is_connected(Peer), case Connected of true -> ?LOG_DEBUG( "Adding peer node ~p to the active view", [Peer] ), %% Add to active view. State1 = add_to_active_view(Peer, PeerTag, State0), LastDisconnectId = get_current_id(Peer, RecvMessageMap0), %% Send the NEIGHBOR message to origin, that will update it's view. _ = do_send_message( Peer, {neighbor, Myself0, Tag0, LastDisconnectId, Peer} ), %% Random walk for forward join. %% Since we might have dropped peers from the active view when %% adding this one we need to use the most up to date active view, %% and that's the one that's currently in the state %% also disregard the the new joiner node Peers = (members(State1#state.active) -- [Myself0]) -- [Peer], ok = lists:foreach( fun(P) -> %% Establish connections. ok = partisan_peer_service_manager:connect(P), ?LOG_DEBUG(#{ description => "Forwarding join of to active view peer", from => Peer, to => P }), Message = { forward_join, Peer, PeerTag, PeerEpoch, arwl(), Myself0 }, _ = do_send_message(P, Message), ok end, Peers ), ?LOG_DEBUG( "Node ~p active view: ~p", [Myself0, members(State1#state.active)] ), %% Notify with event. notify(State1), %% Return. State1; false -> State0 end; false -> ?LOG_DEBUG( "Peer node ~p will not be added to the active view", [Peer] ), State0 end, {noreply, State}; %% @private handle_message( {neighbor, Peer, PeerTag, DisconnectId, _Sender}, #state{myself=Myself0, sent_message_map=SentMessageMap0}=State0) -> ?LOG_DEBUG( "Node ~p received the NEIGHBOR message from ~p with ~p", [Myself0, Peer, PeerTag] ), State = case is_addable(DisconnectId, Peer, SentMessageMap0) of true -> %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), case partisan_peer_connections:is_connected(Peer) of true -> %% Add node into the active view. State1 = add_to_active_view(Peer, PeerTag, State0), ?LOG_DEBUG( "Node ~p active view: ~p", [Myself0, members(State1#state.active)] ), State1; false -> State0 end; false -> State0 end, %% Notify with event. notify(State), {noreply, State}; %% @private handle_message({forward_join, Peer, PeerTag, PeerEpoch, TTL, Sender}, #state{myself=Myself0, active=Active0, tag=Tag0, sent_message_map=SentMessageMap0, recv_message_map=RecvMessageMap0}=State0) -> ?LOG_DEBUG("Node ~p received the FORWARD_JOIN message from ~p about ~p", [Myself0, Sender, Peer]), ActiveViewSize = sets:size(Active0), State = case TTL =:= 0 orelse ActiveViewSize =:= 1 of true -> ?LOG_DEBUG("FORWARD_JOIN: ttl(~p) expired or only one peer in active view (~p), " "adding ~p tagged ~p to active view", [TTL, ActiveViewSize, Peer, PeerTag]), IsAddable0 = is_addable(PeerEpoch, Peer, SentMessageMap0), NotInActiveView0 = not sets:is_element(Peer, Active0), case IsAddable0 andalso NotInActiveView0 of true -> %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), Connected = partisan_peer_connections:is_connected(Peer), case Connected of true -> %% Add to our active view. State1 = add_to_active_view(Peer, PeerTag, State0), LastDisconnectId = get_current_id( Peer, RecvMessageMap0 ), %% Send neighbor message to origin, that will update it's view. _ = do_send_message( Peer, {neighbor, Myself0, Tag0, LastDisconnectId, Peer} ), ?LOG_DEBUG( "Node ~p active view: ~p", [Myself0, members(State1#state.active)] ), State1; false -> State0 end; false -> ?LOG_DEBUG("Peer node ~p will not be added to the active view", [Peer]), State0 end; false -> %% If we run out of peers before we hit the PRWL, that's %% fine, because exchanges between peers will eventually %% repair the passive view during shuffles. State2 = case TTL =:= prwl() of true -> ?LOG_DEBUG("FORWARD_JOIN: Passive walk ttl expired, adding ~p to " "the passive view", [Peer]), add_to_passive_view(Peer, State0); false -> State0 end, %% Don't forward the join to the sender, ourself, or the joining peer. case select_random(Active0, [Sender, Myself0, Peer]) of undefined -> IsAddable1 = is_addable(PeerEpoch, Peer, SentMessageMap0), NotInActiveView1 = not sets:is_element(Peer, Active0), case IsAddable1 andalso NotInActiveView1 of true -> %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), Connected = partisan_peer_connections:is_connected( Peer ), case Connected of true -> ?LOG_DEBUG( "FORWARD_JOIN: No node for forward, adding ~p to active view", [Peer] ), %% Add to our active view. State3 = add_to_active_view(Peer, PeerTag, State2), LastDisconnectId = get_current_id(Peer, RecvMessageMap0), %% Send neighbor message to origin, that will %% update it's view. _ = do_send_message( Peer, {neighbor, Myself0, Tag0, LastDisconnectId, Peer}), ?LOG_DEBUG("Node ~p active view: ~p", [Myself0, members(State3#state.active)]), State3; false -> State2 end; false -> ?LOG_DEBUG("Peer node ~p will not be added to the active view", [Peer]), State2 end; Random -> %% Establish any new connections. ok = partisan_peer_service_manager:connect(Random), ?LOG_DEBUG("FORWARD_JOIN: forwarding to ~p", [Random]), %% Forward join. _ = do_send_message( Random, {forward_join, Peer, PeerTag, PeerEpoch, TTL - 1, Myself0}), State2 end end, %% Notify with event. notify(State), {noreply, State}; %% @private handle_message({disconnect, Peer, DisconnectId}, #state{myself=Myself0, active=Active0, passive=Passive, recv_message_map=RecvMessageMap0}=State0) -> ?LOG_DEBUG("Node ~p received the DISCONNECT message from ~p with ~p", [Myself0, Peer, DisconnectId]), case is_valid_disconnect(Peer, DisconnectId, RecvMessageMap0) of false -> %% Ignore the older disconnect message. {noreply, State0}; true -> %% Remove from active Active = sets:del_element(Peer, Active0), ?LOG_DEBUG("Node ~p active view: ~p", [Myself0, members(Active)]), %% Add to passive view. State1 = add_to_passive_view(Peer, State0#state{active=Active}), %% Update the AckMessageMap. RecvMessageMap = dict:store(Peer, DisconnectId, RecvMessageMap0), %% Trigger disconnection. ok = disconnect(Peer), State = case sets:size(Active) == 1 of true -> %% the peer that disconnected us just got moved to the %% passive view, exclude it when selecting a new one to %% move back into the active view RandomPeer = select_random(Passive, [Myself0, Peer]), ?LOG_DEBUG("Node ~p is isolated, moving random peer ~p from passive " "to active view", [RandomPeer, Myself0]), move_peer_from_passive_to_active( RandomPeer, State1#state{recv_message_map=RecvMessageMap} ); false -> State1#state{recv_message_map=RecvMessageMap} end, {noreply, State} end; %% @private handle_message({neighbor_request, Peer, Priority, PeerTag, DisconnectId, Exchange}, #state{myself=Myself0, active=Active0, passive=Passive0, tag=Tag0, sent_message_map=SentMessageMap0, recv_message_map=RecvMessageMap0}=State0) -> ?LOG_DEBUG("Node ~p received the NEIGHBOR_REQUEST message from ~p with ~p", [Myself0, Peer, DisconnectId]), %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), Exchange_Ack0 = %% Myself. [Myself0] ++ % Random members of the active list. select_random_sublist(Active0, k_active()) ++ %% Random members of the passive list. select_random_sublist(Passive0, k_passive()), Exchange_Ack = lists:usort(Exchange_Ack0), State2 = case neighbor_acceptable(Priority, PeerTag, State0) of true -> case is_addable(DisconnectId, Peer, SentMessageMap0) of true -> ?LOG_DEBUG("Node ~p accepted neighbor peer ~p", [Myself0, Peer]), LastDisconnectId = get_current_id(Peer, RecvMessageMap0), %% Reply to acknowledge the neighbor was accepted. _ = do_send_message( Peer, {neighbor_accepted, Myself0, Tag0, LastDisconnectId, Exchange_Ack}), State1 = add_to_active_view(Peer, PeerTag, State0), ?LOG_DEBUG( "Node ~p active view: ~p", [Myself0, members(State1#state.active)] ), State1; false -> ?LOG_DEBUG("Node ~p rejected neighbor peer ~p", [Myself0, Peer]), %% Reply to acknowledge the neighbor was rejected. _ = do_send_message(Peer, {neighbor_rejected, Myself0, Exchange_Ack}), State0 end; false -> ?LOG_DEBUG("Node ~p rejected neighbor peer ~p", [Myself0, Peer]), %% Reply to acknowledge the neighbor was rejected. _ = do_send_message(Peer, {neighbor_rejected, Myself0}), State0 end, State = merge_exchange(Exchange, State2), %% Notify with event. notify(State), {noreply, State}; %% @private handle_message({neighbor_rejected, Peer, Exchange}, #state{myself=Myself0} = State0) -> ?LOG_DEBUG("Node ~p received the NEIGHBOR_REJECTED message from ~p", [Myself0, Peer]), %% Trigger disconnection. ok = disconnect(Peer), State = merge_exchange(Exchange, State0), {noreply, State}; %% @private handle_message({neighbor_accepted, Peer, PeerTag, DisconnectId, Exchange}, #state{myself=Myself0, sent_message_map=SentMessageMap0} = State0) -> ?LOG_DEBUG("Node ~p received the NEIGHBOR_ACCEPTED message from ~p with ~p", [Myself0, Peer, DisconnectId]), State1 = case is_addable(DisconnectId, Peer, SentMessageMap0) of true -> %% Add node into the active view. add_to_active_view(Peer, PeerTag, State0); false -> State0 end, State = merge_exchange(Exchange, State1), %% Notify with event. notify(State), {noreply, State}; handle_message({shuffle_reply, Exchange, _Sender}, State0) -> State = merge_exchange(Exchange, State0), {noreply, State}; handle_message({shuffle, Exchange, TTL, Sender}, #state{myself=Myself, active=Active0, passive=Passive0}=State0) -> ?LOG_DEBUG("Node ~p received the SHUFFLE message from ~p", [Myself, Sender]), %% Forward to random member of the active view. State = case TTL > 0 andalso sets:size(Active0) > 1 of true -> State1 = case select_random(Active0, [Sender, Myself]) of undefined -> State0; Random -> %% Trigger connection. ok = partisan_peer_service_manager:connect(Random), %% Forward shuffle until random walk complete. _ = do_send_message( Random, {shuffle, Exchange, TTL - 1, Myself} ), State0 end, State1; false -> %% Randomly select nodes from the passive view and respond. ResponseExchange = select_random_sublist( Passive0, length(Exchange) ), %% Trigger connection. ok = partisan_peer_service_manager:connect(Sender), _ = do_send_message( Sender, {shuffle_reply, ResponseExchange, Myself} ), State2 = merge_exchange(Exchange, State0), State2 end, {noreply, State}; handle_message({relay_message, Node, Message}, #state{out_links=OutLinks}=State) -> ?LOG_DEBUG("Node ~p received tree relay to ~p", [partisan:node(), Node]), %% Attempt to deliver, or recurse and forward on through a relay. _ = do_send_message( Node, Message, #{out_links => OutLinks, transitive => true} ), {noreply, State}; handle_message({forward_message, ServerRef, Message}, State) -> ServerRef ! Message, {noreply, State}; %% Optimization Reply handle_message({optimization_reply, true, _, InitiatorNode, CandidateNode, undefined}, State) ->%#state{active=Active}=State) -> #{name := InitiatorName} = InitiatorNode, #{name := CandidateName} = CandidateNode, ?LOG_DEBUG("XBOT: Received optimization reply message at Node ~p from ~p", [InitiatorName, CandidateName]), %% Revise this behaviour, when candidate accepts inmediatly because it has availability in his active view %% what to do with old node?? we cannot disconnect from it because maybe it will be isolated %Check = is_in_active_view(OldNode, Active), %if Check -> %remove_from_active_view(OldNode, Active), %add_to_passive_view(OldNode, State) % do_disconnect(OldNode, State) %end, %move_peer_from_passive_to_active(CandidateNode, State), _ = send_join(CandidateNode, State), ?LOG_DEBUG("XBOT: Finished optimization round started by Node ~p ", [InitiatorName]), {noreply, State}; handle_message({optimization_reply, true, OldNode, InitiatorNode, CandidateNode, _}, #state{active=Active}=State) -> #{name := InitiatorName} = InitiatorNode, #{name := CandidateName} = CandidateNode, ?LOG_DEBUG("XBOT: Received optimization reply message at Node ~p from ~p", [InitiatorName, CandidateName]), Check = is_in_active_view(OldNode, Active), if Check -> %remove_from_active_view(OldNode, Active); do_disconnect(OldNode, State); true -> ok end, %move_peer_from_passive_to_active(CandidateNode, State), _ = send_join(CandidateNode, State), ?LOG_DEBUG("XBOT: Finished optimization round started by Node ~p ", [InitiatorName]), {noreply, State}; handle_message({optimization_reply, false, _, _, _, _}, State) -> {noreply, State}; %% Optimization handle_message({optimization, _, OldNode, InitiatorNode, CandidateNode, undefined}, #state{ active=Active, max_active_size=MaxActiveSize, reserved=Reserved}=State) -> #{name := CandidateName} = CandidateNode, #{name := InitiatorName} = InitiatorNode, ?LOG_DEBUG("XBOT: Received optimization message at Node ~p from ~p", [CandidateName, InitiatorName]), Check = is_full({active, Active, Reserved},MaxActiveSize), if not Check -> %add_to_active_view(CandidateNode, MyTag, State), _ = send_join(InitiatorNode, State), ok = partisan_peer_service_manager:connect(InitiatorNode), Message = { optimization_reply, true, OldNode, InitiatorNode, CandidateNode, undefined }, _ = do_send_message(InitiatorNode, Message), ?LOG_DEBUG( "XBOT: Sending optimization reply message to Node ~p from ~p", [InitiatorName, CandidateName] ); true -> DisconnectNode = select_disconnect_node(sets:to_list(Active)), #{name := DisconnectName} = DisconnectNode, ok = partisan_peer_service_manager:connect(DisconnectNode), Message = { replace, undefined, OldNode, InitiatorNode, CandidateNode, DisconnectNode }, _ = do_send_message(DisconnectNode, Message), ?LOG_DEBUG( "XBOT: Sending replace message to Node ~p from ~p", [DisconnectName, CandidateName] ) end, {noreply, State}; %% Replace Reply handle_message({replace_reply, true, OldNode, InitiatorNode, CandidateNode, DisconnectNode}, #state{}=State) -> #{name := InitiatorName} = InitiatorNode, #{name := DisconnectName} = DisconnectNode, #{name := CandidateName} = CandidateNode, ?LOG_DEBUG("XBOT: Received replace reply message at Node ~p from ~p", [CandidateName, DisconnectName]), %remove_from_active_view(DisconnectNode, Active), do_disconnect(DisconnectNode, State), %add_to_active_view(InitiatorNode, MyTag, State), _ = send_join(InitiatorNode, State), ok = partisan_peer_service_manager:connect(InitiatorNode), _ = do_send_message(InitiatorNode,{optimization_reply, true, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending optimization reply to Node ~p from ~p", [InitiatorName, CandidateName]), {noreply, State}; handle_message({replace_reply, false, OldNode, InitiatorNode, CandidateNode, DisconnectNode}, #state{}=State) -> #{name := InitiatorName} = InitiatorNode, #{name := DisconnectName} = DisconnectNode, #{name := CandidateName} = CandidateNode, ?LOG_DEBUG("XBOT: Received replace reply message at Node ~p from ~p", [CandidateName, DisconnectName]), ok = partisan_peer_service_manager:connect(InitiatorNode), _ = do_send_message(InitiatorNode,{optimization_reply, false, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending optimization reply to Node ~p from ~p", [InitiatorName, CandidateName]), {noreply, State}; %% Replace handle_message({replace, _, OldNode, InitiatorNode, CandidateNode, DisconnectNode}, #state{}=State) -> #{name := DisconnectName} = DisconnectNode, #{name := CandidateName} = CandidateNode, #{name := OldName} = OldNode, ?LOG_DEBUG("XBOT: Received replace message at Node ~p from ~p", [DisconnectName, CandidateName]), Check = is_better(?XPARAM, OldNode, CandidateNode), if not Check -> ok = partisan_peer_service_manager:connect(CandidateNode), _ = do_send_message(CandidateNode,{replace_reply, false, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending replace reply to Node ~p from ~p", [CandidateName, DisconnectName]); true -> ok = partisan_peer_service_manager:connect(OldNode), _ = do_send_message(OldNode,{switch, undefined, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending switch to Node ~p from ~p", [OldName, DisconnectName]) end, {noreply, State}; %% Switch Reply handle_message({switch_reply, true, OldNode, InitiatorNode, CandidateNode, DisconnectNode}, #state{}=State) -> #{name := DisconnectName} = DisconnectNode, #{name := CandidateName} = CandidateNode, #{name := OldName} = OldNode, ?LOG_DEBUG("XBOT: Received switch reply message at Node ~p from ~p", [DisconnectName, OldName]), %remove_from_active_view(CandidateNode, Active), do_disconnect(CandidateNode, State), %add_to_active_view(OldNode, MyTag, State), _ = send_join(OldNode, State), ok = partisan_peer_service_manager:connect(CandidateNode), _ = do_send_message(CandidateNode,{replace_reply, true, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending replace reply to Node ~p from ~p", [CandidateName, DisconnectName]), {noreply, State}; handle_message({switch_reply, false, OldNode, InitiatorNode, CandidateNode, DisconnectNode}, #state{}=State) -> #{name := DisconnectName} = DisconnectNode, #{name := CandidateName} = CandidateNode, #{name := OldName} = OldNode, ?LOG_DEBUG("XBOT: Received switch reply message at Node ~p from ~p", [DisconnectName, OldName]), ok = partisan_peer_service_manager:connect(CandidateNode), _ = do_send_message(CandidateNode, {replace_reply, false, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending replace reply to Node ~p from ~p", [CandidateName, DisconnectName]), {noreply, State}; %% Switch handle_message({switch, _, OldNode, InitiatorNode, CandidateNode, DisconnectNode}, #state{active=Active}=State) -> #{name := DisconnectName} = DisconnectNode, #{name := OldName} = OldNode, ?LOG_DEBUG("XBOT: Received switch message at Node ~p from ~p", [OldName, DisconnectName]), Check = is_in_active_view(InitiatorNode, Active), if Check -> %remove_from_active_view(InitiatorNode, Active), do_disconnect(InitiatorNode, State), %add_to_active_view(DisconnectNode, MyTag, State), _ = send_join(DisconnectNode, State), ok = partisan_peer_service_manager:connect(DisconnectNode), _ = do_send_message(DisconnectNode, {switch_reply, true, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending switch reply to Node ~p from ~p", [DisconnectName, OldName]); true -> ok = partisan_peer_service_manager:connect(DisconnectNode), _ = do_send_message(DisconnectNode, {switch_reply, false, OldNode, InitiatorNode, CandidateNode, DisconnectNode}), ?LOG_DEBUG("XBOT: Sending switch reply to Node ~p from ~p", [DisconnectName, OldName]) end, {noreply, State}. %% Determine if New node is better than Old node based on ping (latency) %% @private is_better(latency, #{name := NewNodeName}, #{name := OldNodeName}) -> is_better_node_by_latency(timer:tc(net_adm, ping, [NewNodeName]), timer:tc(net_adm, ping, [OldNodeName])); is_better(_, _, _) -> true. %% @private is_better_node_by_latency({_, pang}, {_, _}) -> %% if we do not get ping response from new node false; is_better_node_by_latency({_, pong}, {_, pang}) -> %% if we cannot get response from old node but we got response from new (this should never happen, in general) true; is_better_node_by_latency({NewTime, pong}, {OldTime, pong}) -> ?LOG_DEBUG("XBOT: Checking is better - OldTime ~p - NewTime ~p", [OldTime, NewTime]), %% otherwise check lower ping response (OldTime-NewTime) > 0. %% @private select_disconnect_node([H | T]) -> select_worst_in_active_view(T, H). %% @private select_worst_in_active_view([], Worst) -> Worst; select_worst_in_active_view([H | T], Worst) -> Check = is_better(?XPARAM, H, Worst), if Check -> select_worst_in_active_view(T, Worst); true -> select_worst_in_active_view(T, H) end. %% @private send_join(Peer, #state{myself=Myself0, tag=Tag0, epoch=Epoch0}) -> %% Trigger connection. ok = partisan_peer_service_manager:connect(Peer), ?LOG_DEBUG("Node ~p sends the JOIN message to ~p", [Myself0, Peer]), %% Send the JOIN message to the peer. do_send_message(Peer, {join, Myself0, Tag0, Epoch0}). %% Send a disconnect message to a peer %% @private do_disconnect(Peer, #state{active=Active0}=State0) -> case sets:is_element(Peer, Active0) of true -> %% If a member of the active view, remove it. Active = sets:del_element(Peer, Active0), State = add_to_passive_view(Peer, State0#state{active=Active}), ok = disconnect(Peer), {noreply, State}; false -> {noreply, State0} end. %% @private zero_epoch() -> Epoch = 0, persist_epoch(Epoch), Epoch. %% @private data_root() -> case application:get_env(partisan, partisan_data_dir) of {ok, PRoot} -> filename:join(PRoot, "peer_service"); undefined -> undefined end. %% @private write_state_to_disk(Epoch) -> case data_root() of undefined -> ok; Dir -> File = filename:join(Dir, "cluster_state"), ok = filelib:ensure_dir(File), ok = file:write_file(File, term_to_binary(Epoch)) end. %% @private delete_state_from_disk() -> case data_root() of undefined -> ok; Dir -> File = filename:join(Dir, "cluster_state"), ok = filelib:ensure_dir(File), case file:delete(File) of ok -> ?LOG_DEBUG(#{description => "Leaving cluster, removed cluster_state"}); {error, Reason} -> ?LOG_DEBUG("Unable to remove cluster_state for reason ~p", [Reason]) end end. %% @private maybe_load_epoch_from_disk() -> case data_root() of undefined -> zero_epoch(); Dir -> case filelib:is_regular(filename:join(Dir, "cluster_state")) of true -> {ok, Bin} = file:read_file(filename:join(Dir, "cluster_state")), binary_to_term(Bin); false -> zero_epoch() end end. %% @private persist_epoch(Epoch) -> write_state_to_disk(Epoch). %% @private members(Set) -> sets:to_list(Set). %% @private -spec disconnect(Node :: partisan:node_spec()) -> ok. disconnect(Node) -> try partisan_peer_connections:prune(Node) of {_Info, Connections} -> [ begin Pid = partisan_peer_connections:pid(Connection), ?LOG_DEBUG( "disconnecting node ~p by stopping connection pid ~p", [Node, Pid] ), unlink(Pid), _ = catch gen_server:stop(Pid) end || Connection <- Connections ], ok catch error:badarg -> ok end. %% @private -spec do_send_message(Node :: atom() | partisan:node_spec(), Message :: term()) -> ok | {error, disconnected} | {error, not_yet_connected} | {error, term()}. do_send_message(Node, Message) -> do_send_message(Node, Message, #{}). %% @private -spec do_send_message( Node :: atom() | partisan:node_spec(), Message :: term(), Options :: map()) -> ok | {error, disconnected} | {error, not_yet_connected} | {error, term()}. do_send_message(Node, Message, Options) -> Broadcast = partisan_config:get(broadcast, false), Transitive = maps:get(transitive, Options, false), case partisan_peer_connections:dispatch_pid(Node) of {ok, Pid} -> try gen_server:call(Pid, {send_message, Message}) catch Class:EReason -> ?LOG_DEBUG( "failed to send a message to ~p due to ~p:~p", [Node, Class, EReason] ), {error, EReason} end; {error, Reason} -> case Reason of not_yet_connected -> ?LOG_DEBUG(#{ description => "Node not yet connected when sending message to peer node.", message => Message, peer_node => Node, options => #{broadcast => Broadcast, transitive => Transitive} }); disconnected -> ?LOG_DEBUG(#{ description => "Node disconnected when sending message to peer node.", message => Message, peer_node => Node, options => #{broadcast => Broadcast, transitive => Transitive} }) end, case {Broadcast, Transitive} of {true, true} -> do_tree_forward(Node, Message, Options); {true, false} -> ok; {false, _} -> {error, Reason} end end. %% @private select_random(View, Omit) -> List = members(View) -- lists:flatten([Omit]), %% Catch exceptions where there may not be enough members. try Index = rand:uniform(length(List)), lists:nth(Index, List) catch _:_ -> undefined end. %% @private select_random_sublist(View, K) -> List = members(View), lists:sublist(shuffle(List), K). %% @doc Add to the active view. %% %% However, interesting race condition here: if the passive random walk %% timer exceeded and the node was added to the passive view, we might %% also have the active random walk timer exceed *after* because of a %% network delay; if so, we have to remove this element from the passive %% view, otherwise it will exist in both places. %% add_to_active_view(#{name := Node} = Peer, Tag, #state{} = State0) -> #state{ active = Active0, myself = Myself, passive = Passive0, reserved = Reserved0, max_active_size = MaxActiveSize } = State0, IsNotMyself = not (Node =:= partisan:node()), NotInActiveView = not sets:is_element(Peer, Active0), case IsNotMyself andalso NotInActiveView of true -> %% See above for more information. Passive = remove_from_passive_view(Peer, Passive0), State1 = case is_full({active, Active0, Reserved0}, MaxActiveSize) of true -> drop_random_element_from_active_view(State0#state{passive=Passive}); false -> State0#state{passive=Passive} end, ?LOG_DEBUG( "Node ~p adds ~p to active view with tag ~p", [Myself, Peer, Tag] ), %% Add to the active view. Active = sets:add_element(Peer, State1#state.active), %% Fill reserved slot if necessary. Reserved = case dict:find(Tag, Reserved0) of {ok, undefined} -> ?LOG_DEBUG(#{ description => "Node added to reserved slot!" }), dict:store(Tag, Peer, Reserved0); {ok, _} -> %% Slot already filled, treat this as a normal peer. ?LOG_DEBUG(#{ description => "Node added to active view, but reserved slot already full!" }), Reserved0; error -> ?LOG_DEBUG("Tag is not reserved: ~p ~p", [Tag, Reserved0]), Reserved0 end, State2 = State1#state{ active = Active, passive = Passive, reserved = Reserved }, persist_epoch(State2#state.epoch), State2; false -> State0 end. %% @doc Add to the passive view. add_to_passive_view(#{name := Node} = Peer, #state{} = State0) -> #state{ myself = Myself, active = Active0, passive = Passive0, max_passive_size = MaxPassiveSize } = State0, IsNotMyself = not (Node =:= partisan:node()), NotInActiveView = not sets:is_element(Peer, Active0), NotInPassiveView = not sets:is_element(Peer, Passive0), Passive = case IsNotMyself andalso NotInActiveView andalso NotInPassiveView of true -> Passive1 = case is_full({passive, Passive0}, MaxPassiveSize) of true -> Random = select_random(Passive0, [Myself]), sets:del_element(Random, Passive0); false -> Passive0 end, sets:add_element(Peer, Passive1); false -> Passive0 end, State = State0#state{passive=Passive}, persist_epoch(State#state.epoch), State. %% @private is_full({active, Active, Reserved}, MaxActiveSize) -> %% Find the slots that are reserved, but not filled. Open = dict:fold(fun(Key, Value, Acc) -> case Value of undefined -> [Key | Acc]; _ -> Acc end end, [], Reserved), sets:size(Active) + length(Open) >= MaxActiveSize; is_full({passive, Passive}, MaxPassiveSize) -> sets:size(Passive) >= MaxPassiveSize. %% @doc Process of removing a random element from the active view. drop_random_element_from_active_view( #state{myself=Myself0, active=Active0, reserved=Reserved0, epoch=Epoch0, sent_message_map=SentMessageMap0}=State0) -> ReservedPeers = dict:fold(fun(_K, V, Acc) -> [V | Acc] end, [], Reserved0), %% Select random peer, but omit the peers in reserved slots and omit %% ourself from the active view. case select_random(Active0, [Myself0, ReservedPeers]) of undefined -> State0; Peer -> ?LOG_DEBUG("Removing and disconnecting peer: ~p", [Peer]), %% Remove from the active view. Active = sets:del_element(Peer, Active0), %% Add to the passive view. State = add_to_passive_view(Peer, State0#state{active=Active}), %% Trigger connection. ok = partisan_peer_service_manager:connect(Peer), %% Get next disconnect id for the peer. NextId = get_next_id(Peer, Epoch0, SentMessageMap0), %% Update the SentMessageMap. SentMessageMap = dict:store(Peer, NextId, SentMessageMap0), %% Let peer know we are disconnecting them. do_send_message(Peer, {disconnect, Myself0, NextId}), %% Trigger disconnection. ok = disconnect(Peer), ?LOG_DEBUG("Node ~p active view: ~p", [Myself0, members(Active)]), State#state{sent_message_map=SentMessageMap} end. %% @private arwl() -> partisan_config:get(arwl, 6). %% @private prwl() -> partisan_config:get(prwl, 6). %% @private remove_from_passive_view(Peer, Passive) -> sets:del_element(Peer, Passive). %% @private is_in_passive_view(Peer, Passive) -> sets:is_element(Peer, Passive). %% @private remove_from_active_view(Peer, Active) -> sets:del_element(Peer, Active). %% @private is_in_active_view(Peer, Active) -> sets:is_element(Peer, Active). %% @private neighbor_acceptable(Priority, Tag, #state{active=Active, reserved=Reserved, max_active_size=MaxActiveSize}) -> %% Broken down for readability. case Priority of high -> %% Always true. true; _ -> case reserved_slot_available(Tag, Reserved) of true -> %% Always take. true; _ -> %% Otherwise, only if we have a slot available. not is_full({active, Active, Reserved}, MaxActiveSize) end end. %% @private k_active() -> 3. %% @private k_passive() -> 4. %% @private schedule_tree_refresh() -> Period = partisan_config:get(tree_refresh, 1000), erlang:send_after(Period, ?MODULE, tree_refresh). %% @private schedule_passive_view_maintenance() -> Period = partisan_config:get(passive_view_shuffle_period, ?DEFAULT_PASSIVE_VIEW_MAINTENANCE_INTERVAL), erlang:send_after(Period, ?MODULE, passive_view_maintenance). %% @private schedule_xbot_execution() -> Interval = partisan_config:get(xbot_interval), erlang:send_after(Interval, ?MODULE, xbot_execution). %% ----------------------------------------------------------------------------- %% @doc %% http://stackoverflow.com/questions/8817171/shuffling-elements-in-a-list-randomly-re-arrange-list-elements/8820501#8820501 %% @end %% ----------------------------------------------------------------------------- shuffle(L) -> [X || {_, X} <- lists:sort([{rand:uniform(), N} || N <- L])]. %% @private merge_exchange(Exchange, #state{myself=Myself, active=Active}=State0) -> %% Remove ourself and active set members from the exchange. ToAdd = lists:usort(Exchange -- ([Myself] ++ members(Active))), %% Add to passive set. lists:foldl(fun(X, P) -> add_to_passive_view(X, P) end, State0, ToAdd). %% @private notify(#state{active=Active}) -> partisan_peer_service_events:update(Active). %% @private reserved_slot_available(Tag, Reserved) -> case dict:find(Tag, Reserved) of {ok, undefined} -> true; _ -> false end. %% %% @private %%remove_from_reserved(Peer, Reserved) -> %% dict:fold(fun(K, V, Acc) -> %% case V of %% Peer -> %% Acc; %% _ -> %% dict:store(K, V, Acc) %% end %% end, dict:new(), Reserved). %% @private get_current_id(Peer, MessageMap) -> case dict:find(Peer, MessageMap) of {ok, Id} -> Id; error -> %% Default value for the messageId: %% {First start, No disconnect} {1, 0} end. %% @private get_next_id(Peer, MyEpoch, SentMessageMap) -> case dict:find(Peer, SentMessageMap) of {ok, {MyEpoch, Cnt}} -> {MyEpoch, Cnt + 1}; error -> {MyEpoch, 1} end. %% @private is_valid_disconnect(Peer, {DisconnectIdEpoch, DisconnectIdCnt}, AckMessageMap) -> case dict:find(Peer, AckMessageMap) of error -> true; {ok, {Epoch, Cnt}} -> case DisconnectIdEpoch > Epoch of true -> true; false -> DisconnectIdCnt > Cnt end end. %% @private is_addable({DisconnectIdEpoch, DisconnectIdCnt}, Peer, SentMessageMap) -> case dict:find(Peer, SentMessageMap) of error -> true; {ok, {Epoch, Cnt}} -> case DisconnectIdEpoch > Epoch of true -> true; false when DisconnectIdEpoch == Epoch -> DisconnectIdCnt >= Cnt; false -> false end end; is_addable(PeerEpoch, Peer, SentMessageMap) -> case dict:find(Peer, SentMessageMap) of error -> true; {ok, {Epoch, _Cnt}} -> PeerEpoch >= Epoch end. %% @private move_peer_from_passive_to_active(undefined, State) -> State; move_peer_from_passive_to_active(Peer, #state{myself=Myself0, active=Active0, passive=Passive0, tag=Tag0, recv_message_map=RecvMessageMap0}=State0) -> ?LOG_DEBUG("Node ~p sends the NEIGHBOR_REQUEST to ~p", [Myself0, Peer]), Exchange0 = %% Myself. [Myself0] ++ % Random members of the active list. select_random_sublist(Active0, k_active()) ++ %% Random members of the passive list. select_random_sublist(Passive0, k_passive()), Exchange = lists:usort(Exchange0), %% Trigger connection. ok = partisan_peer_service_manager:connect(Peer), LastDisconnectId = get_current_id(Peer, RecvMessageMap0), do_send_message( Peer, {neighbor_request, Myself0, high, Tag0, LastDisconnectId, Exchange} ), State0. %% @private schedule_random_promotion() -> erlang:send_after(?RANDOM_PROMOTION_INTERVAL, ?MODULE, random_promotion). %% @private has_reached_the_limit({active, Active, Reserved}, LimitActiveSize) -> %% Find the slots that are reserved, but not filled. Open = dict:fold(fun(Key, Value, Acc) -> case Value of undefined -> [Key | Acc]; _ -> Acc end end, [], Reserved), sets:size(Active) + length(Open) >= LimitActiveSize. %% @private propagate_partition_injection(Ref, Origin, TTL, Peer) -> ?LOG_DEBUG("Forwarding partition request to: ~p", [Peer]), do_send_message(Peer, {inject_partition, Ref, Origin, TTL}). %% @private propagate_partition_resolution(Reference, Peer) -> ?LOG_DEBUG("Forwarding partition request to: ~p", [Peer]), do_send_message(Peer, {resolve_partition, Reference}). %% @private handle_partition_injection(Reference, _Origin, TTL, #state{active=Active, myself=Myself, partitions=Partitions0}) -> %% If the TTL hasn't expired, re-forward the partition injection %% request. case TTL > 0 of true -> [ propagate_partition_injection(Reference, Myself, TTL - 1, Peer) || Peer <- members(Active) ]; false -> ok end, %% Update partition table marking all immediate neighbors as %% partitioned. Partitions0 ++ lists:map(fun(Peer) -> {Reference, Peer} end, members(Active)). %% @private handle_partition_resolution(Reference, #state{active=Active, partitions=Partitions0}) -> %% Remove partitions. Partitions = lists:foldl(fun({Ref, Peer}, Acc) -> case Reference of Ref -> Acc; _ -> Acc ++ [{Ref, Peer}] end end, [], Partitions0), %% If the list hasn't changed, then don't further propagate %% the message. case Partitions of Partitions0 -> ok; _ -> [ propagate_partition_resolution(Reference, Peer) || Peer <- members(Active) ] end, Partitions. %% @private do_tree_forward(Node, Message, Options) -> ?LOG_DEBUG( "Attempting to forward message from ~p to ~p.", [partisan:node(), Node] ), %% Preempt with user-supplied outlinks. UserOutLinks = maps:get(out_links, Options, undefined), OutLinks = case UserOutLinks of undefined -> try retrieve_outlinks() of Value -> Value catch _:_ -> ?LOG_ERROR(#{description => "Outlinks retrieval failed!"}), [] end; OL -> OL end, %% Send messages, but don't attempt to forward again, if we aren't connected. ok = lists:foreach( fun(N) -> ?LOG_DEBUG( "Forwarding relay message to node ~p for node ~p from node ~p", [N, Node, partisan:node()] ), RelayMessage = {relay_message, Node, Message}, RelayOptions = maps:without([transitive], Options), do_send_message(N, RelayMessage, RelayOptions) end, OutLinks ). %% @private retrieve_outlinks() -> %% Use our tree for the root. Root = partisan:node(), {EagerPeers, _LazyPeers} = partisan_plumtree_broadcast:debug_get_peers(partisan:node(), Root), ordsets:to_list(EagerPeers).