<commit_before>%  @copyright 2009-2014 Zuse Institute Berlin

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

%% @author <NAME> <<EMAIL>>
%% @doc    T-Man ring maintenance
%% @end
%% @reference <NAME>, <NAME>. T-Man: Gossip-Based Overlay
%% Topology Management. Engineering Self-Organising Systems 2005:1-15
%% @version $Id$
-module(rm_tman).
-author('<EMAIL>').
-vsn('$Id$').

-include("scalaris.hrl").

-behavior(rm_beh).

-type state_t() :: {Neighbors      :: nodelist:neighborhood(),
                    RandomViewSize :: pos_integer(),
                    Cache          :: [node:node_type()], % random cyclon nodes
                    Churn          :: boolean()}.
-opaque state() :: state_t().

% accepted messages of an initialized rm_tman process in addition to rm_loop
-type(custom_message() ::
    {rm, once, {cy_cache, Cache::[node:node_type()]}} |
    {rm, {cy_cache, Cache::[node:node_type()]}} |
    {rm, {get_node_details_response, NodeDetails::node_details:node_details()}} |
    {rm, buffer, OtherNeighbors::nodelist:neighborhood(), RequestPredsMinCount::non_neg_integer(), RequestSuccsMinCount::non_neg_integer()} |
    {rm, buffer_response, OtherNeighbors::nodelist:neighborhood()}).

-define(SEND_OPTIONS, [{channel, prio}, {?quiet}]).

% note include after the type definitions for erlang < R13B04!
-include("rm_beh.hrl").

-spec get_neighbors(state()) -> nodelist:neighborhood().
get_neighbors({Neighbors, _RandViewSize, _Cache, _Churn}) ->
    Neighbors.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Startup
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% @doc Initializes the cyclon cache retrieval "trigger" (un-infected).
-spec init_first() -> ok.
init_first() ->
    cyclon:get_subset_rand_next_interval(1, comm:reply_as(self(), 2, {rm, '_'})),
    ok.

%% @doc Initialises the state when rm_loop receives an init_rm message.
-spec init(Me::node:node_type(), Pred::node:node_type(),
           Succ::node:node_type()) -> state().
init(Me, Pred, Succ) ->
    Neighborhood = nodelist:new_neighborhood(Pred, Me, Succ),
    % ask cyclon once (a repeating trigger is already started in init_first/0)
    cyclon:get_subset_rand_next_interval(1, comm:reply_as(self(), 2, {rm, once, '_'})),
    {Neighborhood, config:read(cyclon_cache_size), [], true}.

-spec unittest_create_state(Neighbors::nodelist:neighborhood()) -> state().
unittest_create_state(Neighbors) ->
    {Neighbors, 1, [], true}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Message Loop (custom messages not already handled by rm_loop:on/2)
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% @doc Message handler when the module is fully initialized.
-spec handle_custom_message(custom_message(), state())
        -> {ChangeReason::rm_loop:reason(), state()} | unknown_event.
% got empty cyclon cache
handle_custom_message({rm, once, {cy_cache, []}}, State) ->
    % ignore empty cache from cyclon
    {{unknown}, State};

% got cyclon cache
handle_custom_message({rm, once, {cy_cache, NewCache}},
   {Neighborhood, RandViewSize, _Cache, Churn}) ->
    % increase RandViewSize (no error detected):
    RandViewSizeNew =
        case (RandViewSize < config:read(cyclon_cache_size)) of
            true  -> RandViewSize + 1;
            false -> RandViewSize
        end,
    MyRndView = get_RndView(RandViewSizeNew, NewCache),
    OtherNeighborhood =
        nodelist:mk_neighborhood(NewCache, nodelist:node(Neighborhood),
                                 get_pred_list_length(), get_succ_list_length()),
    NewNeighborhood = trigger_update(Neighborhood, MyRndView, OtherNeighborhood),
    {{node_discovery}, {NewNeighborhood, RandViewSizeNew, NewCache, Churn}};

% got cyclon cache (as part of a repeating call)
handle_custom_message({rm, {cy_cache, _NewCache} = CyMsg}, State) ->
    NewState = handle_custom_message({rm, once, CyMsg}, State),
    NewRandViewSize = element(2, element(2, NewState)),
    % trigger new cyclon cache request
    cyclon:get_subset_rand_next_interval(NewRandViewSize,
                                         comm:reply_as(self(), 2, {rm, '_'})),
    NewState;

% got shuffle request
handle_custom_message({rm, buffer, OtherNeighbors, RequestPredsMinCount, RequestSuccsMinCount},
   {Neighborhood, RandViewSize, Cache, Churn}) ->
    MyRndView = get_RndView(RandViewSize, Cache),
    MyView = lists:append(MyRndView, nodelist:to_list(Neighborhood)),
    OtherNode = nodelist:node(OtherNeighbors),
    OtherNodeId = node:id(OtherNode),
    OtherLastPredId = node:id(lists:last(nodelist:preds(OtherNeighbors))),
    OtherLastSuccId = node:id(lists:last(nodelist:succs(OtherNeighbors))),
    % note: the buffer message, esp. OtherNode, might already be outdated
    % and our own view may contain a newer version of the node
    {[OtherNodeUpd], MyViewUpd} = nodelist:lupdate_ids([OtherNode], MyView),
    NeighborsToSendTmp = nodelist:mk_neighborhood(MyViewUpd, OtherNodeUpd,
                                                  get_pred_list_length(),
                                                  get_succ_list_length()),
    NeighborsToSend =
        nodelist:filter_min_length(
          NeighborsToSendTmp,
          fun(N) ->
                  intervals:in(node:id(N), intervals:new('(', OtherNodeId, OtherLastSuccId, ')')) orelse
                      intervals:in(node:id(N), intervals:new('(', OtherLastPredId, OtherNodeId, ')'))
          end,
          RequestPredsMinCount, RequestSuccsMinCount),
    comm:send(node:pidX(nodelist:node(OtherNeighbors)),
              {rm, buffer_response, NeighborsToSend}, ?SEND_OPTIONS),
    NewNeighborhood = trigger_update(Neighborhood, MyRndView, OtherNeighbors),
    {{node_discovery}, {NewNeighborhood, RandViewSize, Cache, Churn}};

handle_custom_message({rm, buffer_response, OtherNeighbors},
   {Neighborhood, RandViewSize, Cache, Churn}) ->
    MyRndView = get_RndView(RandViewSize, Cache),
    NewNeighborhood = trigger_update(Neighborhood, MyRndView, OtherNeighbors),
    % increase RandViewSize (no error detected):
    NewRandViewSize =
        case RandViewSize < config:read(cyclon_cache_size) of
            true ->  RandViewSize + 1;
            false -> RandViewSize
        end,
    {{node_discovery}, {NewNeighborhood, NewRandViewSize, Cache, Churn}};

% we asked another node we wanted to add for its node object -> now add it
% (if it is not in the process of leaving the system)
handle_custom_message({rm, {get_node_details_response, NodeDetails}}, State) ->
    case node_details:get(NodeDetails, is_leaving) of
        false ->
            NewState =
                update_nodes(State, [node_details:get(NodeDetails, node)], [], null),
            {{node_discovery}, NewState};
        true ->
            {{unknown}, State}
    end;

handle_custom_message(_, _State) -> unknown_event.

-spec trigger_action(State::state_t())
        -> {ChangeReason::rm_loop:reason(), state()}.
trigger_action({Neighborhood, RandViewSize, Cache, Churn} = State) ->
    % Trigger an update of the Random view
    % Test for being alone:
    case nodelist:has_real_pred(Neighborhood) andalso
             nodelist:has_real_succ(Neighborhood) of
        false -> % our node is the only node in the system
            % no need to set a new trigger - we will be actively called by
            % any new node and set the trigger then (see handling of
            % notify_new_succ and notify_new_pred)
            {{unknown}, State};
        _ -> % there is another node in the system
            RndView = get_RndView(RandViewSize, Cache),
            %log:log(debug, " [RM | ~p ] RNDVIEW: ~p", [self(),RndView]),
            {Pred, Succ} = get_safe_pred_succ(Neighborhood, RndView),
            %io:format("~p~n",[{Preds,Succs,RndView,Me}]),
            RequestPredsMinCount =
                case nodelist:has_real_pred(Neighborhood) of
                    true -> get_pred_list_length() - length(nodelist:preds(Neighborhood));
                    _    -> get_pred_list_length()
                end,
            RequestSuccsMinCount =
                case nodelist:has_real_succ(Neighborhood) of
                    true -> get_succ_list_length() - length(nodelist:succs(Neighborhood));<commit_msg>rm_tman: make dialyzer happy about opaque types<commit_after>%  @copyright 2009-2014 Zuse Institute Berlin

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

%% @author <NAME> <<EMAIL>>
%% @doc    T-Man ring maintenance
%% @end
%% @reference <NAME>, <NAME>. T-Man: Gossip-Based Overlay
%% Topology Management. Engineering Self-Organising Systems 2005:1-15
%% @version $Id$
-module(rm_tman).
-author('<EMAIL>').
-vsn('$Id$').

-include("scalaris.hrl").

-behavior(rm_beh).

-type state_t() :: {Neighbors      :: nodelist:neighborhood(),
                    RandomViewSize :: pos_integer(),
                    Cache          :: [node:node_type()], % random cyclon nodes
                    Churn          :: boolean()}.
-opaque state() :: state_t().

% accepted messages of an initialized rm_tman process in addition to rm_loop
-type(custom_message() ::
    {rm, once, {cy_cache, Cache::[node:node_type()]}} |
    {rm, {cy_cache, Cache::[node:node_type()]}} |
    {rm, {get_node_details_response, NodeDetails::node_details:node_details()}} |
    {rm, buffer, OtherNeighbors::nodelist:neighborhood(), RequestPredsMinCount::non_neg_integer(), RequestSuccsMinCount::non_neg_integer()} |
    {rm, buffer_response, OtherNeighbors::nodelist:neighborhood()}).

-define(SEND_OPTIONS, [{channel, prio}, {?quiet}]).

% note include after the type definitions for erlang < R13B04!
-include("rm_beh.hrl").

-spec get_neighbors(state()) -> nodelist:neighborhood().
get_neighbors({Neighbors, _RandViewSize, _Cache, _Churn}) ->
    Neighbors.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Startup
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% @doc Initializes the cyclon cache retrieval "trigger" (un-infected).
-spec init_first() -> ok.
init_first() ->
    cyclon:get_subset_rand_next_interval(1, comm:reply_as(self(), 2, {rm, '_'})),
    ok.

%% @doc Initialises the state when rm_loop receives an init_rm message.
-spec init(Me::node:node_type(), Pred::node:node_type(),
           Succ::node:node_type()) -> state().
init(Me, Pred, Succ) ->
    Neighborhood = nodelist:new_neighborhood(Pred, Me, Succ),
    % ask cyclon once (a repeating trigger is already started in init_first/0)
    cyclon:get_subset_rand_next_interval(1, comm:reply_as(self(), 2, {rm, once, '_'})),
    {Neighborhood, config:read(cyclon_cache_size), [], true}.

-spec unittest_create_state(Neighbors::nodelist:neighborhood()) -> state().
unittest_create_state(Neighbors) ->
    {Neighbors, 1, [], true}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Message Loop (custom messages not already handled by rm_loop:on/2)
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% @doc Message handler when the module is fully initialized.
-spec handle_custom_message(custom_message(), state())
        -> {ChangeReason::rm_loop:reason(), state()} | unknown_event.
% got empty cyclon cache
handle_custom_message({rm, once, {cy_cache, NewCache}}, State) ->
    add_cyclon_cache(NewCache, State);

% got cyclon cache (as part of a repeating call)
handle_custom_message({rm, {cy_cache, NewCache}}, State) ->
    NewState = add_cyclon_cache(NewCache, State),
    NewRandViewSize = element(2, element(2, NewState)),
    % trigger new cyclon cache request
    cyclon:get_subset_rand_next_interval(NewRandViewSize,
                                         comm:reply_as(self(), 2, {rm, '_'})),
    NewState;

% got shuffle request
handle_custom_message({rm, buffer, OtherNeighbors, RequestPredsMinCount, RequestSuccsMinCount},
   {Neighborhood, RandViewSize, Cache, Churn}) ->
    MyRndView = get_RndView(RandViewSize, Cache),
    MyView = lists:append(MyRndView, nodelist:to_list(Neighborhood)),
    OtherNode = nodelist:node(OtherNeighbors),
    OtherNodeId = node:id(OtherNode),
    OtherLastPredId = node:id(lists:last(nodelist:preds(OtherNeighbors))),
    OtherLastSuccId = node:id(lists:last(nodelist:succs(OtherNeighbors))),
    % note: the buffer message, esp. OtherNode, might already be outdated
    % and our own view may contain a newer version of the node
    {[OtherNodeUpd], MyViewUpd} = nodelist:lupdate_ids([OtherNode], MyView),
    NeighborsToSendTmp = nodelist:mk_neighborhood(MyViewUpd, OtherNodeUpd,
                                                  get_pred_list_length(),
                                                  get_succ_list_length()),
    NeighborsToSend =
        nodelist:filter_min_length(
          NeighborsToSendTmp,
          fun(N) ->
                  intervals:in(node:id(N), intervals:new('(', OtherNodeId, OtherLastSuccId, ')')) orelse
                      intervals:in(node:id(N), intervals:new('(', OtherLastPredId, OtherNodeId, ')'))
          end,
          RequestPredsMinCount, RequestSuccsMinCount),
    comm:send(node:pidX(nodelist:node(OtherNeighbors)),
              {rm, buffer_response, NeighborsToSend}, ?SEND_OPTIONS),
    NewNeighborhood = trigger_update(Neighborhood, MyRndView, OtherNeighbors),
    {{node_discovery}, {NewNeighborhood, RandViewSize, Cache, Churn}};

handle_custom_message({rm, buffer_response, OtherNeighbors},
   {Neighborhood, RandViewSize, Cache, Churn}) ->
    MyRndView = get_RndView(RandViewSize, Cache),
    NewNeighborhood = trigger_update(Neighborhood, MyRndView, OtherNeighbors),
    % increase RandViewSize (no error detected):
    NewRandViewSize =
        case RandViewSize < config:read(cyclon_cache_size) of
            true ->  RandViewSize + 1;
            false -> RandViewSize
        end,
    {{node_discovery}, {NewNeighborhood, NewRandViewSize, Cache, Churn}};

% we asked another node we wanted to add for its node object -> now add it
% (if it is not in the process of leaving the system)
handle_custom_message({rm, {get_node_details_response, NodeDetails}}, State) ->
    case node_details:get(NodeDetails, is_leaving) of
        false ->
            NewState =
                update_nodes(State, [node_details:get(NodeDetails, node)], [], null),
            {{node_discovery}, NewState};
        true ->
            {{unknown}, State}
    end;

handle_custom_message(_, _State) -> unknown_event.

%% @doc Integrates a non-empty cyclon cache into the own random view (empty
%%      cyclon caches are ignored) and updates the random view size accordingly.
-spec add_cyclon_cache(Cache::[node:node_type()], state_t())
        -> {ChangeReason::{unknown} | {node_discovery}, state_t()}.
add_cyclon_cache([], State) ->
    % ignore empty cache from cyclon
    {{unknown}, State};
add_cyclon_cache(NewCache, {Neighborhood, RandViewSize, _Cache, Churn}) ->
    % increase RandViewSize (no error detected):
    RandViewSizeNew =
        case (RandViewSize < config:read(cyclon_cache_size)) of
            true  -> RandViewSize + 1;
            false -> RandViewSize
        end,
    MyRndView = get_RndView(RandViewSizeNew, NewCache),
    OtherNeighborhood =
        nodelist:mk_neighborhood(NewCache, nodelist:node(Neighborhood),
                                 get_pred_list_length(), get_succ_list_length()),
    NewNeighborhood = trigger_update(Neighborhood, MyRndView, OtherNeighborhood),
    {{node_discovery}, {NewNeighborhood, RandViewSizeNew, NewCache, Churn}}.

-spec trigger_action(State::state_t())
        -> {ChangeReason::rm_loop:reason(), state()}.
trigger_action({Neighborhood, RandViewSize, Cache, Churn} = State) ->
    % Trigger an update of the Random view
    % Test for being alone:
    case nodelist:has_real_pred(Neighborhood) andalso
             nodelist:has_real_succ(Neighborhood) of
        false -> % our node is the only node in the system
            % no need to set a new trigger - we will be actively called by
            % any new node and set the trigger then (see handling of
            % notify_new_succ and notify_new_pred)
            {{unknown}, State};
        _ -> % there is another node in the system
            RndView = get_RndView(RandViewSize, Cache),
            %log:log(debug, " [RM | ~p ] RNDVIEW: ~p", [self(),RndView]),
            {Pred, Succ} = get_safe_pred_succ(Neighborhood, RndView),
            %io:format("~p~n",[{Preds,Succs,RndView,Me}]),
            RequestPredsMinCount =
                case nodelist:has_real_pred(Neighborhood) of
                    true -> get_pred_list_length() - length(nodelist:preds(Neighborhood));
                    _    -> get_pred_list_length()
                end,
            RequestSuccsMinCount =
                case nodelist:has_real_succ(Neighborhood) of
                    true -> get_succ_list_length() - length(nodelist:succs(Neighborhood));
{'fraction_non_alphanumeric': 0.1470522554711925, 'fraction_numerical': 0.003907994640464493, 'mean_word_length': 3.566148355850115, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2797609', 'n_tokens_mistral': 5586, 'n_tokens_neox': 5277, 'n_words': 1296}