%% ------------------------------------------------------------------- %% %% 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. %% %% ------------------------------------------------------------------- %% ----------------------------------------------------------------------------- %% @doc This module realises the {@link partisan_peer_service_manager} %% behaviour implementing a peer-to-peer partial mesh topology using the %% Hyparview membership protocol. %% %% == Characteristics == %% %% %% @end %% ----------------------------------------------------------------------------- -module(partisan_hyparview_peer_service_manager). -author("Christopher S. Meiklejohn "). -behaviour(gen_server). -behaviour(partisan_peer_service_manager). -include("partisan_logger.hrl"). -include("partisan.hrl"). -define(DEFAULT_PASSIVE_VIEW_MAINTENANCE_INTERVAL, 10000). -define(RANDOM_PROMOTION_INTERVAL, 5000). %% 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 peer service manager. %% @end %% ----------------------------------------------------------------------------- send_message(Name, Message) -> Cmd = {send_message, Name, Message}, gen_server:call(?MODULE, Cmd, 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(Node, ServerRef, Message, Options) -> FullMessage = {'$gen_cast', Message}, forward_message(Node, 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, Options) when is_map(Options) -> ?LOG_TRACE(#{ description => "About to send message", node => partisan:node(), process => ServerRef, message => Message }), %% We ignore channel -> Why? FullMessage = {forward_message, Node, ServerRef, Message, Options}, %% 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, {forward_message, ServerRef, Message} = FullMessage) -> case partisan_config:get(disable_fast_receive, true) of true -> gen_server:call(?MODULE, {receive_message, Peer, FullMessage}, infinity); false -> partisan_peer_service_manager:process_forward(ServerRef, Message) end; receive_message(Peer, Message) -> ?LOG_TRACE(#{ description => "Manager received message from peer", peer_node => Peer, message => Message }), Result = gen_server:call(?MODULE, {receive_message, Message}, infinity), ?LOG_TRACE(#{ description => "Processed message from peer", peer_node => Peer, message => Message }), Result. %% @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}) -> decode(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(), %% Set logger metadata logger:set_process_metadata(#{node => partisan:node()}), %% Process connection exits. process_flag(trap_exit, true), Epoch = maybe_load_epoch_from_disk(), Myself = myself(), Active = sets:add_element(Myself, sets:new()), Passive = sets:new(), 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 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(#{ description => "Injecting partition", origin => Origin, myself => Myself, ttl => 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(#{ description => "Node active view", myself => Myself, members => 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(#{ description => "Peer connection processes", peer_node => Peer, connection_processes => 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(#{ description => "Sending JOIN message", node => Myself0, peer_node => 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 cast 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) -> %% Get lazily computed outlinks. OutLinks = retrieve_outlinks(), %% 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_info({'EXIT', From, Reason}, State0) -> ?LOG_DEBUG(#{ description => "Connection process died active view", process => From, reason => Reason }), #state{ myself = Myself, active = Active0, passive = Passive0 } = State0, %% 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(#{ description => "Active view", myself => Myself, active_view => members(State#state.active) }), {noreply, State} catch error:badarg -> {noreply, State0} end; handle_info( {connected, Peer, _Channel, _Tag, _PeerEpoch, _RemoteState}, State) -> ?LOG_DEBUG(#{ description => "Node is now connected", peer_node => 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(_Info, Connections) -> lists:foreach( fun(Connection) -> Pid = partisan_peer_connections:pid(Connection), catch gen_server:stop(Pid, normal, infinity), ok end, Connections ) end, partisan_peer_connections:foreach(Fun), ok. %% @private -spec code_change(term() | {down, term()}, state_t(), term()) -> {ok, state_t()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== %% @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(#{ description => "Node is now connected", mysefl => Myself0, peer_node => Peer, peer_epoch => PeerEpoch }), IsAddable = is_addable(PeerEpoch, Peer, SentMessageMap0), NotInActiveView = not sets:is_element(Peer, Active0), State = case IsAddable andalso NotInActiveView of true -> ?LOG_DEBUG(#{ description => "Addding peer node to the active view", peer_node => Peer }), %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), Connected = partisan_peer_connections:is_connected(Peer), case Connected of true -> %% only find the peer connection will add the peer to the active %% 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(#{ description => "Active view", myself => Myself0, active_view => members(State1#state.active) }), %% Notify with event. notify(State1), %% Return. State1; false -> State0 end; false -> ?LOG_DEBUG(#{ description => "Peer node will not be added to the active view", peer_node => Peer }), State0 end, {noreply, State}; %% @private handle_message({neighbor, Peer, PeerTag, DisconnectId, _Sender}, #state{myself=Myself0, sent_message_map=SentMessageMap0}=State0) -> ?LOG_DEBUG(#{ description => "Node received the NEIGHBOR message from peer", myself => Myself0, peer_node => Peer, peer_tag => PeerTag }), State = case is_addable(DisconnectId, Peer, SentMessageMap0) of true -> %% Establish connections. ok = partisan_peer_service_manager:connect(Peer), Connected = partisan_peer_connections:is_connected(Peer), case Connected of true -> %% Add node into the active view. State1 = add_to_active_view( Peer, PeerTag, State0 ), ?LOG_DEBUG(#{ description => "Active view", myself => Myself0, active_view => 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. Message = { neighbor, Myself0, Tag0, LastDisconnectId, Peer }, do_send_message(Peer, Message), ?LOG_DEBUG(#{ description => "Active view", myself => Myself0, active_view => 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 -> ?LOG_DEBUG("FORWARD_JOIN: No node for forward, adding ~p to active view", [Peer]), %% 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. 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. Message = { neighbor, Myself0, Tag0, LastDisconnectId, Peer }, do_send_message(Peer, Message), ?LOG_DEBUG(#{ description => "Active view", myself => Myself0, active_view => members(State3#state.active) }), State3; false -> State0 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]), Message = {forward_join, Peer, PeerTag, PeerEpoch, TTL - 1, Myself0}, %% Forward join. do_send_message(Random, Message), 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(#{ description => "Active view", myself => Myself0, active_view => 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 -> Connected = partisan_peer_connections:is_connected( Peer ), case Connected 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(#{ description => "Active view", myself => Myself0, active_view => members(State1#state.active) }), State1; false -> %% the connections does not change, the peer can not be connected 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, 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, NodeSpec, Message, TTL}, #state{} = State) -> ?LOG_TRACE( "Node ~p received tree relay to ~p", [partisan:node(), NodeSpec] ), OutLinks = State#state.out_links, Active = State#state.active, ActiveMembers = [P || #{name := P} <- members(Active)], case lists:member(NodeSpec, ActiveMembers) of true -> do_send_message( NodeSpec, Message, #{out_links => OutLinks, transitive => true} ); false -> case TTL of 0 -> %% No longer forward. ?LOG_DEBUG( "TTL expired, dropping message for node ~p: ~p", [NodeSpec, Message] ), ok; _ -> Opts = #{out_links => OutLinks}, do_tree_forward(NodeSpec, Message, Opts, TTL), ok end end, {noreply, State}; handle_message({forward_message, ServerRef, Message}, State) -> partisan_peer_service_manager:process_forward(ServerRef, Message), {noreply, State}. %% @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 :: partisan:message()) -> 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 :: partisan:message(), Options :: map()) -> ok | {error, disconnected} | {error, not_yet_connected} | {error, term()}. do_send_message(Node, Message, Options) when is_atom(Node) -> 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} -> TTL = partisan_config:get(relay_ttl, ?RELAY_TTL), do_tree_forward(Node, Message, Options, TTL); {true, false} -> ok; {false, _} -> {error, Reason} end end; do_send_message(#{name := Node}, Message, Options) -> do_send_message(Node, Message, Options). %% @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 := Name}=Peer, Tag, #state{active=Active0, myself=Myself, passive=Passive0, reserved=Reserved0, max_active_size=MaxActiveSize}=State0) -> IsNotMyself = not (Name =:= 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), #state{active=Active1} = 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, Active1), %% 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 := Name}=Peer, #state{myself=Myself, active=Active0, passive=Passive0, max_passive_size=MaxPassiveSize}=State0) -> IsNotMyself = not (Name =:= 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(#{ description => "Active view", myself => Myself0, active_view => 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() -> case partisan_config:get(broadcast, false) of true -> Period = partisan_config:get(tree_refresh, 1000), erlang:send_after(Period, ?MODULE, tree_refresh); false -> ok end. %% @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). %% ----------------------------------------------------------------------------- %% @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}) -> catch 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, TTL) -> MyNode = partisan:node(), ?LOG_TRACE( "Attempting to forward message ~p from ~p to ~p.", [Message, MyNode, 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 _:Reason -> ?LOG_INFO(#{ description => "Outlinks retrieval failed", reason => Reason }), [] end; OL -> OL -- [MyNode] end, %% Send messages, but don't attempt to forward again, if we aren't connected. _ = lists:foreach( fun(N) -> ?LOG_TRACE( "Forwarding relay message ~p to node ~p for node ~p from node ~p", [Message, N, Node, MyNode] ), RelayMessage = {relay_message, Node, Message, TTL - 1}, do_send_message( N, RelayMessage, maps:without([transitive], Options) ) end, OutLinks ), ok. %% @private retrieve_outlinks() -> ?LOG_TRACE(#{description => "About to retrieve outlinks..."}), Root = partisan:node(), OutLinks = try partisan_plumtree_broadcast:debug_get_peers(partisan:node(), Root, 1000) of {EagerPeers, _LazyPeers} -> ordsets:to_list(EagerPeers) catch _:_ -> ?LOG_INFO(#{ description => "Request to get outlinks timed out..." }), [] end, ?LOG_TRACE("Finished getting outlinks: ~p", [OutLinks]), OutLinks -- [partisan:node()].