%% @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">>).
