%% ------------------------------------------------------------------- %% %% Copyright (c) 2018 Christopher S. 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 %% @reference https://people.maths.bris.ac.uk/~maajg/hiscamp-sigops.pdf %% %% %% ==Isolation prevention == %% %% Protocol has no specific description of how detection of recovery or %% isolation is performed, so we use the protocol specification for that taken %% from the scamp_v1 membership strategy. %% %% @todo Join of InView. %% @todo Node unsubscript. %% @end %% ----------------------------------------------------------------------------- -module(partisan_scamp_v2_membership_strategy). -behaviour(partisan_membership_strategy). -author("Christopher S. Meiklejohn "). -include("partisan.hrl"). -include("partisan_logger.hrl"). -record(scamp_v2, { myself :: partisan:node_spec(), actor :: partisan:actor(), partial_view :: [partisan:node_spec()], in_view :: [partisan:node_spec()], last_message_time :: term() }). -export([init/1]). -export([join/3]). -export([leave/2]). -export([prune/2]). -export([periodic/1]). -export([handle_message/2]). %% ============================================================================= %% API %% ============================================================================= %% @doc Initialize the strategy state. %% Start with an empty state with only ourselves known. init(Identity) -> Myself = partisan:node_spec(), PartialView = [Myself], State = #scamp_v2{ myself = Myself, in_view = [], partial_view = PartialView, actor = Identity }, {ok, PartialView, State}. %% @doc When a remote node is connected, notify that node to add us. Then, perform forwarding, if necessary. join(Node, _NodeState, #scamp_v2{} = State0) -> PartialView0 = State0#scamp_v2.partial_view, OutgoingMessages0 = [], %% 1. Add node to our state. ?LOG_INFO("~p: Adding node ~p to our partial_view.", [partisan:node(), Node]), PartialView = [Node|PartialView0], %% 2. Notify node to add us to its state. %% This is lazily done to ensure we can setup the TCP connection both ways, first. Myself = partisan:node_spec(), OutgoingMessages1 = OutgoingMessages0 ++ [ {Node, {membership_strategy, {forward_subscription, Myself}}} ], %% 3. Notify all members we know about to add node to their partial_view. OutgoingMessages2 = lists:foldl( fun(N, OM) -> ?LOG_DEBUG( "~p: Forwarding subscription for ~p to node: ~p", [partisan:node(), Node, N] ), OM ++ [{N, {membership_strategy, {forward_subscription, Node}}}] end, OutgoingMessages1, PartialView0 ), %% 4. Use 'c - 1' (failure tolerance value) to send forwarded subscriptions for node. %% %% @todo: Ambiguity here: "These forwarded subscriptions may be kept by the neighbours or %% "forwarded, but are not destroyed until some node keeps them?" %% %% What does this mean? %% C = partisan_config:get(scamp_c, ?SCAMP_C_VALUE), ForwardMessages = lists:map(fun(N) -> ?LOG_INFO( "~p: Forwarding additional subscription for ~p to node: ~p", [partisan:node(), Node, N] ), {N, {membership_strategy, {forward_subscription, Node}}} end, select_random_sublist(State0, C - 1)), %% Important difference from scamp_v1: (c - 1) additional copies instead of c! OutgoingMessages = OutgoingMessages2 ++ ForwardMessages, {ok, PartialView, OutgoingMessages, State0#scamp_v2{partial_view=PartialView}}. %% @doc Leave a node from the cluster. leave(Node, #scamp_v2{partial_view=PartialView}=State0) -> ?LOG_INFO("~p: Issuing remove_subscription for node ~p.", [partisan:node(), Node]), %% Begin unsubcription process: send a bootstrap message to the node that is being removed. Message = {bootstrap_remove_subscription, Node}, OutgoingMessages = lists:map(fun(Peer) -> {Peer, {membership_strategy, Message}} end, PartialView), {ok, PartialView, OutgoingMessages, State0}. %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- prune(_Nodes, #scamp_v2{partial_view = PartialView} = State) -> %% Not implemented {ok, PartialView, State}. %% @doc Periodic protocol maintenance. periodic(#scamp_v2{partial_view=PartialView, last_message_time=LastMessageTime}=State) -> SourceNode = partisan:node_spec(), %% Isolation detection: %% %% Since we do not know the rate of message transmission by other nodes in the system, %% periodically transmit a message to all known nodes. Each node will keep track of the %% last message received, and if we don't receive one after X interval, then we know %% we are isolated. OutgoingPingMessages = lists:map(fun(Peer) -> {Peer, {membership_strategy, {ping, SourceNode}}} end, PartialView), Difference = case LastMessageTime of undefined -> 0; _ -> CurrentTime = erlang:timestamp(), timer:now_diff(CurrentTime, LastMessageTime) end, OutgoingSubscriptionMessages = case Difference > (?PERIODIC_INTERVAL * ?SCAMP_MESSAGE_WINDOW) of true -> %% Node is isolated. ?LOG_TRACE("~p: Node is possibly isolated.", [partisan:node()]), Myself = partisan:node_spec(), lists:map(fun(N) -> ?LOG_TRACE( "~p: Forwarding additional subscription for ~p to node: ~p", [partisan:node(), Myself, N] ), {N, {membership_strategy, {forward_subscription, Myself}}} end, select_random_sublist(State, 1)); false -> %% Node is not isolated. [] end, {ok, PartialView, OutgoingSubscriptionMessages ++ OutgoingPingMessages, State}. %% @doc Handling incoming protocol message. handle_message({ping, SourceNode}, #scamp_v2{partial_view=PartialView0}=State) -> ?LOG_TRACE("~p: Received ping from node ~p.", [partisan:node(), SourceNode]), LastMessageTime = erlang:timestamp(), OutgoingMessages = [], {ok, PartialView0, OutgoingMessages, State#scamp_v2{last_message_time=LastMessageTime}}; handle_message({bootstrap_remove_subscription, Node}, #scamp_v2{partial_view=PartialView0, in_view=InView0}=State0) -> ?LOG_TRACE( "~p: Received bootstrap_remove_subscription from node ~p.", [partisan:node(), Node] ), Myself = partisan:node_spec(), C = partisan_config:get(scamp_c, ?SCAMP_C_VALUE), case Node of %% Remove ourselves, but attempt to preserve the scaling relation. Myself -> %% 1. Notify InView[0 - (L - C - 1)] to replace with PartialView[0 - (L - C - 1)] NumToIterate = length(InView0) - (C - 1), ReplacementMessages = case NumToIterate > 0 of true -> lists:map(fun(N) -> Nth = lists:nth(N, InView0), Replacement = lists:nth(N div length(PartialView0), PartialView0), {Nth, {membership_strategy, {replace_subscription, Node, Replacement}}} end, lists:seq(1, NumToIterate)); false -> [] end, %% 2. Notify InView[(L - C - 1) - ] to remove. RemainderToIterate = length(InView0) - NumToIterate, RemovalMessages = case RemainderToIterate > 0 of true -> lists:map(fun(N) -> Nth = lists:nth(N, InView0), {Nth, {membership_strategy, {remove_subscription, Node}}} end, lists:seq(1, RemainderToIterate)); false -> [] end, %% Reset our state. {ok, [], ReplacementMessages ++ RemovalMessages, State0#scamp_v2{in_view=[], partial_view=[]}}; _ -> %% Not us, do nothing. {ok, PartialView0, [], State0} end; handle_message({replace_subscription, Node, Replacement}, #scamp_v2{partial_view=PartialView0}=State0) -> ?LOG_TRACE( "~p: Received replace_subscription for node ~p => ~p.", [partisan:node(), Node, Replacement] ), %% Replacement reorganizes the graphs so that the removed nodes parents connect to %% its children; but, this doesn't update in links, right? Is that missing in the %% protocol description? PartialView = lists:map(fun(N) -> case N of Node -> Replacement; _ -> N end end, PartialView0), {ok, PartialView, [], State0#scamp_v2{partial_view=PartialView}}; handle_message({remove_subscription, Node}, #scamp_v2{partial_view=PartialView0}=State0) -> ?LOG_TRACE("~p: Received remove_subscription for node ~p.", [partisan:node(), Node]), case lists:member(Node, PartialView0) of true -> %% Remove. PartialView = PartialView0 -- [Node], %% Gossip removals. Message = {remove_subscription, Node}, OutgoingMessages = lists:map(fun(Peer) -> {Peer, {membership_strategy, Message}} end, PartialView0), %% Update state. {ok, PartialView, OutgoingMessages, State0#scamp_v2{partial_view=PartialView}}; false -> OutgoingMessages = [], {ok, PartialView0, OutgoingMessages, State0} end; handle_message({forward_subscription, Node}, #scamp_v2{partial_view=PartialView0}=State0) -> ?LOG_TRACE("~p: Received subscription for node ~p.", [partisan:node(), Node]), %% Probability: P = 1 / (1 + sizeOf(View)) Random = random_0_or_1(), Keep = trunc((length(PartialView0) + 1) * Random), case Keep =:= 0 andalso not lists:member(Node, PartialView0) of true -> ?LOG_TRACE("~p: Adding subscription for node: ~p", [partisan:node(), Node]), PartialView = [Node|PartialView0], %% Respond to the node that's joining and tell them to keep us. ?LOG_TRACE( "~p: Notifying ~p to keep us: ~p", [partisan:node(), Node, partisan:node()] ), OutgoingMessages = [{Node, {membership_strategy, {keep_subscription, partisan:node_spec()}}}], {ok, PartialView, OutgoingMessages, State0#scamp_v2{partial_view=PartialView}}; false -> OutgoingMessages = lists:map(fun(N) -> ?LOG_TRACE( "~p: Forwarding subscription for ~p to node: ~p", [partisan:node(), Node, N] ), {N, {membership_strategy, {forward_subscription, Node}}} end, select_random_sublist(State0, 1)), {ok, PartialView0, OutgoingMessages, State0} end; handle_message({keep_subscription, Node}, #scamp_v2{partial_view=PartialView0, in_view=InView0}=State) -> ?LOG_TRACE("~p: Received keep_subscription for node ~p.", [partisan:node(), Node]), InView = [Node|InView0], OutgoingMessages = [], {ok, PartialView0, OutgoingMessages, State#scamp_v2{in_view=InView}}. %%%=================================================================== %%% Internal functions %%%=================================================================== %% @private select_random_sublist(#scamp_v2{partial_view=PartialView}, K) -> lists:sublist(shuffle(PartialView), K). %% ----------------------------------------------------------------------------- %% @private %% @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 random_0_or_1() -> Rand = rand:uniform(10), case Rand >= 5 of true -> 1; false -> 0 end.