Make accept asynchronous

Ranch now accepts connection asynchronously through a separate
process. The accept process is linked to the acceptor, calls
accept and does nothing else but send the socket back to the
acceptor. This allows us to receive messages in the acceptor
to handle upgrades instead of polling. This will also allow us
later to make acceptors system processes.

Remove support for connection pools in favor of a simpler
max_connections setting. Connections can be removed from the
count, allowing us to have as many long-lived connections as
we want while still limiting the number of short-lived ones.

Add max_connections, max_connections with long-lived connections,
and upgrade tests.

1 files changed, 28 insertions, 1 deletions

diff --git a/src/ranch_server.erl b/src/ranch_server.erl
index 16e892d..faec9b6 100644
--- a/src/ranch_server.erl
+++ b/src/ranch_server.erl
@@ -22,6 +22,9 @@
 -export([lookup_listener/1]).
 -export([add_acceptor/2]).
 -export([send_to_acceptors/2]).
+-export([add_connection/1]).
+-export([count_connections/1]).
+-export([remove_connection/1]).
 
 %% gen_server.
 -export([init/1]).
@@ -69,12 +72,31 @@ send_to_acceptors(Ref, Msg) ->
 	_ = [Pid ! Msg || Pid <- Acceptors],
 	ok.
 
+%% @doc Add a connection to the connection pool.
+%%
+%% Also return the number of connections in the pool after this operation.
+-spec add_connection(pid()) -> non_neg_integer().
+add_connection(ListenerPid) ->
+	ets:update_counter(?TAB, {connections, ListenerPid}, 1).
+
+%% @doc Count the number of connections in the connection pool.
+-spec count_connections(pid()) -> non_neg_integer().
+count_connections(ListenerPid) ->
+	ets:update_counter(?TAB, {connections, ListenerPid}, 0).
+
+%% @doc Remove a connection from the connection pool.
+%%
+%% Also return the number of connections in the pool after this operation.
+-spec remove_connection(pid()) -> non_neg_integer().
+remove_connection(ListenerPid) ->
+	ets:update_counter(?TAB, {connections, ListenerPid}, -1).
+
 %% gen_server.
 
 %% @private
 init([]) ->
 	?TAB = ets:new(?TAB, [
-		ordered_set, public, named_table, {read_concurrency, true}]),
+		ordered_set, public, named_table, {write_concurrency, true}]),
 	{ok, #state{}}.
 
 %% @private
@@ -84,6 +106,7 @@ handle_call(_Request, _From, State) ->
 %% @private
 handle_cast({insert_listener, Ref, Pid}, State=#state{monitors=Monitors}) ->
 	true = ets:insert_new(?TAB, {{acceptors, Ref}, []}),
+	true = ets:insert_new(?TAB, {{connections, Pid}, 0}),
 	MonitorRef = erlang:monitor(process, Pid),
 	{noreply, State#state{
 		monitors=[{{MonitorRef, Pid}, {listener, Ref}}|Monitors]}};
@@ -93,6 +116,9 @@ handle_cast({add_acceptor, Ref, Pid}, State=#state{monitors=Monitors}) ->
 	true = ets:insert(?TAB, {{acceptors, Ref}, [Pid|Acceptors]}),
 	{noreply, State#state{
 		monitors=[{{MonitorRef, Pid}, {acceptors, Ref}}|Monitors]}};
+handle_cast({add_connection, Pid}, State) ->
+	_ = erlang:monitor(process, Pid),
+	{noreply, State};
 handle_cast(_Request, State) ->
 	{noreply, State}.
 
@@ -120,6 +146,7 @@ code_change(_OldVsn, State, _Extra) ->
 remove_process(Key = {listener, Ref}, MonitorRef, Pid, Monitors) ->
 	true = ets:delete(?TAB, Key),
 	true = ets:delete(?TAB, {acceptors, Ref}),
+	true = ets:delete(?TAB, {connections, Pid}),
 	lists:keydelete({MonitorRef, Pid}, 1, Monitors);
 remove_process(Key = {acceptors, _}, MonitorRef, Pid, Monitors) ->
 	Acceptors = ets:lookup_element(?TAB, Key, 2),