%% %% %CopyrightBegin% %% %% Copyright Ericsson AB 1996-2021. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. %% You may obtain a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, software %% distributed under the License is distributed on an "AS IS" BASIS, %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% See the License for the specific language governing permissions and %% limitations under the License. %% %% %CopyrightEnd% %% ----------------------------------------------------------------------------- %% @doc This module is an adaptation of Erlang's `gen' module which %% implements the really generic stuff of the generic standard behaviours (e.g. %% gen_server, gen_fsm). %% %% It replaces all instances of `erlang:send/2` and `erlang:monitor/2` with %% their Partisan counterparts. %% %% NOTICE: %% At the moment this only works for `partisan_pluggable_peer_service_manager'. %% @end %% ----------------------------------------------------------------------------- -module(partisan_gen). -include("partisan_logger.hrl"). % -compile({inline,[get_node/1]}). %%%----------------------------------------------------------------- %%% This module implements the really generic stuff of the generic %%% standard behaviours (e.g. gen_server, gen_fsm). %%% %%% The standard behaviour should export init_it/6. %%%----------------------------------------------------------------- -export([start/5, start/6, debug_options/2, hibernate_after/1, name/1, unregister_name/1, get_proc_name/1, get_parent/0, call/3, call/4, reply/2, send_request/3, wait_response/2, receive_response/2, check_response/2, stop/1, stop/3]). -export([init_it/6, init_it/7]). -export([format_status_header/2]). -export([get_channel/0]). -define(default_timeout, 5000). %%----------------------------------------------------------------- -type linkage() :: 'monitor' | 'link' | 'nolink'. % -type linkage() :: 'link' | 'nolink'. -type emgr_name() :: {'local', atom()} | {'global', term()} | {'via', Module :: module(), Name :: term()}. -type start_ret() :: {'ok', pid()} | {'ok', {pid(), reference()}} | 'ignore' | {'error', term()}. -type debug_flag() :: 'trace' | 'log' | 'statistics' | 'debug' | {'logfile', string()}. -type option() :: {'timeout', timeout()} | {'debug', [debug_flag()]} | {'hibernate_after', timeout()} | {'spawn_opt', [proc_lib:spawn_option()]}. -type options() :: [option()]. -type server_ref() :: pid() | atom() | {atom(), node()} | {global, term()} | {via, module(), term()}. -type request_id() :: term(). %%----------------------------------------------------------------- %% Starts a generic process. %% start(GenMod, LinkP, Mod, Args, Options) %% start(GenMod, LinkP, Name, Mod, Args, Options) %% GenMod = atom(), callback module implementing the 'real' fsm %% LinkP = link | nolink %% Name = {local, atom()} | {global, term()} | {via, atom(), term()} %% Args = term(), init arguments (to Mod:init/1) %% Options = [{timeout, Timeout} | {debug, [Flag]} | {spawn_opt, OptionList}] %% Flag = trace | log | {logfile, File} | statistics | debug %% (debug == log && statistics) %% Returns: {ok, Pid} | ignore |{error, Reason} | %% {error, {already_started, Pid}} | %% The 'already_started' is returned only if Name is given %%----------------------------------------------------------------- -spec start(module(), linkage(), emgr_name(), module(), term(), options()) -> start_ret(). start(GenMod, LinkP, Name, Mod, Args, Options) -> case where(Name) of undefined -> do_spawn(GenMod, LinkP, Name, Mod, Args, Options); Pid -> {error, {already_started, Pid}} end. -spec start(module(), linkage(), module(), term(), options()) -> start_ret(). start(GenMod, LinkP, Mod, Args, Options) -> do_spawn(GenMod, LinkP, Mod, Args, Options). %%----------------------------------------------------------------- %% Spawn the process (and link) maybe at another node. %% If spawn without link, set parent to ourselves 'self'!!! %%----------------------------------------------------------------- do_spawn(GenMod, link, Mod, Args, Options) -> Time = timeout(Options), proc_lib:start_link(?MODULE, init_it, [GenMod, self(), self(), Mod, Args, Options], Time, spawn_opts(Options)); do_spawn(GenMod, monitor, Mod, Args, Options) -> Time = timeout(Options), Ret = proc_lib:start_monitor(?MODULE, init_it, [GenMod, self(), self(), Mod, Args, Options], Time, spawn_opts(Options)), monitor_return(Ret); do_spawn(GenMod, _, Mod, Args, Options) -> Time = timeout(Options), proc_lib:start(?MODULE, init_it, [GenMod, self(), 'self', Mod, Args, Options], Time, spawn_opts(Options)). do_spawn(GenMod, link, Name, Mod, Args, Options) -> Time = timeout(Options), proc_lib:start_link(?MODULE, init_it, [GenMod, self(), self(), Name, Mod, Args, Options], Time, spawn_opts(Options)); do_spawn(GenMod, monitor, Name, Mod, Args, Options) -> Time = timeout(Options), Ret = proc_lib:start_monitor(?MODULE, init_it, [GenMod, self(), self(), Name, Mod, Args, Options], Time, spawn_opts(Options)), monitor_return(Ret); do_spawn(GenMod, _, Name, Mod, Args, Options) -> Time = timeout(Options), proc_lib:start(?MODULE, init_it, [GenMod, self(), 'self', Name, Mod, Args, Options], Time, spawn_opts(Options)). %% %% Adjust monitor returns for OTP gen behaviours... %% %% If an OTP behaviour is introduced that 'init_ack's %% other results, this has code has to be moved out %% into all behaviours as well as adjusted... %% % monitor_return({{ok, Pid}, Mon}) when is_pid(Pid), is_reference(Mon) -> % %% Successful start_monitor()... % {ok, {Pid, Mon}}; % monitor_return({Error, Mon}) when is_reference(Mon) -> % %% Failure; wait for spawned process to terminate % %% and release resources, then return the error... % receive % {'DOWN', Mon, process, _Pid, _Reason} -> % ok % end, % Error. monitor_return({{ok, Pid}, Mon}) -> %% Successful start_monitor()... case partisan:is_pid(Pid) andalso partisan:is_reference(Mon) of true -> {ok, {Pid, Mon}}; false -> %% Simulate same error we would have if this was the original %% commented code above error(function_clause) end; monitor_return({Error, Mon}) when is_reference(Mon) -> %% Failure; wait for spawned process to terminate %% and release resources, then return the error... case partisan:is_reference(Mon) of true -> receive {'DOWN', Mon, process, _Pid, _Reason} -> ok end, Error; false -> %% Simulate same error we would have if this was the original %% commented code above error(function_clause) end. %%----------------------------------------------------------------- %% Initiate the new process. %% Register the name using the Rfunc function %% Calls the Mod:init/Args function. %% Finally an acknowledge is sent to Parent and the main %% loop is entered. %%----------------------------------------------------------------- init_it(GenMod, Starter, Parent, Mod, Args, Options) -> init_it2(GenMod, Starter, Parent, self(), Mod, Args, Options). init_it(GenMod, Starter, Parent, Name, Mod, Args, Options) -> case register_name(Name) of true -> init_it2(GenMod, Starter, Parent, Name, Mod, Args, Options); {false, Pid} -> proc_lib:init_ack(Starter, {error, {already_started, Pid}}) end. init_it2(GenMod, Starter, Parent, Name, Mod, Args, Options) -> set_channel(Options), GenMod:init_it(Starter, Parent, Name, Mod, Args, Options). %%----------------------------------------------------------------- %% Makes a synchronous call to a generic process. %% Request is sent to the Pid, and the response must be %% {Tag, Reply}. %%----------------------------------------------------------------- %%% New call function which uses the new monitor BIF %%% call(ServerId, Label, Request) call(Process, Label, Request) -> call(Process, Label, Request, ?default_timeout). %% Optimize a common case. call(Process, Label, Request, Timeout) when is_pid(Process), Timeout =:= infinity orelse is_integer(Timeout) andalso Timeout >= 0 -> do_call(Process, Label, Request, Timeout); call(Process, Label, Request, Timeout) when Timeout =:= infinity; is_integer(Timeout), Timeout >= 0 -> Fun = fun(Arg) -> do_call(Arg, Label, Request, Timeout) end, do_for_proc(Process, Fun). -dialyzer({no_improper_lists, do_call/4}). % do_call(Process, Label, Request, Timeout) when is_atom(Process) =:= false -> % Mref = erlang:monitor(process, Process, [{alias,demonitor}]), % Tag = [alias | Mref], % %% OTP-24: % %% Using alias to prevent responses after 'noconnection' and timeouts. % %% We however still may call nodes responding via process identifier, so % %% we still use 'noconnect' on send in order to try to send on the % %% monitored connection, and not trigger a new auto-connect. % %% % erlang:send(Process, {Label, {self(), Tag}, Request}, [noconnect]), % receive % {[alias | Mref], Reply} -> % erlang:demonitor(Mref, [flush]), % {ok, Reply}; % {'DOWN', Mref, _, _, noconnection} -> % Node = get_node(Process), % exit({nodedown, Node}); % {'DOWN', Mref, _, _, Reason} -> % exit(Reason) % after Timeout -> % erlang:demonitor(Mref, [flush]), % receive % {[alias | Mref], Reply} -> % {ok, Reply} % after 0 -> % exit(timeout) % end % end. do_call(Process, Label, Request, Timeout) -> %% For remote calls we use Partisan Mref = do_send_request(Process, Label, Request), %% Wait for reply. receive {Mref, Reply} -> partisan:demonitor(Mref, [flush]), {ok, Reply}; {'DOWN', Mref, _, _, noconnection} -> Node = get_node(Process), exit({nodedown, Node}); {'DOWN', Mref, _, _, Reason} -> exit(Reason) after Timeout -> partisan:demonitor(Mref, [flush]), ?LOG_DEBUG(#{ description => "Timed out at while waiting for response to message", message => Request }), receive {Mref, Reply} -> {ok, Reply} after 0 -> exit(timeout) end end. get_node(Process) -> %% We trust the arguments to be correct, i.e %% Process is either a local or remote pid, %% or a {Name, Node} tuple (of atoms) and in this %% case this node (node()) _is_ distributed and Node =/= node(). case Process of {_S, N} when is_atom(N) -> N; _ when is_pid(Process) -> node(Process); _ -> %% Process is partisan_remote_ref partisan:node(Process) end. -spec send_request(Name::server_ref(), Label::term(), Request::term()) -> request_id(). send_request(Process, Label, Request) when is_pid(Process) -> do_send_request(Process, Label, Request); send_request(Process, Label, Request) -> Fun = fun(Pid) -> do_send_request(Pid, Label, Request) end, try do_for_proc(Process, Fun) catch exit:Reason -> %% Make send_request async and fake a down message Ref = partisan:make_ref(), self() ! {'DOWN', Ref, process, Process, Reason}, Ref end. % -dialyzer({no_improper_lists, do_send_request/3}). % do_send_request(Process, Label, Request) -> % Mref = erlang:monitor(process, Process, [{alias, demonitor}]), % erlang:send(Process, {Label, {self(), [alias|Mref]}, Request}, [noconnect]), % Mref. % %% % %% Wait for a reply to the client. % %% Note: if timeout is returned monitors are kept. -spec wait_response(RequestId::request_id(), timeout()) -> {reply, Reply::term()} | 'timeout' | {error, {term(), server_ref()}}. wait_response(Mref, Timeout) -> receive % {[alias|Mref], Reply} -> {Mref, Reply} -> partisan:demonitor(Mref, [flush]), {reply, Reply}; {'DOWN', Mref, _, Object, Reason} -> {error, {Reason, Object}} after Timeout -> timeout end. -spec receive_response(RequestId::request_id(), timeout()) -> {reply, Reply::term()} | 'timeout' | {error, {term(), server_ref()}}. receive_response(Mref, Timeout) when is_reference(Mref) -> receive % {[alias|Mref], Reply} -> {Mref, Reply} -> partisan:demonitor(Mref, [flush]), {reply, Reply}; {'DOWN', Mref, _, Object, Reason} -> {error, {Reason, Object}} after Timeout -> partisan:demonitor(Mref, [flush]), receive % {[alias|Mref], Reply} -> {Mref, Reply} -> {reply, Reply} after 0 -> timeout end end. -spec check_response(RequestId::term(), Key::request_id()) -> {reply, Reply::term()} | 'no_reply' | {error, {term(), server_ref()}}. check_response(Msg, Mref) -> case Msg of % {[alias|Mref], Reply} -> {Mref, Reply} -> partisan:demonitor(Mref, [flush]), {reply, Reply}; {'DOWN', Mref, _, Object, Reason} -> {error, {Reason, Object}}; {Mref, Reply} -> {reply, Reply}; _ -> no_reply end. %% %% Send a reply to the client. %% % reply({_To, [alias|Alias] = Tag}, Reply) when is_reference(Alias) -> % Alias ! {Tag, Reply}, ok; % reply({_To, [[alias|Alias] | _] = Tag}, Reply) when is_reference(Alias) -> % Alias ! {Tag, Reply}, ok; reply({To, Tag}, Reply) -> % try To ! {Tag, Reply}, ok catch _:_ -> ok end. try partisan:forward_message(To, {Tag, Reply}) catch _:_ -> ok end. %%----------------------------------------------------------------- %% Synchronously stop a generic process %%----------------------------------------------------------------- stop(Process) -> stop(Process, normal, infinity). stop(Process, Reason, Timeout) when Timeout =:= infinity; is_integer(Timeout), Timeout >= 0 -> Fun = fun(Pid) -> proc_lib:stop(Pid, Reason, Timeout) end, do_for_proc(Process, Fun). %%----------------------------------------------------------------- %% Map different specifications of a process to either Pid or %% {Name,Node}. Execute the given Fun with the process as only %% argument. %% ----------------------------------------------------------------- %% Local or remote by pid do_for_proc(Pid, Fun) when is_pid(Pid) -> Fun(Pid); %% Local by name do_for_proc(Name, Fun) when is_atom(Name) -> case whereis(Name) of Pid when is_pid(Pid) -> Fun(Pid); undefined -> exit(noproc) end; %% Global by name do_for_proc(Process, Fun) when ((tuple_size(Process) == 2 andalso element(1, Process) == global) orelse (tuple_size(Process) == 3 andalso element(1, Process) == via)) -> case where(Process) of Pid when is_pid(Pid) -> Node = node(Pid), try Fun(Pid) catch exit:{nodedown, Node} -> %% A nodedown not yet detected by global, %% pretend that it was. exit(noproc) end; undefined -> exit(noproc) end; %% Remote by name do_for_proc({Name, Node} = Process, Fun) when is_atom(Node) -> case partisan:node() of Node -> %% Local by name in disguise do_for_proc(Name, Fun); nonode@nohost -> exit({nodedown, Node}); _ -> Fun(Process) end. %%%----------------------------------------------------------------- %%% Misc. functions. %%%----------------------------------------------------------------- where({global, Name}) -> global:whereis_name(Name); where({via, Module, Name}) -> Module:whereis_name(Name); where({local, Name}) -> whereis(Name). register_name({local, Name} = LN) -> try register(Name, self()) of true -> true catch error:_ -> {false, where(LN)} end; register_name({global, Name} = GN) -> case global:register_name(Name, self()) of yes -> true; no -> {false, where(GN)} end; register_name({via, Module, Name} = GN) -> case Module:register_name(Name, self()) of yes -> true; no -> {false, where(GN)} end. name({local,Name}) -> Name; name({global,Name}) -> Name; name({via,_, Name}) -> Name; name(Pid) when is_pid(Pid) -> Pid. unregister_name({local,Name}) -> try unregister(Name) of _ -> ok catch _:_ -> ok end; unregister_name({global,Name}) -> _ = global:unregister_name(Name), ok; unregister_name({via, Mod, Name}) -> _ = Mod:unregister_name(Name), ok; unregister_name(Pid) when is_pid(Pid) -> ok. get_proc_name(Pid) when is_pid(Pid) -> Pid; get_proc_name({local, Name}) -> case process_info(self(), registered_name) of {registered_name, Name} -> Name; {registered_name, _Name} -> exit(process_not_registered); [] -> exit(process_not_registered) end; get_proc_name({global, Name}) -> case global:whereis_name(Name) of undefined -> exit(process_not_registered_globally); Pid when Pid =:= self() -> Name; _Pid -> exit(process_not_registered_globally) end; get_proc_name({via, Mod, Name}) -> case Mod:whereis_name(Name) of undefined -> exit({process_not_registered_via, Mod}); Pid when Pid =:= self() -> Name; _Pid -> exit({process_not_registered_via, Mod}) end. get_parent() -> case get('$ancestors') of [Parent | _] when is_pid(Parent) -> Parent; [Parent | _] when is_atom(Parent) -> name_to_pid(Parent); _ -> exit(process_was_not_started_by_proc_lib) end. name_to_pid(Name) -> case whereis(Name) of undefined -> case global:whereis_name(Name) of undefined -> exit(could_not_find_registered_name); Pid -> Pid end; Pid -> Pid end. timeout(Options) -> case lists:keyfind(timeout, 1, Options) of {_,Time} -> Time; false -> infinity end. spawn_opts(Options) -> case lists:keyfind(spawn_opt, 1, Options) of {_,Opts} -> Opts; false -> [] end. hibernate_after(Options) -> case lists:keyfind(hibernate_after, 1, Options) of {_,HibernateAfterTimeout} -> HibernateAfterTimeout; false -> infinity end. debug_options(Name, Opts) -> case lists:keyfind(debug, 1, Opts) of {_,Options} -> try sys:debug_options(Options) catch _:_ -> error_logger:format( "~tp: ignoring erroneous debug options - ~tp~n", [Name,Options]), [] end; false -> [] end. format_status_header(TagLine, Pid) when is_pid(Pid) -> lists:concat([TagLine, " ", pid_to_list(Pid)]); format_status_header(TagLine, RegName) when is_atom(RegName) -> lists:concat([TagLine, " ", RegName]); format_status_header(TagLine, Name) -> {TagLine, Name}. %% ============================================================================= %% PARTISAN %% ============================================================================= %% @private do_send_request(Process, Label, Request) when is_pid(Process) orelse is_atom(Process) -> Mref = partisan:monitor(process, Process), Message = {Label, {partisan:self(), Mref}, Request}, Node = partisan:node(), forward_message(Node, Process, Message), Mref; do_send_request({ServerRef, Node} = Process, Label, Request) -> Mref = partisan:monitor(process, Process), Message = {Label, {partisan:self(), Mref}, Request}, forward_message(Node, ServerRef, Message), Mref; do_send_request(RemoteRef, Label, Request) -> Mref = partisan:monitor(process, RemoteRef), Message = {Label, {partisan:self(), Mref}, Request}, ?LOG_DEBUG(#{ description => "Sending message", remote_ref => RemoteRef, message => Message }), partisan:forward_message(RemoteRef, Message, #{channel => get_channel()}), Mref. %% @private forward_message(Node, ServerRef, Message) -> ?LOG_DEBUG(#{ description => "Sending message", peer_node => Node, process => ServerRef, message => Message }), partisan:forward_message( Node, ServerRef, Message, #{channel => get_channel()} ). %% @doc Returns channel from the process dictionary get_channel() -> erlang:get(partisan_channel). set_channel(Options) -> _ = case lists:keyfind(channel, 1, Options) of {_, Channel} -> erlang:put(partisan_channel, Channel); false -> erlang:put(partisan_channel, partisan:default_channel()) end, ok.