%% -*- erlang-indent-level: 4;indent-tabs-mode: nil -*- %% -------------------------------------------------- %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file %% except in compliance with the License. You may obtain %% a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, %% software distributed under the License is distributed on an %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. %% -------------------------------------------------- %% %% @author Ulf Wiger %% %% @doc Load balancing functions based on Gproc. %% %% This module implements support for load-balancing server pools. It was %% originally intended mainly as an example of how to use various Gproc %% resources (e.g. counters and shared properties), but is fully integrated %% into Gproc, and fully functional. %% %%

Concepts

%% %% Each pool has a list of 'named' workers (defined using `add_worker/2') and %% a load-balancing strategy. Processes can then 'connect' to the pool (with %% `connect_worker/2'), using one of the defined names. %% %% Users then 'pick' one of the currently connected processes in the pool. Which %% process is picked depends on the load-balancing strategy. %% %% The whole representation of the pool and its connected workers is in gproc. %% The server `gproc_pool' is used to serialize pool management updates, but %% worker selection is performed entirely in the calling process, and can be %% performed by several processes concurrently. %% %%

Load-balancing strategies

%% %% * `round_robin' is the default. A wrapping gproc counter keeps track of the %% latest worker picked, and `gproc:next()' is used to find the next worker. %% * `random' picks a random worker from the pool. %% * `hash' requires a value (`pick/2'), and picks a worker based on the hash of %% that value. %% * `direct' takes an integer as an argument, and picks the next worker (modulo %% the size of the pool). This is mainly for implementations that implement %% a load-balancing strategy on top of `gproc_pool'. %% * `claim' picks the first available worker and 'claims' it while executing %% a user-provided fun. This means that the number of concurrently executing %% jobs will not exceed the size of the pool. %% @end -module(gproc_pool). -behavior(gen_server). %% gproc round-robin name lookup -export([new/1, % (Pool) -> (Pool, round_robin, []) new/3, % (Pool, Type, Opts) delete/1, % (Pool) force_delete/1, % (Pool) add_worker/2, % (Pool, Name) -> Pos add_worker/3, % (Pool, Name, Pos) -> Pos remove_worker/2, % (Pool, Name) connect_worker/2, % (Pool, Name) disconnect_worker/2, % (Pool, Name) whereis_worker/2, % (Pool, Name) worker_id/2, % (Pool, Name) active_workers/1, % (Pool) defined_workers/1, % (Pool) worker_pool/1, % (Pool) pick/1, % (Pool) pick/2, % (Pool, Value) pick_worker/1, % (Pool) pick_worker/2, % (Pool, Value) claim/2, % (Pool, Fun) claim/3, % (Pool, Fun, Wait) log/1, % (WorkerId) randomize/1]). % (Pool) -export([start_link/0]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([test/1, test/3, ptest/4, test_run/2, test_run1/2, test_run2/2, test_run0/2, setup_test_pool/3, setup_test_pool/4, remove_test_pool/1]). -include("gproc_int.hrl"). -define(POOL(Pool), {p,l,{?MODULE,Pool}}). -define(POOL_CUR(Pool), {c,l,{?MODULE,Pool,cur}}). -define(POOL_WRK(Pool,Name), {c,l,{?MODULE,Pool,w,Name}}). -record(st, {}). %% @spec new(Pool::any()) -> ok %% %% @equiv new(Pool, round_robin, []) new(Pool) -> new(Pool, round_robin, []). %% @spec new(Pool::any(), Type, Opts) -> true %% Type = round_robin | random | hash | direct | claim %% Opts = [{size, integer()} | {auto_size, boolean()}] %% %% @doc Create a new pool. %% %% The pool starts out empty. If a size is not given, the pool size is set to %% 0 initially. `auto_size' is `true' by default if size is not specified, but %% `false' by default otherwise. If `auto_size == true', the pool will be %% enlarged to accommodate new workers, when necessary. Otherwise, trying to add %% a worker when the pool is full will raise an exception, as will trying to add %% a worker on a specific position beyond the current size of the pool. %% %% If the given pool already exists, this function will raise an exception. %% @end new(Pool, Type, Opts) when Type == round_robin; Type == random; Type == hash; Type == direct; Type == claim -> call({new, Pool, Type, Opts}). %% @spec delete(Pool::any()) -> true %% @doc Delete an existing pool. %% %% This function will delete a pool, only if there are no connected workers. %% Ensure that workers have been disconnected before deleting the pool. %% @end %% delete(Pool) -> call({delete, Pool}). %% @spec force_delete(Pool::any()) -> true %% @doc Forcibly remove a pool, terminating all active workers %% %% This function is primarily intended for cleanup of any pools that might have %% become inconsistent (for whatever reason). It will clear out all resources %% belonging to the pool and send `exit(Pid, kill)' signals to all connected %% workers (except the calling process). %% @end %% force_delete(Pool) -> %% This is not pretty, but this function is mainly intended to clean up %% a pool that's not used, with no regard to connected workers, except self(), %% (that is, we kill each connected worker). We don't worry about races, %% so don't go to the server (which doesn't have own state on the pool %% anyway). force_delete_(Pool). %% @spec add_worker(Pool::any(), Name::any()) -> integer() %% %% @doc Assign a worker name to the pool, returning the worker's position. %% %% Before a worker can connect to the pool, its name must be added. If no explicit %% position is given (see {@link add_worker/3}), the most suitable position, %% depending on load-balancing algorithm, is selected: for round_robin and direct %% pools, names are packed tightly from the beginning; for hash and random pools, %% slots are filled as sparsely as possible, in order to maintain an even %% likelihood of hitting each worker. %% %% An exception is raised if the pool is full (and `auto_size' is false), or if %% `Name' already exists in the pool. %% %% Before a worker can be used, a process must connect to it (see %% {@link connect_worker/2}. %% @end add_worker(Pool, Name) -> call({add_worker, Pool, Name}). %% @spec add_worker(Pool::any(), Name::any(), Slot::integer()) -> integer() %% %% @doc Assign a worker name to a given slot in the pool, returning the slot. %% %% This function allows the pool maintainer to exactly position each worker %% inside the pool. An exception is raised if the position is already taken, %% or if `Name' already exists in the pool. If `Slot' is larger than the current %% size of the pool, an exception is raised iff `auto_size' is `false'; %% otherwise the pool is expanded to accommodate the new position. %% @end add_worker(Pool, Name, Slot) -> call({add_worker, Pool, Name, Slot}). %% @spec connect_worker(Pool::any(), Name::any()) -> true %% @doc Connect the current process to `Name' in `Pool'. %% %% Typically, a server will call this function as it starts, similarly to when %% it registers itself. In fact, calling `connect_worker/2' leads to the process %% being registered as `{n,l,[gproc_pool,N,Name]}', where `N' is the position of %% `Name' in the pool. This means (a) that gproc monitors the worker, and %% removes the connection automatically if it dies, and (b) that the registered %% names can be listed in order of their positions in the pool. %% %% This function raises an exception if `Name' does not exist in `Pool' (or %% there is no such pool), or if another worker is already connected to %% `Name'. %% @end %% connect_worker(Pool, Name) -> gproc:reg(worker_id(Pool, Name), 0). %% @spec disconnect_worker(Pool, Name) -> true %% %% @doc Disconnect the current process from `Name' in `Pool'. %% %% This function is similar to a `gproc:unreg()' call. It removes the %% connection between `Pool', `Name' and pid, and makes it possible for another %% process to connect to `Name'. %% %% An exception is raised if there is no prior connection between `Pool', %% `Name' and the current process. %% @end %% disconnect_worker(Pool, Name) -> gproc:unreg(worker_id(Pool, Name)). %% @spec remove_worker(Pool::any(), Name::any()) -> true %% @doc Remove a previously added worker. %% %% This function will assume that any connected worker is disconnected first. %% It will fail if there is no such pool, but will return `true' in the case %% when `Name' did not exist in the pool in the first place. %% @end remove_worker(Pool, Name) -> call({remove_worker, Pool, Name}). %% @spec whereis_worker(Pool::any(), Name::any()) -> pid() | undefined %% @doc Look up the pid of a connected worker. %% %% This function works similarly to `gproc:where/1': it will return the pid %% of the worker connected as `Pool / Name', if there is such a worker; otherwise %% it will return `undefined'. It will raise an exception if `Name' has not been %% added to the pool. %% @end whereis_worker(Pool, Name) -> ID = worker_id(Pool, Name), gproc:where(ID). %% @spec worker_id(Pool, Name) -> GprocName %% @doc Return the unique gproc name corresponding to a name in the pool. %% %% This function assumes that `Name' has been added to `Pool'. It returns the %% unique name that a connected worker will be registered as. This doesn't mean %% that there is, in fact, such a connected worker. %% @end worker_id(Pool, Name) -> N = gproc:get_attribute(?POOL_WRK(Pool, Name), shared, n), {n, l, [?MODULE, Pool, N, Name]}. %% @spec active_workers(Pool::any()) -> [{Name, Pid}] %% @doc Return a list of currently connected workers in the pool. %% active_workers(Pool) -> gproc:select( {l,n}, [{ {{n,l,[?MODULE,Pool,'$1','$2']},'$3','_'}, [{is_integer, '$1'}], [{{'$2', '$3'}}] }]). %% @spec defined_workers(Pool::any()) -> [{Name, Pos, Count}] %% @doc Return a list of added workers in the pool. %% %% The added workers are slots in the pool that have been given names, and thus %% can be connected to. This function doesn't detect whether or not there are %% any connected (active) workers. %% %% The list contains `{Name, Pos, Count}', where `Name' is the name of the added %% worker, `Pos' is its position in the pool, and `Count' represents the number %% of times the worker has been picked (assuming callers keep count by explicitly %% calling {@link log/1}). %% @end defined_workers(Pool) -> K = ?POOL(Pool), [{N, Pos, gproc:get_value(?POOL_WRK(Pool, N), shared)} || {N, Pos} <- get_workers_(K)]. %% @spec worker_pool(Pool::any()) -> [integer() | {Name, Pos}] %% @doc Return a list of slots and/or named workers in the pool. %% %% This function is mainly for testing, but can also be useful when implementing %% your own worker placement algorithm on top of gproc_pool. %% %% A plain integer represents an unfilled slot, and `{Name, Pos}' represents an %% added worker. The pool is always filled to the current size. %% @end worker_pool(Pool) -> get_workers_(?POOL(Pool)). %% @spec pick(Pool::any()) -> GprocName | false %% @doc Pick a worker from the pool given the pool's load-balancing algorithm. %% %% The pool types that allows picking without an extra argument are %% round_robin and random. This function returns `false' if there is no available %% worker, or if `Pool' is not a valid pool. %% @end pick(Pool) -> case gproc:get_value(?POOL(Pool), shared) of {0, _} -> false; {Sz, Type} when Type == round_robin; Type == random -> pick(Pool, Sz, Type, name); _ -> error(badarg) end. %% @spec pick_worker(Pool::any()) -> pid() | false %% @doc Pick a worker pid from the pool given the pool's load-balancing algorithm. %% %% Like {@link pick/1}, but returns the worker pid instead of the name. %% @end pick_worker(Pool) -> case gproc:get_value(?POOL(Pool), shared) of {0, _} -> false; {Sz, Type} when Type == round_robin; Type == random -> pick(Pool, Sz, Type, pid); _ -> error(badarg) end. %% @spec pick(Pool::any(), Value::any()) -> GprocName | false %% @doc Pick a worker from the pool based on `Value'. %% %% The pool types that allows picking based on an extra argument are %% hash and direct. This function returns `false' if there is no available %% worker, or if `Pool' is not a valid pool. %% %% If the pool is of type `direct', `Value' must be an integer corresponding to %% a position in the pool (modulo the size of the pool). If the type is %% `hash', `Value' may be any term, and its hash value will serve as a guide for %% selecting a worker. %% @end pick(Pool, N) -> case gproc:get_value(?POOL(Pool), shared) of {0, _} -> false; {Sz, Type} when Type == hash; Type == direct -> pick(Pool, Sz, Type, N, name); _ -> error(badarg) end. %% @spec pick_worker(Pool::any(), Value::any()) -> pid() | false %% @doc Pick a worker pid from the pool given the pool's load-balancing algorithm. %% %% Like {@link pick/2}, but returns the worker pid instead of the name. %% @end pick_worker(Pool, N) -> case gproc:get_value(?POOL(Pool), shared) of {0, _} -> false; {Sz, Type} when Type == hash; Type == direct -> pick(Pool, Sz, Type, N, pid); _ -> error(badarg) end. pick(Pool, Sz, round_robin, Ret) -> Next = incr(Pool, 1, Sz), case ets:next(gproc, {{n,l,[?MODULE,Pool,Next]},n}) of {{n,l,[?MODULE,Pool,Actual,_Name]} = Pick, _} -> case Actual - Next of Diff when Diff > 0 -> incr(Pool, Diff, Sz), ret(Pick, Ret); _ -> ret(Pick, Ret) end; _ -> case ets:next(gproc, {{n,l,[?MODULE,Pool,0]}, n}) of {{n,l,[?MODULE,Pool,Actual1,_Name1]} = Pick, _} -> incr(Pool, Sz-Next+Actual1, Sz), ret(Pick, Ret); _ -> false end end; pick(Pool, Sz, random, Ret) -> pick_near(Pool, rand:uniform(Sz), Ret). pick(Pool, Sz, hash, Val, Ret) -> pick_near(Pool, erlang:phash2(Val, Sz) + 1, Ret); pick(Pool, Sz, direct, N, Ret) when is_integer(N), N > 0 -> pick_near(Pool, case (N rem Sz-1) + 1 of 0 -> Sz; N1 -> N1 end, Ret). pick_near(Pool, N, Ret) -> case ets:next(gproc, {{n,l,[?MODULE,Pool,N]}, n}) of {{n,l,[?MODULE,Pool,_,_]} = Pick, _} -> ret(Pick, Ret); _ -> %% wrap case ets:next(gproc, {{n,l,[?MODULE,Pool,1]}, n}) of {{n,l,[?MODULE,Pool,_,_]} = Pick, _} -> ret(Pick, Ret); _ -> false end end. ret(Name, name) -> Name; ret(Name, pid) -> case ets:lookup(gproc, {Name,n}) of [{_, Pid, _}] -> Pid; [] -> %% possible race false end. %% @equiv claim(Pool, F, nowait) claim(Pool, F) when is_function(F, 2) -> claim(Pool, F, nowait). %% @spec claim(Pool, Fun, Wait) -> {true, Res} | false %% Pool = any() %% Fun = function() %% Wait = nowait | {busy_wait, integer()} %% %% @doc Picks the first available worker in the pool and applies `Fun'. %% %% A `claim' pool allows the caller to "claim" a worker during a short span %% (essentially, a lock is set and released as soon as `Fun' returns). %% Once a worker is selected, `Fun(Name, Pid)' is called, where `Name' is a %% unique gproc name of the worker, and `Pid' is its process identifier. %% The gproc name of the worker serves as a mutex, where its value is 0 (zero) %% if the worker is free, and 1 (one) if it is busy. The mutex operation is %% implemented using `gproc:update_counter/2'. %% %% `Wait == nowait' means that the call will return `false' immediately if %% there is no available worker. %% %% `Wait == {busy_wait, Timeout}' will keep repeating the claim attempt %% for `Timeout' milliseconds. If still no worker is available, it will %% return `false'. %% @end claim(Pool, F, Wait) -> case gproc:get_value(?POOL(Pool), shared) of {0, _} -> false; {_, claim} -> W = setup_wait(Wait, Pool), claim_w(Pool, F, W); _ -> error(badarg) end. claim_w(_Pool, _F, timeout) -> false; claim_w(Pool, F, W) -> case claim_(Pool, F) of false -> claim_w(Pool, F, do_wait(W)); Other -> clear_wait(W), Other end. %% Define how many workers to select in each chunk. We want to strike %% a good compromise between the cost of succeeding on the first try %% (likely a common event) and the cost of retrying. In my measurements, %% if a chunk size of 1 costs ca 30 us (on my Macbook), a chunk size of 5 %% adds only ca 20% to the cost, i.e. a few us. -define(CLAIM_CHUNK, 5). claim_(Pool, F) -> %% Sorry, but we use ets:select/3 here in order to shave off a few us. case ets:select(gproc, [{ {{{n,l,[?MODULE,Pool,'_','_']},n}, '$1', 0}, [], [{{ {element,1,{element,1,'$_'}}, '$1' }}]}], ?CLAIM_CHUNK) of {[_|_] = Workers, Cont} -> case try_claim(Workers, F) of {true, _} = True -> True; false -> claim_cont(Cont, F) end; _ -> false end. claim_cont('$end_of_table', _) -> false; claim_cont(Cont, F) -> case ets:select(Cont) of {[_|_] = Workers, Cont1} -> case try_claim(Workers, F) of {true, _} = True -> True; false -> claim_cont(Cont1, F) end; _ -> false end. try_claim([], _) -> false; try_claim([{K,Pid}|T], F) -> case try_claim(K, Pid, F) of false -> try_claim(T, F); Other -> Other end. try_claim(K, Pid, F) -> case gproc:update_counter(K, [0, {1, 1, 1}]) of [0, 1] -> %% have lock execute_claim(F, K, Pid); [1, 1] -> %% no false end. %% Wrapper to handle the case where the claimant gets killed by another %% process while executing within the critical section. %% This is likely a rare case, but if it happens, the claim would never %% get released. %% Solution: %% - spawn a monitoring process which resets the claim if the parent dies %% (spawn_link() might be more efficient, but we cannot enable trap_exit %% atomically, which introduces a race condition). %% - for all return types, kill the monitor and release the claim. %% - the one case where the monitor *isn't* killed is when Parent itself %% is killed before executing the `after' clause. In this case, it should %% be safe to release the claim from the monitoring process. %% %% Overhead in the normal case: %% - spawning the monitoring process %% - (possibly scheduling the monitoring process to set up the monitor) %% - killing the monitoring process (untrappably) %% Timing the overhead over 100,000 claims on a Core i7 MBP running OTP 17, %% this wrapper increases the cost of a minimal claim from ca 3 us to %% ca 7-8 us. execute_claim(F, K, Pid) -> Parent = self(), Mon = spawn( fun() -> Ref = erlang:monitor(process, Parent), receive {'DOWN', Ref, _, _, _} -> gproc:reset_counter(K) end end), try begin Res = F(K, Pid), {true, Res} end after exit(Mon, kill), %% Can crash when Pid dies and dissociates from K. ?MAY_FAIL(gproc:reset_counter(K)) end. setup_wait(nowait, _) -> nowait; setup_wait({busy_wait, MS}, Pool) -> Ref = erlang:start_timer(MS, self(), {claim, Pool}), {busy_wait, Ref}. do_wait(nowait) -> timeout; do_wait({busy_wait, Ref} = W) -> %% Yielding here serves two purposes: %% 1) Increase the chance that whoever's before us can finish %% 2) The value of read_timer/1 only refreshes after yield (so I've heard) erlang:yield(), case erlang:read_timer(Ref) of false -> receive {timeout, Ref, _} -> ok after 0 -> ok end, timeout; _ -> W end. clear_wait(nowait) -> ok; clear_wait({busy_wait, Ref}) -> erlang:cancel_timer(Ref), ok. %% @spec log(GprocKey) -> integer() %% @doc Update a counter associated with a worker name. %% %% Each added worker has a gproc counter that can be used e.g. to keep track of %% the number of times the worker has been picked. Since it's associated with the %% named 'slot', and not to the connected worker, its value will persist even %% if the currently connected worker dies. %% @end log({n,l,[?MODULE,Pool,_,Name]}) -> gproc:update_shared_counter(?POOL_WRK(Pool,Name), 1). %% @spec randomize(Pool::any()) -> integer() %% @doc Randomizes the "next" pointer for the pool. %% %% This function only has an effect for `round_robin' pools, which have a %% reference to the next worker to be picked. Without randomizing, the load %% balancing will always start with the first worker in the pool. %% @end randomize(Pool) -> case pool_size(Pool) of 0 -> 0; 1 -> 1; Sz -> incr(Pool, rand:uniform(Sz) - 1, Sz) end. %% @spec pool_size(Pool::any()) -> integer() %% @doc Return the size of the pool. %% pool_size(Pool) -> {Sz, _} = gproc:get_value(?POOL(Pool), shared), Sz. %% =================================================================== %% Start, stop, call gen_server %% @private start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %% @private init([]) -> {ok, #st{}}. %% @private call(Req) -> case gen_server:call(?MODULE, Req) of badarg -> error(badarg); {badarg, Reason} -> error(Reason); Reply -> Reply end. %% =================================================================== %% Gen_server callbacks %% @private handle_call(Req, From, S) -> try handle_call_(Req, From, S) catch error:Reason -> {reply, {badarg, Reason}, S} end. handle_call_({new, Pool, Type, Opts}, _, S) -> new_(Pool, Type, Opts), {reply, ok, S}; handle_call_({delete, Pool}, _, S) -> delete_(Pool), {reply, ok, S}; handle_call_({force_delete, Pool}, _, S) -> force_delete_(Pool), {reply, ok, S}; handle_call_({add_worker, Pool, Name}, _, S) -> N = add_worker_(Pool, Name), {reply, N, S}; handle_call_({add_worker, Pool, Name, Pos}, _, S) -> N = add_worker_(Pool, Name, Pos), {reply, N, S}; handle_call_({set_pool_size, Pool, Sz}, _, S) -> Workers = get_workers_(Pool), case get_last_worker_n(Workers) of N when N > Sz -> {reply, badarg, S}; _ -> set_pool_size_(?POOL(Pool), Sz, Workers), {reply, true, S} end; handle_call_({remove_worker, Pool, Name}, _, S) -> ok = remove_worker_(Pool, Name), {reply, true, S}. %% @private handle_cast(_, S) -> {noreply, S}. %% @private handle_info(_, S) -> {noreply, S}. %% @private terminate(_, _) -> ok. %% @private code_change(_, S, _) -> {ok, S}. %% =================================================================== %% Internal functions new_(Pool, Type, Opts) -> valid_type(Type), Size = proplists:get_value(size, Opts, 0), Workers = lists:seq(1, Size), K = ?POOL(Pool), try gproc:reg_shared(K, {Size, Type}) catch error:_ -> error(exists) end, Opts1 = case lists:keyfind(auto_size, 1, Opts) of false -> Opts ++ [{auto_size, not lists:keymember(size, 1, Opts)}]; {_, Bool} when is_boolean(Bool) -> Opts end, gproc:set_attributes_shared(K, Opts1), set_workers(K, Workers), gproc:reg_shared(?POOL_CUR(Pool), Size). valid_type(T) when T==round_robin; T==hash; T==random; T==direct; T==claim -> true; valid_type(_) -> error(invalid_type). set_pool_size_(K, Sz, Workers) -> {_, Type} = gproc:get_value(K, shared), case length(Workers) of Sz -> set_workers(K, Workers); Len when Len > Sz -> Workers1 = lists:sublist(Workers, 1, Sz), set_workers(K, Workers1); Len when Len < Sz -> Workers1 = Workers ++ lists:seq(Len+1, Sz), set_workers(K, Workers1) end, gproc:set_value_shared(K, {Sz, Type}). delete_(Pool) -> K = ?POOL(Pool), Ws = get_workers_(K), case [1 || {_,_} <- Ws] of [] -> gproc:unreg_shared(K), gproc:unreg_shared(?POOL_CUR(Pool)); [_|_] -> error(not_empty) end. force_delete_(Pool) -> Props = gproc:select({l,p}, [{ {?POOL(Pool), '_', '_'}, [], ['$_']}]), Cur = gproc:select({l,c}, [{ {?POOL_CUR(Pool), '_', '_'}, [], ['$_']}]), Workers = gproc:select( {l,c}, [{ {?POOL_WRK(Pool,'_'), '_', '_'}, [], ['$_']}]), Names = find_names(Pool, '_'), lists:foreach( fun({Key, Pid, _}) when Pid == self() -> gproc:unreg(Key); ({_, Pid, _}) when is_pid(Pid) -> exit(Pid, kill) end, Names), [gproc:unreg_shared(W) || {W,shared,_} <- Cur ++ Props ++ Workers], true. find_names(Pool, Pid) -> gproc:select( {l,n}, [{ {{n,l,[?MODULE,Pool,Pid,'_']}, '_', '_'}, [], ['$_']}]). add_worker_(Pool, Name) -> K = ?POOL(Pool), {Sz, Type} = gproc:get_value(K, shared), AutoSz = gproc:get_attribute(K, shared, auto_size), Ws0 = get_workers_(K), {N,Ws1} = case lists:keymember(Name, 1, Ws0) of false -> case find_slot(Name, K, Ws0, Sz, Type, AutoSz) of {_, _} = Res -> Res; false -> error(pool_full) end; true -> error(exists) end, if N > Sz -> set_pool_size_(K, N, Ws1); % also calls set_workers/2 true -> %% size not changed set_workers(K, Ws1) end, reg_worker(Pool, Name, N), N. add_worker_(Pool, Name, Pos) -> K = ?POOL(Pool), {Sz, _} = gproc:get_value(K, shared), Ws0 = get_workers_(K), if Pos > Sz -> case gproc:get_attribute(K, shared, auto_size) of true -> Ws1 = Ws0 ++ lists:seq(Sz+1,Pos-1) ++ [{Name, Pos}], set_pool_size_(K, Pos, Ws1); false -> error(out_of_range) end; true -> case lists:nth(Pos, Ws0) of {_,_} -> error(exists); P when is_integer(P) -> Ws1 = set_pos(Pos, Ws0, {Name, Pos}), set_workers(K, Ws1) end end, reg_worker(Pool, Name, Pos), Pos. reg_worker(Pool, Name, Pos) -> gproc:reg_shared(Wrk = ?POOL_WRK(Pool, Name), 0), gproc:set_attributes_shared(Wrk, [{n, Pos}]). remove_worker_(Pool, Name) -> case whereis_worker(Pool, Name) of Pid when is_pid(Pid) -> error({worker_connected, Pid}); undefined -> do_remove_worker_(Pool, Name) end. do_remove_worker_(Pool, Name) -> K = ?POOL(Pool), AutoSize = gproc:get_attribute(K, shared, auto_size), Ws0 = get_workers_(K), Ws1 = del_slot(Name, Ws0, AutoSize), gproc:unreg_shared(?POOL_WRK(Pool, Name)), case AutoSize of false -> ok; true -> case (NewLen = length(Ws1)) - length(Ws0) of 0 -> ok; Diff when Diff < 0 -> {_, Type} = gproc:get_value(K, shared), gproc:set_value_shared(K, {NewLen, Type}) end end, gproc:set_attributes_shared(K, [{workers, Ws1}]), ok. del_slot(Name, [{Name,_}], true) -> []; del_slot(Name, [{Name, Pos}|T], _AutoSize) -> [Pos|T]; del_slot(Name, [H|T], AutoSize) -> [H|del_slot(Name, T, AutoSize)]. find_slot(Name, _, [], Sz, _, Auto) -> case {Sz, Auto} of {0, false} -> false; {_, _} -> {1, [{Name, 1}]} end; find_slot(Name, Key, Workers, Sz, Type, AutoSz) -> case get_strategy(Key, Type) of packed -> find_slot_packed(Name, Workers, AutoSz); sparse -> find_slot_sparse(Name, Workers, Sz, AutoSz) end. %% find_slot(Name, Key, Workers, Sz, Type, AutoSz, Strategy). %% find_slot(Name, []) -> %% {1, [{Name, 1}]}; %% find_slot(Name, Slots) -> %% find_slot(Name, Slots, []). get_last_worker_n(Ws) -> get_last_worker_n(Ws, 0, 1). get_last_worker_n([{_,_}|T], _, P) -> get_last_worker_n(T, P, P+1); get_last_worker_n([H|T], Last, P) when is_integer(H) -> get_last_worker_n(T, Last, P+1); get_last_worker_n([], Last, _) -> Last. find_slot_packed(Name, Workers, AutoSz) -> find_slot_packed(Name, Workers, AutoSz, []). find_slot_packed(Name, [N|T], _, Acc) when is_integer(N) -> % empty slot {N, lists:reverse(Acc) ++ [{Name, N}|T]}; find_slot_packed(Name, [{_,Prev} = Last], true, Acc) -> % last elem; expand pool New = Prev+1, {New, lists:reverse([{Name, New}, Last|Acc])}; find_slot_packed(_, [_], false, _) -> false; find_slot_packed(Name, [{_,_} = H|T], Auto, Acc) -> find_slot_packed(Name, T, Auto, [H|Acc]). find_slot_sparse(Name, Ws, Sz, Auto) -> %% Collect the position of the first and last filled slots, as well as %% the largest gap between filled slots case lists:foldl( fun(N, {Prev, StartP, First, Last, Max, MaxP}) when is_integer(N) -> case Prev+1 of Gap when Gap > Max -> {Gap, StartP, First, Last, Gap, StartP}; Gap -> {Gap, StartP, First, Last, Max, MaxP} end; (N, []) when is_integer(N) -> %% skip []; ({_, Pos}, []) -> {0, Pos, _First = Pos, _Last = Pos, 0, 0}; ({_, Pos}, {Prev, StartP, First, _PrevLast, Max, MaxP}) -> if Prev > Max -> {0, Pos, First, Pos, Prev, StartP}; true -> {0, Pos, First, Pos, Max, MaxP} end end, [], Ws) of [] -> %% all empty slots case {Sz, Auto} of {0, false} -> false; {0, true} -> {1, [{Name, 1}]}; {_, _} when is_integer(Sz), Sz > 0 -> {1, [{Name, 1}|tl(Ws)]} end; {_, _, 1, Last, 0, _} -> %% Pool full if Auto -> NewPos = Last + 1, {NewPos, Ws ++ [{Name, NewPos}]}; true -> false end; {_, _, First, Last, MaxGap, StartPos} -> WrapGap = (Sz - Last) + First - 1, NewPos = if WrapGap >= MaxGap -> (Last + (WrapGap div 2) + 1) rem (Sz+1); true -> (StartPos + (MaxGap div 2) + 1) rem (Sz+1) end, {NewPos, set_pos(NewPos, Ws, {Name, NewPos})} end. set_pos(P, L, X) when P > 0, is_list(L) -> set_pos(P, 1, L, X). set_pos(P, P, [_|T], X) -> [X|T]; set_pos(P, C, [H|T], X) when C < P -> [H|set_pos(P, C+1, T, X)]. get_workers_(K) -> case gproc:get_attribute(K, shared, workers) of undefined -> []; L when is_list(L) -> L end. set_workers(K, L) when is_list(L) -> gproc:set_attributes_shared(K, [{workers, L}]). get_strategy(Key, Type) -> Default = case Type of round_robin -> packed; random -> sparse; hash -> sparse; direct -> packed; claim -> packed end, attribute(Key, fill_strategy, Default). attribute(Key, A, Default) -> case gproc:get_attribute(Key, shared, A) of undefined -> Default; Value -> Value end. incr(Pool, Incr, Sz) -> gproc:update_counter(?POOL_CUR(Pool), shared, {Incr, Sz, 1}). %% find_worker(Pool, Name) -> %% case gproc:select(n, [{ {{n, l, {?MODULE, Pool, '_'}}, '_', Name}, %% [], ['$_'] }]) of %% [] -> %% undefined; %% [{{n,l,{?MODULE,_,N}}, Pid, _}] -> %% {N, Pid} %% end. %% ============================= Test code =========================== %% @private test(N) when N > 0 -> test(N, round_robin, []). %% @private test(N, Type, Opts) when Type==round_robin; Type==random; Type==hash; Type==direct; Type==claim -> P = ?LINE, setup_test_pool(P, Type, Opts), try timer:tc(?MODULE, f(Type), [N, P]) after remove_test_pool(P) end. ptest(N, I, Type, Opts) -> P = ?LINE, setup_test_pool(P, Type, Opts), F = f(Type), Pids = [spawn_monitor(fun() -> exit({ok, timer:tc(?MODULE, F, [I, P])}) end) || _ <- lists:seq(1, N)], try collect(Pids) after remove_test_pool(P) end. collect(Pids) -> Results = [receive {'DOWN', Ref, _, _, Reason} -> Reason end || {_, Ref} <- Pids], {Times, Avgs} = lists:foldr(fun({ok, {T, Avg}}, {A,B}) -> {[T|A], [Avg|B]} end, {[],[]}, Results), {Times, lists:sum(Times)/length(Times), lists:sum(Avgs)/length(Avgs)}. f(Type) when Type==hash; Type==direct -> test_run1; f(Type) when Type==claim -> test_run2; f({empty,_}) -> test_run0; f(_) -> test_run. %% @private setup_test_pool(P, Type, Opts) -> setup_test_pool(P, Type, Opts, test_workers()). setup_test_pool(P, Type0, Opts, Workers) -> Type = case Type0 of {_, T} -> T; T when is_atom(T) -> T end, new(P, Type, Opts), [begin R = add_worker(P, W), io:fwrite("add_worker(~p, ~p) -> ~p; Ws = ~p~n", [P, W, R, get_workers_(?POOL(P))]), connect_worker(P, W) end || W <- Workers]. %% @private remove_test_pool(P) -> io:fwrite("worker stats (~p):~n" "~p~n", [P, gproc:select( {l,c}, [{ {{c,l,{?MODULE,P,w,'$1'}},'_','$2'}, [], [{{'$1','$2'}}] }])]), [begin disconnect_worker(P, W), remove_worker(P, W) end || W <- test_workers()], delete(P). test_workers() -> [a,b,c,d,e,f]. %% @private test_run(N, P) -> test_run(N, P, 0, 0). test_run(N, P, S, M) when N > 0 -> {T, Worker} = timer:tc(?MODULE, pick, [P]), true = (Worker =/= false), log(Worker), timer:sleep(rand:uniform(50)), test_run(N-1, P, S+T, M+1); test_run(_, _, S, M) -> S/M. %% @private test_run1(N, P) -> test_run1(N, P, 0, 0). test_run1(N, P, S, M) when N > 0 -> {T, Worker} = timer:tc(?MODULE, pick, [P, N]), true = (Worker =/= false), log(Worker), timer:sleep(rand:uniform(50)), test_run1(N-1, P, S+T, M+1); test_run1(_, _, S, M) -> S/M. %% @private test_run2(N, P) -> test_run2(N, P, fun(K,_) -> R = log(K), timer:sleep(rand:uniform(50)), R end, 0, 0). test_run2(N, P, F, S, M) when N > 0 -> {T, {true, _}} = timer:tc(?MODULE, claim, [P, F, {busy_wait, 5000}]), test_run2(N-1, P, F, S+T, M+1); test_run2(_, _, _, S, M) -> S/M. test_run0(N, X) when N > 0 -> test_run0(N-1, X); test_run0(_, _) -> ok.