emqx / esockd / 354

Committed 14 Dec 2023 12:54PM UTC coverage: 71.491%. First build

Build # 354

Build Type

Pull #183

github

Committed by

web-flow

Commit Message

Merge 3297859f4 into 5cb22a8b1

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

Run Details

170 of 192 new or added lines in 10 files covered. (88.54%)

820 of 1147 relevant lines covered (71.49%)

60.53 hits per line

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

88.24

/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).

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

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.

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) ->
    set_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) ->
    set_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]
         }}.

-spec start_acceptors(esockd:listener_ref()) -> ignore.
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.

%%--------------------------------------------------------------------
%% 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},	56✔
82	supervisor:start_link(?MODULE, {ListenerRef, MFA}).	56✔
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)).	8✔
99
100	%% @doc Get child pid with id.
101	get_sup_child(Sup, ChildId) ->
102	lists:keyfind(ChildId, 1, supervisor:which_children(Sup)).	37✔
103
104	get_child_pid({_, Pid, _, _}) -> Pid.	37✔
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	set_options(ListenerRef, Sup, Opts) ->
115	OptsWas = esockd_server:get_listener_prop(ListenerRef, options),	9✔
116	OptsWas = esockd_server:set_listener_prop(ListenerRef, options,	9✔
117	esockd:merge_opts(OptsWas, Opts)),
118	ConnSup = esockd_server:get_listener_prop(ListenerRef, connection_sup),	9✔
119	{Listener, ListenerPid} = esockd_server:get_listener_prop(ListenerRef, listener),	9✔
120	try	9✔
121	_ = ensure_ok(esockd_connection_sup:set_options(ConnSup, Opts)),	9✔
122	_ = ensure_ok(Listener:set_options(ListenerPid, Opts)),	8✔
123	_ = ensure_ok(restart_acceptor_sup(ListenerRef, Sup)),	7✔
124	ok	7✔
125	catch
126	throw:{?MODULE, Error} ->
127	%% Restore previous options
128	_ = esockd_server:set_listener_prop(ListenerRef, options, OptsWas),	2✔
129	ok = esockd_connection_sup:set_options(ConnSup, OptsWas),	2✔
130	ok = Listener:set_options(ListenerPid, OptsWas),	2✔
131	ok = restart_acceptor_sup(ListenerRef, Sup),	2✔
132	Error	2✔
133	end.
134
135	restart_acceptor_sup(ListenerRef, Sup) ->
136	_ = supervisor:terminate_child(Sup, acceptor_sup),	9✔
137	case supervisor:restart_child(Sup, acceptor_sup) of	9✔
138	{ok, _Child} ->
139	_ = start_acceptors(ListenerRef),	9✔
140	ok;	9✔
141	Error = {error, _} ->
NEW 142	Error	×
143	end.
144
145	ensure_ok(ok) ->
146	ok;	22✔
147	ensure_ok({error, _} = Error) ->
148	throw({?MODULE, Error}).	2✔
149
150	get_acceptors(Sup) ->
151	esockd_acceptor_sup:count_acceptors(acceptor_sup(Sup)).	8✔
152
153	get_max_connections(Sup) ->
154	esockd_connection_sup:get_max_connections(connection_sup(Sup)).	5✔
155
156	set_max_connections(ListenerRef, Sup, MaxConns) ->
157	set_options(ListenerRef, Sup, [{max_connections, MaxConns}]).	2✔
158
159	get_max_conn_rate(_Sup, Proto, ListenOn) ->
160	case esockd_limiter:lookup({listener, Proto, ListenOn}) of	4✔
161	undefined ->
162	{error, not_found};	×
163	#{capacity := Capacity, interval := Interval} ->
164	{Capacity, Interval}	4✔
165	end.
166
167	set_max_conn_rate(ListenerRef, Sup, Opt) ->
168	set_options(ListenerRef, Sup, [{limiter, Opt}]).	2✔
169
170	get_current_connections(Sup) ->
171	esockd_connection_sup:count_connections(connection_sup(Sup)).	3✔
172
173	get_shutdown_count(Sup) ->
174	esockd_connection_sup:get_shutdown_count(connection_sup(Sup)).	2✔
175
176	get_access_rules(Sup) ->
177	esockd_connection_sup:access_rules(connection_sup(Sup)).	9✔
178
179	allow(Sup, CIDR) ->
180	esockd_connection_sup:allow(connection_sup(Sup), CIDR).	3✔
181
182	deny(Sup, CIDR) ->
183	esockd_connection_sup:deny(connection_sup(Sup), CIDR).	3✔
184
185	%%--------------------------------------------------------------------
186	%% Supervisor callbacks
187	%%--------------------------------------------------------------------
188
189	init({ListenerRef, MFA}) ->
190	ConnSup = #{id => connection_sup,	56✔
191	start => {esockd_connection_sup, start_supervised, [ListenerRef, MFA]},
192	restart => transient,
193	shutdown => infinity,
194	type => supervisor,
195	modules => [esockd_connection_sup]},
196	ListenerMod = case esockd_server:get_listener_prop(ListenerRef, type) of	56✔
197	dtls -> esockd_dtls_listener;	10✔
198	_ -> esockd_listener	46✔
199	end,
200	Listener = #{id => listener,	56✔
201	start => {ListenerMod, start_supervised, [ListenerRef]},
202	restart => transient,
203	shutdown => 16#ffffffff,
204	type => worker,
205	modules => [ListenerMod]},
206	AcceptorSup = #{id => acceptor_sup,	56✔
207	start => {esockd_acceptor_sup, start_supervised, [ListenerRef]},
208	restart => transient,
209	shutdown => infinity,
210	type => supervisor,
211	modules => [esockd_acceptor_sup]},
212	Starter = #{id => starter,	56✔
213	start => {?MODULE, start_acceptors, [ListenerRef]},
214	restart => transient,
215	shutdown => infinity,
216	type => worker,
217	modules => []},
218	{ok, { {rest_for_one, 10, 3600}	56✔
219	, [ConnSup, Listener, AcceptorSup, Starter]
220	}}.
221
222	-spec start_acceptors(esockd:listener_ref()) -> ignore.
223	start_acceptors(ListenerRef) ->
224	{LMod, LPid} = esockd_server:get_listener_prop(ListenerRef, listener),	67✔
225	LState = LMod:get_state(LPid),	67✔
226	LSock = proplists:get_value(listen_sock, LState),	67✔
227	ok = esockd_acceptor_sup:start_acceptors(ListenerRef, LSock),	67✔
228	ignore.	67✔
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),	77✔
239	conn_limiter_opt(Opt, DefName).	77✔
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.	72✔
247
248	conn_rate_limiter(undefined) ->
249	undefined;	72✔
250	conn_rate_limiter(Opts) ->
251	esockd_generic_limiter:create(Opts).	6✔

emqx / esockd / 354

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