From 5f7273b4b4b7686dc95783635a7e4bd016e164a0 Mon Sep 17 00:00:00 2001 From: elbrujohalcon Date: Wed, 26 Aug 2026 16:34:38 +0200 Subject: [PATCH] brujo-otp29 Fully move into OTP28/29 --- .github/workflows/erlang.yml | 12 +- elvis.config | 17 ++- rebar.config | 14 +-- src/wpool.erl | 4 +- src/wpool_pool.erl | 37 ++++-- src/wpool_process.erl | 6 +- src/wpool_process_callbacks.erl | 14 +-- src/wpool_process_sup.erl | 2 +- src/wpool_queue_manager.erl | 10 +- src/wpool_sup.erl | 6 +- src/wpool_time_checker.erl | 8 +- test/crashy_server.erl | 2 +- test/echo_server.erl | 2 +- test/wpool_SUITE.erl | 32 +++--- test/wpool_bench.erl | 2 +- test/wpool_pool_SUITE.erl | 153 ++++++++++++------------- test/wpool_process_SUITE.erl | 29 ++--- test/wpool_process_callbacks_SUITE.erl | 34 ++++-- test/wpool_test_utils.erl | 36 ++++++ test/wpool_worker_SUITE.erl | 3 +- 20 files changed, 249 insertions(+), 174 deletions(-) create mode 100644 test/wpool_test_utils.erl diff --git a/.github/workflows/erlang.yml b/.github/workflows/erlang.yml index 569c2cf..ddfac15 100644 --- a/.github/workflows/erlang.yml +++ b/.github/workflows/erlang.yml @@ -15,22 +15,22 @@ jobs: name: OTP:${{matrix.otp}} strategy: matrix: - otp: ['28', '27'] - rebar3: ['3.26'] + otp: ['28', '29'] + rebar3: ['3.27'] steps: - - uses: actions/checkout@v6 + - uses: actions/checkout@v7 - uses: erlef/setup-beam@v1 id: setup-beam with: otp-version: ${{matrix.otp}} rebar3-version: ${{matrix.rebar3}} - name: Restore _build - uses: actions/cache@v5 + uses: actions/cache@v6 with: path: _build key: _build-cache-for-os-${{runner.os}}-otp-${{steps.setup-beam.outputs.otp-version}}-rebar3-${{steps.setup-beam.outputs.rebar3-version}}-hash-${{hashFiles('rebar.lock')}} - name: Restore rebar3's cache - uses: actions/cache@v5 + uses: actions/cache@v6 with: path: ~/.cache/rebar3 key: rebar3-cache-for-os-${{runner.os}}-otp-${{steps.setup-beam.outputs.otp-version}}-rebar3-${{steps.setup-beam.outputs.rebar3-version}}-hash-${{hashFiles('rebar.lock')}} @@ -41,7 +41,7 @@ jobs: - name: Run tests and verifications run: rebar3 test - name: Upload code coverage - uses: codecov/codecov-action@v5 + uses: codecov/codecov-action@v7 with: files: ./_build/test/covertool/worker_pool.covertool.xml token: ${{ secrets.CODECOV_TOKEN }} diff --git a/elvis.config b/elvis.config index 9d9f136..7b39282 100644 --- a/elvis.config +++ b/elvis.config @@ -2,27 +2,32 @@ {config, [ #{ files => ["src/**/*.erl"], - ruleset => erl_files, + ruleset => erl_files_strict, rules => [ + {elvis_style, no_macros, disable}, {elvis_style, no_invalid_dynamic_calls, #{ ignore => [wpool_process, wpool_time_checker] }}, {elvis_style, no_god_modules, #{limit => 30}}, {elvis_style, state_record_and_type, disable}, - {elvis_style, dont_repeat_yourself, #{ - ignore => [wpool_SUITE], min_complexity => 13 - }} + {elvis_style, dont_repeat_yourself, #{min_complexity => 13}}, + {elvis_style, max_function_length, disable}, + {elvis_style, max_function_clause_length, #{max_length => 75}} ] }, #{ files => ["test/*.erl"], - ruleset => erl_files, + ruleset => erl_files_strict, rules => [ {elvis_style, no_debug_call, disable}, {elvis_style, state_record_and_type, disable}, - {elvis_style, dont_repeat_yourself, #{min_complexity => 13}} + {elvis_style, dont_repeat_yourself, #{min_complexity => 13}}, + {elvis_style, max_module_length, disable}, + {elvis_style, max_function_length, disable}, + {elvis_style, max_function_clause_length, disable}, + {elvis_style, abc_size, disable} ] } ]} diff --git a/rebar.config b/rebar.config index 4ec387b..0f48700 100644 --- a/rebar.config +++ b/rebar.config @@ -12,10 +12,10 @@ {cover_export_enabled, true}, {cover_opts, [verbose, {min_coverage, 92}]}, {ct_opts, [{verbose, true}]}, - {deps, [{katana, "1.0.0"}, {mixer, "1.2.0", {pkg, inaka_mixer}}, {meck, "~> 1.1"}]}, + {deps, [{mixer, "2.0.0", {pkg, inaka_mixer}}, {meck, "~> 1.2"}]}, {dialyzer, [ {warnings, [no_return, unmatched_returns, error_handling, underspecs, unknown]}, - {plt_extra_apps, [common_test, katana, meck]} + {plt_extra_apps, [common_test, meck]} ]} ]} ]}. @@ -40,11 +40,11 @@ %% == Dependencies and plugins == {project_plugins, [ - {rebar3_hank, "~> 1.6"}, - {rebar3_depup, "~> 0.4.0"}, - {rebar3_hex, "~> 7.1"}, - {rebar3_ex_doc, "~> 0.2"}, - {rebar3_lint, "~> 5.0"}, + {rebar3_hank, "~> 2.0.0"}, + {rebar3_depup, "~> 1.0.0"}, + {rebar3_hex, "~> 7.2"}, + {rebar3_ex_doc, "~> 0.3"}, + {rebar3_lint, "~> 6.0"}, {erlfmt, "~> 1.8"}, {covertool, "~> 2.0"} ]}. diff --git a/src/wpool.erl b/src/wpool.erl index 289f7b9..2e0a88d 100644 --- a/src/wpool.erl +++ b/src/wpool.erl @@ -348,12 +348,12 @@ stop() -> %% BEHAVIOUR CALLBACKS %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% @private --spec start(any(), any()) -> supervisor:startlink_ret(). +-spec start(term(), term()) -> supervisor:startlink_ret(). start(_StartType, _StartArgs) -> wpool_sup:start_link(). %% @private --spec stop(any()) -> ok. +-spec stop(term()) -> ok. stop(_State) -> ok. diff --git a/src/wpool_pool.erl b/src/wpool_pool.erl index 4c59689..2ea0194 100644 --- a/src/wpool_pool.erl +++ b/src/wpool_pool.erl @@ -21,6 +21,27 @@ %%% `t:wpool:pool_sup_intensity()' options respectively. -module(wpool_pool). +%% We use <- to filter and we use list_to_atom/1 to build atoms from other atoms. +-elvis( + [ + {elvis_style, prefer_strict_generators, disable}, + {elvis_style, no_common_caveats_call, #{ + caveat_functions => [ + {timer, send_after, 2}, + {timer, send_after, 3}, + {timer, send_interval, 2}, + {timer, send_interval, 3}, + {erlang, size, 1}, + {gen_statem, call, 2}, + {gen_server, call, 2}, + {gen_event, call, 3}, + {erlang, binary_to_atom, 1}, + {erlang, binary_to_atom, 2} + ] + }} + ] +). + -include_lib("kernel/include/logger.hrl"). -behaviour(supervisor). @@ -145,7 +166,7 @@ run_with_available_worker(Name, Run, Timeout) -> %% The timeout provided includes the time it takes to get a worker %% and for it to process the call. %% @throws no_workers | timeout --spec call_available_worker(wpool:name(), any(), timeout()) -> any(). +-spec call_available_worker(wpool:name(), term(), timeout()) -> term(). call_available_worker(Name, Call, Timeout) -> case find_wpool(Name) of undefined -> @@ -163,7 +184,7 @@ call_available_worker(Name, Call, Timeout) -> %% @doc Picks the first available worker and sends the request to it. %% The timeout provided considers only the time it takes to get a worker --spec send_request_available_worker(wpool:name(), any(), timeout()) -> +-spec send_request_available_worker(wpool:name(), term(), timeout()) -> noproc | timeout | gen_server:request_id(). send_request_available_worker(Name, Call, Timeout) -> wpool_queue_manager:send_request_available_worker( @@ -233,7 +254,7 @@ all() -> Name || {{?MODULE, Name}, _} <- persistent_term:get(), is_atom(Name), - find_wpool(Name) /= undefined + find_wpool(Name) =/= undefined ]. %% @doc Retrieves the list of worker registered names. @@ -245,7 +266,7 @@ get_workers(Name) -> %% @doc Retrieves the pool stats for all pools -spec stats() -> [wpool:stats()]. stats() -> - [stats(Name) || Name <- all()]. + lists:map(fun stats/1, all()). %% @doc Retrieves a snapshot of the pool stats %% @throws no_workers @@ -350,10 +371,10 @@ remove_callback_module(Pool, Module) -> %% @doc Get values from the worker pool record. Useful when using a custom %% strategy function. -spec wpool_get - (atom(), wpool()) -> any(); - ([atom()], wpool()) -> any(). + (atom(), wpool()) -> term(); + ([atom()], wpool()) -> term(). wpool_get(List, Wpool) when is_list(List) -> - [g(Atom, Wpool) || Atom <- List]; + [g(Atom, Wpool) || Atom <:- List]; wpool_get(Atom, Wpool) when is_atom(Atom) -> g(Atom, Wpool). @@ -556,7 +577,7 @@ all_workers(Name) -> store_wpool(Name, Size, Options) -> Atomic = atomics:new(1, [{signed, false}]), atomics:put(Atomic, 1, 1), - WorkerNames = list_to_tuple([worker_name(Name, I) || I <- lists:seq(1, Size)]), + WorkerNames = list_to_tuple([worker_name(Name, I) || I <:- lists:seq(1, Size)]), Wpool = #wpool{ name = Name, diff --git a/src/wpool_process.erl b/src/wpool_process.erl index e32c077..efe12da 100644 --- a/src/wpool_process.erl +++ b/src/wpool_process.erl @@ -191,7 +191,7 @@ terminate(Reason, State) -> end. %% @private --spec code_change(string() | {down, string()}, state(), any()) -> +-spec code_change(string() | {down, string()}, state(), term()) -> {ok, state()} | {error, term()}. code_change(OldVsn, #state{mod = #callback_cache{module = Mod}} = State, Extra) -> case erlang:function_exported(Mod, code_change, 3) of @@ -207,7 +207,7 @@ code_change(OldVsn, #state{mod = #callback_cache{module = Mod}} = State, Extra) end. %% @private --spec handle_info(any(), state()) -> +-spec handle_info(term(), state()) -> {noreply, state()} | {noreply, state(), next_step()} | {stop, term(), state()}. handle_info(Info, #state{mod = CbCache} = State) -> #callback_cache{module = Mod, handle_info = HandleInfo} = CbCache, @@ -235,7 +235,7 @@ handle_info(Info, #state{mod = CbCache} = State) -> end. %% @private --spec handle_continue(any(), state()) -> +-spec handle_continue(term(), state()) -> {noreply, state()} | {noreply, state(), next_step()} | {stop, term(), state()}. diff --git a/src/wpool_process_callbacks.erl b/src/wpool_process_callbacks.erl index 4b442b0..cfba517 100644 --- a/src/wpool_process_callbacks.erl +++ b/src/wpool_process_callbacks.erl @@ -18,9 +18,9 @@ -export_type([event/0]). --callback handle_init_start(wpool:name()) -> any(). --callback handle_worker_creation(wpool:name()) -> any(). --callback handle_worker_death(wpool:name(), term()) -> any(). +-callback handle_init_start(wpool:name()) -> term(). +-callback handle_worker_creation(wpool:name()) -> term(). +-callback handle_worker_death(wpool:name(), term()) -> term(). -optional_callbacks([ handle_init_start/1, @@ -34,7 +34,7 @@ init(Module) -> {ok, Module}. %% @private --spec handle_event({event(), [any()]}, state()) -> {ok, state()}. +-spec handle_event({event(), [term()]}, state()) -> {ok, state()}. handle_event({Event, Args}, Module) -> call(Module, Event, Args), {ok, Module}. @@ -45,14 +45,14 @@ handle_call(Msg, State) -> {ok, {error, {unexpected_call, Msg}}, State}. %% @doc Sends a notification to all registered callback modules. --spec notify(event(), undefined | atom(), [any()]) -> ok. +-spec notify(event(), undefined | atom(), [term()]) -> ok. notify(_, undefined, _) -> ok; notify(Event, EventMgr, Args) -> gen_event:notify(EventMgr, {Event, Args}). %% @doc Adds a callback module. --spec add_callback_module(wpool:name(), module()) -> ok | {error, any()}. +-spec add_callback_module(wpool:name(), module()) -> ok | {error, term()}. add_callback_module(EventManager, Module) -> case ensure_loaded(Module) of ok -> @@ -62,7 +62,7 @@ add_callback_module(EventManager, Module) -> end. %% @doc Removes a callback module. --spec remove_callback_module(wpool:name(), module()) -> ok | {error, any()}. +-spec remove_callback_module(wpool:name(), module()) -> ok | {error, term()}. remove_callback_module(EventManager, Module) -> gen_event:delete_handler(EventManager, {wpool_process_callbacks, Module}, Module). diff --git a/src/wpool_process_sup.erl b/src/wpool_process_sup.erl index bbaf228..df4e4a6 100644 --- a/src/wpool_process_sup.erl +++ b/src/wpool_process_sup.erl @@ -50,7 +50,7 @@ init({Name, Options}) -> type => worker, modules => [Worker] } - || I <- lists:seq(1, Workers) + || I <:- lists:seq(1, Workers) ], {ok, {Strategy, WorkerSpecs}}. diff --git a/src/wpool_queue_manager.erl b/src/wpool_queue_manager.erl index a7526ac..ccee45c 100644 --- a/src/wpool_queue_manager.erl +++ b/src/wpool_queue_manager.erl @@ -91,7 +91,7 @@ run_with_available_worker(QueueManager, Call, Timeout) -> end. %% @doc returns the first available worker in the pool --spec call_available_worker(queue_mgr(), any(), timeout()) -> noproc | timeout | any(). +-spec call_available_worker(queue_mgr(), term(), timeout()) -> noproc | timeout | term(). call_available_worker(QueueManager, Call, Timeout) -> case get_available_worker(QueueManager, Call, Timeout) of {ok, Worker, TimeLeft} when TimeLeft > 0 -> @@ -112,7 +112,7 @@ cast_to_available_worker(QueueManager, Cast) -> gen_server:cast(QueueManager, {cast_to_available_worker, Cast}). %% @doc returns the first available worker in the pool --spec send_request_available_worker(queue_mgr(), any(), timeout()) -> +-spec send_request_available_worker(queue_mgr(), term(), timeout()) -> noproc | timeout | gen_server:request_id(). send_request_available_worker(QueueManager, Call, Timeout) -> case get_available_worker(QueueManager, Call, Timeout) of @@ -147,7 +147,7 @@ worker_dead(QueueManager, Worker) -> -spec pending_task_count(queue_mgr()) -> non_neg_integer(). pending_task_count(QueueManager) -> try - gen_server:call(QueueManager, pending_task_count) + gen_server:call(QueueManager, pending_task_count, 5000) catch _:{noproc, _} -> 0 @@ -246,7 +246,7 @@ handle_call({available_worker, ExpiresAt}, {ClientPid, _Ref} = Client, State) -> handle_call(pending_task_count, _From, State) -> {reply, get(pending_tasks), State}. --spec handle_info(any(), state()) -> {noreply, state()}. +-spec handle_info(term(), state()) -> {noreply, state()}. handle_info({'DOWN', Ref, Type, {Worker, _Node}, Exit}, State) -> handle_info({'DOWN', Ref, Type, Worker, Exit}, State); handle_info({'DOWN', _, _, Worker, Exit}, #state{monitors = Mons} = State) -> @@ -263,7 +263,7 @@ handle_info(_Info, State) -> %%%=================================================================== %%% private %%%=================================================================== --spec get_available_worker(queue_mgr(), any(), timeout()) -> +-spec get_available_worker(queue_mgr(), term(), timeout()) -> noproc | timeout | {ok, atom(), timeout()}. get_available_worker(QueueManager, Call, Timeout) -> ExpiresAt = expires(Timeout), diff --git a/src/wpool_sup.erl b/src/wpool_sup.erl index 9522cfe..fd28801 100644 --- a/src/wpool_sup.erl +++ b/src/wpool_sup.erl @@ -27,7 +27,7 @@ %% @doc Starts the supervisor -spec start_link() -> supervisor:startlink_ret(). start_link() -> - supervisor:start_link({local, ?MODULE}, ?MODULE, []). + supervisor:start_link({local, ?MODULE}, ?MODULE, #{}). %% @doc Starts a new pool -spec start_pool(wpool:name(), wpool:options()) -> supervisor:startchild_ret(). @@ -55,8 +55,8 @@ stop_pool(Name) -> %%---------------------------------------------------------------------- %% Supervisor behaviour callbacks %%---------------------------------------------------------------------- --spec init([]) -> {ok, {{simple_one_for_one, 5, 60}, [supervisor:child_spec()]}}. -init([]) -> +-spec init(#{}) -> {ok, {{simple_one_for_one, 5, 60}, [supervisor:child_spec()]}}. +init(#{}) -> {ok, {{simple_one_for_one, 5, 60}, [ {wpool_pool, {wpool_pool, start_link, []}, permanent, 2000, supervisor, dynamic} diff --git a/src/wpool_time_checker.erl b/src/wpool_time_checker.erl index 80c347a..6343ccd 100644 --- a/src/wpool_time_checker.erl +++ b/src/wpool_time_checker.erl @@ -48,7 +48,7 @@ start_link(WPool, Name, Handler) when is_tuple(Handler) -> %% @private -spec add_handler(atom(), handler()) -> ok. add_handler(Name, Handler) -> - gen_server:call(Name, {add_handler, Handler}). + gen_server:call(Name, {add_handler, Handler}, 5000). %%%=================================================================== %%% simple callbacks @@ -71,7 +71,7 @@ handle_call({add_handler, Handler}, _, #state{handlers = Handlers} = State) -> %%% real (i.e. interesting) callbacks %%%=================================================================== %% @private --spec handle_info(any(), state()) -> {noreply, state()}. +-spec handle_info(term(), state()) -> {noreply, state()}. handle_info({check, Pid, TaskId, Runtime, WarningsLeft}, State) -> case erlang:process_info(Pid, dictionary) of {dictionary, Values} -> @@ -124,7 +124,7 @@ decrease_warnings(N) -> N - 1. -spec send_reports( - [{atom(), atom()}], + [{module(), atom()}], atom(), atom(), pid(), @@ -134,7 +134,7 @@ decrease_warnings(N) -> ok. send_reports(Handlers, Alert, Pool, Pid, Task, Runtime) -> Args = [{alert, Alert}, {pool, Pool}, {worker, Pid}, {task, Task}, {runtime, Runtime}], - _ = [safe_apply(Mod, Fun, Args) || {Mod, Fun} <- Handlers], + _ = [safe_apply(Mod, Fun, Args) || {Mod, Fun} <:- Handlers], ok. safe_apply(Mod, Fun, Args) -> diff --git a/test/crashy_server.erl b/test/crashy_server.erl index e0bf9a3..394e8ed 100644 --- a/test/crashy_server.erl +++ b/test/crashy_server.erl @@ -39,7 +39,7 @@ init(Something) -> terminate(Reason, _State) -> Reason. --spec code_change(string(), State, any()) -> {ok, State}. +-spec code_change(string(), State, term()) -> {ok, State}. code_change(_OldVsn, State, _Extra) -> {ok, State}. diff --git a/test/echo_server.erl b/test/echo_server.erl index 2965374..80e6d77 100644 --- a/test/echo_server.erl +++ b/test/echo_server.erl @@ -46,7 +46,7 @@ init(Something) -> terminate(Reason, _State) -> Reason. --spec code_change(string(), State, any()) -> any() | {ok, State}. +-spec code_change(string(), State, term()) -> term() | {ok, State}. code_change(_OldVsn, _State, Extra) -> Extra. diff --git a/test/wpool_SUITE.erl b/test/wpool_SUITE.erl index 3633de4..58d2b28 100644 --- a/test/wpool_SUITE.erl +++ b/test/wpool_SUITE.erl @@ -117,7 +117,7 @@ too_much_overrun(_Config) -> ct:comment("Start a long running task..."), ok = wpool:cast(wpool_SUITE_too_much_overrun, {timer, sleep, [5000]}), TaskId = - ktn_task:wait_for_success(fun() -> + wpool_test_utils:wait_for_success(fun() -> {dictionary, Dict} = erlang:process_info(Worker, dictionary), {TId, _, _} = proplists:get_value(wpool_task, Dict), TId @@ -350,15 +350,15 @@ stats(_Config) -> 0 = Get(message_queue_len, WorkerStats), [] = lists:keydelete(message_queue_len, 1, lists:keydelete(memory, 1, WorkerStats)) end - || I <- lists:seq(1, 10) + || I <:- lists:seq(1, 10) ], % Start a long task on every worker Sleep = {timer, sleep, [2000]}, - [wpool:cast(wpool_SUITE_stats_pool, Sleep, next_worker) || _ <- lists:seq(1, 10)], + [wpool:cast(wpool_SUITE_stats_pool, Sleep, next_worker) || _ <:- lists:seq(1, 10)], ok = - ktn_task:wait_for_success(fun() -> + wpool_test_utils:wait_for_success(fun() -> WorkingStats = wpool:stats(wpool_SUITE_stats_pool), wpool_SUITE_stats_pool = Get(pool, WorkingStats), PoolPid = Get(supervisor, WorkingStats), @@ -376,7 +376,7 @@ stats(_Config) -> {cast, Sleep} = Get(task, WorkerStats), true = is_number(Get(runtime, WorkerStats)) end - || I <- lists:seq(1, 10) + || I <:- lists:seq(1, 10) ], ok end), @@ -384,7 +384,7 @@ stats(_Config) -> wpool:stop_sup_pool(wpool_SUITE_stats_pool), no_workers = - ktn_task:wait_for( + wpool_test_utils:wait_for( fun() -> try wpool:stats(wpool_SUITE_stats_pool) @@ -460,7 +460,7 @@ broadcall(_Config) -> % Broadcall x:x() execution to workers. {[_ | _], _} = wpool:broadcall(Pool, {x, x, []}, infinity), % Give some time for the workers to perform the calls. - WorkersCount = ktn_task:wait_for(fun() -> meck:num_calls(x, x, '_') end, WorkersCount), + WorkersCount = wpool_test_utils:wait_for(fun() -> meck:num_calls(x, x, '_') end, WorkersCount), ct:comment("Check they all are \"working\""), try @@ -492,7 +492,7 @@ broadcast(_Config) -> % Broadcast x:x() execution to workers. wpool:broadcast(Pool, {x, x, []}), % Give some time for the workers to perform the calls. - WorkersCount = ktn_task:wait_for(fun() -> meck:num_calls(x, x, '_') end, WorkersCount), + WorkersCount = wpool_test_utils:wait_for(fun() -> meck:num_calls(x, x, '_') end, WorkersCount), ct:comment("Check they all are \"working\""), % Make all the workers sleep for 1.5 seconds @@ -547,7 +547,7 @@ worker_killed_stats(_Config) -> {workers, [_, _]} = Workers(), ct:comment("Once the process is alive again, we should see it at the stats"), - true = ktn_task:wait_for(fun() -> is_pid(whereis(WorkerName)) end, true, 10, 75), + true = wpool_test_utils:wait_for(fun() -> is_pid(whereis(WorkerName)) end, true, 10, 75), {workers, [_, _, _]} = Workers(), {comment, []}. @@ -596,7 +596,7 @@ pool_of_supervisors(_Config) -> Supervisors = wpool:get_workers(pool_of_supervisors), [ 3 = proplists:get_value(active, supervisor:count_children(Supervisor)) - || Supervisor <- Supervisors + || Supervisor <:- Supervisors ], {comment, "Nicely load-balanced childrens across supervisors"}. @@ -605,13 +605,17 @@ pool_of_supervisors(_Config) -> %% Helpers %% ============================================================================= get_time_checker(PoolPid) -> - [TCPid] = - [P || {_, P, worker, [wpool_time_checker]} <- supervisor:which_children(PoolPid)], + [TCPid] = [ + P + || {_, P, _, [M]} <:- supervisor:which_children(PoolPid), M =:= wpool_time_checker + ], TCPid. get_queue_manager(PoolPid) -> - [QMPid] = - [P || {_, P, worker, [wpool_queue_manager]} <- supervisor:which_children(PoolPid)], + [QMPid] = [ + P + || {_, P, _, [M]} <:- supervisor:which_children(PoolPid), M =:= wpool_queue_manager + ], QMPid. get_messages(MaxTimeout) -> diff --git a/test/wpool_bench.erl b/test/wpool_bench.erl index 4b19c37..75ec2a6 100644 --- a/test/wpool_bench.erl +++ b/test/wpool_bench.erl @@ -10,7 +10,7 @@ ) -> float(). run_tasks(TaskGroups, Strategy, Options) -> - Tasks = lists:flatten([lists:duplicate(N, Type) || {Type, N} <- TaskGroups]), + Tasks = lists:flatten([lists:duplicate(N, Type) || {Type, N} <:- TaskGroups]), {ok, _Pool} = wpool:start_sup_pool(?MODULE, Options), try lists:foldl(fun(Task, Acc) -> run_task(Task, Strategy, Acc) end, [], Tasks) of [] -> diff --git a/test/wpool_pool_SUITE.erl b/test/wpool_pool_SUITE.erl index 09a68e5..6d15925 100644 --- a/test/wpool_pool_SUITE.erl +++ b/test/wpool_pool_SUITE.erl @@ -21,8 +21,6 @@ -export_type([config/0]). --define(WORKERS, 6). - -export([all/0]). -export([init_per_suite/1, end_per_suite/1, init_per_testcase/2, end_per_testcase/2]). -export([ @@ -49,7 +47,8 @@ all() -> [ Fun - || {Fun, 1} <- module_info(exports), + || {Fun, Arity} <:- module_info(exports), + Arity =:= 1, not lists:member(Fun, [init_per_suite, end_per_suite, module_info]) ]. @@ -74,25 +73,32 @@ init_per_testcase(queue_type_fifo = TestCase, Config) -> {ok, _} = wpool:start_pool(TestCase, [{workers, 1}, {queue_type, fifo}]), Config; init_per_testcase(TestCase, Config) -> - {ok, _} = wpool:start_pool(TestCase, [{workers, ?WORKERS}]), + {ok, _} = wpool:start_pool(TestCase, [{workers, workers()}]), Config. -spec end_per_testcase(atom(), config()) -> config(). end_per_testcase(TestCase, Config) -> - catch wpool:stop_sup_pool(TestCase), + try + wpool:stop_sup_pool(TestCase) + catch + _:_ -> no_problem + end, Config. -spec stop_worker(config()) -> {comment, []}. stop_worker(_Config) -> - true = undefined /= wpool_pool:find_wpool(stop_worker), + true = undefined =/= wpool_pool:find_wpool(stop_worker), true = wpool:stop_pool(stop_worker), - undefined = ktn_task:wait_for(fun() -> wpool_pool:find_wpool(stop_worker) end, undefined), + undefined = wpool_test_utils:wait_for( + fun() -> wpool_pool:find_wpool(stop_worker) end, undefined + ), true = wpool:stop_pool(stop_worker), undefined = wpool_pool:find_wpool(stop_worker), {comment, ""}. -spec available_worker(config()) -> {comment, []}. available_worker(_Config) -> + NumWorkers = workers(), Pool = available_worker, Run = fun(Worker, Timeout) -> gen_server:call(Worker, {erlang, self, []}, Timeout) end, try wpool:call(not_a_pool, x) of @@ -114,29 +120,29 @@ available_worker(_Config) -> {ok, _} = wpool:run(Pool, Run, available_worker), ct:log("Put them all to work, each request should go to a different worker"), - [wpool:cast(Pool, {timer, sleep, [5000]}) || _ <- lists:seq(1, ?WORKERS)], + [wpool:cast(Pool, {timer, sleep, [5000]}) || _ <:- lists:seq(1, NumWorkers)], - [0] = ktn_task:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [0]), + [0] = wpool_test_utils:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [0]), ct:log( "Now send another round of messages,\n" " the workers queues should still be empty" ), - [wpool:cast(Pool, {timer, sleep, [100 * I]}) || I <- lists:seq(1, ?WORKERS)], + [wpool:cast(Pool, {timer, sleep, [100 * I]}) || I <:- lists:seq(1, NumWorkers)], - % Check that we have ?WORKERS pending tasks - ?WORKERS = - ktn_task:wait_for( + % Check that we have NumWorkers pending tasks + NumWorkers = + wpool_test_utils:wait_for( fun() -> Stats1 = wpool:stats(Pool), [0] = lists:usort([ proplists:get_value(message_queue_len, WS) - || {_, WS} <- proplists:get_value(workers, Stats1) + || {_, WS} <:- proplists:get_value(workers, Stats1) ]), proplists:get_value(total_message_queue_len, Stats1) end, - ?WORKERS + NumWorkers ), ct:log("If we can't wait we get no workers"), @@ -165,11 +171,11 @@ available_worker(_Config) -> ct:log("Now they all should be free"), ct:log("We get half of them working for a while"), - [wpool:cast(Pool, {timer, sleep, [60000]}) || _ <- lists:seq(1, ?WORKERS, 2)], + [wpool:cast(Pool, {timer, sleep, [60000]}) || _ <:- lists:seq(1, NumWorkers, 2)], % Check we have no pending tasks 0 = - ktn_task:wait_for( + wpool_test_utils:wait_for( fun() -> proplists:get_value(total_message_queue_len, wpool:stats(Pool)) end, 0 ), @@ -181,19 +187,17 @@ available_worker(_Config) -> Workers = [ wpool:call(Pool, {erlang, self, []}, available_worker, 5000) - || _ <- lists:seq(1, 20 * ?WORKERS) + || _ <:- lists:seq(1, 20 * NumWorkers) ], - UniqueWorkers = - sets:to_list( - sets:from_list(Workers) - ), - {?WORKERS, UniqueWorkers, true} = - {?WORKERS, UniqueWorkers, ?WORKERS / 2 >= length(UniqueWorkers)}, + UniqueWorkers = sets:to_list(sets:from_list(Workers)), + {Workers, UniqueWorkers, true} = + {Workers, UniqueWorkers, NumWorkers / 2 >= length(UniqueWorkers)}, {comment, []}. -spec best_worker(config()) -> {comment, []}. best_worker(_Config) -> + NumWorkers = workers(), Pool = best_worker, try wpool:call(not_a_pool, x, best_worker) of Result -> @@ -210,17 +214,17 @@ best_worker(_Config) -> {reply, {ok, _}} = gen_server:wait_response(Req, 5000), %% Fill up their message queues... - [wpool:cast(Pool, {timer, sleep, [60000]}, next_worker) || _ <- lists:seq(1, ?WORKERS)], - [0] = ktn_task:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [0]), + [wpool:cast(Pool, {timer, sleep, [60000]}, next_worker) || _ <:- lists:seq(1, NumWorkers)], + [0] = wpool_test_utils:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [0]), - [wpool:cast(Pool, {timer, sleep, [60000]}, best_worker) || _ <- lists:seq(1, ?WORKERS)], + [wpool:cast(Pool, {timer, sleep, [60000]}, best_worker) || _ <:- lists:seq(1, NumWorkers)], - [1] = ktn_task:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [1]), + [1] = wpool_test_utils:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [1]), %% Now try best worker once per worker - [wpool:cast(Pool, {timer, sleep, [60000]}, best_worker) || _ <- lists:seq(1, ?WORKERS)], + [wpool:cast(Pool, {timer, sleep, [60000]}, best_worker) || _ <:- lists:seq(1, NumWorkers)], %% The load should be evenly distributed... - [2] = ktn_task:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [2]), + [2] = wpool_test_utils:wait_for(fun() -> worker_msg_queue_lengths(Pool) end, [2]), {comment, []}. @@ -242,20 +246,20 @@ next_available_worker(_Config) -> ct:log("Put them all to work..."), [ wpool:cast(Pool, {timer, sleep, [1500 + I]}, next_available_worker) - || I <- lists:seq(0, (?WORKERS - 1) * 60000, 60000) + || I <:- lists:seq(0, (workers() - 1) * 60000, 60000) ], AvailableWorkers = fun() -> length([ a_worker - || {_, WS} <- proplists:get_value(workers, wpool:stats(Pool)), - proplists:get_value(task, WS) == undefined + || {_, WS} <:- proplists:get_value(workers, wpool:stats(Pool)), + proplists:get_value(task, WS) =:= undefined ]) end, ct:log("All busy..."), - 0 = ktn_task:wait_for(AvailableWorkers, 0), + 0 = wpool_test_utils:wait_for(AvailableWorkers, 0), ct:log("No available workers..."), try wpool:cast(Pool, {timer, sleep, [60000]}, next_available_worker) of @@ -267,7 +271,7 @@ next_available_worker(_Config) -> end, ct:log("Wait until the first frees up..."), - 1 = ktn_task:wait_for(AvailableWorkers, 1), + 1 = wpool_test_utils:wait_for(AvailableWorkers, 1), Req = wpool:send_request(Pool, {erlang, self, []}, next_available_worker), {reply, {ok, _}} = gen_server:wait_response(Req, 5000), @@ -287,6 +291,7 @@ next_available_worker(_Config) -> -spec next_worker(config()) -> {comment, []}. next_worker(_Config) -> + NumWorkers = workers(), Pool = next_worker, try wpool:call(not_a_pool, x, next_worker) of @@ -304,12 +309,9 @@ next_worker(_Config) -> I = proplists:get_value(next_worker, Stats), wpool:call(Pool, {erlang, self, []}, next_worker, infinity) end - || I <- lists:seq(1, ?WORKERS) + || I <:- lists:seq(1, NumWorkers) ], - ?WORKERS = - sets:size( - sets:from_list(Res0) - ), + NumWorkers = sets:size(sets:from_list(Res0)), Res0 = [ @@ -318,7 +320,7 @@ next_worker(_Config) -> I = proplists:get_value(next_worker, Stats), wpool:call(Pool, {erlang, self, []}, next_worker) end - || I <- lists:seq(1, ?WORKERS) + || I <:- lists:seq(1, NumWorkers) ], Req = wpool:send_request(Pool, {erlang, self, []}, next_worker), @@ -331,6 +333,7 @@ next_worker(_Config) -> -spec random_worker(config()) -> {comment, []}. random_worker(_Config) -> + NumWorkers = workers(), Pool = random_worker, try wpool:call(not_a_pool, x, random_worker) of @@ -347,16 +350,16 @@ random_worker(_Config) -> %% Ask for a random worker's identity 20x more than the number of workers %% and expect to get an answer from every worker at least once. Serial = - [wpool:call(Pool, {erlang, self, []}, random_worker) || _ <- lists:seq(1, 20 * ?WORKERS)], - ?WORKERS = - sets:size( - sets:from_list(Serial) - ), + [ + wpool:call(Pool, {erlang, self, []}, random_worker) + || _ <:- lists:seq(1, 20 * NumWorkers) + ], + NumWorkers = sets:size(sets:from_list(Serial)), %% Randomly ask a lot of workers to send ourselves the atom true [ wpool:cast(Pool, {erlang, send, [self(), true]}, random_worker) - || _ <- lists:seq(1, 20 * ?WORKERS) + || _ <:- lists:seq(1, 20 * NumWorkers) ], Results = [ @@ -366,7 +369,7 @@ random_worker(_Config) -> after 5000 -> ct:fail("Didn't receive 'true' in time") end - || _ <- lists:seq(1, 20 * ?WORKERS) + || _ <:- lists:seq(1, 20 * NumWorkers) ], true = lists:all(fun(Value) -> Value end, Results), @@ -382,18 +385,16 @@ random_worker(_Config) -> WorkerId = wpool:call(Pool, {erlang, self, []}, random_worker), Self ! {worker, WorkerId} end) - || _ <- lists:seq(1, 20 * ?WORKERS) + || _ <:- lists:seq(1, 20 * NumWorkers) ], - Concurrent = collect_results(20 * ?WORKERS, []), - ?WORKERS = - sets:size( - sets:from_list(Concurrent) - ), + Concurrent = collect_results(20 * NumWorkers, []), + NumWorkers = sets:size(sets:from_list(Concurrent)), {comment, []}. -spec hash_worker(config()) -> {comment, []}. hash_worker(_Config) -> + NumWorkers = workers(), Pool = hash_worker, try wpool:call(not_a_pool, x, {hash_worker, 1}) of @@ -409,40 +410,37 @@ hash_worker(_Config) -> Targeted = [ wpool:call(Pool, {erlang, self, []}, {hash_worker, I rem 2}) - || I <- lists:seq(1, 20 * ?WORKERS) + || I <:- lists:seq(1, 20 * NumWorkers) ], - 2 = - sets:size( - sets:from_list(Targeted) - ), + 2 = sets:size(sets:from_list(Targeted)), %% Now use many different hash keys. All workers should be hit. Spread = [ wpool:call(Pool, {erlang, self, []}, {hash_worker, I}) - || I <- lists:seq(1, 20 * ?WORKERS) + || I <:- lists:seq(1, 20 * NumWorkers) ], - ?WORKERS = - sets:size( - sets:from_list(Spread) - ), + NumWorkers = sets:size(sets:from_list(Spread)), Run = fun(Worker, Timeout) -> gen_server:call(Worker, {erlang, self, []}, Timeout) end, - [{ok, _} = wpool:run(Pool, Run, {hash_worker, I}) || I <- lists:seq(1, 20 * ?WORKERS)], + [{ok, _} = wpool:run(Pool, Run, {hash_worker, I}) || I <:- lists:seq(1, 20 * NumWorkers)], %% Fill up their message queues... [ wpool:cast(Pool, {timer, sleep, [60000]}, {hash_worker, I}) - || I <- lists:seq(1, 20 * ?WORKERS) + || I <:- lists:seq(1, 20 * NumWorkers) ], false = - ktn_task:wait_for(fun() -> lists:member(0, worker_msg_queue_lengths(Pool)) end, false), + wpool_test_utils:wait_for( + fun() -> lists:member(0, worker_msg_queue_lengths(Pool)) end, false + ), {comment, []}. -spec custom_worker(config()) -> {comment, []}. custom_worker(_Config) -> + NumWorkers = workers(), Pool = custom_worker, Strategy = fun wpool_pool:next_worker/1, @@ -461,7 +459,7 @@ custom_worker(_Config) -> I = proplists:get_value(next_worker, Stats), wpool:cast(Pool, {io, format, ["ok!"]}, Strategy) end - || I <- lists:seq(1, ?WORKERS) + || I <:- lists:seq(1, NumWorkers) ], Res0 = @@ -471,12 +469,9 @@ custom_worker(_Config) -> I = proplists:get_value(next_worker, Stats), wpool:call(Pool, {erlang, self, []}, Strategy, infinity) end - || I <- lists:seq(1, ?WORKERS) + || I <:- lists:seq(1, NumWorkers) ], - ?WORKERS = - sets:size( - sets:from_list(Res0) - ), + NumWorkers = sets:size(sets:from_list(Res0)), Res0 = [ begin @@ -484,7 +479,7 @@ custom_worker(_Config) -> I = proplists:get_value(next_worker, Stats), wpool:call(Pool, {erlang, self, []}, Strategy) end - || I <- lists:seq(1, ?WORKERS) + || I <:- lists:seq(1, NumWorkers) ], Req = wpool:send_request(Pool, {erlang, self, []}, Strategy), @@ -509,7 +504,7 @@ manager_crash(_Config) -> exit(whereis(QueueManager), kill), false = - ktn_task:wait_for( + wpool_test_utils:wait_for( fun() -> lists:member(whereis(QueueManager), [OldPid, undefined]) end, false ), @@ -625,7 +620,7 @@ get_workers(_Config) -> ct:log("Verify that there's the correct number of workers"), Workers = wpool:get_workers(Pool), - ?WORKERS = length(Workers), + true = workers() =:= length(Workers), ct:log("All workers are alive"), true = lists:all(fun(Whereis) -> Whereis =/= undefined end, Workers), @@ -673,7 +668,7 @@ mess_up_with_store(_Config) -> Flag = process_flag(trap_exit, true), exit(whereis(Pool), kill), ok = - ktn_task:wait_for( + wpool_test_utils:wait_for( fun() -> try wpool:call(Pool, {io, format, ["1!~n"]}, random_worker) of X -> @@ -732,8 +727,10 @@ send_io_format(Pool) -> worker_msg_queue_lengths(Pool) -> lists:usort([ proplists:get_value(message_queue_len, WS) - || {_, WS} <- proplists:get_value(workers, wpool:stats(Pool)) + || {_, WS} <:- proplists:get_value(workers, wpool:stats(Pool)) ]). store_mess_up(Pool) -> true = persistent_term:erase({wpool_pool, Pool}). + +workers() -> 6. diff --git a/test/wpool_process_SUITE.erl b/test/wpool_process_SUITE.erl index 7d662c1..f24a49f 100644 --- a/test/wpool_process_SUITE.erl +++ b/test/wpool_process_SUITE.erl @@ -43,7 +43,8 @@ all() -> [ Fun - || {Fun, 1} <- module_info(exports), + || {Fun, Arity} <:- module_info(exports), + Arity =:= 1, not lists:member(Fun, [init_per_suite, end_per_suite, module_info]) ]. @@ -85,7 +86,7 @@ init_timeout(_Config) -> {ok, Pid} = wpool_process:start_link(?MODULE, echo_server, {ok, state, 0}, opts()), timeout = get_state(?MODULE), Pid ! {stop, normal, state}, - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. @@ -95,9 +96,9 @@ info(_Config) -> Pid ! {noreply, newstate}, newstate = get_state(?MODULE), Pid ! {noreply, newerstate, 1}, - timeout = ktn_task:wait_for(fun() -> get_state(?MODULE) end, timeout), + timeout = wpool_test_utils:wait_for(fun() -> get_state(?MODULE) end, timeout), Pid ! {stop, normal, state}, - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. @@ -107,9 +108,9 @@ cast(_Config) -> wpool_process:cast(Pid, {noreply, newstate}), newstate = get_state(?MODULE), wpool_process:cast(Pid, {noreply, newerstate, 0}), - timeout = ktn_task:wait_for(fun() -> get_state(?MODULE) end, timeout), + timeout = wpool_test_utils:wait_for(fun() -> get_state(?MODULE) end, timeout), wpool_process:cast(Pid, {stop, normal, state}), - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. @@ -122,7 +123,7 @@ send_request(_Config) -> ok2 = wait_response(Req2), Req3 = wpool_process:send_request(Pid, {stop, normal, ok3, state}), ok3 = wait_response(Req3), - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. @@ -182,11 +183,11 @@ continue(_Config) -> %% handle_continue/2 returns timeout = 0 wpool_process:cast(Pid, {noreply, state, {continue, {noreply, continue_state_g, 0}}}), - timeout = ktn_task:wait_for(fun() -> get_state(?MODULE) end, timeout), + timeout = wpool_test_utils:wait_for(fun() -> get_state(?MODULE) end, timeout), %% handle_continue/2 returns {stop, normal, state} wpool_process:cast(Pid, {noreply, state, {continue, {stop, normal, state}}}), - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. @@ -202,7 +203,7 @@ handle_info_fails(_Config) -> %% sleepy_server does not implement handle_info/2 {ok, Pid} = wpool_process:start_link(?MODULE, crashy_server, {ok, state}, opts()), Pid ! undef, - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. -spec format_status(config()) -> {comment, []}. @@ -228,9 +229,9 @@ call(_Config) -> ok1 = wpool_process:call(Pid, {reply, ok1, newstate}, 5000), newstate = get_state(?MODULE), ok2 = wpool_process:call(Pid, {reply, ok2, newerstate, 1}, 5000), - timeout = ktn_task:wait_for(fun() -> get_state(?MODULE) end, timeout), + timeout = wpool_test_utils:wait_for(fun() -> get_state(?MODULE) end, timeout), ok3 = wpool_process:call(Pid, {stop, normal, ok3, state}, 5000), - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. @@ -251,7 +252,7 @@ pool_restart_crash(_Config) -> ct:log("Check that the pool didn't die"), {error, {timeout, {badmatch, true}}} = - ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), hello = wpool:call(Pool, hello), {comment, []}. @@ -277,7 +278,7 @@ pool_norestart_crash(_Config) -> wpool:cast(Pool, crash), ct:log("Check that the pool is not working"), - false = ktn_task:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), + false = wpool_test_utils:wait_for(fun() -> erlang:is_process_alive(Pid) end, false), {comment, []}. diff --git a/test/wpool_process_callbacks_SUITE.erl b/test/wpool_process_callbacks_SUITE.erl index fc4752d..3a118fc 100644 --- a/test/wpool_process_callbacks_SUITE.erl +++ b/test/wpool_process_callbacks_SUITE.erl @@ -59,12 +59,16 @@ complete_callback_passed_when_starting_pool(_Config) -> ), WorkersCount = - ktn_task:wait_for(function_calls(callbacks, handle_init_start, ['_']), WorkersCount), + wpool_test_utils:wait_for( + function_calls(callbacks, handle_init_start, ['_']), WorkersCount + ), WorkersCount = - ktn_task:wait_for(function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount), + wpool_test_utils:wait_for( + function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount + ), Worker = wpool_pool:random_worker(Pool), Worker ! crash, - 1 = ktn_task:wait_for(function_calls(callbacks, handle_worker_death, ['_', '_']), 1), + 1 = wpool_test_utils:wait_for(function_calls(callbacks, handle_worker_death, ['_', '_']), 1), wpool:stop_pool(Pool), meck:unload(callbacks), @@ -87,7 +91,9 @@ partial_callback_passed_when_starting_pool(_Config) -> ] ), WorkersCount = - ktn_task:wait_for(function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount), + wpool_test_utils:wait_for( + function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount + ), wpool:stop_pool(Pool), meck:unload(callbacks), @@ -117,8 +123,8 @@ callback_can_be_added_and_removed_after_pool_is_started(_Config) -> Worker ! crash, %% they both are called - 1 = ktn_task:wait_for(function_calls(callbacks, handle_worker_death, ['_', '_']), 1), - 1 = ktn_task:wait_for(function_calls(callbacks2, handle_worker_death, ['_', '_']), 1), + 1 = wpool_test_utils:wait_for(function_calls(callbacks, handle_worker_death, ['_', '_']), 1), + 1 = wpool_test_utils:wait_for(function_calls(callbacks2, handle_worker_death, ['_', '_']), 1), %% then the first module is removed _ = wpool_pool:remove_callback_module(Pool, callbacks), @@ -126,8 +132,8 @@ callback_can_be_added_and_removed_after_pool_is_started(_Config) -> Worker2 ! crash, %% and only the scond one is called - 1 = ktn_task:wait_for(function_calls(callbacks, handle_worker_death, ['_', '_']), 1), - 2 = ktn_task:wait_for(function_calls(callbacks2, handle_worker_death, ['_', '_']), 2), + 1 = wpool_test_utils:wait_for(function_calls(callbacks, handle_worker_death, ['_', '_']), 1), + 2 = wpool_test_utils:wait_for(function_calls(callbacks2, handle_worker_death, ['_', '_']), 2), wpool:stop_pool(Pool), meck:unload(callbacks), @@ -159,9 +165,11 @@ crashing_callback_does_not_affect_others(_Config) -> ), WorkersCount = - ktn_task:wait_for(function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount), + wpool_test_utils:wait_for( + function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount + ), WorkersCount = - ktn_task:wait_for( + wpool_test_utils:wait_for( function_calls(callbacks2, handle_worker_creation, ['_']), WorkersCount ), @@ -192,7 +200,9 @@ non_existsing_module_does_not_affect_others(_Config) -> {error, nofile} = wpool_pool:add_callback_module(Pool, non_existing_m2), WorkersCount = - ktn_task:wait_for(function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount), + wpool_test_utils:wait_for( + function_calls(callbacks, handle_worker_creation, ['_']), WorkersCount + ), wpool:stop_pool(Pool), meck:unload(callbacks), @@ -207,5 +217,5 @@ complete_coverage(_Config) -> {ok, EventManager} = gen_event:start_link(), gen_event:add_handler(EventManager, {wpool_process_callbacks, ?MODULE}, ?MODULE), {error, {unexpected_call, call}} = - gen_event:call(EventManager, {wpool_process_callbacks, ?MODULE}, call), + gen_event:call(EventManager, {wpool_process_callbacks, ?MODULE}, call, 5000), ok. diff --git a/test/wpool_test_utils.erl b/test/wpool_test_utils.erl new file mode 100644 index 0000000..2d76193 --- /dev/null +++ b/test/wpool_test_utils.erl @@ -0,0 +1,36 @@ +-module(wpool_test_utils). + +-export([wait_for/2, wait_for/4, wait_for_success/1]). + +-type task(T) :: fun(() -> T). +-export_type([task/1]). + +-spec wait_for(task(T1), T2) -> {error, {timeout, {badmatch, T1}}} | T2 when is_subtype(T2, T1). +wait_for(Task, ExpectedAnswer) -> + wait_for(Task, ExpectedAnswer, 200, 10). + +-spec wait_for(task(T1), T2, pos_integer(), pos_integer()) -> + {error, {timeout, {badmatch, T1}}} | T2 +when + is_subtype(T2, T1). +wait_for(Task, ExpectedAnswer, SleepTime, Retries) -> + wait_for_success(fun() -> ExpectedAnswer = Task() end, SleepTime, Retries). + +-spec wait_for_success(task(T)) -> {error, {timeout, term()}} | T. +wait_for_success(Task) -> + wait_for_success(Task, 200, 10). + +-spec wait_for_success(task(T), pos_integer(), pos_integer()) -> {error, {timeout, term()}} | T. +wait_for_success(Task, SleepTime, Retries) -> + wait_for_success(Task, undefined, SleepTime, Retries). + +wait_for_success(_Task, Exception, _SleepTime, 0) -> + {error, {timeout, Exception}}; +wait_for_success(Task, _Exception, SleepTime, Retries) -> + try + Task() + catch + _:NewException -> + timer:sleep(SleepTime), + wait_for_success(Task, NewException, SleepTime, Retries - 1) + end. diff --git a/test/wpool_worker_SUITE.erl b/test/wpool_worker_SUITE.erl index 39ef0c6..159833d 100644 --- a/test/wpool_worker_SUITE.erl +++ b/test/wpool_worker_SUITE.erl @@ -30,7 +30,8 @@ all() -> [ Fun - || {Fun, 1} <- module_info(exports), + || {Fun, Arity} <:- module_info(exports), + Arity =:= 1, not lists:member(Fun, [init_per_suite, end_per_suite, module_info]) ].