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

-export([declare/0,
         bind/2,
         bind/4,
         read/1,
         produce/2,
         produce/3,
         consume/1,
         extend/1,
         is_det/1,
         wait_needed/1,
         spawn_mon/4,
         get_stream/1]).

%% Public API

declare() ->
    _ = derflow_declare_fsm_sup:start_child([self()]),
    receive
        {ok, Id} ->
            {ok, Id}
    end.

bind(Id, Value) ->
    case derflow_vnode:bind(Id, Value) of
        {ok, _} ->
            ok;
        error ->
            error
    end.

bind(Id, Module, Function, Args) ->
    Value = Module:Function(Args),
    bind(Id, Value).

read(Id) ->
    {ok, Value, _Next}= derflow_vnode:read(Id),
    {ok, Value}.

produce(Id, Value) ->
    derflow_vnode:bind(Id, Value).

produce(Id, Function, Args) ->
    derflow_vnode:bind(Id, Function, Args).

consume(Id) ->
    derflow_vnode:read(Id).

extend(Id) ->
    derflow_vnode:next(Id).

is_det(Id) ->
    derflow_vnode:is_det(Id).

wait_needed(Id) ->
    derflow_vnode:wait_needed(Id).

%% The rest of primitives are not in the paper. They are just utilities.
%% maybe we should just remove them

spawn_mon(Supervisor, Module, Function, Args) ->
    Pid = spawn(Module, Function, Args),
    Supervisor ! {'SUPERVISE', Pid, Module, Function, Args}.

get_stream(Stream)->
    internal_get_stream(Stream, []).

%% Internal functions

internal_get_stream(Head, Output) ->
    case consume(Head) of
        {ok, nil, _} ->
            Output;
        {ok, Value, Next} ->
            internal_get_stream(Next, lists:append(Output, [Value]))
    end.
<commit_msg>Fix function to compute value prior to replication.<commit_after>-module(derflow).
-include("derflow.hrl").
-include_lib("riak_core/include/riak_core_vnode.hrl").

-export([declare/0,
         bind/2,
         bind/4,
         read/1,
         produce/2,
         produce/4,
         consume/1,
         extend/1,
         is_det/1,
         wait_needed/1,
         spawn_mon/4,
         get_stream/1]).

%% Public API

declare() ->
    _ = derflow_declare_fsm_sup:start_child([self()]),
    receive
        {ok, Id} ->
            {ok, Id}
    end.

bind(Id, Value) ->
    case derflow_vnode:bind(Id, Value) of
        {ok, _} ->
            ok;
        error ->
            error
    end.

bind(Id, Module, Function, Args) ->
    Value = Module:Function(Args),
    bind(Id, Value).

read(Id) ->
    {ok, Value, _Next}= derflow_vnode:read(Id),
    {ok, Value}.

produce(Id, Value) ->
    derflow_vnode:bind(Id, Value).

produce(Id, Module, Function, Args) ->
    Value = Module:Function(Args),
    derflow_vnode:bind(Id, Value).

consume(Id) ->
    derflow_vnode:read(Id).

extend(Id) ->
    derflow_vnode:next(Id).

is_det(Id) ->
    derflow_vnode:is_det(Id).

wait_needed(Id) ->
    derflow_vnode:wait_needed(Id).

%% The rest of primitives are not in the paper. They are just utilities.
%% maybe we should just remove them

spawn_mon(Supervisor, Module, Function, Args) ->
    Pid = spawn(Module, Function, Args),
    Supervisor ! {'SUPERVISE', Pid, Module, Function, Args}.

get_stream(Stream)->
    internal_get_stream(Stream, []).

%% Internal functions

internal_get_stream(Head, Output) ->
    case consume(Head) of
        {ok, nil, _} ->
            Output;
        {ok, Value, Next} ->
            internal_get_stream(Next, lists:append(Output, [Value]))
    end.

{'fraction_non_alphanumeric': 0.14655172413793102, 'fraction_numerical': 0.006896551724137931, 'mean_word_length': 2.4060665362035225, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 39, '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': '756069', 'n_tokens_mistral': 1245, 'n_tokens_neox': 1244, 'n_words': 251}