<commit_before>%  @copyright 2010-2012 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    Passive load balancing using gossip to sample O(log(N)) nodes (at
%%         least lb_psv_samples samples) and choose the
%%         one that reduces the standard deviation the most.
%%         If there is no load address ranges are split according to
%%         lb_psv_split_fallback, i.e. in halves or using the selected keys,
%%         otherwise no more load is moved than is needed to have the average
%%         load on one of the two nodes.
%% @end
%% @version $Id$
-module(lb_psv_gossip).
-author('<EMAIL>').
-vsn('$Id$').

%-define(TRACE(X,Y), log:pal(X,Y)).
-define(TRACE(X,Y), ok).
-define(TRACE_SEND(Pid, Msg), ?TRACE("[ ~.0p ] to ~.0p: ~.0p)~n", [self(), Pid, Msg])).
-define(TRACE1(Msg, State), ?TRACE("[ ~.0p ]~n  Msg: ~.0p~n  State: ~.0p)~n", [self(), Msg, State])).

-behaviour(lb_psv_beh).
-include("lb_psv_beh.hrl").

-ifdef(with_export_type_support).
-export_type([custom_message/0]).
-endif.

-type custom_message() ::
          {gossip_get_values_best_response, gossip_load:load_info()} |
          {get_split_key_response, SplitKey::{?RT:key(), TakenLoad::non_neg_integer()}}.

%% @doc Gets the number of IDs to sample during join.
%%      Note: this is executed at the joining node.
-spec get_number_of_samples(Conn::dht_node_join:connection()) -> ok.
get_number_of_samples({_, ContactNode} = Connection) ->
    comm:send(ContactNode, {join, number_of_samples_request, comm:this(),
                            ?MODULE, Connection}).

%% @doc Sends the number of IDs to sample during join to the joining node.
%%      Note: this is executed at the existing node.
-spec get_number_of_samples_remote(
        SourcePid::comm:mypid(), Conn::dht_node_join:connection()) -> ok.
get_number_of_samples_remote(SourcePid, Connection) ->
    SPid = comm:reply_as(self(), 3,
                         {join, ?MODULE, '_',
                          {get_samples, SourcePid, Connection}}),
    GossipPid = pid_groups:get_my(gossip),
    comm:send_local(GossipPid, {get_values_best, {gossip_load, default}, SPid}).

%% @doc Creates a join operation if a node would join at my node with the
%%      given key. This will simulate the join operation and return a lb_op()
%%      with all the data needed in sort_candidates/1.
%%      Note: this is executed at an existing node.
-spec create_join(DhtNodeState::dht_node_state:state(), SelectedKey::?RT:key(),
                  SourcePid::comm:mypid(), Conn::dht_node_join:connection())
        -> dht_node_state:state().
create_join(DhtNodeState, SelectedKey, SourcePid, Conn) ->
    case dht_node_state:get(DhtNodeState, slide_pred) of
        null ->
            SPid = comm:reply_as(self(), 3,
                                 {join, ?MODULE, '_',
                                  {create_join2, SelectedKey, SourcePid, Conn}}),
            GossipPid = pid_groups:get_my(gossip),
            Msg = {get_values_best, {gossip_load, default}, SPid},
            ?TRACE_SEND(GossipPid, Msg),
            comm:send_local(GossipPid, Msg);
        _ ->
            % postpone message:
            Msg = {join, get_candidate, SourcePid, SelectedKey, ?MODULE, Conn},
            ?TRACE_SEND(SourcePid, Msg),
            _ = comm:send_local_after(100, self(), Msg),
            ok
    end,
    DhtNodeState.

-spec create_join2(DhtNodeState::dht_node_state:state(), SelectedKey::?RT:key(),
                   SourcePid::comm:mypid(), BestValues::gossip_load:load_info(),
                   Conn::dht_node_join:connection()) -> dht_node_state:state().
create_join2(DhtNodeState, SelectedKey, SourcePid, BestValues, Conn) ->
    case dht_node_state:get(DhtNodeState, slide_pred) of
        null ->
            Neighbors = dht_node_state:get(DhtNodeState, neighbors),
            MyNodeId = nodelist:nodeid(Neighbors),
            try
                MyLoad = dht_node_state:get(DhtNodeState, load),
                if MyLoad >= 2 ->
%%                        log:pal("[ ~.0p ] trying split by load", [self()]),
                       TargetLoad =
                           case gossip_load:load_info_get(avgLoad, BestValues) of
                               unknown -> util:floor(MyLoad / 2);
                               AvgLoad when AvgLoad > 0 andalso MyLoad > AvgLoad ->
                                   util:floor(erlang:min(MyLoad - AvgLoad, AvgLoad));
                               _       -> util:floor(MyLoad / 2)
                           end,
%%                        log:pal("T: ~.0p, My: ~.0p, Avg: ~.0p~n", [TargetLoad, MyLoad, gossip_load:load_info_get(avgLoad, BestValues)]),
                       MyPredId = node:id(nodelist:pred(Neighbors)),
                       SPid = comm:reply_as(self(), 3,
                                            {join, ?MODULE, '_',<commit_msg>lb_psv_gossip: complete and fix debugging output<commit_after>%  @copyright 2010-2012 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    Passive load balancing using gossip to sample O(log(N)) nodes (at
%%         least lb_psv_samples samples) and choose the
%%         one that reduces the standard deviation the most.
%%         If there is no load address ranges are split according to
%%         lb_psv_split_fallback, i.e. in halves or using the selected keys,
%%         otherwise no more load is moved than is needed to have the average
%%         load on one of the two nodes.
%% @end
%% @version $Id$
-module(lb_psv_gossip).
-author('<EMAIL>').
-vsn('$Id$').

%-define(TRACE(X,Y), log:pal(X,Y)).
-define(TRACE(X,Y), ok).
-define(TRACE_SEND(Pid, Msg), ?TRACE("[ ~.0p ] to ~.0p: ~.0p)~n", [self(), Pid, Msg])).
-define(TRACE1(Msg, State), ?TRACE("[ ~.0p ]~n  Msg: ~.0p~n  State: ~.0p)~n", [self(), Msg, State])).

-behaviour(lb_psv_beh).
-include("lb_psv_beh.hrl").

-ifdef(with_export_type_support).
-export_type([custom_message/0]).
-endif.

-type custom_message() ::
          {gossip_get_values_best_response, gossip_load:load_info()} |
          {get_split_key_response, SplitKey::{?RT:key(), TakenLoad::non_neg_integer()}}.

%% @doc Gets the number of IDs to sample during join.
%%      Note: this is executed at the joining node.
-spec get_number_of_samples(Conn::dht_node_join:connection()) -> ok.
get_number_of_samples({_, ContactNode} = Connection) ->
    Msg = {join, number_of_samples_request, comm:this(), ?MODULE, Connection},
    ?TRACE_SEND(ContactNode, Msg),
    comm:send(ContactNode, Msg).

%% @doc Sends the number of IDs to sample during join to the joining node.
%%      Note: this is executed at the existing node.
-spec get_number_of_samples_remote(
        SourcePid::comm:mypid(), Conn::dht_node_join:connection()) -> ok.
get_number_of_samples_remote(SourcePid, Connection) ->
    SPid = comm:reply_as(self(), 3,
                         {join, ?MODULE, '_',
                          {get_samples, SourcePid, Connection}}),
    GossipPid = pid_groups:get_my(gossip),
    Msg = {get_values_best, {gossip_load, default}, SPid},
    ?TRACE_SEND(GossipPid, Msg),
    comm:send_local(GossipPid, Msg).

%% @doc Creates a join operation if a node would join at my node with the
%%      given key. This will simulate the join operation and return a lb_op()
%%      with all the data needed in sort_candidates/1.
%%      Note: this is executed at an existing node.
-spec create_join(DhtNodeState::dht_node_state:state(), SelectedKey::?RT:key(),
                  SourcePid::comm:mypid(), Conn::dht_node_join:connection())
        -> dht_node_state:state().
create_join(DhtNodeState, SelectedKey, SourcePid, Conn) ->
    case dht_node_state:get(DhtNodeState, slide_pred) of
        null ->
            SPid = comm:reply_as(self(), 3,
                                 {join, ?MODULE, '_',
                                  {create_join2, SelectedKey, SourcePid, Conn}}),
            GossipPid = pid_groups:get_my(gossip),
            Msg = {get_values_best, {gossip_load, default}, SPid},
            ?TRACE_SEND(GossipPid, Msg),
            comm:send_local(GossipPid, Msg);
        _ ->
            % postpone message:
            Msg = {join, get_candidate, SourcePid, SelectedKey, ?MODULE, Conn},
            ?TRACE_SEND(self(), Msg),
            _ = comm:send_local_after(100, self(), Msg),
            ok
    end,
    DhtNodeState.

-spec create_join2(DhtNodeState::dht_node_state:state(), SelectedKey::?RT:key(),
                   SourcePid::comm:mypid(), BestValues::gossip_load:load_info(),
                   Conn::dht_node_join:connection()) -> dht_node_state:state().
create_join2(DhtNodeState, SelectedKey, SourcePid, BestValues, Conn) ->
    case dht_node_state:get(DhtNodeState, slide_pred) of
        null ->
            Neighbors = dht_node_state:get(DhtNodeState, neighbors),
            MyNodeId = nodelist:nodeid(Neighbors),
            try
                MyLoad = dht_node_state:get(DhtNodeState, load),
                if MyLoad >= 2 ->
%%                        log:pal("[ ~.0p ] trying split by load", [self()]),
                       TargetLoad =
                           case gossip_load:load_info_get(avgLoad, BestValues) of
                               unknown -> util:floor(MyLoad / 2);
                               AvgLoad when AvgLoad > 0 andalso MyLoad > AvgLoad ->
                                   util:floor(erlang:min(MyLoad - AvgLoad, AvgLoad));
                               _       -> util:floor(MyLoad / 2)
                           end,
%%                        log:pal("T: ~.0p, My: ~.0p, Avg: ~.0p~n", [TargetLoad, MyLoad, gossip_load:load_info_get(avgLoad, BestValues)]),
                       MyPredId = node:id(nodelist:pred(Neighbors)),
                       SPid = comm:reply_as(self(), 3,
                                            {join, ?MODULE, '_',
{'fraction_non_alphanumeric': 0.12348956426217503, 'fraction_numerical': 0.006957158549981692, 'mean_word_length': 2.52874677002584, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5793094', 'n_tokens_mistral': 3489, 'n_tokens_neox': 3353, 'n_words': 889}