Initial pass at solution. Seems to work. Interesting realisation that the fhc server was never previously running (post new boot sequence).

4 files changed, 54 insertions, 29 deletions

diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl
index e8d7cf6e85..ac6de519b7 100644
--- a/src/file_handle_cache.erl
+++ b/src/file_handle_cache.erl
@@ -124,7 +124,7 @@
 -export([start_link/0, init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--export([decrement/0, increment/0]).
+-export([release/0, obtain/0]).
 
 -define(SERVER, ?MODULE).
 -define(RESERVED_FOR_OTHERS, 50).
@@ -159,7 +159,8 @@
 -record(fhc_state,
         { elders,
           limit,
-          count
+          count,
+          obtains
         }).
 
 %%----------------------------------------------------------------------------
@@ -431,11 +432,11 @@ set_maximum_since_use(MaximumAge) ->
         false -> ok
     end.
 
-decrement() ->
-    gen_server:cast(?SERVER, decrement).
+release() ->
+    gen_server:cast(?SERVER, release).
 
-increment() ->
-    gen_server:cast(?SERVER, increment).
+obtain() ->
+    gen_server:call(?SERVER, obtain, infinity).
 
 %%----------------------------------------------------------------------------
 %% Internal functions
@@ -662,10 +663,17 @@ init([]) ->
                     ulimit()
             end,
     error_logger:info_msg("Limiting to approx ~p file handles~n", [Limit]),
-    {ok, #fhc_state { elders = dict:new(), limit = Limit, count = 0}}.
-
-handle_call(_Msg, _From, State) ->
-    {reply, message_not_understood, State}.
+    {ok, #fhc_state { elders = dict:new(), limit = Limit, count = 0,
+                      obtains = [] }}.
+
+handle_call(obtain, From, State = #fhc_state { count = Count }) ->
+    State1 = #fhc_state { count = Count1, limit = Limit, obtains = Obtains } =
+        maybe_reduce(State #fhc_state { count = Count + 1 }),
+    case Limit /= infinity andalso Count1 >= Limit of
+        true  -> {noreply, State1 #fhc_state { obtains = [From | Obtains],
+                                               count = Count1 - 1 }};
+        false -> {reply, ok, State1}
+    end.
 
 handle_cast({open, Pid, EldestUnusedSince}, State =
             #fhc_state { elders = Elders, count = Count }) ->
@@ -686,13 +694,11 @@ handle_cast({close, Pid, EldestUnusedSince}, State =
                   undefined -> dict:erase(Pid, Elders);
                   _         -> dict:store(Pid, EldestUnusedSince, Elders)
               end,
-    {noreply, State #fhc_state { elders = Elders1, count = Count - 1 }};
+    {noreply, process_obtains(State #fhc_state { elders = Elders1,
+                                                 count = Count - 1 })};
 
-handle_cast(increment, State = #fhc_state { count = Count }) ->
-    {noreply, maybe_reduce(State #fhc_state { count = Count + 1 })};
-
-handle_cast(decrement, State = #fhc_state { count = Count }) ->
-    {noreply, State #fhc_state { count = Count - 1 }};
+handle_cast(release, State = #fhc_state { count = Count }) ->
+    {noreply, process_obtains(State #fhc_state { count = Count - 1 })};
 
 handle_cast(check_counts, State) ->
     {noreply, maybe_reduce(State)}.
@@ -710,6 +716,19 @@ code_change(_OldVsn, State, _Extra) ->
 %% server helpers
 %%----------------------------------------------------------------------------
 
+process_obtains(State = #fhc_state { obtains = [] }) ->
+    State;
+process_obtains(State = #fhc_state { limit = Limit, count = Count })
+  when Limit /= infinity andalso Count >= Limit ->
+    State;
+process_obtains(State = #fhc_state { limit = Limit, count = Count,
+                                     obtains = Obtains }) ->
+    Take = lists:min([length(Obtains), Limit - Count]),
+    {Obtainable, ObtainsNewRev} = lists:split(Take, lists:reverse(Obtains)),
+    [gen_server:reply(From, ok) || From <- Obtainable],
+    State #fhc_state { count = Count + Take,
+                       obtains = lists:reverse(ObtainsNewRev) }.
+
 maybe_reduce(State = #fhc_state { limit = Limit, count = Count,
                                   elders = Elders })
   when Limit /= infinity andalso Count >= Limit ->
diff --git a/src/rabbit.erl b/src/rabbit.erl
index ac7ad04660..a0cc14367f 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -63,6 +63,11 @@
                     {mfa,         {rabbit_hooks, start, []}},
                     {pre,         kernel_ready}]}).
 
+-rabbit_boot_step({file_handle_cache,
+                   [{description, "file handle cache server"},
+                    {mfa,         {rabbit_sup, start_child, [file_handle_cache]}},
+                    {pre,         kernel_ready}]}).
+
 -rabbit_boot_step({kernel_ready,
                    [{description, "kernel ready"}]}).
 
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 49e66e328e..c2818be4dd 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -213,8 +213,7 @@ start_connection(Parent, Deb, Sock, SockTransform) ->
     erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
                       handshake_timeout),
     ProfilingValue = setup_profiling(),
-    try 
-        file_handle_cache:increment(),
+    try
         mainloop(Parent, Deb, switch_callback(
                                 #v1{sock = ClientSock,
                                     connection = #connection{
@@ -235,7 +234,7 @@ start_connection(Parent, Deb, Sock, SockTransform) ->
                end)("exception on TCP connection ~p from ~s:~p~n~p~n",
                     [self(), PeerAddressS, PeerPort, Ex])
     after
-        file_handle_cache:decrement(),
+        file_handle_cache:release(),
         rabbit_log:info("closing TCP connection ~p from ~s:~p~n",
                         [self(), PeerAddressS, PeerPort]),
         %% We don't close the socket explicitly. The reader is the
diff --git a/src/tcp_acceptor.erl b/src/tcp_acceptor.erl
index bc7425613f..9e7962338a 100644
--- a/src/tcp_acceptor.erl
+++ b/src/tcp_acceptor.erl
@@ -48,16 +48,14 @@ start_link(Callback, LSock) ->
 %%--------------------------------------------------------------------
 
 init({Callback, LSock}) ->
-    case prim_inet:async_accept(LSock, -1) of
-        {ok, Ref} -> {ok, #state{callback=Callback, sock=LSock, ref=Ref}};
-        Error -> {stop, {cannot_accept, Error}}
-    end.
+    gen_server:cast(self(), accept),
+    {ok, #state{callback=Callback, sock=LSock, ref=undefined}}.
 
 handle_call(_Request, _From, State) ->
     {noreply, State}.
 
-handle_cast(_Msg, State) ->
-    {noreply, State}.
+handle_cast(accept, State) ->
+    accept(State).
 
 handle_info({inet_async, LSock, Ref, {ok, Sock}},
             State = #state{callback={M,F,A}, sock=LSock, ref=Ref}) ->
@@ -83,10 +81,7 @@ handle_info({inet_async, LSock, Ref, {ok, Sock}},
     end,
 
     %% accept more
-    case prim_inet:async_accept(LSock, -1) of
-        {ok, NRef} -> {noreply, State#state{ref=NRef}};
-        Error -> {stop, {cannot_accept, Error}, none}
-    end;
+    accept(State);
 handle_info({inet_async, LSock, Ref, {error, closed}},
             State=#state{sock=LSock, ref=Ref}) ->
     %% It would be wrong to attempt to restart the acceptor when we
@@ -104,3 +99,10 @@ code_change(_OldVsn, State, _Extra) ->
 %%--------------------------------------------------------------------
 
 inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
+
+accept(State = #state{sock=LSock}) ->
+    ok = file_handle_cache:obtain(),
+    case prim_inet:async_accept(LSock, -1) of
+        {ok, Ref} -> {noreply, State#state{ref=Ref}};
+        Error     -> {stop,    {cannot_accept, Error}}
+    end.