Asynchronously register users after interarrival throttle allows

Unfortunately we cannot set the interarrival throttle to occur before
the `proc_lib:init_ack/2` call because this would block the supervisor.

This change also allows us to init processes completely asynchronously,
as proc_lib is sync

Author: NelsonVides

src/amoc_scenario.erl

@@ -91,7 +91,6 @@ start(Scenario, Id, State) ->
                {false, false} ->
                    exit("the scenario module must export either start/2 or start/1 function")
            end,
-    infinity =:= amoc_controller:get_interarrival() orelse amoc_throttle:wait(interarrival),
     telemetry:span([amoc, scenario, start], Metadata, Span).
 
 %% ------------------------------------------------------------------

src/users/amoc_user.erl

@@ -9,10 +9,9 @@
 
 -type state() :: term().
 
--spec start_link(amoc:scenario(), amoc_scenario:user_id(), state()) ->
-    {ok, pid()} | {error, term()}.
+-spec start_link(amoc:scenario(), amoc_scenario:user_id(), state()) -> {pid(), reference()}.
 start_link(Scenario, Id, State) ->
-    proc_lib:start_link(?MODULE, init, [self(), Scenario, Id, State]).
+    proc_lib:spawn_opt(?MODULE, init, [self(), Scenario, Id, State], [link, monitor]).
 
 -spec stop() -> ok.
 stop() ->
@@ -24,6 +23,7 @@ stop(Pid, Force) when is_pid(Pid) ->
 
 -spec init(pid(), amoc:scenario(), amoc_scenario:user_id(), state()) -> term().
 init(Parent, Scenario, Id, State) ->
-    proc_lib:init_ack(Parent, {ok, self()}),
     process_flag(trap_exit, true),
+    infinity =:= amoc_controller:get_interarrival() orelse amoc_throttle:wait(interarrival),
+    amoc_users_worker_sup:child_up(Parent, Id),
     amoc_scenario:start(Scenario, Id, State).

src/users/amoc_users_worker_sup.erl

@@ -17,13 +17,15 @@
 -export([start_link/1]).
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
 
--export([start_child/4, stop_child/2, start_children/4, stop_children/3, terminate_all_children/1]).
+-export([start_child/4, stop_child/2, child_up/2,
+         start_children/4, stop_children/3, terminate_all_children/1]).
 
 -export([get_all_children/1]).
 
 -record(state, {
           index :: non_neg_integer(),
           table :: ets:table(),
+          waiting = #{} :: #{reference() := pid()},
           tasks = #{} :: #{reference() := pid()}
          }).
 -type state() :: #state{}.
@@ -39,6 +41,10 @@ start_link(N) ->
 start_child(Sup, Scenario, Id, ScenarioState) ->
     gen_server:cast(Sup, {start_child, Scenario, Id, ScenarioState}).
 
+-spec child_up(pid(), amoc_scenario:user_id()) -> any().
+child_up(Sup, Id) ->
+    gen_server:cast(Sup, {child_up, self(), Id}).
+
 -spec start_children(pid(), amoc:scenario(), [amoc_scenario:user_id()], any()) -> ok.
 start_children(Sup, Scenario, UserIds, ScenarioState) ->
     gen_server:cast(Sup, {start_children, Scenario, UserIds, ScenarioState}).
@@ -84,10 +90,13 @@ handle_call(get_all_children, _From, #state{table = Table} = State) ->
              {stop_children, non_neg_integer(), boolean()} |
              terminate_all_children.
 handle_cast({start_child, Scenario, Id, ScenarioState}, State) ->
-    do_start_child(Scenario, Id, ScenarioState, State),
-    {noreply, State};
+    NewState = do_start_child(Scenario, [Id], ScenarioState, State),
+    {noreply, NewState};
 handle_cast({start_children, Scenario, Ids, ScenarioState}, State) ->
-    [ do_start_child(Scenario, Id, ScenarioState, State) || Id <- Ids],
+    NewState = do_start_child(Scenario, Ids, ScenarioState, State),
+    {noreply, NewState};
+handle_cast({child_up, Pid, Id}, #state{index = N, table = Tid} = State) ->
+    handle_up_user(Tid, Pid, Id, N),
     {noreply, State};
 handle_cast({stop_children, 0, _}, State) ->
     {noreply, State};
@@ -102,19 +111,26 @@ handle_cast({stop_children, Int, ForceRemove}, #state{table = Table} = State) ->
     {noreply, NewState};
 handle_cast(terminate_all_children, State) ->
     NewState = do_terminate_all_my_children(State),
-    {noreply, NewState};
-handle_cast(_Msg, State) ->
-    {noreply, State}.
+    {noreply, NewState}.
 
 %% @private
 -spec handle_info(Request, state()) -> {noreply, state()} when
       Request :: {child_up, pid(), amoc_scenario:user_id()} |
                  {'DOWN', reference(), process, pid(), term()} |
                  {'EXIT', pid(), term()}.
-handle_info({'DOWN', Ref, process, _Pid, _Reason}, #state{tasks = Tasks} = State) ->
-    {noreply, State#state{tasks = maps:remove(Ref, Tasks)}};
-handle_info({'EXIT', Pid, _Reason}, #state{index = N, table = Table} = State) ->
-    handle_down_user(Table, Pid, N),
+handle_info({'DOWN', Ref, process, Pid, _Reason},
+            #state{waiting = Waiting, tasks = Tasks} = State) ->
+    case {Waiting, Tasks} of
+        {#{Pid := Ref}, _} ->
+            flush_exit(Pid),
+            {noreply, State#state{waiting = maps:remove(Pid, Waiting)}};
+        {_, #{Ref := Pid}} ->
+            {noreply, State#state{tasks = maps:remove(Ref, Tasks)}};
+        _ ->
+            {noreply, State}
+    end;
+handle_info({'EXIT', Pid, _Reason}, #state{index = N, table = Tid} = State) ->
+    handle_down_user(Tid, Pid, N),
     {noreply, State};
 handle_info(_Info, State) ->
     {noreply, State}.
@@ -126,14 +142,19 @@ terminate(_Reason, State) ->
 
 %% Helpers
 
--spec do_start_child(module(), amoc_scenario:user_id(), term(), state()) -> any().
-do_start_child(Scenario, Id, ScenarioState, #state{index = N, table = Table}) ->
-    case amoc_user:start_link(Scenario, Id, ScenarioState) of
-        {ok, Pid} ->
-            handle_up_user(Table, Pid, Id, N);
-        _ ->
-            ok
-    end.
+-spec flush_exit(pid()) -> ok.
+flush_exit(Pid) ->
+    unlink(Pid),
+    receive {'EXIT', Pid, _} -> ok after 0 -> ok end.
+
+-spec do_start_child(amoc:scenario(), [amoc_scenario:user_id()], term(), state()) -> state().
+do_start_child(Scenario, Ids, ScenarioState, #state{waiting = Waiting} = State) ->
+    Fun = fun(Id, Acc) ->
+                  {Pid, Ref} = amoc_user:start_link(Scenario, Id, ScenarioState),
+                  Acc#{Pid => Ref}
+          end,
+    NewWaiting = lists:foldl(Fun, Waiting, Ids),
+    State#state{waiting = NewWaiting}.
 
 -spec handle_up_user(ets:table(), pid(), amoc_scenario:user_id(), non_neg_integer()) -> any().
 handle_up_user(Table, Pid, Id, SupNum) ->
@@ -156,7 +177,7 @@ maybe_track_task_to_stop_my_children(State, Pids, false) ->
     State;
 maybe_track_task_to_stop_my_children(#state{tasks = Tasks} = State, Pids, true) ->
     {Pid, Ref} = spawn_monitor(shutdown_and_kill_after_timeout_fun(Pids)),
-    State#state{tasks = Tasks#{Pid => Ref}}.
+    State#state{tasks = Tasks#{Ref => Pid}}.
 
 -spec shutdown_and_kill_after_timeout_fun(pid() | [pid()]) -> fun(() -> term()).
 shutdown_and_kill_after_timeout_fun([_ | _] = Pids) ->