<commit_before>% @copyright 2009-2011 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    API for transactions on replicated DHT items.
%% @version $Id$
-module(rdht_tx).
-author('<EMAIL>').
-vsn('$Id$').

%-define(TRACE(X,Y), io:format(X,Y)).
-define(TRACE(X,Y), ok).

-export([req_list/2]).
-export([check_config/0]).

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

-ifdef(with_export_type_support).
-export_type([req_id/0, request/0, result_entry/0, result/0]).
-endif.

-type req_id() :: util:global_uid().
-type request() :: {rdht_tx_read, client_key()}
                 | {rdht_tx_write, client_key(), ?DB:value()}
                 | {commit}.

-type result_entry_read() :: {ok, ?DB:value()} | {fail, timeout | not_found}.
-type result_entry_write() :: {ok} | {fail, timeout}.
-type result_entry_commit() :: {ok} | {fail, abort | timeout}.
-type result_entry() :: result_entry_read()
                      | result_entry_write()
                      | result_entry_commit().
-type result() :: [ result_entry() ].

-spec req_list(tx_tlog:tlog(), [request()]) ->
        {tx_tlog:tlog(), result()}.
%% single request and empty translog, done separately for optimization only
req_list([], [SingleReq]) ->
    RdhtOpWithReqId = initiate_rdht_ops([{1, SingleReq}]),
    {NewTLog, TmpResultList, [], [], []} =
        collect_results_and_do_translogops({[], [], RdhtOpWithReqId, [], []}),
    {_ReqNum, Result} = lists:unzip(TmpResultList),
    {NewTLog, Result};

req_list([], [{rdht_tx_write, _K, _V} = SingleReq, {commit}]) ->
    {TLog, [Res1]} = req_list(tx_tlog:empty(), [SingleReq]),
    {TLog, [Res1, commit(TLog)]};

req_list(TLog, PlainReqList) ->
    ?TRACE("rdht_tx:req_list(~p, ~p)~n", [TLog, PlainReqList]),
    %% PRE: no 'abort/tx_failed' in TLog
    %% rdht requests for independent keys are processed in parallel.
    %% requests for the same key are executed in order with
    %% accumulated translog.
    %% if translog entry exists, do work_phase inside this process to
    %% reduce overall latency.
    NumReqs = length(PlainReqList),
    ReqList = lists:zip(lists:seq(1, NumReqs), PlainReqList),
    %% split into 'rdht', 'delayed' and 'translog based' operations
    {RdhtOps, Delayed, TransLogOps, Commit} = my_split_ops(TLog, ReqList),

    RdhtOpsWithReqIds = initiate_rdht_ops(RdhtOps),
    {NewTLog, TmpResultList, [], [], []} =
        collect_results_and_do_translogops({TLog, [], RdhtOpsWithReqIds,
                                            Delayed, TransLogOps}),
    %% Now all op lists are empty.
    %% @TODO only if TransLog is ok, do the validation here if requested
    CommitResults =
        case Commit of
            []               -> [];
            [{NumReqs,{commit}}] -> [{NumReqs, commit(NewTLog)}];
            [{Pos,{commit}}] ->
                log:log(warn, "Commit not at end of a request list. "
                        "Deciding abort."),
                [{Pos, {fail, abort}}];
            Commits          ->
                log:log(warn, "Multiple commits in a request list. "
                        "Deciding abort."),
                [ {Num, {fail, abort}} || {Num, {commit}} <- Commits ]
        end,
    %% Sort resultlist and eliminate numbering
    {_, ResultList} = lists:unzip(
                        lists:sort(fun(A, B) ->
                                           element(1, A) =< element(1, B)
                                   end, TmpResultList ++ CommitResults)),
    %% return the NewTLog and a result list
    {NewTLog, ResultList}.

%% implementation
-spec my_split_ops(tx_tlog:tlog(), [{pos_integer(), request()}])
                  -> {[{pos_integer(), request()}],
                      [{pos_integer(), request()}],
                      [{pos_integer(), request()}],
                      [{pos_integer(), request()}]}.
my_split_ops(TLog, ReqList) ->
    ?TRACE("rdht_tx:my_split_ops(~p, ~p)~n", [TLog, ReqList]),
    Splitter =
        fun(ReqEntry, {RdhtOps, Delayed, TransLogOps, Commit}) ->
          {_Num, Entry} = ReqEntry,
          case element(1, Entry) of
              commit ->
                  {RdhtOps, Delayed, TransLogOps, [ReqEntry | Commit]};
              _ -> case {lists:keymember(element(2, Entry), 2, TLog),
                         my_key_in_numbered_reqlist(element(2, Entry), RdhtOps)}
                   of
                       {true,_} ->
                           {RdhtOps, Delayed, [ ReqEntry | TransLogOps ],
                            Commit};
                       {false, true} -><commit_msg>- if a commit was done, cleanup the TLog to save bw, its of no value anymore.<commit_after>% @copyright 2009-2011 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    API for transactions on replicated DHT items.
%% @version $Id$
-module(rdht_tx).
-author('<EMAIL>').
-vsn('$Id$').

%-define(TRACE(X,Y), io:format(X,Y)).
-define(TRACE(X,Y), ok).

-export([req_list/2]).
-export([check_config/0]).

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

-ifdef(with_export_type_support).
-export_type([req_id/0, request/0, result_entry/0, result/0]).
-endif.

-type req_id() :: util:global_uid().
-type request() :: {rdht_tx_read, client_key()}
                 | {rdht_tx_write, client_key(), ?DB:value()}
                 | {commit}.

-type result_entry_read() :: {ok, ?DB:value()} | {fail, timeout | not_found}.
-type result_entry_write() :: {ok} | {fail, timeout}.
-type result_entry_commit() :: {ok} | {fail, abort | timeout}.
-type result_entry() :: result_entry_read()
                      | result_entry_write()
                      | result_entry_commit().
-type result() :: [ result_entry() ].

-spec req_list(tx_tlog:tlog(), [request()]) ->
        {tx_tlog:tlog(), result()}.
%% single request and empty translog, done separately for optimization only
req_list([], [SingleReq]) ->
    RdhtOpWithReqId = initiate_rdht_ops([{1, SingleReq}]),
    {NewTLog, TmpResultList, [], [], []} =
        collect_results_and_do_translogops({[], [], RdhtOpWithReqId, [], []}),
    {_ReqNum, Result} = lists:unzip(TmpResultList),
    {NewTLog, Result};

req_list([], [{rdht_tx_write, _K, _V} = SingleReq, {commit}]) ->
    {TLog, [Res1]} = req_list(tx_tlog:empty(), [SingleReq]),
    {TLog, [Res1, commit(TLog)]};

req_list(TLog, PlainReqList) ->
    ?TRACE("rdht_tx:req_list(~p, ~p)~n", [TLog, PlainReqList]),
    %% PRE: no 'abort/tx_failed' in TLog
    %% rdht requests for independent keys are processed in parallel.
    %% requests for the same key are executed in order with
    %% accumulated translog.
    %% if translog entry exists, do work_phase inside this process to
    %% reduce overall latency.
    NumReqs = length(PlainReqList),
    ReqList = lists:zip(lists:seq(1, NumReqs), PlainReqList),
    %% split into 'rdht', 'delayed' and 'translog based' operations
    {RdhtOps, Delayed, TransLogOps, Commit} = my_split_ops(TLog, ReqList),

    RdhtOpsWithReqIds = initiate_rdht_ops(RdhtOps),
    {NewTLog, TmpResultList, [], [], []} =
        collect_results_and_do_translogops({TLog, [], RdhtOpsWithReqIds,
                                            Delayed, TransLogOps}),
    %% Now all op lists are empty.
    %% @TODO only if TransLog is ok, do the validation here if requested
    CommitResults =
        case Commit of
            []               -> [];
            [{NumReqs,{commit}}] -> [{NumReqs, commit(NewTLog)}];
            [{Pos,{commit}}] ->
                log:log(warn, "Commit not at end of a request list. "
                        "Deciding abort."),
                [{Pos, {fail, abort}}];
            Commits          ->
                log:log(warn, "Multiple commits in a request list. "
                        "Deciding abort."),
                [ {Num, {fail, abort}} || {Num, {commit}} <- Commits ]
        end,
    %% Sort resultlist and eliminate numbering
    {_, ResultList} = lists:unzip(
                        lists:sort(fun(A, B) ->
                                           element(1, A) =< element(1, B)
                                   end, TmpResultList ++ CommitResults)),
    %% return the NewTLog and a result list
    %% if a commit was done, cleanup the TLog, its of no value anymore.
    case Commit of
        [] -> {NewTLog, ResultList};
        _  -> {tx_tlog:empty(), ResultList}
    end.

%% implementation
-spec my_split_ops(tx_tlog:tlog(), [{pos_integer(), request()}])
                  -> {[{pos_integer(), request()}],
                      [{pos_integer(), request()}],
                      [{pos_integer(), request()}],
                      [{pos_integer(), request()}]}.
my_split_ops(TLog, ReqList) ->
    ?TRACE("rdht_tx:my_split_ops(~p, ~p)~n", [TLog, ReqList]),
    Splitter =
        fun(ReqEntry, {RdhtOps, Delayed, TransLogOps, Commit}) ->
          {_Num, Entry} = ReqEntry,
          case element(1, Entry) of
              commit ->
                  {RdhtOps, Delayed, TransLogOps, [ReqEntry | Commit]};
              _ -> case {lists:keymember(element(2, Entry), 2, TLog),
                         my_key_in_numbered_reqlist(element(2, Entry), RdhtOps)}
                   of
                       {true,_} ->
                           {RdhtOps, Delayed, [ ReqEntry | TransLogOps ],
                            Commit};
                       {false, true} ->
{'fraction_non_alphanumeric': 0.14625332826169646, 'fraction_numerical': 0.006085964244960061, 'mean_word_length': 2.4024587512131994, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3194220', 'n_tokens_mistral': 3341, 'n_tokens_neox': 3222, 'n_words': 849}