eharbor

%% @author Eric des Courtis <eric.descourtis@fireworkhq.com>
%% @doc eharbor is an Erlang library that allows for running a
%% function in a controlled manner, with the ability to deduplicate
%% function calls with the same arguments. It features configurable
%% deduplication, ordered function calls, and buffering to prevent
%% system overload. This library is useful for running expensive tasks,
%% where deduplication of the same inputs can lead to significant
%% performance gains.
%%
%% The config type is a map with the following keys:
%%
%% `name' (default: `default') The name of the harbor instance. This name is used to derive
%%        a coordinator name and a pobox name.
%%
%% `backlog' (default: `1000') How big should our pobox buffer be? This is
%%           effectively going to determine how much memory the pobox will
%%           be allowed to consume. The underlying pobox acts as our load
%%           shedding for overload scenarios. Don't set this parameter to
%%           an arbitrary value instead try to calculate how much memory will
%%           use when the pobox will be full to make sure that the system
%%           will survive.
%%
%% `piers' (default: `100') How many unique parameters to process at once. This
%%         effectively limits concurrency on the backend. It can be used as a
%%         bulkhead by providing a limit that is smaller than connection pool
%%         size for a database like Postgres for example.
%%
%% `breakwater_limit' (default: `2000') Maximum number of processes waiting
%%                    on a result being processed by the backend. This does
%%                    not include the `backlog' processes. Keep in mind this
%%                    value should not be to large otherwise memory spikes
%%                    could occur during fanout of the results from conductor
%%                    to follower.
%%
%% `dedup' (default: `true') Is this instance being used as a simple bulkhead
%%         or are we tring to accelerate the requests by deduplicating them.
%%
%% `ordered' (default: `true') For a sequential process P where no other sequential
%%           process writes. If the process reads and then writes when this parameter
%%           is set to false the write may appear to have occured before the read.
%%           This results in lower latency and lower overhead at the cost of ordering.
%%           The parameter is set to true by default since it that behavior is
%%           more intuitive. But if maximum performance is desired set it to false.
%%
%% `group_by_key_fun' (default: `fun(Key) -> Key end') Sometimes some parameters need
%%                    to be passed into harbor but shouldn't be considered for
%%                    pier assignment or deduplication. This allows you to ignore
%%                    some parameters while still passing them into your function.
%%
%% `error_type' (default: `value') How to handle scenarios where the pobox is full
%%              what type of error to return (should be one of these `throw | error | raise | exit | value').
%%
%% `error_value' (default: `{error, full}') The value of the throw error, exit etc.
%%               In this case we return a value because the default is `value'.
%%
%% `buffer_insert_timeout' (default: `5000') How long to wait for insertion into the
%%                         pobox buffer normally changing this should not be required.
%%
%% `follower_wait_for_conductor_timeout' (default: `infinity') How long the follower should
%%                                       wait after the conductor. Normally you shouldn't
%%                                       have to modify this value since crashes are detected
%%                                       and will not cause followers to wait until timeout.
%%
%% `conductor_wait_for_coordinator_followers_timeout' (default: `infinity') Don't touch unless
%%                                                    you know exactly what you are doing.
%% `assign_role_timeout' (default: `infinity') Don't touch unless you know exactly what you
%%                       are doing.
%%
-module(eharbor).

-export([run/2, run/3, config_defaults/0, merge_with_defaults/1, pobox_name/1,
         coordinator_name/1]).

-export_type([incomplete_config/0, config/0]).

-type config() ::
    #{name := atom(),
      backlog := non_neg_integer(),
      piers := non_neg_integer(),
      breakwater_limit := non_neg_integer(),
      dedup := boolean(),
      ordered := boolean(),
      group_by_key_fun := fun((any()) -> any()),
      error_type := throw | error | raise | exit | value,
      error_value := any(),
      buffer_insert_timeout := timeout(),
      follower_wait_for_conductor_timeout := timeout(),
      conductor_wait_for_coordinator_followers_timeout := timeout(),
      assign_role_timeout := timeout()}.
-type incomplete_config() ::
    #{name => atom(),
      backlog => non_neg_integer(),
      piers => non_neg_integer(),
      breakwater_limit => non_neg_integer(),
      dedup => boolean(),
      ordered => boolean(),
      group_by_key_fun => fun((any()) -> any()),
      error_type => throw | error | raise | exit | value,
      error_value => any(),
      buffer_insert_timeout => timeout(),
      follower_wait_for_conductor_timeout => timeout(),
      conductor_wait_for_coordinator_followers_timeout => timeout(),
      assign_role_timeout => timeout()}.

-define(CONFIG_DEFAULTS,
        #{name => default,
          backlog => 1000,
          piers => 100,
          breakwater_limit => 2000,
          dedup => true,
          ordered => true,
          group_by_key_fun => fun(Key) -> Key end,
          error_type => value,
          error_value => {error, full},
          buffer_insert_timeout => 5000,
          follower_wait_for_conductor_timeout => infinity,
          conductor_wait_for_coordinator_followers_timeout => infinity,
          assign_role_timeout => infinity}).

%% @doc start_link the harbor instance using a configuration
%%
%% @end
-spec start_link(Config :: config()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}.
start_link(Config) ->
    eharbor_coordinator:start_link(Config).

%% @doc Return the configuration defaults for harbor.
%%
%% `config_defaults/2' returns the default configuration parameters.
%% This is later used to convert an `incomplete_config()' into a
%% `config()'.
%%
%% @see merge_with_defaults/1
%% @end
-spec config_defaults() -> config().
config_defaults() ->
    ?CONFIG_DEFAULTS.

%% @doc Merge an incomplete configuration with the defaults to produce a proper config.
%%
%% `merge_with_defaults/2' returns overridden parameters merged with the default
%%  configuration parameters when the paramter is missing.
%%
%% @see config_defaults/0
%% @end
-spec merge_with_defaults(incomplete_config()) -> config().
merge_with_defaults(Map) when is_map(Map) ->
    maps:merge_with(fun(_Key, Value1, Value2) ->
                       case Value2 of
                           undefined ->
                               Value1;
                           _ ->
                               Value2
                       end
                    end,
                    eharbor:config_defaults(),
                    Map).

%% @doc Run a function using a function reference and arguments with the
%%      config using a harbor instance.
%%
%% @end
-spec run(fun((any()) -> any()), [any()], config()) -> any() | no_return().
run(Function, Arguments, Config = #{ordered := Ordered}) ->
    do_run(Function, Arguments, Config, Ordered).

do_run(Function,
       Arguments,
       Config =
           #{name := Name,
             buffer_insert_timeout := BufferInsertTimeout,
             follower_wait_for_conductor_timeout := FollowerWaitForConductorTimeout,
             conductor_wait_for_coordinator_followers_timeout :=
                 ConductorWaitForCoordinatorFollowersTimeout,
             assign_role_timeout := AssignRoleTimeout,
             group_by_key_fun := GroupByKeyFun,
             error_type := ErrorType,
             error_value := ErrorValue},
       Ordered) ->
    PoboxName = pobox_name(Name),
    CoordinatorName = coordinator_name(Name),
    %% Start monitoring the coordinator before we start so we don't
    %% wait forever if it dies
    MRef = erlang:monitor(process, erlang:whereis(CoordinatorName)),

    %% Ask pobox to store our arguments (transformed by group_by_key_fun)
    %% This is how we dock into the harbor. If the buffer is full it will
    %% return :full
    case pobox:post_sync(PoboxName,
                         {'$habor_dock', {self(), MRef}, apply(GroupByKeyFun, Arguments)},
                         BufferInsertTimeout)
    of
        %% We successfully docked into the harbor
        ok ->
            %% Wait until we are assigned a role
            receive
                {MRef, {follower, ConductorPid}} ->
                    %% We got assigned a follower role lets monitor the conductor for this pier
                    %% since he is doing the work
                    ConductorMRef = erlang:monitor(process, ConductorPid),
                    receive
                        {MRef, Result0} ->
                            %% We have our result computed by the conductor
                            %% We can stop monitoring and flush monitoring messages
                            erlang:demonitor(MRef, [flush]),
                            erlang:demonitor(ConductorMRef, [flush]),
                            %% Ensure ordering (only the follower needs to care
                            %% because it may be joining an in progress operation
                            case Ordered of
                                true ->
                                    %% Run again we have to wait for the current operation
                                    %% the next one with be synchronized
                                    do_run(Function, Arguments, Config, false);
                                false ->
                                    %% Error was encapsulated by the conductor lets unpack it so we behave
                                    %% as if we executed it
                                    case Result0 of
                                        %% Return the result
                                        {success, Result1} ->
                                            Result1;
                                        {value, Val} ->
                                            Val;
                                        {Class, Reason, Stacktrace} ->
                                            erlang:raise(Class, Reason, Stacktrace)
                                    end
                            end;
                        {'DOWN', MRef, _, _, Reason} ->
                            %% Coordinator died clean up and exit with the same reason
                            erlang:demonitor(MRef, [flush]),
                            erlang:demonitor(ConductorMRef, [flush]),
                            erlang:exit(Reason);
                        {'DOWN', ConductorMRef, _, _, Reason} ->
                            %% Conductor died clean up and exit
                            erlang:demonitor(MRef, [flush]),
                            erlang:demonitor(ConductorMRef, [flush]),
                            erlang:exit({conductor_died, Reason})
                    after FollowerWaitForConductorTimeout ->
                        erlang:demonitor(MRef, [flush]),
                        erlang:exit(timeout)
                    end;
                {MRef, conductor} ->
                    %%  We are assigned the conductor role by the coordinator
                    Result0 =
                        try
                            %% Run the real function specified by the user and encapsulate it
                            %% in a success tuple
                            {success, apply(Function, Arguments)}
                        catch
                            Class0:Reason0:Stacktrace0 ->
                                {Class0, Reason0, Stacktrace0}
                        end,

                    %% Let the coordinator know that we are done the work so it can stop
                    %% accumulating followers for this pier and send use the list.
                    %% We are sending it a reply to reference to call us back with the details.
                    erlang:send(CoordinatorName, {'$harbor_work_completed', {erlang:self(), MRef}}),

                    receive
                        {MRef, {followers, Followers}} ->
                            %% We have the list of followers from the coordinator now we can fan out
                            %% the result and stop monitoring the coordinator since we are now undocked
                            %% from the harbor
                            erlang:demonitor(MRef, [flush]),
                            lists:foreach(fun(Follower) -> gen:reply(Follower, Result0) end,
                                          Followers);
                        {'DOWN', MRef, _, _, Reason1} ->
                            %% Coordinator died clean up and exit
                            erlang:demonitor(MRef, [flush]),
                            erlang:exit(Reason1)
                    after ConductorWaitForCoordinatorFollowersTimeout ->
                        erlang:demonitor(MRef, [flush]),
                        erlang:exit(timeout)
                    end,

                    %% Error was encapsulated by the conductor lets unpack it so we behave
                    %% as if we executed it
                    case Result0 of
                        %% Return the result
                        {success, Result1} ->
                            Result1;
                        {value, Val} ->
                            Val;
                        {Class2, Reason2, Stacktrace2} ->
                            erlang:raise(Class2, Reason2, Stacktrace2)
                    end;
                {'DOWN', MRef, _, _, Reason} ->
                    %% Coordinator died clean up and exit
                    erlang:demonitor(MRef, [flush]),
                    erlang:exit(Reason)
            after AssignRoleTimeout ->
                erlang:demonitor(MRef, [flush]),
                erlang:exit(timeout)
            end;
        full ->
            %% We can't dock into the harbor the backlog is full, there is no
            %% space available in the buffer. Emit the user chosen error result
            erlang:demonitor(MRef, [flush]),
            case ErrorType of
                throw ->
                    erlang:throw(ErrorValue);
                error ->
                    erlang:error(ErrorValue);
                exit ->
                    erlang:exit(ErrorValue);
                value ->
                    ErrorValue
            end
    end.

%% @doc Run a function using an MFA with the config using a harbor instance.
%%
%% @end
-spec run(MFA :: mfa(), Config :: config()) -> any() | no_return().
run({Module, Function, Arguments}, Config) ->
    F = fun(Args) -> erlang:apply(Module, Function, Args) end,
    run(F, Arguments, Config).

%% @doc From the harbor name return the derived pobox name.
%%
%% The pobox is a named process and to talk to it directly
%% some operations require its name.
%%
%% @end
-spec pobox_name(Name :: atom()) -> atom().
pobox_name(Name) ->
    erlang:binary_to_atom(<<(erlang:atom_to_binary(Name))/binary, "_pobox">>).

%% @doc From the harbor name return the derived coordinator name.
%%
%% The coordinator is a named process and to talk to it directly
%% some operations require its name.
%%
%% @end
-spec coordinator_name(Name :: atom()) -> atom().
coordinator_name(Name) ->
    erlang:binary_to_atom(<<(erlang:atom_to_binary(Name))/binary, "_coordinator">>).