<commit_before>% @copyright 2012-2018 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>
%% @doc    Acceptor implementation from "Generalized Lattice Agreement"
%%  TODO:
%%      - right now proposerlist is lazily extended
%% @end
-module(gla_acceptor).
-author('<EMAIL>').

%-define(TRACE(X,Y), ct:pal(X,Y)).
-define(TRACE(X,Y), ok).
-include("scalaris.hrl").

-define(PDB, db_prbr).

%%% functions for module where embedded into
-export([on/2, init/1, close/1, close_and_delete/1]).
-export([check_config/0]).
-export([new/2]).
-export([set_entry/2]).
-export([get_entry/2]).
-export([entry_key/1]).
-export([entry_val/2]).
-export([entry_set_val/2]).

-export_type([state/0]).
-export_type([entry/0]).

-type state() :: {?PDB:db(), [comm:mypid()]}.

%% so there are no longer any read denies, all reads succeed.
-type entry() :: {
                   any(), %% key
                   any() %% value
                 }.

%% Messages to expect from this module
-spec msg_get_val_reply(comm:mypid(), ?RT:key(), crdt:crdt_module(), crdt:crdt()) -> ok.
msg_get_val_reply(Client, Key, DataType, UpdatedVal) ->
    comm:send(Client, {proposal_val_reply, Key, DataType, UpdatedVal}).

-spec msg_ack_reply(comm:mypid(), ?RT:key(), pr:pr(), crdt:crdt()) -> ok.
msg_ack_reply(Client, Key, ProposalNumber, ProposalValue) ->
    comm:send(Client, {ack_reply, Key, ProposalNumber, ProposalValue}).

-spec msg_learner_ack_reply(comm:mypid(), ?RT:key(), [comm:mypid()], crdt:crdt_module(), pr:pr(), crdt:crdt(), any()) -> ok.
msg_learner_ack_reply(Proposer, Key, Clients, DataType, ProposalNumber, ProposalValue, ProposerId) ->
    comm:send(Proposer, {learner_ack_reply, Key, Clients, DataType, ProposalNumber, ProposalValue, ProposerId}).

-spec msg_nack_reply(comm:mypid(), ?RT:key(), pr:pr(), crdt:crdt()) -> ok.
msg_nack_reply(Client, Key, ProposalNumber, ProposalValue) ->
    comm:send(Client, {nack_reply, Key, ProposalNumber, ProposalValue}).


%% initialize: return initial state.
-spec init(atom() | tuple()) -> state().
init(DBName) -> {?PDB:new(DBName), []}.

%% @doc Closes the given DB (it may be recoverable using open/1 depending on
%%      the DB back-end).
-spec close(state()) -> true.
close(_State={TableName, _}) -> ?PDB:close(TableName).

%% @doc Closes the given DB and deletes all contents (this DB can thus not be
%%      re-opened using open/1).
-spec close_and_delete(state()) -> true.
close_and_delete(_State={TableName, _}) -> ?PDB:close_and_delete(TableName).

-spec on(tuple(), state()) -> state().
on({gla_acceptor, get_proposal_value, _Cons, Proposer, Key, DataType, UpdateFun}, State={TableName, _ProposerList}) ->
    ?TRACE("gla_acceptor:update: ~p ~p ", [Key, Proposer]),
    Entry = get_entry(Key, TableName),

    Keys = lists:sort(replication:get_keys(Key)),
    Tmp = lists:dropwhile(fun(E) -> E =/= Key end, Keys),
    ThisReplicaId = length(Keys) - length(Tmp) + 1,

    CVal = entry_val(Entry, DataType),
    NewCVal = DataType:apply_update(UpdateFun, ThisReplicaId, CVal),
    NewEntry = entry_set_val(Entry, NewCVal),
    _ = set_entry(NewEntry, TableName),

    msg_get_val_reply(Proposer, Key, DataType, NewCVal),
    trace_mpath:log_info(self(), {acceptor_update,
                                  key, Key,
                                  old_value, CVal,
                                  new_value, NewCVal}),
    State;


on({gla_acceptor, propose, _Cons, Proposer, Key, Clients, ProposalNumber, ProposedValue, DataType},
   _State={TableName, ProposerList}) ->
    ?TRACE("crdt_acceptor:propose: ~p ~p ~n ~p ~p", [Key, Proposer, ProposalNumber, ProposedValue]),
    NewProposerList = case lists:member(Proposer, ProposerList) of
                          true -> ProposerList;
                          false -> [Proposer | ProposerList]
                      end,

    Entry = get_entry(Key, TableName),
    AcceptedValue = entry_val(Entry, DataType),

    case DataType:lteq(AcceptedValue, ProposedValue) of
        true ->
            NewEntry = entry_set_val(Entry, ProposedValue),
            _ = set_entry(NewEntry, TableName),

            _ = [msg_learner_ack_reply(Learner, Key, Clients, DataType, ProposalNumber, ProposedValue, Proposer)
                 || Learner <- NewProposerList], %% each proposer is also a learner
            msg_ack_reply(Proposer, Key, ProposalNumber, ProposedValue);
        false ->
            NewEntry = entry_set_val(Entry, DataType:merge(AcceptedValue, ProposedValue)),
            _ = set_entry(NewEntry, TableName),
            msg_nack_reply(Proposer, Key, ProposalNumber, AcceptedValue)
    end,

    {TableName, NewProposerList}.

-spec get_entry(any(), ?PDB:db()) -> entry().
get_entry(Id, TableName) ->
    case ?PDB:get(TableName, Id) of
        {}    -> new(Id);
        Entry -> Entry
    end.

-spec set_entry(entry(), ?PDB:db()) -> ?PDB:db().
set_entry(NewEntry, TableName) ->
    _ = ?PDB:set(TableName, NewEntry),
    TableName.


-spec new(any()) -> entry().
new(Key) ->
    new(Key, gla_bottom).

-spec new(any(), any()) -> entry().
new(Key, Val) -><commit_msg>crdt: gla: send correct value in nack<commit_after>% @copyright 2012-2018 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>
%% @doc    Acceptor implementation from "Generalized Lattice Agreement"
%%  TODO:
%%      - right now proposerlist is lazily extended
%% @end
-module(gla_acceptor).
-author('<EMAIL>').

%-define(TRACE(X,Y), ct:pal(X,Y)).
-define(TRACE(X,Y), ok).
-include("scalaris.hrl").

-define(PDB, db_prbr).

%%% functions for module where embedded into
-export([on/2, init/1, close/1, close_and_delete/1]).
-export([check_config/0]).
-export([new/2]).
-export([set_entry/2]).
-export([get_entry/2]).
-export([entry_key/1]).
-export([entry_val/2]).
-export([entry_set_val/2]).

-export_type([state/0]).
-export_type([entry/0]).

-type state() :: {?PDB:db(), [comm:mypid()]}.

%% so there are no longer any read denies, all reads succeed.
-type entry() :: {
                   any(), %% key
                   any() %% value
                 }.

%% Messages to expect from this module
-spec msg_get_val_reply(comm:mypid(), ?RT:key(), crdt:crdt_module(), crdt:crdt()) -> ok.
msg_get_val_reply(Client, Key, DataType, UpdatedVal) ->
    comm:send(Client, {proposal_val_reply, Key, DataType, UpdatedVal}).

-spec msg_ack_reply(comm:mypid(), ?RT:key(), pr:pr(), crdt:crdt()) -> ok.
msg_ack_reply(Client, Key, ProposalNumber, ProposalValue) ->
    comm:send(Client, {ack_reply, Key, ProposalNumber, ProposalValue}).

-spec msg_learner_ack_reply(comm:mypid(), ?RT:key(), [comm:mypid()], crdt:crdt_module(), pr:pr(), crdt:crdt(), any()) -> ok.
msg_learner_ack_reply(Proposer, Key, Clients, DataType, ProposalNumber, ProposalValue, ProposerId) ->
    comm:send(Proposer, {learner_ack_reply, Key, Clients, DataType, ProposalNumber, ProposalValue, ProposerId}).

-spec msg_nack_reply(comm:mypid(), ?RT:key(), pr:pr(), crdt:crdt()) -> ok.
msg_nack_reply(Client, Key, ProposalNumber, ProposalValue) ->
    comm:send(Client, {nack_reply, Key, ProposalNumber, ProposalValue}).


%% initialize: return initial state.
-spec init(atom() | tuple()) -> state().
init(DBName) -> {?PDB:new(DBName), []}.

%% @doc Closes the given DB (it may be recoverable using open/1 depending on
%%      the DB back-end).
-spec close(state()) -> true.
close(_State={TableName, _}) -> ?PDB:close(TableName).

%% @doc Closes the given DB and deletes all contents (this DB can thus not be
%%      re-opened using open/1).
-spec close_and_delete(state()) -> true.
close_and_delete(_State={TableName, _}) -> ?PDB:close_and_delete(TableName).

-spec on(tuple(), state()) -> state().
on({gla_acceptor, get_proposal_value, _Cons, Proposer, Key, DataType, UpdateFun}, State={TableName, _ProposerList}) ->
    ?TRACE("gla_acceptor:update: ~p ~p ", [Key, Proposer]),
    Entry = get_entry(Key, TableName),

    Keys = lists:sort(replication:get_keys(Key)),
    Tmp = lists:dropwhile(fun(E) -> E =/= Key end, Keys),
    ThisReplicaId = length(Keys) - length(Tmp) + 1,

    CVal = entry_val(Entry, DataType),
    NewCVal = DataType:apply_update(UpdateFun, ThisReplicaId, CVal),
    NewEntry = entry_set_val(Entry, NewCVal),
    _ = set_entry(NewEntry, TableName),

    msg_get_val_reply(Proposer, Key, DataType, NewCVal),
    trace_mpath:log_info(self(), {acceptor_update,
                                  key, Key,
                                  old_value, CVal,
                                  new_value, NewCVal}),
    State;


on({gla_acceptor, propose, _Cons, Proposer, Key, Clients, ProposalNumber, ProposedValue, DataType},
   _State={TableName, ProposerList}) ->
    ?TRACE("crdt_acceptor:propose: ~p ~p ~n ~p ~p", [Key, Proposer, ProposalNumber, ProposedValue]),
    NewProposerList = case lists:member(Proposer, ProposerList) of
                          true -> ProposerList;
                          false -> [Proposer | ProposerList]
                      end,

    Entry = get_entry(Key, TableName),
    AcceptedValue = entry_val(Entry, DataType),

    case DataType:lteq(AcceptedValue, ProposedValue) of
        true ->
            NewEntry = entry_set_val(Entry, ProposedValue),
            _ = set_entry(NewEntry, TableName),

            _ = [msg_learner_ack_reply(Learner, Key, Clients, DataType, ProposalNumber, ProposedValue, Proposer)
                 || Learner <- NewProposerList], %% each proposer is also a learner
            msg_ack_reply(Proposer, Key, ProposalNumber, ProposedValue);
        false ->
            MergedValue = DataType:merge(AcceptedValue, ProposedValue),
            NewEntry = entry_set_val(Entry, MergedValue),
            _ = set_entry(NewEntry, TableName),
            msg_nack_reply(Proposer, Key, ProposalNumber, MergedValue)
    end,

    {TableName, NewProposerList}.

-spec get_entry(any(), ?PDB:db()) -> entry().
get_entry(Id, TableName) ->
    case ?PDB:get(TableName, Id) of
        {}    -> new(Id);
        Entry -> Entry
    end.

-spec set_entry(entry(), ?PDB:db()) -> ?PDB:db().
set_entry(NewEntry, TableName) ->
    _ = ?PDB:set(TableName, NewEntry),
    TableName.


-spec new(any()) -> entry().
new(Key) ->
    new(Key, gla_bottom).

-spec new(any(), any()) -> entry().
new(Key, Val) ->
{'fraction_non_alphanumeric': 0.14997786631252766, 'fraction_numerical': 0.004957945993802567, 'mean_word_length': 3.4932378679395386, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3363766', 'n_tokens_mistral': 3788, 'n_tokens_neox': 3737, 'n_words': 914}