%% @copyright 2024 Erlang Solutions Ltd. %% @doc Allows limiting the number of users' actions per interval. -module(amoc_throttle). %% API -export([start/2, stop/1, send/2, send/3, wait/1, run/2, pause/1, resume/1, unlock/1, change_rate/2, change_rate_gradually/2]). -type name() :: atom(). %% Atom representing the name of the throttle. -type rate() :: infinity | non_neg_integer(). %% Number of events per given `t:interval/0'. %% %% It can also be: %% -type interarrival() :: infinity | non_neg_integer(). %% Time in milliseconds between two events. %% %% It can also be: %% -type interval() :: non_neg_integer(). %% In milliseconds, defaults to 60000 (one minute). %% %% Note that an interval of zero means effectively allowing `t:rate/0' number of executions in %% parallel. It might be expected for this to be always `infinity' as a result of the limit when %% dividing by zero, but this needs to be made explicit in the `t:rate/0' by setting it to infinity. -type t() :: #{rate := rate(), interval => interval()} | #{interarrival := interarrival()}. %% Throttle unit of measurement -type gradual() :: #{from_rate := non_neg_integer(), to_rate := non_neg_integer(), interval => interval()} | #{from_interarrival := non_neg_integer(), to_interarrival := non_neg_integer()}. %% Configuration throttle for a gradual rate change. %% %% "from" and "to" prefixed parameters, whether rates or interarrivals, are required. %% `interval' applies only to rate and defaults to 1s. -type plan() :: #{step_interval := pos_integer(), step_count := pos_integer()} | #{duration := pos_integer()}. %% Configuration plan for a gradual rate change. %% %% The throttle mechanism will take a series of discrete steps, %% for as long as the duration given, %% or in the shape of the `step_interval' and `step_count'. -type gradual_plan() :: #{throttle := gradual(), plan := plan()}. %% Gradual plan details. Must specify a `t:gradual/0', and a `t:plan/0'. -export_type([t/0, name/0, rate/0, interval/0, interarrival/0, gradual_plan/0]). %% @doc Starts the throttle mechanism for a given `Name' with a given config. %% %% `Name' is needed to identify the rate as a single test can have different rates for different tasks. -spec start(name(), t() | rate()) -> {ok, started | already_started} | {error, any()}. start(Name, #{} = Config) -> case amoc_throttle_config:verify_config(Config) of {error, Error} -> {error, Error}; VerifiedConfig -> amoc_throttle_controller:ensure_throttle_processes_started(Name, VerifiedConfig) end; start(Name, Rate) when is_integer(Rate) -> start(Name, #{rate => Rate}). %% @doc Pauses executions for the given `Name' as if `Rate' was set to `0'. %% %% Does not stop the scheduled rate changes. `resume/1' undoes the pausing. -spec pause(name()) -> ok | {error, any()}. pause(Name) -> amoc_throttle_controller:pause(Name). %% @doc Resumes the executions for the given `Name', to their original configuration value. %% %% It is the counterpart to the `pause/1' API, resuming the execution of what that mechanism paused. -spec resume(name()) -> ok | {error, any()}. resume(Name) -> amoc_throttle_controller:resume(Name). %% @doc Unlocks executions for the given `Name' by setting `Rate' to `infinity'. -spec unlock(name()) -> ok | {error, any()}. unlock(Name) -> change_rate(Name, #{rate => infinity, interval => 0}). %% @doc Sets the throttle `Config' for `Name' according to the given values. -spec change_rate(name(), t() | rate()) -> ok | {error, any()}. change_rate(Name, #{} = Config) -> case amoc_throttle_config:verify_config(Config) of {error, Error} -> {error, Error}; VerifiedConfig -> amoc_throttle_controller:change_rate(Name, VerifiedConfig) end; change_rate(Name, Rate) when is_integer(Rate) -> change_rate(Name, #{rate => Rate}). %% @doc Allows to set a plan of gradual rate changes for a given `Name'. %% %% The configuration will be changed in a series of consecutive steps. %% Rates can be changed upwards as well as downwards. %% See the documentation for `t:gradual_plan/0' for more info. %% %% Be aware that, at first, the rate will be changed to the initial point given %% in the configuration, and this is not considered a step. -spec change_rate_gradually(name(), gradual_plan()) -> ok | {error, any()}. change_rate_gradually(Name, Config) -> case amoc_throttle_config:verify_gradual_config(Config) of {error, _} = Error -> Error; VerifiedConfig -> amoc_throttle_controller:change_rate_gradually(Name, VerifiedConfig) end. %% @doc Executes a given function `Fn' when it does not exceed the rate for `Name'. %% %% `Fn' is executed in the context of a new process spawned on the same node on which %% the process executing `run/2' runs, so a call to `run/2' is non-blocking. %% %% Diagram %% showing function execution flow in distributed environment. %% ``` %% title Amoc distributed %% participantgroup **Slave node** %% participant User %% participant Async runner %% end %% participantgroup **Master node** %% participant Throttle process %% end %% box left of User: request telemetry event %% %% User -> *Async runner : Fun %% %% User -> Throttle process : {schedule, Async runner PID} %% box right of Throttle process : request telemetry event %% %% ==throtlling delay== %% %% Throttle process -> Async runner: scheduled %% box right of Throttle process : execution telemetry event %% space -5 %% box left of Async runner : execution telemetry event %% abox over Async runner : scheduled action %% activate Async runner %% space %% deactivate Async runner %% Async runner ->Throttle process:'DOWN' %% destroy Async runner %% ''' -spec run(name(), fun(() -> any())) -> ok | {error, any()}. run(Name, Fn) -> amoc_throttle_runner:throttle(Name, Fn). %% @see send/3 %% @doc Sends a given message to `erlang:self()' -spec send(name(), any()) -> ok | {error, any()}. send(Name, Msg) -> amoc_throttle_runner:throttle(Name, {self(), Msg}). %% @doc Sends a given message `Msg' to a given `Pid' when the rate for `Name' allows for that. %% %% May be used to schedule tasks. -spec send(name(), pid(), any()) -> ok | {error, any()}. send(Name, Pid, Msg) -> amoc_throttle_runner:throttle(Name, {Pid, Msg}). %% @doc Blocks the caller until the throttle mechanism allows. -spec wait(name()) -> ok | {error, any()}. wait(Name) -> amoc_throttle_runner:throttle(Name, wait). %% @doc Stops the throttle mechanism for the given `Name'. -spec stop(name()) -> ok | {error, any()}. stop(Name) -> amoc_throttle_controller:stop(Name).