partisan

%% -------------------------------------------------------------------
%%
%% 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/hiscamp-sigops.pdf

%% @todo Join of InView.
%% @todo Node unsubscript.

-module(partisan_scamp_v2_membership_strategy).

-author("Christopher S. Meiklejohn <christopher.meiklejohn@gmail.com>").

-behaviour(partisan_membership_strategy).

-include("partisan.hrl").
-include("partisan_logger.hrl").

-export([init/1,
         join/3,
         leave/2,
         prune/2,
         periodic/1,
         handle_message/2]).

%% 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.



-record(scamp_v2, {actor :: term(),
                   partial_view :: [node_spec()],
                   in_view :: [node_spec()],
                   last_message_time :: term()}).

%%%===================================================================
%%% API
%%%===================================================================

%% @doc Initialize the strategy state.
%%      Start with an empty state with only ourselves known.
init(Identity) ->
    PartialView = [myself()],
    InView = [],
    State = #scamp_v2{in_view=InView, 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(#scamp_v2{partial_view=PartialView0}=State0, Node, _NodeState) ->
    OutgoingMessages0 = [],

    %% 1. Add node to our state.
    ?LOG_INFO("~p: Adding node ~p to our partial_view.", [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_INFO(
            "~p: Forwarding subscription for ~p to node: ~p", [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",
            [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(#scamp_v2{partial_view=PartialView}=State0, Node) ->
    ?LOG_INFO("~p: Issuing remove_subscription for node ~p.", [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(#scamp_v2{partial_view = PartialView} = State, _Nodes) ->
    %% Not implemented
    {ok, PartialView, State}.


%% @doc Periodic protocol maintenance.
periodic(#scamp_v2{partial_view=PartialView, last_message_time=LastMessageTime}=State) ->
    SourceNode = myself(),

    %% 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 possibily isolated.", [node()]),

            Myself = myself(),

            lists:map(fun(N) ->
                ?LOG_TRACE(
                    "~p: Forwarding additional subscription for ~p to node: ~p",
                    [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(#scamp_v2{partial_view=PartialView0}=State, {ping, SourceNode}) ->
    ?LOG_TRACE("~p: Received ping from node ~p.", [node(), SourceNode]),

    LastMessageTime = erlang:timestamp(),
    OutgoingMessages = [],
    {ok, PartialView0, OutgoingMessages, State#scamp_v2{last_message_time=LastMessageTime}};
handle_message(#scamp_v2{partial_view=PartialView0, in_view=InView0}=State0, {bootstrap_remove_subscription, Node}) ->
    ?LOG_TRACE(
        "~p: Received bootstrap_remove_subscription from node ~p.",
        [node(), Node]
    ),

    Myself = myself(),
    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(#scamp_v2{partial_view=PartialView0}=State0, {replace_subscription, Node, Replacement}) ->
    ?LOG_TRACE(
        "~p: Received replace_subscription for node ~p => ~p.",
        [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(#scamp_v2{partial_view=PartialView0}=State0, {remove_subscription, Node}) ->
    ?LOG_TRACE("~p: Received remove_subscription for node ~p.", [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(#scamp_v2{partial_view=PartialView0}=State0, {forward_subscription, Node}) ->
    ?LOG_TRACE("~p: Received subscription for node ~p.", [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", [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", [node(), Node, node()]
            ),
            OutgoingMessages = [{Node, {membership_strategy, {keep_subscription, myself()}}}],

            {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",
                    [node(), Node, N]
                ),
                {N, {membership_strategy, {forward_subscription, Node}}}
            end, select_random_sublist(State0, 1)),
            {ok, PartialView0, OutgoingMessages, State0}
    end;
handle_message(#scamp_v2{partial_view=PartialView0, in_view=InView0}=State, {keep_subscription, Node}) ->
    ?LOG_TRACE("~p: Received keep_subscription for node ~p.", [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.

%% @private
myself() ->
    partisan:node_spec().