syn/src/syn_backbone.erl

%% ==========================================================================================================
%% Syn - A global process registry.
%%
%% Copyright (C) 2015, Roberto Ostinelli <roberto@ostinelli.net>.
%% All rights reserved.
%%
%% The MIT License (MIT)
%%
%% Copyright (c) 2015 Roberto Ostinelli
%%
%% Permission is hereby granted, free of charge, to any person obtaining a copy
%% of this software and associated documentation files (the "Software"), to deal
%% in the Software without restriction, including without limitation the rights
%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%% copies of the Software, and to permit persons to whom the Software is
%% furnished to do so, subject to the following conditions:
%%
%% The above copyright notice and this permission notice shall be included in
%% all copies or substantial portions of the Software.
%%
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
%% THE SOFTWARE.
%% ==========================================================================================================
-module(syn_backbone).
-behaviour(gen_server).

%% API
-export([start_link/0]).
-export([initdb/0]).
-export([register/2, register/3]).
-export([unregister/1]).
-export([find_by_key/1, find_by_key/2]).
-export([find_by_pid/1, find_by_pid/2]).
-export([count/0, count/1]).

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

%% records
-record(state, {
    process_exit_callback_module = undefined :: atom(),
    process_exit_callback_function = undefined :: atom()
}).

%% include
-include("syn.hrl").


%% ===================================================================
%% API
%% ===================================================================
-spec start_link() -> {ok, pid()} | {error, any()}.
start_link() ->
    Options = [],
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], Options).

-spec initdb() -> ok | {error, any()}.
initdb() ->
    initdb_do().

-spec find_by_key(Key :: any()) -> pid() | undefined.
find_by_key(Key) ->
    case i_find_by_key(Key) of
        undefined -> undefined;
        Process -> Process#syn_processes_table.pid
    end.

-spec find_by_key(Key :: any(), with_meta) -> {pid(), Meta :: any()} | undefined.
find_by_key(Key, with_meta) ->
    case i_find_by_key(Key) of
        undefined -> undefined;
        Process -> {Process#syn_processes_table.pid, Process#syn_processes_table.meta}
    end.

-spec find_by_pid(Pid :: pid()) -> Key :: any() | undefined.
find_by_pid(Pid) ->
    case i_find_by_pid(Pid) of
        undefined -> undefined;
        Process -> Process#syn_processes_table.key
    end.

-spec find_by_pid(Pid :: pid(), with_meta) -> {Key :: any(), Meta :: any()} | undefined.
find_by_pid(Pid, with_meta) ->
    case i_find_by_pid(Pid) of
        undefined -> undefined;
        Process -> {Process#syn_processes_table.key, Process#syn_processes_table.meta}
    end.

-spec register(Key :: any(), Pid :: pid()) -> ok | {error, taken | pid_already_registered}.
register(Key, Pid) ->
    register(Key, Pid, undefined).

-spec register(Key :: any(), Pid :: pid(), Meta :: any()) -> ok | {error, taken}.
register(Key, Pid, Meta) ->
    Node = node(Pid),
    gen_server:call({?MODULE, Node}, {register_on_node, Key, Pid, Meta}).

-spec unregister(Key :: any()) -> ok | {error, undefined}.
unregister(Key) ->
    case find_by_key(Key) of
        undefined ->
            {error, undefined};
        Pid ->
            Node = node(Pid),
            gen_server:call({?MODULE, Node}, {unregister_on_node, Key, Pid})
    end.

-spec count() -> non_neg_integer().
count() ->
    mnesia:table_info(syn_processes_table, size).

-spec count(Node :: atom()) -> non_neg_integer().
count(Node) ->
    %% build match specs
    MatchHead = #syn_processes_table{node = '$2', _ = '_'},
    Guard = {'=:=', '$2', Node},
    Result = '$2',
    %% select
    Processes = mnesia:dirty_select(syn_processes_table, [{MatchHead, [Guard], [Result]}]),
    length(Processes).

%% ===================================================================
%% Callbacks
%% ===================================================================

%% ----------------------------------------------------------------------------------------------------------
%% Init
%% ----------------------------------------------------------------------------------------------------------
-spec init([]) ->
    {ok, #state{}} |
    {ok, #state{}, Timeout :: non_neg_integer()} |
    ignore |
    {stop, Reason :: any()}.
init([]) ->
    %% trap linked processes signal
    process_flag(trap_exit, true),

    %% get options
    {ok, [ProcessExitCallbackModule, ProcessExitCallbackFunction]} = syn_utils:get_env_value(
        process_exit_callback,
        [undefined, undefined]
    ),

    %% build state
    {ok, #state{
        process_exit_callback_module = ProcessExitCallbackModule,
        process_exit_callback_function = ProcessExitCallbackFunction
    }}.

%% ----------------------------------------------------------------------------------------------------------
%% Call messages
%% ----------------------------------------------------------------------------------------------------------
-spec handle_call(Request :: any(), From :: any(), #state{}) ->
    {reply, Reply :: any(), #state{}} |
    {reply, Reply :: any(), #state{}, Timeout :: non_neg_integer()} |
    {noreply, #state{}} |
    {noreply, #state{}, Timeout :: non_neg_integer()} |
    {stop, Reason :: any(), Reply :: any(), #state{}} |
    {stop, Reason :: any(), #state{}}.

handle_call({register_on_node, Key, Pid, Meta}, _From, State) ->
    %% check & register in gen_server process to ensure atomicity at node level without transaction lock
    %% atomicity is obviously not at cluster level, which is covered by syn_consistency.
    case i_find_by_key(Key) of
        undefined ->
            case i_find_by_pid(Pid) of
                undefined ->
                    %% add to table
                    mnesia:dirty_write(#syn_processes_table{
                        key = Key,
                        pid = Pid,
                        node = node(),
                        meta = Meta
                    }),
                    %% link
                    erlang:link(Pid),
                    %% return
                    {reply, ok, State};
                _ ->
                    {reply, {error, pid_already_registered}, State}
            end;
        _ ->
            {reply, {error, taken}, State}
    end;

handle_call({unregister_on_node, Key, Pid}, _From, State) ->
    remove_process_by_key(Key),
    erlang:unlink(Pid),
    {reply, ok, State};

handle_call({unlink_process, Pid}, _From, State) ->
    erlang:unlink(Pid),
    {reply, ok, State};

handle_call(Request, From, State) ->
    error_logger:warning_msg("Received from ~p an unknown call message: ~p", [Request, From]),
    {reply, undefined, State}.

%% ----------------------------------------------------------------------------------------------------------
%% Cast messages
%% ----------------------------------------------------------------------------------------------------------
-spec handle_cast(Msg :: any(), #state{}) ->
    {noreply, #state{}} |
    {noreply, #state{}, Timeout :: non_neg_integer()} |
    {stop, Reason :: any(), #state{}}.

handle_cast(Msg, State) ->
    error_logger:warning_msg("Received an unknown cast message: ~p", [Msg]),
    {noreply, State}.

%% ----------------------------------------------------------------------------------------------------------
%% All non Call / Cast messages
%% ----------------------------------------------------------------------------------------------------------
-spec handle_info(Info :: any(), #state{}) ->
    {noreply, #state{}} |
    {noreply, #state{}, Timeout :: non_neg_integer()} |
    {stop, Reason :: any(), #state{}}.

handle_info({'EXIT', Pid, Reason}, #state{
    process_exit_callback_module = ProcessExitCallbackModule,
    process_exit_callback_function = ProcessExitCallbackFunction
} = State) ->
    %% do not lock backbone
    spawn(fun() ->
        %% check if pid is in table
        case find_by_pid(Pid, with_meta) of
            undefined ->
                case Reason of
                    normal -> ok;
                    killed -> ok;
                    _ ->
                        error_logger:warning_msg("Received an exit message from an unlinked process ~p with reason: ~p", [Pid, Reason])
                end;
            {Key, Meta} ->
                %% delete from table
                remove_process_by_key(Key),
                %% log
                case Reason of
                    normal -> ok;
                    killed -> ok;
                    _ ->
                        error_logger:error_msg("Process with key ~p exited with reason: ~p", [Key, Reason])
                end,
                %% callback
                case ProcessExitCallbackModule of
                    undefined -> ok;
                    _ -> ProcessExitCallbackModule:ProcessExitCallbackFunction(Key, Pid, Meta, Reason)
                end
        end
    end),
    %% return
    {noreply, State};

handle_info(Info, State) ->
    error_logger:warning_msg("Received an unknown info message: ~p", [Info]),
    {noreply, State}.

%% ----------------------------------------------------------------------------------------------------------
%% Terminate
%% ----------------------------------------------------------------------------------------------------------
-spec terminate(Reason :: any(), #state{}) -> terminated.
terminate(Reason, _State) ->
    error_logger:info_msg("Terminating syn backbone with reason: ~p", [Reason]),
    terminated.

%% ----------------------------------------------------------------------------------------------------------
%% Convert process state when code is changed.
%% ----------------------------------------------------------------------------------------------------------
-spec code_change(OldVsn :: any(), #state{}, Extra :: any()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% ===================================================================
%% Internal
%% ===================================================================
-spec initdb_do() -> ok | {error, any()}.
initdb_do() ->
    %% get nodes
    CurrentNode = node(),
    ClusterNodes = [CurrentNode | nodes()],
    %% ensure all nodes are added
    mnesia:change_config(extra_db_nodes, ClusterNodes),
    %% ensure table exists
    case mnesia:create_table(syn_processes_table, [
        {type, set},
        {ram_copies, ClusterNodes},
        {attributes, record_info(fields, syn_processes_table)},
        {index, [#syn_processes_table.pid]},
        {storage_properties, [{ets, [{read_concurrency, true}]}]}
    ]) of
        {atomic, ok} ->
            error_logger:info_msg("syn_processes_table was successfully created"),
            ok;
        {aborted, {already_exists, syn_processes_table}} ->
            %% table already exists, try to add current node as copy
            add_table_copy_to_current_node();
        {aborted, {already_exists, syn_processes_table, CurrentNode}} ->
            %% table already exists, try to add current node as copy
            add_table_copy_to_current_node();
        Other ->
            error_logger:error_msg("Error while creating syn_processes_table: ~p", [Other]),
            {error, Other}
    end.

-spec add_table_copy_to_current_node() -> ok | {error, any()}.
add_table_copy_to_current_node() ->
    %% wait for table
    mnesia:wait_for_tables([syn_processes_table], 10000),
    %% add copy
    CurrentNode = node(),
    case mnesia:add_table_copy(syn_processes_table, CurrentNode, ram_copies) of
        {atomic, ok} ->
            error_logger:info_msg("Copy of syn_processes_table was successfully added to current node"),
            ok;
        {aborted, {already_exists, syn_processes_table}} ->
            error_logger:info_msg("Copy of syn_processes_table is already added to current node"),
            ok;
        {aborted, {already_exists, syn_processes_table, CurrentNode}} ->
            error_logger:info_msg("Copy of syn_processes_table is already added to current node"),
            ok;
        {aborted, Reason} ->
            error_logger:error_msg("Error while creating copy of syn_processes_table: ~p", [Reason]),
            {error, Reason}
    end.

-spec i_find_by_key(Key :: any()) -> Process :: #syn_processes_table{} | undefined.
i_find_by_key(Key) ->
    case mnesia:dirty_read(syn_processes_table, Key) of
        [Process] -> return_if_on_connected_node(Process);
        _ -> undefined
    end.

-spec i_find_by_pid(Pid :: pid()) -> Process :: #syn_processes_table{} | undefined.
i_find_by_pid(Pid) ->
    case mnesia:dirty_index_read(syn_processes_table, Pid, #syn_processes_table.pid) of
        [Process] -> return_if_on_connected_node(Process);
        _ -> undefined
    end.

-spec return_if_on_connected_node(Process :: #syn_processes_table{}) -> Process :: #syn_processes_table{} | undefined.
return_if_on_connected_node(Process) ->
    case lists:member(Process#syn_processes_table.node, [node() | nodes()]) of
        true -> Process;
        _ -> undefined
    end.

-spec remove_process_by_key(Key :: any()) -> ok.
remove_process_by_key(Key) ->
    mnesia:dirty_delete(syn_processes_table, Key).