<commit_before>-module(derflow_vnode).
-behaviour(riak_core_vnode).
-include("derflow.hrl").
-include_lib("riak_core/include/riak_core_vnode.hrl").

-export([async_bind/2,
         async_bind/3,
         bind/2,
         bind/3,
         read/1,
         touch/1,
         next/1,
         is_det/1,
         wait_needed/1,
         declare/1,
         declare/2,
         get_new_id/0,
         put/4,
         execute_and_put/5]).

-export([start_vnode/1,
         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]).

-ignore_xref([start_vnode/1]).

-record(state, {partition, clock, table}).
-record(dv, {value, next = empty, waiting_threads = [], binding_list = [], creator, lazy= false, bounded = false}).

%% Extrenal API
async_bind(Id, Value) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {async_bind,
                                                          Id, Value},
                                              derflow_vnode_master).

async_bind(Id, Function, Args) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {async_bind,
                                                          Id, Function,
                                                          Args}, derflow_vnode_master).

bind(Id, Value) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {bind, Id,
                                                          Value},
                                              derflow_vnode_master).

bind(Id, Function, Args) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {bind, Id,
                                                          Function,
                                                          Args}, derflow_vnode_master).

read(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {read, Id},
                                              derflow_vnode_master).

touch(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {touch, Id},
                                              derflow_vnode_master).

next(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {next, Id},
                                              derflow_vnode_master).

is_det(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {is_det, Id},
                                              derflow_vnode_master).

declare(Id, Partition) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    io:format("I am gonna send it to ~w and my partition is ~w~n",[IndexNode, Partition]),
    riak_core_vnode_master:sync_spawn_command(IndexNode, {declare, Id},
                                              derflow_vnode_master).

declare(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {declare, Id},
                                              derflow_vnode_master).

fetch(Id, FromId, FromP) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:command(IndexNode, {fetch, Id, FromId,
                                               FromP}, derflow_vnode_master).

replyFetch(Id, FromP, DV) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:command(IndexNode, {replyFetch, Id, FromP,
                                               DV}, derflow_vnode_master).

notifyValue(Id, Value) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:command(IndexNode, {notifyValue, Id, Value},
                                   derflow_vnode_master).

get_new_id() ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(now())}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, get_new_id,
                                              derflow_vnode_master).

wait_needed(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {wait_needed,
                                                          Id}, derflow_vnode_master).

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

init([Partition]) ->
    Table=string:concat(integer_to_list(Partition), "dvstore"),
    Table_atom=list_to_atom(Table),
    ets:new(Table_atom, [set, named_table, public, {write_concurrency, true}]),
    {ok, #state { partition=Partition, clock=0, table=Table_atom }}.

handle_command(get_new_id, _From, State=#state{partition=Partition}) ->
    Clock = State#state.clock +1,
    {reply, {Clock,Partition}, State#state{clock=Clock}};

handle_command({declare, Id}, _From, State=#state{table=Table}) ->
    %io:format("Procces ~w declaring ~w~n",[From, Id]),
    V = #dv{value=empty},<commit_msg>Resolve dialyzer error.<commit_after>-module(derflow_vnode).
-behaviour(riak_core_vnode).
-include("derflow.hrl").
-include_lib("riak_core/include/riak_core_vnode.hrl").

-export([async_bind/2,
         async_bind/3,
         bind/2,
         bind/3,
         read/1,
         touch/1,
         next/1,
         is_det/1,
         wait_needed/1,
         declare/1,
         declare/2,
         get_new_id/0,
         put/4,
         execute_and_put/5]).

-export([start_vnode/1,
         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]).

-ignore_xref([start_vnode/1]).

-record(state, {partition, clock, table}).
-record(dv, {value, next = empty, waiting_threads = [], binding_list = [], creator, lazy= false, bounded = false}).

%% Extrenal API
async_bind(Id, Value) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {async_bind,
                                                          Id, Value},
                                              derflow_vnode_master).

async_bind(Id, Function, Args) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {async_bind,
                                                          Id, Function,
                                                          Args}, derflow_vnode_master).

bind(Id, Value) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {bind, Id,
                                                          Value},
                                              derflow_vnode_master).

bind(Id, Function, Args) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {bind, Id,
                                                          Function,
                                                          Args}, derflow_vnode_master).

read(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {read, Id},
                                              derflow_vnode_master).

touch(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {touch, Id},
                                              derflow_vnode_master).

next(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {next, Id},
                                              derflow_vnode_master).

is_det(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {is_det, Id},
                                              derflow_vnode_master).

declare(Id, Partition) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    io:format("I am gonna send it to ~w and my partition is ~w~n",[IndexNode, Partition]),
    riak_core_vnode_master:sync_spawn_command(IndexNode, {declare, Id},
                                              derflow_vnode_master).

declare(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {declare, Id},
                                              derflow_vnode_master).

fetch(Id, FromId, FromP) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:command(IndexNode, {fetch, Id, FromId,
                                               FromP}, derflow_vnode_master).

replyFetch(Id, FromP, DV) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:command(IndexNode, {replyFetch, Id, FromP,
                                               DV}, derflow_vnode_master).

notifyValue(Id, Value) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:command(IndexNode, {notifyValue, Id, Value},
                                   derflow_vnode_master).

get_new_id() ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(now())}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, get_new_id,
                                              derflow_vnode_master).

wait_needed(Id) ->
    DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Id)}),
    PrefList = riak_core_apl:get_primary_apl(DocIdx, 1, derflow),
    [{IndexNode, _Type}] = PrefList,
    riak_core_vnode_master:sync_spawn_command(IndexNode, {wait_needed,
                                                          Id}, derflow_vnode_master).

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

init([Partition]) ->
    Table = string:concat(integer_to_list(Partition), "dvstore"),
    TableAtom = list_to_atom(Table),
    TableAtom = ets:new(TableAtom, [set, named_table, public, {write_concurrency, true}]),
    {ok, #state { partition=Partition, clock=0, table=TableAtom }}.

handle_command(get_new_id, _From, State=#state{partition=Partition}) ->
    Clock = State#state.clock +1,
    {reply, {Clock,Partition}, State#state{clock=Clock}};

handle_command({declare, Id}, _From, State=#state{table=Table}) ->
    %io:format("Procces ~w declaring ~w~n",[From, Id]),
    V = #dv{value=empty},
{'fraction_non_alphanumeric': 0.1201933179239336, 'fraction_numerical': 0.00644393079778665, 'mean_word_length': 2.564153769345981, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 74, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3586632', 'n_tokens_mistral': 4951, 'n_tokens_neox': 4864, 'n_words': 704}