<commit_before>%  @copyright 2010 Konrad-Zuse-Zentrum fuer Informationstechnik 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    Slide operation structure for node moves, joins and leaves.
%% @end
%% @version $Id$
-module(slide_op).
-author('<EMAIL>').
-vsn('$Id$').

-export([new_receiving_slide/6, new_sending_slide/6,
         new_receiving_slide_join/4, new_sending_slide_join/4,
         is_join/1,
         get_id/1, get_node/1, get_interval/1, get_target_id/1,
         get_source_pid/1, get_tag/1, get_type/1,
         get_timer/1, set_timer/3, reset_timer/1,
         get_timeouts/1, inc_timeouts/1, reset_timeouts/1,
         get_phase/1, set_phase/2]).

-include("scalaris.hrl").
-include("record_helpers.hrl").

-ifdef(with_export_type_support).
-export_type([slide_op/0, id/0, slide_phase/0]).
-endif.

-define(special_tags, ['$join$', '$leave$']).

-type id() :: util:global_uid().

-type slide_phase() ::
        null | % should only occur as an intermediate state, otherwise equal to "no slide op"
        wait_for_succ_ack | % pred initiated a slide and has notified its succ
        wait_for_node_update | % pred changing its id
        wait_for_pred_update | % wait for the local rm process to know about a joining node (pred)
        wait_for_req_data | wait_for_data_ack | wait_for_delta_ack | % sending node
        wait_for_data | wait_for_delta. % receiving node

-record(slide_op, {type       = ?required(slide_op, type)      :: 'send' | 'rcv',
                   id         = ?required(slide_op, id)        :: id(),
                   node       = ?required(slide_op, node)      :: comm:mypid(), % the node, data is sent to/received from
                   interval   = ?required(slide_op, interval)  :: intervals:interval(), % send/receive data in this range
                   target_id  = ?required(slide_op, target_id) :: ?RT:key(), % ID to move the predecessor of the two participating nodes to
                   tag        = ?required(slide_op, tag)       :: any(),
                   source_pid = null          :: comm:erl_local_pid() | null, % pid of the process that requested the move (and will thus receive a message about its state)
                   timer      = {null, nomsg} :: {reference(), comm:message()} | {null, nomsg}, % timeout timer
                   timeouts   = 0             :: non_neg_integer(),
                   phase      = null          :: slide_phase()
                  }).
-opaque slide_op() :: #slide_op{}.

%% @doc Sets up a new slide operation for a joining node (see
%%      dht_node_join.erl). MyKey is the joining node's new Id and will be used
%%      as the target id of the slide operation.
-spec new_receiving_slide_join(MoveId::util:global_uid(), Pred::node:node_type(),
                               Succ::node:node_type(), MyKey::?RT:key())
        -> slide_op().
new_receiving_slide_join(MoveId, Pred, Succ, MyKey) ->
    IntervalToReceive = node:mk_interval_between_ids(node:id(Pred), MyKey),
    #slide_op{type = 'rcv',
              id = MoveId,
              node = node:pidX(Succ),
              interval = IntervalToReceive,
              target_id = MyKey,
              tag = '$join$',
              source_pid = null}.

-spec is_join(SlideOp::slide_op()) -> boolean().
is_join(SlideOp) -> get_tag(SlideOp) =:= '$join$'.

%% @doc Sets up a slide operation where the current node is receiving data from
%%      the given target node. One of the nodes will change its ID to TargetId.
-spec new_receiving_slide(MoveId::util:global_uid(),
                          PredOrSucc::pred | succ, TargetId::?RT:key(), Tag::any(),
                          SourcePid::comm:erl_local_pid() | null,
                          State::dht_node_state:state()) -> slide_op().
new_receiving_slide(MoveId, PredOrSucc, TargetId, Tag, SourcePid, State) ->
    case lists:member(Tag, ?special_tags) of
        true ->
            log:log(warn, "Using reserved tag in receiving slide(~.0p, ~.0p, ~.0p, ~.0p, ~.0p, ~.0p)",
                    [MoveId, PredOrSucc, TargetId, Tag, SourcePid]);
        _ -> ok
    end,
    {IntervalToReceive, TargetNodePid} =
        case PredOrSucc of
            pred ->
                Pred = dht_node_state:get(State, pred),
                Interval = node:mk_interval_between_ids(TargetId, node:id(Pred)),
                {Interval, node:pidX(Pred)};
            succ ->
                Interval = node:mk_interval_between_ids(
                             dht_node_state:get(State, node_id), TargetId),
                {Interval, dht_node_state:get(State, succ_pid)}
        end,
    #slide_op{type = 'rcv',
              id = MoveId,
              node = TargetNodePid,
              interval = IntervalToReceive,
              target_id = TargetId,
              tag = Tag,
              source_pid = SourcePid}.

%% @doc Sets up a slide operation where the current node is sending data to
%%      the node given by PredOrSucc. One of the nodes will change its ID to TargetId.
-spec new_sending_slide(MoveId::util:global_uid(),
                        PredOrSucc::pred | succ, TargetId::?RT:key(), Tag::any(),
                        SourcePid::comm:erl_local_pid() | null,
                        State::dht_node_state:state()) -> slide_op().
new_sending_slide(MoveId, PredOrSucc, TargetId, Tag, SourcePid, State) ->
    case lists:member(Tag, ?special_tags) of
        true ->
            log:log(warn, "Using reserved tag in sending slide(~.0p, ~.0p, ~.0p, ~.0p, ~.0p, ~.0p)",
                    [MoveId, PredOrSucc, TargetId, Tag, SourcePid]);
        _ -> ok
    end,
    {IntervalToSend, TargetNodePid} =
        case PredOrSucc of
            pred ->
                Pred = dht_node_state:get(State, pred),
                Interval = node:mk_interval_between_ids(node:id(Pred), TargetId),
                {Interval, node:pidX(Pred)};
            succ ->
                Interval = node:mk_interval_between_ids(
                             TargetId, dht_node_state:get(State, node_id)),
                {Interval, dht_node_state:get(State, succ_pid)}
        end,
    #slide_op{type = 'send',
              id = MoveId,
              node = TargetNodePid,
              interval = IntervalToSend,
              target_id = TargetId,<commit_msg>- fixed error message in slide_op:new_receiving_slide/6 (parameter State was missing)<commit_after>%  @copyright 2010 Konrad-Zuse-Zentrum fuer Informationstechnik 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    Slide operation structure for node moves, joins and leaves.
%% @end
%% @version $Id$
-module(slide_op).
-author('<EMAIL>').
-vsn('$Id$').

-export([new_receiving_slide/6, new_sending_slide/6,
         new_receiving_slide_join/4, new_sending_slide_join/4,
         is_join/1,
         get_id/1, get_node/1, get_interval/1, get_target_id/1,
         get_source_pid/1, get_tag/1, get_type/1,
         get_timer/1, set_timer/3, reset_timer/1,
         get_timeouts/1, inc_timeouts/1, reset_timeouts/1,
         get_phase/1, set_phase/2]).

-include("scalaris.hrl").
-include("record_helpers.hrl").

-ifdef(with_export_type_support).
-export_type([slide_op/0, id/0, slide_phase/0]).
-endif.

-define(special_tags, ['$join$', '$leave$']).

-type id() :: util:global_uid().

-type slide_phase() ::
        null | % should only occur as an intermediate state, otherwise equal to "no slide op"
        wait_for_succ_ack | % pred initiated a slide and has notified its succ
        wait_for_node_update | % pred changing its id
        wait_for_pred_update | % wait for the local rm process to know about a joining node (pred)
        wait_for_req_data | wait_for_data_ack | wait_for_delta_ack | % sending node
        wait_for_data | wait_for_delta. % receiving node

-record(slide_op, {type       = ?required(slide_op, type)      :: 'send' | 'rcv',
                   id         = ?required(slide_op, id)        :: id(),
                   node       = ?required(slide_op, node)      :: comm:mypid(), % the node, data is sent to/received from
                   interval   = ?required(slide_op, interval)  :: intervals:interval(), % send/receive data in this range
                   target_id  = ?required(slide_op, target_id) :: ?RT:key(), % ID to move the predecessor of the two participating nodes to
                   tag        = ?required(slide_op, tag)       :: any(),
                   source_pid = null          :: comm:erl_local_pid() | null, % pid of the process that requested the move (and will thus receive a message about its state)
                   timer      = {null, nomsg} :: {reference(), comm:message()} | {null, nomsg}, % timeout timer
                   timeouts   = 0             :: non_neg_integer(),
                   phase      = null          :: slide_phase()
                  }).
-opaque slide_op() :: #slide_op{}.

%% @doc Sets up a new slide operation for a joining node (see
%%      dht_node_join.erl). MyKey is the joining node's new Id and will be used
%%      as the target id of the slide operation.
-spec new_receiving_slide_join(MoveId::util:global_uid(), Pred::node:node_type(),
                               Succ::node:node_type(), MyKey::?RT:key())
        -> slide_op().
new_receiving_slide_join(MoveId, Pred, Succ, MyKey) ->
    IntervalToReceive = node:mk_interval_between_ids(node:id(Pred), MyKey),
    #slide_op{type = 'rcv',
              id = MoveId,
              node = node:pidX(Succ),
              interval = IntervalToReceive,
              target_id = MyKey,
              tag = '$join$',
              source_pid = null}.

-spec is_join(SlideOp::slide_op()) -> boolean().
is_join(SlideOp) -> get_tag(SlideOp) =:= '$join$'.

%% @doc Sets up a slide operation where the current node is receiving data from
%%      the given target node. One of the nodes will change its ID to TargetId.
-spec new_receiving_slide(MoveId::util:global_uid(),
                          PredOrSucc::pred | succ, TargetId::?RT:key(), Tag::any(),
                          SourcePid::comm:erl_local_pid() | null,
                          State::dht_node_state:state()) -> slide_op().
new_receiving_slide(MoveId, PredOrSucc, TargetId, Tag, SourcePid, State) ->
    case lists:member(Tag, ?special_tags) of
        true ->
            log:log(warn, "Using reserved tag in receiving slide(~.0p, ~.0p, ~.0p, ~.0p, ~.0p, ~.0p)",
                    [MoveId, PredOrSucc, TargetId, Tag, SourcePid, State]);
        _ -> ok
    end,
    {IntervalToReceive, TargetNodePid} =
        case PredOrSucc of
            pred ->
                Pred = dht_node_state:get(State, pred),
                Interval = node:mk_interval_between_ids(TargetId, node:id(Pred)),
                {Interval, node:pidX(Pred)};
            succ ->
                Interval = node:mk_interval_between_ids(
                             dht_node_state:get(State, node_id), TargetId),
                {Interval, dht_node_state:get(State, succ_pid)}
        end,
    #slide_op{type = 'rcv',
              id = MoveId,
              node = TargetNodePid,
              interval = IntervalToReceive,
              target_id = TargetId,
              tag = Tag,
              source_pid = SourcePid}.

%% @doc Sets up a slide operation where the current node is sending data to
%%      the node given by PredOrSucc. One of the nodes will change its ID to TargetId.
-spec new_sending_slide(MoveId::util:global_uid(),
                        PredOrSucc::pred | succ, TargetId::?RT:key(), Tag::any(),
                        SourcePid::comm:erl_local_pid() | null,
                        State::dht_node_state:state()) -> slide_op().
new_sending_slide(MoveId, PredOrSucc, TargetId, Tag, SourcePid, State) ->
    case lists:member(Tag, ?special_tags) of
        true ->
            log:log(warn, "Using reserved tag in sending slide(~.0p, ~.0p, ~.0p, ~.0p, ~.0p, ~.0p)",
                    [MoveId, PredOrSucc, TargetId, Tag, SourcePid, State]);
        _ -> ok
    end,
    {IntervalToSend, TargetNodePid} =
        case PredOrSucc of
            pred ->
                Pred = dht_node_state:get(State, pred),
                Interval = node:mk_interval_between_ids(node:id(Pred), TargetId),
                {Interval, node:pidX(Pred)};
            succ ->
                Interval = node:mk_interval_between_ids(
                             TargetId, dht_node_state:get(State, node_id)),
                {Interval, dht_node_state:get(State, succ_pid)}
        end,
    #slide_op{type = 'send',
              id = MoveId,
              node = TargetNodePid,
              interval = IntervalToSend,
              target_id = TargetId,
{'fraction_non_alphanumeric': 0.11500875145857643, 'fraction_numerical': 0.006490665110851808, 'mean_word_length': 2.271230916030534, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4249041', 'n_tokens_mistral': 4222, 'n_tokens_neox': 4112, 'n_words': 1116}