pooler

%% @author Seth Falcon <seth@userprimary.net>
%% @copyright 2012-2013 Seth Falcon
%% @doc Helper gen_server to manage async member lifecycle (start and stop)
%%
-module(pooler_starter).
-behaviour(gen_server).

-include_lib("kernel/include/logger.hrl").

-define(POOLER_POOL_NAME, '$pooler_pool_name').
-define(POOLER_PID, '$pooler_pid').
-define(DEFAULT_STOP_MFA, {supervisor, terminate_child, [?POOLER_POOL_NAME, ?POOLER_PID]}).

%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export([
    start_link/1,
    start_member/2,
    start_member/3,
    start_member/4,
    stop_member_async/1,
    stop/1,
    stop_spec/3,
    default_stop_mfa/0,
    replace_placeholders/3
]).

%% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------

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

-export_type([start_spec/0, start_result/0, stop_spec/0, stop_mfa/0]).

-type pool_member_sup() :: pid() | atom().
-type parent() :: pid() | pool.
-type initialize_mfa() :: undefined | {module(), atom(), [term()]}.
-type stop_mfa() :: {module(), atom(), ['$pooler_pid' | '$pooler_pool_name' | term()]}.
-type start_result() :: {StarterPid :: pid(), Result :: pid() | {error, _}}.
-opaque start_spec() :: {pooler:pool_name(), pool_member_sup(), parent(), initialize_mfa()}.
%% {PoolName, MemberPid, StopMFA}
-opaque stop_spec() :: {pooler:pool_name(), pid(), stop_mfa()}.

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

-spec start_link(start_spec() | stop_spec()) -> {ok, pid()}.
start_link({_, _, _} = Spec) ->
    gen_server:start_link(?MODULE, Spec, []);
start_link({_, _, _, _} = Spec) ->
    gen_server:start_link(?MODULE, Spec, []).

stop(Starter) ->
    gen_server:cast(Starter, stop).

%% @doc Start a member for the specified `Pool'.
%%
%% Member creation with this call is async. This function returns
%% immediately with create process' pid. When the member has been
%% created it is sent to the specified pool via
%% {@link pooler:accept_member/2}.
%%
%% Each call starts a single use `pooler_starter' instance via
%% `pooler_starter_sup'. The instance terminates normally after
%% creating a single member.
-spec start_member(pooler:pool_name(), pool_member_sup()) -> pid().
start_member(PoolName, PoolMemberSup) ->
    start_member(PoolName, PoolMemberSup, undefined).

-spec start_member(pooler:pool_name(), pool_member_sup(), initialize_mfa()) -> pid().
start_member(PoolName, PoolMemberSup, InitMFA) ->
    {ok, Pid} = pooler_starter_sup:new_starter({PoolName, PoolMemberSup, pool, InitMFA}),
    Pid.

%% @doc Same as {@link start_member/2} except that instead of calling
%% {@link pooler:accept_member/2} a raw message is sent to `Parent' of
%% the form `{accept_member, {Ref, Member}'. Where `Member' will
%% either be the member pid or an error term and `Ref' will be the
%% Pid of the starter.
%%
%% This is used by the init function in the `pooler' to start the
%% initial set of pool members in parallel.
-spec start_member(pooler:pool_name(), pool_member_sup(), pid(), initialize_mfa()) -> pid().
start_member(PoolName, PoolMemberSup, Parent, InitMFA) ->
    {ok, Pid} = pooler_starter_sup:new_starter({PoolName, PoolMemberSup, Parent, InitMFA}),
    Pid.

%% @doc Stop a member in the pool

%% Member creation can take too long. In this case, the starter
%% needs to be informed that even if creation succeeds, the
%% started child should be not be sent back and should be
%% cleaned up
-spec stop_member_async(pid()) -> ok.
stop_member_async(Pid) ->
    gen_server:cast(Pid, stop_member).

%% ------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------
-record(starter, {
    %% start mode
    parent :: parent() | undefined,
    pool_name :: pooler:pool_name(),
    pool_member_sup :: pool_member_sup() | undefined,
    initialize_mfa :: initialize_mfa(),
    msg :: start_result() | undefined,
    %% stop mode
    stopping_pid = undefined :: pid() | undefined,
    stopping_mfa = undefined :: stop_mfa() | undefined
}).

-spec init(start_spec() | stop_spec()) -> {ok, #starter{}, {continue, start | stop}}.
init({PoolName, MemberPid, StopMFA}) ->
    {ok,
        #starter{
            pool_name = PoolName,
            stopping_pid = MemberPid,
            stopping_mfa = StopMFA
        },
        {continue, stop}};
init({PoolName, PoolMemberSup, Parent, InitMFA}) ->
    {ok, #starter{pool_name = PoolName, pool_member_sup = PoolMemberSup, parent = Parent, initialize_mfa = InitMFA},
        {continue, start}}.

handle_continue(
    start,
    #starter{pool_member_sup = PoolSup, pool_name = PoolName, initialize_mfa = InitMFA} = State
) ->
    Msg = do_start_member(PoolSup, PoolName, InitMFA),
    % asynchronously in order to receive potential `stop*'
    accept_member_async(self()),
    {noreply, State#starter{msg = Msg}};
handle_continue(
    stop,
    #starter{pool_name = PoolName, stopping_pid = MemberPid, stopping_mfa = StopMFA} = State
) ->
    terminate_pid(PoolName, MemberPid, StopMFA),
    {stop, normal, State}.

handle_call(_Request, _From, State) ->
    {noreply, State}.

handle_cast(stop_member, #starter{msg = {_Me, Pid}, pool_member_sup = MemberSup} = State) when is_pid(Pid) ->
    %% The process we were starting is no longer valid for the pool.
    %% Cleanup the process and stop normally.
    supervisor:terminate_child(MemberSup, Pid),
    {stop, normal, State};
handle_cast(stop_member, State) ->
    %% Either init failed (msg is an error) or start_member hasn't completed yet.
    {stop, normal, State};
handle_cast(accept_member, #starter{msg = Msg, parent = Parent, pool_name = PoolName} = State) ->
    %% Process creation has succeeded. Send the member to the pooler
    %% gen_server to be accepted. Pooler gen_server will notify
    %% us if the member was accepted or needs to cleaned up.
    send_accept_member(Parent, PoolName, Msg),
    {noreply, State};
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(_Request, State) ->
    {noreply, State}.

handle_info(_Info, State) ->
    {noreply, State}.

-spec terminate(_, _) -> 'ok'.
terminate(_Reason, _State) ->
    ok.

-spec code_change(_, _, _) -> {'ok', _}.
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

do_start_member(PoolSup, PoolName, InitMFA) ->
    case supervisor:start_child(PoolSup, []) of
        {ok, Pid} ->
            case call_initialize_mfa(PoolName, Pid, InitMFA) of
                ok ->
                    {self(), Pid};
                Error ->
                    ?LOG_ERROR(
                        #{
                            label => "failed to initialize member",
                            pool => PoolName,
                            pid => Pid,
                            error => Error
                        },
                        #{domain => [pooler]}
                    ),
                    supervisor:terminate_child(PoolSup, Pid),
                    {self(), Error}
            end;
        Error ->
            ?LOG_ERROR(
                #{
                    label => "failed to start member",
                    pool => PoolName,
                    error => Error
                },
                #{domain => [pooler]}
            ),
            {self(), Error}
    end.

-spec default_stop_mfa() -> stop_mfa().
default_stop_mfa() ->
    ?DEFAULT_STOP_MFA.

-spec stop_spec(pooler:pool_name(), pid(), stop_mfa()) -> stop_spec().
stop_spec(PoolName, MemberPid, StopMFA) ->
    {PoolName, MemberPid, StopMFA}.

%% @doc Best-effort termination for a pool member: applies the given MFA with
%% `?POOLER_PID' and `?POOLER_POOL_NAME' placeholders replaced by the actual
%% pid and pool name. Falls back to the default stop MFA on any failure.
-spec terminate_pid(pooler:pool_name(), pid(), stop_mfa()) -> ok.
terminate_pid(PoolName, Pid, {Mod, Fun, Args}) when is_list(Args) ->
    NewArgs = replace_placeholders(PoolName, Pid, Args),
    try erlang:apply(Mod, Fun, NewArgs) of
        _ -> ok
    catch
        _:_ -> terminate_pid(PoolName, Pid, ?DEFAULT_STOP_MFA)
    end.

-spec replace_placeholders(pooler:pool_name(), pid(), [term()]) -> [term()].
replace_placeholders(PoolName, Pid, Args) ->
    [
        case Arg of
            ?POOLER_POOL_NAME -> pooler_pool_sup:build_member_sup_name(PoolName);
            ?POOLER_PID -> Pid;
            _ -> Arg
        end
     || Arg <- Args
    ].

-spec call_initialize_mfa(pooler:pool_name(), pid(), initialize_mfa()) -> ok | {error, term()}.
call_initialize_mfa(_PoolName, _Pid, undefined) ->
    ok;
call_initialize_mfa(PoolName, Pid, {Mod, Fun, Args}) ->
    NewArgs = replace_placeholders(PoolName, Pid, Args),
    try erlang:apply(Mod, Fun, NewArgs) of
        ok -> ok;
        {error, _} = Err -> Err;
        Other -> {error, {unexpected_initialize_mfa_return, Other}}
    catch
        _:Reason -> {error, {initialize_mfa_exit, Reason}}
    end.

-spec send_accept_member(parent(), pooler:pool_name(), start_result()) -> ok.
send_accept_member(pool, PoolName, Msg) ->
    %% used to grow pool
    pooler:accept_member(PoolName, Msg);
send_accept_member(Pid, _PoolName, Msg) ->
    %% used during pool initialization
    Pid ! {accept_member, Msg},
    ok.

accept_member_async(Pid) ->
    gen_server:cast(Pid, accept_member).