Apply review

Author: NelsonVides

src/amoc_controller.erl

@@ -130,9 +130,7 @@ disable() ->
 %% @private
 -spec verify_user_rate(any()) -> boolean().
 verify_user_rate(UserRate) ->
-    (infinity =:= UserRate)
-    orelse is_integer(UserRate)
-    andalso (0 =< UserRate).
+    (is_integer(UserRate) andalso (0 =< UserRate)) orelse (infinity =:= UserRate).
 
 %% @private
 -spec update_user_rate(user_rate, user_rate()) -> ok.
@@ -255,7 +253,7 @@ handle_add(StartId, EndId, #state{status       = running,
   when StartId =< EndId, LastId < StartId ->
     amoc_telemetry:execute([controller, users], #{count => EndId - StartId + 1},
                            #{scenario => Scenario, type => add}),
-    amoc_users_sup:start_children(Scenario, {StartId, EndId}, ScenarioState),
+    amoc_users_sup:start_children(Scenario, StartId, EndId, ScenarioState),
     {ok, State#state{last_user_id = EndId}};
 handle_add(_StartId, _EndId, #state{status = running} = State) ->
     {{error, invalid_range}, State};
@@ -325,7 +323,7 @@ get_user_rate() ->
 
 -spec wait_user_rate() -> boolean().
 wait_user_rate() ->
-    0 =:= get_user_rate()
+    infinity =:= get_user_rate()
     orelse ok =:= amoc_throttle:wait(user_rate).
 
 -spec start_user_rate() -> any().

src/throttle/amoc_throttle_controller.erl

@@ -11,7 +11,7 @@
          ensure_throttle_processes_started/2,
          pause/1, resume/1, stop/1, get_info/1,
          change_rate/2, change_rate_gradually/2,
-         pg_scope/0, consume_all_timer_ticks/1,
+         pg_scope/0,
          get_throttle_process/1,
          raise_event_on_slave_node/2, telemetry_event/2]).
 

src/throttle/amoc_throttle_process.erl

@@ -143,7 +143,14 @@ maybe_stop_timer(#state{tref = undefined}) ->
     ok;
 maybe_stop_timer(#state{tref = TRef}) ->
     {ok, cancel} = timer:cancel(TRef),
-    amoc_throttle_controller:consume_all_timer_ticks(delay_between_executions).
+    consume_all_timer_ticks(delay_between_executions).
+
+-spec consume_all_timer_ticks(any()) -> ok.
+consume_all_timer_ticks(Msg) ->
+    receive
+        Msg -> consume_all_timer_ticks(Msg)
+    after 0 -> ok
+    end.
 
 timeout(#state{delay_between_executions = infinity}) ->
     infinity;

src/users/amoc_users_sup.erl

@@ -4,7 +4,7 @@
 %%
 %% It spawns a pool of workers as big as online schedulers. When starting a new user, as the user is
 %% identified by ID, a worker will be chosen for this user based on its ID
-%% (see `gen_sup_from_userid/1').
+%% (see `get_sup_from_user_id/1').
 %%
 %% The currently running number of users is stored in an atomic that all workers update and the
 %% controller can read.
@@ -17,7 +17,7 @@
 
 %% API
 -export([handle_up_user/3, handle_down_user/2, count_no_of_users/0]).
--export([start_children/3, stop_child/2, stop_children/2, terminate_all_children/0]).
+-export([start_children/4, stop_child/2, stop_children/2, terminate_all_children/0]).
 -export([distribute/2, get_all_children/0]).
 
 -type count() :: non_neg_integer().
@@ -42,8 +42,7 @@
 start_link() ->
     Ret = supervisor:start_link({local, ?MODULE}, ?MODULE, no_args),
     UserSups = supervisor:which_children(?MODULE),
-    IndexedSupsUnsorted = [ {Pid, N} || {{amoc_users_worker_sup, N}, Pid, _, _} <- UserSups,
-                                        is_integer(N), is_pid(Pid)],
+    IndexedSupsUnsorted = [ {Pid, N} || {{amoc_users_worker_sup, N}, Pid, _, _} <- UserSups ],
     IndexedSups = lists:keysort(2, IndexedSupsUnsorted),
     UserSupPidsTuple = list_to_tuple([ Pid || {Pid, _} <- IndexedSups ]),
     SupCount = tuple_size(UserSupPidsTuple),
@@ -72,6 +71,7 @@ init(no_args) ->
     Strategy = #{strategy => one_for_one, intensity => 0},
     {ok, {Strategy, Specs}}.
 
+%% We start from 2 to simplify user_count atomics management.
 indexes() ->
     lists:seq(2, erlang:system_info(schedulers_online) + 1).
 
@@ -104,7 +104,7 @@ handle_down_user(SupNum, Pid) when SupNum > 1 ->
 stop_child(Pid, Force) ->
     case ets:lookup(?TABLE, Pid) of
         [Object] ->
-            Sup = gen_sup_from_userid(Object),
+            Sup = get_sup_from_user_id(Object),
             amoc_users_worker_sup:stop_children(Sup, [Pid], Force);
         _ ->
             ok
@@ -113,11 +113,11 @@ stop_child(Pid, Force) ->
 %% Group all children based on ID to their respective worker supervisor and cast a request with each
 %% group at once. This way we reduce the number of casts to each worker to always one, instead of
 %% depending on the number of users.
--spec start_children(amoc:scenario(), {amoc_scenario:user_id(), amoc_scenario:user_id()}, any()) ->
+-spec start_children(amoc:scenario(), amoc_scenario:user_id(), amoc_scenario:user_id(), any()) ->
     ok.
-start_children(Scenario, {StartId, EndId}, ScenarioState) ->
+start_children(Scenario, StartId, EndId, ScenarioState) ->
     UserIds = lists:seq(StartId, EndId),
-    Assignments = maps:groups_from_list(fun gen_sup_from_userid/1, UserIds),
+    Assignments = maps:groups_from_list(fun get_sup_from_user_id/1, UserIds),
     CastFun = fun(Sup, Users) ->
                       amoc_users_worker_sup:start_children(Sup, Scenario, Users, ScenarioState)
               end,
@@ -147,7 +147,7 @@ terminate_all_children() ->
 
 -spec stop_children_assignments([{pid(), amoc_scenario:user_id()}], boolean()) -> ok.
 stop_children_assignments(Users, Force) ->
-    Assignments = maps:groups_from_list(fun gen_sup_from_userid/1, fun get_pid/1, Users),
+    Assignments = maps:groups_from_list(fun get_sup_from_user_id/1, fun get_pid/1, Users),
     CastFun = fun(Sup, Assignment) ->
                       amoc_users_worker_sup:stop_children(Sup, Assignment, Force)
               end,
@@ -163,10 +163,10 @@ do_terminate_all_my_children('$end_of_table') ->
     ok.
 
 %% Helpers
--spec gen_sup_from_userid({pid(), amoc_scenario:user_id()} | amoc_scenario:user_id()) -> pid().
-gen_sup_from_userid({_Pid, Id}) ->
-    gen_sup_from_userid(Id);
-gen_sup_from_userid(Id) ->
+-spec get_sup_from_user_id({pid(), amoc_scenario:user_id()} | amoc_scenario:user_id()) -> pid().
+get_sup_from_user_id({_Pid, Id}) ->
+    get_sup_from_user_id(Id);
+get_sup_from_user_id(Id) ->
     #storage{sups = Supervisors, sups_count = SupCount} = persistent_term:get(?MODULE),
     Index = erlang:phash2(Id, SupCount) + 1,
     element(Index, Supervisors).

test/controller_SUITE.erl

@@ -46,7 +46,8 @@ all_tests() ->
      stop_running_scenario_with_no_users_immediately_terminates,
      stop_running_scenario_with_users_stays_in_finished,
      stop_running_scenario_with_users_eventually_terminates,
-     user_rate_equal_zero_starts_all_users_at_once,
+     user_rate_equal_zero_starts_no_users,
+     user_rate_equal_infinity_starts_all_users_at_once,
      scenario_with_state_and_crashing_in_terminate_run_fine,
      scenario_missing_start_callback_fails,
      scenario_with_failing_init_fails
@@ -215,13 +216,25 @@ stop_running_scenario_with_users_eventually_terminates(_) ->
     WaitUntilValue = {finished, testing_scenario},
     wait_helper:wait_until(WaitUntilFun, WaitUntilValue).
 
-user_rate_equal_zero_starts_all_users_at_once(_) ->
+user_rate_equal_zero_starts_no_users(_) ->
     Vars = [{user_rate, 0}, {testing_var1, def1}
             | test_helpers:other_vars_to_keep_quiet()],
     do_start_scenario(testing_scenario, Vars),
+    amoc_controller:add_users(1, 1000),
+    ct:sleep(100),
+    Status = amoc_controller:get_status(),
+    ?assertMatch(
+       {running, #{scenario := testing_scenario,
+                   currently_running_users := 0}},
+       Status).
+
+user_rate_equal_infinity_starts_all_users_at_once(_) ->
+    Vars = [{user_rate, infinity}, {testing_var1, def1}
+            | test_helpers:other_vars_to_keep_quiet()],
+    do_start_scenario(testing_scenario, Vars),
     NumOfUsers = 1000,
     amoc_controller:add_users(1, NumOfUsers),
-    Extra = #{time_left => 25, sleep_time => 5},
+    Extra = #{time_left => 5, sleep_time => 1},
     test_helpers:wait_until_scenario_has_users(testing_scenario, NumOfUsers, NumOfUsers, Extra).
 
 scenario_with_state_and_crashing_in_terminate_run_fine(_) ->