diff options
author | Matthew Sackman <matthew@lshift.net> | 2010-01-19 16:00:00 +0000 |
---|---|---|
committer | Matthew Sackman <matthew@lshift.net> | 2010-01-19 16:00:00 +0000 |
commit | f439db9f457c1c8697f9b5252d8579bd426945f7 (patch) | |
tree | a21f9d1a9225a22b4c5dda7a6c6edddc37ca5340 | |
parent | e787774a8d8ceb671f623b25d7991ce8b3786e97 (diff) | |
parent | 82982eca2d9b8214fe14aadbb8c3f12ef270175c (diff) | |
download | rabbitmq-server-f439db9f457c1c8697f9b5252d8579bd426945f7.tar.gz |
merging in from bug21673
-rw-r--r-- | src/file_handle_cache.erl | 59 | ||||
-rw-r--r-- | src/rabbit_reader.erl | 4 | ||||
-rw-r--r-- | src/tcp_acceptor.erl | 26 |
3 files changed, 55 insertions, 34 deletions
diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl index aa01a508..b289e842 100644 --- a/src/file_handle_cache.erl +++ b/src/file_handle_cache.erl @@ -125,7 +125,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_on_death/1, obtain/0]). -define(SERVER, ?MODULE). -define(RESERVED_FOR_OTHERS, 50). @@ -160,7 +160,8 @@ -record(fhc_state, { elders, limit, - count + count, + obtains }). %%---------------------------------------------------------------------------- @@ -432,11 +433,11 @@ set_maximum_since_use(MaximumAge) -> false -> ok end. -decrement() -> - gen_server:cast(?SERVER, decrement). +release_on_death(Pid) when is_pid(Pid) -> + gen_server:cast(?SERVER, {release_on_death, Pid}). -increment() -> - gen_server:cast(?SERVER, increment). +obtain() -> + gen_server:call(?SERVER, obtain, infinity). %%---------------------------------------------------------------------------- %% Internal functions @@ -663,10 +664,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 }) -> @@ -687,20 +695,20 @@ 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 }}; - -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 }}; + {noreply, process_obtains(State #fhc_state { elders = Elders1, + count = Count - 1 })}; handle_cast(check_counts, State) -> - {noreply, maybe_reduce(State)}. + {noreply, maybe_reduce(State)}; -handle_info(_Msg, State) -> +handle_cast({release_on_death, Pid}, State) -> + _MRef = erlang:monitor(process, Pid), {noreply, State}. +handle_info({'DOWN', _MRef, process, _Pid, _Reason}, + State = #fhc_state { count = Count }) -> + {noreply, process_obtains(State #fhc_state { count = Count - 1 })}. + terminate(_Reason, State) -> State. @@ -711,6 +719,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_reader.erl b/src/rabbit_reader.erl index 49e66e32..adfd412f 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,6 @@ 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(), 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 bc742561..6de6ac3e 100644 --- a/src/tcp_acceptor.erl +++ b/src/tcp_acceptor.erl @@ -48,22 +48,20 @@ 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}) -> %% patch up the socket so it looks like one we got from - %% gen_tcp:accept/1 + %% gen_tcp:accept/1 {ok, Mod} = inet_db:lookup_socket(LSock), inet_db:register_socket(Sock, Mod), @@ -75,7 +73,7 @@ handle_info({inet_async, LSock, Ref, {ok, Sock}}, [inet_parse:ntoa(Address), Port, inet_parse:ntoa(PeerAddress), PeerPort]), %% handle - apply(M, F, A ++ [Sock]) + file_handle_cache:release_on_death(apply(M, F, A ++ [Sock])) catch {inet_error, Reason} -> gen_tcp:close(Sock), error_logger:error_msg("unable to accept TCP connection: ~p~n", @@ -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. |