pooler.erl 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656
  1. %% @author Seth Falcon <seth@userprimary.net>
  2. %% @copyright 2011-2013 Seth Falcon
  3. %% @doc This is the main interface to the pooler application
  4. %%
  5. %% To integrate with your application, you probably want to call
  6. %% application:start(pooler) after having specified appropriate
  7. %% configuration for the pooler application (either via a config file
  8. %% or appropriate calls to the application module to set the
  9. %% application's config).
  10. %%
  11. -module(pooler).
  12. -behaviour(gen_server).
  13. -include("pooler.hrl").
  14. -include_lib("eunit/include/eunit.hrl").
  15. %% type specs for pool metrics
  16. -type metric_value() :: 'unknown_pid' |
  17. non_neg_integer() |
  18. {'add_pids_failed', non_neg_integer(), non_neg_integer()} |
  19. {'inc',1} |
  20. 'error_no_members'.
  21. -type metric_type() :: 'counter' | 'histogram' | 'history' | 'meter'.
  22. %% ------------------------------------------------------------------
  23. %% API Function Exports
  24. %% ------------------------------------------------------------------
  25. -export([accept_member/2,
  26. start_link/1,
  27. take_member/1,
  28. take_group_member/1,
  29. return_group_member/2,
  30. return_group_member/3,
  31. return_member/2,
  32. return_member/3,
  33. pool_stats/1,
  34. manual_start/0,
  35. new_pool/1,
  36. rm_pool/1]).
  37. %% ------------------------------------------------------------------
  38. %% gen_server Function Exports
  39. %% ------------------------------------------------------------------
  40. -export([init/1,
  41. handle_call/3,
  42. handle_cast/2,
  43. handle_info/2,
  44. terminate/2,
  45. code_change/3]).
  46. %% To help with testing internal functions
  47. -ifdef(TEST).
  48. -compile([export_all]).
  49. -endif.
  50. %% ------------------------------------------------------------------
  51. %% API Function Definitions
  52. %% ------------------------------------------------------------------
  53. start_link(#pool{name = Name} = Pool) ->
  54. gen_server:start_link({local, Name}, ?MODULE, Pool, []).
  55. manual_start() ->
  56. application:start(sasl),
  57. application:start(pooler).
  58. %% @doc Start a new pool described by the proplist `PoolConfig'. The
  59. %% following keys are required in the proplist:
  60. %%
  61. %% <dl>
  62. %% <dt>`name'</dt>
  63. %% <dd>An atom giving the name of the pool.</dd>
  64. %% <dt>`init_count'</dt>
  65. %% <dd>Number of members to add to the pool at start. When the pool is
  66. %% started, `init_count' members will be started in parallel.</dd>
  67. %% <dt>`max_count'</dt>
  68. %% <dd>Maximum number of members in the pool.</dd>
  69. %% <dt>`start_mfa'</dt>
  70. %% <dd>A tuple of the form `{Mod, Fun, Args}' describing how to start
  71. %% new pool members.</dd>
  72. %% </dl>
  73. %%
  74. %% In addition, you can specify any of the following optional
  75. %% configuration options:
  76. %%
  77. %% <dl>
  78. %% <dt>`group'</dt>
  79. %% <dd>An atom giving the name of the group this pool belongs
  80. %% to. Pools sharing a common `group' value can be accessed using
  81. %% {@link take_group_member/1} and {@link return_group_member/2}.</dd>
  82. %% <dt>`cull_interval'</dt>
  83. %% <dd>Time between checks for stale pool members. Specified as
  84. %% `{Time, Unit}' where `Time' is a non-negative integer and `Unit'
  85. %% is one of `min', `sec', `ms', or `mu'. The default value of `{0,
  86. %% min}' disables stale member checking. When `Time' is greater than
  87. %% zero, a message will be sent to the pool at the configured interval
  88. %% to trigger the removal of members that have not been accessed in
  89. %% `max_age' time units.</dd>
  90. %% <dt>`max_age'</dt>
  91. %% <dd>Members idle longer than `max_age' time units are removed from
  92. %% the pool when stale checking is enabled via
  93. %% `cull_interval'. Culling of idle members will never reduce the pool
  94. %% below `init_count'. The value is specified as `{Time, Unit}'. Note
  95. %% that timers are not set on individual pool members and may remain
  96. %% in the pool beyond the configured `max_age' value since members are
  97. %% only removed on the interval configured via `cull_interval'.</dd>
  98. %% </dl>
  99. new_pool(PoolConfig) ->
  100. pooler_sup:new_pool(PoolConfig).
  101. %% @doc Terminate the named pool.
  102. rm_pool(PoolName) ->
  103. pooler_sup:rm_pool(PoolName).
  104. %% @doc For INTERNAL use. Adds `MemberPid' to the pool.
  105. -spec accept_member(atom() | pid(), pid() | {noproc, _}) -> ok.
  106. accept_member(PoolName, MemberPid) ->
  107. gen_server:call(PoolName, {accept_member, MemberPid}).
  108. %% @doc Obtain exclusive access to a member from `PoolName'.
  109. %%
  110. %% If no free members are available, 'error_no_members' is returned.
  111. %%
  112. -spec take_member(atom() | pid()) -> pid() | error_no_members.
  113. take_member(PoolName) when is_atom(PoolName) orelse is_pid(PoolName) ->
  114. gen_server:call(PoolName, take_member, infinity).
  115. %% @doc Take a member from a randomly selected member of the group
  116. %% `GroupName'. Returns `MemberPid' or `error_no_members'. If no
  117. %% members are available in the randomly chosen pool, all other pools
  118. %% in the group are tried in order.
  119. -spec take_group_member(atom()) -> pid() | error_no_members | {error_no_group, atom()}.
  120. take_group_member(GroupName) ->
  121. case pg2:get_local_members(GroupName) of
  122. {error, {no_such_group, GroupName}} ->
  123. {error_no_group, GroupName};
  124. [] ->
  125. error_no_members;
  126. Pools ->
  127. %% Put a random member at the front of the list and then
  128. %% return the first member you can walking the list.
  129. {_, _, X} = erlang:now(),
  130. Idx = (X rem length(Pools)) + 1,
  131. {PoolPid, Rest} = extract_nth(Idx, Pools),
  132. take_first_pool([PoolPid | Rest])
  133. end.
  134. take_first_pool([PoolPid | Rest]) ->
  135. case take_member(PoolPid) of
  136. error_no_members ->
  137. take_first_pool(Rest);
  138. Member ->
  139. ets:insert(?POOLER_GROUP_TABLE, {Member, PoolPid}),
  140. Member
  141. end;
  142. take_first_pool([]) ->
  143. error_no_members.
  144. %% this helper function returns `{Nth_Elt, Rest}' where `Nth_Elt' is
  145. %% the nth element of `L' and `Rest' is `L -- [Nth_Elt]'.
  146. extract_nth(N, L) ->
  147. extract_nth(N, L, []).
  148. extract_nth(1, [H | T], Acc) ->
  149. {H, Acc ++ T};
  150. extract_nth(N, [H | T], Acc) ->
  151. extract_nth(N - 1, T, [H | Acc]);
  152. extract_nth(_, [], _) ->
  153. error(badarg).
  154. %% @doc Return a member that was taken from the group
  155. %% `GroupName'. This is a convenience function for
  156. %% `return_group_member/3' with `Status' of `ok'.
  157. -spec return_group_member(atom(), pid() | error_no_members) -> ok.
  158. return_group_member(GroupName, MemberPid) ->
  159. return_group_member(GroupName, MemberPid, ok).
  160. %% @doc Return a member that was taken from the group `GroupName'. If
  161. %% `Status' is `ok' the member is returned to the pool from which is
  162. %% came. If `Status' is `fail' the member will be terminated and a new
  163. %% member added to the appropriate pool.
  164. -spec return_group_member(atom(), pid() | error_no_members, ok | fail) -> ok.
  165. return_group_member(_, error_no_members, _) ->
  166. ok;
  167. return_group_member(_GroupName, MemberPid, Status) ->
  168. case ets:lookup(?POOLER_GROUP_TABLE, MemberPid) of
  169. [{MemberPid, PoolPid}] ->
  170. return_member(PoolPid, MemberPid, Status);
  171. [] ->
  172. ok
  173. end.
  174. %% @doc Return a member to the pool so it can be reused.
  175. %%
  176. %% If `Status' is 'ok', the member is returned to the pool. If
  177. %% `Status' is 'fail', the member is destroyed and a new member is
  178. %% added to the pool in its place.
  179. -spec return_member(atom() | pid(), pid() | error_no_members, ok | fail) -> ok.
  180. return_member(PoolName, Pid, Status) when is_pid(Pid) andalso
  181. (is_atom(PoolName) orelse
  182. is_pid(PoolName)) andalso
  183. (Status =:= ok orelse
  184. Status =:= fail) ->
  185. gen_server:call(PoolName, {return_member, Pid, Status}, infinity),
  186. ok;
  187. return_member(_, error_no_members, _) ->
  188. ok.
  189. %% @doc Return a member to the pool so it can be reused.
  190. %%
  191. -spec return_member(atom() | pid(), pid() | error_no_members) -> ok.
  192. return_member(PoolName, Pid) when is_pid(Pid) andalso
  193. (is_atom(PoolName) orelse is_pid(PoolName)) ->
  194. gen_server:call(PoolName, {return_member, Pid, ok}, infinity),
  195. ok;
  196. return_member(_, error_no_members) ->
  197. ok.
  198. %% @doc Obtain runtime state info for all pools.
  199. %%
  200. %% Format of the return value is subject to change.
  201. -spec pool_stats(atom() | pid()) -> [tuple()].
  202. pool_stats(PoolName) ->
  203. gen_server:call(PoolName, pool_stats).
  204. %% ------------------------------------------------------------------
  205. %% gen_server Function Definitions
  206. %% ------------------------------------------------------------------
  207. -spec init(#pool{}) -> {'ok', #pool{}, 0}.
  208. init(#pool{}=Pool) ->
  209. #pool{init_count = N} = Pool,
  210. MemberSup = pooler_pool_sup:member_sup_name(Pool),
  211. Pool1 = set_member_sup(Pool, MemberSup),
  212. %% This schedules the next cull when the pool is configured for
  213. %% such and is otherwise a no-op.
  214. Pool2 = cull_members_from_pool(Pool1),
  215. {ok, NewPool} = init_members_sync(N, Pool2),
  216. %% trigger an immediate timeout, handled by handle_info to allow
  217. %% us to register with pg2. We use the timeout mechanism to ensure
  218. %% that a server is added to a group only when it is ready to
  219. %% process messages.
  220. {ok, NewPool, 0}.
  221. set_member_sup(#pool{} = Pool, MemberSup) ->
  222. Pool#pool{member_sup = MemberSup}.
  223. handle_call(take_member, {CPid, _Tag}, #pool{} = Pool) ->
  224. {Member, NewPool} = take_member_from_pool(Pool, CPid),
  225. {reply, Member, NewPool};
  226. handle_call({return_member, Pid, Status}, {_CPid, _Tag}, Pool) ->
  227. {reply, ok, do_return_member(Pid, Status, Pool)};
  228. handle_call({accept_member, Pid}, _From, Pool) ->
  229. {reply, ok, do_accept_member(Pid, Pool)};
  230. handle_call(stop, _From, Pool) ->
  231. {stop, normal, stop_ok, Pool};
  232. handle_call(pool_stats, _From, Pool) ->
  233. {reply, dict:to_list(Pool#pool.all_members), Pool};
  234. handle_call(dump_pool, _From, Pool) ->
  235. {reply, Pool, Pool};
  236. handle_call(_Request, _From, Pool) ->
  237. {noreply, Pool}.
  238. -spec handle_cast(_,_) -> {'noreply', _}.
  239. handle_cast(_Msg, Pool) ->
  240. {noreply, Pool}.
  241. -spec handle_info(_, _) -> {'noreply', _}.
  242. handle_info(timeout, #pool{group = undefined} = Pool) ->
  243. %% ignore
  244. {noreply, Pool};
  245. handle_info(timeout, #pool{group = Group} = Pool) ->
  246. ok = pg2:create(Group),
  247. ok = pg2:join(Group, self()),
  248. {noreply, Pool};
  249. handle_info({'DOWN', MRef, process, Pid, Reason}, State) ->
  250. State1 =
  251. case dict:find(Pid, State#pool.all_members) of
  252. {ok, {_PoolName, _ConsumerPid, _Time}} ->
  253. do_return_member(Pid, fail, State);
  254. error ->
  255. case dict:find(Pid, State#pool.consumer_to_pid) of
  256. {ok, {MRef, Pids}} ->
  257. IsOk = case Reason of
  258. normal -> ok;
  259. _Crash -> fail
  260. end,
  261. lists:foldl(
  262. fun(P, S) -> do_return_member(P, IsOk, S) end,
  263. State, Pids);
  264. error ->
  265. State
  266. end
  267. end,
  268. {noreply, State1};
  269. handle_info(cull_pool, Pool) ->
  270. {noreply, cull_members_from_pool(Pool)};
  271. handle_info(_Info, State) ->
  272. {noreply, State}.
  273. -spec terminate(_, _) -> 'ok'.
  274. terminate(_Reason, _State) ->
  275. ok.
  276. -spec code_change(_, _, _) -> {'ok', _}.
  277. code_change(_OldVsn, State, _Extra) ->
  278. {ok, State}.
  279. %% ------------------------------------------------------------------
  280. %% Internal Function Definitions
  281. %% ------------------------------------------------------------------
  282. do_accept_member({Ref, Pid},
  283. #pool{
  284. all_members = AllMembers,
  285. free_pids = Free,
  286. free_count = NumFree,
  287. starting_members = StartingMembers0
  288. } = Pool) when is_pid(Pid) ->
  289. %% make sure we don't accept a timedout member
  290. StartingMembers = remove_stale_starting_members(Pool, StartingMembers0,
  291. ?DEFAULT_MEMBER_START_TIMEOUT),
  292. case lists:keymember(Ref, 1, StartingMembers) of
  293. false ->
  294. %% a pid we didn't ask to start, ignore it.
  295. %% should we log it?
  296. Pool;
  297. true ->
  298. StartingMembers1 = lists:keydelete(Ref, 1, StartingMembers),
  299. MRef = erlang:monitor(process, Pid),
  300. Entry = {MRef, free, os:timestamp()},
  301. AllMembers1 = store_all_members(Pid, Entry, AllMembers),
  302. Pool#pool{free_pids = Free ++ [Pid],
  303. free_count = NumFree + 1,
  304. all_members = AllMembers1,
  305. starting_members = StartingMembers1}
  306. end;
  307. do_accept_member({Ref, _Reason}, #pool{starting_members = StartingMembers0} = Pool) ->
  308. %% member start failed, remove in-flight ref and carry on.
  309. StartingMembers = remove_stale_starting_members(Pool, StartingMembers0,
  310. ?DEFAULT_MEMBER_START_TIMEOUT),
  311. StartingMembers1 = lists:keydelete(Ref, 1, StartingMembers),
  312. Pool#pool{starting_members = StartingMembers1}.
  313. -spec remove_stale_starting_members(#pool{}, [{reference(), erlang:timestamp()}],
  314. time_spec()) -> [{reference(), erlang:timestamp()}].
  315. remove_stale_starting_members(Pool, StartingMembers, MaxAge) ->
  316. Now = os:timestamp(),
  317. MaxAgeSecs = time_as_secs(MaxAge),
  318. lists:filter(fun(SM) ->
  319. starting_member_not_stale(Pool, Now, SM, MaxAgeSecs)
  320. end, StartingMembers).
  321. starting_member_not_stale(Pool, Now, {_Ref, StartTime}, MaxAgeSecs) ->
  322. case secs_between(StartTime, Now) < MaxAgeSecs of
  323. true ->
  324. true;
  325. false ->
  326. error_logger:error_msg("pool '~s': starting member timeout", [Pool#pool.name]),
  327. send_metric(Pool, starting_member_timeout, {inc, 1}, counter),
  328. false
  329. end.
  330. init_members_sync(N, #pool{name = PoolName} = Pool) ->
  331. Self = self(),
  332. StartTime = os:timestamp(),
  333. StartRefs = [ {pooler_starter:start_member(Pool, Self), StartTime}
  334. || _I <- lists:seq(1, N) ],
  335. Pool1 = Pool#pool{starting_members = StartRefs},
  336. case collect_init_members(Pool1) of
  337. timeout ->
  338. error_logger:error_msg("pool '~s': exceeded timeout waiting for ~B members",
  339. [PoolName, Pool1#pool.init_count]),
  340. error({timeout, "unable to start members"});
  341. #pool{} = Pool2 ->
  342. {ok, Pool2}
  343. end.
  344. collect_init_members(#pool{starting_members = []} = Pool) ->
  345. Pool;
  346. collect_init_members(#pool{} = Pool) ->
  347. Timeout = time_as_millis(?DEFAULT_MEMBER_START_TIMEOUT),
  348. receive
  349. {accept_member, {Ref, Member}} ->
  350. collect_init_members(do_accept_member({Ref, Member}, Pool))
  351. after
  352. Timeout ->
  353. timeout
  354. end.
  355. -spec take_member_from_pool(#pool{}, {pid(), term()}) ->
  356. {error_no_members | pid(), #pool{}}.
  357. take_member_from_pool(#pool{init_count = InitCount,
  358. max_count = Max,
  359. free_pids = Free,
  360. in_use_count = NumInUse,
  361. free_count = NumFree,
  362. consumer_to_pid = CPMap,
  363. starting_members = StartingMembers0} = Pool,
  364. From) ->
  365. send_metric(Pool, take_rate, 1, meter),
  366. StartingMembers = remove_stale_starting_members(Pool, StartingMembers0,
  367. ?DEFAULT_MEMBER_START_TIMEOUT),
  368. NumCanAdd = Max - (NumInUse + NumFree + length(StartingMembers)),
  369. case Free of
  370. [] when NumCanAdd =< 0 ->
  371. send_metric(Pool, error_no_members_count, {inc, 1}, counter),
  372. send_metric(Pool, events, error_no_members, history),
  373. {error_no_members, Pool};
  374. [] when NumCanAdd > 0 ->
  375. %% Limit concurrently starting members to init_count. Add
  376. %% up to init_count members. Starting members here means
  377. %% we always return an error_no_members for a take request
  378. %% when all members are in-use. By adding a batch of new
  379. %% members, the pool should reach a steady state with
  380. %% unused members culled over time (if scheduled cull is
  381. %% enabled).
  382. NumToAdd = min(InitCount - length(StartingMembers), NumCanAdd),
  383. Pool1 = add_members_async(NumToAdd, Pool),
  384. send_metric(Pool, error_no_members_count, {inc, 1}, counter),
  385. send_metric(Pool, events, error_no_members, history),
  386. {error_no_members, Pool1};
  387. [Pid|Rest] ->
  388. Pool1 = Pool#pool{free_pids = Rest, in_use_count = NumInUse + 1,
  389. free_count = NumFree - 1},
  390. send_metric(Pool, in_use_count, Pool1#pool.in_use_count, histogram),
  391. send_metric(Pool, free_count, Pool1#pool.free_count, histogram),
  392. {Pid, Pool1#pool{
  393. consumer_to_pid = add_member_to_consumer(Pid, From, CPMap),
  394. all_members = set_cpid_for_member(Pid, From,
  395. Pool1#pool.all_members)
  396. }}
  397. end.
  398. %% @doc Add `Count' members to `Pool' asynchronously. Returns updated
  399. %% `Pool' record with starting member refs added to field
  400. %% `starting_members'.
  401. add_members_async(Count, #pool{starting_members = StartingMembers} = Pool) ->
  402. StartTime = os:timestamp(),
  403. StartRefs = [ {pooler_starter:start_member(Pool), StartTime}
  404. || _I <- lists:seq(1, Count) ],
  405. Pool#pool{starting_members = StartRefs ++ StartingMembers}.
  406. -spec do_return_member(pid(), ok | fail, #pool{}) -> #pool{}.
  407. do_return_member(Pid, ok, #pool{all_members = AllMembers} = Pool) ->
  408. clean_group_table(Pid, Pool),
  409. case dict:find(Pid, AllMembers) of
  410. {ok, {MRef, CPid, _}} ->
  411. #pool{free_pids = Free, in_use_count = NumInUse,
  412. free_count = NumFree} = Pool,
  413. Pool1 = Pool#pool{free_pids = [Pid | Free], in_use_count = NumInUse - 1,
  414. free_count = NumFree + 1},
  415. Entry = {MRef, free, os:timestamp()},
  416. Pool1#pool{all_members = store_all_members(Pid, Entry, AllMembers),
  417. consumer_to_pid = cpmap_remove(Pid, CPid,
  418. Pool1#pool.consumer_to_pid)};
  419. error ->
  420. Pool
  421. end;
  422. do_return_member(Pid, fail, #pool{all_members = AllMembers} = Pool) ->
  423. % for the fail case, perhaps the member crashed and was alerady
  424. % removed, so use find instead of fetch and ignore missing.
  425. clean_group_table(Pid, Pool),
  426. case dict:find(Pid, AllMembers) of
  427. {ok, {_MRef, _, _}} ->
  428. Pool1 = remove_pid(Pid, Pool),
  429. add_members_async(1, Pool1);
  430. error ->
  431. Pool
  432. end.
  433. clean_group_table(_MemberPid, #pool{group = undefined}) ->
  434. ok;
  435. clean_group_table(MemberPid, #pool{group = _GroupName}) ->
  436. ets:delete(?POOLER_GROUP_TABLE, MemberPid).
  437. % @doc Remove `Pid' from the pid list associated with `CPid' in the
  438. % consumer to member map given by `CPMap'.
  439. %
  440. % If `Pid' is the last element in `CPid's pid list, then the `CPid'
  441. % entry is removed entirely.
  442. %
  443. -spec cpmap_remove(pid(), pid() | free, dict()) -> dict().
  444. cpmap_remove(_Pid, free, CPMap) ->
  445. CPMap;
  446. cpmap_remove(Pid, CPid, CPMap) ->
  447. case dict:find(CPid, CPMap) of
  448. {ok, {MRef, Pids0}} ->
  449. Pids1 = lists:delete(Pid, Pids0),
  450. case Pids1 of
  451. [_H|_T] ->
  452. dict:store(CPid, {MRef, Pids1}, CPMap);
  453. [] ->
  454. %% no more members for this consumer
  455. erlang:demonitor(MRef),
  456. dict:erase(CPid, CPMap)
  457. end;
  458. error ->
  459. % FIXME: this shouldn't happen, should we log or error?
  460. CPMap
  461. end.
  462. % @doc Remove and kill a pool member.
  463. %
  464. % Handles in-use and free members. Logs an error if the pid is not
  465. % tracked in state.all_members.
  466. %
  467. -spec remove_pid(pid(), #pool{}) -> #pool{}.
  468. remove_pid(Pid, Pool) ->
  469. #pool{name = PoolName,
  470. all_members = AllMembers,
  471. consumer_to_pid = CPMap} = Pool,
  472. case dict:find(Pid, AllMembers) of
  473. {ok, {MRef, free, _Time}} ->
  474. % remove an unused member
  475. erlang:demonitor(MRef),
  476. FreePids = lists:delete(Pid, Pool#pool.free_pids),
  477. NumFree = Pool#pool.free_count - 1,
  478. Pool1 = Pool#pool{free_pids = FreePids, free_count = NumFree},
  479. exit(Pid, kill),
  480. send_metric(Pool1, killed_free_count, {inc, 1}, counter),
  481. Pool1#pool{all_members = dict:erase(Pid, AllMembers)};
  482. {ok, {MRef, CPid, _Time}} ->
  483. %% remove a member being consumed. No notice is sent to
  484. %% the consumer.
  485. erlang:demonitor(MRef),
  486. Pool1 = Pool#pool{in_use_count = Pool#pool.in_use_count - 1},
  487. exit(Pid, kill),
  488. send_metric(Pool1, killed_in_use_count, {inc, 1}, counter),
  489. Pool1#pool{consumer_to_pid = cpmap_remove(Pid, CPid, CPMap),
  490. all_members = dict:erase(Pid, AllMembers)};
  491. error ->
  492. error_logger:error_report({{pool, PoolName}, unknown_pid, Pid,
  493. erlang:get_stacktrace()}),
  494. send_metric(Pool, events, unknown_pid, history),
  495. Pool
  496. end.
  497. -spec store_all_members(pid(),
  498. {reference(), free | pid(), {_, _, _}}, dict()) -> dict().
  499. store_all_members(Pid, Val = {_MRef, _CPid, _Time}, AllMembers) ->
  500. dict:store(Pid, Val, AllMembers).
  501. -spec set_cpid_for_member(pid(), pid(), dict()) -> dict().
  502. set_cpid_for_member(MemberPid, CPid, AllMembers) ->
  503. dict:update(MemberPid,
  504. fun({MRef, free, Time = {_, _, _}}) ->
  505. {MRef, CPid, Time}
  506. end, AllMembers).
  507. -spec add_member_to_consumer(pid(), pid(), dict()) -> dict().
  508. add_member_to_consumer(MemberPid, CPid, CPMap) ->
  509. %% we can't use dict:update here because we need to create the
  510. %% monitor if we aren't already tracking this consumer.
  511. case dict:find(CPid, CPMap) of
  512. {ok, {MRef, MList}} ->
  513. dict:store(CPid, {MRef, [MemberPid | MList]}, CPMap);
  514. error ->
  515. MRef = erlang:monitor(process, CPid),
  516. dict:store(CPid, {MRef, [MemberPid]}, CPMap)
  517. end.
  518. -spec cull_members_from_pool(#pool{}) -> #pool{}.
  519. cull_members_from_pool(#pool{cull_interval = {0, _}} = Pool) ->
  520. %% 0 cull_interval means do not cull
  521. Pool;
  522. cull_members_from_pool(#pool{name = PoolName,
  523. free_count = FreeCount,
  524. init_count = InitCount,
  525. in_use_count = InUseCount,
  526. cull_interval = Delay,
  527. max_age = MaxAge,
  528. all_members = AllMembers} = Pool) ->
  529. MaxCull = FreeCount - (InitCount - InUseCount),
  530. Pool1 = case MaxCull > 0 of
  531. true ->
  532. MemberInfo = member_info(Pool#pool.free_pids, AllMembers),
  533. ExpiredMembers =
  534. expired_free_members(MemberInfo, os:timestamp(), MaxAge),
  535. CullList = lists:sublist(ExpiredMembers, MaxCull),
  536. lists:foldl(fun({CullMe, _}, S) -> remove_pid(CullMe, S) end,
  537. Pool, CullList);
  538. false ->
  539. Pool
  540. end,
  541. schedule_cull(PoolName, Delay),
  542. Pool1.
  543. -spec schedule_cull(PoolName :: atom() | pid(),
  544. Delay :: time_spec()) -> reference().
  545. %% @doc Schedule a pool cleaning or "cull" for `PoolName' in which
  546. %% members older than `max_age' will be removed until the pool has
  547. %% `init_count' members. Uses `erlang:send_after/3' for light-weight
  548. %% timer that will be auto-cancelled upon pooler shutdown.
  549. schedule_cull(PoolName, Delay) ->
  550. DelayMillis = time_as_millis(Delay),
  551. %% use pid instead of server name atom to take advantage of
  552. %% automatic cancelling
  553. erlang:send_after(DelayMillis, PoolName, cull_pool).
  554. -spec member_info([pid()], dict()) -> [{pid(), member_info()}].
  555. member_info(Pids, AllMembers) ->
  556. [ {P, dict:fetch(P, AllMembers)} || P <- Pids ].
  557. -spec expired_free_members(Members :: [{pid(), member_info()}],
  558. Now :: {_, _, _},
  559. MaxAge :: time_spec()) -> [{pid(), free_member_info()}].
  560. expired_free_members(Members, Now, MaxAge) ->
  561. MaxMicros = time_as_micros(MaxAge),
  562. [ MI || MI = {_, {_, free, LastReturn}} <- Members,
  563. timer:now_diff(Now, LastReturn) >= MaxMicros ].
  564. %% Send a metric using the metrics module from application config or
  565. %% do nothing.
  566. -spec send_metric(Pool :: #pool{},
  567. Label :: atom(),
  568. Value :: metric_value(),
  569. Type :: metric_type()) -> ok.
  570. send_metric(#pool{metrics_mod = pooler_no_metrics}, _Label, _Value, _Type) ->
  571. ok;
  572. send_metric(#pool{name = PoolName, metrics_mod = MetricsMod}, Label, Value, Type) ->
  573. MetricName = pool_metric(PoolName, Label),
  574. MetricsMod:notify(MetricName, Value, Type),
  575. ok.
  576. -spec pool_metric(atom(), atom()) -> binary().
  577. pool_metric(PoolName, Metric) ->
  578. iolist_to_binary([<<"pooler.">>, atom_to_binary(PoolName, utf8),
  579. ".", atom_to_binary(Metric, utf8)]).
  580. -spec time_as_secs(time_spec()) -> non_neg_integer().
  581. time_as_secs({Time, Unit}) ->
  582. time_as_micros({Time, Unit}) div 1000000.
  583. -spec time_as_millis(time_spec()) -> non_neg_integer().
  584. %% @doc Convert time unit into milliseconds.
  585. time_as_millis({Time, Unit}) ->
  586. time_as_micros({Time, Unit}) div 1000.
  587. -spec time_as_micros(time_spec()) -> non_neg_integer().
  588. %% @doc Convert time unit into microseconds
  589. time_as_micros({Time, min}) ->
  590. 60 * 1000 * 1000 * Time;
  591. time_as_micros({Time, sec}) ->
  592. 1000 * 1000 * Time;
  593. time_as_micros({Time, ms}) ->
  594. 1000 * Time;
  595. time_as_micros({Time, mu}) ->
  596. Time.
  597. secs_between({Mega1, Secs1, _}, {Mega2, Secs2, _}) ->
  598. (Mega2 - Mega1) * 1000000 + (Secs2 - Secs1).