<commit_before>% @copyright 2009-2012 Zuse Institute Berlin,
%            2009 onScale solutions GmbH

%   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 Part of a generic implementation of transactions using Paxos Commit -
%%      the roles of the (replicated) transaction manager TM and RTM.
%% @end
%% @version $Id$
-module(tx_tm_rtm).
-author('<EMAIL>').
-vsn('$Id$ ').

%%-define(TRACE_RTM_MGMT(X,Y), io:format(X,Y)).
%%-define(TRACE_RTM_MGMT(X,Y), ct:pal(X,Y)).
-define(TRACE_RTM_MGMT(X,Y), ok).
%-define(TRACE(X,Y), ct:pal(X,Y)).
-define(TRACE(_X,_Y), ok).
-behaviour(gen_component).
-include("scalaris.hrl").
-include("client_types.hrl").

%% public interface for transaction validation using Paxos-Commit.
-export([commit/4]).
-export([msg_commit_reply/3]).
-export([rm_send_update/4]).

%% functions for gen_component module, supervisor callbacks and config
-export([start_link/2]).
-export([on/2, init/1]).
-export([on_init/2]).
-export([get_my/2]).
-export([check_config/0]).

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

%% messages a client has to expect when using this module
-spec msg_commit_reply(comm:mypid(), any(), any()) -> ok.
msg_commit_reply(Client, ClientsID, Result) ->
    comm:send(Client, {tx_tm_rtm_commit_reply, ClientsID, Result}).

-spec msg_tp_do_commit_abort(comm:mypid(), any(), ?commit | ?abort) -> ok.
msg_tp_do_commit_abort(TP, Id, Result) ->
    comm:send(TP, {?tp_do_commit_abort, Id, Result}).

%% public interface for transaction validation using Paxos-Commit.
%% ClientsID may be nil, its not used by tx_tm. It will be repeated in
%% replies to allow to map replies to the right requests in the
%% client.
-spec commit(comm:erl_local_pid(), comm:mypid(), any(), tx_tlog:tlog()) -> ok.
commit(TM, Client, ClientsID, TLog) ->
    Msg = {tx_tm_rtm_commit, Client, ClientsID, TLog},
    comm:send_local(TM, Msg).

%% @doc Notifies the tx_tm_rtm of a changed node ID.
-spec rm_send_update(Subscriber::pid(), Tag::?MODULE,
                     OldNeighbors::nodelist:neighborhood(),
                     NewNeighbors::nodelist:neighborhood()) -> ok.
rm_send_update(Pid, ?MODULE, OldNeighbors, NewNeighbors) ->
    OldId = node:id(nodelist:node(OldNeighbors)),
    NewId = node:id(nodelist:node(NewNeighbors)),
    case OldId =/= NewId of
        true  -> comm:send_local(Pid, {new_node_id, NewId});
        false -> ok
    end.

%% be startable via supervisor, use gen_component
-spec start_link(pid_groups:groupname(), any()) -> {ok, pid()}.
start_link(DHTNodeGroup, Role) ->
    gen_component:start_link(?MODULE, fun ?MODULE:on/2,
                             [],
                             [{pid_groups_join_as, DHTNodeGroup, Role}]).

-type rtm() :: {?RT:key(),
                {comm:mypid()} | unknown,
                Role :: 0..3, %% non_neg_integer(),
                {Acceptor :: comm:mypid()} | unknown}.

-type rtms() :: [rtm()].

-type state() ::
    {RTMs           :: rtms(),
     TableName      :: pdb:tableid(),
     Role           :: pid_groups:pidname(),
     LocalAcceptor  :: pid(),
     GLocalLearner  :: comm:mypid(),
     OpenTxNum      :: non_neg_integer()}.

%% initialize: return initial state.
-spec init([]) -> state() | {'$gen_component', [{on_handler, gen_component:handler()},...],
                             state()}.
init([]) ->
    Role = pid_groups:my_pidname(),
    DHTNodeGroup = pid_groups:my_groupname(),
    ?TRACE("tx_tm_rtm:init for instance: ~p ~p~n", [DHTNodeGroup, Role]),
    %% For easier debugging, use a named table (generates an atom)
    Table = pdb:new(DHTNodeGroup ++ "_tx_tm_rtm_" ++ atom_to_list(Role),
                    [set, protected, named_table]),
    %% use random table name provided by ets to *not* generate an atom
    %% Table = pdb:new(?MODULE, [set, private]),
    LAcceptor = get_my(Role, acceptor),
    GLLearner = comm:make_global(get_my(Role, learner)),
    %% start getting rtms and maintain them.
    case Role of
        tx_tm ->
            comm:send_local(self(), {get_node_details}),
            State = {_RTMs = [], Table, Role, LAcceptor, GLLearner, 0},
            %% subscribe to id changes
            rm_loop:subscribe(self(), ?MODULE,
                              fun rm_loop:subscribe_dneighbor_change_filter/3,
                              fun ?MODULE:rm_send_update/4, inf),
            gen_component:change_handler(State, fun ?MODULE:on_init/2);
        _ -> {_RTMs = [], Table, Role, LAcceptor, GLLearner, 0}
    end.

-spec on(comm:message(), state()) -> state().
%% a TP comitted/roled back the item as requested
on({?tp_committed, ItemId} = _Msg, State) ->
    ?TRACE("tx_tm_rtm:on(~p)~n", [_Msg]),

    %% retrieve the item and handle the commit ack therein
    {OrigItemResult, ItemState} = get_item_entry(ItemId, State),
    case OrigItemResult of
        new -> State;
        _ ->
            {ItemResult, NewItemState} = tx_item_state:add_commit_ack(ItemState),
            _ = set_entry(NewItemState, State),
            %% retrieve the tx entry and increment the number of commit acks
            TxId = tx_item_state:get_txid(NewItemState),
            {_, TxState} = get_tx_entry(TxId, State),
            _ = case ItemResult of
                    state_updated ->
                        _ = set_entry(TxState, State),
                        trigger_delete_if_done(TxState, State);
                    item_newly_safe ->
                        {TxResult, TmpTxState} = tx_state:add_item_acked(TxState),
                        NewTxState =<commit_msg>tx_tm: fixed missing {update_RTMs} periodic message NK&FS<commit_after>% @copyright 2009-2012 Zuse Institute Berlin,
%            2009 onScale solutions GmbH

%   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 Part of a generic implementation of transactions using Paxos Commit -
%%      the roles of the (replicated) transaction manager TM and RTM.
%% @end
%% @version $Id$
-module(tx_tm_rtm).
-author('<EMAIL>').
-vsn('$Id$ ').

%%-define(TRACE_RTM_MGMT(X,Y), io:format(X,Y)).
%%-define(TRACE_RTM_MGMT(X,Y), ct:pal(X,Y)).
-define(TRACE_RTM_MGMT(X,Y), ok).
%-define(TRACE(X,Y), ct:pal(X,Y)).
-define(TRACE(_X,_Y), ok).
-behaviour(gen_component).
-include("scalaris.hrl").
-include("client_types.hrl").

%% public interface for transaction validation using Paxos-Commit.
-export([commit/4]).
-export([msg_commit_reply/3]).
-export([rm_send_update/4]).

%% functions for gen_component module, supervisor callbacks and config
-export([start_link/2]).
-export([on/2, init/1]).
-export([on_init/2]).
-export([get_my/2]).
-export([check_config/0]).

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

%% messages a client has to expect when using this module
-spec msg_commit_reply(comm:mypid(), any(), any()) -> ok.
msg_commit_reply(Client, ClientsID, Result) ->
    comm:send(Client, {tx_tm_rtm_commit_reply, ClientsID, Result}).

-spec msg_tp_do_commit_abort(comm:mypid(), any(), ?commit | ?abort) -> ok.
msg_tp_do_commit_abort(TP, Id, Result) ->
    comm:send(TP, {?tp_do_commit_abort, Id, Result}).

%% public interface for transaction validation using Paxos-Commit.
%% ClientsID may be nil, its not used by tx_tm. It will be repeated in
%% replies to allow to map replies to the right requests in the
%% client.
-spec commit(comm:erl_local_pid(), comm:mypid(), any(), tx_tlog:tlog()) -> ok.
commit(TM, Client, ClientsID, TLog) ->
    Msg = {tx_tm_rtm_commit, Client, ClientsID, TLog},
    comm:send_local(TM, Msg).

%% @doc Notifies the tx_tm_rtm of a changed node ID.
-spec rm_send_update(Subscriber::pid(), Tag::?MODULE,
                     OldNeighbors::nodelist:neighborhood(),
                     NewNeighbors::nodelist:neighborhood()) -> ok.
rm_send_update(Pid, ?MODULE, OldNeighbors, NewNeighbors) ->
    OldId = node:id(nodelist:node(OldNeighbors)),
    NewId = node:id(nodelist:node(NewNeighbors)),
    case OldId =/= NewId of
        true  -> comm:send_local(Pid, {new_node_id, NewId});
        false -> ok
    end.

%% be startable via supervisor, use gen_component
-spec start_link(pid_groups:groupname(), any()) -> {ok, pid()}.
start_link(DHTNodeGroup, Role) ->
    gen_component:start_link(?MODULE, fun ?MODULE:on/2,
                             [],
                             [{pid_groups_join_as, DHTNodeGroup, Role}]).

-type rtm() :: {?RT:key(),
                {comm:mypid()} | unknown,
                Role :: 0..3, %% non_neg_integer(),
                {Acceptor :: comm:mypid()} | unknown}.

-type rtms() :: [rtm()].

-type state() ::
    {RTMs           :: rtms(),
     TableName      :: pdb:tableid(),
     Role           :: pid_groups:pidname(),
     LocalAcceptor  :: pid(),
     GLocalLearner  :: comm:mypid(),
     OpenTxNum      :: non_neg_integer()}.

%% initialize: return initial state.
-spec init([]) -> state() | {'$gen_component', [{on_handler, gen_component:handler()},...],
                             state()}.
init([]) ->
    Role = pid_groups:my_pidname(),
    DHTNodeGroup = pid_groups:my_groupname(),
    ?TRACE("tx_tm_rtm:init for instance: ~p ~p~n", [DHTNodeGroup, Role]),
    %% For easier debugging, use a named table (generates an atom)
    Table = pdb:new(DHTNodeGroup ++ "_tx_tm_rtm_" ++ atom_to_list(Role),
                    [set, protected, named_table]),
    %% use random table name provided by ets to *not* generate an atom
    %% Table = pdb:new(?MODULE, [set, private]),
    LAcceptor = get_my(Role, acceptor),
    GLLearner = comm:make_global(get_my(Role, learner)),
    %% start getting rtms and maintain them.
    case Role of
        tx_tm ->
            comm:send_local(self(), {get_node_details}),
            State = {_RTMs = [], Table, Role, LAcceptor, GLLearner, 0},
            rtm_update(state_get_RTMs(State),
                       config:read(tx_rtm_update_interval) div 1000,
                       {update_RTMs}),
            %% subscribe to id changes
            rm_loop:subscribe(self(), ?MODULE,
                              fun rm_loop:subscribe_dneighbor_change_filter/3,
                              fun ?MODULE:rm_send_update/4, inf),
            gen_component:change_handler(State, fun ?MODULE:on_init/2);
        _ -> {_RTMs = [], Table, Role, LAcceptor, GLLearner, 0}
    end.

-spec on(comm:message(), state()) -> state().
%% a TP comitted/roled back the item as requested
on({?tp_committed, ItemId} = _Msg, State) ->
    ?TRACE("tx_tm_rtm:on(~p)~n", [_Msg]),

    %% retrieve the item and handle the commit ack therein
    {OrigItemResult, ItemState} = get_item_entry(ItemId, State),
    case OrigItemResult of
        new -> State;
        _ ->
            {ItemResult, NewItemState} = tx_item_state:add_commit_ack(ItemState),
            _ = set_entry(NewItemState, State),
            %% retrieve the tx entry and increment the number of commit acks
            TxId = tx_item_state:get_txid(NewItemState),
            {_, TxState} = get_tx_entry(TxId, State),
            _ = case ItemResult of
                    state_updated ->
                        _ = set_entry(TxState, State),
                        trigger_delete_if_done(TxState, State);
                    item_newly_safe ->
                        {TxResult, TmpTxState} = tx_state:add_item_acked(TxState),
                        NewTxState =
{'fraction_non_alphanumeric': 0.13915438198606836, 'fraction_numerical': 0.005831848371942329, 'mean_word_length': 2.9346717654557044, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3285860', 'n_tokens_mistral': 4080, 'n_tokens_neox': 3940, 'n_words': 1005}