emqx / esockd / 352

Committed 12 Dec 2023 09:17PM UTC coverage: 70.175%. First build

Build # 352

Build Type

Pull #183

github

Committed by

web-flow

Commit Message

Merge c60472aab into 5cb22a8b1

Pull Request Pull Request #183: feat(listener): support changing options on the fly

Run Details

143 of 177 new or added lines in 10 files covered. (80.79%)

800 of 1140 relevant lines covered (70.18%)

59.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.6

/src/esockd_listener_sup.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(esockd_listener_sup).

-behaviour(supervisor).

-include("esockd.hrl").

-export([ start_link/5
        , start_link/3
        , listener/1
        , acceptor_sup/1
        , connection_sup/1
       ]).

%% get/set
-export([ get_options/2
        , get_acceptors/1
        , get_max_connections/1
        , get_max_conn_rate/3
        , get_current_connections/1
        , get_shutdown_count/1
        ]).

-export([ set_options/3
        , set_max_connections/3
        , set_max_conn_rate/3
        ]).

-export([ get_access_rules/1
        , allow/2
        , deny/2
        ]).

-export([conn_rate_limiter/1, conn_limiter_opts/2, conn_limiter_opt/2]).

%% supervisor callbacks
-export([init/1]).

%% callbacks
-export([ tune_socket/2
        , start_acceptors/1
        ]).

-type listen_type() :: tcp | dtls.

%%--------------------------------------------------------------------
%% APIs
%%--------------------------------------------------------------------

%% @doc Start listener supervisor
-spec start_link(listen_type(), atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
      -> {ok, pid()} | {error, term()}.
start_link(Type, Proto, ListenOn, Opts, MFA) ->
    %% NOTE
    %% `Opts` are still part of the childspec, which means if whole listener supervisor
    %% restarts, any changes done through `set_options` will be lost.
    ListenerRef = {Proto, ListenOn},
    _ = esockd_server:set_listener_prop(ListenerRef, type, Type),
    _ = esockd_server:set_listener_prop(ListenerRef, options, Opts),
    supervisor:start_link(?MODULE, {ListenerRef, MFA}).

%% @doc Start listener supervisor
-spec start_link(atom(), esockd:listen_on(), esockd:mfargs())
      -> {ok, pid()} | {error, term()}.
start_link(Proto, ListenOn, MFA) ->
    ListenerRef = {Proto, ListenOn},
    supervisor:start_link(?MODULE, {ListenerRef, MFA}).

%% @doc Get listener.
-spec(listener(pid()) -> {module(), pid()}).
listener(Sup) ->
    Child = get_sup_child(Sup, listener),
    {get_child_mod(Child), get_child_pid(Child)}.

%% @doc Get connection supervisor.
-spec(connection_sup(pid()) -> pid()).
connection_sup(Sup) ->
    get_child_pid(get_sup_child(Sup, connection_sup)).

%% @doc Get acceptor supervisor.
-spec(acceptor_sup(pid()) -> pid()).
acceptor_sup(Sup) ->
    get_child_pid(get_sup_child(Sup, acceptor_sup)).

%% @doc Get child pid with id.
get_sup_child(Sup, ChildId) ->
    lists:keyfind(ChildId, 1, supervisor:which_children(Sup)).

get_child_pid({_, Pid, _, _}) -> Pid.
get_child_mod({_, _, _, [Mod | _]}) -> Mod.

%%--------------------------------------------------------------------
%% Get/Set APIs
%%--------------------------------------------------------------------

get_options(ListenerRef, _Sup) ->
    esockd_server:get_listener_prop(ListenerRef, options).

update_options(ListenerRef, Sup, UpdateOpts) ->
    Opts = esockd_server:get_listener_prop(ListenerRef, options),
    set_options(ListenerRef, Sup, merge_options(Opts, UpdateOpts)).

set_options(ListenerRef, Sup, Opts) ->
    OptsWas = esockd_server:set_listener_prop(ListenerRef, options, Opts),
    ConnSup = esockd_server:get_listener_prop(ListenerRef, connection_sup),
    try
        _ = ensure_ok(esockd_connection_sup:set_options(ConnSup, Opts)),
        _ = ensure_ok(restart_acceptor_sup(ListenerRef, Sup)),
        ok
    catch
        throw:{?MODULE, Error} ->
            _ = esockd_server:set_listener_prop(ListenerRef, options, OptsWas),
            ok = esockd_connection_sup:set_options(ConnSup, OptsWas),
            ok = restart_acceptor_sup(ListenerRef, Sup),
            Error
    end.

merge_options(Opts1, Opts2) ->
    Opts2 ++ lists:foldl(fun proplists:delete/2, Opts1, proplists:get_keys(Opts2)).

restart_acceptor_sup(ListenerRef, Sup) ->
    _ = supervisor:terminate_child(Sup, acceptor_sup),
    case supervisor:restart_child(Sup, acceptor_sup) of
        {ok, _Child} ->
            _ = start_acceptors(ListenerRef),
            ok;
        Error = {error, _} ->
            Error
    end.

start_acceptors(ListenerRef) ->
    {LMod, LPid} = esockd_server:get_listener_prop(ListenerRef, listener),
    LState = LMod:get_state(LPid),
    LSock = proplists:get_value(listen_sock, LState),
    ok = esockd_acceptor_sup:start_acceptors(ListenerRef, LSock),
    ignore.

ensure_ok(ok) ->
    ok;
ensure_ok({error, _} = Error) ->
    throw({?MODULE, Error}).

get_acceptors(Sup) ->
    esockd_acceptor_sup:count_acceptors(acceptor_sup(Sup)).

get_max_connections(Sup) ->
    esockd_connection_sup:get_max_connections(connection_sup(Sup)).

set_max_connections(ListenerRef, Sup, MaxConns) ->
    update_options(ListenerRef, Sup, [{max_connections, MaxConns}]).

get_max_conn_rate(_Sup, Proto, ListenOn) ->
    case esockd_limiter:lookup({listener, Proto, ListenOn}) of
        undefined ->
            {error, not_found};
        #{capacity := Capacity, interval := Interval} ->
            {Capacity, Interval}
    end.

set_max_conn_rate(ListenerRef, Sup, Opt) ->
    update_options(ListenerRef, Sup, [{limiter, Opt}]).

get_current_connections(Sup) ->
    esockd_connection_sup:count_connections(connection_sup(Sup)).

get_shutdown_count(Sup) ->
    esockd_connection_sup:get_shutdown_count(connection_sup(Sup)).

get_access_rules(Sup) ->
    esockd_connection_sup:access_rules(connection_sup(Sup)).

allow(Sup, CIDR) ->
    esockd_connection_sup:allow(connection_sup(Sup), CIDR).

deny(Sup, CIDR) ->
    esockd_connection_sup:deny(connection_sup(Sup), CIDR).

%%--------------------------------------------------------------------
%% Supervisor callbacks
%%--------------------------------------------------------------------

init({ListenerRef, MFA}) ->
    ConnSup = #{id => connection_sup,
                start => {esockd_connection_sup, start_supervised, [ListenerRef, MFA]},
                restart => transient,
                shutdown => infinity,
                type => supervisor,
                modules => [esockd_connection_sup]},
    ListenerMod = case esockd_server:get_listener_prop(ListenerRef, type) of
                    dtls -> esockd_dtls_listener;
                    _ -> esockd_listener
                end,
    Listener = #{id => listener,
                 start => {ListenerMod, start_supervised, [ListenerRef]},
                 restart => transient,
                 shutdown => 16#ffffffff,
                 type => worker,
                 modules => [ListenerMod]},
    AcceptorSup = #{id => acceptor_sup,
                    start => {esockd_acceptor_sup, start_supervised, [ListenerRef]},
                    restart => transient,
                    shutdown => infinity,
                    type => supervisor,
                    modules => [esockd_acceptor_sup]},
    Starter = #{id => starter,
                start => {?MODULE, start_acceptors, [ListenerRef]},
                restart => transient,
                shutdown => infinity,
                type => worker,
                modules => []},
    {ok, { {rest_for_one, 10, 3600}
         , [ConnSup, Listener, AcceptorSup, Starter]
         }}.

%%--------------------------------------------------------------------
%% Sock tune/upgrade functions
%%--------------------------------------------------------------------

tune_socket(Sock, Tunings) ->
    esockd_acceptor_sup:tune_socket(Sock, Tunings).

conn_limiter_opts(Opts, DefName) ->
    Opt =  proplists:get_value(limiter, Opts, undefined),
    conn_limiter_opt(Opt, DefName).

conn_limiter_opt(#{name := _} = Opt, _DefName) ->
    Opt;
conn_limiter_opt(Opt, DefName) when is_map(Opt) ->
    Opt#{name => DefName};
conn_limiter_opt(_Opt, _DefName) ->
    undefined.

conn_rate_limiter(undefined) ->
    undefined;
conn_rate_limiter(Opts) ->
    esockd_generic_limiter:create(Opts).

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16
17	-module(esockd_listener_sup).
18
19	-behaviour(supervisor).
20
21	-include("esockd.hrl").
22
23	-export([ start_link/5
24	, start_link/3
25	, listener/1
26	, acceptor_sup/1
27	, connection_sup/1
28	]).
29
30	%% get/set
31	-export([ get_options/2
32	, get_acceptors/1
33	, get_max_connections/1
34	, get_max_conn_rate/3
35	, get_current_connections/1
36	, get_shutdown_count/1
37	]).
38
39	-export([ set_options/3
40	, set_max_connections/3
41	, set_max_conn_rate/3
42	]).
43
44	-export([ get_access_rules/1
45	, allow/2
46	, deny/2
47	]).
48
49	-export([conn_rate_limiter/1, conn_limiter_opts/2, conn_limiter_opt/2]).
50
51	%% supervisor callbacks
52	-export([init/1]).
53
54	%% callbacks
55	-export([ tune_socket/2
56	, start_acceptors/1
57	]).
58
59	-type listen_type() :: tcp \| dtls.
60
61	%%--------------------------------------------------------------------
62	%% APIs
63	%%--------------------------------------------------------------------
64
65	%% @doc Start listener supervisor
66	-spec start_link(listen_type(), atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
67	-> {ok, pid()} \| {error, term()}.
68	start_link(Type, Proto, ListenOn, Opts, MFA) ->
69	%% NOTE
70	%% `Opts` are still part of the childspec, which means if whole listener supervisor
71	%% restarts, any changes done through `set_options` will be lost.
NEW 72	ListenerRef = {Proto, ListenOn},	×
NEW 73	_ = esockd_server:set_listener_prop(ListenerRef, type, Type),	×
NEW 74	_ = esockd_server:set_listener_prop(ListenerRef, options, Opts),	×
NEW 75	supervisor:start_link(?MODULE, {ListenerRef, MFA}).	×
76
77	%% @doc Start listener supervisor
78	-spec start_link(atom(), esockd:listen_on(), esockd:mfargs())
79	-> {ok, pid()} \| {error, term()}.
80	start_link(Proto, ListenOn, MFA) ->
81	ListenerRef = {Proto, ListenOn},	55✔
82	supervisor:start_link(?MODULE, {ListenerRef, MFA}).	55✔
83
84	%% @doc Get listener.
85	-spec(listener(pid()) -> {module(), pid()}).
86	listener(Sup) ->
87	Child = get_sup_child(Sup, listener),	4✔
88	{get_child_mod(Child), get_child_pid(Child)}.	4✔
89
90	%% @doc Get connection supervisor.
91	-spec(connection_sup(pid()) -> pid()).
92	connection_sup(Sup) ->
93	get_child_pid(get_sup_child(Sup, connection_sup)).	25✔
94
95	%% @doc Get acceptor supervisor.
96	-spec(acceptor_sup(pid()) -> pid()).
97	acceptor_sup(Sup) ->
98	get_child_pid(get_sup_child(Sup, acceptor_sup)).	6✔
99
100	%% @doc Get child pid with id.
101	get_sup_child(Sup, ChildId) ->
102	lists:keyfind(ChildId, 1, supervisor:which_children(Sup)).	35✔
103
104	get_child_pid({_, Pid, _, _}) -> Pid.	35✔
105	get_child_mod({_, _, _, [Mod \| _]}) -> Mod.	4✔
106
107	%%--------------------------------------------------------------------
108	%% Get/Set APIs
109	%%--------------------------------------------------------------------
110
111	get_options(ListenerRef, _Sup) ->
112	esockd_server:get_listener_prop(ListenerRef, options).	2✔
113
114	update_options(ListenerRef, Sup, UpdateOpts) ->
115	Opts = esockd_server:get_listener_prop(ListenerRef, options),	4✔
116	set_options(ListenerRef, Sup, merge_options(Opts, UpdateOpts)).	4✔
117
118	set_options(ListenerRef, Sup, Opts) ->
119	OptsWas = esockd_server:set_listener_prop(ListenerRef, options, Opts),	7✔
120	ConnSup = esockd_server:get_listener_prop(ListenerRef, connection_sup),	7✔
121	try	7✔
122	_ = ensure_ok(esockd_connection_sup:set_options(ConnSup, Opts)),	7✔
123	_ = ensure_ok(restart_acceptor_sup(ListenerRef, Sup)),	7✔
124	ok	7✔
125	catch
126	throw:{?MODULE, Error} ->
NEW 127	_ = esockd_server:set_listener_prop(ListenerRef, options, OptsWas),	×
NEW 128	ok = esockd_connection_sup:set_options(ConnSup, OptsWas),	×
NEW 129	ok = restart_acceptor_sup(ListenerRef, Sup),	×
NEW 130	Error	×
131	end.
132
133	merge_options(Opts1, Opts2) ->
134	Opts2 ++ lists:foldl(fun proplists:delete/2, Opts1, proplists:get_keys(Opts2)).	4✔
135
136	restart_acceptor_sup(ListenerRef, Sup) ->
137	_ = supervisor:terminate_child(Sup, acceptor_sup),	7✔
138	case supervisor:restart_child(Sup, acceptor_sup) of	7✔
139	{ok, _Child} ->
140	_ = start_acceptors(ListenerRef),	7✔
141	ok;	7✔
142	Error = {error, _} ->
NEW 143	Error	×
144	end.
145
146	start_acceptors(ListenerRef) ->
147	{LMod, LPid} = esockd_server:get_listener_prop(ListenerRef, listener),	64✔
148	LState = LMod:get_state(LPid),	64✔
149	LSock = proplists:get_value(listen_sock, LState),	64✔
150	ok = esockd_acceptor_sup:start_acceptors(ListenerRef, LSock),	64✔
151	ignore.	64✔
152
153	ensure_ok(ok) ->
154	ok;	14✔
155	ensure_ok({error, _} = Error) ->
NEW 156	throw({?MODULE, Error}).	×
157
158	get_acceptors(Sup) ->
159	esockd_acceptor_sup:count_acceptors(acceptor_sup(Sup)).	6✔
160
161	get_max_connections(Sup) ->
162	esockd_connection_sup:get_max_connections(connection_sup(Sup)).	5✔
163
164	set_max_connections(ListenerRef, Sup, MaxConns) ->
165	update_options(ListenerRef, Sup, [{max_connections, MaxConns}]).	2✔
166
167	get_max_conn_rate(_Sup, Proto, ListenOn) ->
168	case esockd_limiter:lookup({listener, Proto, ListenOn}) of	4✔
169	undefined ->
170	{error, not_found};	×
171	#{capacity := Capacity, interval := Interval} ->
172	{Capacity, Interval}	4✔
173	end.
174
175	set_max_conn_rate(ListenerRef, Sup, Opt) ->
176	update_options(ListenerRef, Sup, [{limiter, Opt}]).	2✔
177
178	get_current_connections(Sup) ->
179	esockd_connection_sup:count_connections(connection_sup(Sup)).	3✔
180
181	get_shutdown_count(Sup) ->
182	esockd_connection_sup:get_shutdown_count(connection_sup(Sup)).	2✔
183
184	get_access_rules(Sup) ->
185	esockd_connection_sup:access_rules(connection_sup(Sup)).	9✔
186
187	allow(Sup, CIDR) ->
188	esockd_connection_sup:allow(connection_sup(Sup), CIDR).	3✔
189
190	deny(Sup, CIDR) ->
191	esockd_connection_sup:deny(connection_sup(Sup), CIDR).	3✔
192
193	%%--------------------------------------------------------------------
194	%% Supervisor callbacks
195	%%--------------------------------------------------------------------
196
197	init({ListenerRef, MFA}) ->
198	ConnSup = #{id => connection_sup,	55✔
199	start => {esockd_connection_sup, start_supervised, [ListenerRef, MFA]},
200	restart => transient,
201	shutdown => infinity,
202	type => supervisor,
203	modules => [esockd_connection_sup]},
204	ListenerMod = case esockd_server:get_listener_prop(ListenerRef, type) of	55✔
205	dtls -> esockd_dtls_listener;	10✔
206	_ -> esockd_listener	45✔
207	end,
208	Listener = #{id => listener,	55✔
209	start => {ListenerMod, start_supervised, [ListenerRef]},
210	restart => transient,
211	shutdown => 16#ffffffff,
212	type => worker,
213	modules => [ListenerMod]},
214	AcceptorSup = #{id => acceptor_sup,	55✔
215	start => {esockd_acceptor_sup, start_supervised, [ListenerRef]},
216	restart => transient,
217	shutdown => infinity,
218	type => supervisor,
219	modules => [esockd_acceptor_sup]},
220	Starter = #{id => starter,	55✔
221	start => {?MODULE, start_acceptors, [ListenerRef]},
222	restart => transient,
223	shutdown => infinity,
224	type => worker,
225	modules => []},
226	{ok, { {rest_for_one, 10, 3600}	55✔
227	, [ConnSup, Listener, AcceptorSup, Starter]
228	}}.
229
230	%%--------------------------------------------------------------------
231	%% Sock tune/upgrade functions
232	%%--------------------------------------------------------------------
233
234	tune_socket(Sock, Tunings) ->
NEW 235	esockd_acceptor_sup:tune_socket(Sock, Tunings).	×
236
237	conn_limiter_opts(Opts, DefName) ->
238	Opt = proplists:get_value(limiter, Opts, undefined),	74✔
239	conn_limiter_opt(Opt, DefName).	74✔
240
241	conn_limiter_opt(#{name := _} = Opt, _DefName) ->
242	Opt;	×
243	conn_limiter_opt(Opt, DefName) when is_map(Opt) ->
244	Opt#{name => DefName};	6✔
245	conn_limiter_opt(_Opt, _DefName) ->
246	undefined.	69✔
247
248	conn_rate_limiter(undefined) ->
249	undefined;	69✔
250	conn_rate_limiter(Opts) ->
251	esockd_generic_limiter:create(Opts).	6✔

emqx / esockd / 352

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous