%% @author Seth Falcon %% @copyright 2011-2013 Seth Falcon %% @doc This is the main interface to the pooler application %% %% To integrate with your application, you probably want to call %% `application:start(pooler)' after having specified appropriate %% configuration for the pooler application (either via a config file %% or appropriate calls to the application module to set the %% application's config). %% -module(pooler). -behaviour(gen_server). -include_lib("kernel/include/logger.hrl"). %% ------------------------------------------------------------------ %% API Function Exports %% ------------------------------------------------------------------ -export([ start/0, stop/0 ]). -export([ accept_member/2, start_link/1, take_member/1, take_member/2, take_group_member/1, return_group_member/2, return_group_member/3, group_pools/1, return_member/2, return_member/3, pool_stats/1, pool_utilization/1, manual_start/0, new_pool/1, pool_child_spec/1, pool_reconfigure/2, rm_pool/1, rm_group/1, call_free_members/2, call_free_members/3, metrics/0 ]). -export([create_group_table/0, config_as_map/1, to_map/1]). %% ------------------------------------------------------------------ %% gen_server Function Exports %% ------------------------------------------------------------------ -export([ init/1, handle_continue/2, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3 ]). -vsn(4). %% Bump this value and add a new clause to `code_change', if the format of `#pool{}' record changed %% ------------------------------------------------------------------ %% Types %% ------------------------------------------------------------------ -export_type([ pool_config/0, pool_config_legacy/0, pool_name/0, group_name/0, member_info/0, time_unit/0, time_spec/0, metric_type/0, label_values/0, label_spec/0, metric_name/0, metric_info/0 ]). -define(DEFAULT_ADD_RETRY, 1). -define(DEFAULT_CULL_INTERVAL, {1, min}). -define(DEFAULT_MAX_AGE, {30, sec}). -define(DEFAULT_MEMBER_START_TIMEOUT, {1, min}). -define(DEFAULT_AUTO_GROW_THRESHOLD, undefined). -define(POOLER_GROUP_TABLE, pooler_group_table). -define(DEFAULT_POOLER_QUEUE_MAX, 50). %% Per-pool TTL configuration and timer state. %% `undefined' in #pool.ttl means TTL is disabled — all TTL functions short-circuit immediately. -record(ttl, { max_lifetime :: time_spec(), jitter = {0, sec} :: time_spec(), timer = undefined :: reference() | undefined, timer_target = undefined :: pid() | undefined }). -record(pool, { name :: atom(), group :: atom(), max_count = 100 :: non_neg_integer(), init_count = 10 :: non_neg_integer(), start_mfa :: {atom(), atom(), [term()]}, free_pids = [] :: [pid()], in_use_count = 0 :: non_neg_integer(), free_count = 0 :: non_neg_integer(), %% The number times to attempt adding a pool member if the %% pool size is below max_count and there are no free %% members. After this many tries, error_no_members will be %% returned by a call to take_member. NOTE: this value %% should be >= 2 or else the pool will not grow on demand %% when max_count is larger than init_count. %% TODO: seems to be not in use anymore add_member_retry = ?DEFAULT_ADD_RETRY :: non_neg_integer(), %% The interval to schedule a cull message. Both %% 'cull_interval' and 'max_age' are specified using a %% `time_spec()' type. cull_interval = ?DEFAULT_CULL_INTERVAL :: time_spec(), %% The maximum age for members. max_age = ?DEFAULT_MAX_AGE :: time_spec(), cull_timer :: reference() | undefined, %% Optional TTL config and timer state. undefined = feature disabled (zero overhead). %% ExpiresAt per member is stored as the 4th element of all_members tuples. ttl = undefined :: undefined | #ttl{}, %% The supervisor used to start new members member_sup :: atom() | pid(), %% The supervisor used to start starter servers that start %% new members. This is what enables async member starts. starter_sup :: atom() | pid(), %% Maps member pid to a tuple of the form: %% {MonitorRef, Status, Time}, %% where MonitorRef is a monitor reference for the member,, %% Status is either 'free' or the consumer pid, and Time is %% an Erlang timestamp that records when the member became %% free. all_members = #{} :: members_map(), %% Maps consumer pid to a tuple of the form: %% {MonitorRef, MemberList} where MonitorRef is a monitor %% reference for the consumer and MemberList is a list of %% members being consumed. consumer_to_pid = #{} :: consumers_map(), %% A list of `{References, Timestamp}' tuples representing %% new member start requests that are in-flight. The %% timestamp records when the start request was initiated %% and is used to implement start timeout. starting_members = [] :: [{pid(), erlang:timestamp()}], stopping_count = 0 :: non_neg_integer(), %% The maximum amount of time to allow for member start. member_start_timeout = ?DEFAULT_MEMBER_START_TIMEOUT :: time_spec(), %% The optional threshold at which more members will be started if %% free_count drops to this value. Normally undefined, but may be %% set to a non-negative integer in order to enable "anticipatory" %% behavior (start members before they're actually needed). auto_grow_threshold = ?DEFAULT_AUTO_GROW_THRESHOLD :: undefined | non_neg_integer(), %% Stop callback to gracefully attempt to terminate pool members. %% The list of arguments must contain the fixed atom '$pooler_pid'. stop_mfa = pooler_starter:default_stop_mfa() :: pooler_starter:stop_mfa(), %% Optional callback invoked by pooler_starter after supervisor:start_child %% returns but before the member is accepted into the pool. Use this to %% perform slow initialization (e.g. a network handshake) outside the %% member supervisor, so concurrent starts do not serialize through it. %% Arguments may contain the placeholders '$pooler_pid' and '$pooler_pool_name'. %% Must return 'ok' on success or '{error, Reason}' on failure. initialize_mfa = undefined :: undefined | {atom(), atom(), [term()]}, %% The module to use for collecting metrics. If set to %% 'pooler_no_metrics', then metric sending calls do %% nothing. A typical value to actually capture metrics is %% folsom_metrics. metrics_mod = pooler_no_metrics :: atom(), %% The API used to call the metrics system. It supports both Folsom %% and Exometer format. metrics_api = folsom :: 'folsom' | 'exometer' | 'telemetry', %% A queue of requestors for blocking take member requests queued_requestors = queue:new() :: requestor_queue(), %% The max depth of the queue queue_max = 50 :: non_neg_integer() }). -type pool_name() :: atom(). %% The name of the pool -type group_name() :: atom(). %% The name of the group pool belongs to -type time_unit() :: hour | min | sec | ms | mu. -type time_spec() :: {non_neg_integer(), time_unit()} | non_neg_integer(). %% Time value: either a `{Amount, Unit}' tuple or a plain non-negative integer %% (interpreted as milliseconds, consistent with Erlang timeout conventions). -type pool_config() :: #{ name := pool_name(), init_count := non_neg_integer(), max_count := non_neg_integer(), start_mfa := {module(), atom(), [any()]}, group => group_name(), cull_interval => time_spec(), max_age => time_spec(), member_start_timeout => time_spec(), queue_max => non_neg_integer(), metrics_api => folsom | exometer | telemetry, metrics_mod => module(), stop_mfa => pooler_starter:stop_mfa(), initialize_mfa => {module(), atom(), ['$pooler_pid' | '$pooler_pool_name' | any(), ...]}, auto_grow_threshold => non_neg_integer(), add_member_retry => non_neg_integer(), max_lifetime => time_spec(), max_lifetime_jitter => time_spec() }. %% See {@link pooler:new_pool/1} -type pool_config_legacy() :: [{atom(), any()}]. %% Can be provided as a proplist, but is not recommended -type reconfigure_action() :: {start_workers, pos_integer()} | {stop_free_workers, pos_integer()} | {shrink_queue, pos_integer()} | {reset_cull_timer, time_spec()} | {cull, _} | {leave_group, group_name()} | {join_group, group_name()} | {set_parameter, {group, group_name() | undefined} | {init_count, non_neg_integer()} | {max_count, non_neg_integer()} | {cull_interval, time_spec()} | {max_age, time_spec()} | {member_start_timeout, time_spec()} | {queue_max, non_neg_integer()} | {metrics_api, folsom | exometer | telemetry} | {metrics_mod, module()} | {stop_mfa, pooler_starter:stop_mfa()} | {initialize_mfa, undefined | {module(), atom(), ['$pooler_pid' | '$pooler_pool_name' | any(), ...]}} | {auto_grow_threshold, non_neg_integer()}} | {update_ttl, undefined | #ttl{}}. -type member_expiry() :: integer() | infinity. %% erlang:monotonic_time(millisecond) deadline, or `infinity' when TTL is disabled. -type member_status() :: free | pid() | {stopping, replace | no_replace}. -type free_member_info() :: {reference(), free, erlang:timestamp(), member_expiry()}. -type member_info() :: {reference(), member_status(), erlang:timestamp(), member_expiry()}. %% See {@link pool_stats/1} -type members_map() :: #{pid() => member_info()}. -type consumers_map() :: #{pid() => {reference(), [pid()]}}. -if(?OTP_RELEASE >= 25). -type gen_server_from() :: gen_server:from(). -else. -type gen_server_from() :: {pid(), any()}. -endif. -type requestor_queue() :: queue:queue({gen_server_from(), reference()}). %% Internal % type specs for pool metrics -type metric_value() :: 'unknown_pid' | non_neg_integer() | {'add_pids_failed', non_neg_integer(), non_neg_integer()} | {'inc', 1} | 'error_no_members'. -type metric_type() :: 'counter' | 'gauge' | 'histogram' | 'history' | 'meter'. %% Metric descriptor types for pooler:metrics/0. %% label_values: 'any' = open-ended (e.g. pool_name), list = fixed known set. %% Order of label_spec list is significant — matches positional label order in prometheus.erl. -type label_values() :: any | [atom()]. -type label_spec() :: {Name :: atom(), label_values()}. -type metric_name() :: take | free | in_use | stopping | queue | error_no_members | killed_free | killed_in_use | queue_max_reached | starting_member_timeout. -type metric_info() :: #{ name := metric_name(), type := metric_type(), help := binary(), labels := [label_spec()] }. %% ------------------------------------------------------------------ %% Application API %% ------------------------------------------------------------------ -spec start() -> 'ok'. start() -> {ok, _} = application:ensure_all_started(pooler), ok. -spec stop() -> 'ok'. stop() -> ok = application:stop(pooler). %% ------------------------------------------------------------------ %% API Function Definitions %% ------------------------------------------------------------------ -spec start_link(pool_config()) -> {ok, pid()} | {error, any()}. start_link(#{name := Name, max_count := _, init_count := _, start_mfa := _} = PoolConfig) -> %% PoolConfig may include `metrics_mod' and `metrics_api' at this point gen_server:start_link({local, Name}, ?MODULE, PoolConfig, []). manual_start() -> application:start(sasl), application:start(pooler). %% @private create_group_table() -> ets:new(?POOLER_GROUP_TABLE, [set, public, named_table, {write_concurrency, true}]). %% @doc Start a new pool described by the map `PoolConfig'. The %% following keys are required in the map: %% %%
%%
`name'
%%
An atom giving the name of the pool.
%%
`init_count'
%%
Number of members to add to the pool at start. When the pool is %% started, `init_count' members will be started in parallel.
%%
`max_count'
%%
Maximum number of members in the pool.
%%
`start_mfa'
%%
A tuple of the form `{Mod, Fun, Args}' describing how to start %% new pool members.
%%
%% %% In addition, you can specify any of the following optional %% configuration options: %% %%
%%
`group'
%%
An atom giving the name of the group this pool belongs %% to. Pools sharing a common `group' value can be accessed using %% {@link take_group_member/1}, {@link return_group_member/2} and {@link group_pools/1}.
%%
`cull_interval'
%%
Default: `{1, min}'. Time between checks for stale pool members. Specified as %% a `{Time, Unit}' tuple where `Unit' is one of `hour', `min', `sec', `ms', `mu', %% or as a plain non-negative integer (milliseconds). %% Triggers a once per `cull_interval' check to remove members that have not %% been accessed in `max_age' time units. Culling can be disabled by %% specifying a zero time vaule (e.g. `{0, min}'). Culling will also be %% disabled if `init_count' is the same as `max_count'.
%%
`max_age'
%%
Default: `{30, sec}'. Members idle longer than `max_age' time units are removed from %% the pool when stale checking is enabled via %% `cull_interval'. Culling of idle members will never reduce the pool %% below `init_count'. The value is specified as `{Time, Unit}'. Note %% that timers are not set on individual pool members and may remain %% in the pool beyond the configured `max_age' value since members are %% only removed on the interval configured via `cull_interval'.
%%
`member_start_timeout'
%%
Default: `{1, min}'. Time limit for member starts. Specified as `{Time, Unit}'.
%%
`queue_max'
%%
Default: 50. When pool is empty and client is asking for a member with timeout %% (using {@link take_member/2}), this client will be put into a "waiting queue", served in a FIFO order. %% This queue lenght is bound by `queue_max'. When queue is full, any new queries will instantly get %% `error_no_members'
%%
`metrics_api', `metrics_mod'
%%
Pooler can export some internal metrics. It currently can export using API similar to `folsom' %% or API similar to `exometer'. Use `metrics_api' to specify API style and `metrics_mod' to specify %% the module implementing this API.
%%
`stop_mfa'
%%
By default when `pooler' needs to terminate one of its workers (when it is returned with `fail' status %% or `max_age' is reached), pooler calls %% `supervisor:terminate_child(pooler__member_sup, )'. If worker shutdown requires some %% more complex preparatons, a custom stop `{Module, Function, Arguments}' callback can be provided. %% `Arguments' can contain placeholders: `$pooler_pool_name' - name of the pool, `$pooler_pid' - pid of the worker to %% terminate. This callback have to terminate this process and remove it from pooler worker supervisor.
%%
`auto_grow_threshold'
%%
Default: `undefined' (disabled). Threshold at which more members (capped to `max_count') will be started %% if the number of free workers drops to this value - "anticipatory" behavior (start members before they're %% actually needed). Might be usefull when the worker initialization is relatively slow and we want to keep %% latency at minimum.
%%
`max_lifetime'
%%
Default: `undefined' (disabled). Maximum wall-clock lifetime of a pool member, specified as %% `{Time, Unit}'. When a member reaches its lifetime it is stopped and replaced with a fresh one. %% Eviction happens proactively via a timer while the member is idle, inline at the next %% {@link take_member/1} call if the timer fires late, or at {@link return_member/2} if the %% member expired while in use. Members held by a consumer past their TTL are never forcibly %% evicted. Useful when firewalls or database backends impose hard TCP connection lifetime limits. %% Has zero per-operation overhead when not set.
%%
`max_lifetime_jitter'
%%
Default: `{0, sec}' (no jitter). Random spread applied to each member's expiry time, %% specified as `{Time, Unit}'. Each member's lifetime is offset by a value drawn uniformly %% from `[-jitter, +jitter]', so members started together do not all expire simultaneously. %% Only meaningful when `max_lifetime' is set. Must be strictly less than `max_lifetime'; %% pool start and reconfigure will fail with `jitter_must_be_less_than_max_lifetime' otherwise.
%%
-spec new_pool(pool_config() | pool_config_legacy()) -> {ok, pid()} | {error, {already_started, pid()}}. new_pool(PoolConfig) -> pooler_sup:new_pool(config_as_map(PoolConfig)). %% @doc Terminate the named pool. -spec rm_pool(pool_name()) -> ok | {error, not_found | running | restarting}. rm_pool(PoolName) -> pooler_sup:rm_pool(PoolName). %% @doc Terminates the group and all pools in that group. %% %% If termination of any member pool fails, `rm_group/1' returns %% `{error, {failed_delete_pools, Pools}}', where `Pools' is a list %% of pools that failed to terminate. %% %% The group is NOT terminated if any member pool did not %% successfully terminate. %% -spec rm_group(group_name()) -> ok | {error, {failed_rm_pools, [atom()]}}. rm_group(GroupName) -> Pools = pg_get_local_members(GroupName), case rm_group_members(Pools) of [] -> pg_delete(GroupName); Failures -> {error, {failed_rm_pools, Failures}} end. -spec rm_group_members([pid()]) -> [atom()]. rm_group_members(MemberPids) -> lists:foldl( fun(MemberPid, Acc) -> #{name := PoolName} = gen_server:call(MemberPid, dump_pool), case pooler_sup:rm_pool(PoolName) of ok -> Acc; _ -> [PoolName | Acc] end end, [], MemberPids ). %% @doc Get child spec described by the `PoolConfig'. %% %% See {@link pooler:new_pool/1} for info about `PoolConfig'. -spec pool_child_spec(pool_config() | pool_config_legacy()) -> supervisor:child_spec(). pool_child_spec(PoolConfig) -> pooler_sup:pool_child_spec(config_as_map(PoolConfig)). %% @doc Updates the pool's state so it starts to behave like it was started with the new configuration without restart -spec pool_reconfigure(pool_name() | pid(), pool_config()) -> {ok, [reconfigure_action()]} | {error, any()}. pool_reconfigure(Pool, NewConfig) -> gen_server:call(Pool, {reconfigure, NewConfig}). %% @doc For INTERNAL use. Adds `MemberPid' to the pool. -spec accept_member(pool_name(), pooler_starter:start_result()) -> ok. accept_member(PoolName, StartResult) -> gen_server:call(PoolName, {accept_member, StartResult}). %% @doc Obtain exclusive access to a member from `PoolName'. %% %% If no free members are available, 'error_no_members' is returned. %% -spec take_member(pool_name() | pid()) -> pid() | error_no_members. take_member(PoolName) when is_atom(PoolName) orelse is_pid(PoolName) -> gen_server:call(PoolName, {take_member, 0}, infinity). %% @doc Obtain exclusive access to a member of 'PoolName'. %% %% If no members are available, wait for up to Timeout milliseconds for a member %% to become available. Waiting requests are served in FIFO order. If no member %% is available within the specified timeout, error_no_members is returned. %% `Timeout' can be either milliseconds as integer or `{duration, time_unit}' %% -spec take_member(pool_name() | pid(), non_neg_integer() | time_spec()) -> pid() | error_no_members. take_member(PoolName, Timeout) when is_atom(PoolName) orelse is_pid(PoolName) -> gen_server:call(PoolName, {take_member, time_as_millis(Timeout)}, infinity). %% @doc Take a member from a randomly selected member of the group %% `GroupName'. Returns `MemberPid' or `error_no_members'. If no %% members are available in the randomly chosen pool, all other pools %% in the group are tried in order. -spec take_group_member(group_name()) -> pid() | error_no_members. take_group_member(GroupName) -> case pg_get_local_members(GroupName) of [] -> error_no_members; Pools -> %% Put a random member at the front of the list and then %% return the first member you can walking the list. {_, _, X} = os:timestamp(), Idx = (X rem length(Pools)) + 1, {PoolPid, Rest} = extract_nth(Idx, Pools), take_first_pool([PoolPid | Rest]) end. take_first_pool([PoolPid | Rest]) -> case take_member(PoolPid) of error_no_members -> take_first_pool(Rest); Member -> ets:insert(?POOLER_GROUP_TABLE, {Member, PoolPid}), Member end; take_first_pool([]) -> error_no_members. %% this helper function returns `{Nth_Elt, Rest}' where `Nth_Elt' is %% the nth element of `L' and `Rest' is `L -- [Nth_Elt]'. extract_nth(N, L) -> extract_nth(N, L, []). extract_nth(1, [H | T], Acc) -> {H, Acc ++ T}; extract_nth(N, [H | T], Acc) -> extract_nth(N - 1, T, [H | Acc]); extract_nth(_, [], _) -> error(badarg). %% @doc Return a member that was taken from the group %% `GroupName'. This is a convenience function for %% `return_group_member/3' with `Status' of `ok'. -spec return_group_member(group_name(), pid() | error_no_members) -> ok. return_group_member(GroupName, MemberPid) -> return_group_member(GroupName, MemberPid, ok). %% @doc Return a member that was taken from the group `GroupName'. If %% `Status' is `ok' the member is returned to the pool from which is %% came. If `Status' is `fail' the member will be terminated and a new %% member added to the appropriate pool. -spec return_group_member(group_name(), pid() | error_no_members, ok | fail) -> ok. return_group_member(_, error_no_members, _) -> ok; return_group_member(_GroupName, MemberPid, Status) when is_pid(MemberPid) -> case ets:lookup(?POOLER_GROUP_TABLE, MemberPid) of [{MemberPid, PoolPid}] -> return_member(PoolPid, MemberPid, Status); [] -> ok end. %% @doc Return a member to the pool so it can be reused. %% %% If `Status' is 'ok', the member is returned to the pool. If %% `Status' is 'fail', the member is destroyed and a new member is %% added to the pool in its place. -spec return_member(pool_name() | pid(), pid() | error_no_members, ok | fail) -> ok. return_member(PoolName, Pid, Status) when is_pid(Pid) andalso (is_atom(PoolName) orelse is_pid(PoolName)) andalso (Status =:= ok orelse Status =:= fail) -> gen_server:call(PoolName, {return_member, Pid, Status}, infinity), ok; return_member(_, error_no_members, _) -> ok. %% @doc Return a member to the pool so it can be reused. %% -spec return_member(pool_name() | pid(), pid() | error_no_members) -> ok. return_member(PoolName, Pid) when is_pid(Pid) andalso (is_atom(PoolName) orelse is_pid(PoolName)) -> gen_server:call(PoolName, {return_member, Pid, ok}, infinity), ok; return_member(_, error_no_members) -> ok. %% @doc Obtain runtime state info for all workers. %% %% Format of the return value is subject to change. -spec pool_stats(pool_name() | pid()) -> [{pid(), member_info()}]. pool_stats(PoolName) -> gen_server:call(PoolName, pool_stats). %% @doc Obtain the pids of all pools which are members of the group. -spec group_pools(group_name()) -> [pid()]. group_pools(GroupName) -> pg_get_local_members(GroupName). %% @doc Obtain utilization info for a pool. %% %% Format of the return value is subject to change, but for now it %% will be a proplist to maintain backcompat with R16. -spec pool_utilization(pool_name() | pid()) -> [ {max_count, pos_integer()} | {in_use_count, non_neg_integer()} | {free_count, non_neg_integer()} | {starting_count, non_neg_integer()} | {stopping_count, non_neg_integer()} | {queued_count, non_neg_integer()} | {queue_max, non_neg_integer()} ]. pool_utilization(PoolName) -> gen_server:call(PoolName, pool_utilization). %% @doc Returns descriptors for all metrics emitted by pooler when `metrics_api = telemetry'. %% %% Each descriptor is a map with keys `name', `type', `help', and `labels'. %% `name' is the atom used in the telemetry event name `[pooler, Name]'. %% `labels' is a proplist `[{LabelName, LabelValues}]' where `LabelValues' is %% `any' for open-ended values or a list of the known atoms. %% The order of entries in `labels' matches the positional label order expected by prometheus.erl. %% %% Intended use: pre-declare prometheus metrics at application startup, e.g.: %% ``` %% [declare(M) || M <- pooler:metrics()] %% ''' %% %% Note: for gauge-type metrics (`free', `in_use', `stopping', `queue'), consider using %% `pool_utilization/1' with a prometheus_collector instead — it reads current values %% on each scrape rather than relying on push events from take/return calls. -spec metrics() -> [metric_info()]. metrics() -> [ #{ name => take, type => counter, labels => [{pool_name, any}], help => <<"Total take_member calls (compute rate at query time)">> }, #{ name => free, type => gauge, labels => [{pool_name, any}], help => <<"Number of free pool members">> }, #{ name => in_use, type => gauge, labels => [{pool_name, any}], help => <<"Number of in-use pool members">> }, #{ name => stopping, type => gauge, labels => [{pool_name, any}], help => <<"Number of pool members being stopped">> }, #{ name => error_no_members, type => counter, labels => [{pool_name, any}, {reason, [at_capacity, all_in_use]}], help => <<"Count of take_member calls returning error_no_members">> }, #{ name => killed_free, type => counter, labels => [{pool_name, any}, {reason, [max_lifetime, max_age, crashed, reconfigured]}], help => <<"Count of free pool members removed">> }, #{ name => killed_in_use, type => counter, labels => [{pool_name, any}, {reason, [max_lifetime, crashed, failed]}], help => <<"Count of in-use pool members removed">> }, #{ name => queue_max_reached, type => counter, labels => [{pool_name, any}], help => <<"Count of times the request queue reached its maximum depth">> }, #{ name => queue, type => gauge, labels => [{pool_name, any}], help => <<"Current depth of the request queue">> }, #{ name => starting_member_timeout, type => counter, labels => [{pool_name, any}], help => <<"Count of pool member start timeouts">> } ]. %% @doc Invokes `Fun' with arity 1 over all free members in pool with `PoolName'. %% -spec call_free_members(pool_name() | pid(), fun((pid()) -> term())) -> Res when Res :: [{ok, term()} | {error, term()}]. call_free_members(PoolName, Fun) when (is_atom(PoolName) orelse is_pid(PoolName)) andalso is_function(Fun, 1) -> call_free_members(PoolName, Fun, infinity). %% @doc Invokes `Fun' with arity 1 over all free members in pool with `PoolName'. %% `Timeout' sets the timeout of gen_server call. -spec call_free_members(pool_name() | pid(), Fun, timeout()) -> Res when Fun :: fun((pid()) -> term()), Res :: [{ok, term()} | {error, term()}]. call_free_members(PoolName, Fun, Timeout) when (is_atom(PoolName) orelse is_pid(PoolName)) andalso is_function(Fun, 1) -> gen_server:call(PoolName, {call_free_members, Fun}, Timeout). %% ------------------------------------------------------------------ %% gen_server Function Definitions %% ------------------------------------------------------------------ init(#{name := Name, max_count := MaxCount, init_count := InitCount, start_mfa := StartMFA} = P) -> TTL = case ttl_from_config(P) of {error, Err} -> exit({error, Err}); {ok, T} -> T end, Pool = #pool{ name = Name, group = maps:get(group, P, undefined), max_count = MaxCount, init_count = InitCount, start_mfa = StartMFA, add_member_retry = maps:get(add_member_retry, P, ?DEFAULT_ADD_RETRY), cull_interval = maps:get(cull_interval, P, ?DEFAULT_CULL_INTERVAL), max_age = maps:get(max_age, P, ?DEFAULT_MAX_AGE), member_start_timeout = maps:get(member_start_timeout, P, ?DEFAULT_MEMBER_START_TIMEOUT), auto_grow_threshold = maps:get(auto_grow_threshold, P, ?DEFAULT_AUTO_GROW_THRESHOLD), stop_mfa = maps:get(stop_mfa, P, pooler_starter:default_stop_mfa()), initialize_mfa = maps:get(initialize_mfa, P, undefined), metrics_mod = maps:get(metrics_mod, P, pooler_no_metrics), metrics_api = maps:get(metrics_api, P, folsom), queue_max = maps:get(queue_max, P, ?DEFAULT_POOLER_QUEUE_MAX), ttl = TTL }, MemberSup = pooler_pool_sup:build_member_sup_name(Name), Pool1 = set_member_sup(Pool, MemberSup), %% This schedules the next cull when the pool is configured for %% such and is otherwise a no-op. Pool2 = cull_members_from_pool(Pool1), {ok, NewPool} = init_members_sync(InitCount, Pool2), {ok, NewPool, {continue, join_group}}. handle_continue(join_group, #pool{group = undefined} = Pool) -> %% ignore {noreply, Pool}; handle_continue(join_group, #pool{group = Group} = Pool) -> ok = pg_create(Group), ok = pg_join(Group, self()), {noreply, Pool}. set_member_sup(#pool{} = Pool, MemberSup) -> Pool#pool{member_sup = MemberSup}. handle_call({take_member, Timeout}, From = {APid, _}, #pool{} = Pool) when is_pid(APid) -> maybe_reply(take_member_from_pool_queued(Pool, From, Timeout)); handle_call({return_member, Pid, Status}, {_CPid, _Tag}, Pool) -> {reply, ok, do_return_member(Pid, Status, Pool)}; handle_call({accept_member, StartResult}, _From, Pool) -> {reply, ok, do_accept_member(StartResult, Pool)}; handle_call(stop, _From, Pool) -> {stop, normal, stop_ok, Pool}; handle_call(pool_stats, _From, Pool) -> {reply, maps:to_list(Pool#pool.all_members), Pool}; handle_call(pool_utilization, _From, Pool) -> {reply, compute_utilization(Pool), Pool}; handle_call(dump_pool, _From, Pool) -> {reply, to_map(Pool), Pool}; handle_call({call_free_members, Fun}, _From, #pool{free_pids = Pids} = Pool) -> {reply, do_call_free_members(Fun, Pids), Pool}; handle_call({reconfigure, NewConfig}, _From, Pool) -> case calculate_reconfigure_actions(NewConfig, Pool) of {ok, Actions} = Res -> NewPool = lists:foldl(fun apply_reconfigure_action/2, Pool, Actions), {reply, Res, NewPool}; {error, _} = Res -> {reply, Res, Pool} end; handle_call(_Request, _From, Pool) -> {noreply, Pool}. -spec handle_cast(_, _) -> {'noreply', _}. handle_cast(_Msg, Pool) -> {noreply, Pool}. -spec handle_info(_, _) -> {'noreply', _}. handle_info({requestor_timeout, From}, Pool = #pool{queued_requestors = RequestorQueue}) -> NewQueue = queue:filter( fun ({RequestorFrom, _TRef}) when RequestorFrom =:= From -> gen_server:reply(RequestorFrom, error_no_members), false; ({_, _}) -> true end, RequestorQueue ), {noreply, Pool#pool{queued_requestors = NewQueue}}; handle_info({'DOWN', MRef, process, Pid, Reason}, State) -> State1 = case maps:get(Pid, State#pool.all_members, undefined) of {MRef, {stopping, Flag}, _Time, _ExpTs} -> %% Expected death: member was being stopped asynchronously. Pool1 = State#pool{ all_members = maps:remove(Pid, State#pool.all_members), stopping_count = State#pool.stopping_count - 1 }, send_metric(Pool1, stopping_count, Pool1#pool.stopping_count, gauge), case Flag of replace -> add_members_async(1, Pool1); no_replace -> Pool1 end; {MRef, _Status, _Time, _ExpTs} -> %% Unexpected death while free or in_use. Process is already %% dead — clean up directly without the async stop path. handle_unexpected_member_down(Pid, State); undefined -> case maps:get(Pid, State#pool.consumer_to_pid, undefined) of {MRef, Pids} -> IsOk = case Reason of normal -> ok; _Crash -> fail end, lists:foldl( fun(P, S) -> do_return_member(P, IsOk, S) end, State, Pids ); undefined -> State end end, {noreply, State1}; handle_info(cull_pool, Pool) -> {noreply, cull_members_from_pool(Pool)}; handle_info({ttl_expired, Pid}, #pool{ttl = TTL} = Pool) -> Pool1 = Pool#pool{ttl = TTL#ttl{timer = undefined, timer_target = undefined}}, Pool2 = case maps:get(Pid, Pool1#pool.all_members, undefined) of {_, free, _, _} -> remove_pid(Pid, Pool1, replace, max_lifetime); _ -> %% In-use, stopping, or already gone — return path handles it Pool1 end, {noreply, reschedule_ttl_timer(Pool2)}; handle_info(_Info, State) -> {noreply, State}. -spec terminate(_, _) -> 'ok'. terminate(_Reason, _State) -> ok. -spec code_change(_, _, _) -> {'ok', _}. %% pre-v2 tuple (24 elements, 1.5.2) → v2 (25 elements, same shape as 1.6.0) code_change(_OldVsn, OldState, Extra) when tuple_size(OldState) =:= 24 -> code_change(2, do_upgrade_pre_v2_to_v2(OldState), Extra); %% v2 tuple (25 elements, 1.6.0) → v3 code_change(2, OldState, Extra) when tuple_size(OldState) =:= 25 -> code_change(3, do_upgrade_to_v3(OldState), Extra); %% v3 tuple (27 elements) → v4 code_change(3, OldState, _Extra) when tuple_size(OldState) =:= 27 -> {ok, do_upgrade_to_v4(OldState)}; code_change(_, State, _Extra) -> {ok, State}. %% Converts the pre-v2 (1.5.2) 24-element pool tuple into the 25-element v2 shape %% (matching the 1.6.0 release): adds cull_timer=undefined and converts dict-based maps. do_upgrade_pre_v2_to_v2( {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, MaxAge, MemberSup, StarterSup, AllMembers, ConsumerToPid, StartingMembers, MemberStartTimeout, AutoGrowThreshold, StopMFA, MetricsMod, MetricsAPI, QueuedRequestors, QueueMax} ) -> {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, MaxAge, undefined, MemberSup, StarterSup, maps:from_list(dict:to_list(AllMembers)), maps:from_list(dict:to_list(ConsumerToPid)), StartingMembers, MemberStartTimeout, AutoGrowThreshold, StopMFA, MetricsMod, MetricsAPI, QueuedRequestors, QueueMax}. %% Converts a v2 25-element pool tuple to a v3 27-element pool tuple, %% adding initialize_mfa=undefined and stopping_count=0. do_upgrade_to_v3( {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, MaxAge, CullTimer, MemberSup, StarterSup, AllMembers, ConsumerToPid, StartingMembers, MemberStartTimeout, AutoGrowThreshold, StopMFA, MetricsMod, MetricsAPI, QueuedRequestors, QueueMax} ) -> {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, MaxAge, CullTimer, MemberSup, StarterSup, AllMembers, ConsumerToPid, StartingMembers, 0, MemberStartTimeout, AutoGrowThreshold, StopMFA, undefined, MetricsMod, MetricsAPI, QueuedRequestors, QueueMax}. %% Converts a v3 27-element pool tuple to a v4 #pool{} record: inserts ttl=undefined %% and extends all_members entries from 3-tuples to 4-tuples. do_upgrade_to_v4( {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, MaxAge, CullTimer, MemberSup, StarterSup, AllMembers, ConsumerToPid, StartingMembers, StoppingCount, MemberStartTimeout, AutoGrowThreshold, StopMFA, InitializeMFA, MetricsMod, MetricsAPI, QueuedRequestors, QueueMax} ) -> #pool{ name = Name, group = Group, max_count = MaxCount, init_count = InitCount, start_mfa = StartMFA, free_pids = FreePids, in_use_count = InUseCount, free_count = FreeCount, add_member_retry = AddMemberRetry, cull_interval = CullInterval, max_age = MaxAge, cull_timer = CullTimer, ttl = undefined, member_sup = MemberSup, starter_sup = StarterSup, all_members = maps:map( fun(_Pid, {MRef, Status, Ts}) -> {MRef, Status, Ts, infinity} end, AllMembers ), consumer_to_pid = ConsumerToPid, starting_members = StartingMembers, stopping_count = StoppingCount, member_start_timeout = MemberStartTimeout, auto_grow_threshold = AutoGrowThreshold, stop_mfa = StopMFA, initialize_mfa = InitializeMFA, metrics_mod = MetricsMod, metrics_api = MetricsAPI, queued_requestors = QueuedRequestors, queue_max = QueueMax }. %% ------------------------------------------------------------------ %% Internal Function Definitions %% ------------------------------------------------------------------ do_accept_member( {StarterPid, Pid}, #pool{ all_members = AllMembers, starting_members = StartingMembers0, member_start_timeout = StartTimeout } = Pool ) when is_pid(Pid) -> %% make sure we don't accept a timedout member Pool1 = #pool{starting_members = StartingMembers} = remove_stale_starting_members(Pool, StartingMembers0, StartTimeout), case lists:keytake(StarterPid, 1, StartingMembers) of false -> %% A starter completed even though we invalidated the pid %% Ask the starter to kill the child and stop. In most cases, the %% starter has already received this message. However, when pools %% are dynamically re-created with the same name, it is possible %% to receive an accept from a pool that has since gone away. %% In this case, we should cleanup. pooler_starter:stop_member_async(StarterPid), Pool1; {value, _, StartingMembers1} -> MRef = erlang:monitor(process, Pid), ExpTs = compute_expiry(Pool1#pool.ttl), Entry = {MRef, free, os:timestamp(), ExpTs}, AllMembers1 = store_all_members(Pid, Entry, AllMembers), pooler_starter:stop(StarterPid), Pool2 = Pool1#pool{ all_members = AllMembers1, starting_members = StartingMembers1 }, Pool3 = maybe_advance_ttl_timer(Pid, ExpTs, Pool2), maybe_reply_with_pid(Pid, Pool3) end; do_accept_member( {StarterPid, _Reason}, #pool{ starting_members = StartingMembers0, member_start_timeout = StartTimeout } = Pool ) -> %% member start failed, remove in-flight ref and carry on. pooler_starter:stop(StarterPid), Pool1 = #pool{starting_members = StartingMembers} = remove_stale_starting_members( Pool, StartingMembers0, StartTimeout ), StartingMembers1 = lists:keydelete(StarterPid, 1, StartingMembers), Pool1#pool{starting_members = StartingMembers1}. maybe_reply_with_pid( Pid, Pool = #pool{ queued_requestors = QueuedRequestors, free_pids = Free, free_count = NumFree } ) when is_pid(Pid) -> case queue:out(QueuedRequestors) of {empty, _} -> Pool#pool{ free_pids = [Pid | Free], free_count = NumFree + 1 }; {{value, {From = {APid, _}, TRef}}, NewQueuedRequestors} when is_pid(APid) -> reply_to_queued_requestor(TRef, Pid, From, NewQueuedRequestors, Pool) end. reply_to_queued_requestor(TRef, Pid, From = {APid, _}, NewQueuedRequestors, Pool) when is_pid(APid) -> erlang:cancel_timer(TRef), Pool1 = take_member_bookkeeping(Pid, From, NewQueuedRequestors, Pool), send_metric(Pool, in_use_count, Pool1#pool.in_use_count, gauge), send_metric(Pool, free_count, Pool1#pool.free_count, gauge), send_metric(Pool, events, error_no_members, history), gen_server:reply(From, Pid), Pool1. -spec take_member_bookkeeping( pid(), gen_server_from(), [pid()] | requestor_queue(), #pool{} ) -> #pool{}. take_member_bookkeeping( MemberPid, {CPid, _}, Rest, Pool = #pool{ in_use_count = NumInUse, free_count = NumFree, consumer_to_pid = CPMap, all_members = AllMembers } ) when is_pid(MemberPid), is_pid(CPid), is_list(Rest) -> Pool#pool{ free_pids = Rest, in_use_count = NumInUse + 1, free_count = NumFree - 1, consumer_to_pid = add_member_to_consumer(MemberPid, CPid, CPMap), all_members = set_cpid_for_member(MemberPid, CPid, AllMembers) }; take_member_bookkeeping( MemberPid, {ReplyPid, _Tag}, NewQueuedRequestors, Pool = #pool{ in_use_count = NumInUse, all_members = AllMembers, consumer_to_pid = CPMap } ) -> Pool#pool{ in_use_count = NumInUse + 1, all_members = set_cpid_for_member(MemberPid, ReplyPid, AllMembers), consumer_to_pid = add_member_to_consumer(MemberPid, ReplyPid, CPMap), queued_requestors = NewQueuedRequestors }. -spec remove_stale_starting_members( #pool{}, [{pid(), erlang:timestamp()}], time_spec() ) -> #pool{}. remove_stale_starting_members(Pool, StartingMembers, MaxAge) -> Now = os:timestamp(), MaxAgeSecs = time_as_secs(MaxAge), FilteredStartingMembers = lists:foldl( fun(SM, AccIn) -> accumulate_starting_member_not_stale(Pool, Now, SM, MaxAgeSecs, AccIn) end, [], StartingMembers ), Pool#pool{starting_members = FilteredStartingMembers}. accumulate_starting_member_not_stale(Pool, Now, SM = {Pid, StartTime}, MaxAgeSecs, AccIn) -> case secs_between(StartTime, Now) < MaxAgeSecs of true -> [SM | AccIn]; false -> ?LOG_ERROR( #{ label => "starting member timeout", pool => Pool#pool.name }, #{domain => [pooler]} ), send_metric(Pool, starting_member_timeout, {inc, 1}, counter), pooler_starter:stop_member_async(Pid), AccIn end. init_members_sync(N, #pool{name = PoolName, member_sup = MemberSup, initialize_mfa = InitMFA} = Pool) -> Self = self(), StartTime = os:timestamp(), StartRefs = [ {pooler_starter:start_member(PoolName, MemberSup, Self, InitMFA), StartTime} || _I <- lists:seq(1, N) ], Pool1 = Pool#pool{starting_members = StartRefs}, case collect_init_members(Pool1) of timeout -> ?LOG_ERROR( #{ label => "exceeded timeout waiting for members", pool => PoolName, init_count => Pool1#pool.init_count }, #{domain => [pooler]} ), error({timeout, "unable to start members"}); #pool{} = Pool2 -> {ok, Pool2} end. collect_init_members(#pool{starting_members = Empty} = Pool) when Empty =:= [] -> Pool; collect_init_members(#pool{member_start_timeout = StartTimeout} = Pool) -> Timeout = time_as_millis(StartTimeout), receive {accept_member, {_, _} = StartResult} -> collect_init_members(do_accept_member(StartResult, Pool)) after Timeout -> timeout end. -spec take_member_from_pool(#pool{}, gen_server_from()) -> {error_no_members | pid(), #pool{}}. take_member_from_pool( #pool{ init_count = InitCount, max_count = Max, member_start_timeout = StartTimeout } = Pool, From ) -> send_metric(Pool, take_rate, {inc, 1}, counter), Pool0 = evict_expired_heads(Pool), Pool1 = remove_stale_starting_members(Pool0, Pool0#pool.starting_members, StartTimeout), NonStaleStartingMemberCount = length(Pool1#pool.starting_members), NumInUse = Pool1#pool.in_use_count, NumFree = Pool1#pool.free_count, StoppingCount = Pool1#pool.stopping_count, NumCanAdd = Max - (NumInUse + NumFree + NonStaleStartingMemberCount + StoppingCount), case Pool1#pool.free_pids of [] when NumCanAdd =< 0 -> send_metric(Pool, error_no_members_count, {inc, 1}, counter, #{reason => at_capacity}), send_metric(Pool, events, error_no_members, history), {error_no_members, Pool1}; [] when NumCanAdd > 0 -> %% Limit concurrently starting members to init_count. Add %% up to init_count members. Starting members here means %% we always return an error_no_members for a take request %% when all members are in-use. By adding a batch of new %% members, the pool should reach a steady state with %% unused members culled over time (if scheduled cull is %% enabled). NumToAdd = max(min(InitCount - NonStaleStartingMemberCount, NumCanAdd), 1), Pool2 = add_members_async(NumToAdd, Pool1), send_metric(Pool, error_no_members_count, {inc, 1}, counter, #{reason => all_in_use}), send_metric(Pool, events, error_no_members, history), {error_no_members, Pool2}; [Pid | Rest] -> Pool2 = take_member_bookkeeping(Pid, From, Rest, Pool1), Pool3 = case Pool2#pool.auto_grow_threshold of N when is_integer(N) andalso Pool2#pool.free_count =< N andalso NumCanAdd > 0 -> NumToAdd = max(min(InitCount - NonStaleStartingMemberCount, NumCanAdd), 0), add_members_async(NumToAdd, Pool2); _ -> Pool2 end, send_metric(Pool, in_use_count, Pool3#pool.in_use_count, gauge), send_metric(Pool, free_count, Pool3#pool.free_count, gauge), {Pid, Pool3} end. -spec take_member_from_pool_queued( #pool{}, gen_server_from(), non_neg_integer() ) -> {error_no_members | queued | pid(), #pool{}}. take_member_from_pool_queued( Pool0 = #pool{ queue_max = QMax, queued_requestors = Requestors }, From = {CPid, _}, Timeout ) when is_pid(CPid) -> case {take_member_from_pool(Pool0, From), queue:len(Requestors)} of {{error_no_members, Pool1}, QLen} when QLen >= QMax -> send_metric(Pool1, events, error_no_members, history), send_metric(Pool1, queue_max_reached, {inc, 1}, counter), {error_no_members, Pool1}; {{error_no_members, Pool1}, _} when Timeout =:= 0 -> {error_no_members, Pool1}; {{error_no_members, Pool1 = #pool{queued_requestors = QueuedRequestors}}, QueueCount} -> TRef = erlang:send_after(Timeout, self(), {requestor_timeout, From}), send_metric(Pool1, queue_count, QueueCount, gauge), {queued, Pool1#pool{queued_requestors = queue:in({From, TRef}, QueuedRequestors)}}; {{Member, NewPool}, _} when is_pid(Member) -> {Member, NewPool} end. %% @doc Add `Count' members to `Pool' asynchronously. Returns updated %% `Pool' record with starting member refs added to field %% `starting_members'. add_members_async( Count, #pool{name = PoolName, member_sup = MemberSup, starting_members = StartingMembers, initialize_mfa = InitMFA} = Pool ) -> StartTime = os:timestamp(), StartRefs = [ {pooler_starter:start_member(PoolName, MemberSup, InitMFA), StartTime} || _I <- lists:seq(1, Count) ], Pool#pool{starting_members = StartRefs ++ StartingMembers}. -spec do_return_member(pid(), ok | fail, #pool{}) -> #pool{}. do_return_member( Pid, ok, #pool{ name = PoolName, all_members = AllMembers, queued_requestors = QueuedRequestors } = Pool ) -> clean_group_table(Pid, Pool), case maps:get(Pid, AllMembers, undefined) of {_, free, _, _} -> ?LOG_WARNING( #{ label => "ignored return of free member", pool => PoolName, pid => Pid }, #{domain => [pooler]} ), Pool; {_, {stopping, _}, _, _} -> %% member is being stopped asynchronously — ignore this return Pool; {MRef, CPid, _FreeTs, ExpTs} -> #pool{ free_pids = Free, in_use_count = NumInUse, free_count = NumFree } = Pool, Pool1 = Pool#pool{ in_use_count = NumInUse - 1, consumer_to_pid = cpmap_remove(Pid, CPid, Pool#pool.consumer_to_pid) }, case is_member_expired(Pid, Pool1) of true -> %% Member expired while in use — evict inline (not in free_pids, %% so no lists:delete needed → true O(1)) Pool2 = Pool1#pool{ stopping_count = Pool1#pool.stopping_count + 1, all_members = AllMembers#{ Pid => {MRef, {stopping, replace}, os:timestamp(), ExpTs} } }, pooler_starter_sup:new_stopper( pooler_starter:stop_spec(PoolName, Pid, Pool2#pool.stop_mfa) ), send_metric(Pool2, killed_in_use_count, {inc, 1}, counter, #{reason => max_lifetime}), send_metric(Pool2, stopping_count, Pool2#pool.stopping_count, gauge), Pool2; false -> Entry = {MRef, free, os:timestamp(), ExpTs}, Pool2 = Pool1#pool{ all_members = store_all_members(Pid, Entry, AllMembers) }, case queue:out(QueuedRequestors) of {empty, _} -> Pool3 = Pool2#pool{free_pids = [Pid | Free], free_count = NumFree + 1}, maybe_advance_ttl_timer(Pid, ExpTs, Pool3); {{value, {From = {APid, _}, TRef}}, NewQueuedRequestors} when is_pid(APid) -> reply_to_queued_requestor(TRef, Pid, From, NewQueuedRequestors, Pool2) end end; undefined -> Pool end; do_return_member(Pid, fail, #pool{all_members = AllMembers} = Pool) -> % for the fail case, perhaps the member crashed and was already % removed, so use find instead of fetch and ignore missing. clean_group_table(Pid, Pool), case maps:get(Pid, AllMembers, undefined) of {_MRef, {stopping, _}, _, _} -> %% already being stopped asynchronously — ignore Pool; {_MRef, _, _, _} -> %% replacement is triggered when the member's DOWN arrives remove_pid(Pid, Pool, replace, failed); undefined -> Pool end. clean_group_table(_MemberPid, #pool{group = undefined}) -> ok; clean_group_table(MemberPid, #pool{group = _GroupName}) -> ets:delete(?POOLER_GROUP_TABLE, MemberPid). % @doc Remove `Pid' from the pid list associated with `CPid' in the % consumer to member map given by `CPMap'. % % If `Pid' is the last element in `CPid's pid list, then the `CPid' % entry is removed entirely. % -spec cpmap_remove(pid(), pid() | free, consumers_map()) -> consumers_map(). cpmap_remove(_Pid, free, CPMap) -> CPMap; cpmap_remove(Pid, CPid, CPMap) -> case maps:get(CPid, CPMap, undefined) of {MRef, Pids0} -> Pids1 = lists:delete(Pid, Pids0), case Pids1 of [_H | _T] -> CPMap#{CPid => {MRef, Pids1}}; [] -> %% no more members for this consumer erlang:demonitor(MRef, [flush]), maps:remove(CPid, CPMap) end; undefined -> % FIXME: this shouldn't happen, should we log or error? CPMap end. % @doc Handle a member process that died unexpectedly (not via async stop). % % The process is already dead so we clean up pool state directly and % start a replacement, bypassing the async stop machinery. handle_unexpected_member_down(Pid, Pool) -> clean_group_table(Pid, Pool), AllMembers = Pool#pool.all_members, case maps:get(Pid, AllMembers, undefined) of {_, free, _, _} -> Pool1 = Pool#pool{ free_pids = lists:delete(Pid, Pool#pool.free_pids), free_count = Pool#pool.free_count - 1, all_members = maps:remove(Pid, AllMembers) }, send_metric(Pool1, killed_free_count, {inc, 1}, counter, #{reason => crashed}), Pool2 = case Pool1#pool.ttl of #ttl{timer_target = Pid} = TTL -> reschedule_ttl_timer(Pool1#pool{ttl = TTL#ttl{timer_target = undefined}}); _ -> Pool1 end, add_members_async(1, Pool2); {_, CPid, _, _} -> Pool1 = Pool#pool{ in_use_count = Pool#pool.in_use_count - 1, all_members = maps:remove(Pid, AllMembers), consumer_to_pid = cpmap_remove(Pid, CPid, Pool#pool.consumer_to_pid) }, send_metric(Pool1, killed_in_use_count, {inc, 1}, counter, #{reason => crashed}), add_members_async(1, Pool1); undefined -> Pool end. % @doc Initiate async removal of a pool member. % % Tags the member as stopping (keeping it in all_members so it counts % against max_count), spawns a supervised stopper, and returns % immediately. The pool learns the member has actually died via its % monitor DOWN message. % -spec remove_pid(pid(), #pool{}, replace | no_replace, atom()) -> #pool{}. remove_pid(Pid, Pool, Flag, Reason) -> #pool{ name = PoolName, all_members = AllMembers, consumer_to_pid = CPMap, stop_mfa = StopMFA } = Pool, case maps:get(Pid, AllMembers, undefined) of {MRef, free, Time, ExpTs} -> Pool1 = Pool#pool{ free_pids = lists:delete(Pid, Pool#pool.free_pids), free_count = Pool#pool.free_count - 1, stopping_count = Pool#pool.stopping_count + 1, all_members = AllMembers#{Pid => {MRef, {stopping, Flag}, Time, ExpTs}} }, pooler_starter_sup:new_stopper(pooler_starter:stop_spec(PoolName, Pid, StopMFA)), send_metric(Pool1, killed_free_count, {inc, 1}, counter, #{reason => Reason}), send_metric(Pool1, stopping_count, Pool1#pool.stopping_count, gauge), case Pool1#pool.ttl of #ttl{timer_target = Pid} = TTL -> reschedule_ttl_timer(Pool1#pool{ttl = TTL#ttl{timer_target = undefined}}); _ -> Pool1 end; {MRef, CPid, Time, ExpTs} -> Pool1 = Pool#pool{ in_use_count = Pool#pool.in_use_count - 1, stopping_count = Pool#pool.stopping_count + 1, all_members = AllMembers#{Pid => {MRef, {stopping, Flag}, Time, ExpTs}}, consumer_to_pid = cpmap_remove(Pid, CPid, CPMap) }, pooler_starter_sup:new_stopper(pooler_starter:stop_spec(PoolName, Pid, StopMFA)), send_metric(Pool1, killed_in_use_count, {inc, 1}, counter, #{reason => Reason}), send_metric(Pool1, stopping_count, Pool1#pool.stopping_count, gauge), Pool1; undefined -> ?LOG_ERROR( #{ label => unknown_pid, pool => PoolName, pid => Pid }, #{domain => [pooler]} ), send_metric(Pool, events, unknown_pid, history), Pool end. -spec store_all_members( pid(), member_info(), members_map() ) -> members_map(). store_all_members(Pid, Val = {_MRef, _CPid, _Time, _ExpTs}, AllMembers) -> AllMembers#{Pid => Val}. -spec set_cpid_for_member(pid(), pid(), members_map()) -> members_map(). set_cpid_for_member(MemberPid, CPid, AllMembers) -> maps:update_with( MemberPid, fun({MRef, free, Time = {_, _, _}, ExpTs}) -> {MRef, CPid, Time, ExpTs} end, AllMembers ). -spec add_member_to_consumer(pid(), pid(), consumers_map()) -> consumers_map(). add_member_to_consumer(MemberPid, CPid, CPMap) -> %% we can't use maps:update_with here because we need to create the %% monitor if we aren't already tracking this consumer. case maps:get(CPid, CPMap, undefined) of {MRef, MList} -> CPMap#{CPid => {MRef, [MemberPid | MList]}}; undefined -> MRef = erlang:monitor(process, CPid), CPMap#{CPid => {MRef, [MemberPid]}} end. -spec cull_members_from_pool(#pool{}) -> #pool{}. cull_members_from_pool(#pool{cull_interval = {0, _}} = Pool) -> %% 0 cull_interval means do not cull Pool; cull_members_from_pool(#pool{init_count = C, max_count = C} = Pool) -> %% if init_count matches max_count, then we will not dynamically %% add capacity and should not schedule culling regardless of %% cull_interval config. Pool; cull_members_from_pool( #pool{ free_count = FreeCount, init_count = InitCount, in_use_count = InUseCount, cull_interval = Delay, cull_timer = CullTRef, max_age = MaxAge, all_members = AllMembers } = Pool ) -> case is_reference(CullTRef) of true -> erlang:cancel_timer(CullTRef); false -> noop end, MaxCull = FreeCount - (InitCount - InUseCount), Pool1 = case MaxCull > 0 of true -> MemberInfo = member_info(Pool#pool.free_pids, AllMembers), ExpiredMembers = expired_free_members(MemberInfo, os:timestamp(), MaxAge), CullList = lists:sublist(ExpiredMembers, MaxCull), lists:foldl( fun({CullMe, _}, S) -> remove_pid(CullMe, S, no_replace, max_age) end, Pool, CullList ); false -> Pool end, Pool1#pool{cull_timer = schedule_cull(self(), Delay)}. -spec schedule_cull( Pool :: atom() | pid(), Delay :: time_spec() ) -> reference(). %% @doc Schedule a pool cleaning or "cull" for `PoolName' in which %% members older than `max_age' will be removed until the pool has %% `init_count' members. Uses `erlang:send_after/3' for light-weight %% timer that will be auto-cancelled upon pooler shutdown. schedule_cull(Pool, Delay) -> DelayMillis = time_as_millis(Delay), erlang:send_after(DelayMillis, Pool, cull_pool). -spec member_info([pid()], members_map()) -> [{pid(), member_info()}]. member_info(Pids, AllMembers) -> maps:to_list(maps:with(Pids, AllMembers)). -spec expired_free_members( Members :: [{pid(), member_info()}], Now :: {_, _, _}, MaxAge :: time_spec() ) -> [{pid(), free_member_info()}]. expired_free_members(Members, Now, MaxAge) -> MaxMicros = time_as_micros(MaxAge), [ MI || MI = {_, {_, free, LastReturn, _}} <- Members, timer:now_diff(Now, LastReturn) >= MaxMicros ]. -spec calculate_reconfigure_actions(pool_config(), #pool{}) -> {ok, [reconfigure_action()]} | {error, any()}. calculate_reconfigure_actions( #{name := Name, start_mfa := MFA} = NewConfig, #pool{name = PName, start_mfa = PMFA} = Pool ) when Name =:= PName, MFA =:= PMFA -> Defaults = #{ group => undefined, cull_interval => ?DEFAULT_CULL_INTERVAL, max_age => ?DEFAULT_MAX_AGE, member_start_timeout => ?DEFAULT_MEMBER_START_TIMEOUT, auto_grow_threshold => ?DEFAULT_AUTO_GROW_THRESHOLD, stop_mfa => pooler_starter:default_stop_mfa(), initialize_mfa => undefined, metrics_mod => pooler_no_metrics, metrics_api => folsom, queue_max => ?DEFAULT_POOLER_QUEUE_MAX }, NewWithDefaults0 = maps:merge(Defaults, NewConfig), try NewWithDefaults = case ttl_from_config(NewConfig) of {error, _} = Err -> throw(Err); {ok, T} -> NewWithDefaults0#{ttl => T} end, lists:flatmap( fun(Param) -> mk_rec_action(Param, maps:get(Param, NewWithDefaults), NewConfig, Pool) end, [ group, init_count, max_count, cull_interval, max_age, member_start_timeout, queue_max, metrics_api, metrics_mod, stop_mfa, initialize_mfa, auto_grow_threshold, ttl ] ) of Actions -> {ok, Actions} catch throw:{error, _} = E -> E end; calculate_reconfigure_actions(_, _) -> {error, changed_unsupported_parameter}. mk_rec_action(group, New, _, #pool{group = Old}) when New =/= Old -> [{set_parameter, {group, New}}] ++ case Old of undefined -> []; _ -> [{leave_group, Old}] end ++ case New of undefined -> []; _ -> [{join_group, New}] end; mk_rec_action(init_count, NewInitCount, _, #pool{init_count = OldInitCount, in_use_count = InUse, free_count = Free}) when NewInitCount > OldInitCount -> AliveCount = InUse + Free, [ {set_parameter, {init_count, NewInitCount}} | case AliveCount < NewInitCount of true -> [{start_workers, NewInitCount - AliveCount}]; false -> [] end ]; mk_rec_action(init_count, NewInitCount, _, #pool{init_count = OldInitCount}) when NewInitCount < OldInitCount -> [{set_parameter, {init_count, NewInitCount}}]; mk_rec_action(max_count, NewMaxCount, _, #pool{max_count = OldMaxCount, in_use_count = InUse, free_count = Free}) when NewMaxCount < OldMaxCount -> AliveCount = InUse + Free, [ {set_parameter, {max_count, NewMaxCount}} | case AliveCount > NewMaxCount of true when Free >= (AliveCount - NewMaxCount) -> %% We have enough free workers to shut down [{stop_free_workers, AliveCount - NewMaxCount}]; true -> %% We don't have enough free workers to shutdown throw({error, {max_count, not_enough_free_workers_to_shutdown}}); false -> [] end ]; mk_rec_action(max_count, NewMaxCount, _, #pool{max_count = OldMaxCount}) when NewMaxCount > OldMaxCount -> [{set_parameter, {max_count, NewMaxCount}}]; mk_rec_action(cull_interval, New, _, #pool{cull_interval = Old, cull_timer = _Timer}) when New =/= Old -> [ {set_parameter, {cull_interval, New}} | case time_as_millis(New) < time_as_millis(Old) of true -> [{reset_cull_timer, New}]; false -> [] end ]; mk_rec_action(max_age, New, _, #pool{max_age = Old}) when New =/= Old -> [ {set_parameter, {max_age, New}} | case time_as_millis(New) < time_as_millis(Old) of true -> [{cull, []}]; false -> [] end ]; mk_rec_action(member_start_timeout = P, New, _, #pool{member_start_timeout = Old}) when New =/= Old -> [{set_parameter, {P, New}}]; mk_rec_action(queue_max = P, New, _, #pool{queue_max = Old, queued_requestors = Queue}) when New < Old -> QLen = queue:len(Queue), [ {set_parameter, {P, New}} | case QLen > New of true -> [{shrink_queue, QLen - New}]; false -> [] end ]; mk_rec_action(queue_max = P, New, _, #pool{queue_max = Old}) when New > Old -> [{set_parameter, {P, New}}]; mk_rec_action(metrics_api = P, New, _, #pool{metrics_api = Old}) when New =/= Old -> [{set_parameter, {P, New}}]; mk_rec_action(metrics_mod = P, New, _, #pool{metrics_mod = Old}) when New =/= Old -> [{set_parameter, {P, New}}]; mk_rec_action(stop_mfa = P, New, _, #pool{stop_mfa = Old}) when New =/= Old -> [{set_parameter, {P, New}}]; mk_rec_action(initialize_mfa = P, New, _, #pool{initialize_mfa = Old}) when New =/= Old -> [{set_parameter, {P, New}}]; mk_rec_action(auto_grow_threshold = P, New, _, #pool{auto_grow_threshold = Old}) when New =/= Old -> [{set_parameter, {P, New}}]; mk_rec_action(ttl, NewTTL, _Config, #pool{ttl = OldTTL}) -> case {ttl_static(OldTTL), ttl_static(NewTTL)} of {Same, Same} -> []; _ -> [{update_ttl, NewTTL}] end; mk_rec_action(_Param, _NewVal, _, _Pool) -> %% not changed []. -spec apply_reconfigure_action(reconfigure_action(), #pool{}) -> #pool{}. apply_reconfigure_action({set_parameter, {Name, Value}}, Pool) -> set_parameter(Name, Value, Pool); apply_reconfigure_action({start_workers, Count}, Pool) -> add_members_async(Count, Pool); apply_reconfigure_action({stop_free_workers, Count}, #pool{free_pids = Free} = Pool) -> lists:foldl(fun(P, S) -> remove_pid(P, S, no_replace, reconfigured) end, Pool, lists:sublist(Free, Count)); apply_reconfigure_action({shrink_queue, Count}, #pool{queued_requestors = Q} = Pool) -> {ToShrink, ToKeep} = lists:split(Count, queue:to_list(Q)), [gen_server:reply(From, error_no_members) || {From, _TRef} <- ToShrink], Pool#pool{queued_requestors = queue:from_list(ToKeep)}; apply_reconfigure_action({reset_cull_timer, Interval}, #pool{cull_timer = TRef} = Pool) -> case is_reference(TRef) of true -> erlang:cancel_timer(TRef); false -> noop end, Pool#pool{cull_timer = schedule_cull(self(), Interval)}; apply_reconfigure_action({cull, _}, Pool) -> cull_members_from_pool(Pool); apply_reconfigure_action({join_group, Group}, Pool) -> ok = pg_create(Group), ok = pg_join(Group, self()), Pool; apply_reconfigure_action({leave_group, Group}, Pool) -> ok = pg_leave(Group, self()), Pool; apply_reconfigure_action({update_ttl, NewTTL}, Pool) -> case Pool#pool.ttl of #ttl{timer = TRef} when is_reference(TRef) -> erlang:cancel_timer(TRef); _ -> ok end, NewTTLConfig = case NewTTL of undefined -> undefined; #ttl{} -> NewTTL#ttl{timer = undefined, timer_target = undefined} end, NewAllMembers = recompute_member_expiries(Pool#pool.all_members, Pool#pool.ttl, NewTTLConfig), Pool1 = Pool#pool{ttl = NewTTLConfig, all_members = NewAllMembers}, reschedule_ttl_timer(Pool1). set_parameter(group, Value, Pool) -> Pool#pool{group = Value}; set_parameter(init_count, Value, Pool) -> Pool#pool{init_count = Value}; set_parameter(max_count, Value, Pool) -> Pool#pool{max_count = Value}; set_parameter(cull_interval, Value, Pool) -> Pool#pool{cull_interval = Value}; set_parameter(max_age, Value, Pool) -> Pool#pool{max_age = Value}; set_parameter(member_start_timeout, Value, Pool) -> Pool#pool{member_start_timeout = Value}; set_parameter(queue_max, Value, Pool) -> Pool#pool{queue_max = Value}; set_parameter(metrics_api, Value, Pool) -> Pool#pool{metrics_api = Value}; set_parameter(metrics_mod, Value, Pool) -> Pool#pool{metrics_mod = Value}; set_parameter(stop_mfa, Value, Pool) -> Pool#pool{stop_mfa = Value}; set_parameter(initialize_mfa, Value, Pool) -> Pool#pool{initialize_mfa = Value}; set_parameter(auto_grow_threshold, Value, Pool) -> Pool#pool{auto_grow_threshold = Value}. %% Send a metric using the metrics module from application config or %% do nothing. -spec send_metric( Pool :: #pool{}, Label :: atom(), Value :: metric_value(), Type :: metric_type() ) -> ok. send_metric(Pool, Label, Value, Type) -> send_metric(Pool, Label, Value, Type, #{}). -spec send_metric( Pool :: #pool{}, Label :: atom(), Value :: metric_value(), Type :: metric_type(), ExtraTags :: #{atom() => term()} ) -> ok. send_metric(#pool{metrics_mod = pooler_no_metrics}, _Label, _Value, _Type, _ExtraTags) -> ok; %% Folsom/exometer backward compat: gauge was historically histogram for these APIs. send_metric(#pool{metrics_api = A} = Pool, Label, Value, gauge, ExtraTags) when A =:= folsom; A =:= exometer -> send_metric(Pool, Label, Value, histogram, ExtraTags); %% Telemetry and exometer do not support history type metrics. send_metric(#pool{metrics_api = A}, _Label, _Value, history, _ExtraTags) when A =:= telemetry; A =:= exometer -> ok; %% Exometer counter: unwrap {inc, N} to a plain value. send_metric( #pool{name = PoolName, metrics_mod = MetricsMod, metrics_api = exometer}, Label, {inc, Value}, counter, _ExtraTags ) -> MetricName = pool_metric_exometer(PoolName, Label), MetricsMod:update_or_create(MetricName, Value, counter, []), ok; send_metric( #pool{name = PoolName, metrics_mod = MetricsMod, metrics_api = exometer}, Label, Value, Type, _ExtraTags ) -> MetricName = pool_metric_exometer(PoolName, Label), MetricsMod:update_or_create(MetricName, Value, Type, []), ok; %% Folsom API is the default one. send_metric( #pool{name = PoolName, metrics_mod = MetricsMod, metrics_api = folsom}, Label, Value, Type, _ExtraTags ) -> MetricName = pool_metric(PoolName, Label), MetricsMod:notify(MetricName, Value, Type), ok; %% Telemetry API. send_metric( #pool{name = PoolName, metrics_mod = MetricsMod, metrics_api = telemetry}, Label, Value, Type, ExtraTags ) -> {Measurements, EventMeta} = telemetry_measurements(Value, Type), Metadata = maps:merge(ExtraTags, EventMeta#{pool_name => PoolName, type => Type}), MetricsMod:execute([pooler, telemetry_label(Label)], Measurements, Metadata), ok. -spec pool_metric(atom(), atom()) -> binary(). pool_metric(PoolName, Metric) -> iolist_to_binary([ <<"pooler.">>, atom_to_binary(PoolName, utf8), ".", atom_to_binary(Metric, utf8) ]). %% Exometer metric names are lists, not binaries. -spec pool_metric_exometer(atom(), atom()) -> nonempty_list(binary()). pool_metric_exometer(PoolName, Metric) -> [ <<"pooler">>, atom_to_binary(PoolName, utf8), atom_to_binary(Metric, utf8) ]. -spec telemetry_label(atom()) -> atom(). telemetry_label(take_rate) -> take; telemetry_label(free_count) -> free; telemetry_label(in_use_count) -> in_use; telemetry_label(stopping_count) -> stopping; telemetry_label(error_no_members_count) -> error_no_members; telemetry_label(killed_free_count) -> killed_free; telemetry_label(killed_in_use_count) -> killed_in_use; telemetry_label(queue_count) -> queue; telemetry_label(Label) -> Label. -spec telemetry_measurements(metric_value(), metric_type()) -> {#{atom() => term()}, #{atom() => term()}}. telemetry_measurements({inc, N}, counter) -> {#{count => N}, #{}}; telemetry_measurements(N, _) when is_integer(N) -> {#{value => N}, #{}}; telemetry_measurements(Reason, history) -> {#{count => 1}, #{reason => Reason}}. -spec time_as_secs(time_spec()) -> non_neg_integer(). time_as_secs({Time, Unit}) -> time_as_micros({Time, Unit}) div 1000000. -spec time_as_millis(time_spec()) -> non_neg_integer(). %% @doc Convert time unit into milliseconds. time_as_millis({Time, Unit}) -> time_as_micros({Time, Unit}) div 1000; %% Allows blind convert time_as_millis(Time) when is_integer(Time) -> Time. -spec time_as_micros(time_spec()) -> non_neg_integer(). %% @doc Convert time unit into microseconds time_as_micros({Time, hour}) -> 60 * 60 * 1000 * 1000 * Time; time_as_micros({Time, min}) -> 60 * 1000 * 1000 * Time; time_as_micros({Time, sec}) -> 1000 * 1000 * Time; time_as_micros({Time, ms}) -> 1000 * Time; time_as_micros({Time, mu}) -> Time; time_as_micros(Time) when is_integer(Time) -> Time * 1000. secs_between({Mega1, Secs1, _}, {Mega2, Secs2, _}) -> (Mega2 - Mega1) * 1000000 + (Secs2 - Secs1). %% ------------------------------------------------------------------ %% TTL helpers — all short-circuit on #pool{ttl = undefined} %% ------------------------------------------------------------------ %% @doc Compute an absolute expiry timestamp (erlang:monotonic_time(millisecond)) for a -spec ttl_from_config(pool_config()) -> {ok, undefined | #ttl{}} | {error, jitter_must_be_less_than_max_lifetime}. ttl_from_config(#{max_lifetime := ML} = Config) -> Jitter = maps:get(max_lifetime_jitter, Config, {0, sec}), case time_as_millis(Jitter) >= time_as_millis(ML) of true -> {error, jitter_must_be_less_than_max_lifetime}; false -> {ok, #ttl{max_lifetime = ML, jitter = Jitter}} end; ttl_from_config(#{} = _Config) -> {ok, undefined}. %% newly accepted member. Returns `infinity' when TTL is disabled. -spec compute_expiry(undefined | #ttl{}) -> member_expiry(). compute_expiry(undefined) -> infinity; compute_expiry(#ttl{max_lifetime = MaxLifetime, jitter = Jitter}) -> LifetimeMs = time_as_millis(MaxLifetime), JitterMs = time_as_millis(Jitter), RandJitter = case JitterMs of 0 -> 0; _ -> rand:uniform(2 * JitterMs + 1) - JitterMs - 1 end, erlang:monotonic_time(millisecond) + max(LifetimeMs + RandJitter, 1). %% @doc Returns true if Pid's assigned expiry has passed. O(1). -spec is_member_expired(pid(), #pool{}) -> boolean(). is_member_expired(_Pid, #pool{ttl = undefined}) -> false; is_member_expired(Pid, #pool{all_members = AllMembers}) -> case maps:get(Pid, AllMembers, undefined) of {_, _, _, infinity} -> false; {_, _, _, ExpTs} -> erlang:monotonic_time(millisecond) >= ExpTs; undefined -> false end. %% @doc Evict an expired member that was just dequeued from the head of free_pids. %% Unlike remove_pid/3, does NOT call lists:delete (we already have Rest). -spec remove_free_head(pid(), [pid()], #pool{}) -> #pool{}. remove_free_head( Pid, Rest, #pool{ all_members = AllMembers, ttl = TTL, name = PoolName, stop_mfa = StopMFA } = Pool ) -> {MRef, free, Time, ExpTs} = maps:get(Pid, AllMembers), Pool1 = Pool#pool{ free_pids = Rest, free_count = Pool#pool.free_count - 1, stopping_count = Pool#pool.stopping_count + 1, all_members = AllMembers#{Pid => {MRef, {stopping, replace}, Time, ExpTs}} }, pooler_starter_sup:new_stopper(pooler_starter:stop_spec(PoolName, Pid, StopMFA)), send_metric(Pool1, killed_free_count, {inc, 1}, counter, #{reason => max_lifetime}), send_metric(Pool1, stopping_count, Pool1#pool.stopping_count, gauge), case TTL#ttl.timer_target of Pid -> case TTL#ttl.timer of TRef when is_reference(TRef) -> erlang:cancel_timer(TRef); undefined -> ok end, Pool1#pool{ttl = TTL#ttl{timer = undefined, timer_target = undefined}}; _ -> Pool1 end. %% @doc Check the head of free_pids and evict all consecutive expired members. %% Calls reschedule_ttl_timer at most once regardless of how many heads are evicted. -spec evict_expired_heads(#pool{}) -> #pool{}. evict_expired_heads(#pool{ttl = undefined} = Pool) -> Pool; evict_expired_heads(#pool{ttl = #ttl{timer_target = BeforeTarget}} = Pool) -> Pool1 = evict_expired_heads_loop(Pool), case Pool1#pool.ttl of #ttl{timer_target = undefined} when BeforeTarget =/= undefined -> reschedule_ttl_timer(Pool1); _ -> Pool1 end. -spec evict_expired_heads_loop(#pool{}) -> #pool{}. evict_expired_heads_loop(#pool{free_pids = []} = Pool) -> Pool; evict_expired_heads_loop(#pool{free_pids = [Pid | Rest]} = Pool) -> case is_member_expired(Pid, Pool) of false -> Pool; true -> evict_expired_heads_loop(remove_free_head(Pid, Rest, Pool)) end. %% @doc Schedule or advance the TTL timer when a member joins free_pids. %% Reschedules only if the new member expires sooner than the current target — O(1). -spec maybe_advance_ttl_timer(pid(), member_expiry(), #pool{}) -> #pool{}. maybe_advance_ttl_timer(_Pid, _ExpTs, #pool{ttl = undefined} = Pool) -> Pool; maybe_advance_ttl_timer(_Pid, infinity, Pool) -> Pool; maybe_advance_ttl_timer(Pid, ExpTs, #pool{ttl = #ttl{timer_target = undefined} = TTL} = Pool) -> schedule_ttl_timer_for(Pid, ExpTs, TTL, Pool); maybe_advance_ttl_timer( Pid, NewExpTs, #pool{ ttl = #ttl{timer_target = TgtPid} = TTL, all_members = AllMembers } = Pool ) -> {_, _, _, TgtExpTs} = maps:get(TgtPid, AllMembers), case NewExpTs < TgtExpTs of true -> schedule_ttl_timer_for(Pid, NewExpTs, TTL, Pool); false -> Pool end. %% @doc Cancel any existing TTL timer and schedule a new one for Pid at ExpTs. -spec schedule_ttl_timer_for(pid(), integer(), #ttl{}, #pool{}) -> #pool{}. schedule_ttl_timer_for(Pid, ExpTs, TTL, Pool) -> case TTL#ttl.timer of TRef when is_reference(TRef) -> erlang:cancel_timer(TRef); undefined -> ok end, DelayMs = max(0, ExpTs - erlang:monotonic_time(millisecond)), NewTRef = erlang:send_after(DelayMs, self(), {ttl_expired, Pid}), Pool#pool{ttl = TTL#ttl{timer = NewTRef, timer_target = Pid}}. %% @doc Cancel the current TTL timer and schedule a new one for the earliest-expiring %% free member. O(n) over free_pids — call only when the current target is removed. -spec reschedule_ttl_timer(#pool{}) -> #pool{}. reschedule_ttl_timer(#pool{ttl = undefined} = Pool) -> Pool; reschedule_ttl_timer(#pool{ttl = TTL, free_pids = FreePids, all_members = AllMembers} = Pool) -> case TTL#ttl.timer of TRef when is_reference(TRef) -> erlang:cancel_timer(TRef); undefined -> ok end, TTL1 = TTL#ttl{timer = undefined, timer_target = undefined}, Pool1 = Pool#pool{ttl = TTL1}, case find_earliest_expiry(FreePids, AllMembers) of none -> Pool1; {Pid, ExpTs} -> schedule_ttl_timer_for(Pid, ExpTs, TTL1, Pool1) end. %% @doc Scan free_pids for the member with the smallest finite ExpiresAt. O(n). -spec find_earliest_expiry([pid()], members_map()) -> {pid(), integer()} | none. find_earliest_expiry(Pids, AllMembers) -> lists:foldl( fun(Pid, Acc) -> case maps:get(Pid, AllMembers, undefined) of {_, _, _, infinity} -> Acc; {_, _, _, ExpTs} -> case Acc of none -> {Pid, ExpTs}; {_, Best} when ExpTs < Best -> {Pid, ExpTs}; _ -> Acc end; _ -> Acc end end, none, Pids ). %% @doc Extract only the static config fields of a #ttl{} for change-detection in %% reconfigure. Must not include timer or timer_target (those change at runtime). -spec ttl_static(undefined | #ttl{}) -> undefined | {time_spec(), time_spec()}. ttl_static(undefined) -> undefined; ttl_static(#ttl{max_lifetime = ML, jitter = J}) -> {ML, J}. %% @doc Recompute ExpiresAt for every member in AllMembers when TTL config changes. %% unset→set : each member gets now+lifetime+rand_jitter (spread by jitter). %% set→unset : every member becomes immortal (infinity). %% changed : shift all existing expiries by the delta; floor at now+(new_lifetime-jitter). -spec recompute_member_expiries(members_map(), undefined | #ttl{}, undefined | #ttl{}) -> members_map(). recompute_member_expiries(AllMembers, _OldTTL, undefined) -> maps:map(fun(_Pid, {MRef, S, Ts, _}) -> {MRef, S, Ts, infinity} end, AllMembers); recompute_member_expiries(AllMembers, undefined, NewTTL) -> maps:map( fun(_Pid, {MRef, S, Ts, _}) -> {MRef, S, Ts, compute_expiry(NewTTL)} end, AllMembers ); recompute_member_expiries(AllMembers, #ttl{max_lifetime = OldML}, #ttl{max_lifetime = NewML} = NewTTL) -> Delta = time_as_millis(NewML) - time_as_millis(OldML), Floor = erlang:monotonic_time(millisecond) + time_as_millis(NewML) - time_as_millis(NewTTL#ttl.jitter), maps:map( fun(_Pid, {MRef, S, Ts, OldExpTs}) -> NewExpTs = case OldExpTs of infinity -> compute_expiry(NewTTL); _ -> max(OldExpTs + Delta, Floor) end, {MRef, S, Ts, NewExpTs} end, AllMembers ). -spec maybe_reply({'queued' | 'error_no_members' | pid(), #pool{}}) -> {noreply, #pool{}} | {reply, 'error_no_members' | pid(), #pool{}}. maybe_reply({Member, NewPool}) -> case Member of queued -> {noreply, NewPool}; error_no_members -> {reply, error_no_members, NewPool}; Member when is_pid(Member) -> {reply, Member, NewPool} end. compute_utilization(#pool{ max_count = MaxCount, in_use_count = InUseCount, free_count = FreeCount, starting_members = Starting, stopping_count = StoppingCount, queued_requestors = Queue, queue_max = QueueMax }) -> [ {max_count, MaxCount}, {in_use_count, InUseCount}, {free_count, FreeCount}, {starting_count, length(Starting)}, {stopping_count, StoppingCount}, %% Note not O(n), so in pathological cases this might be expensive {queued_count, queue:len(Queue)}, {queue_max, QueueMax} ]. do_call_free_members(Fun, Pids) -> [do_call_free_member(Fun, P) || P <- Pids]. do_call_free_member(Fun, Pid) -> try {ok, Fun(Pid)} catch _Class:Reason -> {error, Reason} end. %% @private to_map(#pool{} = Pool) -> [Name | Values] = tuple_to_list(Pool), maps:from_list( [{'$record_name', Name} | lists:zip(record_info(fields, pool), Values)] ). %% @private -spec config_as_map(pool_config() | pool_config_legacy()) -> pool_config(). config_as_map(Conf) when is_map(Conf) -> Conf; config_as_map(LegacyConf) when is_list(LegacyConf) -> maps:from_list(LegacyConf). pg_get_local_members(GroupName) -> pg:get_local_members(GroupName). pg_delete(_GroupName) -> ok. pg_create(_Group) -> ok. pg_join(Group, Pid) -> pg:join(Group, Pid). pg_leave(Group, Pid) -> pg:leave(Group, Pid).