File pooler-1.5.2-git.patch of Package pooler
diff --git a/.travis.yml b/.travis.yml
index 7657fc6..edc4a9b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,8 +1,12 @@
language: erlang
otp_release:
- - 18.0
- - 17.0
+ - 21.0
+ - 20.3
+ - 19.3
+ - 18.3
+ - 17.5
- R16B03
- - R15B03
script:
- make test
+before_script:
+ - "kerl list installations"
diff --git a/Makefile b/Makefile
index c4f2adf..06d5930 100644
--- a/Makefile
+++ b/Makefile
@@ -11,7 +11,8 @@ run: $(REBAR)
@$(REBAR) as dev shell --apps pooler --config config/demo.config
test: $(REBAR)
- $(REBAR) eunit skip_deps=true verbose=3
+ $(REBAR) eunit --cover skip_deps=true verbose=3
+ $(REBAR) cover
doc: $(REBAR)
$(REBAR) as dev edoc
diff --git a/rebar.config b/rebar.config
index cf1e643..c017208 100644
--- a/rebar.config
+++ b/rebar.config
@@ -1,5 +1,8 @@
{erl_opts, [
{platform_define, "^[0-9]+", namespaced_types},
+ {platform_define, "^(19|2)", rand_only},
+ {platform_define, "^(R|1|20)", fun_stacktrace},
+
debug_info,
@@ -22,8 +25,8 @@
warn_unused_function,
warn_deprecated_function,
- strict_validation,
- warn_missing_spec
+ strict_validation
+ %, warn_missing_spec
]}.
{deps, [
@@ -35,9 +38,14 @@
{deps, [
{edown, ".*",
- {git, "git://github.com/uwiger/edown.git", {tag, "0.8"}}}
- ]}
- ]}
+ {git, "https://github.com/uwiger/edown.git", {tag, "0.8"}}}
+ ]},
+ {erl_opts, [nowarn_export_all]}
+ ]},
+ {test, [
+ {erl_opts, [nowarn_export_all]}
+ ]
+ }
]}.
{eunit_opts, [{report, {eunit_progress, [colored, profile]}}]}.
diff --git a/src/pooler.erl b/src/pooler.erl
index 2d271ed..ff52b33 100644
--- a/src/pooler.erl
+++ b/src/pooler.erl
@@ -38,6 +38,7 @@
return_member/2,
return_member/3,
pool_stats/1,
+ pool_utilization/1,
manual_start/0,
new_pool/1,
pool_child_spec/1,
@@ -58,11 +59,6 @@
terminate/2,
code_change/3]).
-%% To help with testing internal functions
--ifdef(TEST).
--compile([export_all]).
--endif.
-
%% ------------------------------------------------------------------
%% Application API
%% ------------------------------------------------------------------
@@ -306,6 +302,14 @@ return_member(_, error_no_members) ->
pool_stats(PoolName) ->
gen_server:call(PoolName, pool_stats).
+%% @doc Obtain utilization info for a pool.
+%%
+%% Format of the return value is subject to change, but for now it
+%% will be a proplist to maintain backcompat with R16.
+-spec pool_utilization(atom() | pid()) -> [{atom(), integer()}].
+pool_utilization(PoolName) ->
+ gen_server:call(PoolName, pool_utilization).
+
%% @doc Invokes `Fun' with arity 1 over all free members in pool with `PoolName'.
%%
-spec call_free_members(atom() | pid(), fun((pid()) -> term())) -> Res when
@@ -356,6 +360,8 @@ handle_call(stop, _From, Pool) ->
{stop, normal, stop_ok, Pool};
handle_call(pool_stats, _From, Pool) ->
{reply, dict:to_list(Pool#pool.all_members), Pool};
+handle_call(pool_utilization, _From, Pool) ->
+ {reply, compute_utilization(Pool), Pool};
handle_call(dump_pool, _From, Pool) ->
{reply, Pool, Pool};
handle_call({call_free_members, Fun}, _From, #pool{free_pids = Pids} = Pool) ->
@@ -480,7 +486,7 @@ reply_to_queued_requestor(TRef, Pid, From = {APid, _}, NewQueuedRequestors, Pool
send_metric(Pool, events, error_no_members, history),
gen_server:reply(From, Pid),
Pool1.
-
+
-spec take_member_bookkeeping(pid(),
{pid(), _},
@@ -553,7 +559,8 @@ init_members_sync(N, #pool{name = PoolName} = Pool) ->
{ok, Pool2}
end.
-collect_init_members(#pool{starting_members = []} = Pool) ->
+collect_init_members(#pool{starting_members = Empty} = Pool)
+ when Empty =:= [] ->
Pool;
collect_init_members(#pool{member_start_timeout = StartTimeout} = Pool) ->
Timeout = time_as_millis(StartTimeout),
@@ -733,7 +740,7 @@ remove_pid(Pid, Pool) ->
FreePids = lists:delete(Pid, Pool#pool.free_pids),
NumFree = Pool#pool.free_count - 1,
Pool1 = Pool#pool{free_pids = FreePids, free_count = NumFree},
- terminate_pid(Pid, StopMFA),
+ terminate_pid(PoolName, Pid, StopMFA),
send_metric(Pool1, killed_free_count, {inc, 1}, counter),
Pool1#pool{all_members = dict:erase(Pid, AllMembers)};
{ok, {MRef, CPid, _Time}} ->
@@ -741,13 +748,13 @@ remove_pid(Pid, Pool) ->
%% the consumer.
erlang:demonitor(MRef, [flush]),
Pool1 = Pool#pool{in_use_count = Pool#pool.in_use_count - 1},
- terminate_pid(Pid, StopMFA),
+ terminate_pid(PoolName, Pid, StopMFA),
send_metric(Pool1, killed_in_use_count, {inc, 1}, counter),
Pool1#pool{consumer_to_pid = cpmap_remove(Pid, CPid, CPMap),
all_members = dict:erase(Pid, AllMembers)};
error ->
error_logger:error_report({{pool, PoolName}, unknown_pid, Pid,
- erlang:get_stacktrace()}),
+ ?GET_STACKTRACE}),
send_metric(Pool, events, unknown_pid, history),
Pool
end.
@@ -844,11 +851,6 @@ send_metric(#pool{name = PoolName, metrics_mod = MetricsMod,
MetricName = pool_metric_exometer(PoolName, Label),
MetricsMod:update_or_create(MetricName, Value, counter, []),
ok;
-send_metric(#pool{name = PoolName, metrics_mod = MetricsMod,
- metrics_api = exometer}, Label, {dec, Value}, counter) ->
- MetricName = pool_metric_exometer(PoolName, Label),
- MetricsMod:update_or_create(MetricName, - Value, counter, []),
- ok;
% Exometer does not support 'history' type metrics right now.
send_metric(#pool{name = _PoolName, metrics_mod = _MetricsMod,
metrics_api = exometer}, _Label, _Value, history) ->
@@ -920,21 +922,37 @@ maybe_reply({Member, NewPool}) ->
%% Terminates the pid's pool member given a MFA that gets applied. The list
%% of arguments must contain the fixed atom ?POOLER_PID, which is replaced
%% by the target pid. Failure to provide a valid MFA will lead to use the
-%% default callback, i.e `erlang:exit(Pid, kill)`.
--spec terminate_pid(pid(), {atom(), atom(), [term()]}) -> ok.
-terminate_pid(Pid, {Mod, Fun, Args}) when is_list(Args) ->
- NewArgs = [case Arg of
- ?POOLER_PID -> Pid;
- _ -> Arg
- end || Arg <- Args],
+%% default callback.
+-spec terminate_pid(atom(), pid(), {atom(), atom(), [term()]}) -> ok.
+terminate_pid(PoolName, Pid, {Mod, Fun, Args}) when is_list(Args) ->
+ NewArgs = replace_placeholders(PoolName, Pid, Args),
case catch erlang:apply(Mod, Fun, NewArgs) of
{'EXIT', _} ->
- terminate_pid(Pid, ?DEFAULT_STOP_MFA);
+ terminate_pid(PoolName, Pid, ?DEFAULT_STOP_MFA);
_Result ->
ok
- end;
-terminate_pid(Pid, _) ->
- terminate_pid(Pid, ?DEFAULT_STOP_MFA).
+ end.
+
+replace_placeholders(Name, Pid, Args) ->
+ [case Arg of
+ ?POOLER_POOL_NAME ->
+ pooler_pool_sup:build_member_sup_name(Name);
+ ?POOLER_PID ->
+ Pid;
+ _ ->
+ Arg
+ end || Arg <- Args].
+
+compute_utilization(#pool{max_count = MaxCount,
+ in_use_count = InUseCount,
+ free_count = FreeCount,
+ queued_requestors = Queue,
+ queue_max = QueueMax}) ->
+ [{max_count, MaxCount},
+ {in_use_count, InUseCount},
+ {free_count, FreeCount},
+ {queued_count, queue:len(Queue)}, %% Note not O(n), so in pathological cases this might be expensive
+ {queue_max, QueueMax}].
do_call_free_members(Fun, Pids) ->
[do_call_free_member(Fun, P) || P <- Pids].
diff --git a/src/pooler.hrl b/src/pooler.hrl
index ffac0e0..741ca2f 100644
--- a/src/pooler.hrl
+++ b/src/pooler.hrl
@@ -5,8 +5,11 @@
-define(DEFAULT_AUTO_GROW_THRESHOLD, undefined).
-define(POOLER_GROUP_TABLE, pooler_group_table).
-define(DEFAULT_POOLER_QUEUE_MAX, 50).
+-define(POOLER_POOL_NAME, '$pooler_pool_name').
-define(POOLER_PID, '$pooler_pid').
--define(DEFAULT_STOP_MFA, {erlang, exit, [?POOLER_PID, kill]}).
+-define(DEFAULT_STOP_MFA, {supervisor,
+ terminate_child,
+ [?POOLER_POOL_NAME, ?POOLER_PID]}).
-type member_info() :: {string(), free | pid(), {_, _, _}}.
-type free_member_info() :: {string(), free, {_, _, _}}.
@@ -21,6 +24,20 @@
-type p_requestor_queue() :: queue().
-endif.
+-ifdef(rand_only).
+-define(RANDOM_SEED(X), rand:seed(exs1024s, X)).
+-define(RANDOM_UNIFORM(X), rand:uniform(X)).
+-else.
+-define(RANDOM_SEED(X), random:seed(X)).
+-define(RANDOM_UNIFORM(X), random:uniform(X)).
+-endif.
+
+-ifdef(fun_stacktrace).
+-define(GET_STACKTRACE, erlang:get_stacktrace()).
+-else.
+-define(GET_STACKTRACE, try throw(fake_stacktrace) catch _:_:S -> S end).
+-endif.
+
-record(pool, {
name :: atom(),
group :: atom(),
@@ -99,7 +116,7 @@
%% A queue of requestors for blocking take member requests
queued_requestors = queue:new() :: p_requestor_queue(),
%% The max depth of the queue
- queue_max = 50
+ queue_max = 50 :: non_neg_integer()
}).
-define(gv(X, Y), proplists:get_value(X, Y)).
diff --git a/src/pooler_pool_sup.erl b/src/pooler_pool_sup.erl
index e08faa7..6720412 100644
--- a/src/pooler_pool_sup.erl
+++ b/src/pooler_pool_sup.erl
@@ -4,7 +4,8 @@
-export([start_link/1, init/1,
pool_sup_name/1,
- member_sup_name/1]).
+ member_sup_name/1,
+ build_member_sup_name/1]).
-include("pooler.hrl").
@@ -25,8 +26,10 @@ init(#pool{} = Pool) ->
Restart = {one_for_all, 5, 60},
{ok, {Restart, [MemberSupSpec, PoolerSpec]}}.
-
member_sup_name(#pool{name = PoolName}) ->
+ build_member_sup_name(PoolName).
+
+build_member_sup_name(PoolName) ->
list_to_atom("pooler_" ++ atom_to_list(PoolName) ++ "_member_sup").
pool_sup_name(#pool{name = PoolName}) ->
diff --git a/src/pooler_starter.erl b/src/pooler_starter.erl
index 52141e4..f3e4dac 100644
--- a/src/pooler_starter.erl
+++ b/src/pooler_starter.erl
@@ -6,7 +6,6 @@
-behaviour(gen_server).
-include("pooler.hrl").
--include_lib("eunit/include/eunit.hrl").
%% ------------------------------------------------------------------
%% API Function Exports
@@ -29,11 +28,6 @@
terminate/2,
code_change/3]).
-%% To help with testing internal functions
--ifdef(TEST).
--compile([export_all]).
--endif.
-
%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------
@@ -84,9 +78,9 @@ stop_member_async(Pid) ->
%% ------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------
--record(starter, {pool,
- parent,
- msg}).
+-record(starter, {pool :: #pool{},
+ parent :: pid() | atom(),
+ msg :: term()}).
-spec init({#pool{}, pid() | atom()}) -> {'ok', #starter{}, 0}.
init({Pool, Parent}) ->
diff --git a/test/error_logger_mon.erl b/test/error_logger_mon.erl
new file mode 100644
index 0000000..1439401
--- /dev/null
+++ b/test/error_logger_mon.erl
@@ -0,0 +1,65 @@
+%%% A gen_server to check if we get any error_logger messages during test to see if
+%%% any messages gets generated when they shouldn't
+
+-module(error_logger_mon).
+
+-behaviour(gen_server).
+-define(SERVER, ?MODULE).
+
+-record(state, {count = 0 :: integer()}).
+
+%% ------------------------------------------------------------------
+%% API Function Exports
+%% ------------------------------------------------------------------
+%% gen_server
+-export([start_link/0,
+ report/0,
+ get_msg_count/0,
+ stop/0
+ ]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+ terminate/2, code_change/3]).
+
+%% ------------------------------------------------------------------
+%% API Function Definitions
+%% ------------------------------------------------------------------
+
+start_link() ->
+ gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+report() ->
+ gen_server:call(?SERVER, report).
+
+get_msg_count() ->
+ gen_server:call(?SERVER, get_count).
+
+stop() ->
+ gen_server:call(?SERVER, stop).
+
+%% ------------------------------------------------------------------
+%% gen_server Function Definitions
+%% ------------------------------------------------------------------
+init([]) ->
+ {ok, #state{}}.
+
+handle_call(get_count, _From, #state{count = C} = State) ->
+ {reply, C, State};
+handle_call(report, _From, #state{count = C} = State) ->
+ {reply, ok, State#state{count = C+1}};
+handle_call(stop, _From, State) ->
+ {stop, normal, ok, State};
+handle_call(_Request, _From, State) ->
+ {reply, error, State}.
+
+handle_cast(_Msg, State) ->
+ {noreply, State}.
+
+handle_info(_Info, State) ->
+ {noreply, State}.
+
+terminate(_Reason, _State) ->
+ ok.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
diff --git a/test/error_logger_pooler_h.erl b/test/error_logger_pooler_h.erl
new file mode 100644
index 0000000..ce037ae
--- /dev/null
+++ b/test/error_logger_pooler_h.erl
@@ -0,0 +1,25 @@
+%%% report handler to add to error_logger for calling error_logger_mon
+%%% during test
+-module(error_logger_pooler_h).
+
+-export([init/1,
+ handle_event/2,
+ handle_call/2,
+ handle_info/2,
+ terminate/2]).
+
+init(T) ->
+ {ok, T}.
+
+handle_event(_Event, Type) ->
+ error_logger_mon:report(),
+ {ok, Type}.
+
+handle_info(_, Type) ->
+ {ok, Type}.
+
+handle_call(_Query, _Type) ->
+ {error, bad_query}.
+
+terminate(_Reason, _Type) ->
+ [].
diff --git a/test/fake_metrics.erl b/test/fake_metrics.erl
index fdc8b72..61d820e 100644
--- a/test/fake_metrics.erl
+++ b/test/fake_metrics.erl
@@ -48,7 +48,7 @@ get_metrics() ->
%% gen_server Function Definitions
%% ------------------------------------------------------------------
-record(state, {
- metrics = []
+ metrics = [] :: list()
}).
init(_) ->
diff --git a/src/pooled_gs.erl b/test/pooled_gs.erl
similarity index 95%
rename from src/pooled_gs.erl
rename to test/pooled_gs.erl
index 994a1d7..43765f9 100644
--- a/src/pooled_gs.erl
+++ b/test/pooled_gs.erl
@@ -10,6 +10,8 @@
-behaviour(gen_server).
-define(SERVER, ?MODULE).
+-include("pooler.hrl").
+
%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------
@@ -75,9 +77,9 @@ stop(S) ->
%% gen_server Function Definitions
%% ------------------------------------------------------------------
-record(state, {
- type = "",
- id,
- ping_count = 0
+ type = "" :: string(),
+ id :: reference(),
+ ping_count = 0 :: non_neg_integer()
}).
init({Type}) ->
@@ -91,7 +93,7 @@ init({Type, StartFun}) ->
handle_call(get_id, _From, State) ->
{reply, {State#state.type, State#state.id}, State};
handle_call({do_work, T}, _From, State) ->
- Sleep = random:uniform(T),
+ Sleep = ?RANDOM_UNIFORM(T),
timer:sleep(Sleep),
{reply, {ok, Sleep}, State};
handle_call(ping, _From, #state{ping_count = C } = State) ->
diff --git a/test/pooler_tests.erl b/test/pooler_tests.erl
index 43bb11a..d84d235 100644
--- a/test/pooler_tests.erl
+++ b/test/pooler_tests.erl
@@ -402,6 +402,17 @@ basic_tests() ->
Bad = spawn(fun() -> ok end),
FakeStarter = spawn(fun() -> starter end),
?assertEqual(ok, pooler:accept_member(test_pool_1, {FakeStarter, Bad}))
+ end},
+
+ {"utilization returns sane results",
+ fun() ->
+ #pool{max_count = MaxCount, queue_max = QueueMax} = Pool = sys:get_state(test_pool_1),
+
+ ?assertEqual(MaxCount, ?gv(max_count, pooler:pool_utilization(test_pool_1))),
+ ?assertEqual(0, ?gv(in_use_count, pooler:pool_utilization(test_pool_1))),
+ ?assertEqual(2, ?gv(free_count, pooler:pool_utilization(test_pool_1))),
+ ?assertEqual(0, ?gv(queued_count, pooler:pool_utilization(test_pool_1))),
+ ?assertEqual(QueueMax, ?gv(queue_max, pooler:pool_utilization(test_pool_1)))
end}
].
@@ -621,7 +632,8 @@ pooler_scheduled_cull_test_() ->
[ pooler:return_member(test_pool_1, P) || P <- Pids ],
%% wait for longer than cull delay
timer:sleep(250),
- ?assertEqual(2, length(pooler:pool_stats(test_pool_1)))
+ ?assertEqual(2, length(pooler:pool_stats(test_pool_1))),
+ ?assertEqual(2, ?gv(free_count,pooler:pool_utilization(test_pool_1)))
end}
end,
@@ -631,7 +643,8 @@ pooler_scheduled_cull_test_() ->
[ pooler:return_member(test_pool_1, P) || P <- Pids ],
%% wait for longer than cull delay
timer:sleep(250),
- ?assertEqual(2, length(pooler:pool_stats(test_pool_1)))
+ ?assertEqual(2, length(pooler:pool_stats(test_pool_1))),
+ ?assertEqual(2, ?gv(free_count,pooler:pool_utilization(test_pool_1)))
end}
end,
@@ -668,6 +681,8 @@ in_use_members_not_culled(Pids, N) ->
PidCount = length(Pids),
?assertEqual(PidCount,
length(pooler:pool_stats(test_pool_1))),
+ ?assertEqual(0, ?gv(free_count,pooler:pool_utilization(test_pool_1))),
+ ?assertEqual(PidCount, ?gv(in_use_count,pooler:pool_utilization(test_pool_1))),
Returns = lists:sublist(Pids, N),
[ pooler:return_member(test_pool_1, P)
|| P <- Returns ],
@@ -769,7 +784,7 @@ sleep_for_configured_timeout() ->
_ ->
0
end,
- timer:sleep(SleepTime).
+ timer:sleep(SleepTime).
pooler_integration_queueing_test_() ->
{foreach,
@@ -814,6 +829,7 @@ pooler_integration_queueing_test_() ->
?assertEqual(0, (dump_pool(test_pool_1))#pool.free_count),
Val = pooler:take_member(test_pool_1, 0),
?assertEqual(error_no_members, Val),
+ ?assertEqual(0, ?gv(free_count,pooler:pool_utilization(test_pool_1))),
timer:sleep(50),
%Next request should be available
Pid = pooler:take_member(test_pool_1, 0),
@@ -847,8 +863,14 @@ pooler_integration_queueing_test_() ->
end)
|| _ <- lists:seq(1, (dump_pool(test_pool_1))#pool.max_count)
],
+ ?assertEqual(0, ?gv(free_count,pooler:pool_utilization(test_pool_1))),
+ ?assert(?gv(queued_count,pooler:pool_utilization(test_pool_1)) >= 1),
+ ?assertEqual(10, ?gv(queue_max,pooler:pool_utilization(test_pool_1))),
+
timer:sleep(50),
?assertEqual(10, queue:len((dump_pool(test_pool_1))#pool.queued_requestors)),
+ ?assertEqual(10, ?gv(queue_max,pooler:pool_utilization(test_pool_1))),
+
?assertEqual(pooler:take_member(test_pool_1, 500), error_no_members),
ExpectKeys = lists:sort([<<"pooler.test_pool_1.error_no_members_count">>,
<<"pooler.test_pool_1.events">>,
@@ -1098,6 +1120,53 @@ pooler_custom_stop_mfa_test_() ->
?assertEqual(normal, Reason)
end}]}.
+no_error_logger_reports_after_culling_test_() ->
+ %% damn those wraiths! This is the cure
+ {foreach,
+ fun() ->
+ {ok, _Pid} = error_logger_mon:start_link(),
+ Pool = [{name, test_pool_1},
+ {max_count, 3},
+ {init_count, 2},
+ {cull_interval, {200, ms}},
+ {max_age, {0, min}},
+ {start_mfa, {pooled_gs, start_link, [{type}]}}],
+ application:set_env(pooler, pools, [Pool])
+ end,
+ fun(_) ->
+ ok = error_logger_mon:stop(),
+ error_logger:delete_report_handler(error_logger_pooler_h),
+ application:unset_env(pooler, pools)
+ end,
+ [
+ {"Force supervisor to report by using exit/2 instead of terminate_child",
+ fun() ->
+ {ok, [Pool]} = application:get_env(pooler, pools),
+ Stop = {stop_mfa, {erlang, exit, ['$pooler_pid', kill]}},
+ application:set_env(pooler, pools, [[Stop | Pool]]),
+ ok = application:start(pooler),
+ error_logger:add_report_handler(error_logger_pooler_h),
+ Reason = monitor_members_trigger_culling_and_return_reason(),
+ %% we need to wait for the message to arrive before deleting handler
+ timer:sleep(250),
+ error_logger:delete_report_handler(error_logger_pooler_h),
+ ok = application:stop(pooler),
+ ?assertEqual(killed, Reason),
+ ?assertEqual(1, error_logger_mon:get_msg_count())
+ end},
+ {"Default MFA shouldn't generate any reports during culling",
+ fun() ->
+ ok = application:start(pooler),
+ error_logger:add_report_handler(error_logger_pooler_h),
+ Reason = monitor_members_trigger_culling_and_return_reason(),
+ error_logger:delete_report_handler(error_logger_pooler_h),
+ ok = application:stop(pooler),
+ ?assertEqual(killed, Reason),
+ ?assertEqual(0, error_logger_mon:get_msg_count())
+ end}
+ ]}.
+
+
monitor_members_trigger_culling_and_return_reason() ->
Pids = get_n_pids(test_pool_1, 3, []),
[ erlang:monitor(process, P) || P <- Pids ],
