Cosmetics changes to improve readability, types, and upgrades

.github/workflows/ci.yml

@@ -4,17 +4,18 @@ on:
  push:
  branches: [ master ]
  pull_request:
- branches: [ master ]
+ workflow_dispatch:
+
 
 jobs:
  test:
  name: ${{ matrix.test-type }} test on OTP ${{matrix.otp_vsn}}
  strategy:
  matrix:
- otp_vsn: ['26.2', '25.3', '24.3']
- rebar_vsn: ['3.22.0']
+ otp_vsn: ['27', '26', '25']
+ rebar_vsn: ['3.23.0']
  test-type: ['regular', 'integration']
- runs-on: 'ubuntu-22.04'
+ runs-on: 'ubuntu-24.04'
  steps:
  - uses: actions/checkout@v4
  - uses: erlef/setup-beam@v1

guides/configuration.md

@@ -9,9 +9,6 @@ Amoc supports the following generic configuration parameters:
  * default value - empty list (`[]`)
  * example: `AMOC_NODES="['amoc@amoc-1', 'amoc@amoc-2']"`
 
-* `api_port` - a port for the amoc REST interfaces:
- * default value - 4000
- * example: `AMOC_API_PORT="4000"`
 
 * `interarrival` - a delay (in ms, for each node in the cluster independently) between creating the processes
  for two consecutive users:

integration_test/extra_code_paths/path1/dummy_helper.erl

@@ -1,7 +1,12 @@
 -module(dummy_helper).
 
--required_variable(#{name=>dummy_var, description=>"dummy_var",
- default_value=>default_value}).
+-include_lib("stdlib/include/assert.hrl").
+
+-required_variable(#{name => dummy_var,
+ description => "dummy_var",
+ default_value => default_value}).
+
+-define(comment(U), io_lib:format("Condition failed with last users distribution ~n~p", [U])).
 
 %% amoc_dist testing function
 -export([test_amoc_dist/0]).
@@ -11,65 +16,73 @@ test_amoc_dist() ->
  Master = amoc_cluster:master_node(),
  Slaves = amoc_cluster:slave_nodes(),
  %% check the status of the nodes
- disabled = rpc:call(Master, amoc_controller, get_status, []),
- [{running, #{scenario := dummy_scenario}} = rpc:call(Node, amoc_controller, get_status, [])
- || Node <- Slaves],
- %% check user ids
- {N1, Nodes1, Ids1, Max1} = get_users_info(Slaves),
- true = N1 > 0,
- N1 = Max1,
- Ids1 = lists:seq(1, N1),
+ ?assertEqual(disabled, get_status(Master)),
+ [ ?assertMatch({running, #{scenario := dummy_scenario}}, get_status(Node)) || Node <- Slaves],
+ %% check user ids, users have all been started at the first two nodes
+ {N1, Max1, Nodes1, Ids1, Users1} = get_users_info(Slaves),
+ ?assert(N1 > 0),
+ ?assertEqual(N1, Max1, ?comment(Users1)),
+ ?assertEqual(Ids1, lists:seq(1, N1), ?comment(Users1)),
  [AddedNode] = Slaves -- Nodes1,
  %% add 20 users
- {ok, _} = rpc:call(Master, amoc_dist, add, [20]),
- timer:sleep(3000),
- {N2, Nodes2, Ids2, Max2} = get_users_info(Slaves),
- N2 = Max2,
- Ids2 = lists:seq(1, N2),
- [AddedNode] = Nodes2 -- Nodes1,
- N2 = N1 + 20,
+ add_and_wait(Master, 20),
+ {N2, Max2, Nodes2, Ids2, Users2} = get_users_info(Slaves),
+ ?assertEqual(N2, Max2, ?comment(Users2)),
+ ?assertEqual(Ids2, lists:seq(1, N2), ?comment(Users2)),
+ ?assertEqual([AddedNode], Nodes2 -- Nodes1, ?comment(Users2)),
+ ?assertEqual(N2, N1 + 20, ?comment(Users2)),
  %% remove 10 users
- {ok, _} = rpc:call(Master, amoc_dist, remove, [10, true]),
- timer:sleep(3000),
- {N3, Nodes3, _Ids3, Max3} = get_users_info(Slaves),
- Nodes2 = Nodes3,
- Max3 = Max2,
- N2 = N3 + 10,
+ remove_and_wait(Master, 10),
+ {N3, Max3, Nodes3, _Ids3, Users3} = get_users_info(Slaves),
+ ?assertEqual(N2 - 10, N3, ?comment(Users3)),
+ ?assertEqual(Max2, Max3, ?comment(Users3)),
+ ?assertEqual(Nodes2, Nodes3, ?comment(Users3)),
  %% try to remove N3 users
- {ok, Ret} = rpc:call(Master, amoc_dist, remove, [N3, true]),
+ Ret = remove_and_wait(Master, N3),
  RemovedN = lists:sum([N || {_, N} <- Ret]),
- timer:sleep(3000),
- {N4, Nodes4, Ids4, _Max4} = get_users_info(Slaves),
- Nodes1 = Nodes4,
- N3 = N4 + RemovedN,
- true = RemovedN < N3,
+ {N4, _Max4, Nodes4, Ids4, Users4} = get_users_info(Slaves),
+ ?assertEqual(N3 - RemovedN, N4, ?comment(Users4)),
+ ?assertEqual(Nodes1, Nodes4, ?comment(Users4)),
+ ?assert(RemovedN < N3),
  %% add 20 users
- {ok, _} = rpc:call(Master, amoc_dist, add, [20]),
- timer:sleep(3000),
- {N5, Nodes5, Ids5, Max5} = get_users_info(Slaves),
- Nodes2 = Nodes5,
- Max5 = Max2 + 20,
- N5 = N4 + 20,
- true = Ids5 -- Ids4 =:= lists:seq(Max2 + 1, Max5),
+ add_and_wait(Master, 20),
+ {N5, Max5, Nodes5, Ids5, Users5} = get_users_info(Slaves),
+ ?assertEqual(Nodes2, Nodes5, ?comment(Users5)),
+ ?assertEqual(Max5, Max2 + 20, ?comment(Users5)),
+ ?assertEqual(N5, N4 + 20, ?comment(Users5)),
+ ?assertEqual(Ids5 -- Ids4, lists:seq(Max2 + 1, Max5), ?comment(Users5)),
  %% terminate scenario
- {ok,_} = rpc:call(Master, amoc_dist, stop, []),
- timer:sleep(3000),
- [{finished, dummy_scenario} = rpc:call(Node, amoc_controller, get_status, [])
- || Node <- Slaves],
+ stop(Master),
+ [ ?assertEqual({finished, dummy_scenario}, get_status(Node)) || Node <- Slaves],
  %% return expected value
  amoc_dist_works_as_expected
  catch
  C:E:S ->
- {error, {C, E, S}}
+ {C, E, S}
  end.
 
 get_users_info(SlaveNodes) ->
- Users = [{Node, Id} ||
- Node <- SlaveNodes,
- {_Pid, Id} <- rpc:call(Node, amoc_users_sup, get_all_children, [])],
- Ids = lists:usort([Id || {_, Id} <- Users]),
- Nodes = lists:usort([Node || {Node, _} <- Users]),
+ Distrib = [ {Node, erpc:call(Node, amoc_users_sup, get_all_children, [])} || Node <- SlaveNodes ],
+ Ids = lists:usort([Id || {_Node, Users} <- Distrib, {_, Id} <- Users]),
+ Nodes = lists:usort([Node || {Node, Users} <- Distrib, [] =/= Users]),
  N = length(Ids),
- N = length(Users),
  MaxId = lists:max(Ids),
- {N, Nodes, Ids, MaxId}.
+ {N, MaxId, Nodes, Ids, Distrib}.
+
+add_and_wait(Master, Num) ->
+ {ok, Ret} = erpc:call(Master, amoc_dist, add, [Num]),
+ timer:sleep(3000),
+ Ret.
+
+remove_and_wait(Master, Num) ->
+ {ok, Ret} = erpc:call(Master, amoc_dist, remove, [Num, true]),
+ timer:sleep(3000),
+ Ret.
+
+stop(Master) ->
+ {ok, Ret} = erpc:call(Master, amoc_dist, stop, []),
+ timer:sleep(3000),
+ Ret.
+
+get_status(Node) ->
+ erpc:call(Node, amoc_controller, get_status, []).

src/amoc_controller.erl

@@ -79,7 +79,7 @@
 %% ------------------------------------------------------------------
 
 %% @private
--spec start_link() -> {ok, pid()}.
+-spec start_link() -> gen_server:start_ret().
 start_link() ->
  gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 

src/amoc_scenario.erl

@@ -73,7 +73,7 @@ terminate(Scenario, State) ->
 %% if scenario module exports both functions, `Scenario:start/2' is used.
 %%
 %% Runs on the user process and spans a `[amoc, scenario, user, _]' telemetry event.
--spec start(amoc:scenario(), user_id(), state()) -> any().
+-spec start(amoc:scenario(), user_id(), state()) -> term().
 start(Scenario, Id, State) ->
  Metadata = #{scenario => Scenario, state => State, user_id => Id},
  Span = case {erlang:function_exported(Scenario, start, 2),

src/amoc_sup.erl

@@ -11,14 +11,14 @@
 -export([init/1]).
 
 %% Helper macro for declaring children of supervisor
--define(WORKER(I, Type), {I, {I, start_link, []}, permanent, 5000, Type, [I]}).
--define(SUP(I, Type), {I, {I, start_link, []}, permanent, infinity, Type, [I]}).
+-define(WORKER(I), {I, {I, start_link, []}, permanent, 5000, worker, [I]}).
+-define(SUP(I), {I, {I, start_link, []}, permanent, infinity, supervisor, [I]}).
 
 %% ===================================================================
 %% API functions
 %% ===================================================================
 
--spec start_link() -> {ok, Pid :: pid()}.
+-spec start_link() -> supervisor:startlink_ret().
 start_link() ->
  supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
@@ -30,10 +30,10 @@ start_link() ->
 init([]) ->
  {ok, {#{strategy => one_for_all, intensity => 0},
  [
- ?SUP(amoc_users_sup, supervisor),
- ?SUP(amoc_throttle_sup, supervisor),
- ?SUP(amoc_coordinator_sup, supervisor),
- ?WORKER(amoc_controller, worker),
- ?WORKER(amoc_cluster, worker),
- ?WORKER(amoc_code_server, worker)
+ ?SUP(amoc_users_sup),
+ ?SUP(amoc_throttle_sup),
+ ?SUP(amoc_coordinator_sup),
+ ?WORKER(amoc_controller),
+ ?WORKER(amoc_cluster),
+ ?WORKER(amoc_code_server)
  ]}}.

src/dist/amoc_dist.erl

@@ -94,7 +94,7 @@ get_state() ->
  case {amoc_cluster:master_node(), get_param(state)} of
  {undefined, undefined} -> idle;
  {_, {ok, State}} -> State;
- {Node, undefined} -> rpc:call(Node, ?MODULE, ?FUNCTION_NAME, [])
+ {Node, undefined} -> erpc:call(Node, ?MODULE, ?FUNCTION_NAME, [])
  end.
 
 %% ------------------------------------------------------------------
@@ -172,7 +172,7 @@ setup_slave_node(Node) ->
  {ok, _} ->
  {ok, Scenario} = get_param(scenario),
  {ok, Settings} = get_param(settings),
- rpc:call(Node, amoc_controller, start_scenario, [Scenario, Settings]);
+ erpc:call(Node, amoc_controller, start_scenario, [Scenario, Settings]);
  Error -> Error
  end.
 
@@ -196,7 +196,7 @@ add_users(Result, LastId, Count, [Node | T] = Nodes) ->
  0 ->
  add_users([{Node, {ok, node_skipped}} | Result], LastId, Count, T);
  N ->
- Ret = rpc:call(Node, amoc_controller, add_users, [LastId + 1, LastId + N]),
+ Ret = erpc:call(Node, amoc_controller, add_users, [LastId + 1, LastId + N]),
  add_users([{Node, Ret} | Result], LastId + N, Count - N, T)
  end.
 
@@ -213,7 +213,7 @@ remove_users(Result, Count, ForceRemove, [Node | T] = Nodes) ->
  0 ->
  remove_users([{Node, {ok, node_skipped}} | Result], Count, ForceRemove, T);
  N ->
- Ret = rpc:call(Node, amoc_controller, remove_users, [N, ForceRemove]),
+ Ret = erpc:call(Node, amoc_controller, remove_users, [N, ForceRemove]),
  remove_users([{Node, Ret} | Result], Count - N, ForceRemove, T)
  end.
 
@@ -225,7 +225,7 @@ update_settings_on_nodes(Settings, Nodes) ->
 -spec update_settings_on_node(amoc_config:settings(), node()) ->
  ok | {badrpc, any()} | {error, any()}.
 update_settings_on_node(Settings, Node) ->
- rpc:call(Node, amoc_controller, update_settings, [Settings]).
+ erpc:call(Node, amoc_controller, update_settings, [Settings]).
 
 -spec stop_cluster() -> {ok, any()} | {error, any()}.
 stop_cluster() ->
@@ -234,7 +234,7 @@ stop_cluster() ->
  {_, []} -> {error, no_slave_nodes};
  {MasterNode, Slaves} ->
  set_state(stopped),
- Result = [{Node, rpc:call(Node, amoc_controller, stop_scenario, [])} ||
+ Result = [{Node, erpc:call(Node, amoc_controller, stop_scenario, [])} ||
  Node <- Slaves],
  maybe_error(Result);
  {_, _} -> {error, not_a_master}

src/throttle/amoc_throttle_process.erl

@@ -24,7 +24,7 @@
  handle_info/2,
  handle_cast/2,
  handle_continue/2,
- format_status/2]).
+ format_status/1]).
 
 -define(PG_SCOPE, amoc_throttle).
 -define(DEFAULT_MSG_TIMEOUT, 60000).%% one minute
@@ -143,12 +143,13 @@ handle_continue(maybe_run_fn, State) ->
  NewState = maybe_run_fn(State),
  {noreply, NewState, timeout(NewState)}.
 
--spec format_status(term(), term()) -> term().
-format_status(_Opt, [_PDict, State]) ->
+-spec format_status(gen_server:format_status()) -> gen_server:format_status().
+format_status(#{state := #state{} = State} = FormatStatus) ->
  ScheduleLen = length(State#state.schedule),
  ScheduleRevLen = length(State#state.schedule_reversed),
  State1 = setelement(#state.schedule, State, ScheduleLen),
- setelement(#state.schedule_reversed, State1, ScheduleRevLen).
+ State2 = setelement(#state.schedule_reversed, State1, ScheduleRevLen),
+ FormatStatus#{state := State2}.
 
 %%------------------------------------------------------------------------------
 %% internal functions

src/users/amoc_user.erl

@@ -14,16 +14,15 @@
 start_link(Scenario, Id, State) ->
  proc_lib:start_link(?MODULE, init, [self(), Scenario, Id, State]).
 
--spec stop() -> no_return().
+-spec stop() -> ok.
 stop() ->
  stop(self(), false).
 
 -spec stop(pid(), boolean()) -> ok.
 stop(Pid, Force) when is_pid(Pid) ->
  amoc_users_sup:stop_child(Pid, Force).
 
--spec init(pid(), amoc:scenario(), amoc_scenario:user_id(), state()) ->
- no_return().
+-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),

src/users/amoc_users_sup.erl

@@ -139,14 +139,14 @@ terminate_all_children() ->
 -spec get_sup_for_user_id(amoc_scenario:user_id()) -> pid().
 get_sup_for_user_id(Id) ->
  #storage{sups = Supervisors, sups_count = SupCount} = persistent_term:get(?MODULE),
- Index = Id rem SupCount + 1,
+ Index = erlang:phash2(Id, SupCount) + 1,
  element(Index, Supervisors).
 
 %% assign which users each worker will be requested to add
 -spec assign_users_to_sups(pos_integer(), tuple(), [amoc_scenario:user_id()], Acc) ->
  Acc when Acc :: #{pid() := [amoc_scenario:user_id()]}.
 assign_users_to_sups(SupCount, Supervisors, [Id | Ids], Acc) ->
- Index = Id rem SupCount + 1,
+ Index = erlang:phash2(Id, SupCount) + 1,
  ChosenSup = element(Index, Supervisors),
  Vs = maps:get(ChosenSup, Acc),
  NewAcc = Acc#{ChosenSup := [Id | Vs]},