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

93.88

/src/esockd_acceptor_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_acceptor_sup).

-behaviour(supervisor).

-export([ start_link/6
        , start_supervised/1
        ]).

-export([ start_acceptors/2
        , start_acceptor/2
        , count_acceptors/1
        ]).

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

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

-define(ACCEPTOR_POOL, 16).

%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------

%% @doc Start Acceptor Supervisor.
-spec(start_link(atom(), esockd:listen_on(), pid(),
                 esockd:sock_fun(), [esockd:sock_fun()], esockd_generic_limiter:limiter())
     -> {ok, pid()}).
start_link(Proto, ListenOn, ConnSup, TuneFun, UpgradeFuns, Limiter) ->
    supervisor:start_link(?MODULE, { esockd_acceptor
                                   , [Proto, ListenOn, ConnSup,
                                      TuneFun, UpgradeFuns, Limiter]}).

%% @doc Start Acceptor Supervisor.
-spec start_supervised(esockd:listener_ref()) -> {ok, pid()}.
start_supervised(ListenerRef = {Proto, ListenOn}) ->
    Type = esockd_server:get_listener_prop(ListenerRef, type),
    Opts = esockd_server:get_listener_prop(ListenerRef, options),
    TuneFun = tune_socket_fun(Opts),
    UpgradeFuns = upgrade_funs(Type, Opts),
    LimiterOpts = esockd_listener_sup:conn_limiter_opts(Opts, {listener, Proto, ListenOn}),
    Limiter = esockd_listener_sup:conn_rate_limiter(LimiterOpts),
    AcceptorMod = case Type of
                         dtls -> esockd_dtls_acceptor;
                         _ -> esockd_acceptor
                     end,
    ConnSup = esockd_server:get_listener_prop(ListenerRef, connection_sup),
    AcceptorArgs = [Proto, ListenOn, ConnSup, TuneFun, UpgradeFuns, Limiter],
    case supervisor:start_link(?MODULE, {AcceptorMod, AcceptorArgs}) of
        {ok, Pid} ->
            _ = esockd_server:set_listener_prop(ListenerRef, acceptor_sup, Pid),
            {ok, Pid};
        {error, _} = Error ->
            Error
    end.

%% @doc Start acceptors.
-spec start_acceptors(esockd:listener_ref(), inet:socket()) -> ok.
start_acceptors(ListenerRef, LSock) ->
    Opts = esockd_server:get_listener_prop(ListenerRef, options),
    AcceptorNum = proplists:get_value(acceptors, Opts, ?ACCEPTOR_POOL),
    AcceptorSup = esockd_server:get_listener_prop(ListenerRef, acceptor_sup),
    lists:foreach(
        fun(_) -> {ok, _} = start_acceptor(AcceptorSup, LSock) end,
        lists:seq(1, AcceptorNum)
    ).

%% @doc Start a acceptor.
-spec(start_acceptor(pid(), inet:socket()) -> {ok, pid()} | {error, term()}).
start_acceptor(AcceptorSup, LSock) ->
    supervisor:start_child(AcceptorSup, [LSock]).

%% @doc Count acceptors.
-spec(count_acceptors(AcceptorSup :: pid()) -> pos_integer()).
count_acceptors(AcceptorSup) ->
    proplists:get_value(active, supervisor:count_children(AcceptorSup), 0).

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

init({AcceptorMod, AcceptorArgs}) ->
    SupFlags = #{strategy => simple_one_for_one,
                 intensity => 100000,
                 period => 1
                },
    Acceptor = #{id => acceptor,
                 start => {AcceptorMod, start_link, AcceptorArgs},
                 restart => transient,
                 shutdown => 1000,
                 type => worker,
                 modules => [AcceptorMod]
                },
    {ok, {SupFlags, [Acceptor]}}.

%%--------------------------------------------------------------------
%% Internal functions
%% -------------------------------------------------------------------

tune_socket_fun(Opts) ->
    TuneOpts = [ {tune_buffer, proplists:get_bool(tune_buffer, Opts)}
                 %% optional callback, returns ok | {error, Reason}
               , {tune_fun, proplists:get_value(tune_fun, Opts, undefined)}],
    {fun ?MODULE:tune_socket/2, [TuneOpts]}.

upgrade_funs(Type, Opts) ->
    lists:append([proxy_upgrade_fun(Opts), ssl_upgrade_fun(Type, Opts)]).

proxy_upgrade_fun(Opts) ->
    case proplists:get_bool(proxy_protocol, Opts) of
        false -> [];
        true  -> [esockd_transport:proxy_upgrade_fun(Opts)]
    end.

ssl_upgrade_fun(Type, Opts) ->
    Key = case Type of
              dtls -> dtls_options;
              _ -> ssl_options
          end,
    case proplists:get_value(Key, Opts) of
        undefined -> [];
        SslOpts -> [esockd_transport:ssl_upgrade_fun(SslOpts)]
    end.

tune_socket(Sock, []) ->
    {ok, Sock};
tune_socket(Sock, [{tune_buffer, true}|More]) ->
    case esockd_transport:getopts(Sock, [sndbuf, recbuf, buffer]) of
        {ok, BufSizes} ->
            BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]),
            _ = esockd_transport:setopts(Sock, [{buffer, BufSz}]),
            tune_socket(Sock, More);
        Error -> Error
   end;
tune_socket(Sock, [{tune_fun, undefined} | More]) ->
    tune_socket(Sock, More);
tune_socket(Sock, [{tune_fun, {M, F, A}} | More]) ->
    case apply(M, F, A) of
        ok ->
            tune_socket(Sock, More);
        Error -> Error
    end;
tune_socket(Sock, [_|More]) ->
    tune_socket(Sock, More).

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_acceptor_sup).
18
19	-behaviour(supervisor).
20
21	-export([ start_link/6
22	, start_supervised/1
23	]).
24
25	-export([ start_acceptors/2
26	, start_acceptor/2
27	, count_acceptors/1
28	]).
29
30	%% Supervisor callbacks
31	-export([init/1]).
32
33	%% callbacks
34	-export([tune_socket/2]).
35
36	-define(ACCEPTOR_POOL, 16).
37
38	%%--------------------------------------------------------------------
39	%% API
40	%%--------------------------------------------------------------------
41
42	%% @doc Start Acceptor Supervisor.
43	-spec(start_link(atom(), esockd:listen_on(), pid(),
44	esockd:sock_fun(), [esockd:sock_fun()], esockd_generic_limiter:limiter())
45	-> {ok, pid()}).
46	start_link(Proto, ListenOn, ConnSup, TuneFun, UpgradeFuns, Limiter) ->
NEW 47	supervisor:start_link(?MODULE, { esockd_acceptor	×
48	, [Proto, ListenOn, ConnSup,
49	TuneFun, UpgradeFuns, Limiter]}).
50
51	%% @doc Start Acceptor Supervisor.
52	-spec start_supervised(esockd:listener_ref()) -> {ok, pid()}.
53	start_supervised(ListenerRef = {Proto, ListenOn}) ->
54	Type = esockd_server:get_listener_prop(ListenerRef, type),	67✔
55	Opts = esockd_server:get_listener_prop(ListenerRef, options),	67✔
56	TuneFun = tune_socket_fun(Opts),	67✔
57	UpgradeFuns = upgrade_funs(Type, Opts),	67✔
58	LimiterOpts = esockd_listener_sup:conn_limiter_opts(Opts, {listener, Proto, ListenOn}),	67✔
59	Limiter = esockd_listener_sup:conn_rate_limiter(LimiterOpts),	67✔
60	AcceptorMod = case Type of	67✔
61	dtls -> esockd_dtls_acceptor;	12✔
62	_ -> esockd_acceptor	55✔
63	end,
64	ConnSup = esockd_server:get_listener_prop(ListenerRef, connection_sup),	67✔
65	AcceptorArgs = [Proto, ListenOn, ConnSup, TuneFun, UpgradeFuns, Limiter],	67✔
66	case supervisor:start_link(?MODULE, {AcceptorMod, AcceptorArgs}) of	67✔
67	{ok, Pid} ->
68	_ = esockd_server:set_listener_prop(ListenerRef, acceptor_sup, Pid),	67✔
69	{ok, Pid};	67✔
70	{error, _} = Error ->
NEW 71	Error	×
72	end.
73
74	%% @doc Start acceptors.
75	-spec start_acceptors(esockd:listener_ref(), inet:socket()) -> ok.
76	start_acceptors(ListenerRef, LSock) ->
77	Opts = esockd_server:get_listener_prop(ListenerRef, options),	67✔
78	AcceptorNum = proplists:get_value(acceptors, Opts, ?ACCEPTOR_POOL),	67✔
79	AcceptorSup = esockd_server:get_listener_prop(ListenerRef, acceptor_sup),	67✔
80	lists:foreach(	67✔
81	fun(_) -> {ok, _} = start_acceptor(AcceptorSup, LSock) end,	928✔
82	lists:seq(1, AcceptorNum)
83	).
84
85	%% @doc Start a acceptor.
86	-spec(start_acceptor(pid(), inet:socket()) -> {ok, pid()} \| {error, term()}).
87	start_acceptor(AcceptorSup, LSock) ->
88	supervisor:start_child(AcceptorSup, [LSock]).	928✔
89
90	%% @doc Count acceptors.
91	-spec(count_acceptors(AcceptorSup :: pid()) -> pos_integer()).
92	count_acceptors(AcceptorSup) ->
93	proplists:get_value(active, supervisor:count_children(AcceptorSup), 0).	8✔
94
95	%%--------------------------------------------------------------------
96	%% Supervisor callbacks
97	%%--------------------------------------------------------------------
98
99	init({AcceptorMod, AcceptorArgs}) ->
100	SupFlags = #{strategy => simple_one_for_one,	67✔
101	intensity => 100000,
102	period => 1
103	},
104	Acceptor = #{id => acceptor,	67✔
105	start => {AcceptorMod, start_link, AcceptorArgs},
106	restart => transient,
107	shutdown => 1000,
108	type => worker,
109	modules => [AcceptorMod]
110	},
111	{ok, {SupFlags, [Acceptor]}}.	67✔
112
113	%%--------------------------------------------------------------------
114	%% Internal functions
115	%% -------------------------------------------------------------------
116
117	tune_socket_fun(Opts) ->
118	TuneOpts = [ {tune_buffer, proplists:get_bool(tune_buffer, Opts)}	67✔
119	%% optional callback, returns ok \| {error, Reason}
120	, {tune_fun, proplists:get_value(tune_fun, Opts, undefined)}],
121	{fun ?MODULE:tune_socket/2, [TuneOpts]}.	67✔
122
123	upgrade_funs(Type, Opts) ->
124	lists:append([proxy_upgrade_fun(Opts), ssl_upgrade_fun(Type, Opts)]).	67✔
125
126	proxy_upgrade_fun(Opts) ->
127	case proplists:get_bool(proxy_protocol, Opts) of	67✔
128	false -> [];	54✔
129	true -> [esockd_transport:proxy_upgrade_fun(Opts)]	13✔
130	end.
131
132	ssl_upgrade_fun(Type, Opts) ->
133	Key = case Type of	67✔
134	dtls -> dtls_options;	12✔
135	_ -> ssl_options	55✔
136	end,
137	case proplists:get_value(Key, Opts) of	67✔
138	undefined -> [];	54✔
139	SslOpts -> [esockd_transport:ssl_upgrade_fun(SslOpts)]	13✔
140	end.
141
142	tune_socket(Sock, []) ->
143	{ok, Sock};	50✔
144	tune_socket(Sock, [{tune_buffer, true}\|More]) ->
145	case esockd_transport:getopts(Sock, [sndbuf, recbuf, buffer]) of	1✔
146	{ok, BufSizes} ->
147	BufSz = lists:max([Sz \|\| {_Opt, Sz} <- BufSizes]),	1✔
148	_ = esockd_transport:setopts(Sock, [{buffer, BufSz}]),	1✔
149	tune_socket(Sock, More);	1✔
NEW 150	Error -> Error	×
151	end;
152	tune_socket(Sock, [{tune_fun, undefined} \| More]) ->
153	tune_socket(Sock, More);	49✔
154	tune_socket(Sock, [{tune_fun, {M, F, A}} \| More]) ->
155	case apply(M, F, A) of	3✔
156	ok ->
157	tune_socket(Sock, More);	1✔
158	Error -> Error	2✔
159	end;
160	tune_socket(Sock, [_\|More]) ->
161	tune_socket(Sock, More).	51✔

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