diff --git a/README.org b/README.org index f2d3847..08acc1a 100644 --- a/README.org +++ b/README.org @@ -254,9 +254,15 @@ a TCP handshake or a database login — still needs to happen before the member is safe to use. The value is an ={M, F, A}= tuple. Before calling, pooler replaces placeholder -atoms in =A=: ='$pooler_pid'= with the pid of the newly started member, and -='$pooler_pool_name'= with the member supervisor name for the pool (same -placeholders supported by =stop_mfa=): +atoms in =A= with the corresponding runtime values (same placeholders are +supported by =stop_mfa=): + +- ='$pooler_pid'= — the pid of the newly started member +- ='$pooler_member_sup'= — the name of the member supervisor that owns this + worker (shard-specific when [[#numbermembersups][=num_member_sups > 1=]]) +- ='$pooler_pool'= — the pool name atom +- ='$pooler_pool_name'= — *deprecated* alias for ='$pooler_member_sup'= kept + for backward compatibility; existing configs continue to work unchanged #+BEGIN_SRC erlang #{name => pg_pool, @@ -295,6 +301,41 @@ The callback is called from the helper process instead of =supervisor:terminate_child/2=. For maximum throughput under churn, prefer a fast or fire-and-forget teardown so that slots are freed quickly. +**** Graceful member shutdown (member_shutdown) + +By default pooler kills member processes with =brutal_kill= (immediate exit, +no cleanup). Set =member_shutdown= to a millisecond timeout to allow the +member's =terminate/2= callback to run before the supervisor considers it done: + +#+BEGIN_SRC erlang +#{name => pg_pool, + ... + member_shutdown => 5000} %% wait up to 5 s for clean shutdown +#+END_SRC + +The value maps directly to the [[https://www.erlang.org/doc/apps/stdlib/supervisor.html][OTP supervisor child spec]] =Shutdown= field: +=brutal_kill= sends =exit(Pid, kill)= (immediate, the default); a +=pos_integer()= sends =exit(Pid, shutdown)= and waits up to that many +milliseconds. + +Two caveats: + +- *Requires the worker to trap exits.* =exit(Pid, shutdown)= kills a process + that has not called =process_flag(trap_exit, true)= immediately — including + gen_server-based workers, which do /not/ set trap_exit automatically. + Workers must explicitly call =process_flag(trap_exit, true)= in their + =init/1= for =terminate/2= to run on shutdown. + +- *Ignored by custom =stop_mfa= that bypasses the supervisor.* If =stop_mfa= + calls the worker directly (e.g. =epgsql:close/1=) rather than going through + =supervisor:terminate_child/2=, the =Shutdown= field is never consulted. + The default =stop_mfa= uses =supervisor:terminate_child/2= and respects + this setting. + +Graceful shutdown interacts well with =num_member_sups=: stops across +different shards proceed in parallel, reducing total teardown time from +O(N × timeout) to O(N × timeout / num_member_sups). + **** Pools with slow-starting or slow-stopping workers For pools whose workers take significant time to start (e.g. opening a database @@ -314,6 +355,55 @@ or network connection), the recommended configuration pattern is: For pools with workers that are slow to stop, use =stop_mfa= with a fast asynchronous teardown so that the pool is not blocked while workers drain. +**** Sharded member supervisors (num_member_sups) +<> + +This feature mirrors the =num_conns_sups= option introduced in +[[https://ninenines.eu/articles/ranch-2.0.0/][Ranch 2.0]] to address the same supervisor mailbox bottleneck in +connection listeners. + +By default each pool has a single =simple_one_for_one= supervisor that owns all +member processes. Concurrent =start_child= / =terminate_child= calls (from +parallel starters, cull events, and async stoppers) serialize through that +supervisor's mailbox, so a slow =start_mfa= can stall the pool under load even +when individual workers are fine. + +The =num_member_sups= pool configuration option splits ownership across N +parallel supervisors. New starts are distributed round-robin and each worker +remembers which shard owns it, so terminations route to the correct supervisor. +Defaults to =1= (single supervisor, identical to the legacy layout). + +#+BEGIN_SRC erlang +#{name => pg_pool, + init_count => 32, + max_count => 128, + start_mfa => {my_conn, start_link, []}, + num_member_sups => 8} %% 8 parallel member supervisors +#+END_SRC + +When to use it: + +- Workloads with non-trivial =start_mfa= cost where =initialize_mfa= is not + an option (e.g. third-party libraries that only expose a monolithic + =start_link/N=). With =initialize_mfa= available, the supervisor is held only + for the cheap =start_link= portion and a single shard is usually sufficient. +- Workloads using a graceful =member_shutdown= timeout: with a single + supervisor all shutdowns serialise, so total stop time is O(N × timeout); + with M shards it becomes O(N × timeout / M). + +The cost of extra shards is negligible — each shard is one lightweight +supervisor process (~2–3 KB). When choosing N, the relevant figure is the +expected peak number of *concurrent* starts or stops rather than the total pool +size. For I/O-bound =start_link= (e.g. opening TCP connections), N can safely +exceed the CPU scheduler count since blocked processes yield the scheduler. +=N > max_count= is never useful — at most =max_count= workers can exist at any +moment, so surplus shards sit permanently empty. + +=num_member_sups= can be /increased/ via =pooler:pool_reconfigure/2=; existing +members stay on their original shards and new starts fill the added shards +round-robin until churn rebalances things. /Decreasing/ is not supported and +returns ={error, num_member_sups_cannot_be_decreased}=. + *** Pool Configuration via =pooler:new_pool= You can create pools using =pooler:new_pool/1= when accepts a map of pool configuration. Here's an example: @@ -337,6 +427,8 @@ pooler:pool_reconfigure(rc8081, PoolConfig#{max_count => 10, init_count => 4}). It will update the pool's state and will start/stop workers if necessary, join/leave group, reschedule the cull timer etc. The only parameters that can't be updated are ~name~ and ~start_mfa~. +~num_member_sups~ can only be increased (not decreased) — see +[[#numbermembersups][the sharding section]] for details. However, updated configuration won't survive pool crash (it will be restarted with old config by supervisor). But this should not normally happen. @@ -598,6 +690,11 @@ to start and supervise the members of this pool. The pooler_starter_sup is used to start temporary workers used for managing async member start. +With =num_member_sups > 1= (see [[#numbermembersups][Sharded member supervisors]]), the pool +supervisor starts N member supervisors. Shard 1 keeps the legacy name +=pooler_NAME_member_sup= shown above; additional shards are named +=pooler_NAME_member_sup_2=, =pooler_NAME_member_sup_3=, and so on. + pooler_sup: one_for_one pooler_NAME_pool_sup: all_for_one pooler_NAME_member_sup: simple_one_for_one diff --git a/src/pooler.appup.src b/src/pooler.appup.src index 6ba7088..a8c4c6a 100644 --- a/src/pooler.appup.src +++ b/src/pooler.appup.src @@ -1,4 +1,10 @@ % -*- mode: erlang -*- +%% When bumping the version (e.g. 1.7.0 → 1.8.0) and adding a `1\\.7\\.0.*' entry: +%% remember to include `{update, pooler_pool_sup, supervisor}'. pool_sup's +%% `init/1' builds child specs dynamically from `num_member_sups', so any change +%% to that logic (or to the MFA / id / count of the children it returns) requires +%% the supervisor to re-evaluate `init/1' on hot upgrade — `{load_module, ...}' +%% alone leaves the running supervisor with stale specs. {"1.7.0", [{<<"1\\.6\\.0.*">>, [{update, pooler, {advanced, []}}, diff --git a/src/pooler.erl b/src/pooler.erl index 6c5da14..63be676 100644 --- a/src/pooler.erl +++ b/src/pooler.erl @@ -61,7 +61,7 @@ code_change/3 ]). --vsn(4). +-vsn(5). %% Bump this value and add a new clause to `code_change', if the format of `#pool{}' record changed %% ------------------------------------------------------------------ @@ -99,6 +99,22 @@ timer_target = undefined :: pid() | undefined }). +%% Per-member bookkeeping. Stored as the value of #pool.all_members. +-record(member, { + mref :: reference(), + %% `free' when checked into the pool, the consumer's pid when checked out, + %% or `{stopping, replace | no_replace}' while async termination is in flight. + status :: free | pid() | {stopping, replace | no_replace}, + %% Timestamp the member entered its current `status' (used by cull/max_age). + time :: erlang:timestamp(), + %% `erlang:monotonic_time(millisecond)' deadline, or `infinity' when TTL is disabled. + expires_at :: integer() | infinity, + %% 1-based index into `#pool.member_sups' identifying which member supervisor + %% owns this worker. Pinned at start time and never changes; used to route + %% async terminations to the correct shard. + shard_idx = 1 :: pos_integer() +}). + -record(pool, { name :: atom(), group :: atom(), @@ -129,8 +145,16 @@ %% 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(), + %% Tuple of `pooler_pooled_worker_sup' names, one per shard. With the default + %% `num_member_sups = 1' this is a 1-tuple containing the legacy unsuffixed + %% supervisor name (`pooler__member_sup'); with N > 1, names are + %% `pooler__member_sup_1..N'. `tuple_size/1' is the source of truth for + %% the shard count. Stored as a tuple (not a list) for O(1) index access via + %% `element(ShardIdx, member_sups)' when routing starts and stops to a shard. + member_sups :: tuple(), + %% Round-robin counter for selecting the next shard to receive a new start. + %% 1-based; rotates modulo `tuple_size(member_sups)' after each pick. + next_shard = 1 :: pos_integer(), %% The supervisor used to start starter servers that start %% new members. This is what enables async member starts. @@ -151,11 +175,12 @@ %% 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()}], + %% A list of `{StarterPid, Timestamp, ShardIdx}' 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; + %% `ShardIdx' carries the shard the starter was assigned to so it can be + %% recorded on the resulting `#member{}' once `accept_member' fires. + starting_members = [] :: [{pid(), erlang:timestamp(), pos_integer()}], stopping_count = 0 :: non_neg_integer(), %% The maximum amount of time to allow for member start. @@ -218,11 +243,17 @@ metrics_api => folsom | exometer | telemetry, metrics_mod => module(), stop_mfa => pooler_starter:stop_mfa(), - initialize_mfa => {module(), atom(), ['$pooler_pid' | '$pooler_pool_name' | any(), ...]}, + initialize_mfa => + {module(), atom(), [ + '$pooler_pid' | '$pooler_member_sup' | '$pooler_pool' | '$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() + max_lifetime_jitter => time_spec(), + num_member_sups => pos_integer(), + member_shutdown => brutal_kill | pos_integer() }. %% See {@link pooler:new_pool/1} @@ -237,6 +268,7 @@ | {cull, _} | {leave_group, group_name()} | {join_group, group_name()} + | {add_member_sups, pos_integer(), pos_integer()} | {set_parameter, {group, group_name() | undefined} | {init_count, non_neg_integer()} @@ -248,18 +280,22 @@ | {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(), ...]}} + | {initialize_mfa, + undefined + | {module(), atom(), [ + '$pooler_pid' | '$pooler_member_sup' | '$pooler_pool' | '$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} +%% The 4-tuple shape returned by {@link pool_stats/1} for backward compatibility. +%% Internally, members are tracked as `#member{}' records. --type members_map() :: #{pid() => member_info()}. +-type members_map() :: #{pid() => #member{}}. -type consumers_map() :: #{pid() => {reference(), [pid()]}}. -if(?OTP_RELEASE >= 25). @@ -721,6 +757,8 @@ init(#{name := Name, max_count := MaxCount, init_count := InitCount, start_mfa : {error, Err} -> exit({error, Err}); {ok, T} -> T end, + NumMemberSups = maps:get(num_member_sups, P, 1), + MemberSups = pooler_pool_sup:member_sup_names(Name, NumMemberSups), Pool = #pool{ name = Name, group = maps:get(group, P, undefined), @@ -737,14 +775,13 @@ init(#{name := Name, max_count := MaxCount, init_count := InitCount, start_mfa : 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 + ttl = TTL, + member_sups = MemberSups }, - 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), + Pool1 = cull_members_from_pool(Pool), + {ok, NewPool} = init_members_sync(InitCount, Pool1), {ok, NewPool, {continue, join_group}}. handle_continue(join_group, #pool{group = undefined} = Pool) -> @@ -755,9 +792,6 @@ handle_continue(join_group, #pool{group = Group} = Pool) -> 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) -> @@ -767,7 +801,12 @@ handle_call({accept_member, StartResult}, _From, 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}; + Stats = maps:fold( + fun(Pid, M, Acc) -> [{Pid, member_to_info_tuple(M)} | Acc] end, + [], + Pool#pool.all_members + ), + {reply, Stats, Pool}; handle_call(pool_utilization, _From, Pool) -> {reply, compute_utilization(Pool), Pool}; handle_call(dump_pool, _From, Pool) -> @@ -805,7 +844,7 @@ handle_info({requestor_timeout, From}, Pool = #pool{queued_requestors = Requesto handle_info({'DOWN', MRef, process, Pid, Reason}, State) -> State1 = case maps:get(Pid, State#pool.all_members, undefined) of - {MRef, {stopping, Flag}, _Time, _ExpTs} -> + #member{mref = MRef, status = {stopping, Flag}} -> %% Expected death: member was being stopped asynchronously. Pool1 = State#pool{ all_members = maps:remove(Pid, State#pool.all_members), @@ -816,7 +855,7 @@ handle_info({'DOWN', MRef, process, Pid, Reason}, State) -> replace -> add_members_async(1, Pool1); no_replace -> Pool1 end; - {MRef, _Status, _Time, _ExpTs} -> + #member{mref = MRef} -> %% 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); @@ -844,7 +883,7 @@ 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, _, _} -> + #member{status = free} -> remove_pid(Pid, Pool1, replace, max_lifetime); _ -> %% In-use, stopping, or already gone — return path handles it @@ -866,8 +905,13 @@ code_change(_OldVsn, OldState, Extra) when tuple_size(OldState) =:= 24 -> 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(3, OldState, Extra) when tuple_size(OldState) =:= 27 -> + code_change(4, do_upgrade_to_v4(OldState), Extra); +%% v4 #pool{} (28-tuple: singular `member_sup', 4-tuple all_members entries, +%% 2-tuple starting_members) → v5 (`member_sups' tuple + `next_shard', +%% `#member{}' record entries, 3-tuple starting_members). +code_change(4, OldState, _Extra) when is_tuple(OldState), tuple_size(OldState) =:= 28, element(1, OldState) =:= pool -> + {ok, do_upgrade_to_v5(OldState)}; code_change(_, State, _Extra) -> {ok, State}. @@ -894,14 +938,44 @@ do_upgrade_to_v3( 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. +%% Converts a v3 27-element pool tuple to a v4 28-element pool tuple: inserts +%% `ttl=undefined' and extends `all_members' entries from 3-tuples to 4-tuples. +%% Output is positional (matching v4 `#pool{}' shape) so the next step +%% `do_upgrade_to_v5/1' can destructure it — the current `#pool{}' record def +%% has moved on to v5 shape and would mis-construct a v4 tuple via record syntax. 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} ) -> + NewAllMembers = maps:map( + fun(_Pid, {MRef, Status, Ts}) -> {MRef, Status, Ts, infinity} end, + AllMembers + ), + {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, + MaxAge, CullTimer, undefined, MemberSup, StarterSup, NewAllMembers, ConsumerToPid, StartingMembers, + StoppingCount, MemberStartTimeout, AutoGrowThreshold, StopMFA, InitializeMFA, MetricsMod, MetricsAPI, + QueuedRequestors, QueueMax}. + +%% Converts v4 `#pool{}' (singular `member_sup' field, 4-tuple `all_members' +%% entries, 2-tuple `starting_members' entries) to v5 (`member_sups' tuple + +%% `next_shard' counter, `#member{}' records, 3-tuple `starting_members'). +%% The new `#pool{}' record has a different shape so the old state is +%% destructured positionally. +do_upgrade_to_v5( + {pool, Name, Group, MaxCount, InitCount, StartMFA, FreePids, InUseCount, FreeCount, AddMemberRetry, CullInterval, + MaxAge, CullTimer, TTL, MemberSup, StarterSup, AllMembers, ConsumerToPid, StartingMembers, StoppingCount, + MemberStartTimeout, AutoGrowThreshold, StopMFA, InitializeMFA, MetricsMod, MetricsAPI, QueuedRequestors, + QueueMax} +) -> + NewAllMembers = maps:map( + fun(_Pid, {MRef, Status, Ts, ExpTs}) -> + #member{mref = MRef, status = Status, time = Ts, expires_at = ExpTs, shard_idx = 1} + end, + AllMembers + ), + NewStartingMembers = [{P, T, 1} || {P, T} <- StartingMembers], #pool{ name = Name, group = Group, @@ -915,15 +989,13 @@ do_upgrade_to_v4( cull_interval = CullInterval, max_age = MaxAge, cull_timer = CullTimer, - ttl = undefined, - member_sup = MemberSup, + ttl = TTL, + member_sups = {MemberSup}, + next_shard = 1, starter_sup = StarterSup, - all_members = maps:map( - fun(_Pid, {MRef, Status, Ts}) -> {MRef, Status, Ts, infinity} end, - AllMembers - ), + all_members = NewAllMembers, consumer_to_pid = ConsumerToPid, - starting_members = StartingMembers, + starting_members = NewStartingMembers, stopping_count = StoppingCount, member_start_timeout = MemberStartTimeout, auto_grow_threshold = AutoGrowThreshold, @@ -961,10 +1033,16 @@ do_accept_member( %% In this case, we should cleanup. pooler_starter:stop_member_async(StarterPid), Pool1; - {value, _, StartingMembers1} -> + {value, {_, _, ShardIdx}, StartingMembers1} -> MRef = erlang:monitor(process, Pid), ExpTs = compute_expiry(Pool1#pool.ttl), - Entry = {MRef, free, os:timestamp(), ExpTs}, + Entry = #member{ + mref = MRef, + status = free, + time = os:timestamp(), + expires_at = ExpTs, + shard_idx = ShardIdx + }, AllMembers1 = store_all_members(Pid, Entry, AllMembers), pooler_starter:stop(StarterPid), Pool2 = Pool1#pool{ @@ -1067,7 +1145,7 @@ take_member_bookkeeping( -spec remove_stale_starting_members( #pool{}, - [{pid(), erlang:timestamp()}], + [{pid(), erlang:timestamp(), pos_integer()}], time_spec() ) -> #pool{}. remove_stale_starting_members(Pool, StartingMembers, MaxAge) -> @@ -1082,7 +1160,7 @@ remove_stale_starting_members(Pool, StartingMembers, MaxAge) -> ), Pool#pool{starting_members = FilteredStartingMembers}. -accumulate_starting_member_not_stale(Pool, Now, SM = {Pid, StartTime}, MaxAgeSecs, AccIn) -> +accumulate_starting_member_not_stale(Pool, Now, SM = {Pid, StartTime, _ShardIdx}, MaxAgeSecs, AccIn) -> case secs_between(StartTime, Now) < MaxAgeSecs of true -> [SM | AccIn]; @@ -1099,27 +1177,33 @@ accumulate_starting_member_not_stale(Pool, Now, SM = {Pid, StartTime}, MaxAgeSec AccIn end. -init_members_sync(N, #pool{name = PoolName, member_sup = MemberSup, initialize_mfa = InitMFA} = Pool) -> +init_members_sync(N, #pool{name = PoolName, 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 + {StartRefs, Pool1} = lists:foldl( + fun(_I, {Acc, P}) -> + {ShardIdx, P1} = pick_shard(P), + MemberSup = member_sup_for(ShardIdx, P1), + StarterPid = pooler_starter:start_member(PoolName, MemberSup, Self, InitMFA), + {[{StarterPid, StartTime, ShardIdx} | Acc], P1} + end, + {[], Pool}, + lists:seq(1, N) + ), + Pool2 = Pool1#pool{starting_members = StartRefs}, + case collect_init_members(Pool2) of timeout -> ?LOG_ERROR( #{ label => "exceeded timeout waiting for members", pool => PoolName, - init_count => Pool1#pool.init_count + init_count => Pool2#pool.init_count }, #{domain => [pooler]} ), error({timeout, "unable to start members"}); - #pool{} = Pool2 -> - {ok, Pool2} + #pool{} = Pool3 -> + {ok, Pool3} end. collect_init_members(#pool{starting_members = Empty} = Pool) when @@ -1224,14 +1308,40 @@ take_member_from_pool_queued( %% `starting_members'. add_members_async( Count, - #pool{name = PoolName, member_sup = MemberSup, starting_members = StartingMembers, initialize_mfa = InitMFA} = Pool + #pool{name = PoolName, 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}. + {StartRefs, Pool1} = lists:foldl( + fun(_I, {Acc, P}) -> + {ShardIdx, P1} = pick_shard(P), + MemberSup = member_sup_for(ShardIdx, P1), + StarterPid = pooler_starter:start_member(PoolName, MemberSup, InitMFA), + {[{StarterPid, StartTime, ShardIdx} | Acc], P1} + end, + {[], Pool}, + lists:seq(1, Count) + ), + Pool1#pool{starting_members = StartRefs ++ StartingMembers}. + +%% @doc Round-robin shard selection. Returns `{ShardIdx, Pool'}' where the +%% returned `Pool'' has its `next_shard' counter advanced. +-spec pick_shard(#pool{}) -> {pos_integer(), #pool{}}. +pick_shard(#pool{next_shard = I, member_sups = Sups} = Pool) -> + N = tuple_size(Sups), + NextI = + case I >= N of + true -> 1; + false -> I + 1 + end, + {I, Pool#pool{next_shard = NextI}}. + +-spec member_sup_for(pos_integer(), #pool{}) -> atom() | pid(). +member_sup_for(ShardIdx, #pool{member_sups = Sups}) -> + element(ShardIdx, Sups). + +-spec shard_of(pid(), #pool{}) -> pos_integer(). +shard_of(Pid, #pool{all_members = AllMembers}) -> + (maps:get(Pid, AllMembers))#member.shard_idx. -spec do_return_member(pid(), ok | fail, #pool{}) -> #pool{}. do_return_member( @@ -1245,7 +1355,7 @@ do_return_member( ) -> clean_group_table(Pid, Pool), case maps:get(Pid, AllMembers, undefined) of - {_, free, _, _} -> + #member{status = free} -> ?LOG_WARNING( #{ label => "ignored return of free member", @@ -1255,10 +1365,10 @@ do_return_member( #{domain => [pooler]} ), Pool; - {_, {stopping, _}, _, _} -> + #member{status = {stopping, _}} -> %% member is being stopped asynchronously — ignore this return Pool; - {MRef, CPid, _FreeTs, ExpTs} -> + #member{status = CPid, expires_at = ExpTs} = Member -> #pool{ free_pids = Free, in_use_count = NumInUse, @@ -1275,17 +1385,22 @@ do_return_member( Pool2 = Pool1#pool{ stopping_count = Pool1#pool.stopping_count + 1, all_members = AllMembers#{ - Pid => {MRef, {stopping, replace}, os:timestamp(), ExpTs} + Pid => Member#member{status = {stopping, replace}, time = os:timestamp()} } }, pooler_starter_sup:new_stopper( - pooler_starter:stop_spec(PoolName, Pid, Pool2#pool.stop_mfa) + pooler_starter:stop_spec( + PoolName, + member_sup_for(shard_of(Pid, Pool2), Pool2), + 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}, + Entry = Member#member{status = free, time = os:timestamp(), expires_at = ExpTs}, Pool2 = Pool1#pool{ all_members = store_all_members(Pid, Entry, AllMembers) }, @@ -1305,10 +1420,10 @@ do_return_member(Pid, fail, #pool{all_members = AllMembers} = Pool) -> % removed, so use find instead of fetch and ignore missing. clean_group_table(Pid, Pool), case maps:get(Pid, AllMembers, undefined) of - {_MRef, {stopping, _}, _, _} -> + #member{status = {stopping, _}} -> %% already being stopped asynchronously — ignore Pool; - {_MRef, _, _, _} -> + #member{} -> %% replacement is triggered when the member's DOWN arrives remove_pid(Pid, Pool, replace, failed); undefined -> @@ -1354,7 +1469,7 @@ handle_unexpected_member_down(Pid, Pool) -> clean_group_table(Pid, Pool), AllMembers = Pool#pool.all_members, case maps:get(Pid, AllMembers, undefined) of - {_, free, _, _} -> + #member{status = free} -> Pool1 = Pool#pool{ free_pids = lists:delete(Pid, Pool#pool.free_pids), free_count = Pool#pool.free_count - 1, @@ -1369,7 +1484,7 @@ handle_unexpected_member_down(Pid, Pool) -> Pool1 end, add_members_async(1, Pool2); - {_, CPid, _, _} -> + #member{status = CPid} -> Pool1 = Pool#pool{ in_use_count = Pool#pool.in_use_count - 1, all_members = maps:remove(Pid, AllMembers), @@ -1397,14 +1512,16 @@ remove_pid(Pid, Pool, Flag, Reason) -> stop_mfa = StopMFA } = Pool, case maps:get(Pid, AllMembers, undefined) of - {MRef, free, Time, ExpTs} -> + #member{status = free, shard_idx = ShardIdx} = Member -> 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}} + all_members = AllMembers#{Pid => Member#member{status = {stopping, Flag}}} }, - pooler_starter_sup:new_stopper(pooler_starter:stop_spec(PoolName, Pid, StopMFA)), + pooler_starter_sup:new_stopper( + pooler_starter:stop_spec(PoolName, member_sup_for(ShardIdx, Pool1), 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 @@ -1413,14 +1530,16 @@ remove_pid(Pid, Pool, Flag, Reason) -> _ -> Pool1 end; - {MRef, CPid, Time, ExpTs} -> + #member{status = CPid, shard_idx = ShardIdx} = Member -> 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}}, + all_members = AllMembers#{Pid => Member#member{status = {stopping, Flag}}}, consumer_to_pid = cpmap_remove(Pid, CPid, CPMap) }, - pooler_starter_sup:new_stopper(pooler_starter:stop_spec(PoolName, Pid, StopMFA)), + pooler_starter_sup:new_stopper( + pooler_starter:stop_spec(PoolName, member_sup_for(ShardIdx, Pool1), 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; @@ -1439,19 +1558,24 @@ remove_pid(Pid, Pool, Flag, Reason) -> -spec store_all_members( pid(), - member_info(), + #member{}, members_map() ) -> members_map(). -store_all_members(Pid, Val = {_MRef, _CPid, _Time, _ExpTs}, AllMembers) -> +store_all_members(Pid, #member{} = Val, AllMembers) -> AllMembers#{Pid => Val}. +%% @doc Convert a `#member{}' record to the legacy 4-tuple shape exposed via +%% {@link pool_stats/1}. Kept for API stability with code that pattern-matches +%% the tuple form. +-spec member_to_info_tuple(#member{}) -> member_info(). +member_to_info_tuple(#member{mref = MRef, status = Status, time = Time, expires_at = ExpTs}) -> + {MRef, Status, Time, ExpTs}. + -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, + fun(#member{status = free} = M) -> M#member{status = CPid} end, AllMembers ). @@ -1521,20 +1645,20 @@ 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()}]. +-spec member_info([pid()], members_map()) -> [{pid(), #member{}}]. member_info(Pids, AllMembers) -> maps:to_list(maps:with(Pids, AllMembers)). -spec expired_free_members( - Members :: [{pid(), member_info()}], + Members :: [{pid(), #member{}}], Now :: {_, _, _}, MaxAge :: time_spec() -) -> [{pid(), free_member_info()}]. +) -> [{pid(), #member{}}]. expired_free_members(Members, Now, MaxAge) -> MaxMicros = time_as_micros(MaxAge), [ MI - || MI = {_, {_, free, LastReturn, _}} <- Members, + || MI = {_, #member{status = free, time = LastReturn}} <- Members, timer:now_diff(Now, LastReturn) >= MaxMicros ]. @@ -1556,7 +1680,8 @@ calculate_reconfigure_actions( initialize_mfa => undefined, metrics_mod => pooler_no_metrics, metrics_api => folsom, - queue_max => ?DEFAULT_POOLER_QUEUE_MAX + queue_max => ?DEFAULT_POOLER_QUEUE_MAX, + num_member_sups => 1 }, NewWithDefaults0 = maps:merge(Defaults, NewConfig), try @@ -1582,7 +1707,8 @@ calculate_reconfigure_actions( stop_mfa, initialize_mfa, auto_grow_threshold, - ttl + ttl, + num_member_sups ] ) of @@ -1689,6 +1815,13 @@ mk_rec_action(ttl, NewTTL, _Config, #pool{ttl = OldTTL}) -> {Same, Same} -> []; _ -> [{update_ttl, NewTTL}] end; +mk_rec_action(num_member_sups, New, _, #pool{member_sups = Sups}) -> + OldN = tuple_size(Sups), + if + New > OldN -> [{add_member_sups, OldN, New}]; + New < OldN -> throw({error, num_member_sups_cannot_be_decreased}); + true -> [] + end; mk_rec_action(_Param, _NewVal, _, _Pool) -> %% not changed []. @@ -1719,6 +1852,13 @@ apply_reconfigure_action({join_group, Group}, Pool) -> apply_reconfigure_action({leave_group, Group}, Pool) -> ok = pg_leave(Group, self()), Pool; +apply_reconfigure_action( + {add_member_sups, OldN, NewN}, + #pool{name = PoolName, start_mfa = StartMFA, member_sups = OldSups} = Pool +) -> + NewNames = pooler_pool_sup:add_member_sups(PoolName, StartMFA, OldN, NewN), + NewSups = list_to_tuple(tuple_to_list(OldSups) ++ NewNames), + Pool#pool{member_sups = NewSups}; apply_reconfigure_action({update_ttl, NewTTL}, Pool) -> case Pool#pool.ttl of #ttl{timer = TRef} when is_reference(TRef) -> erlang:cancel_timer(TRef); @@ -1933,8 +2073,8 @@ 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; + #member{expires_at = infinity} -> false; + #member{expires_at = ExpTs} -> erlang:monotonic_time(millisecond) >= ExpTs; undefined -> false end. @@ -1951,14 +2091,16 @@ remove_free_head( stop_mfa = StopMFA } = Pool ) -> - {MRef, free, Time, ExpTs} = maps:get(Pid, AllMembers), + #member{status = free, shard_idx = ShardIdx} = Member = 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}} + all_members = AllMembers#{Pid => Member#member{status = {stopping, replace}}} }, - pooler_starter_sup:new_stopper(pooler_starter:stop_spec(PoolName, Pid, StopMFA)), + pooler_starter_sup:new_stopper( + pooler_starter:stop_spec(PoolName, member_sup_for(ShardIdx, Pool1), 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 @@ -2012,7 +2154,7 @@ maybe_advance_ttl_timer( all_members = AllMembers } = Pool ) -> - {_, _, _, TgtExpTs} = maps:get(TgtPid, AllMembers), + #member{expires_at = TgtExpTs} = maps:get(TgtPid, AllMembers), case NewExpTs < TgtExpTs of true -> schedule_ttl_timer_for(Pid, NewExpTs, TTL, Pool); false -> Pool @@ -2052,9 +2194,9 @@ find_earliest_expiry(Pids, AllMembers) -> lists:foldl( fun(Pid, Acc) -> case maps:get(Pid, AllMembers, undefined) of - {_, _, _, infinity} -> + #member{expires_at = infinity} -> Acc; - {_, _, _, ExpTs} -> + #member{expires_at = ExpTs} -> case Acc of none -> {Pid, ExpTs}; {_, Best} when ExpTs < Best -> {Pid, ExpTs}; @@ -2080,23 +2222,23 @@ ttl_static(#ttl{max_lifetime = ML, jitter = J}) -> {ML, J}. %% 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); + maps:map(fun(_Pid, #member{} = M) -> M#member{expires_at = infinity} end, AllMembers); recompute_member_expiries(AllMembers, undefined, NewTTL) -> maps:map( - fun(_Pid, {MRef, S, Ts, _}) -> {MRef, S, Ts, compute_expiry(NewTTL)} end, + fun(_Pid, #member{} = M) -> M#member{expires_at = 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}) -> + fun(_Pid, #member{expires_at = OldExpTs} = M) -> NewExpTs = case OldExpTs of infinity -> compute_expiry(NewTTL); _ -> max(OldExpTs + Delta, Floor) end, - {MRef, S, Ts, NewExpTs} + M#member{expires_at = NewExpTs} end, AllMembers ). diff --git a/src/pooler_pool_sup.erl b/src/pooler_pool_sup.erl index d427433..437f00e 100644 --- a/src/pooler_pool_sup.erl +++ b/src/pooler_pool_sup.erl @@ -7,7 +7,9 @@ init/1, pool_sup_name/1, member_sup_name/1, - build_member_sup_name/1 + member_sup_names/2, + build_member_sup_name/1, + add_member_sups/4 ]). -spec start_link(pooler:pool_config()) -> {ok, pid()}. @@ -24,20 +26,25 @@ init(PoolConf) when is_map(PoolConf) -> type => worker, modules => [pooler] }, - MemberSupName = member_sup_name(PoolConf), - MemberSupSpec = - #{ - id => MemberSupName, - start => {pooler_pooled_worker_sup, start_link, [PoolConf]}, - restart => transient, - shutdown => 5000, - type => supervisor, - modules => [pooler_pooled_worker_sup] - }, - + PoolName = maps:get(name, PoolConf), + N = maps:get(num_member_sups, PoolConf, 1), + MemberSupSpecs = [ + begin + SupName = member_sup_name(PoolName, I, N), + #{ + id => SupName, + start => {pooler_pooled_worker_sup, start_link, [PoolConf, SupName]}, + restart => transient, + shutdown => 5000, + type => supervisor, + modules => [pooler_pooled_worker_sup] + } + end + || I <- lists:seq(1, N) + ], %% five restarts in 60 seconds, then shutdown Restart = #{strategy => one_for_all, intensity => 5, period => 60}, - {ok, {Restart, [MemberSupSpec, PoolerSpec]}}; + {ok, {Restart, MemberSupSpecs ++ [PoolerSpec]}}; init(PoolRecord) when is_tuple(PoolRecord), element(1, PoolRecord) =:= pool -> %% This clause is for the hot code upgrade from pre-1.6.0; %% can be removed when "upgrade-from-version" below 1.6.0 are removed from `pooler.appup.src' @@ -72,9 +79,57 @@ init(PoolRecord) when is_tuple(PoolRecord), element(1, PoolRecord) =:= pool -> member_sup_name(#{name := Name}) -> build_member_sup_name(Name). +%% @doc Return the member sup name for shard `I' of `N' total shards. +%% Shard 1 always uses the legacy unsuffixed name (`pooler__member_sup') +%% regardless of N, for consistency: a pool that grows from N=1 to N=2 via +%% reconfigure keeps its original shard-1 supervisor name unchanged. Additional +%% shards use `_2', `_3', ... suffixes. +-spec member_sup_name(pooler:pool_name(), pos_integer(), pos_integer()) -> atom(). +member_sup_name(PoolName, 1, _N) -> + build_member_sup_name(PoolName); +member_sup_name(PoolName, I, _N) -> + build_member_sup_shard_name(PoolName, I). + +%% @doc Build the tuple of member supervisor names for a pool with `N' shards. +%% With N=1, returns a 1-tuple with the legacy unsuffixed name for backward compatibility. +-spec member_sup_names(pooler:pool_name(), pos_integer()) -> tuple(). +member_sup_names(PoolName, N) -> + list_to_tuple([member_sup_name(PoolName, I, N) || I <- lists:seq(1, N)]). + -spec build_member_sup_name(pooler:pool_name()) -> atom(). build_member_sup_name(PoolName) -> list_to_atom("pooler_" ++ atom_to_list(PoolName) ++ "_member_sup"). +-spec build_member_sup_shard_name(pooler:pool_name(), pos_integer()) -> atom(). +build_member_sup_shard_name(PoolName, I) -> + list_to_atom("pooler_" ++ atom_to_list(PoolName) ++ "_member_sup_" ++ integer_to_list(I)). + +%% @doc Start additional member supervisors for shards `OldN+1..NewN' under +%% the pool's pool_sup. Used by `pooler:pool_reconfigure/2' when `num_member_sups' +%% is increased. New shards use `_I'-suffixed names (I >= 2; shard 1 is reserved +%% for the legacy unsuffixed name which already exists at this point). +-spec add_member_sups(pooler:pool_name(), {atom(), atom(), [term()]}, pos_integer(), pos_integer()) -> [atom()]. +add_member_sups(PoolName, StartMFA, OldN, NewN) when NewN > OldN -> + PoolSupName = pool_sup_name(#{name => PoolName}), + NewNames = [build_member_sup_shard_name(PoolName, I) || I <- lists:seq(OldN + 1, NewN)], + lists:foreach( + fun(SupName) -> + Spec = #{ + id => SupName, + start => {pooler_pooled_worker_sup, start_link, [#{start_mfa => StartMFA}, SupName]}, + restart => transient, + shutdown => 5000, + type => supervisor, + modules => [pooler_pooled_worker_sup] + }, + case supervisor:start_child(PoolSupName, Spec) of + {ok, _} -> ok; + {error, Reason} -> error({failed_to_add_member_sup, SupName, Reason}) + end + end, + NewNames + ), + NewNames. + pool_sup_name(#{name := Name}) -> list_to_atom("pooler_" ++ atom_to_list(Name) ++ "_pool_sup"). diff --git a/src/pooler_pooled_worker_sup.erl b/src/pooler_pooled_worker_sup.erl index 6179d5f..17786d6 100644 --- a/src/pooler_pooled_worker_sup.erl +++ b/src/pooler_pooled_worker_sup.erl @@ -2,19 +2,27 @@ -behaviour(supervisor). --export([start_link/1, init/1]). +-export([start_link/1, start_link/2, init/1]). -spec start_link(pooler:pool_config()) -> {ok, pid()} | {error, any()}. -start_link(#{start_mfa := MFA} = PoolConf) -> - SupName = pooler_pool_sup:member_sup_name(PoolConf), - supervisor:start_link({local, SupName}, ?MODULE, MFA). +start_link(#{start_mfa := _} = PoolConf) -> + start_link(PoolConf, pooler_pool_sup:member_sup_name(PoolConf)). -init({Mod, Fun, Args}) -> +-spec start_link(pooler:pool_config(), atom()) -> {ok, pid()} | {error, any()}. +start_link(#{start_mfa := MFA} = PoolConf, SupName) -> + Shutdown = maps:get(member_shutdown, PoolConf, brutal_kill), + supervisor:start_link({local, SupName}, ?MODULE, {MFA, Shutdown}). + +init({Mod, Fun, Args}) when is_atom(Mod) -> + %% Backward compat: old code passed just the MFA as init arg. + %% Reached during hot upgrade from a release that predates member_shutdown. + init({{Mod, Fun, Args}, brutal_kill}); +init({{Mod, Fun, Args}, Shutdown}) -> Worker = #{ id => Mod, start => {Mod, Fun, Args}, restart => temporary, - shutdown => brutal_kill, + shutdown => Shutdown, type => worker, modules => [Mod] }, diff --git a/src/pooler_starter.erl b/src/pooler_starter.erl index 20626a0..343e9be 100644 --- a/src/pooler_starter.erl +++ b/src/pooler_starter.erl @@ -7,9 +7,16 @@ -include_lib("kernel/include/logger.hrl"). +%% Legacy placeholder: misnomer — resolves to the (shard-specific) member sup name, +%% not the pool name. Kept as a deprecated alias for ?POOLER_MEMBER_SUP for backward +%% compatibility with existing user-provided stop_mfa configurations. -define(POOLER_POOL_NAME, '$pooler_pool_name'). +%% Shard-specific member supervisor name. Use this in new stop_mfa configurations. +-define(POOLER_MEMBER_SUP, '$pooler_member_sup'). +%% The pool name atom. Use this when stop_mfa needs to look up pool state. +-define(POOLER_POOL, '$pooler_pool'). -define(POOLER_PID, '$pooler_pid'). --define(DEFAULT_STOP_MFA, {supervisor, terminate_child, [?POOLER_POOL_NAME, ?POOLER_PID]}). +-define(DEFAULT_STOP_MFA, {supervisor, terminate_child, [?POOLER_MEMBER_SUP, ?POOLER_PID]}). %% ------------------------------------------------------------------ %% API Function Exports @@ -22,9 +29,9 @@ start_member/4, stop_member_async/1, stop/1, - stop_spec/3, + stop_spec/4, default_stop_mfa/0, - replace_placeholders/3 + replace_placeholders/4 ]). %% ------------------------------------------------------------------ @@ -46,20 +53,20 @@ -type pool_member_sup() :: pid() | atom(). -type parent() :: pid() | pool. -type initialize_mfa() :: undefined | {module(), atom(), [term()]}. --type stop_mfa() :: {module(), atom(), ['$pooler_pid' | '$pooler_pool_name' | term()]}. +-type stop_mfa() :: + {module(), atom(), ['$pooler_pid' | '$pooler_member_sup' | '$pooler_pool' | '$pooler_pool_name' | term()]}. -type start_result() :: {StarterPid :: pid(), Result :: pid() | {error, _}}. --opaque start_spec() :: {pooler:pool_name(), pool_member_sup(), parent(), initialize_mfa()}. -%% {PoolName, MemberPid, StopMFA} --opaque stop_spec() :: {pooler:pool_name(), pid(), stop_mfa()}. +-opaque start_spec() :: {starter_spec, pooler:pool_name(), pool_member_sup(), parent(), initialize_mfa()}. +-opaque stop_spec() :: {stopper_spec, pooler:pool_name(), pool_member_sup(), pid(), stop_mfa()}. %% ------------------------------------------------------------------ %% API Function Definitions %% ------------------------------------------------------------------ -spec start_link(start_spec() | stop_spec()) -> {ok, pid()}. -start_link({_, _, _} = Spec) -> +start_link({starter_spec, _, _, _, _} = Spec) -> gen_server:start_link(?MODULE, Spec, []); -start_link({_, _, _, _} = Spec) -> +start_link({stopper_spec, _, _, _, _} = Spec) -> gen_server:start_link(?MODULE, Spec, []). stop(Starter) -> @@ -81,7 +88,7 @@ start_member(PoolName, PoolMemberSup) -> -spec start_member(pooler:pool_name(), pool_member_sup(), initialize_mfa()) -> pid(). start_member(PoolName, PoolMemberSup, InitMFA) -> - {ok, Pid} = pooler_starter_sup:new_starter({PoolName, PoolMemberSup, pool, InitMFA}), + {ok, Pid} = pooler_starter_sup:new_starter({starter_spec, PoolName, PoolMemberSup, pool, InitMFA}), Pid. %% @doc Same as {@link start_member/2} except that instead of calling @@ -94,7 +101,7 @@ start_member(PoolName, PoolMemberSup, InitMFA) -> %% initial set of pool members in parallel. -spec start_member(pooler:pool_name(), pool_member_sup(), pid(), initialize_mfa()) -> pid(). start_member(PoolName, PoolMemberSup, Parent, InitMFA) -> - {ok, Pid} = pooler_starter_sup:new_starter({PoolName, PoolMemberSup, Parent, InitMFA}), + {ok, Pid} = pooler_starter_sup:new_starter({starter_spec, PoolName, PoolMemberSup, Parent, InitMFA}), Pid. %% @doc Stop a member in the pool @@ -123,15 +130,16 @@ stop_member_async(Pid) -> }). -spec init(start_spec() | stop_spec()) -> {ok, #starter{}, {continue, start | stop}}. -init({PoolName, MemberPid, StopMFA}) -> +init({stopper_spec, PoolName, MemberSup, MemberPid, StopMFA}) -> {ok, #starter{ pool_name = PoolName, + pool_member_sup = MemberSup, stopping_pid = MemberPid, stopping_mfa = StopMFA }, {continue, stop}}; -init({PoolName, PoolMemberSup, Parent, InitMFA}) -> +init({starter_spec, PoolName, PoolMemberSup, Parent, InitMFA}) -> {ok, #starter{pool_name = PoolName, pool_member_sup = PoolMemberSup, parent = Parent, initialize_mfa = InitMFA}, {continue, start}}. @@ -145,9 +153,10 @@ handle_continue( {noreply, State#starter{msg = Msg}}; handle_continue( stop, - #starter{pool_name = PoolName, stopping_pid = MemberPid, stopping_mfa = StopMFA} = State + #starter{pool_name = PoolName, pool_member_sup = MemberSup, stopping_pid = MemberPid, stopping_mfa = StopMFA} = + State ) -> - terminate_pid(PoolName, MemberPid, StopMFA), + terminate_pid(PoolName, MemberSup, MemberPid, StopMFA), {stop, normal, State}. handle_call(_Request, _From, State) -> @@ -186,7 +195,7 @@ code_change(_OldVsn, State, _Extra) -> do_start_member(PoolSup, PoolName, InitMFA) -> case supervisor:start_child(PoolSup, []) of {ok, Pid} -> - case call_initialize_mfa(PoolName, Pid, InitMFA) of + case call_initialize_mfa(PoolName, PoolSup, Pid, InitMFA) of ok -> {self(), Pid}; Error -> @@ -218,38 +227,43 @@ do_start_member(PoolSup, PoolName, InitMFA) -> default_stop_mfa() -> ?DEFAULT_STOP_MFA. --spec stop_spec(pooler:pool_name(), pid(), stop_mfa()) -> stop_spec(). -stop_spec(PoolName, MemberPid, StopMFA) -> - {PoolName, MemberPid, StopMFA}. +-spec stop_spec(pooler:pool_name(), pool_member_sup(), pid(), stop_mfa()) -> stop_spec(). +stop_spec(PoolName, MemberSup, MemberPid, StopMFA) -> + {stopper_spec, PoolName, MemberSup, MemberPid, StopMFA}. %% @doc Best-effort termination for a pool member: applies the given MFA with -%% `?POOLER_PID' and `?POOLER_POOL_NAME' placeholders replaced by the actual -%% pid and pool name. Falls back to the default stop MFA on any failure. --spec terminate_pid(pooler:pool_name(), pid(), stop_mfa()) -> ok. -terminate_pid(PoolName, Pid, {Mod, Fun, Args}) when is_list(Args) -> - NewArgs = replace_placeholders(PoolName, Pid, Args), +%% `?POOLER_PID', `?POOLER_MEMBER_SUP', `?POOLER_POOL', and (legacy) +%% `?POOLER_POOL_NAME' placeholders replaced by the actual values. Falls back +%% to the default stop MFA on any failure. +-spec terminate_pid(pooler:pool_name(), pool_member_sup(), pid(), stop_mfa()) -> ok. +terminate_pid(PoolName, MemberSup, Pid, {Mod, Fun, Args}) when is_list(Args) -> + NewArgs = replace_placeholders(PoolName, MemberSup, Pid, Args), try erlang:apply(Mod, Fun, NewArgs) of _ -> ok catch - _:_ -> terminate_pid(PoolName, Pid, ?DEFAULT_STOP_MFA) + _:_ -> terminate_pid(PoolName, MemberSup, Pid, ?DEFAULT_STOP_MFA) end. --spec replace_placeholders(pooler:pool_name(), pid(), [term()]) -> [term()]. -replace_placeholders(PoolName, Pid, Args) -> +-spec replace_placeholders(pooler:pool_name(), pool_member_sup(), pid(), [term()]) -> [term()]. +replace_placeholders(PoolName, MemberSup, Pid, Args) -> [ case Arg of - ?POOLER_POOL_NAME -> pooler_pool_sup:build_member_sup_name(PoolName); + ?POOLER_MEMBER_SUP -> MemberSup; + %% Legacy alias — semantically same as ?POOLER_MEMBER_SUP. The original + %% name was a misnomer (it never resolved to the pool name). + ?POOLER_POOL_NAME -> MemberSup; + ?POOLER_POOL -> PoolName; ?POOLER_PID -> Pid; _ -> Arg end || Arg <- Args ]. --spec call_initialize_mfa(pooler:pool_name(), pid(), initialize_mfa()) -> ok | {error, term()}. -call_initialize_mfa(_PoolName, _Pid, undefined) -> +-spec call_initialize_mfa(pooler:pool_name(), pool_member_sup(), pid(), initialize_mfa()) -> ok | {error, term()}. +call_initialize_mfa(_PoolName, _MemberSup, _Pid, undefined) -> ok; -call_initialize_mfa(PoolName, Pid, {Mod, Fun, Args}) -> - NewArgs = replace_placeholders(PoolName, Pid, Args), +call_initialize_mfa(PoolName, MemberSup, Pid, {Mod, Fun, Args}) -> + NewArgs = replace_placeholders(PoolName, MemberSup, Pid, Args), try erlang:apply(Mod, Fun, NewArgs) of ok -> ok; {error, _} = Err -> Err; diff --git a/test/pooler_tests.erl b/test/pooler_tests.erl index 9b658fa..92a03d0 100644 --- a/test/pooler_tests.erl +++ b/test/pooler_tests.erl @@ -2209,6 +2209,149 @@ get_n_pids_group(Group, N, Acc) -> children_count(SupId) -> length(supervisor:which_children(SupId)). +%% =================================================================== +%% Sharded member supervisor tests (num_member_sups) +%% =================================================================== + +pooler_sharded_member_sups_test_() -> + MFA = {pooled_gs, start_link, [{"shard-type"}]}, + {foreach, + fun() -> + application:set_env(pooler, pools, []), + application:set_env(pooler, metrics_module, pooler_no_metrics), + application:start(pooler) + end, + fun(_) -> + application:stop(pooler) + end, + [ + {"default num_member_sups=1 uses legacy unsuffixed supervisor name", fun() -> + {ok, _} = pooler:new_pool(#{ + name => shard_pool_1, + max_count => 4, + init_count => 2, + start_mfa => MFA + }), + ?assertMatch(P when is_pid(P), whereis(pooler_shard_pool_1_member_sup)), + ?assertEqual(undefined, whereis(pooler_shard_pool_1_member_sup_1)), + ?assertEqual(1, length(shard_sup_names(shard_pool_1))) + end}, + {"num_member_sups=4 starts 4 supervisors with round-robin worker distribution", fun() -> + {ok, _} = pooler:new_pool(#{ + name => shard_pool_2, + max_count => 16, + init_count => 8, + num_member_sups => 4, + start_mfa => MFA + }), + wait_for_dump(shard_pool_2, 5000, fun(#{free_count := C}) -> C =:= 8 end), + Sups = shard_sup_names(shard_pool_2), + ?assertEqual(4, length(Sups)), + [?assertMatch(P when is_pid(P), whereis(S)) || S <- Sups], + %% shard 1 always uses the legacy unsuffixed name; _1 suffix must not exist + ?assertEqual(undefined, whereis(pooler_shard_pool_2_member_sup_1)), + %% 8 workers / 4 shards = 2 per shard (round-robin is exact on init) + Counts = [proplists:get_value(active, supervisor:count_children(S), 0) || S <- Sups], + ?assertEqual([2, 2, 2, 2], Counts) + end}, + {"fail-returned member is replaced within its own shard", fun() -> + {ok, _} = pooler:new_pool(#{ + name => shard_pool_3, + max_count => 4, + init_count => 4, + num_member_sups => 4, + start_mfa => MFA + }), + wait_for_dump(shard_pool_3, 5000, fun(#{free_count := C}) -> C =:= 4 end), + Pids = [pooler:take_member(shard_pool_3, 1000) || _ <- lists:seq(1, 4)], + [?assert(is_pid(P)) || P <- Pids], + [pooler:return_member(shard_pool_3, P, fail) || P <- Pids], + %% all 4 replacements must arrive; no orphan workers + wait_for_dump(shard_pool_3, 5000, fun(#{free_count := C, stopping_count := S}) -> + C =:= 4 andalso S =:= 0 + end), + ?assertEqual(4, total_shard_workers(shard_pool_3)) + end}, + {"cull removes excess workers; supervisor counts match pool state", fun() -> + {ok, _} = pooler:new_pool(#{ + name => shard_pool_4, + max_count => 8, + init_count => 2, + cull_interval => {500, ms}, + max_age => {0, sec}, + num_member_sups => 4, + start_mfa => MFA + }), + %% grow pool past init_count + Pids = [pooler:take_member(shard_pool_4, 1000) || _ <- lists:seq(1, 6)], + [?assert(is_pid(P)) || P <- Pids], + [pooler:return_member(shard_pool_4, P) || P <- Pids], + %% trigger cull immediately; skip the racy wait_for_dump(free_count=6) + %% — the cull interval may fire before we observe that state + shard_pool_4 ! cull_pool, + %% wait for all async stops to drain, then verify supervisor + %% counts match pool accounting (no orphan workers) + #{free_count := FC, in_use_count := IC} = + wait_for_dump(shard_pool_4, 5000, fun(#{stopping_count := S}) -> S =:= 0 end), + ?assertEqual(FC + IC, total_shard_workers(shard_pool_4)), + ?assertEqual(2, FC + IC) + end}, + {"reconfigure can increase num_member_sups", fun() -> + Cfg = #{ + name => shard_pool_5, + max_count => 8, + init_count => 4, + num_member_sups => 2, + start_mfa => MFA + }, + {ok, _} = pooler:new_pool(Cfg), + wait_for_dump(shard_pool_5, 5000, fun(#{free_count := C}) -> C =:= 4 end), + ?assertEqual(2, length(shard_sup_names(shard_pool_5))), + {ok, Actions} = pooler:pool_reconfigure(shard_pool_5, Cfg#{num_member_sups => 4}), + ?assertMatch([{add_member_sups, 2, 4}], Actions), + ?assertEqual(4, length(shard_sup_names(shard_pool_5))) + end}, + {"reconfigure refuses to decrease num_member_sups", fun() -> + Cfg = #{ + name => shard_pool_6, + max_count => 4, + init_count => 2, + num_member_sups => 4, + start_mfa => MFA + }, + {ok, _} = pooler:new_pool(Cfg), + ?assertEqual( + {error, num_member_sups_cannot_be_decreased}, + pooler:pool_reconfigure(shard_pool_6, Cfg#{num_member_sups => 2}) + ) + end}, + {"reconfigure with same num_member_sups emits no shard action", fun() -> + Cfg = #{ + name => shard_pool_7, + max_count => 4, + init_count => 2, + num_member_sups => 2, + start_mfa => MFA + }, + {ok, _} = pooler:new_pool(Cfg), + {ok, Actions} = pooler:pool_reconfigure(shard_pool_7, Cfg), + ?assertEqual(false, lists:keymember(add_member_sups, 1, Actions)), + ?assertEqual(2, length(shard_sup_names(shard_pool_7))) + end} + ]}. + +%% Return the list of member supervisor names for PoolName (all shards). +shard_sup_names(PoolName) -> + #{member_sups := Sups} = dump_pool(PoolName), + tuple_to_list(Sups). + +%% Sum active worker counts across all shards. +total_shard_workers(PoolName) -> + lists:sum([ + proplists:get_value(active, supervisor:count_children(S), 0) + || S <- shard_sup_names(PoolName) + ]). + starting_members(PoolName) -> length(maps:get(starting_members, dump_pool(PoolName))). diff --git a/test/prop_pooler.erl b/test/prop_pooler.erl index 982f203..7302585 100644 --- a/test/prop_pooler.erl +++ b/test/prop_pooler.erl @@ -7,7 +7,8 @@ prop_fixed_take_return/1, prop_fixed_take_return_broken/1, prop_fixed_client_died/1, - prop_group_take_return/1 + prop_group_take_return/1, + prop_sharded_worker_count_invariant/1 ]). -include_lib("proper/include/proper.hrl"). @@ -427,10 +428,50 @@ assert_worker_count_bounded(PoolNameOrPid, MaxCount) -> {registered_name, N} = erlang:process_info(PoolNameOrPid, registered_name), N end, - MemberSup = pooler_pool_sup:build_member_sup_name(PoolName), - Counts = supervisor:count_children(MemberSup), - Active = proplists:get_value(active, Counts, 0), + Active = total_shard_workers(PoolName), ?assert(Active =< MaxCount). +prop_sharded_worker_count_invariant(doc) -> + "For any take/return sequence on a sharded pool, the total active supervisor " + "worker count across all shards always equals the number of tracked members.". + +prop_sharded_worker_count_invariant() -> + ?FORALL( + {Size, NumShards}, + {range(1, 8), range(1, 4)}, + with_pool( + #{ + name => ?FUNCTION_NAME, + init_count => Size * NumShards, + max_count => Size * NumShards, + num_member_sups => NumShards, + start_mfa => {pooled_gs, start_link, [{?FUNCTION_NAME}]} + }, + fun() -> + Pool = ?FUNCTION_NAME, + Total = Size * NumShards, + pool_is_free(Pool, Total), + assert_shard_count_eq(Pool, Total), + Pids = [pooler:take_member(Pool) || _ <- lists:seq(1, Total)], + assert_shard_count_eq(Pool, Total), + [pooler:return_member(Pool, P) || P <- Pids], + assert_shard_count_eq(Pool, Total), + true + end + ) + ). + +%% Sum active worker counts across all shards of PoolName. +total_shard_workers(PoolName) -> + #{member_sups := Sups} = gen_server:call(PoolName, dump_pool), + lists:sum([ + proplists:get_value(active, supervisor:count_children(S), 0) + || S <- tuple_to_list(Sups) + ]). + +assert_shard_count_eq(PoolName, Expected) -> + Active = total_shard_workers(PoolName), + ?assertEqual(Expected, Active). + pg_start() -> pg:start(pg).