<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2019 <NAME>.  All Rights Reserved.
%%
%% This file is provided to you 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.
%%
%% -------------------------------------------------------------------

-module(alsberg_day).
-author("<NAME> <<EMAIL>>").

-behaviour(gen_server).

%% API
-export([start_link/0,
         stop/0,
         timeout/0,
         write/2,
         read/1,
         read_local/1,
         state/0,
         update/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-record(state, {next_id, store, membership=[], outstanding}).

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

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

stop() ->
    gen_server:stop(?MODULE).

timeout() ->
    4000.

%% @doc Notifies us of membership update.
update(LocalState0) ->
    LocalState = partisan_peer_service:decode(LocalState0),
    gen_server:cast(?MODULE, {update, LocalState}).

%% @doc Issue write operations.
state() ->
    gen_server:call(?MODULE, state).

%% @doc Issue write operations.
write(Key, Value) ->
    gen_server:cast(?MODULE, {write, self(), Key, Value}),

    receive
        Response ->
            Response
    end.

%% @doc Issue read operations.
read(Key) ->
    gen_server:cast(?MODULE, {read, self(), Key}),

    receive
        Response ->
            Response
    end.

%% @doc Issue read operations.
read_local(Key) ->
    %% TODO: Bit of a hack just to get this working.
    true = erlang:register(read_coordinator, self()),
    From = partisan_util:registered_name(read_coordinator),

    gen_server:cast(?MODULE, {read_local, From, Key}),

    receive
        Response ->
            Response
    end.

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @private
init([]) ->
    %% Seed the random number generator.
    partisan_config:seed(),

    %% Register membership update callback.
    partisan_peer_service:add_sup_callback(fun ?MODULE:update/1),

    %% Start with initial membership.
    {ok, Membership0} = partisan_peer_service:members(),
    partisan_logger:info("Starting with membership: ~p", [Membership0]),

    %% Initialization values.
    Store = dict:new(),
    Membership = membership(Membership0),
    Outstanding = dict:new(),

    {ok, #state{next_id=0,
                store=Store,
                membership=Membership,
                outstanding=Outstanding}}.

%% @private
handle_call(state, _From, #state{store=Store}=State) ->
    {reply, {ok, Store}, State};

%% @private
handle_call(_Msg, _From, State) ->
    {reply, ok, State}.

%% @private
handle_cast({update, Membership0}, State) ->
    Membership = membership(Membership0),
    {noreply, State#state{membership=Membership}};

%% @private
handle_cast({read_local, From, Key}, #state{store=Store}=State) ->
    %% Get the value.
    Value = read(Key, Store),

    %% Reply to the caller.
    partisan_pluggable_peer_service_manager:forward_message(From, {ok, Value}),

    {noreply, State};
handle_cast({read, From0, Key}, #state{next_id=NextId, membership=[Primary|_Rest], store=Store}=State) ->
    %% TODO: Bit of a hack just to get this working.
    From = list_to_atom("read_coordinator_" ++ integer_to_list(NextId)),
    try
        true = erlang:unregister(From)
    catch
        _:_ ->
            ok
    end,
    true = erlang:register(From, From0),
    EncodedFrom = partisan_util:registered_name(From),

    %% TODO: HACK to get around problem in interprocedural analysis.
    Request = {read, EncodedFrom, Key}, 

    case node() of 
        Primary ->
            %% Get the value.
            Value = read(Key, Store),

            %% Reply to the caller.
            partisan_pluggable_peer_service_manager:forward_message(From, {ok, Value}),

            {noreply, State#state{next_id=NextId+1}};
        _ ->
            %% Reply to caller.
            partisan_logger:info("Node ~p is not the primary for request: ~p", [node(), Request]),
            partisan_pluggable_peer_service_manager:forward_message(From, {error, not_primary}),

            {noreply, State#state{next_id=NextId+1}}
    end;
handle_cast({write, From0, Key, Value}, #state{next_id=NextId, membership=[Primary|Rest]=Membership, outstanding=Outstanding0, store=Store0}=State) ->
    %% TODO: Bit of a hack just to get this working.
    From = list_to_atom("write_coordinator_" ++ integer_to_list(NextId)),
    try
        true = erlang:unregister(From)
    catch
        _:_ ->
            ok
    end,
    true = erlang:register(From, From0),
    EncodedFrom = partisan_util:registered_name(From),

    %% TODO: HACK to get around problem in interprocedural analysis.
    Request = {write, EncodedFrom, Key, Value},

    case node() of 
        Primary ->

            %% Add to list of outstanding requests.
            Replies = [node()],
            Outstanding = dict:store(Request, {Membership, Replies}, Outstanding0),

            %% Send collaboration message to other nodes.
            CoordinatorNode = node(),

            lists:foreach(fun(Node) ->
                partisan_logger:info("sending collaborate message to node ~p for request ~p", [Node, Request]),<commit_msg>Fix reply usage in Alsberg-Day.<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2019 <NAME>.  All Rights Reserved.
%%
%% This file is provided to you 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.
%%
%% -------------------------------------------------------------------

-module(alsberg_day).
-author("<NAME> <<EMAIL>>").

-behaviour(gen_server).

%% API
-export([start_link/0,
         stop/0,
         timeout/0,
         write/2,
         read/1,
         read_local/1,
         state/0,
         update/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-record(state, {next_id, store, membership=[], outstanding}).

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

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

stop() ->
    gen_server:stop(?MODULE).

timeout() ->
    4000.

%% @doc Notifies us of membership update.
update(LocalState0) ->
    LocalState = partisan_peer_service:decode(LocalState0),
    gen_server:cast(?MODULE, {update, LocalState}).

%% @doc Issue write operations.
state() ->
    gen_server:call(?MODULE, state).

%% @doc Issue write operations.
write(Key, Value) ->
    gen_server:cast(?MODULE, {write, self(), Key, Value}),

    receive
        Response ->
            Response
    end.

%% @doc Issue read operations.
read(Key) ->
    gen_server:cast(?MODULE, {read, self(), Key}),

    receive
        Response ->
            Response
    end.

%% @doc Issue read operations.
read_local(Key) ->
    %% TODO: Bit of a hack just to get this working.
    true = erlang:register(read_coordinator, self()),
    From = partisan_util:registered_name(read_coordinator),

    gen_server:cast(?MODULE, {read_local, From, Key}),

    receive
        Response ->
            Response
    end.

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @private
init([]) ->
    %% Seed the random number generator.
    partisan_config:seed(),

    %% Register membership update callback.
    partisan_peer_service:add_sup_callback(fun ?MODULE:update/1),

    %% Start with initial membership.
    {ok, Membership0} = partisan_peer_service:members(),
    partisan_logger:info("Starting with membership: ~p", [Membership0]),

    %% Initialization values.
    Store = dict:new(),
    Membership = membership(Membership0),
    Outstanding = dict:new(),

    {ok, #state{next_id=0,
                store=Store,
                membership=Membership,
                outstanding=Outstanding}}.

%% @private
handle_call(state, _From, #state{store=Store}=State) ->
    {reply, {ok, Store}, State};

%% @private
handle_call(_Msg, _From, State) ->
    {reply, ok, State}.

%% @private
handle_cast({update, Membership0}, State) ->
    Membership = membership(Membership0),
    {noreply, State#state{membership=Membership}};

%% @private
handle_cast({read_local, From, Key}, #state{store=Store}=State) ->
    %% Get the value.
    Value = read(Key, Store),

    %% Reply to the caller.
    partisan_pluggable_peer_service_manager:forward_message(From, {ok, Value}),

    {noreply, State};
handle_cast({read, From0, Key}, #state{next_id=NextId, membership=[Primary|_Rest], store=Store}=State) ->
    %% TODO: Bit of a hack just to get this working.
    From = list_to_atom("read_coordinator_" ++ integer_to_list(NextId)),
    try
        true = erlang:unregister(From)
    catch
        _:_ ->
            ok
    end,
    true = erlang:register(From, From0),
    EncodedFrom = partisan_util:registered_name(From),

    %% TODO: HACK to get around problem in interprocedural analysis.
    Request = {read, EncodedFrom, Key}, 

    case node() of 
        Primary ->
            %% Get the value.
            Value = read(Key, Store),

            %% Reply to the caller.
            partisan_pluggable_peer_service_manager:forward_message(EncodedFrom, {ok, Value}),

            {noreply, State#state{next_id=NextId+1}};
        _ ->
            %% Reply to caller.
            partisan_logger:info("Node ~p is not the primary for request: ~p", [node(), Request]),
            partisan_pluggable_peer_service_manager:forward_message(EncodedFrom, {error, not_primary}),

            {noreply, State#state{next_id=NextId+1}}
    end;
handle_cast({write, From0, Key, Value}, #state{next_id=NextId, membership=[Primary|Rest]=Membership, outstanding=Outstanding0, store=Store0}=State) ->
    %% TODO: Bit of a hack just to get this working.
    From = list_to_atom("write_coordinator_" ++ integer_to_list(NextId)),
    try
        true = erlang:unregister(From)
    catch
        _:_ ->
            ok
    end,
    true = erlang:register(From, From0),
    EncodedFrom = partisan_util:registered_name(From),

    %% TODO: HACK to get around problem in interprocedural analysis.
    Request = {write, EncodedFrom, Key, Value},

    case node() of 
        Primary ->

            %% Add to list of outstanding requests.
            Replies = [node()],
            Outstanding = dict:store(Request, {Membership, Replies}, Outstanding0),

            %% Send collaboration message to other nodes.
            CoordinatorNode = node(),

            lists:foreach(fun(Node) ->
                partisan_logger:info("sending collaborate message to node ~p for request ~p", [Node, Request]),
{'fraction_non_alphanumeric': 0.18902939727377238, 'fraction_numerical': 0.007226145508293644, 'mean_word_length': 2.877427570837313, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4195277', 'n_tokens_mistral': 3574, 'n_tokens_neox': 3394, 'n_words': 903}