%%%-------------------------------------------------------------------
%%% @copyright (C) 2019-2023, Maxim Fedorov
%%% @doc
%%% System monitor: reports scheduler, RAM, and benchmarks.
%%%
%%% Monitor is started by default when {@link erlperf} starts
%%% as an application. Monitor is not started for ad-hoc
%%% benchmarking (e.g. command-line, unless verbose logging
%%% is requested).
%%%
%%% When started, the monitor provides periodic reports
%%% about Erlang VM state, and registered jobs performance.
%%% The reports are sent to all processes that joined
%%% `{erlperf_monitor, Node}' or `cluster_monitor' process
%%% group in `erlperf' scope.
%%%
%%% Reports can be received by any process, even the shell. Run
%%% the following example in `rebar3 shell' of `erlperf':
%%% ```
%%% (erlperf@ubuntu22)1> ok = pg:join(erlperf, cluster_monitor, self()).
%%% ok
%%% (erlperf@ubuntu22)2> erlperf:run(rand, uniform, []).
%%% 14976933
%%% (erlperf@ubuntu22)4> flush().
%%% Shell got {erlperf@ubuntu22,#{dcpu => 0.0,dio => 6.42619095979426e-4,
%%% ets => 44,jobs => [],memory_binary => 928408,
%%% memory_ets => 978056,
%%% memory_processes => 8603392,
%%% memory_total => 34952096,ports => 5,
%%% processes => 95,
%%% sched_util => 0.013187335960637163,
%%% '''
%%%
%%% Note that the monitor may report differently from the benchmark
%%% run results. It is running with lower priority and may be significantly
%%% affected by scheduler starvation, timing issues etc..
%%%
%%%
%%%
%%% @end
-module(erlperf_monitor).
-author("maximfca@gmail.com").
-behaviour(gen_server).
%% API
-export([
start/0,
start/1,
start_link/0,
start_link/1,
register/3,
unregister/1
]).
%% gen_server callbacks
-export([
init/1,
handle_call/3,
handle_cast/2,
handle_info/2
]).
-include_lib("kernel/include/logger.hrl").
-define(DEFAULT_TICK_INTERVAL_MS, 1000).
-type monitor_sample() :: #{
time := integer(),
node := node(),
sched_util := float(),
dcpu := float(),
dio := float(),
processes := integer(),
ports := integer(),
ets := integer(),
memory_total := non_neg_integer(),
memory_processes := non_neg_integer(),
memory_binary := non_neg_integer(),
memory_ets := non_neg_integer(),
jobs => [{Job :: pid(), Cycles :: non_neg_integer()}]
}.
%% Monitoring report
%%
%% <ul>
%% <li>`time': timestamp when the report is generates, wall clock, milliseconds</li>
%% <li>`node': originating Erlang node name</li>
%% <li>`sched_util': normal scheduler utilisation, percentage. See {@link scheduler:utilization/1}</li>
%% <li>`dcpu': dirty CPU scheduler utilisation, percentage.</li>
%% <li>`dio': dirty IO scheduler utilisation, percentage</li>
%% <li>`processes': number of processes in the VM.</li>
%% <li>`ports': number of ports in the VM.</li>
%% <li>`ets': number of ETS tables created in the VM.</li>
%% <li>`memory_total': total VM memory usage, see {@link erlang:memory/1}.</li>
%% <li>`memory_processes': processes memory usage, see {@link erlang:system_info/1}.</li>
%% <li>`memory_binary': binary memory usage.</li>
%% <li>`memory_ets': ETS memory usage.</li>
%% <li>`jobs': a map of job process identifier to the iterations surplus
%% since last sample. If the sampling interval is default 1 second, the
%% value of the map is "requests/queries per second" (RPS/QPS).</li>
%% </ul>
-type start_options() :: #{
interval => pos_integer()
}.
%% Monitor startup options
%%
%% <ul>
%% <li>`interval': monitoring interval, 1000 ms by default</li>
%% </ul>
-export_type([monitor_sample/0, start_options/0]).
%% @equiv start(#{interval => 1000})
-spec start() -> {ok, Pid :: pid()} | {error, Reason :: term()}.
start() ->
start(#{interval => ?DEFAULT_TICK_INTERVAL_MS}).
%% @doc
%% Starts the monitor.
%%
%% `Options' are used to change the monitor behaviour.
%% <ul>
%% <li>`interval': time, in milliseconds, to wait between sample collection</li>
%% </ul>
-spec start(Options :: start_options()) -> {ok, Pid :: pid()} | {error, Reason :: term()}.
start(Options) ->
gen_server:start({local, ?MODULE}, ?MODULE, Options, []).
%% @equiv start_link(#{interval => 1000})
-spec(start_link() -> {ok, Pid :: pid()} | {error, Reason :: term()}).
start_link() ->
start_link(#{interval => ?DEFAULT_TICK_INTERVAL_MS}).
%% @doc
%% Starts the monitor and links it to the current process. See {@link start/1}
%% for options description.
start_link(Options) ->
gen_server:start_link({local, ?MODULE}, ?MODULE, Options, []).
%% @doc
%% Registers an {@link erlperf_job} to monitor.
%%
%% Running monitor queries every registered job, adding
%% the number of iterations performed by all workers of
%% that job to the report.
%% This API is intended to be used by {@link erlperf_job}
%% to enable VM monitoring while benchmarking.
%%
%% `Job' specifies job process identifier, it is only
%% used to detect when job is stopped, to stop reporting
%% counters for that job.
%%
%% `Handle' is the sampling handle, see {@link erlperf_job:handle/1}.
%%
%% `Initial' value should be provided when an existing job
%% is registered, to avoid reporting accumulated counter value
%% in the first report for that job.
%%
%% Always return `ok', even when monitor is not running.
-spec register(pid(), term(), non_neg_integer()) -> ok.
register(Job, Handle, Initial) ->
gen_server:cast(?MODULE, {register, Job, Handle, Initial}).
%% @doc
%% Removes the job from monitoring.
%%
%% Stops reporting this job performance.
%%
%% `Job' is the process identifier of the job.
-spec unregister(pid()) -> ok.
unregister(Job) ->
gen_server:cast(?MODULE, {unregister, Job}).
%%%===================================================================
%%% gen_server callbacks
%% System monitor state
-record(state, {
% bi-map of job processes to counters
jobs :: [{pid(), reference(), Handle :: erlperf_job:handle(), Prev :: integer()}],
% scheduler data saved from last call
sched_data :: [{pos_integer(), integer(), integer()}],
% number of normal schedulers
normal :: pos_integer(),
% number of dirty schedulers
dcpu :: pos_integer(),
%
tick = ?DEFAULT_TICK_INTERVAL_MS :: pos_integer(),
next_tick :: integer()
}).
%% @private
init(#{interval := Tick}) ->
%% TODO: figure out if there is a way to find jobs after restart.
%% ask a supervisor? but not all jobs are supervised...
%% Jobs = [{Pid, erlperf_job:handle(Pid), 0} ||
%% {_, Pid, _, _} <- try supervisor:which_children(erlperf_job_sup) catch exit:{noproc, _} -> [] end],
%% [monitor(process, Pid) || {Pid, _, _} <- Jobs],
Jobs = [],
%% enable scheduler utilisation calculation
erlang:system_flag(scheduler_wall_time, true),
Next = erlang:monotonic_time(millisecond) + Tick,
erlang:start_timer(Next, self(), tick, [{abs, true}]),
{ok, #state{
tick = Tick,
jobs = Jobs,
next_tick = Next,
sched_data = lists:sort(erlang:statistics(scheduler_wall_time_all)),
normal = erlang:system_info(schedulers),
dcpu = erlang:system_info(dirty_cpu_schedulers)}
}.
%% @private
handle_call(_Request, _From, _State) ->
erlang:error(notsup).
%% @private
handle_cast({register, Job, Handle, Initial}, #state{jobs = Jobs} = State) ->
MRef = monitor(process, Job),
{noreply, State#state{jobs = [{Job, MRef, Handle, Initial} | Jobs]}};
handle_cast({unregister, Job}, #state{jobs = Jobs} = State) ->
case lists:keyfind(Job, 1, Jobs) of
{Job, MRef, _, _} ->
demonitor(MRef, [flush]),
{noreply, State#state{jobs = lists:keydelete(Job, 1, Jobs)}};
false ->
{noreply, State}
end.
%% @private
handle_info({'DOWN', _MRef, process, Pid, _Reason}, #state{jobs = Jobs} = State) ->
{noreply, State#state{jobs = lists:keydelete(Pid, 1, Jobs)}};
%% @private
handle_info({timeout, _, tick}, State) ->
{noreply, handle_tick(State)}.
%%%===================================================================
%%% Internal functions
handle_tick(#state{sched_data = Data, normal = Normal, dcpu = Dcpu} = State) ->
NewSched = lists:sort(erlang:statistics(scheduler_wall_time_all)),
{NU, DU, DioU} = fold_normal(Data, NewSched, Normal, Dcpu, 0, 0),
% add benchmarking info
{Jobs, UpdatedJobs} = lists:foldl(
fun ({Pid, MRef, Handle, Prev}, {J, Save}) ->
Cycles =
case erlperf_job:sample(Handle) of
C when is_integer(C) -> C;
undefined -> Prev %% job is stopped, race condition here
end,
{[{Pid, Cycles - Prev} | J], [{Pid, MRef, Handle, Cycles} | Save]}
end, {[], []}, State#state.jobs),
%
Sample = #{
time => erlang:system_time(millisecond),
node => node(),
memory_total => erlang:memory(total),
memory_processes => erlang:memory(processes),
memory_binary => erlang:memory(binary),
memory_ets => erlang:memory(ets),
sched_util => NU * 100,
dcpu => DU * 100,
dio => DioU * 100,
processes => erlang:system_info(process_count),
ports => erlang:system_info(port_count),
ets => erlang:system_info(ets_count),
jobs => Jobs},
% notify local & global subscribers
Subscribers = pg:get_members(erlperf, {erlperf_monitor, node()}) ++ pg:get_members(erlperf, cluster_monitor),
[Pid ! Sample || Pid <- Subscribers],
%%
NextTick = State#state.next_tick + State#state.tick,
erlang:start_timer(NextTick, self(), tick, [{abs, true}]),
State#state{sched_data = NewSched, next_tick = NextTick, jobs = lists:reverse(UpdatedJobs)}.
%% Iterates over normal scheduler
fold_normal(Old, New, 0, Dcpu, AccActive, AccTotal) ->
fold_dirty_cpu(Old, New, Dcpu, AccActive / AccTotal, 0, 0);
fold_normal([{N, OldActive, OldTotal} | Old],
[{N, NewActive, NewTotal} | New], Normal, Dcpu, AccActive, AccTotal) ->
fold_normal(Old, New, Normal - 1, Dcpu, AccActive + (NewActive - OldActive),
AccTotal + (NewTotal - OldTotal)).
%% Iterates over DCPU
fold_dirty_cpu(Old, New, 0, NormalPct, AccActive, AccTotal) ->
fold_dirty_io(Old, New, NormalPct, AccActive / AccTotal, 0, 0);
fold_dirty_cpu([{N, OldActive, OldTotal} | Old],
[{N, NewActive, NewTotal} | New], Dcpu, NormalPct, AccActive, AccTotal) ->
fold_dirty_cpu(Old, New, Dcpu - 1, NormalPct, AccActive + (NewActive - OldActive),
AccTotal + (NewTotal - OldTotal)).
%% Remaining are dirty IO
fold_dirty_io([], [], NormalPct, DcpuPct, AccActive, AccTotal) ->
{NormalPct, DcpuPct, AccActive / AccTotal};
fold_dirty_io([{N, OldActive, OldTotal} | Old],
[{N, NewActive, NewTotal} | New], NormalPct, DcpuPct, AccActive, AccTotal) ->
fold_dirty_io(Old, New, NormalPct, DcpuPct, AccActive + (NewActive - OldActive),
AccTotal + (NewTotal - OldTotal)).
