<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2017 <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(partisan_causality_backend).
-author("<NAME> <<EMAIL>>").

-behaviour(gen_server).

%% API
-export([start_link/0,
         emit/3,
         receive_message/1,
         set_delivery_fun/1]).

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

-record(state, {myself, local_clock, order_buffer, buffered_messages, delivery_fun}).

%% TODO: Needs some sort of persistence for the local causality information.
%% TODO: How do you know if we can use a causal channel or not?
%% TODO: GC.
%% TODO: Add acknowledgements.

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

%% @doc Same as start_link([]).
start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

emit(Node, ServerRef, Message) ->
    gen_server:call(?MODULE, {emit, Node, ServerRef, Message}, infinity).

receive_message(Message) ->
    gen_server:call(?MODULE, {receive_message, Message}, infinity).

set_delivery_fun(DeliveryFun) ->
    gen_server:call(?MODULE, {set_delivery_fun, DeliveryFun}, infinity).

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

%% @private
init([]) ->
    %% Figure out who we are.
    Myself = partisan_peer_service_manager:myself(),

    %% Generate a local clock that's used to track local dependencies.
    LocalClock = vclock:fresh(),

    %% Initiaize order buffer.
    OrderBuffer = orddict:new(),

    %% Generate message buffer.
    BufferedMessages = [],

    %% Schedule delivery attempts.
    schedule_delivery(),

    {ok, #state{myself=Myself, 
                local_clock=LocalClock, 
                order_buffer=OrderBuffer, 
                buffered_messages=BufferedMessages}}.

%% @private

%% Generate a message identifier and a payload to be transmitted on the wire.
handle_call({emit, Node, ServerRef, Message}, 
            _From, 
            #state{myself=#{name := Myself}, local_clock=LocalClock0, order_buffer=OrderBuffer0}=State) ->
    %% Bump our local clock.
    LocalClock = vclock:increment(Myself, LocalClock0),

    %% Only transmit order buffer containing single clock.
    FilteredOrderBuffer = orddict:filter(fun(Key, _Value) -> Key =:= Node end, OrderBuffer0),

    %% Return the message to be transmitted.
    FullMessage = {causal, Node, ServerRef, FilteredOrderBuffer, LocalClock, Message},

    %% Update the order buffer with node and mesage clock.
    OrderBuffer = orddict:store(Node, LocalClock, OrderBuffer0),

    lager:info("Emitting message with clock: ~p", [LocalClock]),

    {reply, {ok, FullMessage}, State#state{local_clock=LocalClock, order_buffer=OrderBuffer}};

%% Receive a causal messag off the wire; deliver or not depending on whether or not
%% the causal dependencies have been satisfied.
handle_call({receive_message, {causal, _Node, _ServerRef, _IncomingOrderBuffer, MessageClock, _Message}=FullMessage}, 
            _From, 
            #state{buffered_messages=BufferedMessages0}=State0) ->
    %% Add to the buffer and try to deliver.
    lager:info("Received message ~p and inserting into buffer.", [MessageClock]),
    BufferedMessages = BufferedMessages0 ++ [FullMessage],
    State = lists:foldl(fun(M, S) -> receive_message(M, S) end, State0#state{buffered_messages=BufferedMessages}, BufferedMessages),
    {reply, ok, State};

handle_call({set_delivery_fun, DeliveryFun}, _From, State) ->
    {reply, ok, State#state{delivery_fun=DeliveryFun}};

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

%% @private
handle_cast(_Msg, State) ->
    {noreply, State}.

%% @private
handle_info(deliver, #state{buffered_messages=BufferedMessages}=State0) ->
    State = lists:foldl(fun(M, S) -> receive_message(M, S) end, State0, BufferedMessages),
    schedule_delivery(),
    {noreply, State};
handle_info(_Msg, State) ->
    {noreply, State}.

%% @private
terminate(_Reason, _State) ->
    ok.

%% @private
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
deliver(#state{myself=Myself, local_clock=LocalClock, order_buffer=OrderBuffer, delivery_fun=DeliveryFun}=State, 
        IncomingOrderBuffer, MessageClock, ServerRef, Message) ->
    %% Merge order buffers.
    MergeFun = fun(_Key, Value1, Value2) ->
        vclock:merge([Value1, Value2])
    end,
    orddict:merge(MergeFun, IncomingOrderBuffer, OrderBuffer),

    %% Merge clocks.
    MergedLocalClock = vclock:merge([LocalClock, MessageClock]),
<commit_msg>Move TODOs.<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2017 <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(partisan_causality_backend).
-author("<NAME> <<EMAIL>>").

-behaviour(gen_server).

%% API
-export([start_link/0,
         emit/3,
         receive_message/1,
         set_delivery_fun/1]).

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

-record(state, {myself, local_clock, order_buffer, buffered_messages, delivery_fun}).

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

%% @doc Same as start_link([]).
start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

emit(Node, ServerRef, Message) ->
    gen_server:call(?MODULE, {emit, Node, ServerRef, Message}, infinity).

receive_message(Message) ->
    gen_server:call(?MODULE, {receive_message, Message}, infinity).

set_delivery_fun(DeliveryFun) ->
    gen_server:call(?MODULE, {set_delivery_fun, DeliveryFun}, infinity).

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

%% @private
init([]) ->
    %% Figure out who we are.
    Myself = partisan_peer_service_manager:myself(),

    %% Generate a local clock that's used to track local dependencies.
    LocalClock = vclock:fresh(),

    %% Initiaize order buffer.
    OrderBuffer = orddict:new(),

    %% Generate message buffer.
    BufferedMessages = [],

    %% Schedule delivery attempts.
    schedule_delivery(),

    {ok, #state{myself=Myself, 
                local_clock=LocalClock, 
                order_buffer=OrderBuffer, 
                buffered_messages=BufferedMessages}}.

%% @private

%% Generate a message identifier and a payload to be transmitted on the wire.
handle_call({emit, Node, ServerRef, Message}, 
            _From, 
            #state{myself=#{name := Myself}, local_clock=LocalClock0, order_buffer=OrderBuffer0}=State) ->
    %% Bump our local clock.
    LocalClock = vclock:increment(Myself, LocalClock0),

    %% Only transmit order buffer containing single clock.
    FilteredOrderBuffer = orddict:filter(fun(Key, _Value) -> Key =:= Node end, OrderBuffer0),

    %% Return the message to be transmitted.
    FullMessage = {causal, Node, ServerRef, FilteredOrderBuffer, LocalClock, Message},

    %% Update the order buffer with node and mesage clock.
    OrderBuffer = orddict:store(Node, LocalClock, OrderBuffer0),

    lager:info("Emitting message with clock: ~p", [LocalClock]),

    {reply, {ok, FullMessage}, State#state{local_clock=LocalClock, order_buffer=OrderBuffer}};

%% Receive a causal messag off the wire; deliver or not depending on whether or not
%% the causal dependencies have been satisfied.
handle_call({receive_message, {causal, _Node, _ServerRef, _IncomingOrderBuffer, MessageClock, _Message}=FullMessage}, 
            _From, 
            #state{buffered_messages=BufferedMessages0}=State0) ->
    %% Add to the buffer and try to deliver.
    lager:info("Received message ~p and inserting into buffer.", [MessageClock]),
    BufferedMessages = BufferedMessages0 ++ [FullMessage],
    State = lists:foldl(fun(M, S) -> receive_message(M, S) end, State0#state{buffered_messages=BufferedMessages}, BufferedMessages),
    {reply, ok, State};

handle_call({set_delivery_fun, DeliveryFun}, _From, State) ->
    {reply, ok, State#state{delivery_fun=DeliveryFun}};

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

%% @private
handle_cast(_Msg, State) ->
    {noreply, State}.

%% @private
handle_info(deliver, #state{buffered_messages=BufferedMessages}=State0) ->
    State = lists:foldl(fun(M, S) -> receive_message(M, S) end, State0, BufferedMessages),
    schedule_delivery(),
    {noreply, State};
handle_info(_Msg, State) ->
    {noreply, State}.

%% @private
terminate(_Reason, _State) ->
    ok.

%% @private
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
deliver(#state{myself=Myself, local_clock=LocalClock, order_buffer=OrderBuffer, delivery_fun=DeliveryFun}=State, 
        IncomingOrderBuffer, MessageClock, ServerRef, Message) ->
    %% Merge order buffers.
    MergeFun = fun(_Key, Value1, Value2) ->
        vclock:merge([Value1, Value2])
    end,
    orddict:merge(MergeFun, IncomingOrderBuffer, OrderBuffer),

    %% Prune order buffers.

    %% Merge clocks.
    MergedLocalClock = vclock:merge([LocalClock, MessageClock]),

{'fraction_non_alphanumeric': 0.21305718144979738, 'fraction_numerical': 0.005943268797838811, 'mean_word_length': 3.8603938730853393, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 49, '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': '5696996', 'n_tokens_mistral': 3236, 'n_tokens_neox': 3037, 'n_words': 832}