<commit_before>%% @doc The coordinator for stat write opeartions.  This example will
%% show how to properly replicate your data in Riak Core by making use
%% of the _preflist_.
-module(clockSI_readitem_fsm).
-behavior(gen_fsm).
-include("floppy.hrl").

%% API
-export([start_link/6]).

%% Callbacks
-export([init/1, code_change/4, handle_event/3, handle_info/3,
         handle_sync_event/4, terminate/3]).

%% States
-export([check_clock/2,
         get_txs_to_check/2,
         commit_notification/2, 
         return/2]).

%% Spawn

-record(state, {type,
                key,
                transaction,
                tx_coordinator,
                vnode,
                updates,
                pending_txs}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Vnode, Coordinator, Tx, Key, Type, Updates) ->
    gen_fsm:start_link(?MODULE, [Vnode, Coordinator, Tx, Key, Type, Updates], []).

now_milisec({MegaSecs,Secs,MicroSecs}) ->
    (MegaSecs*1000000 + Secs)*1000000 + MicroSecs.
%%%===================================================================
%%% States
%%%===================================================================

init([Vnode, Coordinator, Transaction, Key, Type, Updates]) ->
    SD = #state{
            vnode=Vnode,
            type=Type,
            key=Key,
            tx_coordinator=Coordinator, 
            transaction=Transaction,
            updates=Updates,
            pending_txs=[]},
    {ok, check_clock, SD, 0}.

%% @doc check_clock: Compares its local clock with the tx timestamp.
%%	if local clock is behinf, it sleeps the fms until the clock
%%	catches up. CLOCK-SI: clock scew.
check_clock(timeout, SD0=#state{transaction=Transaction}) ->
    TxId = Transaction#transaction.txn_id,
    T_TS = TxId#tx_id.snapshot_time,
    Time = now_milisec(erlang:now()),
    case (T_TS) > Time of true ->
            lager:info("ClockSI ReadItemFSM: waiting for clock to catchUp ~n"),
            timer:sleep(T_TS - Time),	
            lager:info("ClockSI ReadItemFSM: done waiting... ~n"),
            {next_state, get_txs_to_check, SD0, 0};

        false ->
            lager:info("ClockSI ReadItemFSM: no need to wait for clock to catchUp ~n"),
            {next_state, get_txs_to_check, SD0, 0}
    end.		
%% @doc get_txs_to_check: 
%%	- Asks the Vnode for pending txs conflicting the cyrrent one
%%	- If none, goes to return state
%%	- Otherwise, goes to commit_notification
get_txs_to_check(timeout, SD0=#state{transaction= Transaction, vnode=Vnode, key=Key}) ->
    lager:info("ClockSI ReadItemFSM: Calling vnode ~w to get pending txs for Key ~w ~n", [Vnode, Key]),
    TxId = Transaction#transaction.txn_id,
    case clockSI_vnode:get_pending_txs(Vnode, {Key, TxId}) of
        {ok, empty} ->		
            lager:info("ClockSI ReadItemFSM: no txs to wait for ~w ~n", [Key]),
            {next_state, return, SD0, 0};
        {ok, Pending} -><commit_msg>Wait until local vectorclock reaches snapshottime before reading<commit_after>%% @doc The coordinator for stat write opeartions.  This example will
%% show how to properly replicate your data in Riak Core by making use
%% of the _preflist_.
-module(clockSI_readitem_fsm).
-behavior(gen_fsm).
-include("floppy.hrl").

%% API
-export([start_link/6]).

%% Callbacks
-export([init/1, code_change/4, handle_event/3, handle_info/3,
         handle_sync_event/4, terminate/3]).

%% States
-export([check_clock/2,
         get_txs_to_check/2,
         commit_notification/2,
         waiting/2,
         return/2]).

%% Spawn

-record(state, {type,
                key,
                transaction,
                tx_coordinator,
                vnode,
                updates,
                pending_txs}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Vnode, Coordinator, Tx, Key, Type, Updates) ->
    gen_fsm:start_link(?MODULE, [Vnode, Coordinator, Tx, Key, Type, Updates], []).

now_milisec({MegaSecs,Secs,MicroSecs}) ->
    (MegaSecs*1000000 + Secs)*1000000 + MicroSecs.
%%%===================================================================
%%% States
%%%===================================================================

init([Vnode, Coordinator, Transaction, Key, Type, Updates]) ->
    SD = #state{
            vnode=Vnode,
            type=Type,
            key=Key,
            tx_coordinator=Coordinator, 
            transaction=Transaction,
            updates=Updates,
            pending_txs=[]},
    {ok, check_clock, SD, 0}.

%% @doc check_clock: Compares its local clock with the tx timestamp.
%%	if local clock is behinf, it sleeps the fms until the clock
%%	catches up. CLOCK-SI: clock scew.
check_clock(timeout, SD0=#state{transaction=Transaction}) ->
    TxId = Transaction#transaction.txn_id,
    T_TS = TxId#tx_id.snapshot_time,
    Time = now_milisec(erlang:now()),
    case (T_TS) > Time of true ->
            lager:info("ClockSI ReadItemFSM: waiting for clock to catchUp ~n"),
            timer:sleep(T_TS - Time),	
            lager:info("ClockSI ReadItemFSM: done waiting... ~n"),
            {next_state, waiting, SD0, 0};

        false ->
            lager:info("ClockSI ReadItemFSM: no need to wait for clock to catchUp ~n"),
            {next_state, waiting, SD0, 0}
    end.

waiting(timeout, SDO=#state{key = Key, transaction = Transaction}) ->
    {ok, LocalClock} = vectorclock:get_clock_by_key(Key),   
    Snapshottime = Transaction#transaction.vec_snapshot_time, 
    case vectorclock:is_greater_than(LocalClock, Snapshottime) of
        false ->
            clockSI_downstream_generator_vnode:trigger(Key, {dummytx, [], dummycommittime}), %%TODO change this, add a heartbeat to increase vectorclock if there are no pending txns in downstream generator
            {next_state, waiting, SDO, 10};
        true ->
            {next_state, return, SDO, 0} 
    end.
		
%% @doc get_txs_to_check: 
%%	- Asks the Vnode for pending txs conflicting the cyrrent one
%%	- If none, goes to return state
%%	- Otherwise, goes to commit_notification
get_txs_to_check(timeout, SD0=#state{transaction= Transaction, vnode=Vnode, key=Key}) ->
    lager:info("ClockSI ReadItemFSM: Calling vnode ~w to get pending txs for Key ~w ~n", [Vnode, Key]),
    TxId = Transaction#transaction.txn_id,
    case clockSI_vnode:get_pending_txs(Vnode, {Key, TxId}) of
        {ok, empty} ->		
            lager:info("ClockSI ReadItemFSM: no txs to wait for ~w ~n", [Key]),
            {next_state, return, SD0, 0};
        {ok, Pending} ->
{'fraction_non_alphanumeric': 0.19461077844311378, 'fraction_numerical': 0.010778443113772455, 'mean_word_length': 3.129171817058096, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '1527197', 'n_tokens_mistral': 2154, 'n_tokens_neox': 2058, 'n_words': 522}