<commit_before>-module(clocksi_downstream_generator_vnode).

-behaviour(riak_core_vnode).

-include("floppy.hrl").
-include_lib("riak_core/include/riak_core_vnode.hrl").

-record(dstate, {partition,
                 last_commit_time,
                 pending_operations,
                 stable_time}).

-export([start_vnode/1,
         trigger/2]).

-export([init/1,
         terminate/2,
         handle_command/3,
         is_empty/1,
         delete/1,
         handle_handoff_command/3,
         handoff_starting/2,
         handoff_cancelled/1,
         handoff_finished/2,
         handle_handoff_data/2,
         encode_handoff_item/2,
         handle_coverage/4,
         handle_exit/3]).

%% @doc Notify downstream_generator_vnode that an update has been logged.
%%      Downstream_generator_vnode then reads the updates from log, generate
%%      downstream op and write to persistent log
%%      input: Key to identify the partition,
%%             Writeset -> set of updates
-spec trigger(key(),
              {TxId :: term(),
               Updates :: [{term(),
                            {Key :: term(), Type:: term(), Op :: term()}}],
               Vec_snapshot_time :: vectorclock:vectorclock(),
               Commit_time :: non_neg_integer()})
             -> ok | {error, timeout}.
trigger(Key, Writeset) ->
    Logid = log_utilities:get_logid_from_key(Key),
    Preflist = log_utilities:get_preflist_from_logid(Logid),
    IndexNode = hd(Preflist),
    riak_core_vnode_master:command([IndexNode],
                                   {trigger, Writeset, self()},
                                   ?CLOCKSI_GENERATOR_MASTER),
    receive
        {ok, trigger_received} ->
            ok
    after 100 ->
            {error, timeout}
    end.

start_vnode(I) ->
    riak_core_vnode_master:get_vnode_pid(I, ?MODULE).

init([Partition]) ->
    {ok, #dstate{partition = Partition,
                 last_commit_time = 0,
                 pending_operations = []}}.

%% @doc Read client update operations,
%%      generate downstream operations and store it to persistent log.
handle_command({trigger, WriteSet, From}, _Sender,
               State=#dstate{partition = Partition,
                             last_commit_time = LastCommitTime,
                             pending_operations = Pending}) ->
    PendingOperations = add_to_pending_operations(Pending, WriteSet),
    lager:info("PendingOperations: ~p", [PendingOperations]),

    From ! {ok, trigger_received},
    Node = {Partition, node()}, %% Send ack to caller and continue processing
    Stable_time = get_stable_time(Node, LastCommitTime),
    %% Send a message to itself to process operations
    riak_core_vnode_master:command([Node],
                                   {process},
                                   ?CLOCKSI_GENERATOR_MASTER),
    {reply, {ok, trigger_received}, State#dstate{
                                      pending_operations=PendingOperations,
                                      stable_time=Stable_time}};

handle_command({process}, _Sender,
               State=#dstate{partition = Partition,
                             last_commit_time = LastCommitTime,
                             pending_operations = PendingOperations,
                             stable_time = Stable_time}) ->
    lager:info("Take updates before time : ~p",[Stable_time]),
    Sorted_ops = filter_operations(PendingOperations,
                                   Stable_time, LastCommitTime),
    lager:info("Generate downstream for ~p",[Sorted_ops]),
    {Remaining_operations, Last_processed_time} =
        lists:foldl( fun(X, {Ops, LCTS}) ->
                             case process_update(X) of
                                 {ok, CommitTime} ->
                                     %% Remove this op from pending_operations
                                     New_pending = lists:delete(X, Ops),
                                     {New_pending, CommitTime};
                                 {error, _Reason} ->
                                     {Ops, LCTS}
                             end
                     end, {PendingOperations, LastCommitTime}, Sorted_ops),
    DcId = dc_utilities:get_my_dc_id(),
    lager:info("Updating vector clock ~p",[Stable_time]),
    {ok, _Clock} = vectorclock:update_clock(Partition, DcId, Stable_time),
    _ = case PendingOperations of
        [] ->
                %%TODO
            %% inter_dc_repl_vnode:sync_clock(Partition, Clock);
                donothing;
        [H|_T] -> Key = H#clocksi_payload.key,
                  inter_dc_repl_vnode:trigger(Key)
    end,
    {reply, ok, State#dstate{last_commit_time = Last_processed_time,
                             pending_operations = Remaining_operations}};

handle_command(_Message, _Sender, State) ->
    {noreply, State}.

handle_handoff_command( _Message , _Sender, State) ->
    {noreply, State}.

handoff_starting(_TargetNode, State) ->
    {true, State}.

handoff_cancelled(State) ->
    {ok, State}.

handoff_finished(_TargetNode, State) ->
    {ok, State}.

handle_handoff_data(_Data, State) ->
    {reply, ok, State}.<commit_msg>trigger downstream automatically<commit_after>-module(clocksi_downstream_generator_vnode).

-behaviour(riak_core_vnode).

-include("floppy.hrl").
-include_lib("riak_core/include/riak_core_vnode.hrl").

-record(dstate, {partition,
                 last_commit_time,
                 pending_operations,
                 stable_time}).

-export([start_vnode/1,
         trigger/2]).

-export([init/1,
         terminate/2,
         handle_command/3,
         is_empty/1,
         delete/1,
         handle_handoff_command/3,
         handoff_starting/2,
         handoff_cancelled/1,
         handoff_finished/2,
         handle_handoff_data/2,
         encode_handoff_item/2,
         handle_coverage/4,
         handle_exit/3]).

%% @doc Notify downstream_generator_vnode that an update has been logged.
%%      Downstream_generator_vnode then reads the updates from log, generate
%%      downstream op and write to persistent log
%%      input: Key to identify the partition,
%%             Writeset -> set of updates
-spec trigger(key(),
              {TxId :: term(),
               Updates :: [{term(),
                            {Key :: term(), Type:: term(), Op :: term()}}],
               Vec_snapshot_time :: vectorclock:vectorclock(),
               Commit_time :: non_neg_integer()})
             -> ok | {error, timeout}.
trigger(Key, Writeset) ->
    Logid = log_utilities:get_logid_from_key(Key),
    Preflist = log_utilities:get_preflist_from_logid(Logid),
    IndexNode = hd(Preflist),
    riak_core_vnode_master:command([IndexNode],
                                   {trigger, Writeset, self()},
                                   ?CLOCKSI_GENERATOR_MASTER),
    receive
        {ok, trigger_received} ->
            ok
    after 100 ->
            {error, timeout}
    end.

start_vnode(I) ->
    {ok, Pid} = riak_core_vnode_master:get_vnode_pid(I, ?MODULE),
    riak_core_vnode:send_command(Pid, trigger),
    {ok, Pid}.

init([Partition]) ->
    {ok, #dstate{partition = Partition,
                 last_commit_time = 0,
                 pending_operations = []}}.

%%
handle_command(trigger, _Sender,
               State= #dstate{partition = Partition,
                              last_commit_time = LastCommitTime
                              }) ->
    MyNode = node(),
    case MyNode of
        'dev2@127.0.0.1' ->
            {reply, ok, State};
        _ ->
            Node = {Partition, node()},
            Stable_time = get_stable_time(Node, LastCommitTime),
            riak_core_vnode_master:command([Node],
                                           {process},
                                           ?CLOCKSI_GENERATOR_MASTER),
            spawn(fun() ->
                          timer:sleep(1000),
                          riak_core_vnode_master:command([Node],
                                                         {trigger},
                                                         ?CLOCKSI_GENERATOR_MASTER)
                  end
                 ),
            {reply, {ok, trigger_received},
             State#dstate{stable_time=Stable_time}}
        end;

%% @doc Read client update operations,
%%      generate downstream operations and store it to persistent log.
handle_command({trigger, WriteSet, From}, _Sender,
               State=#dstate{partition = Partition,
                             last_commit_time = LastCommitTime,
                             pending_operations = Pending}) ->
    PendingOperations = add_to_pending_operations(Pending, WriteSet),
    lager:info("PendingOperations: ~p", [PendingOperations]),

    From ! {ok, trigger_received},
    Node = {Partition, node()}, %% Send ack to caller and continue processing
    Stable_time = get_stable_time(Node, LastCommitTime),
    %% Send a message to itself to process operations
    riak_core_vnode_master:command([Node],
                                   {process},
                                   ?CLOCKSI_GENERATOR_MASTER),
    {reply, {ok, trigger_received}, State#dstate{
                                      pending_operations=PendingOperations,
                                      stable_time=Stable_time}};

handle_command({process}, _Sender,
               State=#dstate{partition = Partition,
                             last_commit_time = LastCommitTime,
                             pending_operations = PendingOperations,
                             stable_time = Stable_time}) ->
    lager:info("Take updates before time : ~p",[Stable_time]),
    Sorted_ops = filter_operations(PendingOperations,
                                   Stable_time, LastCommitTime),
    lager:info("Generate downstream for ~p",[Sorted_ops]),
    {Remaining_operations, Last_processed_time} =
        lists:foldl( fun(X, {Ops, LCTS}) ->
                             case process_update(X) of
                                 {ok, CommitTime} ->
                                     %% Remove this op from pending_operations
                                     New_pending = lists:delete(X, Ops),
                                     {New_pending, CommitTime};
                                 {error, _Reason} ->
                                     {Ops, LCTS}
                             end
                     end, {PendingOperations, LastCommitTime}, Sorted_ops),
    DcId = dc_utilities:get_my_dc_id(),
    lager:info("Updating vector clock ~p",[Stable_time]),
    {ok, Clock} = vectorclock:update_clock(Partition, DcId, Stable_time),
    _ = case PendingOperations of
            [] ->
                %%TODO
                MyNode = node(),
                case MyNode of
                    'dev2@127.0.0.1' ->
                        {reply, ok, State};
                    _ ->
                        inter_dc_repl_vnode:sync_clock(Partition, Clock)
                end;
            [H|_T] -> Key = H#clocksi_payload.key,
                      inter_dc_repl_vnode:trigger(Key)
        end,
    {reply, ok, State#dstate{last_commit_time = Last_processed_time,
                             pending_operations = Remaining_operations}};

handle_command(_Message, _Sender, State) ->
    {noreply, State}.

handle_handoff_command( _Message , _Sender, State) ->
    {noreply, State}.

handoff_starting(_TargetNode, State) ->
    {true, State}.

handoff_cancelled(State) ->
    {ok, State}.

handoff_finished(_TargetNode, State) ->
    {ok, State}.

handle_handoff_data(_Data, State) ->
    {reply, ok, State}.
{'fraction_non_alphanumeric': 0.10122540559199171, 'fraction_numerical': 0.004832585433206766, 'mean_word_length': 1.7429585798816567, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '965882', 'n_tokens_mistral': 3255, 'n_tokens_neox': 3181, 'n_words': 614}