Merge pull request #167 from inaka/elbrujohalcon.167.wpoo_stats___mig…

…ht_crash_after

wpool:stats() might crash after a worker is killed

elvis.config

@@ -41,6 +41,10 @@
             , dont_repeat_yourself
             , #{min_complexity => 13}
             }
+          , { elvis_style
+            , line_length
+            , #{limit => 100}
+            }
           ]
        },
       #{dirs => ["."],

rebar.config

@@ -19,19 +19,18 @@
            , warn_exported_vars
            , warn_missing_spec
            , warn_untyped_record
-           , debug_info]}.
+           , debug_info
+           ]}.
 
 {profiles, [{test, [{deps, [ {katana_test,  "1.0.1"}
                            , {katana,       "0.4.0"}
-                           , {mixer,        "1.0.1", {pkg, inaka_mixer}}
-                           , {meck,         "0.8.11"}
+                           , {mixer,        "1.1.0", {pkg, inaka_mixer}}
+                           , {meck,         "0.8.13"}
                            ]
                    }]
            }]
 }.
 
-%% == Common Test ==
-
 {ct_compile_opts, [ warn_unused_vars
                   , warn_export_all
                   , warn_shadow_vars
@@ -51,19 +50,16 @@
 
 {ct_opts, []}.
 
-%% == Cover ==
+{alias, [{test, [dialyzer, ct, cover]}]}.
 
-{plugins                , [coveralls,
-                           {rebar3_codecov, "0.1.0"}
-                          ]}.
+{plugins, [ coveralls
+          , {rebar3_codecov, "0.1.0"}
+          ]}.
 
 {cover_enabled          , true}.
 {cover_export_enabled   , true}.
 
-{provider_hooks,
- [
-  {post, [{ct, {codecov, analyze}}]}
- ]}.
+{provider_hooks, [{post, [{ct, {codecov, analyze}}]}]}.
 
 {cover_opts, [verbose]}.
 
@@ -77,16 +73,15 @@
             , {subpackages, false}
             ]}.
 
-{dialyzer, [
-    {warnings, [ race_conditions
-               , no_return
-               , unmatched_returns
-               , error_handling
-               , unknown
-               ]},
-    {plt_apps, all_deps},
-    {plt_extra_apps, [erts, kernel, stdlib]},
-    {plt_location, local},
-    {base_plt_apps, [stdlib, kernel]},
-    {base_plt_location, global}
-]}.
+{dialyzer,  [ {warnings,  [ race_conditions
+                          , no_return
+                          , unmatched_returns
+                          , error_handling
+                          , unknown
+                          ]}
+            , {plt_apps, all_deps}
+            , {plt_extra_apps, [erts, kernel, stdlib]}
+            , {plt_location, local}
+            , {base_plt_apps, [stdlib, kernel]}
+            , {base_plt_location, global}
+            ]}.

src/wpool_pool.erl

@@ -187,20 +187,15 @@ stats(Wpool, Sup) ->
   {Total, WorkerStats} =
     lists:foldl(
       fun(N, {T, L}) ->
-        case erlang:whereis(worker_name(Sup, N)) of
+        case worker_info(Sup, N,  [ message_queue_len
+                                  , memory
+                                  , current_function
+                                  , current_location
+                                  , dictionary
+                                  ]) of
           undefined ->
             {T, L};
-          Worker ->
-            [{message_queue_len, MQL} = MQLT,
-             Memory, Function, Location, {dictionary, Dictionary}] =
-              erlang:process_info(
-                Worker,
-                [ message_queue_len
-                , memory
-                , current_function
-                , current_location
-                , dictionary
-                ]),
+          [{message_queue_len, MQL} = MQLT, Memory, Function, Location, {dictionary, Dictionary}] ->
             WS = [MQLT, Memory] ++
               function_location(Function, Location) ++
               task(proplists:get_value(wpool_task, Dictionary)),
@@ -217,6 +212,14 @@ stats(Wpool, Sup) ->
   , {workers,                  WorkerStats}
   ].
 
+worker_info(Sup, N, Info) ->
+  case erlang:whereis(worker_name(Sup, N)) of
+    undefined ->
+      undefined;
+    Worker ->
+      erlang:process_info(Worker, Info)
+  end.
+
 function_location({current_function, {gen_server, loop, _}}, _) ->
                   [];
 function_location({current_function, {erlang, hibernate, _}}, _) ->

test/sleepy_server.erl

@@ -0,0 +1,47 @@
+% This file is licensed to you under the Apache License,
+% Version 2.0 (the "License"); you may not use this file
+% except in compliance with the License.  You may obtain
+% a copy of the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing,
+% software distributed under the License is distributed on an
+% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+% KIND, either express or implied.  See the License for the
+% specific language governing permissions and limitations
+% under the License.
+%% @doc a gen_server built to test wpool_process
+-module(sleepy_server).
+-author('[email protected]').
+
+-behaviour(gen_server).
+
+%% gen_server callbacks
+-export([ init/1
+        , handle_call/3
+        , handle_cast/2
+        ]).
+
+-dialyzer([no_behaviours]).
+
+%%%===================================================================
+%%% callbacks
+%%%===================================================================
+-spec init(pos_integer()) -> {ok, state}.
+init(TimeToSleep) ->
+  ct:pal("Waiting ~pms to return...", [TimeToSleep]),
+  _ = timer:sleep(TimeToSleep),
+  ct:pal("Done waiting ~pms", [TimeToSleep]),
+  {ok, state}.
+
+-spec handle_cast(pos_integer(), State) -> {noreply, State}.
+handle_cast(TimeToSleep, State) ->
+  _ = timer:sleep(TimeToSleep),
+  {noreply, State}.
+
+-type from() :: {pid(), reference()}.
+-spec handle_call(pos_integer(), from(), State) -> {reply, ok, State}.
+handle_call(TimeToSleep, _From, State) ->
+  _ = timer:sleep(TimeToSleep),
+  {reply, ok, State}.

test/wpool_SUITE.erl

@@ -34,13 +34,14 @@
         , default_options/1
         , complete_coverage/1
         , broadcast/1
+        , worker_killed_stats/1
         ]).
 
 -spec all() -> [atom()].
 all() ->
   [too_much_overrun, overrun, stop_pool, non_brutal_shutdown, stats,
    default_strategy, default_options, complete_coverage, broadcast,
-   kill_on_overrun].
+   kill_on_overrun, worker_killed_stats].
 
 -spec init_per_suite(config()) -> config().
 init_per_suite(Config) ->
@@ -379,6 +380,29 @@ broadcast(_Config) ->
   meck:unload(x),
   {comment, []}.
 
+-spec worker_killed_stats(config()) -> {comment, []}.
+worker_killed_stats(_Config) ->
+  %% Each server will take 100ms to start, but the start_sup_pool/2 call is synchronous anyway
+  {ok, PoolPid} = wpool:start_sup_pool(
+    wpool_SUITE_worker_killed_stats, [{workers, 3}, {worker, {sleepy_server, 500}}]),
+  true = erlang:is_process_alive(PoolPid),
+
+  Workers = fun() -> lists:keyfind(workers, 1, wpool:stats(wpool_SUITE_worker_killed_stats)) end,
+  WorkerName = wpool_pool:worker_name(wpool_SUITE_worker_killed_stats, 1),
+
+  ct:comment("wpool:stats/1 should work normally"),
+  {workers, [_, _, _]} = Workers(),
+
+  ct:comment("wpool:stats/1 should work even if a process just dies and it's not yet back alive"),
+  exit(whereis(WorkerName), kill),
+  {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),
+  {workers, [_, _, _]} = Workers(),
+
+  {comment, []}.
+
 %% =============================================================================
 %% Helpers
 %% =============================================================================