emqx / esockd / 356

Committed 15 Dec 2023 02:41PM UTC coverage: 72.414%. First build

Build # 356

Build Type

Pull #183

github

Committed by

web-flow

Commit Message

Merge 8bfe0c31c into 5cb22a8b1

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

Run Details

191 of 213 new or added lines in 10 files covered. (89.67%)

840 of 1160 relevant lines covered (72.41%)

62.94 hits per line

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

89.39

/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),
    Result = try
        ensure_ok(esockd_connection_sup:set_options(ConnSup, Opts)),
        ensure_ok(Listener:set_options(ListenerPid, Opts))
    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),
            Error
    end,
    ok = restart_acceptor_sup(ListenerRef, Sup),
    Result.

restart_acceptor_sup(ListenerRef, Sup) ->
    _ = supervisor:terminate_child(Sup, acceptor_sup),
    {ok, _Child} = supervisor:restart_child(Sup, acceptor_sup),
    _ = start_acceptors(ListenerRef),
    ok.

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	Result = try	9✔
121	ensure_ok(esockd_connection_sup:set_options(ConnSup, Opts)),	9✔
122	ensure_ok(Listener:set_options(ListenerPid, Opts))	8✔
123	catch
124	throw:{?MODULE, Error} ->
125	%% Restore previous options
126	_ = esockd_server:set_listener_prop(ListenerRef, options, OptsWas),	2✔
127	ok = esockd_connection_sup:set_options(ConnSup, OptsWas),	2✔
128	ok = Listener:set_options(ListenerPid, OptsWas),	2✔
129	Error	2✔
130	end,
131	ok = restart_acceptor_sup(ListenerRef, Sup),	9✔
132	Result.	9✔
133
134	restart_acceptor_sup(ListenerRef, Sup) ->
135	_ = supervisor:terminate_child(Sup, acceptor_sup),	9✔
136	{ok, _Child} = supervisor:restart_child(Sup, acceptor_sup),	9✔
137	_ = start_acceptors(ListenerRef),	9✔
138	ok.	9✔
139
140	ensure_ok(ok) ->
141	ok;	15✔
142	ensure_ok({error, _} = Error) ->
143	throw({?MODULE, Error}).	2✔
144
145	get_acceptors(Sup) ->
146	esockd_acceptor_sup:count_acceptors(acceptor_sup(Sup)).	8✔
147
148	get_max_connections(Sup) ->
149	esockd_connection_sup:get_max_connections(connection_sup(Sup)).	5✔
150
151	set_max_connections(ListenerRef, Sup, MaxConns) ->
152	set_options(ListenerRef, Sup, [{max_connections, MaxConns}]).	2✔
153
154	get_max_conn_rate(_Sup, Proto, ListenOn) ->
155	case esockd_limiter:lookup({listener, Proto, ListenOn}) of	4✔
156	undefined ->
157	{error, not_found};	×
158	#{capacity := Capacity, interval := Interval} ->
159	{Capacity, Interval}	4✔
160	end.
161
162	set_max_conn_rate(ListenerRef, Sup, Opt) ->
163	set_options(ListenerRef, Sup, [{limiter, Opt}]).	2✔
164
165	get_current_connections(Sup) ->
166	esockd_connection_sup:count_connections(connection_sup(Sup)).	3✔
167
168	get_shutdown_count(Sup) ->
169	esockd_connection_sup:get_shutdown_count(connection_sup(Sup)).	2✔
170
171	get_access_rules(Sup) ->
172	esockd_connection_sup:access_rules(connection_sup(Sup)).	9✔
173
174	allow(Sup, CIDR) ->
175	esockd_connection_sup:allow(connection_sup(Sup), CIDR).	3✔
176
177	deny(Sup, CIDR) ->
178	esockd_connection_sup:deny(connection_sup(Sup), CIDR).	3✔
179
180	%%--------------------------------------------------------------------
181	%% Supervisor callbacks
182	%%--------------------------------------------------------------------
183
184	init({ListenerRef, MFA}) ->
185	ConnSup = #{id => connection_sup,	56✔
186	start => {esockd_connection_sup, start_supervised, [ListenerRef, MFA]},
187	restart => transient,
188	shutdown => infinity,
189	type => supervisor,
190	modules => [esockd_connection_sup]},
191	ListenerMod = case esockd_server:get_listener_prop(ListenerRef, type) of	56✔
192	dtls -> esockd_dtls_listener;	10✔
193	_ -> esockd_listener	46✔
194	end,
195	Listener = #{id => listener,	56✔
196	start => {ListenerMod, start_supervised, [ListenerRef]},
197	restart => transient,
198	shutdown => 16#ffffffff,
199	type => worker,
200	modules => [ListenerMod]},
201	AcceptorSup = #{id => acceptor_sup,	56✔
202	start => {esockd_acceptor_sup, start_supervised, [ListenerRef]},
203	restart => transient,
204	shutdown => infinity,
205	type => supervisor,
206	modules => [esockd_acceptor_sup]},
207	Starter = #{id => starter,	56✔
208	start => {?MODULE, start_acceptors, [ListenerRef]},
209	restart => transient,
210	shutdown => infinity,
211	type => worker,
212	modules => []},
213	{ok, { {rest_for_one, 10, 3600}	56✔
214	, [ConnSup, Listener, AcceptorSup, Starter]
215	}}.
216
217	-spec start_acceptors(esockd:listener_ref()) -> ignore.
218	start_acceptors(ListenerRef) ->
219	{LMod, LPid} = esockd_server:get_listener_prop(ListenerRef, listener),	67✔
220	LState = LMod:get_state(LPid),	67✔
221	LSock = proplists:get_value(listen_sock, LState),	67✔
222	ok = esockd_acceptor_sup:start_acceptors(ListenerRef, LSock),	67✔
223	ignore.	67✔
224
225	%%--------------------------------------------------------------------
226	%% Sock tune/upgrade functions
227	%%--------------------------------------------------------------------
228
229	tune_socket(Sock, Tunings) ->
NEW 230	esockd_acceptor_sup:tune_socket(Sock, Tunings).	×
231
232	conn_limiter_opts(Opts, DefName) ->
233	Opt = proplists:get_value(limiter, Opts, undefined),	77✔
234	conn_limiter_opt(Opt, DefName).	77✔
235
236	conn_limiter_opt(#{name := _} = Opt, _DefName) ->
237	Opt;	×
238	conn_limiter_opt(Opt, DefName) when is_map(Opt) ->
239	Opt#{name => DefName};	6✔
240	conn_limiter_opt(_Opt, _DefName) ->
241	undefined.	72✔
242
243	conn_rate_limiter(undefined) ->
244	undefined;	72✔
245	conn_rate_limiter(Opts) ->
246	esockd_generic_limiter:create(Opts).	6✔

emqx / esockd / 356

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