1%% @author Bob Ippolito <bob@mochimedia.com>
2%% @copyright 2007 Mochi Media, Inc.
3
4%% @doc MochiWeb socket server.
5
6-module(mochiweb_socket_server).
7-author('bob@mochimedia.com').
8-behaviour(gen_server).
9
10-include("internal.hrl").
11
12-export([start/1, start_link/1, stop/1]).
13-export([init/1, handle_call/3, handle_cast/2, terminate/2, code_change/3,
14         handle_info/2]).
15-export([get/2, set/3]).
16
17-record(mochiweb_socket_server,
18        {port,
19         loop,
20         name=undefined,
21         max=2048,
22         ip=any,
23         listen=null,
24         nodelay=false,
25         recbuf=?RECBUF_SIZE,
26         backlog=128,
27         active_sockets=0,
28         acceptor_pool_size=16,
29         ssl=false,
30         ssl_opts=[{ssl_imp, new}],
31         acceptor_pool=sets:new(),
32         profile_fun=undefined}).
33
34-define(is_old_state(State), not is_record(State, mochiweb_socket_server)).
35
36start_link(Options) ->
37    start_server(start_link, parse_options(Options)).
38
39start(Options) ->
40    case lists:keytake(link, 1, Options) of
41        {value, {_Key, false}, Options1} ->
42            start_server(start, parse_options(Options1));
43        _ ->
44            %% TODO: https://github.com/mochi/mochiweb/issues/58
45            %%   [X] Phase 1: Add new APIs (Sep 2011)
46            %%   [_] Phase 2: Add deprecation warning
47            %%   [_] Phase 3: Change default to {link, false} and ignore link
48            %%   [_] Phase 4: Add deprecation warning for {link, _} option
49            %%   [_] Phase 5: Remove support for {link, _} option
50            start_link(Options)
51    end.
52
53get(Name, Property) ->
54    gen_server:call(Name, {get, Property}).
55
56set(Name, profile_fun, Fun) ->
57    gen_server:cast(Name, {set, profile_fun, Fun});
58set(Name, Property, _Value) ->
59    error_logger:info_msg("?MODULE:set for ~p with ~p not implemented~n",
60                          [Name, Property]).
61
62stop(Name) when is_atom(Name) orelse is_pid(Name) ->
63    gen_server:call(Name, stop);
64stop({Scope, Name}) when Scope =:= local orelse Scope =:= global ->
65    stop(Name);
66stop(Options) ->
67    State = parse_options(Options),
68    stop(State#mochiweb_socket_server.name).
69
70%% Internal API
71
72parse_options(State=#mochiweb_socket_server{}) ->
73    State;
74parse_options(Options) ->
75    parse_options(Options, #mochiweb_socket_server{}).
76
77parse_options([], State=#mochiweb_socket_server{acceptor_pool_size=PoolSize,
78                                                max=Max}) ->
79    case Max < PoolSize of
80        true ->
81            error_logger:info_report([{warning, "max is set lower than acceptor_pool_size"},
82                                      {max, Max},
83                                      {acceptor_pool_size, PoolSize}]);
84        false ->
85            ok
86    end,
87    State;
88parse_options([{name, L} | Rest], State) when is_list(L) ->
89    Name = {local, list_to_atom(L)},
90    parse_options(Rest, State#mochiweb_socket_server{name=Name});
91parse_options([{name, A} | Rest], State) when A =:= undefined ->
92    parse_options(Rest, State#mochiweb_socket_server{name=A});
93parse_options([{name, A} | Rest], State) when is_atom(A) ->
94    Name = {local, A},
95    parse_options(Rest, State#mochiweb_socket_server{name=Name});
96parse_options([{name, Name} | Rest], State) ->
97    parse_options(Rest, State#mochiweb_socket_server{name=Name});
98parse_options([{port, L} | Rest], State) when is_list(L) ->
99    Port = list_to_integer(L),
100    parse_options(Rest, State#mochiweb_socket_server{port=Port});
101parse_options([{port, Port} | Rest], State) ->
102    parse_options(Rest, State#mochiweb_socket_server{port=Port});
103parse_options([{ip, Ip} | Rest], State) ->
104    ParsedIp = case Ip of
105                   any ->
106                       any;
107                   Ip when is_tuple(Ip) ->
108                       Ip;
109                   Ip when is_list(Ip) ->
110                       {ok, IpTuple} = inet_parse:address(Ip),
111                       IpTuple
112               end,
113    parse_options(Rest, State#mochiweb_socket_server{ip=ParsedIp});
114parse_options([{loop, Loop} | Rest], State) ->
115    parse_options(Rest, State#mochiweb_socket_server{loop=Loop});
116parse_options([{backlog, Backlog} | Rest], State) ->
117    parse_options(Rest, State#mochiweb_socket_server{backlog=Backlog});
118parse_options([{nodelay, NoDelay} | Rest], State) ->
119    parse_options(Rest, State#mochiweb_socket_server{nodelay=NoDelay});
120parse_options([{recbuf, RecBuf} | Rest], State) when is_integer(RecBuf) orelse
121                                                RecBuf == undefined ->
122    %% XXX: `recbuf' value which is passed to `gen_tcp'
123    %% and value reported by `inet:getopts(P, [recbuf])' may
124    %% differ. They depends on underlying OS. From linux mans:
125    %%
126    %% The kernel doubles this value (to allow space for
127    %% bookkeeping overhead) when it is set using setsockopt(2),
128    %% and this doubled value is returned by getsockopt(2).
129    %%
130    %% See: man 7 socket | grep SO_RCVBUF
131    %%
132    %% In case undefined is passed instead of the default buffer
133    %% size ?RECBUF_SIZE, no size is set and the OS can control it dynamically
134    parse_options(Rest, State#mochiweb_socket_server{recbuf=RecBuf});
135parse_options([{acceptor_pool_size, Max} | Rest], State) ->
136    MaxInt = ensure_int(Max),
137    parse_options(Rest,
138                  State#mochiweb_socket_server{acceptor_pool_size=MaxInt});
139parse_options([{max, Max} | Rest], State) ->
140    MaxInt = ensure_int(Max),
141    parse_options(Rest, State#mochiweb_socket_server{max=MaxInt});
142parse_options([{ssl, Ssl} | Rest], State) when is_boolean(Ssl) ->
143    parse_options(Rest, State#mochiweb_socket_server{ssl=Ssl});
144parse_options([{ssl_opts, SslOpts} | Rest], State) when is_list(SslOpts) ->
145    SslOpts1 = [{ssl_imp, new} | proplists:delete(ssl_imp, SslOpts)],
146    parse_options(Rest, State#mochiweb_socket_server{ssl_opts=SslOpts1});
147parse_options([{profile_fun, ProfileFun} | Rest], State) when is_function(ProfileFun) ->
148    parse_options(Rest, State#mochiweb_socket_server{profile_fun=ProfileFun}).
149
150
151start_server(F, State=#mochiweb_socket_server{ssl=Ssl, name=Name}) ->
152    ok = prep_ssl(Ssl),
153    case Name of
154        undefined ->
155            gen_server:F(?MODULE, State, []);
156        _ ->
157            gen_server:F(Name, ?MODULE, State, [])
158    end.
159
160prep_ssl(true) ->
161    ok = mochiweb:ensure_started(crypto),
162    ok = mochiweb:ensure_started(asn1),
163    ok = mochiweb:ensure_started(public_key),
164    ok = mochiweb:ensure_started(ssl);
165prep_ssl(false) ->
166    ok.
167
168ensure_int(N) when is_integer(N) ->
169    N;
170ensure_int(S) when is_list(S) ->
171    list_to_integer(S).
172
173ipv6_supported() ->
174    case (catch inet:getaddr("localhost", inet6)) of
175        {ok, _Addr} ->
176            true;
177        {error, _} ->
178            false
179    end.
180
181init(State=#mochiweb_socket_server{ip=Ip, port=Port, backlog=Backlog,
182                                   nodelay=NoDelay, recbuf=RecBuf}) ->
183    process_flag(trap_exit, true),
184
185    BaseOpts = [binary,
186                {reuseaddr, true},
187                {packet, 0},
188                {backlog, Backlog},
189                {exit_on_close, false},
190                {active, false},
191                {nodelay, NoDelay}],
192    Opts = case Ip of
193        any ->
194            case ipv6_supported() of % IPv4, and IPv6 if supported
195                true -> [inet, inet6 | BaseOpts];
196                _ -> BaseOpts
197            end;
198        {_, _, _, _} -> % IPv4
199            [inet, {ip, Ip} | BaseOpts];
200        {_, _, _, _, _, _, _, _} -> % IPv6
201            [inet6, {ip, Ip} | BaseOpts]
202    end,
203    OptsBuf=case RecBuf of
204        undefined ->
205            Opts;
206        _ ->
207            [{recbuf, RecBuf}|Opts]
208    end,
209    listen(Port, OptsBuf, State).
210
211new_acceptor_pool(State=#mochiweb_socket_server{acceptor_pool_size=Size}) ->
212    lists:foldl(fun (_, S) -> new_acceptor(S) end, State, lists:seq(1, Size)).
213
214new_acceptor(State=#mochiweb_socket_server{acceptor_pool=Pool,
215                                           recbuf=RecBuf,
216                                           loop=Loop,
217                                           listen=Listen}) ->
218    LoopOpts = [{recbuf, RecBuf}],
219    Pid = mochiweb_acceptor:start_link(self(), Listen, Loop, LoopOpts),
220    State#mochiweb_socket_server{
221      acceptor_pool=sets:add_element(Pid, Pool)}.
222
223listen(Port, Opts, State=#mochiweb_socket_server{ssl=Ssl, ssl_opts=SslOpts}) ->
224    case mochiweb_socket:listen(Ssl, Port, Opts, SslOpts) of
225        {ok, Listen} ->
226            {ok, ListenPort} = mochiweb_socket:port(Listen),
227            {ok, new_acceptor_pool(State#mochiweb_socket_server{
228                                     listen=Listen,
229                                     port=ListenPort})};
230        {error, Reason} ->
231            {stop, Reason}
232    end.
233
234do_get(port, #mochiweb_socket_server{port=Port}) ->
235    Port;
236do_get(waiting_acceptors, #mochiweb_socket_server{acceptor_pool=Pool}) ->
237    sets:size(Pool);
238do_get(active_sockets, #mochiweb_socket_server{active_sockets=ActiveSockets}) ->
239    ActiveSockets.
240
241
242state_to_proplist(#mochiweb_socket_server{name=Name,
243                                          port=Port,
244                                          active_sockets=ActiveSockets}) ->
245    [{name, Name}, {port, Port}, {active_sockets, ActiveSockets}].
246
247upgrade_state(State = #mochiweb_socket_server{}) ->
248    State;
249upgrade_state({mochiweb_socket_server, Port, Loop, Name,
250             Max, IP, Listen, NoDelay, Backlog, ActiveSockets,
251             AcceptorPoolSize, SSL, SSL_opts,
252             AcceptorPool}) ->
253    #mochiweb_socket_server{port=Port, loop=Loop, name=Name, max=Max, ip=IP,
254                            listen=Listen, nodelay=NoDelay, backlog=Backlog,
255                            active_sockets=ActiveSockets,
256                            acceptor_pool_size=AcceptorPoolSize,
257                            ssl=SSL,
258                            ssl_opts=SSL_opts,
259                            acceptor_pool=AcceptorPool}.
260
261handle_call(Req, From, State) when ?is_old_state(State) ->
262    handle_call(Req, From, upgrade_state(State));
263handle_call({get, Property}, _From, State) ->
264    Res = do_get(Property, State),
265    {reply, Res, State};
266handle_call(stop, _From, State) ->
267    {stop, normal, ok, State};
268handle_call(_Message, _From, State) ->
269    Res = error,
270    {reply, Res, State}.
271
272
273handle_cast(Req, State) when ?is_old_state(State) ->
274    handle_cast(Req, upgrade_state(State));
275handle_cast({accepted, Pid, Timing},
276            State=#mochiweb_socket_server{active_sockets=ActiveSockets}) ->
277    State1 = State#mochiweb_socket_server{active_sockets=1 + ActiveSockets},
278    case State#mochiweb_socket_server.profile_fun of
279        undefined ->
280            undefined;
281        F when is_function(F) ->
282            catch F([{timing, Timing} | state_to_proplist(State1)])
283    end,
284    {noreply, recycle_acceptor(Pid, State1)};
285handle_cast({set, profile_fun, ProfileFun}, State) ->
286    State1 = case ProfileFun of
287                 ProfileFun when is_function(ProfileFun); ProfileFun =:= undefined ->
288                     State#mochiweb_socket_server{profile_fun=ProfileFun};
289                 _ ->
290                     State
291             end,
292    {noreply, State1}.
293
294
295terminate(Reason, State) when ?is_old_state(State) ->
296    terminate(Reason, upgrade_state(State));
297terminate(_Reason, #mochiweb_socket_server{listen=Listen}) ->
298    mochiweb_socket:close(Listen).
299
300code_change(_OldVsn, State, _Extra) ->
301    State.
302
303recycle_acceptor(Pid, State=#mochiweb_socket_server{
304                        acceptor_pool=Pool,
305                        acceptor_pool_size=PoolSize,
306                        max=Max,
307                        active_sockets=ActiveSockets}) ->
308    %% A socket is considered to be active from immediately after it
309    %% has been accepted (see the {accepted, Pid, Timing} cast above).
310    %% This function will be called when an acceptor is transitioning
311    %% to an active socket, or when either type of Pid dies. An acceptor
312    %% Pid will always be in the acceptor_pool set, and an active socket
313    %% will be in that set during the transition but not afterwards.
314    Pool1 = sets:del_element(Pid, Pool),
315    NewSize = sets:size(Pool1),
316    ActiveSockets1 = case NewSize =:= sets:size(Pool) of
317                         %% Pid has died and it is not in the acceptor set,
318                         %% it must be an active socket.
319                         true -> max(0, ActiveSockets - 1);
320                         false -> ActiveSockets
321                     end,
322    State1 = State#mochiweb_socket_server{
323               acceptor_pool=Pool1,
324               active_sockets=ActiveSockets1},
325    %% Spawn a new acceptor only if it will not overrun the maximum socket
326    %% count or the maximum pool size.
327    case NewSize + ActiveSockets1 < Max andalso NewSize < PoolSize of
328        true -> new_acceptor(State1);
329        false -> State1
330    end.
331
332handle_info(Msg, State) when ?is_old_state(State) ->
333    handle_info(Msg, upgrade_state(State));
334handle_info({'EXIT', Pid, normal}, State) ->
335    {noreply, recycle_acceptor(Pid, State)};
336handle_info({'EXIT', Pid, Reason},
337            State=#mochiweb_socket_server{acceptor_pool=Pool}) ->
338    case sets:is_element(Pid, Pool) of
339        true ->
340            %% If there was an unexpected error accepting, log and sleep.
341            error_logger:error_report({?MODULE, ?LINE,
342                                       {acceptor_error, Reason}}),
343            timer:sleep(100);
344        false ->
345            ok
346    end,
347    {noreply, recycle_acceptor(Pid, State)};
348
349% this is what release_handler needs to get a list of modules,
350% since our supervisor modules list is set to 'dynamic'
351% see sasl-2.1.9.2/src/release_handler_1.erl get_dynamic_mods
352handle_info({From, Tag, get_modules}, State = #mochiweb_socket_server{name={local,Mod}}) ->
353    From ! {element(2,Tag), [Mod]},
354    {noreply, State};
355
356% If for some reason we can't get the module name, send empty list to avoid release_handler timeout:
357handle_info({From, Tag, get_modules}, State) ->
358    error_logger:info_msg("mochiweb_socket_server replying to dynamic modules request as '[]'~n",[]),
359    From ! {element(2,Tag), []},
360    {noreply, State};
361
362handle_info(Info, State) ->
363    error_logger:info_report([{'INFO', Info}, {'State', State}]),
364    {noreply, State}.
365
366
367
368%%
369%% Tests
370%%
371-ifdef(TEST).
372-include_lib("eunit/include/eunit.hrl").
373
374upgrade_state_test() ->
375    OldState = {mochiweb_socket_server,
376                port, loop, name,
377                max, ip, listen,
378                nodelay, backlog,
379                active_sockets,
380                acceptor_pool_size,
381                ssl, ssl_opts, acceptor_pool},
382    State = upgrade_state(OldState),
383    CmpState = #mochiweb_socket_server{port=port, loop=loop,
384                                       name=name, max=max, ip=ip,
385                                       listen=listen, nodelay=nodelay,
386                                       backlog=backlog,
387                                       active_sockets=active_sockets,
388                                       acceptor_pool_size=acceptor_pool_size,
389                                       ssl=ssl, ssl_opts=ssl_opts,
390                                       acceptor_pool=acceptor_pool,
391                                       profile_fun=undefined},
392    ?assertEqual(CmpState, State).
393
394-endif.
395