diff options
author | Simon MacMullen <simon@rabbitmq.com> | 2013-11-08 17:47:48 +0000 |
---|---|---|
committer | Simon MacMullen <simon@rabbitmq.com> | 2013-11-08 17:47:48 +0000 |
commit | 08a4f696987aec9e9721a42d15af5465af4526e0 (patch) | |
tree | 87c3e27475d0cad41497091bcfae642866de8575 /src/worker_pool_worker.erl | |
parent | eae43830da11e157d16f13c8eea0807727ce802a (diff) | |
download | rabbitmq-server-08a4f696987aec9e9721a42d15af5465af4526e0.tar.gz |
Cope with clients dying.
Diffstat (limited to 'src/worker_pool_worker.erl')
-rw-r--r-- | src/worker_pool_worker.erl | 45 |
1 files changed, 36 insertions, 9 deletions
diff --git a/src/worker_pool_worker.erl b/src/worker_pool_worker.erl index a976503f..028a6b3c 100644 --- a/src/worker_pool_worker.erl +++ b/src/worker_pool_worker.erl @@ -18,7 +18,7 @@ -behaviour(gen_server2). --export([start_link/1, submit/2, submit_async/2, run/1]). +-export([start_link/1, next_job_from/2, submit/2, submit_async/2, run/1]). -export([set_maximum_since_use/2]). @@ -32,6 +32,7 @@ -type(mfargs() :: {atom(), atom(), [any()]}). -spec(start_link/1 :: (any()) -> {'ok', pid()} | {'error', any()}). +-spec(next_job_from/2 :: (pid(), pid()) -> 'ok'). -spec(submit/2 :: (pid(), fun (() -> A) | mfargs()) -> A). -spec(submit_async/2 :: (pid(), fun (() -> any()) | mfargs()) -> 'ok'). -spec(run/1 :: (fun (() -> A)) -> A; (mfargs()) -> any()). @@ -44,13 +45,18 @@ -define(HIBERNATE_AFTER_MIN, 1000). -define(DESIRED_HIBERNATE, 10000). +-record(state, {id, next_job_from}). + %%---------------------------------------------------------------------------- start_link(WId) -> gen_server2:start_link(?MODULE, [WId], [{timeout, infinity}]). +next_job_from(Pid, CPid) -> + gen_server2:cast(Pid, {next_job_from, CPid}). + submit(Pid, Fun) -> - gen_server2:call(Pid, {submit, Fun}, infinity). + gen_server2:call(Pid, {submit, Fun, self()}, infinity). submit_async(Pid, Fun) -> gen_server2:cast(Pid, {submit_async, Fun}). @@ -70,32 +76,53 @@ init([WId]) -> [self()]), ok = worker_pool:idle(WId), put(worker_pool_worker, true), - {ok, WId, hibernate, + {ok, #state{id = WId}, hibernate, {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. prioritise_cast({set_maximum_since_use, _Age}, _Len, _State) -> 8; +prioritise_cast({next_job_from, _CPid}, _Len, _State) -> 7; prioritise_cast(_Msg, _Len, _State) -> 0. -handle_call({submit, Fun}, From, WId) -> +handle_call({submit, Fun, CPid}, From, State = #state{id = WId, + next_job_from = NJF}) -> + case NJF of + undefined -> receive {'$gen_cast', {next_job_from, CPid}} -> + ok + end; + {CPid, MRef} -> erlang:demonitor(MRef) + end, gen_server2:reply(From, run(Fun)), ok = worker_pool:idle(WId), - {noreply, WId, hibernate}; + {noreply, State#state{next_job_from = undefined}, hibernate}; handle_call(Msg, _From, State) -> {stop, {unexpected_call, Msg}, State}. -handle_cast({submit_async, Fun}, WId) -> +handle_cast({next_job_from, CPid}, State = #state{next_job_from = undefined}) -> + MRef = erlang:monitor(process, CPid), + {noreply, State#state{next_job_from = {CPid, MRef}}, hibernate}; + +handle_cast({submit_async, Fun}, State = #state{id = WId}) -> run(Fun), ok = worker_pool:idle(WId), - {noreply, WId, hibernate}; + {noreply, State, hibernate}; -handle_cast({set_maximum_since_use, Age}, WId) -> +handle_cast({set_maximum_since_use, Age}, State) -> ok = file_handle_cache:set_maximum_since_use(Age), - {noreply, WId, hibernate}; + {noreply, State, hibernate}; handle_cast(Msg, State) -> {stop, {unexpected_cast, Msg}, State}. +handle_info({'DOWN', MRef, process, CPid, _Reason}, + State = #state{id = WId, + next_job_from = {CPid, MRef}}) -> + ok = worker_pool:idle(WId), + {noreply, State#state{next_job_from = undefined}}; + +handle_info({'DOWN', _MRef, process, _Pid, _Reason}, State) -> + {noreply, State}; + handle_info(Msg, State) -> {stop, {unexpected_info, Msg}, State}. |