pooler_tests.erl 54 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469
  1. -module(pooler_tests).
  2. -include_lib("eunit/include/eunit.hrl").
  3. -export([sleep_for_configured_timeout/0]).
  4. % The `user' processes represent users of the pooler library. A user
  5. % process will take a pid, report details on the pid it has, release
  6. % and take a new pid, stop cleanly, and crash.
  7. start_user() ->
  8. spawn(fun() -> user_loop(start) end).
  9. user_id(Pid) ->
  10. Pid ! {get_tc_id, self()},
  11. receive
  12. {Type, Id} ->
  13. {Type, Id}
  14. end.
  15. user_new_tc(Pid) ->
  16. Pid ! new_tc.
  17. user_stop(Pid) ->
  18. Pid ! stop.
  19. user_crash(Pid) ->
  20. Pid ! crash.
  21. user_loop(Atom) when Atom =:= error_no_members orelse Atom =:= start ->
  22. user_loop(pooler:take_member(test_pool_1));
  23. user_loop(MyTC) ->
  24. receive
  25. {get_tc_id, From} ->
  26. From ! pooled_gs:get_id(MyTC),
  27. user_loop(MyTC);
  28. {ping_tc, From} ->
  29. From ! pooled_gs:ping(MyTC),
  30. user_loop(MyTC);
  31. {ping_count, From} ->
  32. From ! pooled_gs:ping_count(MyTC),
  33. user_loop(MyTC);
  34. new_tc ->
  35. pooler:return_member(test_pool_1, MyTC, ok),
  36. MyNewTC = pooler:take_member(test_pool_1),
  37. user_loop(MyNewTC);
  38. stop ->
  39. pooler:return_member(test_pool_1, MyTC, ok),
  40. stopped;
  41. crash ->
  42. erlang:error({user_loop, kaboom})
  43. end.
  44. % The `tc' processes represent the pids tracked by pooler for testing.
  45. % They have a type and an ID and can report their type and ID and
  46. % stop.
  47. %% tc_loop({Type, Id}) ->
  48. %% receive
  49. %% {get_id, From} ->
  50. %% From ! {ok, Type, Id},
  51. %% tc_loop({Type, Id});
  52. %% stop -> stopped;
  53. %% crash ->
  54. %% erlang:error({tc_loop, kaboom})
  55. %% end.
  56. %% get_tc_id(Pid) ->
  57. %% Pid ! {get_id, self()},
  58. %% receive
  59. %% {ok, Type, Id} ->
  60. %% {Type, Id}
  61. %% after 200 ->
  62. %% timeout
  63. %% end.
  64. %% stop_tc(Pid) ->
  65. %% Pid ! stop.
  66. %% tc_starter(Type) ->
  67. %% Ref = make_ref(),
  68. %% spawn_link(fun() -> tc_loop({Type, Ref}) end).
  69. %% assert_tc_valid(Pid) ->
  70. %% ?assertMatch({_Type, _Ref}, get_tc_id(Pid)),
  71. %% ok.
  72. % tc_sanity_test() ->
  73. % Pid1 = tc_starter("1"),
  74. % {"1", Id1} = get_tc_id(Pid1),
  75. % Pid2 = tc_starter("1"),
  76. % {"1", Id2} = get_tc_id(Pid2),
  77. % ?assertNot(Id1 == Id2),
  78. % stop_tc(Pid1),
  79. % stop_tc(Pid2).
  80. % user_sanity_test() ->
  81. % Pid1 = tc_starter("1"),
  82. % User = spawn(fun() -> user_loop(Pid1) end),
  83. % ?assertMatch({"1", _Ref}, user_id(User)),
  84. % user_crash(User),
  85. % stop_tc(Pid1).
  86. pooler_basics_via_config_test_() ->
  87. {setup,
  88. fun() ->
  89. {ok, _} = error_logger_mon:start_link(),
  90. error_logger_mon:install_handler(pooler),
  91. logger:set_handler_config(default, filters, []),
  92. application:set_env(pooler, metrics_module, fake_metrics),
  93. fake_metrics:start_link()
  94. end,
  95. fun(_X) ->
  96. error_logger_mon:uninstall_handler(),
  97. ok = error_logger_mon:stop(),
  98. fake_metrics:stop()
  99. end,
  100. {foreach,
  101. % setup
  102. fun() ->
  103. error_logger_mon:reset(),
  104. Pools = [
  105. #{
  106. name => test_pool_1,
  107. max_count => 3,
  108. init_count => 2,
  109. cull_interval => {0, min},
  110. start_mfa => {pooled_gs, start_link, [{"type-0"}]}
  111. }
  112. ],
  113. application:set_env(pooler, pools, Pools),
  114. application:start(pooler)
  115. end,
  116. fun(_X) ->
  117. application:stop(pooler)
  118. end,
  119. basic_tests()}}.
  120. pooler_basics_dynamic_test_() ->
  121. {setup,
  122. fun() ->
  123. {ok, _} = error_logger_mon:start_link(),
  124. error_logger_mon:install_handler(pooler),
  125. logger:set_handler_config(default, filters, []),
  126. application:set_env(pooler, metrics_module, fake_metrics),
  127. fake_metrics:start_link()
  128. end,
  129. fun(_X) ->
  130. error_logger_mon:uninstall_handler(),
  131. ok = error_logger_mon:stop(),
  132. fake_metrics:stop()
  133. end,
  134. {foreach,
  135. % setup
  136. fun() ->
  137. error_logger_mon:reset(),
  138. Pool = #{
  139. name => test_pool_1,
  140. max_count => 3,
  141. init_count => 2,
  142. start_mfa => {pooled_gs, start_link, [{"type-0"}]}
  143. },
  144. application:unset_env(pooler, pools),
  145. application:start(pooler),
  146. pooler:new_pool(Pool)
  147. end,
  148. fun(_X) ->
  149. application:stop(pooler)
  150. end,
  151. basic_tests()}}.
  152. pooler_basics_integration_to_other_supervisor_test_() ->
  153. {setup,
  154. fun() ->
  155. {ok, _} = error_logger_mon:start_link(),
  156. error_logger_mon:install_handler(pooler),
  157. logger:set_handler_config(default, filters, []),
  158. application:set_env(pooler, metrics_module, fake_metrics),
  159. fake_metrics:start_link()
  160. end,
  161. fun(_X) ->
  162. error_logger_mon:uninstall_handler(),
  163. ok = error_logger_mon:stop(),
  164. fake_metrics:stop()
  165. end,
  166. {foreach,
  167. % setup
  168. fun() ->
  169. error_logger_mon:reset(),
  170. Pool = #{
  171. name => test_pool_1,
  172. max_count => 3,
  173. init_count => 2,
  174. start_mfa => {pooled_gs, start_link, [{"type-0"}]}
  175. },
  176. application:unset_env(pooler, pools),
  177. application:start(pooler),
  178. supervisor:start_link(fake_external_supervisor, Pool)
  179. end,
  180. fun({ok, SupPid}) ->
  181. exit(SupPid, normal),
  182. application:stop(pooler)
  183. end,
  184. basic_tests()}}.
  185. basic_tests() ->
  186. [
  187. {"there are init_count members at start", fun() ->
  188. Stats = [P || {P, {_, free, _}} <- pooler:pool_stats(test_pool_1)],
  189. ?assertEqual(2, length(Stats))
  190. end},
  191. {"take and return one", fun() ->
  192. P = pooler:take_member(test_pool_1),
  193. ?assertMatch({"type-0", _Id}, pooled_gs:get_id(P)),
  194. ok = pooler:return_member(test_pool_1, P, ok)
  195. end},
  196. {"take and return one, named pool", fun() ->
  197. P = pooler:take_member(test_pool_1),
  198. ?assertMatch({"type-0", _Id}, pooled_gs:get_id(P)),
  199. ok,
  200. pooler:return_member(test_pool_1, P)
  201. end},
  202. {"attempt to take form unknown pool", fun() ->
  203. %% since pools are now servers, an unknown pool will timeout
  204. ?assertExit({noproc, _}, pooler:take_member(bad_pool_name))
  205. end},
  206. {"members creation is triggered after pool exhaustion until max", fun() ->
  207. %% init count is 2
  208. Pids0 = [pooler:take_member(test_pool_1), pooler:take_member(test_pool_1)],
  209. %% since new member creation is async, can only assert
  210. %% that we will get a pid, but may not be first try.
  211. Pids = get_n_pids(1, Pids0),
  212. %% pool is at max now, requests should give error
  213. ?assertEqual(error_no_members, pooler:take_member(test_pool_1)),
  214. ?assertEqual(error_no_members, pooler:take_member(test_pool_1)),
  215. PRefs = [R || {_T, R} <- [pooled_gs:get_id(P) || P <- Pids]],
  216. % no duplicates
  217. ?assertEqual(length(PRefs), length(lists:usort(PRefs)))
  218. end},
  219. {"pids are reused most recent return first", fun() ->
  220. P1 = pooler:take_member(test_pool_1),
  221. P2 = pooler:take_member(test_pool_1),
  222. ?assertNot(P1 == P2),
  223. ok = pooler:return_member(test_pool_1, P1, ok),
  224. ok = pooler:return_member(test_pool_1, P2, ok),
  225. % pids are reused most recent first
  226. ?assertEqual(P2, pooler:take_member(test_pool_1)),
  227. ?assertEqual(P1, pooler:take_member(test_pool_1))
  228. end},
  229. {"if an in-use pid crashes it is replaced", fun() ->
  230. Pids0 = get_n_pids(3, []),
  231. Ids0 = [pooled_gs:get_id(P) || P <- Pids0],
  232. % crash them all
  233. [pooled_gs:crash(P) || P <- Pids0],
  234. Pids1 = get_n_pids(3, []),
  235. Ids1 = [pooled_gs:get_id(P) || P <- Pids1],
  236. [?assertNot(lists:member(I, Ids0)) || I <- Ids1]
  237. end},
  238. {"if a free pid crashes it is replaced", fun() ->
  239. FreePids = [P || {P, {_, free, _}} <- pooler:pool_stats(test_pool_1)],
  240. [exit(P, kill) || P <- FreePids],
  241. Pids1 = get_n_pids(3, []),
  242. ?assertEqual(3, length(Pids1))
  243. end},
  244. {"if a pid is returned with bad status it is replaced", fun() ->
  245. Pids0 = get_n_pids(3, []),
  246. Ids0 = [pooled_gs:get_id(P) || P <- Pids0],
  247. % return them all marking as bad
  248. [pooler:return_member(test_pool_1, P, fail) || P <- Pids0],
  249. Pids1 = get_n_pids(3, []),
  250. Ids1 = [pooled_gs:get_id(P) || P <- Pids1],
  251. [?assertNot(lists:member(I, Ids0)) || I <- Ids1]
  252. end},
  253. {"if a consumer crashes, pid is replaced", fun() ->
  254. Consumer = start_user(),
  255. StartId = user_id(Consumer),
  256. user_crash(Consumer),
  257. NewPid = hd(get_n_pids(1, [])),
  258. NewId = pooled_gs:get_id(NewPid),
  259. ?assertNot(NewId == StartId)
  260. end},
  261. {"it is ok to return an unknown pid", fun() ->
  262. Bogus1 = spawn(fun() -> ok end),
  263. Bogus2 = spawn(fun() -> ok end),
  264. ?assertEqual(ok, pooler:return_member(test_pool_1, Bogus1, ok)),
  265. ?assertEqual(ok, pooler:return_member(test_pool_1, Bogus2, fail))
  266. end},
  267. {"it is ok to return a pid more than once", fun() ->
  268. M = pooler:take_member(test_pool_1),
  269. [
  270. pooler:return_member(test_pool_1, M)
  271. || _I <- lists:seq(1, 37)
  272. ],
  273. ?assertEqual(
  274. 36,
  275. length(
  276. lists:filter(
  277. fun
  278. (
  279. #{
  280. msg :=
  281. {report, #{
  282. label := "ignored return of free member",
  283. pid := Pid
  284. }}
  285. }
  286. ) ->
  287. Pid =:= M;
  288. (_) ->
  289. false
  290. end,
  291. error_logger_mon:get_msgs()
  292. )
  293. )
  294. ),
  295. M1 = pooler:take_member(test_pool_1),
  296. M2 = pooler:take_member(test_pool_1),
  297. ?assert(M1 =/= M2),
  298. Pool1 = gen_server:call(test_pool_1, dump_pool),
  299. ?assertEqual(2, maps:get(in_use_count, Pool1)),
  300. ?assertEqual(0, maps:get(free_count, Pool1)),
  301. pooler:return_member(test_pool_1, M1),
  302. pooler:return_member(test_pool_1, M2),
  303. Pool2 = gen_server:call(test_pool_1, dump_pool),
  304. ?assertEqual(0, maps:get(in_use_count, Pool2)),
  305. ?assertEqual(2, maps:get(free_count, Pool2)),
  306. ok
  307. end},
  308. {"calling return_member on error_no_members is ignored", fun() ->
  309. ?assertEqual(ok, pooler:return_member(test_pool_1, error_no_members)),
  310. ?assertEqual(ok, pooler:return_member(test_pool_1, error_no_members, ok)),
  311. ?assertEqual(ok, pooler:return_member(test_pool_1, error_no_members, fail))
  312. end},
  313. {"dynamic pool creation", fun() ->
  314. PoolSpec = [
  315. {name, dyn_pool_1},
  316. {max_count, 3},
  317. {init_count, 2},
  318. {start_mfa, {pooled_gs, start_link, [{"dyn-0"}]}}
  319. ],
  320. {ok, SupPid1} = pooler:new_pool(PoolSpec),
  321. ?assert(is_pid(SupPid1)),
  322. M = pooler:take_member(dyn_pool_1),
  323. ?assertMatch({"dyn-0", _Id}, pooled_gs:get_id(M)),
  324. ?assertEqual(ok, pooler:rm_pool(dyn_pool_1)),
  325. ?assertExit({noproc, _}, pooler:take_member(dyn_pool_1)),
  326. %% verify pool of same name can be created after removal
  327. {ok, SupPid2} = pooler:new_pool(PoolSpec),
  328. ?assert(is_pid(SupPid2)),
  329. %% remove non-existing pool
  330. ?assertEqual(ok, pooler:rm_pool(dyn_pool_X)),
  331. ?assertEqual(ok, pooler:rm_pool(dyn_pool_1))
  332. end},
  333. {"metrics have been called (no timeout/queue)", fun() ->
  334. %% exercise the API to ensure we have certain keys reported as metrics
  335. fake_metrics:reset_metrics(),
  336. Pids = [pooler:take_member(test_pool_1) || _I <- lists:seq(1, 10)],
  337. [pooler:return_member(test_pool_1, P) || P <- Pids],
  338. catch pooler:take_member(bad_pool_name),
  339. %% kill and unused member
  340. exit(hd(Pids), kill),
  341. %% kill a used member
  342. KillMe = pooler:take_member(test_pool_1),
  343. exit(KillMe, kill),
  344. %% FIXME: We need to wait for pooler to process the
  345. %% exit message. This is ugly, will fix later.
  346. % :(
  347. timer:sleep(200),
  348. ExpectKeys = lists:sort([
  349. <<"pooler.test_pool_1.error_no_members_count">>,
  350. <<"pooler.test_pool_1.events">>,
  351. <<"pooler.test_pool_1.free_count">>,
  352. <<"pooler.test_pool_1.in_use_count">>,
  353. <<"pooler.test_pool_1.killed_free_count">>,
  354. <<"pooler.test_pool_1.killed_in_use_count">>,
  355. <<"pooler.test_pool_1.take_rate">>
  356. ]),
  357. Metrics = fake_metrics:get_metrics(),
  358. GotKeys = lists:usort([Name || {Name, _, _} <- Metrics]),
  359. ?assertEqual(ExpectKeys, GotKeys)
  360. end},
  361. {"metrics have been called (with timeout/queue)", fun() ->
  362. %% exercise the API to ensure we have certain keys reported as metrics
  363. fake_metrics:reset_metrics(),
  364. %% pass a non-zero timeout here to exercise queueing
  365. Pids = [pooler:take_member(test_pool_1, 1) || _I <- lists:seq(1, 10)],
  366. [pooler:return_member(test_pool_1, P) || P <- Pids],
  367. catch pooler:take_member(bad_pool_name),
  368. %% kill and unused member
  369. exit(hd(Pids), kill),
  370. %% kill a used member
  371. KillMe = pooler:take_member(test_pool_1),
  372. exit(KillMe, kill),
  373. %% FIXME: We need to wait for pooler to process the
  374. %% exit message. This is ugly, will fix later.
  375. % :(
  376. timer:sleep(200),
  377. ExpectKeys = lists:sort([
  378. <<"pooler.test_pool_1.error_no_members_count">>,
  379. <<"pooler.test_pool_1.events">>,
  380. <<"pooler.test_pool_1.free_count">>,
  381. <<"pooler.test_pool_1.in_use_count">>,
  382. <<"pooler.test_pool_1.killed_free_count">>,
  383. <<"pooler.test_pool_1.killed_in_use_count">>,
  384. <<"pooler.test_pool_1.take_rate">>,
  385. <<"pooler.test_pool_1.queue_count">>
  386. ]),
  387. Metrics = fake_metrics:get_metrics(),
  388. GotKeys = lists:usort([Name || {Name, _, _} <- Metrics]),
  389. ?assertEqual(ExpectKeys, GotKeys)
  390. end},
  391. {"accept bad member is handled", fun() ->
  392. Bad = spawn(fun() -> ok end),
  393. FakeStarter = spawn(fun() -> starter end),
  394. ?assertEqual(ok, pooler:accept_member(test_pool_1, {FakeStarter, Bad}))
  395. end},
  396. {"utilization returns sane results", fun() ->
  397. #{max_count := MaxCount, queue_max := QueueMax} = dump_pool(test_pool_1),
  398. ?assertEqual(MaxCount, proplists:get_value(max_count, pooler:pool_utilization(test_pool_1))),
  399. ?assertEqual(0, proplists:get_value(in_use_count, pooler:pool_utilization(test_pool_1))),
  400. ?assertEqual(2, proplists:get_value(free_count, pooler:pool_utilization(test_pool_1))),
  401. ?assertEqual(0, proplists:get_value(queued_count, pooler:pool_utilization(test_pool_1))),
  402. ?assertEqual(QueueMax, proplists:get_value(queue_max, pooler:pool_utilization(test_pool_1)))
  403. end}
  404. ].
  405. pooler_groups_test_() ->
  406. {setup,
  407. fun() ->
  408. logger:set_handler_config(default, filters, []),
  409. application:set_env(pooler, metrics_module, fake_metrics),
  410. fake_metrics:start_link()
  411. end,
  412. fun(_X) ->
  413. fake_metrics:stop()
  414. end,
  415. {foreach,
  416. % setup
  417. fun() ->
  418. Pools = [
  419. [
  420. {name, test_pool_1},
  421. {group, group_1},
  422. {max_count, 3},
  423. {init_count, 2},
  424. {start_mfa, {pooled_gs, start_link, [{"type-1-1"}]}}
  425. ],
  426. [
  427. {name, test_pool_2},
  428. {group, group_1},
  429. {max_count, 3},
  430. {init_count, 2},
  431. {start_mfa, {pooled_gs, start_link, [{"type-1-2"}]}}
  432. ],
  433. %% test_pool_3 not part of the group
  434. [
  435. {name, test_pool_3},
  436. {group, undefined},
  437. {max_count, 3},
  438. {init_count, 2},
  439. {start_mfa, {pooled_gs, start_link, [{"type-3"}]}}
  440. ]
  441. ],
  442. application:set_env(pooler, pools, Pools),
  443. pg_start(),
  444. application:start(pooler)
  445. end,
  446. fun(_X) ->
  447. application:stop(pooler),
  448. pg_stop()
  449. end,
  450. [
  451. {"take and return one group member (repeated)", fun() ->
  452. Types = [
  453. begin
  454. Pid = pooler:take_group_member(group_1),
  455. ?assert(is_pid(Pid), [{result, Pid}, {i, I}]),
  456. {Type, _} = pooled_gs:get_id(Pid),
  457. ?assertMatch("type-1" ++ _, Type),
  458. ok = pooler:return_group_member(group_1, Pid, ok),
  459. timer:sleep(10),
  460. Type
  461. end
  462. || I <- lists:seq(1, 50)
  463. ],
  464. Type_1_1 = [X || "type-1-1" = X <- Types],
  465. Type_1_2 = [X || "type-1-2" = X <- Types],
  466. ?assert(length(Type_1_1) > 0, [{types, Types}]),
  467. ?assert(length(Type_1_2) > 0, [{types, Types}])
  468. end},
  469. {"take member from unknown group", fun() ->
  470. ?assertEqual(
  471. error_no_members,
  472. pooler:take_group_member(not_a_group)
  473. )
  474. end},
  475. {"return member to unknown group", fun() ->
  476. Pid = pooler:take_group_member(group_1),
  477. ?assertEqual(ok, pooler:return_group_member(no_such_group, Pid))
  478. end},
  479. {"return member to wrong group", fun() ->
  480. Pid = pooler:take_member(test_pool_3),
  481. ?assertEqual(ok, pooler:return_group_member(group_1, Pid))
  482. end},
  483. {"return member with something which is not a pid", fun() ->
  484. ?assertException(error, _, pooler:return_group_member(group_1, not_pid))
  485. end},
  486. {"take member from empty group", fun() ->
  487. %% artificially empty group member list
  488. [pg_leave(group_1, M) || M <- pooler:group_pools(group_1)],
  489. ?assertEqual(error_no_members, pooler:take_group_member(group_1))
  490. end},
  491. {"return member to group, implied ok", fun() ->
  492. Pid = pooler:take_group_member(group_1),
  493. ?assertEqual(ok, pooler:return_group_member(group_1, Pid))
  494. end},
  495. {"return error_no_member to group", fun() ->
  496. ?assertEqual(ok, pooler:return_group_member(group_1, error_no_members))
  497. end},
  498. {"exhaust pools in group", fun() ->
  499. Pids = get_n_pids_group(group_1, 6, []),
  500. %% they should all be pids
  501. [
  502. begin
  503. {Type, _} = pooled_gs:get_id(P),
  504. ?assertMatch("type-1" ++ _, Type),
  505. ok
  506. end
  507. || P <- Pids
  508. ],
  509. %% further attempts should be error
  510. [
  511. error_no_members,
  512. error_no_members,
  513. error_no_members
  514. ] = [
  515. pooler:take_group_member(group_1)
  516. || _I <- lists:seq(1, 3)
  517. ]
  518. end},
  519. {"rm_group with nonexisting group", fun() ->
  520. ?assertEqual(ok, pooler:rm_group(i_dont_exist))
  521. end},
  522. {"rm_group with existing empty group", fun() ->
  523. ?assertEqual(ok, pooler:rm_pool(test_pool_1)),
  524. ?assertEqual(ok, pooler:rm_pool(test_pool_2)),
  525. % process group de-registration is asynchronous
  526. timer:sleep(100),
  527. ?assertEqual(error_no_members, pooler:take_group_member(group_1)),
  528. ?assertEqual(ok, pooler:rm_group(group_1)),
  529. ?assertExit({noproc, _}, pooler:take_member(test_pool_1)),
  530. ?assertExit({noproc, _}, pooler:take_member(test_pool_2)),
  531. ?assertEqual(
  532. error_no_members,
  533. pooler:take_group_member(group_1)
  534. )
  535. end},
  536. {"rm_group with existing non-empty group", fun() ->
  537. %% Verify that group members exist
  538. MemberPid = pooler:take_group_member(group_1),
  539. ?assert(is_pid(MemberPid)),
  540. pooler:return_group_member(group_1, MemberPid),
  541. Pool1Pid = pooler:take_member(test_pool_1),
  542. ?assert(is_pid(Pool1Pid)),
  543. pooler:return_member(test_pool_1, Pool1Pid),
  544. Pool2Pid = pooler:take_member(test_pool_2),
  545. ?assert(is_pid(Pool2Pid)),
  546. pooler:return_member(test_pool_2, Pool2Pid),
  547. %% Delete and verify that group and pools are destroyed
  548. ?assertEqual(ok, pooler:rm_group(group_1)),
  549. ?assertExit({noproc, _}, pooler:take_member(test_pool_1)),
  550. ?assertExit({noproc, _}, pooler:take_member(test_pool_2)),
  551. ?assertEqual(
  552. error_no_members,
  553. pooler:take_group_member(group_1)
  554. )
  555. end}
  556. ]}}.
  557. pooler_limit_failed_adds_test_() ->
  558. %% verify that pooler crashes completely if too many failures are
  559. %% encountered while trying to add pids.
  560. {setup,
  561. fun() ->
  562. logger:set_handler_config(default, filters, []),
  563. Pools = [
  564. [
  565. {name, test_pool_1},
  566. {max_count, 10},
  567. {init_count, 10},
  568. {start_mfa, {pooled_gs, start_link, [crash]}}
  569. ]
  570. ],
  571. application:set_env(pooler, pools, Pools)
  572. end,
  573. fun(_) ->
  574. application:stop(pooler)
  575. end,
  576. fun() ->
  577. application:start(pooler),
  578. ?assertEqual(error_no_members, pooler:take_member(test_pool_1)),
  579. ?assertEqual(error_no_members, pooler:take_member(test_pool_1))
  580. end}.
  581. pooler_scheduled_cull_test_() ->
  582. {setup,
  583. fun() ->
  584. logger:set_handler_config(default, filters, []),
  585. application:set_env(pooler, metrics_module, fake_metrics),
  586. fake_metrics:start_link(),
  587. Pools = [
  588. [
  589. {name, test_pool_1},
  590. {max_count, 10},
  591. {init_count, 2},
  592. {start_mfa, {pooled_gs, start_link, [{"type-0"}]}},
  593. {cull_interval, {200, ms}},
  594. {max_age, {0, min}}
  595. ]
  596. ],
  597. application:set_env(pooler, pools, Pools),
  598. application:start(pooler)
  599. end,
  600. fun(_X) ->
  601. fake_metrics:stop(),
  602. application:stop(pooler)
  603. end,
  604. [
  605. {foreach,
  606. fun() ->
  607. Pids = get_n_pids(test_pool_1, 10, []),
  608. ?assertEqual(10, length(pooler:pool_stats(test_pool_1))),
  609. ?assertEqual(10, length(Pids)),
  610. Pids
  611. end,
  612. fun(Pids) ->
  613. [pooler:return_member(test_pool_1, P) || P <- Pids]
  614. end,
  615. [
  616. fun(Pids) ->
  617. {"excess members are culled run 1", fun() ->
  618. [pooler:return_member(test_pool_1, P) || P <- Pids],
  619. %% wait for longer than cull delay
  620. timer:sleep(250),
  621. ?assertEqual(2, length(pooler:pool_stats(test_pool_1))),
  622. ?assertEqual(2, proplists:get_value(free_count, pooler:pool_utilization(test_pool_1)))
  623. end}
  624. end,
  625. fun(Pids) ->
  626. {"excess members are culled run 2", fun() ->
  627. [pooler:return_member(test_pool_1, P) || P <- Pids],
  628. %% wait for longer than cull delay
  629. timer:sleep(250),
  630. ?assertEqual(2, length(pooler:pool_stats(test_pool_1))),
  631. ?assertEqual(2, proplists:get_value(free_count, pooler:pool_utilization(test_pool_1)))
  632. end}
  633. end,
  634. fun(Pids) -> in_use_members_not_culled(Pids, 1) end,
  635. fun(Pids) -> in_use_members_not_culled(Pids, 2) end,
  636. fun(Pids) -> in_use_members_not_culled(Pids, 3) end,
  637. fun(Pids) -> in_use_members_not_culled(Pids, 4) end,
  638. fun(Pids) -> in_use_members_not_culled(Pids, 5) end,
  639. fun(Pids) -> in_use_members_not_culled(Pids, 6) end
  640. ]},
  641. {"no cull when init_count matches max_count",
  642. %% not sure how to verify this. But this test at least
  643. %% exercises the code path.
  644. fun() ->
  645. Config = [
  646. {name, test_static_pool_1},
  647. {max_count, 2},
  648. {init_count, 2},
  649. {start_mfa, {pooled_gs, start_link, [{"static-0"}]}},
  650. % ignored
  651. {cull_interval, {200, ms}}
  652. ],
  653. pooler:new_pool(Config),
  654. P = pooler:take_member(test_static_pool_1),
  655. ?assertMatch({"static-0", _}, pooled_gs:get_id(P)),
  656. pooler:return_member(test_static_pool_1, P),
  657. ok
  658. end}
  659. ]}.
  660. in_use_members_not_culled(Pids, N) ->
  661. {"in-use members are not culled " ++ erlang:integer_to_list(N), fun() ->
  662. %% wait for longer than cull delay
  663. timer:sleep(250),
  664. PidCount = length(Pids),
  665. ?assertEqual(
  666. PidCount,
  667. length(pooler:pool_stats(test_pool_1))
  668. ),
  669. ?assertEqual(0, proplists:get_value(free_count, pooler:pool_utilization(test_pool_1))),
  670. ?assertEqual(PidCount, proplists:get_value(in_use_count, pooler:pool_utilization(test_pool_1))),
  671. Returns = lists:sublist(Pids, N),
  672. [
  673. pooler:return_member(test_pool_1, P)
  674. || P <- Returns
  675. ],
  676. timer:sleep(250),
  677. ?assertEqual(
  678. PidCount - N,
  679. length(pooler:pool_stats(test_pool_1))
  680. )
  681. end}.
  682. random_message_test_() ->
  683. {setup,
  684. fun() ->
  685. logger:set_handler_config(default, filters, []),
  686. Pools = [
  687. [
  688. {name, test_pool_1},
  689. {max_count, 2},
  690. {init_count, 1},
  691. {start_mfa, {pooled_gs, start_link, [{"type-0"}]}}
  692. ]
  693. ],
  694. application:set_env(pooler, pools, Pools),
  695. application:start(pooler),
  696. %% now send some bogus messages
  697. %% do the call in a throw-away process to avoid timeout error
  698. spawn(fun() -> catch gen_server:call(test_pool_1, {unexpected_garbage_msg, 5}) end),
  699. gen_server:cast(test_pool_1, {unexpected_garbage_msg, 6}),
  700. whereis(test_pool_1) ! {unexpected_garbage_msg, 7},
  701. ok
  702. end,
  703. fun(_) ->
  704. application:stop(pooler)
  705. end,
  706. [
  707. fun() ->
  708. Pid = spawn(fun() -> ok end),
  709. MonMsg = {'DOWN', erlang:make_ref(), process, Pid, because},
  710. test_pool_1 ! MonMsg
  711. end,
  712. fun() ->
  713. Pid = pooler:take_member(test_pool_1),
  714. {Type, _} = pooled_gs:get_id(Pid),
  715. ?assertEqual("type-0", Type)
  716. end,
  717. fun() ->
  718. RawPool = gen_server:call(test_pool_1, dump_pool),
  719. ?assertEqual(pool, maps:get('$record_name', RawPool))
  720. end
  721. ]}.
  722. pooler_integration_long_init_test_() ->
  723. {foreach,
  724. % setup
  725. fun() ->
  726. logger:set_handler_config(default, filters, []),
  727. {ok, _} = error_logger_mon:start_link(),
  728. error_logger_mon:install_handler(pooler),
  729. Pool = [
  730. {name, test_pool_1},
  731. {max_count, 10},
  732. {init_count, 0},
  733. {member_start_timeout, {10, ms}},
  734. {start_mfa, {pooled_gs, start_link, [{"type-0", fun() -> timer:sleep(15) end}]}}
  735. ],
  736. application:set_env(pooler, pools, [Pool]),
  737. application:start(pooler)
  738. end,
  739. % cleanup
  740. fun(_) ->
  741. error_logger_mon:uninstall_handler(),
  742. ok = error_logger_mon:stop(),
  743. application:stop(pooler)
  744. end,
  745. %
  746. [
  747. fun(_) ->
  748. % Test what happens when pool members take too long to start.
  749. % The pooler_starter should kill off stale members, there by
  750. % reducing the number of children of the member_sup. This
  751. % activity occurs both during take member and accept member.
  752. % Accordingly, the count should go to zero once all starters
  753. % check in.
  754. fun() ->
  755. ?assertEqual(0, children_count(pooler_test_pool_1_member_sup)),
  756. [
  757. begin
  758. ?assertEqual(error_no_members, pooler:take_member(test_pool_1)),
  759. ?assertEqual(1, starting_members(test_pool_1))
  760. end
  761. || _ <- lists:seq(1, 10)
  762. ],
  763. ?assertEqual(10, children_count(pooler_test_pool_1_member_sup)),
  764. timer:sleep(150),
  765. ?assertEqual(0, children_count(pooler_test_pool_1_member_sup)),
  766. ?assertEqual(0, starting_members(test_pool_1)),
  767. %% there is a log when worker start times out
  768. ?assert(
  769. lists:any(
  770. fun
  771. (
  772. #{
  773. level := error,
  774. msg :=
  775. {report, #{
  776. label := "starting member timeout",
  777. pool := test_pool_1
  778. }}
  779. }
  780. ) ->
  781. true;
  782. (_) ->
  783. false
  784. end,
  785. error_logger_mon:get_msgs()
  786. )
  787. )
  788. end
  789. end
  790. ]}.
  791. sleep_for_configured_timeout() ->
  792. SleepTime =
  793. case application:get_env(pooler, sleep_time) of
  794. {ok, Val} ->
  795. Val;
  796. _ ->
  797. 0
  798. end,
  799. timer:sleep(SleepTime).
  800. pooler_integration_queueing_test_() ->
  801. {foreach,
  802. % setup
  803. fun() ->
  804. logger:set_handler_config(default, filters, []),
  805. Pool = [
  806. {name, test_pool_1},
  807. {max_count, 10},
  808. {queue_max, 10},
  809. {init_count, 0},
  810. {metrics, fake_metrics},
  811. {member_start_timeout, {5, sec}},
  812. {start_mfa,
  813. {pooled_gs, start_link, [
  814. {"type-0", fun pooler_tests:sleep_for_configured_timeout/0}
  815. ]}}
  816. ],
  817. application:set_env(pooler, pools, [Pool]),
  818. fake_metrics:start_link(),
  819. application:start(pooler)
  820. end,
  821. % cleanup
  822. fun(_) ->
  823. fake_metrics:stop(),
  824. application:stop(pooler)
  825. end,
  826. [
  827. fun(_) ->
  828. fun() ->
  829. ?assertEqual(0, maps:get(free_count, dump_pool(test_pool_1))),
  830. Val = pooler:take_member(test_pool_1, 10),
  831. ?assert(is_pid(Val)),
  832. pooler:return_member(test_pool_1, Val)
  833. end
  834. end,
  835. fun(_) ->
  836. fun() ->
  837. application:set_env(pooler, sleep_time, 1),
  838. ?assertEqual(0, maps:get(free_count, dump_pool(test_pool_1))),
  839. Val = pooler:take_member(test_pool_1, 0),
  840. ?assertEqual(error_no_members, Val),
  841. ?assertEqual(0, proplists:get_value(free_count, pooler:pool_utilization(test_pool_1))),
  842. timer:sleep(50),
  843. %Next request should be available
  844. Pid = pooler:take_member(test_pool_1, 0),
  845. ?assert(is_pid(Pid)),
  846. pooler:return_member(test_pool_1, Pid)
  847. end
  848. end,
  849. fun(_) ->
  850. fun() ->
  851. application:set_env(pooler, sleep_time, 10),
  852. ?assertEqual(0, maps:get(free_count, dump_pool(test_pool_1))),
  853. [
  854. ?assertEqual(pooler:take_member(test_pool_1, 0), error_no_members)
  855. || _ <- lists:seq(1, maps:get(max_count, dump_pool(test_pool_1)))
  856. ],
  857. timer:sleep(50),
  858. %Next request should be available
  859. Pid = pooler:take_member(test_pool_1, 0),
  860. ?assert(is_pid(Pid)),
  861. pooler:return_member(test_pool_1, Pid)
  862. end
  863. end,
  864. fun(_) ->
  865. fun() ->
  866. % fill to queue_max, next request should return immediately with no_members
  867. % Will return a if queue max is not enforced.
  868. application:set_env(pooler, sleep_time, 100),
  869. [
  870. proc_lib:spawn(fun() ->
  871. Val = pooler:take_member(test_pool_1, 200),
  872. ?assert(is_pid(Val)),
  873. pooler:return_member(Val)
  874. end)
  875. || _ <- lists:seq(1, maps:get(max_count, dump_pool(test_pool_1)))
  876. ],
  877. ?assertEqual(0, proplists:get_value(free_count, pooler:pool_utilization(test_pool_1))),
  878. ?assert(proplists:get_value(queued_count, pooler:pool_utilization(test_pool_1)) >= 1),
  879. ?assertEqual(10, proplists:get_value(queue_max, pooler:pool_utilization(test_pool_1))),
  880. timer:sleep(50),
  881. ?assertEqual(10, queue:len(maps:get(queued_requestors, dump_pool(test_pool_1)))),
  882. ?assertEqual(10, proplists:get_value(queue_max, pooler:pool_utilization(test_pool_1))),
  883. ?assertEqual(pooler:take_member(test_pool_1, 500), error_no_members),
  884. ExpectKeys = lists:sort([
  885. <<"pooler.test_pool_1.error_no_members_count">>,
  886. <<"pooler.test_pool_1.events">>,
  887. <<"pooler.test_pool_1.take_rate">>,
  888. <<"pooler.test_pool_1.queue_count">>,
  889. <<"pooler.test_pool_1.queue_max_reached">>
  890. ]),
  891. Metrics = fake_metrics:get_metrics(),
  892. GotKeys = lists:usort([Name || {Name, _, _} <- Metrics]),
  893. ?assertEqual(ExpectKeys, GotKeys),
  894. timer:sleep(100),
  895. Val = pooler:take_member(test_pool_1, 500),
  896. ?assert(is_pid(Val)),
  897. pooler:return_member(test_pool_1, Val)
  898. end
  899. end
  900. ]}.
  901. pooler_integration_queueing_return_member_test_() ->
  902. {foreach,
  903. % setup
  904. fun() ->
  905. logger:set_handler_config(default, filters, []),
  906. Pool = [
  907. {name, test_pool_1},
  908. {max_count, 10},
  909. {queue_max, 10},
  910. {init_count, 10},
  911. {metrics, fake_metrics},
  912. {member_start_timeout, {5, sec}},
  913. {start_mfa,
  914. {pooled_gs, start_link, [
  915. {"type-0", fun pooler_tests:sleep_for_configured_timeout/0}
  916. ]}}
  917. ],
  918. application:set_env(pooler, pools, [Pool]),
  919. fake_metrics:start_link(),
  920. application:start(pooler)
  921. end,
  922. % cleanup
  923. fun(_) ->
  924. fake_metrics:stop(),
  925. application:stop(pooler)
  926. end,
  927. [
  928. fun(_) ->
  929. fun() ->
  930. application:set_env(pooler, sleep_time, 0),
  931. Parent = self(),
  932. Pids = [
  933. proc_lib:spawn_link(fun() ->
  934. Val = pooler:take_member(test_pool_1, 200),
  935. ?assert(is_pid(Val)),
  936. Parent ! {taken, self()},
  937. receive
  938. return ->
  939. pooler:return_member(test_pool_1, Val)
  940. after 5000 ->
  941. pooler:return_member(test_pool_1, Val)
  942. end,
  943. Parent ! {returned, self()}
  944. end)
  945. || _ <- lists:seq(1, maps:get(max_count, dump_pool(test_pool_1)))
  946. ],
  947. [
  948. receive
  949. {taken, Pid} -> ok
  950. end
  951. || Pid <- Pids
  952. ],
  953. ?assertEqual(error_no_members, pooler:take_member(test_pool_1)),
  954. proc_lib:spawn_link(fun() ->
  955. Val = pooler:take_member(test_pool_1, 200),
  956. Parent ! {extra_result, Val}
  957. end),
  958. [Pid ! return || Pid <- Pids],
  959. [
  960. receive
  961. {returned, Pid} -> ok
  962. end
  963. || Pid <- Pids
  964. ],
  965. receive
  966. {extra_result, Result} ->
  967. ?assert(is_pid(Result)),
  968. pooler:return_member(test_pool_1, Result)
  969. end,
  970. ?assertEqual(
  971. maps:get(max_count, dump_pool(test_pool_1)),
  972. length(maps:get(free_pids, dump_pool(test_pool_1)))
  973. ),
  974. ?assertEqual(
  975. maps:get(max_count, dump_pool(test_pool_1)),
  976. maps:get(free_count, dump_pool(test_pool_1))
  977. )
  978. end
  979. end
  980. ]}.
  981. pooler_integration_test_() ->
  982. {foreach,
  983. % setup
  984. fun() ->
  985. logger:set_handler_config(default, filters, []),
  986. Pools = [
  987. [
  988. {name, test_pool_1},
  989. {max_count, 10},
  990. {init_count, 10},
  991. {start_mfa, {pooled_gs, start_link, [{"type-0"}]}}
  992. ]
  993. ],
  994. application:set_env(pooler, pools, Pools),
  995. application:start(pooler),
  996. Users = [start_user() || _X <- lists:seq(1, 10)],
  997. Users
  998. end,
  999. % cleanup
  1000. fun(Users) ->
  1001. [user_stop(U) || U <- Users],
  1002. application:stop(pooler)
  1003. end,
  1004. %
  1005. [
  1006. fun(Users) ->
  1007. fun() ->
  1008. % each user has a different tc ID
  1009. TcIds = lists:sort([user_id(UPid) || UPid <- Users]),
  1010. ?assertEqual(lists:usort(TcIds), TcIds)
  1011. end
  1012. end,
  1013. fun(Users) ->
  1014. fun() ->
  1015. % users still unique after a renew cycle
  1016. [user_new_tc(UPid) || UPid <- Users],
  1017. TcIds = lists:sort([user_id(UPid) || UPid <- Users]),
  1018. ?assertEqual(lists:usort(TcIds), TcIds)
  1019. end
  1020. end,
  1021. fun(Users) ->
  1022. fun() ->
  1023. % all users crash, pids are replaced
  1024. TcIds1 = lists:sort([user_id(UPid) || UPid <- Users]),
  1025. [user_crash(UPid) || UPid <- Users],
  1026. Seq = lists:seq(1, 5),
  1027. Users2 = [start_user() || _X <- Seq],
  1028. TcIds2 = lists:sort([user_id(UPid) || UPid <- Users2]),
  1029. Both =
  1030. sets:to_list(
  1031. sets:intersection([
  1032. sets:from_list(TcIds1),
  1033. sets:from_list(TcIds2)
  1034. ])
  1035. ),
  1036. ?assertEqual([], Both)
  1037. end
  1038. end
  1039. ]}.
  1040. pooler_auto_grow_disabled_by_default_test_() ->
  1041. {setup,
  1042. fun() ->
  1043. logger:set_handler_config(default, filters, []),
  1044. application:set_env(pooler, metrics_module, fake_metrics),
  1045. fake_metrics:start_link()
  1046. end,
  1047. fun(_X) ->
  1048. fake_metrics:stop()
  1049. end,
  1050. {foreach,
  1051. % setup
  1052. fun() ->
  1053. Pool = [
  1054. {name, test_pool_1},
  1055. {max_count, 5},
  1056. {init_count, 2},
  1057. {start_mfa, {pooled_gs, start_link, [{"type-0"}]}}
  1058. ],
  1059. application:unset_env(pooler, pools),
  1060. application:start(pooler),
  1061. pooler:new_pool(Pool)
  1062. end,
  1063. fun(_X) ->
  1064. application:stop(pooler)
  1065. end,
  1066. [
  1067. {"take one, and it should not auto-grow", fun() ->
  1068. ?assertEqual(2, maps:get(free_count, dump_pool(test_pool_1))),
  1069. P = pooler:take_member(test_pool_1),
  1070. ?assertMatch({"type-0", _Id}, pooled_gs:get_id(P)),
  1071. timer:sleep(100),
  1072. ?assertEqual(1, maps:get(free_count, dump_pool(test_pool_1))),
  1073. ok,
  1074. pooler:return_member(test_pool_1, P)
  1075. end}
  1076. ]}}.
  1077. pooler_auto_grow_enabled_test_() ->
  1078. {setup,
  1079. fun() ->
  1080. logger:set_handler_config(default, filters, []),
  1081. application:set_env(pooler, metrics_module, fake_metrics),
  1082. fake_metrics:start_link()
  1083. end,
  1084. fun(_X) ->
  1085. fake_metrics:stop()
  1086. end,
  1087. {foreach,
  1088. % setup
  1089. fun() ->
  1090. Pool = [
  1091. {name, test_pool_1},
  1092. {max_count, 5},
  1093. {init_count, 2},
  1094. {auto_grow_threshold, 1},
  1095. {start_mfa, {pooled_gs, start_link, [{"type-0"}]}}
  1096. ],
  1097. application:unset_env(pooler, pools),
  1098. application:start(pooler),
  1099. pooler:new_pool(Pool)
  1100. end,
  1101. fun(_X) ->
  1102. application:stop(pooler)
  1103. end,
  1104. [
  1105. {"take one, and it should grow by 2", fun() ->
  1106. ?assertEqual(2, maps:get(free_count, dump_pool(test_pool_1))),
  1107. P = pooler:take_member(test_pool_1),
  1108. ?assertMatch({"type-0", _Id}, pooled_gs:get_id(P)),
  1109. timer:sleep(100),
  1110. ?assertEqual(3, maps:get(free_count, dump_pool(test_pool_1))),
  1111. ok,
  1112. pooler:return_member(test_pool_1, P)
  1113. end}
  1114. ]}}.
  1115. pooler_custom_stop_mfa_test_() ->
  1116. {foreach,
  1117. fun() ->
  1118. logger:set_handler_config(default, filters, []),
  1119. Pool = [
  1120. {name, test_pool_1},
  1121. {max_count, 3},
  1122. {init_count, 2},
  1123. {cull_interval, {200, ms}},
  1124. {max_age, {0, min}},
  1125. {start_mfa, {pooled_gs, start_link, [{foo_type}]}}
  1126. ],
  1127. application:set_env(pooler, pools, [Pool])
  1128. end,
  1129. fun(_) ->
  1130. application:unset_env(pooler, pools)
  1131. end,
  1132. [
  1133. {"default behavior kills the pool member", fun() ->
  1134. ok = application:start(pooler),
  1135. Reason = monitor_members_trigger_culling_and_return_reason(),
  1136. ok = application:stop(pooler),
  1137. ?assertEqual(killed, Reason)
  1138. end},
  1139. {"custom callback terminates the pool member normally", fun() ->
  1140. {ok, [Pool]} = application:get_env(pooler, pools),
  1141. Stop = {stop_mfa, {pooled_gs, stop, ['$pooler_pid']}},
  1142. application:set_env(pooler, pools, [[Stop | Pool]]),
  1143. ok = application:start(pooler),
  1144. Reason = monitor_members_trigger_culling_and_return_reason(),
  1145. ok = application:stop(pooler),
  1146. ?assertEqual(normal, Reason)
  1147. end}
  1148. ]}.
  1149. no_error_logger_reports_after_culling_test_() ->
  1150. %% damn those wraiths! This is the cure
  1151. {foreach,
  1152. fun() ->
  1153. logger:set_handler_config(default, filters, []),
  1154. {ok, _Pid} = error_logger_mon:start_link(),
  1155. Pool = [
  1156. {name, test_pool_1},
  1157. {max_count, 3},
  1158. {init_count, 2},
  1159. {cull_interval, {200, ms}},
  1160. {max_age, {0, min}},
  1161. {start_mfa, {pooled_gs, start_link, [{type}]}}
  1162. ],
  1163. application:set_env(pooler, pools, [Pool])
  1164. end,
  1165. fun(_) ->
  1166. ok = error_logger_mon:stop(),
  1167. error_logger_mon:uninstall_handler(),
  1168. application:unset_env(pooler, pools)
  1169. end,
  1170. [
  1171. {"Force supervisor to report by using exit/2 instead of terminate_child", fun() ->
  1172. {ok, [Pool]} = application:get_env(pooler, pools),
  1173. Stop = {stop_mfa, {erlang, exit, ['$pooler_pid', kill]}},
  1174. application:set_env(pooler, pools, [[Stop | Pool]]),
  1175. ok = application:start(pooler),
  1176. error_logger_mon:install_handler(),
  1177. error_logger_mon:reset(),
  1178. Reason = monitor_members_trigger_culling_and_return_reason(),
  1179. %% we need to wait for the message to arrive before deleting handler
  1180. timer:sleep(250),
  1181. error_logger_mon:uninstall_handler(),
  1182. ok = application:stop(pooler),
  1183. ?assertEqual(killed, Reason),
  1184. ?assertEqual(
  1185. 1,
  1186. error_logger_mon:get_msg_count(),
  1187. [
  1188. {msgs, error_logger_mon:get_msgs()},
  1189. {m, [R || #{msg := {report, R}} <- error_logger_mon:get_msgs()]}
  1190. ]
  1191. )
  1192. end},
  1193. {"Default MFA shouldn't generate any reports during culling", fun() ->
  1194. ok = application:start(pooler),
  1195. error_logger_mon:install_handler(),
  1196. Reason = monitor_members_trigger_culling_and_return_reason(),
  1197. error_logger_mon:uninstall_handler(),
  1198. ok = application:stop(pooler),
  1199. ?assertEqual(killed, Reason),
  1200. ?assertEqual(0, error_logger_mon:get_msg_count())
  1201. end}
  1202. ]}.
  1203. monitor_members_trigger_culling_and_return_reason() ->
  1204. Pids = get_n_pids(test_pool_1, 3, []),
  1205. [erlang:monitor(process, P) || P <- Pids],
  1206. [pooler:return_member(test_pool_1, P) || P <- Pids],
  1207. receive
  1208. {'DOWN', _Ref, process, _Pid, Reason} ->
  1209. Reason
  1210. after 250 -> timeout
  1211. end.
  1212. time_as_millis_test_() ->
  1213. Zeros = [{{0, U}, 0} || U <- [min, sec, ms, mu]],
  1214. Ones = [
  1215. {{1, min}, 60000},
  1216. {{1, sec}, 1000},
  1217. {{1, ms}, 1},
  1218. {{1, mu}, 0}
  1219. ],
  1220. Misc = [{{3000, mu}, 3}],
  1221. Tests = Zeros ++ Ones ++ Misc,
  1222. [?_assertEqual(E, pooler:time_as_millis(I)) || {I, E} <- Tests].
  1223. time_as_micros_test_() ->
  1224. Zeros = [{{0, U}, 0} || U <- [min, sec, ms, mu]],
  1225. Ones = [
  1226. {{1, min}, 60000000},
  1227. {{1, sec}, 1000000},
  1228. {{1, ms}, 1000},
  1229. {{1, mu}, 1}
  1230. ],
  1231. Misc = [{{3000, mu}, 3000}],
  1232. Tests = Zeros ++ Ones ++ Misc,
  1233. [?_assertEqual(E, pooler:time_as_micros(I)) || {I, E} <- Tests].
  1234. call_free_members_test_() ->
  1235. NumWorkers = 10,
  1236. PoolName = test_pool_1,
  1237. {setup,
  1238. fun() ->
  1239. application:set_env(pooler, metrics_module, fake_metrics),
  1240. fake_metrics:start_link()
  1241. end,
  1242. fun(_X) ->
  1243. fake_metrics:stop()
  1244. end,
  1245. {foreach,
  1246. fun() ->
  1247. Pool = [
  1248. {name, PoolName},
  1249. {max_count, NumWorkers},
  1250. {init_count, NumWorkers},
  1251. {start_mfa, {pooled_gs, start_link, [{"type-0"}]}}
  1252. ],
  1253. application:unset_env(pooler, pools),
  1254. application:start(pooler),
  1255. pooler:new_pool(Pool)
  1256. end,
  1257. fun(_X) ->
  1258. application:stop(pooler)
  1259. end,
  1260. [
  1261. {"perform a ping across the pool when all workers are free", fun() ->
  1262. ?assertEqual(NumWorkers, maps:get(free_count, dump_pool(PoolName))),
  1263. Res = pooler:call_free_members(PoolName, fun pooled_gs:ping/1),
  1264. ?assertEqual(NumWorkers, count_pongs(Res))
  1265. end},
  1266. {"perform a ping across the pool when two workers are taken", fun() ->
  1267. ?assertEqual(NumWorkers, maps:get(free_count, dump_pool(PoolName))),
  1268. Pids = [pooler:take_member(PoolName) || _X <- lists:seq(0, 1)],
  1269. Res = pooler:call_free_members(PoolName, fun pooled_gs:ping/1),
  1270. ?assertEqual(NumWorkers - 2, count_pongs(Res)),
  1271. [pooler:return_member(PoolName, P) || P <- Pids]
  1272. end},
  1273. {"perform an op where the op crashes all free members", fun() ->
  1274. ?assertEqual(NumWorkers, maps:get(free_count, dump_pool(PoolName))),
  1275. Res = pooler:call_free_members(
  1276. PoolName,
  1277. fun pooled_gs:error_on_call/1
  1278. ),
  1279. ?assertEqual(NumWorkers, count_errors(Res))
  1280. end}
  1281. ]}}.
  1282. count_pongs(Result) ->
  1283. lists:foldl(
  1284. fun
  1285. ({ok, pong}, Acc) -> Acc + 1;
  1286. ({error, _}, Acc) -> Acc
  1287. end,
  1288. 0,
  1289. Result
  1290. ).
  1291. count_errors(Result) ->
  1292. lists:foldl(
  1293. fun
  1294. ({error, _}, Acc) -> Acc + 1;
  1295. ({ok, _}, Acc) -> Acc
  1296. end,
  1297. 0,
  1298. Result
  1299. ).
  1300. % testing crash recovery means race conditions when either pids
  1301. % haven't yet crashed or pooler hasn't recovered. So this helper loops
  1302. % forver until N pids are obtained, ignoring error_no_members.
  1303. get_n_pids(N, Acc) ->
  1304. get_n_pids(test_pool_1, N, Acc).
  1305. get_n_pids(_Pool, 0, Acc) ->
  1306. Acc;
  1307. get_n_pids(Pool, N, Acc) ->
  1308. case pooler:take_member(Pool) of
  1309. error_no_members ->
  1310. get_n_pids(Pool, N, Acc);
  1311. Pid ->
  1312. get_n_pids(Pool, N - 1, [Pid | Acc])
  1313. end.
  1314. get_n_pids_group(_Group, 0, Acc) ->
  1315. Acc;
  1316. get_n_pids_group(Group, N, Acc) ->
  1317. case pooler:take_group_member(Group) of
  1318. error_no_members ->
  1319. get_n_pids_group(Group, N, Acc);
  1320. Pid ->
  1321. get_n_pids_group(Group, N - 1, [Pid | Acc])
  1322. end.
  1323. children_count(SupId) ->
  1324. length(supervisor:which_children(SupId)).
  1325. starting_members(PoolName) ->
  1326. length(maps:get(starting_members, dump_pool(PoolName))).
  1327. dump_pool(PoolName) ->
  1328. gen_server:call(PoolName, dump_pool).
  1329. % >= OTP-21
  1330. -ifdef(OTP_RELEASE).
  1331. -if(?OTP_RELEASE >= 23).
  1332. -define(USE_PG_NOT_PG2, true).
  1333. -else.
  1334. -undef(USE_PG_NOT_PG2).
  1335. -endif.
  1336. % < OTP-21
  1337. -else.
  1338. -undef(USE_PG_NOT_PG2).
  1339. -endif.
  1340. -ifdef(USE_PG_NOT_PG2).
  1341. pg_start() ->
  1342. pg:start(_Scope = 'pg').
  1343. pg_stop() ->
  1344. lists:foreach(
  1345. fun(Group) -> pg:leave(Group, pg:get_members(Group)) end,
  1346. pg:which_groups()
  1347. ).
  1348. pg_leave(Group, Pid) ->
  1349. pg:leave(Group, Pid).
  1350. -else.
  1351. pg_start() ->
  1352. pg2:start().
  1353. pg_stop() ->
  1354. application:stop(pg2).
  1355. pg_leave(Group, Pid) ->
  1356. pg2:leave(Group, Pid).
  1357. -endif.