diff options
author | Matthias Radestock <matthias@lshift.net> | 2010-04-09 17:12:37 +0100 |
---|---|---|
committer | Matthias Radestock <matthias@lshift.net> | 2010-04-09 17:12:37 +0100 |
commit | 2db1402bf3bf6b1ffe80aa655eb24c0bbd5e4d7f (patch) | |
tree | ce41763ee4f81ca5e5c7a3808c197562ed0020a2 | |
parent | e367a2ecb372af75b234ecd0c2b601dc1f6d7575 (diff) | |
download | rabbitmq-server-2db1402bf3bf6b1ffe80aa655eb24c0bbd5e4d7f.tar.gz |
add worker_pook:submit_async/1
This was on the todo list and got cherry-picked from the bug21673 branch.
-rw-r--r-- | src/worker_pool.erl | 28 | ||||
-rw-r--r-- | src/worker_pool_worker.erl | 12 |
2 files changed, 34 insertions, 6 deletions
diff --git a/src/worker_pool.erl b/src/worker_pool.erl index 1ee958af..97e07545 100644 --- a/src/worker_pool.erl +++ b/src/worker_pool.erl @@ -40,12 +40,10 @@ %% %% 1. Allow priorities (basically, change the pending queue to a %% priority_queue). -%% -%% 2. Allow the submission to the pool_worker to be async. -behaviour(gen_server2). --export([start_link/0, submit/1, idle/1]). +-export([start_link/0, submit/1, submit_async/1, idle/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). @@ -56,6 +54,8 @@ -spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}). -spec(submit/1 :: (fun (() -> A) | {atom(), atom(), [any()]}) -> A). +-spec(submit_async/1 :: + (fun (() -> any()) | {atom(), atom(), [any()]}) -> 'ok'). -endif. @@ -80,6 +80,9 @@ submit(Fun) -> worker_pool_worker:submit(Pid, Fun) end. +submit_async(Fun) -> + gen_server2:cast(?SERVER, {run_async, Fun}). + idle(WId) -> gen_server2:cast(?SERVER, {idle, WId}). @@ -93,7 +96,8 @@ handle_call(next_free, From, State = #state { available = Avail, pending = Pending }) -> case queue:out(Avail) of {empty, _Avail} -> - {noreply, State #state { pending = queue:in(From, Pending) }, + {noreply, + State #state { pending = queue:in({next_free, From}, Pending) }, hibernate}; {{value, WId}, Avail1} -> {reply, get_worker_pid(WId), State #state { available = Avail1 }, @@ -108,11 +112,25 @@ handle_cast({idle, WId}, State = #state { available = Avail, {noreply, case queue:out(Pending) of {empty, _Pending} -> State #state { available = queue:in(WId, Avail) }; - {{value, From}, Pending1} -> + {{value, {next_free, From}}, Pending1} -> gen_server2:reply(From, get_worker_pid(WId)), + State #state { pending = Pending1 }; + {{value, {run_async, Fun}}, Pending1} -> + worker_pool_worker:submit_async(get_worker_pid(WId), Fun), State #state { pending = Pending1 } end, hibernate}; +handle_cast({run_async, Fun}, State = #state { available = Avail, + pending = Pending }) -> + {noreply, + case queue:out(Avail) of + {empty, _Avail} -> + State #state { pending = queue:in({run_async, Fun}, Pending)}; + {{value, WId}, Avail1} -> + worker_pool_worker:submit_async(get_worker_pid(WId), Fun), + State #state { available = Avail1 } + end, hibernate}; + handle_cast(Msg, State) -> {stop, {unexpected_cast, Msg}, State}. diff --git a/src/worker_pool_worker.erl b/src/worker_pool_worker.erl index 3bfcc2d9..d3a48119 100644 --- a/src/worker_pool_worker.erl +++ b/src/worker_pool_worker.erl @@ -33,7 +33,7 @@ -behaviour(gen_server2). --export([start_link/1, submit/2, run/1]). +-export([start_link/1, submit/2, submit_async/2, run/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). @@ -44,6 +44,8 @@ -spec(start_link/1 :: (any()) -> {'ok', pid()} | 'ignore' | {'error', any()}). -spec(submit/2 :: (pid(), fun (() -> A) | {atom(), atom(), [any()]}) -> A). +-spec(submit_async/2 :: + (pid(), fun (() -> any()) | {atom(), atom(), [any()]}) -> 'ok'). -endif. @@ -60,6 +62,9 @@ start_link(WId) -> submit(Pid, Fun) -> gen_server2:call(Pid, {submit, Fun}, infinity). +submit_async(Pid, Fun) -> + gen_server2:cast(Pid, {submit_async, Fun}). + init([WId]) -> ok = worker_pool:idle(WId), put(worker_pool_worker, true), @@ -74,6 +79,11 @@ handle_call({submit, Fun}, From, WId) -> handle_call(Msg, _From, State) -> {stop, {unexpected_call, Msg}, State}. +handle_cast({submit_async, Fun}, WId) -> + run(Fun), + ok = worker_pool:idle(WId), + {noreply, WId, hibernate}; + handle_cast(Msg, State) -> {stop, {unexpected_cast, Msg}, State}. |