%% ------------------------------------------------------------------- %% %% 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. %% %% ------------------------------------------------------------------- %% @reference https://people.maths.bris.ac.uk/~maajg/scamp-ngc.pdf -module(partisan_scamp_v1_membership_strategy). -author("Christopher S. Meiklejohn "). -behaviour(partisan_membership_strategy). -include("partisan.hrl"). -include("partisan_logger.hrl"). -record(scamp_v1, { actor :: partisan:actor(), membership :: sets:set(partisan:node_spec()), last_message_time :: erlang:timestamp() | undefined }). -type t() :: #scamp_v1{}. -export([init/1]). -export([join/3]). -export([leave/2]). -export([compare/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) -> Membership = sets:add_element( partisan:node_spec(), sets:new([{version, 2}]) ), State = #scamp_v1{ membership = Membership, actor = Identity }, {ok, members(State), State}. %% @doc When a remote node is connected, notify that node to add us. Then, perform forwarding, if necessary. join(Node, #scamp_v1{} = _NodeState, #scamp_v1{} = State0) -> Membership0 = State0#scamp_v1.membership, OutgoingMessages0 = [], %% 1. Add node to our state. ?LOG_TRACE( "~p: Adding node ~p to our membership.", [partisan:node(), Node] ), Membership = sets:add_element(Node, Membership0), %% 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 membership. OutgoingMessages2 = sets:fold( fun(N, OM) -> ?LOG_TRACE( "~p: Forwarding subscription for ~p to node: ~p", [partisan:node(), Node, N] ), OM ++ [{N, {membership_strategy, {forward_subscription, Node}}}] end, OutgoingMessages1, Membership0 ), %% 4. Use 'c' (failure tolerance value) to send forwarded subscriptions for node. C = partisan_config:get(scamp_c, ?SCAMP_C_VALUE), ForwardMessages = lists:map( fun(N) -> ?LOG_TRACE( "~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) ), OutgoingMessages = OutgoingMessages2 ++ ForwardMessages, State = State0#scamp_v1{membership = Membership}, {ok, members(State), OutgoingMessages, State}. %% @doc Leave a node from the cluster. leave(Node, #scamp_v1{membership=Membership0}=State0) -> ?LOG_TRACE( "~p: Issuing remove_subscription for node ~p.", [partisan:node(), Node] ), %% Remove node. Membership = sets:del_element(Node, Membership0), %% Gossip to existing cluster members. Message = {remove_subscription, Node}, OutgoingMessages = lists:map( fun(Peer) -> {Peer, {membership_strategy, Message}} end, members(State0) ), %% Return updated membership. State = State0#scamp_v1{membership = Membership}, Members = members(State), {ok, Members, OutgoingMessages, State}. %% ----------------------------------------------------------------------------- %% @doc Returns the tuple `{Joiners, Leavers}' where `Joiners' is the list of %% node specifications that are elements of `List' but are not in the %% membership set, and `Leavers' are the node specifications for the current %% members that are not elements in `List'. %% @end %% ----------------------------------------------------------------------------- -spec compare(Members :: [partisan:node_spec()], t()) -> {Joiners :: [partisan:node_spec()], Leavers :: [partisan:node_spec()]}. compare(_Members, #scamp_v1{}) -> %% TODO at the momebr this is called only when peer_service:jupdate_members %% is called. We need to define what happens in this case as we maintain a %% partial view of the cluster and Members could be the complete cluster %% view as discovered by partisan_peer_discovery_agent or manually by the %% user. {[], []}. %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- prune(_Nodes, #scamp_v1{} = State) -> %% Not implemented {ok, members(State), State}. %% @doc Periodic protocol maintenance. periodic(#scamp_v1{last_message_time=LastMessageTime} = State) -> SourceNode = partisan:node_spec(), Members = members(State), %% 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, Members), 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, Members, OutgoingSubscriptionMessages ++ OutgoingPingMessages, State}. %% @doc Handling incoming protocol message. handle_message({ping, SourceNode}, State) -> ?LOG_TRACE( "~p: Received ping from node ~p.", [partisan:node(), SourceNode] ), Members = members(State), LastMessageTime = erlang:timestamp(), OutgoingMessages = [], {ok, Members, OutgoingMessages, State#scamp_v1{last_message_time=LastMessageTime}}; %% @doc Handling incoming protocol message. handle_message({remove_subscription, Node}, #scamp_v1{} = State0) -> ?LOG_INFO( "~p: Received remove_subscription for node ~p.", [partisan:node(), Node] ), Membership0 = State0#scamp_v1.membership, case sets:is_element(Node, Membership0) of true -> %% Remove. Membership = sets:del_element(Membership0, Node), %% eqwalizer:ignore State = State0#scamp_v1{membership = Membership}, Members = members(State), %% Gossip removals. Message = {remove_subscription, Node}, OutgoingMessages = lists:map( fun(Peer) -> {Peer, {membership_strategy, Message}} end, Members ), {ok, Members, OutgoingMessages, State}; false -> OutgoingMessages = [], {ok, members(State0), OutgoingMessages, State0} end; handle_message({forward_subscription, Node}, #scamp_v1{} = State0) -> ?LOG_TRACE( "~p: Received subscription for node ~p.", [partisan:node(), Node] ), Membership0 = State0#scamp_v1.membership, Members0 = members(State0), %% Probability: P = 1 / (1 + sizeOf(View)) Random = random_0_or_1(), Keep = trunc((sets:size(Membership0) + 1) * Random), case Keep =:= 0 andalso not lists:member(Node, Members0) of true -> ?LOG_TRACE( "~p: Adding subscription for node: ~p", [partisan:node(), Node] ), Membership = sets:add_element(Node, Membership0), State = State0#scamp_v1{membership=Membership}, Members = members(State), OutgoingMessages = [], {ok, Members, OutgoingMessages, State}; 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, Members0, OutgoingMessages, State0} end. %%%=================================================================== %%% Internal functions %%%=================================================================== %% @private members(#scamp_v1{membership = Membership}) -> sets:to_list(Membership). %% @private select_random_sublist(State, K) -> List = members(State), lists:sublist(shuffle(List), K). %% ----------------------------------------------------------------------------- %% @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.