diff options
Diffstat (limited to 'src/cowboy_listener.erl')
| -rw-r--r-- | src/cowboy_listener.erl | 172 |
1 files changed, 105 insertions, 67 deletions
diff --git a/src/cowboy_listener.erl b/src/cowboy_listener.erl index c19d079..b12e059 100644 --- a/src/cowboy_listener.erl +++ b/src/cowboy_listener.erl @@ -16,15 +16,21 @@ -module(cowboy_listener). -behaviour(gen_server). --export([start_link/0, stop/1, - add_connection/3, move_connection/3, remove_connection/2, wait/3]). %% API. +-export([start_link/2, stop/1, + add_connection/4, move_connection/3, remove_connection/2, + get_protocol_options/1, set_protocol_options/2]). %% API. -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). %% gen_server. +-type pools() :: [{atom(), non_neg_integer()}]. + -record(state, { - req_pools = [] :: [{atom(), non_neg_integer()}], + req_pools = [] :: pools(), reqs_table :: ets:tid(), - queue = [] :: [{pid(), reference()}] + queue = undefined :: queue(), + max_conns = undefined :: non_neg_integer(), + proto_opts :: any(), + proto_opts_vsn = 1 :: non_neg_integer() }). %% API. @@ -36,9 +42,10 @@ %% Setting the process priority to high ensures the connection-related code %% will always be executed when a connection needs it, allowing Cowboy to %% scale far beyond what it would with a normal priority. --spec start_link() -> {ok, pid()}. -start_link() -> - gen_server:start_link(?MODULE, [], [{spawn_opt, [{priority, high}]}]). +-spec start_link(non_neg_integer(), any()) -> {ok, pid()}. +start_link(MaxConns, ProtoOpts) -> + gen_server:start_link(?MODULE, [MaxConns, ProtoOpts], + [{spawn_opt, [{priority, high}]}]). %% @private -spec stop(pid()) -> stopped. @@ -50,15 +57,22 @@ stop(ServerPid) -> %% Pools of connections are used to restrict the maximum number of connections %% depending on their type. By default, Cowboy add all connections to the %% pool <em>default</em>. It also checks for the maximum number of connections -%% in that pool before accepting again. +%% in that pool before accepting again. This function only returns when there +%% is free space in the pool. %% %% When a process managing a connection dies, the process is removed from the %% pool. If the socket has been sent to another process, it is up to the %% protocol code to inform the listener of the new <em>ConnPid</em> by removing %% the previous and adding the new one. --spec add_connection(pid(), atom(), pid()) -> {ok, non_neg_integer()}. -add_connection(ServerPid, Pool, ConnPid) -> - gen_server:call(ServerPid, {add_connection, Pool, ConnPid}). +%% +%% This function also returns whether the protocol options have been modified. +%% If so, then an {upgrade, ProtoOpts, OptsVsn} will be returned instead of +%% the atom 'ok'. The acceptor can then continue with the new protocol options. +-spec add_connection(pid(), atom(), pid(), non_neg_integer()) + -> ok | {upgrade, any(), non_neg_integer()}. +add_connection(ServerPid, Pool, ConnPid, OptsVsn) -> + gen_server:call(ServerPid, {add_connection, Pool, ConnPid, OptsVsn}, + infinity). %% @doc Move a connection from one pool to another. -spec move_connection(pid(), atom(), pid()) -> ok. @@ -70,47 +84,48 @@ move_connection(ServerPid, DestPool, ConnPid) -> remove_connection(ServerPid, ConnPid) -> gen_server:cast(ServerPid, {remove_connection, ConnPid}). -%% @doc Wait until the number of connections in the given pool gets below -%% the given threshold. -%% -%% This function will not return until the number of connections in the pool -%% gets below <em>MaxConns</em>. It makes use of <em>gen_server:reply/2</em> -%% to make the process wait for a reply indefinitely. --spec wait(pid(), atom(), non_neg_integer()) -> ok. -wait(ServerPid, Pool, MaxConns) -> - gen_server:call(ServerPid, {wait, Pool, MaxConns}, infinity). +%% @doc Return the current protocol options. +-spec get_protocol_options(pid()) -> {ok, any()}. +get_protocol_options(ServerPid) -> + gen_server:call(ServerPid, get_protocol_options). + +%% @doc Upgrade the protocol options. +-spec set_protocol_options(pid(), any()) -> ok. +set_protocol_options(ServerPid, ProtoOpts) -> + gen_server:call(ServerPid, {set_protocol_options, ProtoOpts}). %% gen_server. %% @private --spec init([]) -> {ok, #state{}}. -init([]) -> - ReqsTablePid = ets:new(requests_table, [set, private]), - {ok, #state{reqs_table=ReqsTablePid}}. +-spec init(list()) -> {ok, #state{}}. +init([MaxConns, ProtoOpts]) -> + ReqsTable = ets:new(requests_table, [set, private]), + Queue = queue:new(), + {ok, #state{reqs_table=ReqsTable, max_conns=MaxConns, + proto_opts=ProtoOpts, queue=Queue}}. %% @private -spec handle_call(_, _, State) -> {reply, ignored, State} | {stop, normal, stopped, State}. -handle_call({add_connection, Pool, ConnPid}, _From, State=#state{ - req_pools=Pools, reqs_table=ReqsTable}) -> - MonitorRef = erlang:monitor(process, ConnPid), - {NbConnsRet, Pools2} = case lists:keyfind(Pool, 1, Pools) of - false -> - {1, [{Pool, 1}|Pools]}; - {Pool, NbConns} -> - NbConns2 = NbConns + 1, - {NbConns2, [{Pool, NbConns2}|lists:keydelete(Pool, 1, Pools)]} - end, - ets:insert(ReqsTable, {ConnPid, {MonitorRef, Pool}}), - {reply, {ok, NbConnsRet}, State#state{req_pools=Pools2}}; -handle_call({wait, Pool, MaxConns}, From, State=#state{ - req_pools=Pools, queue=Queue}) -> - case lists:keyfind(Pool, 1, Pools) of - {Pool, NbConns} when NbConns > MaxConns -> - {noreply, State#state{queue=[From|Queue]}}; - _Any -> - {reply, ok, State} +handle_call({add_connection, Pool, ConnPid, AccOptsVsn}, From, State=#state{ + req_pools=Pools, reqs_table=ReqsTable, + queue=Queue, max_conns=MaxConns, + proto_opts=ProtoOpts, proto_opts_vsn=LisOptsVsn}) -> + {NbConns, Pools2} = add_pid(ConnPid, Pool, Pools, ReqsTable), + State2 = State#state{req_pools=Pools2}, + if AccOptsVsn =/= LisOptsVsn -> + {reply, {ugprade, ProtoOpts, LisOptsVsn}, State2}; + NbConns > MaxConns -> + Queue2 = queue:in(From, Queue), + {noreply, State2#state{queue=Queue2}}; + true -> + {reply, ok, State2} end; +handle_call(get_protocol_options, _From, State=#state{proto_opts=ProtoOpts}) -> + {reply, {ok, ProtoOpts}, State}; +handle_call({set_protocol_options, ProtoOpts}, _From, + State=#state{proto_opts_vsn=OptsVsn}) -> + {reply, ok, State#state{proto_opts=ProtoOpts, proto_opts_vsn=OptsVsn + 1}}; handle_call(stop, _From, State) -> {stop, normal, stopped, State}; handle_call(_Request, _From, State) -> @@ -120,27 +135,21 @@ handle_call(_Request, _From, State) -> -spec handle_cast(_, State) -> {noreply, State}. handle_cast({move_connection, DestPool, ConnPid}, State=#state{ req_pools=Pools, reqs_table=ReqsTable}) -> - {MonitorRef, SrcPool} = ets:lookup_element(ReqsTable, ConnPid, 2), - ets:insert(ReqsTable, {ConnPid, {MonitorRef, DestPool}}), - {SrcPool, SrcNbConns} = lists:keyfind(SrcPool, 1, Pools), - DestNbConns = case lists:keyfind(DestPool, 1, Pools) of - false -> 1; - {DestPool, NbConns} -> NbConns + 1 - end, - Pools2 = lists:keydelete(SrcPool, 1, lists:keydelete(DestPool, 1, Pools)), - Pools3 = [{SrcPool, SrcNbConns - 1}, {DestPool, DestNbConns}|Pools2], - {noreply, State#state{req_pools=Pools3}}; -handle_cast({remove_connection, ConnPid}, State) -> - State2 = remove_pid(ConnPid, State), - {noreply, State2}; + Pools2 = move_pid(ConnPid, DestPool, Pools, ReqsTable), + {noreply, State#state{req_pools=Pools2}}; +handle_cast({remove_connection, ConnPid}, State=#state{ + req_pools=Pools, reqs_table=ReqsTable, queue=Queue}) -> + {Pools2, Queue2} = remove_pid(ConnPid, Pools, ReqsTable, Queue), + {noreply, State#state{req_pools=Pools2, queue=Queue2}}; handle_cast(_Msg, State) -> {noreply, State}. %% @private -spec handle_info(_, State) -> {noreply, State}. -handle_info({'DOWN', _Ref, process, Pid, _Info}, State) -> - State2 = remove_pid(Pid, State), - {noreply, State2}; +handle_info({'DOWN', _Ref, process, Pid, _Info}, State=#state{ + req_pools=Pools, reqs_table=ReqsTable, queue=Queue}) -> + {Pools2, Queue2} = remove_pid(Pid, Pools, ReqsTable, Queue), + {noreply, State#state{req_pools=Pools2, queue=Queue2}}; handle_info(_Info, State) -> {noreply, State}. @@ -157,18 +166,47 @@ code_change(_OldVsn, State, _Extra) -> %% Internal. %% @private --spec remove_pid(pid(), State) -> State. -remove_pid(Pid, State=#state{ - req_pools=Pools, reqs_table=ReqsTable, queue=Queue}) -> +-spec add_pid(pid(), atom(), pools(), ets:tid()) + -> {non_neg_integer(), pools()}. +add_pid(ConnPid, Pool, Pools, ReqsTable) -> + MonitorRef = erlang:monitor(process, ConnPid), + ConnPid ! {shoot, self()}, + {NbConnsRet, Pools2} = case lists:keyfind(Pool, 1, Pools) of + false -> + {1, [{Pool, 1}|Pools]}; + {Pool, NbConns} -> + NbConns2 = NbConns + 1, + {NbConns2, [{Pool, NbConns2}|lists:keydelete(Pool, 1, Pools)]} + end, + ets:insert(ReqsTable, {ConnPid, {MonitorRef, Pool}}), + {NbConnsRet, Pools2}. + +%% @private +-spec move_pid(pid(), atom(), pools(), ets:tid()) -> pools(). +move_pid(ConnPid, DestPool, Pools, ReqsTable) -> + {MonitorRef, SrcPool} = ets:lookup_element(ReqsTable, ConnPid, 2), + ets:insert(ReqsTable, {ConnPid, {MonitorRef, DestPool}}), + {SrcPool, SrcNbConns} = lists:keyfind(SrcPool, 1, Pools), + DestNbConns = case lists:keyfind(DestPool, 1, Pools) of + false -> 1; + {DestPool, NbConns} -> NbConns + 1 + end, + Pools2 = lists:keydelete(SrcPool, 1, lists:keydelete(DestPool, 1, Pools)), + [{SrcPool, SrcNbConns - 1}, {DestPool, DestNbConns}|Pools2]. + +%% @private +-spec remove_pid(pid(), pools(), ets:tid(), queue()) -> {pools(), queue()}. +remove_pid(Pid, Pools, ReqsTable, Queue) -> {MonitorRef, Pool} = ets:lookup_element(ReqsTable, Pid, 2), erlang:demonitor(MonitorRef, [flush]), {Pool, NbConns} = lists:keyfind(Pool, 1, Pools), Pools2 = [{Pool, NbConns - 1}|lists:keydelete(Pool, 1, Pools)], ets:delete(ReqsTable, Pid), - case Queue of - [] -> - State#state{req_pools=Pools2}; - [Client|Queue2] -> + case queue:len(Queue) of + 0 -> + {Pools2, Queue}; + _ -> + {{value, Client}, Queue2} = queue:out(Queue), gen_server:reply(Client, ok), - State#state{req_pools=Pools2, queue=Queue2} + {Pools2, Queue2} end. |