diff options
Diffstat (limited to 'src/delegate.erl')
| -rw-r--r-- | src/delegate.erl | 87 |
1 files changed, 70 insertions, 17 deletions
diff --git a/src/delegate.erl b/src/delegate.erl index f680d94ab3..5277e59fcc 100644 --- a/src/delegate.erl +++ b/src/delegate.erl @@ -18,22 +18,33 @@ -behaviour(gen_server2). --export([start_link/1, invoke_no_result/2, invoke/2, call/2, cast/2]). +-export([start_link/1, invoke_no_result/2, invoke/2, monitor/2, + demonitor/1, demonitor/2, call/2, cast/2]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). +-record(state, {node, monitors, name}). + %%---------------------------------------------------------------------------- -ifdef(use_specs). +-export_type([monitor_ref/0]). + +-type(monitor_ref() :: reference() | {atom(), pid()}). -type(fun_or_mfa(A) :: fun ((pid()) -> A) | {atom(), atom(), [any()]}). + -spec(start_link/1 :: (non_neg_integer()) -> {'ok', pid()} | ignore | {'error', any()}). -spec(invoke/2 :: ( pid(), fun_or_mfa(A)) -> A; ([pid()], fun_or_mfa(A)) -> {[{pid(), A}], [{pid(), term()}]}). -spec(invoke_no_result/2 :: (pid() | [pid()], fun_or_mfa(any())) -> 'ok'). +-spec(monitor/2 :: ('process', pid()) -> monitor_ref()). +-spec(demonitor/1 :: (monitor_ref()) -> 'true'). +-spec(demonitor/2 :: (monitor_ref(), ['flush']) -> 'true'). + -spec(call/2 :: ( pid(), any()) -> any(); ([pid()], any()) -> {[{pid(), any()}], [{pid(), term()}]}). @@ -49,7 +60,8 @@ %%---------------------------------------------------------------------------- start_link(Num) -> - gen_server2:start_link({local, delegate_name(Num)}, ?MODULE, [], []). + Name = delegate_name(Num), + gen_server2:start_link({local, Name}, ?MODULE, [Name], []). invoke(Pid, FunOrMFA) when is_pid(Pid) andalso node(Pid) =:= node() -> apply1(FunOrMFA, Pid); @@ -77,7 +89,7 @@ invoke(Pids, FunOrMFA) when is_list(Pids) -> case orddict:fetch_keys(Grouped) of [] -> {[], []}; RemoteNodes -> gen_server2:multi_call( - RemoteNodes, delegate(RemoteNodes), + RemoteNodes, delegate(self(), RemoteNodes), {invoke, FunOrMFA, Grouped}, infinity) end, BadPids = [{Pid, {exit, {nodedown, BadNode}, []}} || @@ -105,12 +117,27 @@ invoke_no_result(Pids, FunOrMFA) when is_list(Pids) -> {LocalPids, Grouped} = group_pids_by_node(Pids), case orddict:fetch_keys(Grouped) of [] -> ok; - RemoteNodes -> gen_server2:abcast(RemoteNodes, delegate(RemoteNodes), - {invoke, FunOrMFA, Grouped}) + RemoteNodes -> gen_server2:abcast( + RemoteNodes, delegate(self(), RemoteNodes), + {invoke, FunOrMFA, Grouped}) end, safe_invoke(LocalPids, FunOrMFA), %% must not die ok. +monitor(Type, Pid) when node(Pid) =:= node() -> + erlang:monitor(Type, Pid); +monitor(Type, Pid) -> + Name = delegate(Pid, [node(Pid)]), + gen_server2:cast(Name, {monitor, Type, self(), Pid}), + {Name, Pid}. + +demonitor(Ref) -> ?MODULE:demonitor(Ref, []). + +demonitor(Ref, Options) when is_reference(Ref) -> + erlang:demonitor(Ref, Options); +demonitor({Name, Pid}, Options) -> + gen_server2:cast(Name, {demonitor, Pid, Options}). + call(PidOrPids, Msg) -> invoke(PidOrPids, {gen_server2, call, [Msg, infinity]}). @@ -133,10 +160,10 @@ group_pids_by_node(Pids) -> delegate_name(Hash) -> list_to_atom("delegate_" ++ integer_to_list(Hash)). -delegate(RemoteNodes) -> +delegate(Pid, RemoteNodes) -> case get(delegate) of undefined -> Name = delegate_name( - erlang:phash2(self(), + erlang:phash2(Pid, delegate_sup:count(RemoteNodes))), put(delegate, Name), Name; @@ -157,23 +184,49 @@ apply1(Fun, Arg) -> Fun(Arg). %%---------------------------------------------------------------------------- -init([]) -> - {ok, node(), hibernate, +init([Name]) -> + {ok, #state{node = node(), monitors = dict:new(), name = Name}, hibernate, {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. -handle_call({invoke, FunOrMFA, Grouped}, _From, Node) -> - {reply, safe_invoke(orddict:fetch(Node, Grouped), FunOrMFA), Node, +handle_call({invoke, FunOrMFA, Grouped}, _From, State = #state{node = Node}) -> + {reply, safe_invoke(orddict:fetch(Node, Grouped), FunOrMFA), State, hibernate}. -handle_cast({invoke, FunOrMFA, Grouped}, Node) -> +handle_cast({monitor, Type, WantsMonitor, Pid}, + State = #state{monitors = Monitors}) -> + Ref = erlang:monitor(Type, Pid), + Monitors1 = dict:store(Pid, {WantsMonitor, Ref}, Monitors), + {noreply, State#state{monitors = Monitors1}, hibernate}; + +handle_cast({demonitor, Pid, Options}, + State = #state{monitors = Monitors}) -> + {noreply, case dict:find(Pid, Monitors) of + {ok, {_WantsMonitor, Ref}} -> + erlang:demonitor(Ref, Options), + State#state{monitors = dict:erase(Pid, Monitors)}; + error -> + State + end, hibernate}; + +handle_cast({invoke, FunOrMFA, Grouped}, State = #state{node = Node}) -> safe_invoke(orddict:fetch(Node, Grouped), FunOrMFA), - {noreply, Node, hibernate}. + {noreply, State, hibernate}. + +handle_info({'DOWN', Ref, process, Pid, Info}, + State = #state{monitors = Monitors, name = Name}) -> + {noreply, case dict:find(Pid, Monitors) of + {ok, {WantsMonitor, Ref}} -> + WantsMonitor ! {'DOWN', {Name, Pid}, process, Pid, Info}, + State#state{monitors = dict:erase(Pid, Monitors)}; + error -> + State + end, hibernate}; -handle_info(_Info, Node) -> - {noreply, Node, hibernate}. +handle_info(_Info, State) -> + {noreply, State, hibernate}. terminate(_Reason, _State) -> ok. -code_change(_OldVsn, Node, _Extra) -> - {ok, Node}. +code_change(_OldVsn, State, _Extra) -> + {ok, State}. |