Packages
macula
11.2.0
11.4.0
11.3.1
11.3.0
11.2.0
11.1.0
11.0.0
10.25.0
10.24.0
10.23.0
10.22.0
10.21.0
10.20.3
10.20.2
10.20.0
10.19.2
10.19.1
10.19.0
10.18.0
10.17.0
10.16.0
10.15.0
10.14.5
10.14.4
10.14.2
10.14.1
10.14.0
10.13.2
10.13.1
10.11.0
10.10.2
10.10.1
10.10.0
10.9.1
10.9.0
10.8.0
10.7.0
10.5.8
10.5.7
10.5.6
10.5.5
10.5.4
10.5.3
10.5.2
10.5.1
10.5.0
10.4.0
10.2.0
10.1.1
10.1.0
10.0.2
10.0.1
10.0.0
9.13.8
9.8.2
9.8.1
9.8.0
9.5.0
9.4.0
9.3.1
9.3.0
9.2.0
9.1.1
9.1.0
9.0.0
8.7.0
8.6.0
8.5.0
8.4.1
8.4.0
8.3.0
8.2.0
8.1.0
8.0.2
8.0.1
8.0.0
7.1.0
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula_stream_sessions.erl
%%%-------------------------------------------------------------------
%%% @doc The node's count of served stream sessions, and of the bytes their
%%% streams keep unread.
%%%
%%% A station link asks `admit/2' before it serves a session it has
%%% started, naming the verified caller whose STREAM_OPEN the session
%%% serves. The session is admitted while that caller holds fewer than
%%% `max_served_sessions_per_caller' sessions on the node and the node
%%% fewer than `max_served_sessions', both read from the macula application
%%% env when it asks. An admitted session holds its place until its stream
%%% process ends, however it ends: this process monitors it.
%%%
%%% The caps are 16 per caller and 1000 on the node by default. A cap is
%%% kept per caller and not per link, because a link to a station carries
%%% every caller that station sends here: a cap per link would let one busy
%%% caller take the places of all the others. 16 leaves a caller room for
%%% many sessions at once, and it takes at least 63 callers to fill the
%%% node. The node cap bounds the processes all served sessions hold
%%% together, two for each session: its handler and its stream. A node that
%%% serves more sets its own value.
%%%
%%% A served stream charges the bytes of each chunk it queues unread with
%%% `charge/2', and gives them back with `release/2' when a reader takes the
%%% chunk. Both run in the stream's own process against the table, so no
%%% chunk waits on another process. A charge is refused when it would take
%%% the bytes of the stream's caller past `max_served_inbox_bytes_per_caller'
%%% (16 MiB by default) or the node's past `max_served_inbox_bytes' (256 MiB
%%% by default). The caller budget is one stream's own bound, so a caller
%%% with many sessions keeps no more unread than one session may; the node
%%% budget is sixteen callers' worth. What a stream still has charged when
%%% its process ends comes back then, through this process's monitor.
%%%
%%% A charge adds to the caller's count, then the node's, then the stream's
%%% own, and a release takes off in the opposite order; a refused add is
%%% taken back at once. So a stream that ends in the middle of one leaves at
%%% most that chunk counted too much, never too little, and this process
%%% clears what is left when the caller holds no session any more, and on
%%% the node when it holds none. Those clears run where a session ends, and
%%% a new session is admitted by this same process before its stream can
%%% charge anything.
%%%
%%% The sessions and bytes are kept in a table that
%%% `macula_stream_sessions_keeper' holds, so they outlive this process.
%%% When it starts again it counts the kept sessions anew and monitors their
%%% streams, and a stream that ended in between frees its place and its
%%% bytes at once.
%%%
%%% A refusal is counted by its reason: `caller_limit' or `node_limit' for a
%%% session, `caller_budget' or `node_budget' for a charge. A refusal is
%%% logged at once when no refusal was logged within the last
%%% `served_session_refusal_log_interval_ms'; the ones after it are logged
%%% together when that interval has passed.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_stream_sessions).
-behaviour(gen_server).
-export([start_link/0, new_table/0, admit/2, sessions/0,
charge/2, release/2, inbox_bytes/0, refusals/0]).
%% Reads of the node's counts that nothing inside macula calls yet; the tests
%% read them.
-ignore_xref([{macula_stream_sessions, sessions, 0}]).
-ignore_xref([{macula_stream_sessions, inbox_bytes, 0}]).
-ignore_xref([{macula_stream_sessions, refusals, 0}]).
%% Started from macula_root's child spec, which xref does not count as a call.
-ignore_xref([{macula_stream_sessions, start_link, 0}]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-export_type([refusal/0]).
-define(SERVER, ?MODULE).
%% The table's rows: `{{stream, Pid}, Caller}' for each admitted session,
%% `{{caller, Caller}, Count}' for each caller that holds a session,
%% `{sessions, Count}', `{{stream_bytes, Pid}, Bytes}',
%% `{{caller_bytes, Caller}, Bytes}', `{inbox_bytes, Bytes}', and
%% `{{refused, Reason}, Count}'.
-define(TABLE, macula_stream_sessions).
-define(MAX_SESSIONS_PER_CALLER, 16).
-define(MAX_SESSIONS, 1000).
-define(MAX_INBOX_BYTES_PER_CALLER, 16#1000000).
-define(MAX_INBOX_BYTES, 16#10000000).
-define(REFUSAL_LOG_INTERVAL_MS, 60000).
%% The counter only does table operations, so a link waits at most this long
%% for an admission before refusing it as `unavailable'.
-define(ADMIT_TIMEOUT_MS, 1000).
-type refusal() :: caller_limit | node_limit | caller_budget | node_budget.
-record(state, {
%% Refusals not logged yet, when refusals were last logged, and the
%% timer that logs the ones held back.
unlogged = #{} :: #{refusal() => pos_integer()},
logged_at = undefined :: undefined | integer(),
flush = undefined :: undefined | reference()
}).
%%%===================================================================
%%% API
%%%===================================================================
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
%% @doc Create the table the sessions are kept in, owned by the calling
%% process. `macula_stream_sessions_keeper' creates it, so it outlives this
%% process.
-spec new_table() -> ets:table().
new_table() ->
ets:new(?TABLE, [set, public, named_table]).
%% @doc Admit the session that `Stream' serves for the verified `Caller', or
%% refuse it when that caller or the node already holds as many sessions as
%% its cap allows. A stream already admitted keeps its one place.
%%
%% Admission fails closed and never holds up the link that asks: when this
%% process does not answer within a second, because it is restarting, gone
%% or held up, the session is refused as `unavailable'.
-spec admit(binary(), pid()) -> ok | {error, caller_limit | node_limit | unavailable}.
admit(Caller, Stream) when is_binary(Caller), is_pid(Stream) ->
try gen_server:call(?SERVER, {admit, Caller, Stream}, ?ADMIT_TIMEOUT_MS)
catch exit:{_Why, {gen_server, call, _Args}} -> {error, unavailable}
end.
%% @doc The number of served sessions the node holds now.
-spec sessions() -> non_neg_integer().
sessions() ->
gen_server:call(?SERVER, sessions).
%% @doc Charge `Bytes' that `Stream' keeps unread to its caller and the node,
%% or refuse when that would take either past its budget, or when `Stream' is
%% not an admitted session. Called in the stream's own process.
-spec charge(pid(), non_neg_integer()) -> ok | {error, caller_budget | node_budget | not_admitted}.
charge(Stream, Bytes) when is_pid(Stream), is_integer(Bytes), Bytes >= 0 ->
charge_caller(ets:lookup(?TABLE, {stream, Stream}), Stream, Bytes).
%% @doc Give back `Bytes' that `Stream' charged and no longer keeps. Called in
%% the stream's own process.
-spec release(pid(), non_neg_integer()) -> ok.
release(Stream, Bytes) when is_pid(Stream), is_integer(Bytes), Bytes >= 0 ->
release_caller(ets:lookup(?TABLE, {stream, Stream}), Stream, Bytes).
%% @doc The bytes the node's served streams keep unread now.
-spec inbox_bytes() -> integer().
inbox_bytes() ->
ets:lookup_element(?TABLE, inbox_bytes, 2, 0).
%% @doc The refusals since the macula application started, by reason.
-spec refusals() -> #{refusal() => pos_integer()}.
refusals() ->
gen_server:call(?SERVER, refusals).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
init([]) ->
ok = count_kept_sessions(ets:match_object(?TABLE, {{stream, '_'}, '_'})),
{ok, #state{}}.
handle_call({admit, Caller, Stream}, _From, S) ->
on_admission(admission(ets:member(?TABLE, {stream, Stream}), Caller), Caller, Stream, S);
handle_call(sessions, _From, S) ->
{reply, count(sessions), S};
handle_call(refusals, _From, S) ->
{reply, maps:from_list(refusal_counts(ets:match_object(?TABLE, {{refused, '_'}, '_'}))), S}.
handle_cast({charge_refused, Reason}, S) ->
{noreply, count_refusal(Reason, S)};
handle_cast(_Msg, S) ->
{noreply, S}.
handle_info({'DOWN', _Mon, process, Stream, _Reason}, S) ->
ok = release_session(ets:lookup(?TABLE, {stream, Stream})),
{noreply, S};
handle_info(log_refusals, #state{unlogged = Unlogged} = S) when map_size(Unlogged) =:= 0 ->
{noreply, S#state{flush = undefined}};
handle_info(log_refusals, S) ->
{noreply, log_refusals(S#state{flush = undefined}, erlang:monotonic_time(millisecond))};
handle_info(_Info, S) ->
{noreply, S}.
terminate(_Reason, _S) ->
ok.
%%%===================================================================
%%% Internal: sessions
%%%===================================================================
%% Counts the kept sessions anew from their stream rows and monitors each
%% stream again. A stream that ended while no counter ran frees its place
%% when its monitor fires, at once.
count_kept_sessions(Streams) ->
true = ets:match_delete(?TABLE, {{caller, '_'}, '_'}),
true = ets:insert(?TABLE, [{sessions, length(Streams)} | caller_rows(Streams)]),
_ = [erlang:monitor(process, Stream) || {{stream, Stream}, _Caller} <- Streams],
ok.
caller_rows(Streams) ->
Counts = lists:foldl(fun({_Key, Caller}, Acc) -> bump(Caller, Acc) end, #{}, Streams),
[{{caller, Caller}, N} || {Caller, N} <- maps:to_list(Counts)].
refusal_counts(Rows) ->
[{Reason, N} || {{refused, Reason}, N} <- Rows].
admission(true, _Caller) ->
admitted;
admission(false, Caller) ->
admission_for(count(sessions) < limit(max_served_sessions, ?MAX_SESSIONS),
count({caller, Caller}) < limit(max_served_sessions_per_caller,
?MAX_SESSIONS_PER_CALLER)).
admission_for(false, _CallerHasRoom) -> {error, node_limit};
admission_for(true, false) -> {error, caller_limit};
admission_for(true, true) -> ok.
limit(Key, Default) ->
application:get_env(macula, Key, Default).
count(Key) ->
ets:lookup_element(?TABLE, Key, 2, 0).
on_admission(ok, Caller, Stream, S) ->
_ = erlang:monitor(process, Stream),
true = ets:insert(?TABLE, [{{stream, Stream}, Caller},
{{caller, Caller}, count({caller, Caller}) + 1},
{sessions, count(sessions) + 1}]),
{reply, ok, S};
on_admission(admitted, _Caller, _Stream, S) ->
{reply, ok, S};
on_admission({error, Reason} = Refused, _Caller, _Stream, S) ->
{reply, Refused, count_refusal(Reason, S)}.
%% A session's bytes and place free when its stream process ends. The
%% stream's own count is taken first, so a restart in between never gives
%% the same bytes back twice.
release_session([{{stream, Stream} = Key, Caller}]) ->
ok = give_back(Caller, taken_bytes(ets:take(?TABLE, {stream_bytes, Stream}))),
true = ets:delete(?TABLE, Key),
ok = one_less_session(count(sessions)),
one_less(count({caller, Caller}), Caller);
release_session([]) ->
ok.
taken_bytes([{_Key, Bytes}]) -> Bytes;
taken_bytes([]) -> 0.
give_back(Caller, Bytes) ->
ok = add(inbox_bytes, -Bytes),
add({caller_bytes, Caller}, -Bytes).
%% A node that holds no session keeps no unread bytes: what a stream that
%% ended in the middle of an update left counted is cleared with the last one.
one_less_session(1) ->
true = ets:insert(?TABLE, {sessions, 0}),
true = ets:delete(?TABLE, inbox_bytes),
ok;
one_less_session(N) ->
true = ets:insert(?TABLE, {sessions, N - 1}),
ok.
%% A caller that holds no session keeps no unread bytes either.
one_less(1, Caller) ->
true = ets:delete(?TABLE, {caller, Caller}),
true = ets:delete(?TABLE, {caller_bytes, Caller}),
ok;
one_less(N, Caller) ->
true = ets:insert(?TABLE, {{caller, Caller}, N - 1}),
ok.
bump(Key, Counts) ->
maps:update_with(Key, fun(N) -> N + 1 end, 1, Counts).
%%%===================================================================
%%% Internal: the budget on unread bytes
%%%===================================================================
charge_caller([], _Stream, _Bytes) ->
{error, not_admitted};
charge_caller([{_Key, Caller}], Stream, Bytes) ->
charged_caller(add_within({caller_bytes, Caller}, Bytes,
limit(max_served_inbox_bytes_per_caller, ?MAX_INBOX_BYTES_PER_CALLER)),
Caller, Stream, Bytes).
charged_caller(refused, _Caller, _Stream, _Bytes) ->
refuse_charge(caller_budget);
charged_caller(added, Caller, Stream, Bytes) ->
charged_node(add_within(inbox_bytes, Bytes, limit(max_served_inbox_bytes, ?MAX_INBOX_BYTES)),
Caller, Stream, Bytes).
charged_node(refused, Caller, _Stream, Bytes) ->
ok = add({caller_bytes, Caller}, -Bytes),
refuse_charge(node_budget);
charged_node(added, _Caller, Stream, Bytes) ->
add({stream_bytes, Stream}, Bytes).
%% Adds Bytes to the count at Key, and takes them back at once when that took
%% it past Max: for that moment the count is over by only the refused bytes.
add_within(Key, Bytes, Max) ->
within(ets:update_counter(?TABLE, Key, {2, Bytes}, {Key, 0}) =< Max, Key, Bytes).
within(true, _Key, _Bytes) ->
added;
within(false, Key, Bytes) ->
ok = add(Key, -Bytes),
refused.
add(Key, Bytes) ->
_ = ets:update_counter(?TABLE, Key, {2, Bytes}, {Key, 0}),
ok.
%% This process counts a refused charge and logs it with the other refusals.
refuse_charge(Reason) ->
ok = gen_server:cast(?SERVER, {charge_refused, Reason}),
{error, Reason}.
release_caller([], _Stream, _Bytes) ->
ok;
release_caller([{_Key, Caller}], Stream, Bytes) ->
ok = add({stream_bytes, Stream}, -Bytes),
ok = add(inbox_bytes, -Bytes),
add({caller_bytes, Caller}, -Bytes).
%%%===================================================================
%%% Internal: refusals
%%%===================================================================
count_refusal(Reason, #state{unlogged = Unlogged} = S) ->
_ = ets:update_counter(?TABLE, {refused, Reason}, 1, {{refused, Reason}, 0}),
report(S#state{unlogged = bump(Reason, Unlogged)}, erlang:monotonic_time(millisecond)).
%% Logs now when nothing was logged within the interval; otherwise holds the
%% refusals back for one timer that logs them when the interval has passed.
report(#state{logged_at = undefined} = S, Now) ->
log_refusals(S, Now);
report(#state{logged_at = At} = S, Now) ->
report_after(Now - At >= log_interval(), S, Now).
report_after(true, S, Now) ->
log_refusals(S, Now);
report_after(false, #state{flush = undefined, logged_at = At} = S, Now) ->
S#state{flush = erlang:send_after(max(0, At + log_interval() - Now), self(), log_refusals)};
report_after(false, S, _Now) ->
S.
log_refusals(#state{unlogged = Unlogged, flush = Flush} = S, Now) ->
ok = cancel(Flush),
logger:warning("[macula_stream_sessions] served stream sessions refused since the last report: ~p",
[Unlogged]),
S#state{unlogged = #{}, logged_at = Now, flush = undefined}.
cancel(undefined) -> ok;
cancel(Timer) -> _ = erlang:cancel_timer(Timer), ok.
log_interval() ->
limit(served_session_refusal_log_interval_ms, ?REFUSAL_LOG_INTERVAL_MS).