diff options
Diffstat (limited to 'lib/kernel/src/global.erl')
| -rw-r--r-- | lib/kernel/src/global.erl | 2244 |
1 files changed, 2244 insertions, 0 deletions
diff --git a/lib/kernel/src/global.erl b/lib/kernel/src/global.erl new file mode 100644 index 0000000000..cc0402da73 --- /dev/null +++ b/lib/kernel/src/global.erl @@ -0,0 +1,2244 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1996-2009. All Rights Reserved. +%% +%% The contents of this file are subject to the Erlang Public License, +%% Version 1.1, (the "License"); you may not use this file except in +%% compliance with the License. You should have received a copy of the +%% Erlang Public License along with this software. If not, it can be +%% retrieved online at http://www.erlang.org/. +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and limitations +%% under the License. +%% +%% %CopyrightEnd% +%% +-module(global). +-behaviour(gen_server). + +%% Global provides global registration of process names. The names are +%% dynamically kept up to date with the entire network. Global can +%% operate in two modes: in a fully connected network, or in a +%% non-fully connected network. In the latter case, the name +%% registration mechanism won't work. +%% As a separate service Global also provides global locks. + +%% External exports +-export([start/0, start_link/0, stop/0, sync/0, sync/1, + safe_whereis_name/1, whereis_name/1, register_name/2, + register_name/3, register_name_external/2, register_name_external/3, + unregister_name_external/1,re_register_name/2, re_register_name/3, + unregister_name/1, registered_names/0, send/2, node_disconnected/1, + set_lock/1, set_lock/2, set_lock/3, + del_lock/1, del_lock/2, + trans/2, trans/3, trans/4, + random_exit_name/3, random_notify_name/3, notify_all_name/3]). + +%% Internal exports +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3, resolve_it/4]). + +-export([info/0]). + +-include_lib("stdlib/include/ms_transform.hrl"). + +%% Set this variable to 'allow' to allow several names of a process. +%% This is for backward compatibility only; the functionality is broken. +-define(WARN_DUPLICATED_NAME, global_multi_name_action). + +%% Undocumented Kernel variable. Set this to 0 (zero) to get the old +%% behaviour. +-define(N_CONNECT_RETRIES, global_connect_retries). +-define(DEFAULT_N_CONNECT_RETRIES, 5). + +%%% In certain places in the server, calling io:format hangs everything, +%%% so we'd better use erlang:display/1. +%%% my_tracer is used in testsuites +-define(trace(_), ok). + +%-define(trace(T), (catch my_tracer ! {node(), {line,?LINE}, T})). + +%-define(trace(T), erlang:display({format, node(), cs(), T})). +%cs() -> +% {_Big, Small, Tiny} = now(), +% (Small rem 100) * 100 + (Tiny div 10000). + +%% These are the protocol versions: +%% Vsn 1 is the original protocol. +%% Vsn 2 is enhanced with code to take care of registration of names from +%% non erlang nodes, e.g. C-nodes. +%% Vsn 3 is enhanced with a tag in the synch messages to distinguish +%% different synch sessions from each other, see OTP-2766. +%% Vsn 4 uses a single, permanent, locker process, but works like vsn 3 +%% when communicating with vsn 3 nodes. (-R10B) +%% Vsn 5 uses an ordered list of self() and HisTheLocker when locking +%% nodes in the own partition. (R11B-) + +%% Current version of global does not support vsn 4 or earlier. + +-define(vsn, 5). + +%%----------------------------------------------------------------- +%% connect_all = boolean() - true if we are supposed to set up a +%% fully connected net +%% known = [Node] - all nodes known to us +%% synced = [Node] - all nodes that have the same names as us +%% resolvers = [{Node, MyTag, Resolver}] - +%% the tag separating different synch sessions, +%% and the pid of the name resolver process +%% syncers = [pid()] - all current syncers processes +%% node_name = atom() - our node name (can change if distribution +%% is started/stopped dynamically) +%% +%% In addition to these, we keep info about messages arrived in +%% the process dictionary: +%% {pre_connect, Node} = {Vsn, InitMsg} - init_connect msgs that +%% arrived before nodeup +%% {wait_lock, Node} = {exchange, NameList, _NamelistExt} | lock_is_set +%% - see comment below (handle_cast) +%% {save_ops, Node} = {resolved, HisKnown, NamesExt, Res} | [operation()] +%% - save the ops between exchange and resolved +%% {prot_vsn, Node} = Vsn - the exchange protocol version (not used now) +%% {sync_tag_my, Node} = My tag, used at synchronization with Node +%% {sync_tag_his, Node} = The Node's tag, used at synchronization +%% {lock_id, Node} = The resource locking the partitions +%%----------------------------------------------------------------- +-record(state, {connect_all :: boolean(), + known = [] :: [node()], + synced = [] :: [node()], + resolvers = [], + syncers = [] :: [pid()], + node_name = node() :: node(), + the_locker, the_deleter, the_registrar, trace, + global_lock_down = false + }). + +%%% There are also ETS tables used for bookkeeping of locks and names +%%% (the first position is the key): +%%% +%%% global_locks (set): {ResourceId, LockRequesterId, [{Pid,RPid,ref()]} +%%% Pid is locking ResourceId, ref() is the monitor ref. +%%% RPid =/= Pid if there is an extra process calling erlang:monitor(). +%%% global_names (set): {Name, Pid, Method, RPid, ref()} +%%% Registered names. ref() is the monitor ref. +%%% RPid =/= Pid if there is an extra process calling erlang:monitor(). +%%% global_names_ext (set): {Name, Pid, RegNode} +%%% External registered names (C-nodes). +%%% (The RPid:s can be removed when/if erlang:monitor() returns before +%%% trying to connect to the other node.) +%%% +%%% Helper tables: +%%% global_pid_names (bag): {Pid, Name} | {ref(), Name} +%%% Name(s) registered for Pid. +%%% There is one {Pid, Name} and one {ref(), Name} for every Pid. +%%% ref() is the same ref() as in global_names. +%%% global_pid_ids (bag): {Pid, ResourceId} | {ref(), ResourceId} +%%% Resources locked by Pid. +%%% ref() is the same ref() as in global_locks. +%%% +%%% global_pid_names is a 'bag' for backward compatibility. +%%% (Before vsn 5 more than one name could be registered for a process.) +%%% +%%% R11B-3 (OTP-6341): The list of pids in the table 'global_locks' +%%% was replaced by a list of {Pid, Ref}, where Ref is a monitor ref. +%%% It was necessary to use monitors to fix bugs regarding locks that +%%% were never removed. The signal {async_del_lock, ...} has been +%%% kept for backward compatibility. It can be removed later. +%%% +%%% R11B-4 (OTP-6428): Monitors are used for registered names. +%%% The signal {delete_name, ...} has been kept for backward compatibility. +%%% It can be removed later as can the deleter process. +%%% An extra process calling erlang:monitor() is sometimes created. +%%% The new_nodes messages has been augmented with the global lock id. + +start() -> + gen_server:start({local, global_name_server}, ?MODULE, [], []). + +start_link() -> + gen_server:start_link({local, global_name_server}, ?MODULE, [], []). + +stop() -> + gen_server:call(global_name_server, stop, infinity). + +-spec sync() -> 'ok' | {'error', term()}. +sync() -> + case check_sync_nodes() of + {error, _} = Error -> + Error; + SyncNodes -> + gen_server:call(global_name_server, {sync, SyncNodes}, infinity) + end. + +-spec sync([node()]) -> 'ok' | {'error', term()}. +sync(Nodes) -> + case check_sync_nodes(Nodes) of + {error, _} = Error -> + Error; + SyncNodes -> + gen_server:call(global_name_server, {sync, SyncNodes}, infinity) + end. + +-spec send(term(), term()) -> pid(). +send(Name, Msg) -> + case whereis_name(Name) of + Pid when is_pid(Pid) -> + Pid ! Msg, + Pid; + undefined -> + exit({badarg, {Name, Msg}}) + end. + +%% See OTP-3737. +-spec whereis_name(term()) -> pid() | 'undefined'. +whereis_name(Name) -> + where(Name). + +-spec safe_whereis_name(term()) -> pid() | 'undefined'. +safe_whereis_name(Name) -> + gen_server:call(global_name_server, {whereis, Name}, infinity). + +node_disconnected(Node) -> + global_name_server ! {nodedown, Node}. + +%%----------------------------------------------------------------- +%% Method = function(Name, Pid1, Pid2) -> Pid | Pid2 | none +%% Method is called if a name conflict is detected when two nodes +%% are connecting to each other. It is supposed to return one of +%% the Pids or 'none'. If a pid is returned, that pid is +%% registered as Name on all nodes. If 'none' is returned, the +%% Name is unregistered on all nodes. If anything else is returned, +%% the Name is unregistered as well. +%% Method is called once at one of the nodes where the processes reside +%% only. If different Methods are used for the same name, it is +%% undefined which one of them is used. +%% Method blocks the name registration, but does not affect global locking. +%%----------------------------------------------------------------- +-spec register_name(term(), pid()) -> 'yes' | 'no'. +register_name(Name, Pid) when is_pid(Pid) -> + register_name(Name, Pid, fun random_exit_name/3). + +-type method() :: fun((term(), pid(), pid()) -> pid() | 'none'). + +-spec register_name(term(), pid(), method()) -> 'yes' | 'no'. +register_name(Name, Pid, Method) when is_pid(Pid) -> + Fun = fun(Nodes) -> + case (where(Name) =:= undefined) andalso check_dupname(Name, Pid) of + true -> + gen_server:multi_call(Nodes, + global_name_server, + {register, Name, Pid, Method}), + yes; + _ -> + no + end + end, + ?trace({register_name, self(), Name, Pid, Method}), + gen_server:call(global_name_server, {registrar, Fun}, infinity). + +check_dupname(Name, Pid) -> + case ets:lookup(global_pid_names, Pid) of + [] -> + true; + PidNames -> + case application:get_env(kernel, ?WARN_DUPLICATED_NAME) of + {ok, allow} -> + true; + _ -> + S = "global: ~w registered under several names: ~w\n", + Names = [Name | [Name1 || {_Pid, Name1} <- PidNames]], + error_logger:error_msg(S, [Pid, Names]), + false + end + end. + +-spec unregister_name(term()) -> _. +unregister_name(Name) -> + case where(Name) of + undefined -> + ok; + _ -> + Fun = fun(Nodes) -> + gen_server:multi_call(Nodes, + global_name_server, + {unregister, Name}), + ok + end, + ?trace({unregister_name, self(), Name}), + gen_server:call(global_name_server, {registrar, Fun}, infinity) + end. + +-spec re_register_name(term(), pid()) -> _. +re_register_name(Name, Pid) when is_pid(Pid) -> + re_register_name(Name, Pid, fun random_exit_name/3). + +-spec re_register_name(term(), pid(), method()) -> _. +re_register_name(Name, Pid, Method) when is_pid(Pid) -> + Fun = fun(Nodes) -> + gen_server:multi_call(Nodes, + global_name_server, + {register, Name, Pid, Method}), + yes + end, + ?trace({re_register_name, self(), Name, Pid, Method}), + gen_server:call(global_name_server, {registrar, Fun}, infinity). + +-spec registered_names() -> [term()]. +registered_names() -> + MS = ets:fun2ms(fun({Name,_Pid,_M,_RP,_R}) -> Name end), + ets:select(global_names, MS). + +%%----------------------------------------------------------------- +%% The external node (e.g. a C-node) registers the name on an Erlang +%% node which links to the process (an Erlang node has to be used +%% since there is no global_name_server on the C-node). If the Erlang +%% node dies the name is to be unregistered on all nodes. Normally +%% node(Pid) is compared to the node that died, but that does not work +%% for external nodes (the process does not run on the Erlang node +%% that died). Therefore a table of all names registered by external +%% nodes is kept up-to-date on all nodes. +%% +%% Note: if the Erlang node dies an EXIT signal is also sent to the +%% C-node due to the link between the global_name_server and the +%% registered process. [This is why the link has been kept despite +%% the fact that monitors do the job now.] +%%----------------------------------------------------------------- +register_name_external(Name, Pid) when is_pid(Pid) -> + register_name_external(Name, Pid, fun random_exit_name/3). + +register_name_external(Name, Pid, Method) when is_pid(Pid) -> + Fun = fun(Nodes) -> + case where(Name) of + undefined -> + gen_server:multi_call(Nodes, + global_name_server, + {register_ext, Name, Pid, + Method, node()}), + yes; + _Pid -> no + end + end, + ?trace({register_name_external, self(), Name, Pid, Method}), + gen_server:call(global_name_server, {registrar, Fun}, infinity). + +unregister_name_external(Name) -> + unregister_name(Name). + +-type id() :: {term(), term()}. + +-spec set_lock(id()) -> boolean(). +set_lock(Id) -> + set_lock(Id, [node() | nodes()], infinity, 1). + +-type retries() :: non_neg_integer() | 'infinity'. + +-spec set_lock(id(), [node()]) -> boolean(). +set_lock(Id, Nodes) -> + set_lock(Id, Nodes, infinity, 1). + +-spec set_lock(id(), [node()], retries()) -> boolean(). +set_lock(Id, Nodes, Retries) when is_integer(Retries), Retries >= 0 -> + set_lock(Id, Nodes, Retries, 1); +set_lock(Id, Nodes, infinity) -> + set_lock(Id, Nodes, infinity, 1). + +set_lock({_ResourceId, _LockRequesterId}, [], _Retries, _Times) -> + true; +set_lock({_ResourceId, _LockRequesterId} = Id, Nodes, Retries, Times) -> + ?trace({set_lock,{me,self()},Id,{nodes,Nodes}, + {retries,Retries}, {times,Times}}), + case set_lock_on_nodes(Id, Nodes) of + true -> + ?trace({set_lock_true, Id}), + true; + false=Reply when Retries =:= 0 -> + Reply; + false -> + random_sleep(Times), + set_lock(Id, Nodes, dec(Retries), Times+1) + end. + +-spec del_lock(id()) -> 'true'. +del_lock(Id) -> + del_lock(Id, [node() | nodes()]). + +-spec del_lock(id(), [node()]) -> 'true'. +del_lock({_ResourceId, _LockRequesterId} = Id, Nodes) -> + ?trace({del_lock, {me,self()}, Id, {nodes,Nodes}}), + gen_server:multi_call(Nodes, global_name_server, {del_lock, Id}), + true. + +-type trans_fun() :: function() | {module(), atom()}. + +-spec trans(id(), trans_fun()) -> term(). +trans(Id, Fun) -> trans(Id, Fun, [node() | nodes()], infinity). + +-spec trans(id(), trans_fun(), [node()]) -> term(). +trans(Id, Fun, Nodes) -> trans(Id, Fun, Nodes, infinity). + +-spec trans(id(), trans_fun(), [node()], retries()) -> term(). +trans(Id, Fun, Nodes, Retries) -> + case set_lock(Id, Nodes, Retries) of + true -> + try + Fun() + after + del_lock(Id, Nodes) + end; + false -> + aborted + end. + +info() -> + gen_server:call(global_name_server, info, infinity). + +%%%----------------------------------------------------------------- +%%% Call-back functions from gen_server +%%%----------------------------------------------------------------- +init([]) -> + process_flag(trap_exit, true), + _ = ets:new(global_locks, [set, named_table, protected]), + _ = ets:new(global_names, [set, named_table, protected]), + _ = ets:new(global_names_ext, [set, named_table, protected]), + + _ = ets:new(global_pid_names, [bag, named_table, protected]), + _ = ets:new(global_pid_ids, [bag, named_table, protected]), + + %% This is for troubleshooting only. + DoTrace = os:getenv("GLOBAL_HIGH_LEVEL_TRACE") =:= "TRUE", + T0 = case DoTrace of + true -> + send_high_level_trace(), + []; + false -> + no_trace + end, + + S = #state{the_locker = start_the_locker(DoTrace), + trace = T0, + the_deleter = start_the_deleter(self()), + the_registrar = start_the_registrar()}, + S1 = trace_message(S, {init, node()}, []), + + case init:get_argument(connect_all) of + {ok, [["false"]]} -> + {ok, S1#state{connect_all = false}}; + _ -> + {ok, S1#state{connect_all = true}} + end. + +%%----------------------------------------------------------------- +%% Connection algorithm +%% ==================== +%% This algorithm solves the problem with partitioned nets as well. +%% +%% The main idea in the algorithm is that when two nodes connect, they +%% try to set a lock in their own partition (i.e. all nodes already +%% known to them; partitions are not necessarily disjoint). When the +%% lock is set in each partition, these two nodes send each other a +%% list with all registered names in resp partition (*). If no conflict +%% is found, the name tables are just updated. If a conflict is found, +%% a resolve function is called once for each conflict. The result of +%% the resolving is sent to the other node. When the names are +%% exchanged, all other nodes in each partition are informed of the +%% other nodes, and they ping each other to form a fully connected +%% net. +%% +%% A few remarks: +%% +%% (*) When this information is being exchanged, no one is allowed to +%% change the global register table. All calls to register etc are +%% protected by a lock. If a registered process dies during this +%% phase the name is unregistered on the local node immediately, +%% but the unregistration on other nodes will take place when the +%% deleter manages to acquire the lock. This is necessary to +%% prevent names from spreading to nodes where they cannot be +%% deleted. +%% +%% - It is assumed that nodeups and nodedowns arrive in an orderly +%% fashion: for every node, nodeup is followed by nodedown, and vice +%% versa. "Double" nodeups and nodedowns must never occur. It is +%% the responsibility of net_kernel to assure this. +%% +%% - There is always a delay between the termination of a registered +%% process and the removal of the name from Global's tables. This +%% delay can sometimes be quite substantial. Global guarantees that +%% the name will eventually be removed, but there is no +%% synchronization between nodes; the name can be removed from some +%% node(s) long before it is removed from other nodes. Using +%% safe_whereis_name is no cure. +%% +%% - Global cannot handle problems with the distribution very well. +%% Depending on the value of the kernel variable 'net_ticktime' long +%% delays may occur. This does not affect the handling of locks but +%% will block name registration. +%% +%% - Old synch session messages may linger on in the message queue of +%% global_name_server after the sending node has died. The tags of +%% such messages do not match the current tag (if there is one), +%% which makes it possible to discard those messages and cancel the +%% corresponding lock. +%% +%% Suppose nodes A and B connect, and C is connected to A. +%% Here's the algorithm's flow: +%% +%% Node A +%% ------ +%% << {nodeup, B} +%% TheLocker ! {nodeup, ..., Node, ...} (there is one locker per node) +%% B ! {init_connect, ..., {..., TheLockerAtA, ...}} +%% << {init_connect, TheLockerAtB} +%% [The lockers try to set the lock] +%% << {lock_is_set, B, ...} +%% [Now, lock is set in both partitions] +%% B ! {exchange, A, Names, ...} +%% << {exchange, B, Names, ...} +%% [solve conflict] +%% B ! {resolved, A, ResolvedA, KnownAtA, ...} +%% << {resolved, B, ResolvedB, KnownAtB, ...} +%% C ! {new_nodes, ResolvedAandB, [B]} +%% +%% Node C +%% ------ +%% << {new_nodes, ResolvedOps, NewNodes} +%% [insert Ops] +%% ping(NewNodes) +%% << {nodeup, B} +%% <ignore this one> +%% +%% Several things can disturb this picture. +%% +%% First, the init_connect message may arrive _before_ the nodeup +%% message due to delay in net_kernel. We handle this by keeping track +%% of these messages in the pre_connect variable in our state. +%% +%% Of course we must handle that some node goes down during the +%% connection. +%% +%%----------------------------------------------------------------- +%% Messages in the protocol +%% ======================== +%% 1. Between global_name_servers on connecting nodes +%% {init_connect, Vsn, Node, InitMsg} +%% InitMsg = {locker, _Unused, HisKnown, HisTheLocker} +%% {exchange, Node, ListOfNames, _ListOfNamesExt, Tag} +%% {resolved, Node, HisOps, HisKnown, _Unused, ListOfNamesExt, Tag} +%% HisKnown = list of known nodes in Node's partition +%% 2. Between lockers on connecting nodes +%% {his_locker, Pid} (from our global) +%% {lock, Bool} loop until both lockers have lock = true, +%% then send to global_name_server {lock_is_set, Node, Tag} +%% 3. Connecting node's global_name_server informs other nodes in the same +%% partition about hitherto unknown nodes in the other partition +%% {new_nodes, Node, Ops, ListOfNamesExt, NewNodes, ExtraInfo} +%% 4. Between global_name_server and resolver +%% {resolve, NameList, Node} to resolver +%% {exchange_ops, Node, Tag, Ops, Resolved} from resolver +%% 5. sync protocol, between global_name_servers in different partitions +%% {in_sync, Node, IsKnown} +%% sent by each node to all new nodes (Node becomes known to them) +%%----------------------------------------------------------------- + +handle_call({whereis, Name}, From, S) -> + do_whereis(Name, From), + {noreply, S}; + +handle_call({registrar, Fun}, From, S) -> + S#state.the_registrar ! {trans_all_known, Fun, From}, + {noreply, S}; + +%% The pattern {register,'_','_','_'} is traced by the inviso +%% application. Do not change. +handle_call({register, Name, Pid, Method}, {FromPid, _Tag}, S0) -> + S = ins_name(Name, Pid, Method, FromPid, [], S0), + {reply, yes, S}; + +handle_call({unregister, Name}, _From, S0) -> + S = delete_global_name2(Name, S0), + {reply, ok, S}; + +handle_call({register_ext, Name, Pid, Method, RegNode}, {FromPid,_Tag}, S0) -> + S = ins_name_ext(Name, Pid, Method, RegNode, FromPid, [], S0), + {reply, yes, S}; + +handle_call({set_lock, Lock}, {Pid, _Tag}, S0) -> + {Reply, S} = handle_set_lock(Lock, Pid, S0), + {reply, Reply, S}; + +handle_call({del_lock, Lock}, {Pid, _Tag}, S0) -> + S = handle_del_lock(Lock, Pid, S0), + {reply, true, S}; + +handle_call(get_known, _From, S) -> + {reply, S#state.known, S}; + +handle_call(get_synced, _From, S) -> + {reply, S#state.synced, S}; + +handle_call({sync, Nodes}, From, S) -> + %% If we have several global groups, this won't work, since we will + %% do start_sync on a nonempty list of nodes even if the system + %% is quiet. + Pid = start_sync(lists:delete(node(), Nodes) -- S#state.synced, From), + {noreply, S#state{syncers = [Pid | S#state.syncers]}}; + +handle_call(get_protocol_version, _From, S) -> + {reply, ?vsn, S}; + +handle_call(get_names_ext, _From, S) -> + {reply, get_names_ext(), S}; + +handle_call(info, _From, S) -> + {reply, S, S}; + +%% "High level trace". For troubleshooting only. +handle_call(high_level_trace_start, _From, S) -> + S#state.the_locker ! {do_trace, true}, + send_high_level_trace(), + {reply, ok, trace_message(S#state{trace = []}, {init, node()}, [])}; +handle_call(high_level_trace_stop, _From, S) -> + #state{the_locker = TheLocker, trace = Trace} = S, + TheLocker ! {do_trace, false}, + wait_high_level_trace(), + {reply, Trace, S#state{trace = no_trace}}; +handle_call(high_level_trace_get, _From, #state{trace = Trace}=S) -> + {reply, Trace, S#state{trace = []}}; + +handle_call(stop, _From, S) -> + {stop, normal, stopped, S}; + +handle_call(Request, From, S) -> + error_logger:warning_msg("The global_name_server " + "received an unexpected message:\n" + "handle_call(~p, ~p, _)\n", + [Request, From]), + {noreply, S}. + +%%======================================================================== +%% init_connect +%% +%%======================================================================== +handle_cast({init_connect, Vsn, Node, InitMsg}, S) -> + %% Sent from global_name_server at Node. + ?trace({'####', init_connect, {vsn, Vsn}, {node,Node},{initmsg,InitMsg}}), + case Vsn of + %% It is always the responsibility of newer versions to understand + %% older versions of the protocol. + {HisVsn, HisTag} when HisVsn > ?vsn -> + init_connect(?vsn, Node, InitMsg, HisTag, S#state.resolvers, S); + {HisVsn, HisTag} -> + init_connect(HisVsn, Node, InitMsg, HisTag, S#state.resolvers, S); + %% To be future compatible + Tuple when is_tuple(Tuple) -> + List = tuple_to_list(Tuple), + [_HisVsn, HisTag | _] = List, + %% use own version handling if his is newer. + init_connect(?vsn, Node, InitMsg, HisTag, S#state.resolvers, S); + _ -> + Txt = io_lib:format("Illegal global protocol version ~p Node: ~p\n", + [Vsn, Node]), + error_logger:info_report(lists:flatten(Txt)) + end, + {noreply, S}; + +%%======================================================================= +%% lock_is_set +%% +%% Ok, the lock is now set on both partitions. Send our names to other node. +%%======================================================================= +handle_cast({lock_is_set, Node, MyTag, LockId}, S) -> + %% Sent from the_locker at node(). + ?trace({'####', lock_is_set , {node,Node}}), + case get({sync_tag_my, Node}) of + MyTag -> + lock_is_set(Node, S#state.resolvers, LockId), + {noreply, S}; + _ -> %% Illegal tag, delete the old sync session. + NewS = cancel_locker(Node, S, MyTag), + {noreply, NewS} + end; + +%%======================================================================== +%% exchange +%% +%% Here the names are checked to detect name clashes. +%%======================================================================== +handle_cast({exchange, Node, NameList, _NameExtList, MyTag}, S) -> + %% Sent from global_name_server at Node. + case get({sync_tag_my, Node}) of + MyTag -> + exchange(Node, NameList, S#state.resolvers), + {noreply, S}; + _ -> %% Illegal tag, delete the old sync session. + NewS = cancel_locker(Node, S, MyTag), + {noreply, NewS} + end; + +%% {exchange_ops, ...} is sent by the resolver process (which then +%% dies). It could happen that {resolved, ...} has already arrived +%% from the other node. In that case we can go ahead and run the +%% resolve operations. Otherwise we have to save the operations and +%% wait for {resolve, ...}. This is very much like {lock_is_set, ...} +%% and {exchange, ...}. +handle_cast({exchange_ops, Node, MyTag, Ops, Resolved}, S0) -> + %% Sent from the resolver for Node at node(). + ?trace({exchange_ops, {node,Node}, {ops,Ops},{resolved,Resolved}, + {mytag,MyTag}}), + S = trace_message(S0, {exit_resolver, Node}, [MyTag]), + case get({sync_tag_my, Node}) of + MyTag -> + Known = S#state.known, + gen_server:cast({global_name_server, Node}, + {resolved, node(), Resolved, Known, + Known,get_names_ext(),get({sync_tag_his,Node})}), + case get({save_ops, Node}) of + {resolved, HisKnown, Names_ext, HisResolved} -> + put({save_ops, Node}, Ops), + NewS = resolved(Node, HisResolved, HisKnown, Names_ext,S), + {noreply, NewS}; + undefined -> + put({save_ops, Node}, Ops), + {noreply, S} + end; + _ -> %% Illegal tag, delete the old sync session. + NewS = cancel_locker(Node, S, MyTag), + {noreply, NewS} + end; + +%%======================================================================== +%% resolved +%% +%% Here the name clashes are resolved. +%%======================================================================== +handle_cast({resolved, Node, HisResolved, HisKnown, _HisKnown_v2, + Names_ext, MyTag}, S) -> + %% Sent from global_name_server at Node. + ?trace({'####', resolved, {his_resolved,HisResolved}, {node,Node}}), + case get({sync_tag_my, Node}) of + MyTag -> + %% See the comment at handle_case({exchange_ops, ...}). + case get({save_ops, Node}) of + Ops when is_list(Ops) -> + NewS = resolved(Node, HisResolved, HisKnown, Names_ext, S), + {noreply, NewS}; + undefined -> + Resolved = {resolved, HisKnown, Names_ext, HisResolved}, + put({save_ops, Node}, Resolved), + {noreply, S} + end; + _ -> %% Illegal tag, delete the old sync session. + NewS = cancel_locker(Node, S, MyTag), + {noreply, NewS} + end; + +%%======================================================================== +%% new_nodes +%% +%% We get to know the other node's known nodes. +%%======================================================================== +handle_cast({new_nodes, Node, Ops, Names_ext, Nodes, ExtraInfo}, S) -> + %% Sent from global_name_server at Node. + ?trace({new_nodes, {node,Node},{ops,Ops},{nodes,Nodes},{x,ExtraInfo}}), + NewS = new_nodes(Ops, Node, Names_ext, Nodes, ExtraInfo, S), + {noreply, NewS}; + +%%======================================================================== +%% in_sync +%% +%% We are in sync with this node (from the other node's known world). +%%======================================================================== +handle_cast({in_sync, Node, _IsKnown}, S) -> + %% Sent from global_name_server at Node (in the other partition). + ?trace({'####', in_sync, {Node, _IsKnown}}), + lists:foreach(fun(Pid) -> Pid ! {synced, [Node]} end, S#state.syncers), + NewS = cancel_locker(Node, S, get({sync_tag_my, Node})), + reset_node_state(Node), + NSynced = case lists:member(Node, Synced = NewS#state.synced) of + true -> Synced; + false -> [Node | Synced] + end, + {noreply, NewS#state{synced = NSynced}}; + +%% Called when Pid on other node crashed +handle_cast({async_del_name, _Name, _Pid}, S) -> + %% Sent from the_deleter at some node in the partition but node(). + %% The DOWN message deletes the name. + {noreply, S}; + +handle_cast({async_del_lock, _ResourceId, _Pid}, S) -> + %% Sent from global_name_server at some node in the partition but node(). + %% The DOWN message deletes the lock. + {noreply, S}; + +handle_cast(Request, S) -> + error_logger:warning_msg("The global_name_server " + "received an unexpected message:\n" + "handle_cast(~p, _)\n", [Request]), + {noreply, S}. + +handle_info({'EXIT', Deleter, _Reason}=Exit, #state{the_deleter=Deleter}=S) -> + {stop, {deleter_died,Exit}, S#state{the_deleter=undefined}}; +handle_info({'EXIT', Locker, _Reason}=Exit, #state{the_locker=Locker}=S) -> + {stop, {locker_died,Exit}, S#state{the_locker=undefined}}; +handle_info({'EXIT', Registrar, _}=Exit, #state{the_registrar=Registrar}=S) -> + {stop, {registrar_died,Exit}, S#state{the_registrar=undefined}}; +handle_info({'EXIT', Pid, _Reason}, S) when is_pid(Pid) -> + ?trace({global_EXIT,_Reason,Pid}), + %% The process that died was a synch process started by start_sync + %% or a registered process running on an external node (C-node). + %% Links to external names are ignored here (there are also DOWN + %% signals). + Syncers = lists:delete(Pid, S#state.syncers), + {noreply, S#state{syncers = Syncers}}; + +handle_info({nodedown, Node}, S) when Node =:= S#state.node_name -> + %% Somebody stopped the distribution dynamically - change + %% references to old node name (Node) to new node name ('nonode@nohost') + {noreply, change_our_node_name(node(), S)}; + +handle_info({nodedown, Node}, S0) -> + ?trace({'####', nodedown, {node,Node}}), + S1 = trace_message(S0, {nodedown, Node}, []), + S = handle_nodedown(Node, S1), + {noreply, S}; + +handle_info({extra_nodedown, Node}, S0) -> + ?trace({'####', extra_nodedown, {node,Node}}), + S1 = trace_message(S0, {extra_nodedown, Node}, []), + S = handle_nodedown(Node, S1), + {noreply, S}; + +handle_info({nodeup, Node}, S) when Node =:= node() -> + ?trace({'####', local_nodeup, {node, Node}}), + %% Somebody started the distribution dynamically - change + %% references to old node name ('nonode@nohost') to Node. + {noreply, change_our_node_name(Node, S)}; + +handle_info({nodeup, _Node}, S) when not S#state.connect_all -> + {noreply, S}; + +handle_info({nodeup, Node}, S0) when S0#state.connect_all -> + IsKnown = lists:member(Node, S0#state.known) or + %% This one is only for double nodeups (shouldn't occur!) + lists:keymember(Node, 1, S0#state.resolvers), + ?trace({'####', nodeup, {node,Node}, {isknown,IsKnown}}), + S1 = trace_message(S0, {nodeup, Node}, []), + case IsKnown of + true -> + {noreply, S1}; + false -> + resend_pre_connect(Node), + + %% now() is used as a tag to separate different synch sessions + %% from each others. Global could be confused at bursty nodeups + %% because it couldn't separate the messages between the different + %% synch sessions started by a nodeup. + MyTag = now(), + put({sync_tag_my, Node}, MyTag), + ?trace({sending_nodeup_to_locker, {node,Node},{mytag,MyTag}}), + S1#state.the_locker ! {nodeup, Node, MyTag}, + + %% In order to be compatible with unpatched R7 a locker + %% process was spawned. Vsn 5 is no longer compatible with + %% vsn 3 nodes, so the locker process is no longer needed. + %% The permanent locker takes its place. + NotAPid = no_longer_a_pid, + Locker = {locker, NotAPid, S1#state.known, S1#state.the_locker}, + InitC = {init_connect, {?vsn, MyTag}, node(), Locker}, + Rs = S1#state.resolvers, + ?trace({casting_init_connect, {node,Node},{initmessage,InitC}, + {resolvers,Rs}}), + gen_server:cast({global_name_server, Node}, InitC), + Resolver = start_resolver(Node, MyTag), + S = trace_message(S1, {new_resolver, Node}, [MyTag, Resolver]), + {noreply, S#state{resolvers = [{Node, MyTag, Resolver} | Rs]}} + end; + +handle_info({whereis, Name, From}, S) -> + do_whereis(Name, From), + {noreply, S}; + +handle_info(known, S) -> + io:format(">>>> ~p\n",[S#state.known]), + {noreply, S}; + +%% "High level trace". For troubleshooting only. +handle_info(high_level_trace, S) -> + case S of + #state{trace = [{Node, _Time, _M, Nodes, _X} | _]} -> + send_high_level_trace(), + CNode = node(), + CNodes = nodes(), + case {CNode, CNodes} of + {Node, Nodes} -> + {noreply, S}; + _ -> + {New, _, Old} = + sofs:symmetric_partition(sofs:set([CNode|CNodes]), + sofs:set([Node|Nodes])), + M = {nodes_changed, {sofs:to_external(New), + sofs:to_external(Old)}}, + {noreply, trace_message(S, M, [])} + end; + _ -> + {noreply, S} + end; +handle_info({trace_message, M}, S) -> + {noreply, trace_message(S, M, [])}; +handle_info({trace_message, M, X}, S) -> + {noreply, trace_message(S, M, X)}; + +handle_info({'DOWN', MonitorRef, process, _Pid, _Info}, S0) -> + S1 = delete_lock(MonitorRef, S0), + S = del_name(MonitorRef, S1), + {noreply, S}; + +handle_info(Message, S) -> + error_logger:warning_msg("The global_name_server " + "received an unexpected message:\n" + "handle_info(~p, _)\n", [Message]), + {noreply, S}. + + +%%======================================================================== +%%======================================================================== +%%=============================== Internal Functions ===================== +%%======================================================================== +%%======================================================================== + +-define(HIGH_LEVEL_TRACE_INTERVAL, 500). % ms + +wait_high_level_trace() -> + receive + high_level_trace -> + ok + after ?HIGH_LEVEL_TRACE_INTERVAL+1 -> + ok + end. + +send_high_level_trace() -> + erlang:send_after(?HIGH_LEVEL_TRACE_INTERVAL, self(), high_level_trace). + +-define(GLOBAL_RID, global). + +%% Similar to trans(Id, Fun), but always uses global's own lock +%% on all nodes known to global, making sure that no new nodes have +%% become known while we got the list of known nodes. +trans_all_known(Fun) -> + Id = {?GLOBAL_RID, self()}, + Nodes = set_lock_known(Id, 0), + try + Fun(Nodes) + after + delete_global_lock(Id, Nodes) + end. + +set_lock_known(Id, Times) -> + Known = get_known(), + Nodes = [node() | Known], + Boss = the_boss(Nodes), + %% Use the same convention (a boss) as lock_nodes_safely. Optimization. + case set_lock_on_nodes(Id, [Boss]) of + true -> + case lock_on_known_nodes(Id, Known, Nodes) of + true -> + Nodes; + false -> + del_lock(Id, [Boss]), + random_sleep(Times), + set_lock_known(Id, Times+1) + end; + false -> + random_sleep(Times), + set_lock_known(Id, Times+1) + end. + +lock_on_known_nodes(Id, Known, Nodes) -> + case set_lock_on_nodes(Id, Nodes) of + true -> + (get_known() -- Known) =:= []; + false -> + false + end. + +set_lock_on_nodes(_Id, []) -> + true; +set_lock_on_nodes(Id, Nodes) -> + case local_lock_check(Id, Nodes) of + true -> + Msg = {set_lock, Id}, + {Replies, _} = + gen_server:multi_call(Nodes, global_name_server, Msg), + ?trace({set_lock,{me,self()},Id,{nodes,Nodes},{replies,Replies}}), + check_replies(Replies, Id, Replies); + false=Reply -> + Reply + end. + +%% Probe lock on local node to see if one should go on trying other nodes. +local_lock_check(_Id, [_] = _Nodes) -> + true; +local_lock_check(Id, Nodes) -> + not lists:member(node(), Nodes) orelse (can_set_lock(Id) =/= false). + +check_replies([{_Node, true} | T], Id, Replies) -> + check_replies(T, Id, Replies); +check_replies([{_Node, false=Reply} | _T], _Id, [_]) -> + Reply; +check_replies([{_Node, false=Reply} | _T], Id, Replies) -> + TrueReplyNodes = [N || {N, true} <- Replies], + ?trace({check_replies, {true_reply_nodes, TrueReplyNodes}}), + gen_server:multi_call(TrueReplyNodes, global_name_server, {del_lock, Id}), + Reply; +check_replies([], _Id, _Replies) -> + true. + +%%======================================================================== +%% Another node wants to synchronize its registered names with us. +%% Both nodes must have a lock before they are allowed to continue. +%%======================================================================== +init_connect(Vsn, Node, InitMsg, HisTag, Resolvers, S) -> + %% It is always the responsibility of newer versions to understand + %% older versions of the protocol. + put({prot_vsn, Node}, Vsn), + put({sync_tag_his, Node}, HisTag), + case lists:keyfind(Node, 1, Resolvers) of + {Node, MyTag, _Resolver} -> + MyTag = get({sync_tag_my, Node}), % assertion + {locker, _NoLongerAPid, _HisKnown0, HisTheLocker} = InitMsg, + ?trace({init_connect,{histhelocker,HisTheLocker}}), + HisKnown = [], + S#state.the_locker ! {his_the_locker, HisTheLocker, + {Vsn,HisKnown}, S#state.known}; + false -> + ?trace({init_connect,{pre_connect,Node},{histag,HisTag}}), + put({pre_connect, Node}, {Vsn, InitMsg, HisTag}) + end. + +%%======================================================================== +%% In the simple case, we'll get lock_is_set before we get exchange, +%% but we may get exchange before we get lock_is_set from our locker. +%% If that's the case, we'll have to remember the exchange info, and +%% handle it when we get the lock_is_set. We do this by using the +%% process dictionary - when the lock_is_set msg is received, we store +%% this info. When exchange is received, we can check the dictionary +%% if the lock_is_set has been received. If not, we store info about +%% the exchange instead. In the lock_is_set we must first check if +%% exchange info is stored, in that case we take care of it. +%%======================================================================== +lock_is_set(Node, Resolvers, LockId) -> + gen_server:cast({global_name_server, Node}, + {exchange, node(), get_names(), _ExtNames = [], + get({sync_tag_his, Node})}), + put({lock_id, Node}, LockId), + %% If both have the lock, continue with exchange. + case get({wait_lock, Node}) of + {exchange, NameList} -> + put({wait_lock, Node}, lock_is_set), + exchange(Node, NameList, Resolvers); + undefined -> + put({wait_lock, Node}, lock_is_set) + end. + +%%======================================================================== +%% exchange +%%======================================================================== +exchange(Node, NameList, Resolvers) -> + ?trace({'####', exchange, {node,Node}, {namelist,NameList}, + {resolvers, Resolvers}}), + case erase({wait_lock, Node}) of + lock_is_set -> + {Node, _Tag, Resolver} = lists:keyfind(Node, 1, Resolvers), + Resolver ! {resolve, NameList, Node}; + undefined -> + put({wait_lock, Node}, {exchange, NameList}) + end. + +resolved(Node, HisResolved, HisKnown, Names_ext, S0) -> + Ops = erase({save_ops, Node}) ++ HisResolved, + %% Known may have shrunk since the lock was taken (due to nodedowns). + Known = S0#state.known, + Synced = S0#state.synced, + NewNodes = [Node | HisKnown], + sync_others(HisKnown), + ExtraInfo = [{vsn,get({prot_vsn, Node})}, {lock, get({lock_id, Node})}], + S = do_ops(Ops, node(), Names_ext, ExtraInfo, S0), + %% I am synced with Node, but not with HisKnown yet + lists:foreach(fun(Pid) -> Pid ! {synced, [Node]} end, S#state.syncers), + S3 = lists:foldl(fun(Node1, S1) -> + F = fun(Tag) -> cancel_locker(Node1,S1,Tag) end, + cancel_resolved_locker(Node1, F) + end, S, HisKnown), + %% The locker that took the lock is asked to send + %% the {new_nodes, ...} message. This ensures that + %% {del_lock, ...} is received after {new_nodes, ...} + %% (except when abcast spawns process(es)...). + NewNodesF = fun() -> + gen_server:abcast(Known, global_name_server, + {new_nodes, node(), Ops, Names_ext, + NewNodes, ExtraInfo}) + end, + F = fun(Tag) -> cancel_locker(Node, S3, Tag, NewNodesF) end, + S4 = cancel_resolved_locker(Node, F), + %% See (*) below... we're node b in that description + AddedNodes = (NewNodes -- Known), + NewKnown = Known ++ AddedNodes, + S4#state.the_locker ! {add_to_known, AddedNodes}, + NewS = trace_message(S4, {added, AddedNodes}, + [{new_nodes, NewNodes}, {abcast, Known}, {ops,Ops}]), + NewS#state{known = NewKnown, synced = [Node | Synced]}. + +cancel_resolved_locker(Node, CancelFun) -> + Tag = get({sync_tag_my, Node}), + ?trace({calling_cancel_locker,Tag,get()}), + S = CancelFun(Tag), + reset_node_state(Node), + S. + +new_nodes(Ops, ConnNode, Names_ext, Nodes, ExtraInfo, S0) -> + Known = S0#state.known, + %% (*) This one requires some thought... + %% We're node a, other nodes b and c: + %% The problem is that {in_sync, a} may arrive before {resolved, [a]} to + %% b from c, leading to b sending {new_nodes, [a]} to us (node a). + %% Therefore, we make sure we never get duplicates in Known. + AddedNodes = lists:delete(node(), Nodes -- Known), + sync_others(AddedNodes), + S = do_ops(Ops, ConnNode, Names_ext, ExtraInfo, S0), + ?trace({added_nodes_in_sync,{added_nodes,AddedNodes}}), + S#state.the_locker ! {add_to_known, AddedNodes}, + S1 = trace_message(S, {added, AddedNodes}, [{ops,Ops}]), + S1#state{known = Known ++ AddedNodes}. + +do_whereis(Name, From) -> + case is_global_lock_set() of + false -> + gen_server:reply(From, where(Name)); + true -> + send_again({whereis, Name, From}) + end. + +terminate(_Reason, _S) -> + true = ets:delete(global_names), + true = ets:delete(global_names_ext), + true = ets:delete(global_locks), + true = ets:delete(global_pid_names), + true = ets:delete(global_pid_ids). + +code_change(_OldVsn, S, _Extra) -> + {ok, S}. + +%% The resolver runs exchange_names in a separate process. The effect +%% is that locks can be used at the same time as name resolution takes +%% place. +start_resolver(Node, MyTag) -> + spawn(fun() -> resolver(Node, MyTag) end). + +resolver(Node, Tag) -> + receive + {resolve, NameList, Node} -> + ?trace({resolver, {me,self()}, {node,Node}, {namelist,NameList}}), + {Ops, Resolved} = exchange_names(NameList, Node, [], []), + Exchange = {exchange_ops, Node, Tag, Ops, Resolved}, + gen_server:cast(global_name_server, Exchange), + exit(normal); + _ -> % Ignore garbage. + resolver(Node, Tag) + end. + +resend_pre_connect(Node) -> + case erase({pre_connect, Node}) of + {Vsn, InitMsg, HisTag} -> + gen_server:cast(self(), + {init_connect, {Vsn, HisTag}, Node, InitMsg}); + _ -> + ok + end. + +ins_name(Name, Pid, Method, FromPidOrNode, ExtraInfo, S0) -> + ?trace({ins_name,insert,{name,Name},{pid,Pid}}), + S1 = delete_global_name_keep_pid(Name, S0), + S = trace_message(S1, {ins_name, node(Pid)}, [Name, Pid]), + insert_global_name(Name, Pid, Method, FromPidOrNode, ExtraInfo, S). + +ins_name_ext(Name, Pid, Method, RegNode, FromPidOrNode, ExtraInfo, S0) -> + ?trace({ins_name_ext, {name,Name}, {pid,Pid}}), + S1 = delete_global_name_keep_pid(Name, S0), + dolink_ext(Pid, RegNode), + S = trace_message(S1, {ins_name_ext, node(Pid)}, [Name, Pid]), + true = ets:insert(global_names_ext, {Name, Pid, RegNode}), + insert_global_name(Name, Pid, Method, FromPidOrNode, ExtraInfo, S). + +where(Name) -> + case ets:lookup(global_names, Name) of + [{_Name, Pid, _Method, _RPid, _Ref}] -> Pid; + [] -> undefined + end. + +handle_set_lock(Id, Pid, S) -> + ?trace({handle_set_lock, Id, Pid}), + case can_set_lock(Id) of + {true, PidRefs} -> + case pid_is_locking(Pid, PidRefs) of + true -> + {true, S}; + false -> + {true, insert_lock(Id, Pid, PidRefs, S)} + end; + false=Reply -> + {Reply, S} + end. + +can_set_lock({ResourceId, LockRequesterId}) -> + case ets:lookup(global_locks, ResourceId) of + [{ResourceId, LockRequesterId, PidRefs}] -> + {true, PidRefs}; + [{ResourceId, _LockRequesterId2, _PidRefs}] -> + false; + [] -> + {true, []} + end. + +insert_lock({ResourceId, LockRequesterId}=Id, Pid, PidRefs, S) -> + {RPid, Ref} = do_monitor(Pid), + true = ets:insert(global_pid_ids, {Pid, ResourceId}), + true = ets:insert(global_pid_ids, {Ref, ResourceId}), + Lock = {ResourceId, LockRequesterId, [{Pid,RPid,Ref} | PidRefs]}, + true = ets:insert(global_locks, Lock), + trace_message(S, {ins_lock, node(Pid)}, [Id, Pid]). + +is_global_lock_set() -> + is_lock_set(?GLOBAL_RID). + +is_lock_set(ResourceId) -> + ets:member(global_locks, ResourceId). + +handle_del_lock({ResourceId, LockReqId}, Pid, S0) -> + ?trace({handle_del_lock, {pid,Pid},{id,{ResourceId, LockReqId}}}), + case ets:lookup(global_locks, ResourceId) of + [{ResourceId, LockReqId, PidRefs}]-> + remove_lock(ResourceId, LockReqId, Pid, PidRefs, false, S0); + _ -> S0 + end. + +remove_lock(ResourceId, LockRequesterId, Pid, [{Pid,RPid,Ref}], Down, S0) -> + ?trace({remove_lock_1, {id,ResourceId},{pid,Pid}}), + true = erlang:demonitor(Ref, [flush]), + kill_monitor_proc(RPid, Pid), + true = ets:delete(global_locks, ResourceId), + true = ets:delete_object(global_pid_ids, {Pid, ResourceId}), + true = ets:delete_object(global_pid_ids, {Ref, ResourceId}), + S = case ResourceId of + ?GLOBAL_RID -> S0#state{global_lock_down = Down}; + _ -> S0 + end, + trace_message(S, {rem_lock, node(Pid)}, + [{ResourceId, LockRequesterId}, Pid]); +remove_lock(ResourceId, LockRequesterId, Pid, PidRefs0, _Down, S) -> + ?trace({remove_lock_2, {id,ResourceId},{pid,Pid}}), + PidRefs = case lists:keyfind(Pid, 1, PidRefs0) of + {Pid, RPid, Ref} -> + true = erlang:demonitor(Ref, [flush]), + kill_monitor_proc(RPid, Pid), + true = ets:delete_object(global_pid_ids, + {Ref, ResourceId}), + lists:keydelete(Pid, 1, PidRefs0); + false -> + PidRefs0 + end, + Lock = {ResourceId, LockRequesterId, PidRefs}, + true = ets:insert(global_locks, Lock), + true = ets:delete_object(global_pid_ids, {Pid, ResourceId}), + trace_message(S, {rem_lock, node(Pid)}, + [{ResourceId, LockRequesterId}, Pid]). + +kill_monitor_proc(Pid, Pid) -> + ok; +kill_monitor_proc(RPid, _Pid) -> + exit(RPid, kill). + +do_ops(Ops, ConnNode, Names_ext, ExtraInfo, S0) -> + ?trace({do_ops, {ops,Ops}}), + + XInserts = [{Name, Pid, RegNode, Method} || + {Name2, Pid2, RegNode} <- Names_ext, + {insert, {Name, Pid, Method}} <- Ops, + Name =:= Name2, Pid =:= Pid2], + S1 = lists:foldl(fun({Name, Pid, RegNode, Method}, S1) -> + ins_name_ext(Name, Pid, Method, RegNode, + ConnNode, ExtraInfo, S1) + end, S0, XInserts), + + XNames = [Name || {Name, _Pid, _RegNode, _Method} <- XInserts], + Inserts = [{Name, Pid, node(Pid), Method} || + {insert, {Name, Pid, Method}} <- Ops, + not lists:member(Name, XNames)], + S2 = lists:foldl(fun({Name, Pid, _RegNode, Method}, S2) -> + ins_name(Name, Pid, Method, ConnNode, + ExtraInfo, S2) + end, S1, Inserts), + + DelNames = [Name || {delete, Name} <- Ops], + lists:foldl(fun(Name, S) -> delete_global_name2(Name, S) + end, S2, DelNames). + +%% It is possible that a node that was up and running when the +%% operations were assembled has since died. The final {in_sync,...} +%% messages do not generate nodedown messages for such nodes. To +%% compensate "artificial" nodedown messages are created. Since +%% monitor_node may take some time processes are spawned to avoid +%% locking up the global_name_server. Should somehow double nodedown +%% messages occur (one of them artificial), nothing bad can happen +%% (the second nodedown is a no-op). It is assumed that there cannot +%% be a nodeup before the artificial nodedown. +%% +%% The extra nodedown messages generated here also take care of the +%% case that a nodedown message is received _before_ the operations +%% are run. +sync_others(Nodes) -> + N = case application:get_env(kernel, ?N_CONNECT_RETRIES) of + {ok, NRetries} when is_integer(NRetries), + NRetries >= 0 -> NRetries; + _ -> ?DEFAULT_N_CONNECT_RETRIES + end, + lists:foreach(fun(Node) -> + spawn(fun() -> sync_other(Node, N) end) + end, Nodes). + +sync_other(Node, N) -> + erlang:monitor_node(Node, true, [allow_passive_connect]), + receive + {nodedown, Node} when N > 0 -> + sync_other(Node, N - 1); + {nodedown, Node} -> + ?trace({missing_nodedown, {node, Node}}), + error_logger:warning_msg("global: ~w failed to connect to ~w\n", + [node(), Node]), + global_name_server ! {extra_nodedown, Node} + after 0 -> + gen_server:cast({global_name_server,Node}, {in_sync,node(),true}) + end. + % monitor_node(Node, false), + % exit(normal). + +insert_global_name(Name, Pid, Method, FromPidOrNode, ExtraInfo, S) -> + {RPid, Ref} = do_monitor(Pid), + true = ets:insert(global_names, {Name, Pid, Method, RPid, Ref}), + true = ets:insert(global_pid_names, {Pid, Name}), + true = ets:insert(global_pid_names, {Ref, Name}), + case lock_still_set(FromPidOrNode, ExtraInfo, S) of + true -> + S; + false -> + %% The node that took the lock has gone down and then up + %% again. The {register, ...} or {new_nodes, ...} message + %% was delayed and arrived after nodeup (maybe it caused + %% the nodeup). The DOWN signal from the monitor of the + %% lock has removed the lock. + %% Note: it is assumed here that the DOWN signal arrives + %% _before_ nodeup and any message that caused nodeup. + %% This is true of Erlang/OTP. + delete_global_name2(Name, S) + end. + +lock_still_set(PidOrNode, ExtraInfo, S) -> + case ets:lookup(global_locks, ?GLOBAL_RID) of + [{?GLOBAL_RID, _LockReqId, PidRefs}] when is_pid(PidOrNode) -> + %% Name registration. + lists:keymember(PidOrNode, 1, PidRefs); + [{?GLOBAL_RID, LockReqId, PidRefs}] when is_atom(PidOrNode) -> + case extra_info(lock, ExtraInfo) of + {?GLOBAL_RID, LockId} -> % R11B-4 or later + LockReqId =:= LockId; + undefined -> + lock_still_set_old(PidOrNode, LockReqId, PidRefs) + end; + [] -> + %% If the global lock was not removed by a DOWN message + %% then we have a node that do not monitor locking pids + %% (pre R11B-3), or an R11B-3 node (which does not ensure + %% that {new_nodes, ...} arrives before {del_lock, ...}). + not S#state.global_lock_down + end. + +%%% The following is probably overkill. It is possible that this node +%%% has been locked again, but it is a rare occasion. +lock_still_set_old(_Node, ReqId, _PidRefs) when is_pid(ReqId) -> + %% Cannot do better than return true. + true; +lock_still_set_old(Node, ReqId, PidRefs) when is_list(ReqId) -> + %% Connection, version > 4, but before R11B-4. + [P || {P, _RPid, _Ref} <- PidRefs, node(P) =:= Node] =/= []. + +extra_info(Tag, ExtraInfo) -> + %% ExtraInfo used to be a list of nodes (vsn 2). + case catch lists:keyfind(Tag, 1, ExtraInfo) of + {Tag, Info} -> + Info; + _ -> + undefined + end. + +del_name(Ref, S) -> + NameL = [{Name, Pid} || + {_, Name} <- ets:lookup(global_pid_names, Ref), + {_, Pid, _Method, _RPid, Ref1} <- + ets:lookup(global_names, Name), + Ref1 =:= Ref], + ?trace({async_del_name, self(), NameL, Ref}), + case NameL of + [{Name, Pid}] -> + _ = del_names(Name, Pid, S), + delete_global_name2(Name, S); + [] -> + S + end. + +%% Send {async_del_name, ...} to old nodes (pre R11B-3). +del_names(Name, Pid, S) -> + Send = case ets:lookup(global_names_ext, Name) of + [{Name, Pid, RegNode}] -> + RegNode =:= node(); + [] -> + node(Pid) =:= node() + end, + if + Send -> + ?trace({del_names, {pid,Pid}, {name,Name}}), + S#state.the_deleter ! {delete_name, self(), Name, Pid}; + true -> + ok + end. + +%% Keeps the entry in global_names for whereis_name/1. +delete_global_name_keep_pid(Name, S) -> + case ets:lookup(global_names, Name) of + [{Name, Pid, _Method, RPid, Ref}] -> + delete_global_name2(Name, Pid, RPid, Ref, S); + [] -> + S + end. + +delete_global_name2(Name, S) -> + case ets:lookup(global_names, Name) of + [{Name, Pid, _Method, RPid, Ref}] -> + true = ets:delete(global_names, Name), + delete_global_name2(Name, Pid, RPid, Ref, S); + [] -> + S + end. + +delete_global_name2(Name, Pid, RPid, Ref, S) -> + true = erlang:demonitor(Ref, [flush]), + kill_monitor_proc(RPid, Pid), + delete_global_name(Name, Pid), + ?trace({delete_global_name,{item,Name},{pid,Pid}}), + true = ets:delete_object(global_pid_names, {Pid, Name}), + true = ets:delete_object(global_pid_names, {Ref, Name}), + case ets:lookup(global_names_ext, Name) of + [{Name, Pid, RegNode}] -> + true = ets:delete(global_names_ext, Name), + ?trace({delete_global_name, {name,Name,{pid,Pid},{RegNode,Pid}}}), + dounlink_ext(Pid, RegNode); + [] -> + ?trace({delete_global_name,{name,Name,{pid,Pid},{node(Pid),Pid}}}), + ok + end, + trace_message(S, {del_name, node(Pid)}, [Name, Pid]). + +%% delete_global_name/2 is traced by the inviso application. +%% Do not change. +delete_global_name(_Name, _Pid) -> + ok. + +%%----------------------------------------------------------------- +%% The locker is a satellite process to global_name_server. When a +%% nodeup is received from a new node the global_name_server sends a +%% message to the locker. The locker tries to set a lock in our +%% partition, i.e. on all nodes known to us. When the lock is set, it +%% tells global_name_server about it, and keeps the lock set. +%% global_name_server sends a cancel message to the locker when the +%% partitions are connected. + +%% There are two versions of the protocol between lockers on two nodes: +%% Version 1: used by unpatched R7. +%% Version 2: the messages exchanged between the lockers include the known +%% nodes (see OTP-3576). +%%----------------------------------------------------------------- + +-define(locker_vsn, 2). + +-record(multi, + {local = [], % Requests from nodes on the local host. + remote = [], % Other requests. + known = [], % Copy of global_name_server's known nodes. It's + % faster to keep a copy of known than asking + % for it when needed. + the_boss, % max([node() | 'known']) + just_synced = false, % true if node() synced just a moment ago + %% Statistics: + do_trace % bool() + }). + +-record(him, {node, locker, vsn, my_tag}). + +start_the_locker(DoTrace) -> + spawn_link(fun() -> init_the_locker(DoTrace) end). + +init_the_locker(DoTrace) -> + process_flag(trap_exit, true), % needed? + S0 = #multi{do_trace = DoTrace}, + S1 = update_locker_known({add, get_known()}, S0), + loop_the_locker(S1), + erlang:error(locker_exited). + +loop_the_locker(S) -> + ?trace({loop_the_locker,S}), + receive + Message when element(1, Message) =/= nodeup -> + the_locker_message(Message, S) + after 0 -> + Timeout = + case {S#multi.local, S#multi.remote} of + {[],[]} -> + infinity; + _ -> + %% It is important that the timeout is greater + %% than zero, or the chance that some other node + %% in the partition sets the lock once this node + %% has failed after setting the lock is very slim. + if + S#multi.just_synced -> + 0; % no reason to wait after success + S#multi.known =:= [] -> + 200; % just to get started + true -> + erlang:min(1000 + 100*length(S#multi.known), + 3000) + end + end, + S1 = S#multi{just_synced = false}, + receive + Message when element(1, Message) =/= nodeup -> + the_locker_message(Message, S1) + after Timeout -> + case is_global_lock_set() of + true -> + loop_the_locker(S1); + false -> + select_node(S1) + end + end + end. + +the_locker_message({his_the_locker, HisTheLocker, HisKnown0, _MyKnown}, S) -> + ?trace({his_the_locker, HisTheLocker, {node,node(HisTheLocker)}}), + {HisVsn, _HisKnown} = HisKnown0, + true = HisVsn > 4, + receive + {nodeup, Node, MyTag} when node(HisTheLocker) =:= Node -> + ?trace({the_locker_nodeup, {node,Node},{mytag,MyTag}}), + Him = #him{node = node(HisTheLocker), my_tag = MyTag, + locker = HisTheLocker, vsn = HisVsn}, + loop_the_locker(add_node(Him, S)); + {cancel, Node, _Tag, no_fun} when node(HisTheLocker) =:= Node -> + loop_the_locker(S) + after 60000 -> + ?trace({nodeupnevercame, node(HisTheLocker)}), + error_logger:error_msg("global: nodeup never came ~w ~w\n", + [node(), node(HisTheLocker)]), + loop_the_locker(S#multi{just_synced = false}) + end; +the_locker_message({cancel, _Node, undefined, no_fun}, S) -> + ?trace({cancel_the_locker, undefined, {node,_Node}}), + %% If we actually cancel something when a cancel message with the + %% tag 'undefined' arrives, we may be acting on an old nodedown, + %% to cancel a new nodeup, so we can't do that. + loop_the_locker(S); +the_locker_message({cancel, Node, Tag, no_fun}, S) -> + ?trace({the_locker, cancel, {multi,S}, {tag,Tag},{node,Node}}), + receive + {nodeup, Node, Tag} -> + ?trace({cancelnodeup2, {node,Node},{tag,Tag}}), + ok + after 0 -> + ok + end, + loop_the_locker(remove_node(Node, S)); +the_locker_message({lock_set, _Pid, false, _}, S) -> + ?trace({the_locker, spurious, {node,node(_Pid)}}), + loop_the_locker(S); +the_locker_message({lock_set, Pid, true, _HisKnown}, S) -> + Node = node(Pid), + ?trace({the_locker, self(), spontaneous, {node,Node}}), + case find_node_tag(Node, S) of + {true, MyTag, HisVsn} -> + LockId = locker_lock_id(Pid, HisVsn), + {IsLockSet, S1} = lock_nodes_safely(LockId, [], S), + Pid ! {lock_set, self(), IsLockSet, S1#multi.known}, + Known2 = [node() | S1#multi.known], + ?trace({the_locker, spontaneous, {known2, Known2}, + {node,Node}, {is_lock_set,IsLockSet}}), + case IsLockSet of + true -> + gen_server:cast(global_name_server, + {lock_is_set, Node, MyTag, LockId}), + ?trace({lock_sync_done, {pid,Pid}, + {node,node(Pid)}, {me,self()}}), + %% Wait for global to tell us to remove lock. + %% Should the other locker's node die, + %% global_name_server will receive nodedown, and + %% then send {cancel, Node, Tag}. + receive + {cancel, Node, _Tag, Fun} -> + ?trace({cancel_the_lock,{node,Node}}), + call_fun(Fun), + delete_global_lock(LockId, Known2) + end, + S2 = S1#multi{just_synced = true}, + loop_the_locker(remove_node(Node, S2)); + false -> + loop_the_locker(S1#multi{just_synced = false}) + end; + false -> + ?trace({the_locker, not_there, {node,Node}}), + Pid ! {lock_set, self(), false, S#multi.known}, + loop_the_locker(S) + end; +the_locker_message({add_to_known, Nodes}, S) -> + S1 = update_locker_known({add, Nodes}, S), + loop_the_locker(S1); +the_locker_message({remove_from_known, Node}, S) -> + S1 = update_locker_known({remove, Node}, S), + loop_the_locker(S1); +the_locker_message({do_trace, DoTrace}, S) -> + loop_the_locker(S#multi{do_trace = DoTrace}); +the_locker_message(Other, S) -> + unexpected_message(Other, locker), + ?trace({the_locker, {other_msg, Other}}), + loop_the_locker(S). + +%% Requests from nodes on the local host are chosen before requests +%% from other nodes. This should be a safe optimization. +select_node(S) -> + UseRemote = S#multi.local =:= [], + Others1 = if UseRemote -> S#multi.remote; true -> S#multi.local end, + Others2 = exclude_known(Others1, S#multi.known), + S1 = if + UseRemote -> S#multi{remote = Others2}; + true -> S#multi{local = Others2} + end, + if + Others2 =:= [] -> + loop_the_locker(S1); + true -> + Him = random_element(Others2), + #him{locker = HisTheLocker, vsn = HisVsn, + node = Node, my_tag = MyTag} = Him, + HisNode = [Node], + Us = [node() | HisNode], + LockId = locker_lock_id(HisTheLocker, HisVsn), + ?trace({select_node, self(), {us, Us}}), + %% HisNode = [Node] prevents deadlock: + {IsLockSet, S2} = lock_nodes_safely(LockId, HisNode, S1), + case IsLockSet of + true -> + Known1 = Us ++ S2#multi.known, + ?trace({sending_lock_set, self(), {his,HisTheLocker}}), + HisTheLocker ! {lock_set, self(), true, S2#multi.known}, + S3 = lock_is_set(S2, Him, MyTag, Known1, LockId), + loop_the_locker(S3); + false -> + loop_the_locker(S2) + end + end. + +%% Version 5: Both sides use the same requester id. Thereby the nodes +%% common to both sides are locked by both locker processes. This +%% means that the lock is still there when the 'new_nodes' message is +%% received even if the other side has deleted the lock. +locker_lock_id(Pid, Vsn) when Vsn > 4 -> + {?GLOBAL_RID, lists:sort([self(), Pid])}. + +lock_nodes_safely(LockId, Extra, S0) -> + %% Locking node() could stop some node that has already locked the + %% boss, so just check if it is possible to lock node(). + First = delete_nonode([S0#multi.the_boss]), + case ([node()] =:= First) orelse (can_set_lock(LockId) =/= false) of + true -> + %% Locking the boss first is an optimization. + case set_lock(LockId, First, 0) of + true -> + S = update_locker_known(S0), + %% The boss may have changed, but don't bother. + Second = delete_nonode([node() | Extra] -- First), + case set_lock(LockId, Second, 0) of + true -> + Known = S#multi.known, + case set_lock(LockId, Known -- First, 0) of + true -> + _ = locker_trace(S, ok, {First, Known}), + {true, S}; + false -> + %% Since the boss is locked we + %% should have gotten the lock, at + %% least if no one else is locking + %% 'global'. Calling set_lock with + %% Retries > 0 does not seem to + %% speed things up. + SoFar = First ++ Second, + del_lock(LockId, SoFar), + _ = locker_trace(S, not_ok, {Known,SoFar}), + {false, S} + end; + false -> + del_lock(LockId, First), + _ = locker_trace(S, not_ok, {Second, First}), + {false, S} + end; + false -> + _ = locker_trace(S0, not_ok, {First, []}), + {false, S0} + end; + false -> + {false, S0} + end. + +delete_nonode(L) -> + lists:delete(nonode@nohost, L). + +%% Let the server add timestamp. +locker_trace(#multi{do_trace = false}, _, _Nodes) -> + ok; +locker_trace(#multi{do_trace = true}, ok, Ns) -> + global_name_server ! {trace_message, {locker_succeeded, node()}, Ns}; +locker_trace(#multi{do_trace = true}, not_ok, Ns) -> + global_name_server ! {trace_message, {locker_failed, node()}, Ns}; +locker_trace(#multi{do_trace = true}, rejected, Ns) -> + global_name_server ! {trace_message, {lock_rejected, node()}, Ns}. + +update_locker_known(S) -> + receive + {add_to_known, Nodes} -> + S1 = update_locker_known({add, Nodes}, S), + update_locker_known(S1); + {remove_from_known, Node} -> + S1 = update_locker_known({remove, Node}, S), + update_locker_known(S1) + after 0 -> + S + end. + +update_locker_known(Upd, S) -> + Known = case Upd of + {add, Nodes} -> Nodes ++ S#multi.known; + {remove, Node} -> lists:delete(Node, S#multi.known) + end, + TheBoss = the_boss([node() | Known]), + S#multi{known = Known, the_boss = TheBoss}. + +random_element(L) -> + {A,B,C} = now(), + E = (A+B+C) rem length(L), + lists:nth(E+1, L). + +exclude_known(Others, Known) -> + [N || N <- Others, not lists:member(N#him.node, Known)]. + +lock_is_set(S, Him, MyTag, Known1, LockId) -> + Node = Him#him.node, + receive + {lock_set, P, true, _} when node(P) =:= Node -> + gen_server:cast(global_name_server, + {lock_is_set, Node, MyTag, LockId}), + ?trace({lock_sync_done, {p,P, node(P)}, {me,self()}}), + + %% Wait for global to tell us to remove lock. Should the + %% other locker's node die, global_name_server will + %% receive nodedown, and then send {cancel, Node, Tag, Fun}. + receive + {cancel, Node, _, Fun} -> + ?trace({lock_set_loop, {known1,Known1}}), + call_fun(Fun), + delete_global_lock(LockId, Known1) + end, + S#multi{just_synced = true, + local = lists:delete(Him, S#multi.local), + remote = lists:delete(Him, S#multi.remote)}; + {lock_set, P, false, _} when node(P) =:= Node -> + ?trace({not_both_set, {node,Node},{p, P},{known1,Known1}}), + _ = locker_trace(S, rejected, Known1), + delete_global_lock(LockId, Known1), + S; + {cancel, Node, _, Fun} -> + ?trace({the_locker, cancel2, {node,Node}}), + call_fun(Fun), + _ = locker_trace(S, rejected, Known1), + delete_global_lock(LockId, Known1), + remove_node(Node, S); + {'EXIT', _, _} -> + ?trace({the_locker, exit, {node,Node}}), + _ = locker_trace(S, rejected, Known1), + delete_global_lock(LockId, Known1), + S + %% There used to be an 'after' clause (OTP-4902), but it is + %% no longer needed: + %% OTP-5770. Version 5 of the protocol. Deadlock can no longer + %% occur due to the fact that if a partition is locked, one + %% node in the other partition is also locked with the same + %% lock-id, which makes it impossible for any node in the + %% other partition to lock its partition unless it negotiates + %% with the first partition. + end. + +%% The locker does the {new_nodes, ...} call before removing the lock. +call_fun(no_fun) -> + ok; +call_fun(Fun) -> + Fun(). + +%% The lock on the boss is removed last. The purpose is to reduce the +%% risk of failing to lock the known nodes after having locked the +%% boss. (Assumes the boss occurs only once.) +delete_global_lock(LockId, Nodes) -> + TheBoss = the_boss(Nodes), + del_lock(LockId, lists:delete(TheBoss, Nodes)), + del_lock(LockId, [TheBoss]). + +the_boss(Nodes) -> + lists:max(Nodes). + +find_node_tag(Node, S) -> + case find_node_tag2(Node, S#multi.local) of + false -> + find_node_tag2(Node, S#multi.remote); + Reply -> + Reply + end. + +find_node_tag2(_Node, []) -> + false; +find_node_tag2(Node, [#him{node = Node, my_tag = MyTag, vsn = HisVsn} | _]) -> + {true, MyTag, HisVsn}; +find_node_tag2(Node, [_E | Rest]) -> + find_node_tag2(Node, Rest). + +remove_node(Node, S) -> + S#multi{local = remove_node2(Node, S#multi.local), + remote = remove_node2(Node, S#multi.remote)}. + +remove_node2(_Node, []) -> + []; +remove_node2(Node, [#him{node = Node} | Rest]) -> + Rest; +remove_node2(Node, [E | Rest]) -> + [E | remove_node2(Node, Rest)]. + +add_node(Him, S) -> + case is_node_local(Him#him.node) of + true -> + S#multi{local = [Him | S#multi.local]}; + false -> + S#multi{remote = [Him | S#multi.remote]} + end. + +is_node_local(Node) -> + {ok, Host} = inet:gethostname(), + case catch split_node(atom_to_list(Node), $@, []) of + [_, Host] -> + true; + _ -> + false + end. + +split_node([Chr|T], Chr, Ack) -> [lists:reverse(Ack)|split_node(T, Chr, [])]; +split_node([H|T], Chr, Ack) -> split_node(T, Chr, [H|Ack]); +split_node([], _, Ack) -> [lists:reverse(Ack)]. + +cancel_locker(Node, S, Tag) -> + cancel_locker(Node, S, Tag, no_fun). + +cancel_locker(Node, S, Tag, ToBeRunOnLockerF) -> + S#state.the_locker ! {cancel, Node, Tag, ToBeRunOnLockerF}, + Resolvers = S#state.resolvers, + ?trace({cancel_locker, {node,Node},{tag,Tag}, + {sync_tag_my, get({sync_tag_my, Node})},{resolvers,Resolvers}}), + case lists:keyfind(Node, 1, Resolvers) of + {_, Tag, Resolver} -> + ?trace({{resolver, Resolver}}), + exit(Resolver, kill), + S1 = trace_message(S, {kill_resolver, Node}, [Tag, Resolver]), + S1#state{resolvers = lists:keydelete(Node, 1, Resolvers)}; + _ -> + S + end. + +reset_node_state(Node) -> + ?trace({{node,Node}, reset_node_state, get()}), + erase({wait_lock, Node}), + erase({save_ops, Node}), + erase({pre_connect, Node}), + erase({prot_vsn, Node}), + erase({sync_tag_my, Node}), + erase({sync_tag_his, Node}), + erase({lock_id, Node}). + +%% Some node sent us his names. When a name clash is found, the resolve +%% function is called from the smaller node => all resolve funcs are called +%% from the same partition. +exchange_names([{Name, Pid, Method} | Tail], Node, Ops, Res) -> + case ets:lookup(global_names, Name) of + [{Name, Pid, _Method, _RPid2, _Ref2}] -> + exchange_names(Tail, Node, Ops, Res); + [{Name, Pid2, Method2, _RPid2, _Ref2}] when node() < Node -> + %% Name clash! Add the result of resolving to Res(olved). + %% We know that node(Pid) =/= node(), so we don't + %% need to link/unlink to Pid. + Node2 = node(Pid2), %% Node2 is connected to node(). + case rpc:call(Node2, ?MODULE, resolve_it, + [Method2, Name, Pid, Pid2]) of + Pid -> + Op = {insert, {Name, Pid, Method}}, + exchange_names(Tail, Node, [Op | Ops], Res); + Pid2 -> + Op = {insert, {Name, Pid2, Method2}}, + exchange_names(Tail, Node, Ops, [Op | Res]); + none -> + Op = {delete, Name}, + exchange_names(Tail, Node, [Op | Ops], [Op | Res]); + {badrpc, Badrpc} -> + error_logger:info_msg("global: badrpc ~w received when " + "conflicting name ~w was found\n", + [Badrpc, Name]), + Op = {insert, {Name, Pid, Method}}, + exchange_names(Tail, Node, [Op | Ops], Res); + Else -> + error_logger:info_msg("global: Resolve method ~w for " + "conflicting name ~w returned ~w\n", + [Method, Name, Else]), + Op = {delete, Name}, + exchange_names(Tail, Node, [Op | Ops], [Op | Res]) + end; + [{Name, _Pid2, _Method, _RPid, _Ref}] -> + %% The other node will solve the conflict. + exchange_names(Tail, Node, Ops, Res); + _ -> + %% Entirely new name. + exchange_names(Tail, Node, + [{insert, {Name, Pid, Method}} | Ops], Res) + end; +exchange_names([], _, Ops, Res) -> + ?trace({exchange_names_finish,{ops,Ops},{res,Res}}), + {Ops, Res}. + +resolve_it(Method, Name, Pid1, Pid2) -> + catch Method(Name, Pid1, Pid2). + +minmax(P1,P2) -> + if node(P1) < node(P2) -> {P1, P2}; true -> {P2, P1} end. + +-spec random_exit_name(term(), pid(), pid()) -> pid(). +random_exit_name(Name, Pid, Pid2) -> + {Min, Max} = minmax(Pid, Pid2), + error_logger:info_msg("global: Name conflict terminating ~w\n", + [{Name, Max}]), + exit(Max, kill), + Min. + +random_notify_name(Name, Pid, Pid2) -> + {Min, Max} = minmax(Pid, Pid2), + Max ! {global_name_conflict, Name}, + Min. + +-spec notify_all_name(term(), pid(), pid()) -> 'none'. +notify_all_name(Name, Pid, Pid2) -> + Pid ! {global_name_conflict, Name, Pid2}, + Pid2 ! {global_name_conflict, Name, Pid}, + none. + +dolink_ext(Pid, RegNode) when RegNode =:= node() -> + link(Pid); +dolink_ext(_, _) -> + ok. + +dounlink_ext(Pid, RegNode) when RegNode =:= node() -> + unlink_pid(Pid); +dounlink_ext(_Pid, _RegNode) -> + ok. + +unlink_pid(Pid) -> + case ets:member(global_pid_names, Pid) of + false -> + case ets:member(global_pid_ids, Pid) of + false -> + unlink(Pid); + true -> + ok + end; + true -> + ok + end. + +pid_is_locking(Pid, PidRefs) -> + lists:keyfind(Pid, 1, PidRefs) =/= false. + +delete_lock(Ref, S0) -> + Locks = pid_locks(Ref), + del_locks(Locks, Ref, S0#state.known), + F = fun({ResourceId, LockRequesterId, PidRefs}, S) -> + {Pid, _RPid, Ref} = lists:keyfind(Ref, 3, PidRefs), + remove_lock(ResourceId, LockRequesterId, Pid, PidRefs, true,S) + end, + lists:foldl(F, S0, Locks). + +pid_locks(Ref) -> + L = lists:flatmap(fun({_, ResourceId}) -> + ets:lookup(global_locks, ResourceId) + end, ets:lookup(global_pid_ids, Ref)), + [Lock || Lock = {_Id, _Req, PidRefs} <- L, + rpid_is_locking(Ref, PidRefs)]. + +rpid_is_locking(Ref, PidRefs) -> + lists:keyfind(Ref, 3, PidRefs) =/= false. + +%% Send {async_del_lock, ...} to old nodes (pre R11B-3). +del_locks([{ResourceId, _LockReqId, PidRefs} | Tail], Ref, KnownNodes) -> + {Pid, _RPid, Ref} = lists:keyfind(Ref, 3, PidRefs), + case node(Pid) =:= node() of + true -> + gen_server:abcast(KnownNodes, global_name_server, + {async_del_lock, ResourceId, Pid}); + false -> + ok + end, + del_locks(Tail, Ref, KnownNodes); +del_locks([], _Ref, _KnownNodes) -> + ok. + +handle_nodedown(Node, S) -> + %% DOWN signals from monitors have removed locks and registered names. + #state{known = Known, synced = Syncs} = S, + NewS = cancel_locker(Node, S, get({sync_tag_my, Node})), + NewS#state.the_locker ! {remove_from_known, Node}, + reset_node_state(Node), + NewS#state{known = lists:delete(Node, Known), + synced = lists:delete(Node, Syncs)}. + +get_names() -> + ets:select(global_names, + ets:fun2ms(fun({Name, Pid, Method, _RPid, _Ref}) -> + {Name, Pid, Method} + end)). + +get_names_ext() -> + ets:tab2list(global_names_ext). + +get_known() -> + gen_server:call(global_name_server, get_known, infinity). + +random_sleep(Times) -> + case (Times rem 10) of + 0 -> erase(random_seed); + _ -> ok + end, + case get(random_seed) of + undefined -> + {A1, A2, A3} = now(), + random:seed(A1, A2, A3 + erlang:phash(node(), 100000)); + _ -> ok + end, + %% First time 1/4 seconds, then doubling each time up to 8 seconds max. + Tmax = if Times > 5 -> 8000; + true -> ((1 bsl Times) * 1000) div 8 + end, + T = random:uniform(Tmax), + ?trace({random_sleep, {me,self()}, {times,Times}, {t,T}, {tmax,Tmax}}), + receive after T -> ok end. + +dec(infinity) -> infinity; +dec(N) -> N - 1. + +send_again(Msg) -> + Me = self(), + spawn(fun() -> timer(Me, Msg) end). + +timer(Pid, Msg) -> + random_sleep(5), + Pid ! Msg. + +change_our_node_name(NewNode, S) -> + S1 = trace_message(S, {new_node_name, NewNode}, []), + S1#state{node_name = NewNode}. + +trace_message(#state{trace = no_trace}=S, _M, _X) -> + S; +trace_message(S, M, X) -> + S#state{trace = [trace_message(M, X) | S#state.trace]}. + +trace_message(M, X) -> + {node(), now(), M, nodes(), X}. + +%%----------------------------------------------------------------- +%% Each sync process corresponds to one call to sync. Each such +%% process asks the global_name_server on all Nodes if it is in sync +%% with Nodes. If not, that (other) node spawns a syncer process that +%% waits for global to get in sync with all Nodes. When it is in +%% sync, the syncer process tells the original sync process about it. +%%----------------------------------------------------------------- +start_sync(Nodes, From) -> + spawn_link(fun() -> sync_init(Nodes, From) end). + +sync_init(Nodes, From) -> + lists:foreach(fun(Node) -> monitor_node(Node, true) end, Nodes), + sync_loop(Nodes, From). + +sync_loop([], From) -> + gen_server:reply(From, ok); +sync_loop(Nodes, From) -> + receive + {nodedown, Node} -> + monitor_node(Node, false), + sync_loop(lists:delete(Node, Nodes), From); + {synced, SNodes} -> + lists:foreach(fun(N) -> monitor_node(N, false) end, SNodes), + sync_loop(Nodes -- SNodes, From) + end. + +%%%======================================================================= +%%% Get the current global_groups definition +%%%======================================================================= +check_sync_nodes() -> + case get_own_nodes() of + {ok, all} -> + nodes(); + {ok, NodesNG} -> + %% global_groups parameter is defined, we are not allowed to sync + %% with nodes not in our own global group. + intersection(nodes(), NodesNG); + {error, _} = Error -> + Error + end. + +check_sync_nodes(SyncNodes) -> + case get_own_nodes() of + {ok, all} -> + SyncNodes; + {ok, NodesNG} -> + %% global_groups parameter is defined, we are not allowed to sync + %% with nodes not in our own global group. + OwnNodeGroup = intersection(nodes(), NodesNG), + IllegalSyncNodes = (SyncNodes -- [node() | OwnNodeGroup]), + case IllegalSyncNodes of + [] -> SyncNodes; + _ -> {error, {"Trying to sync nodes not defined in " + "the own global group", IllegalSyncNodes}} + end; + {error, _} = Error -> + Error + end. + +get_own_nodes() -> + case global_group:get_own_nodes_with_errors() of + {error, Error} -> + {error, {"global_groups definition error", Error}}; + OkTup -> + OkTup + end. + +%%----------------------------------------------------------------- +%% The deleter process is a satellite process to global_name_server +%% that does background batch deleting of names when a process +%% that had globally registered names dies. It is started by and +%% linked to global_name_server. +%%----------------------------------------------------------------- + +start_the_deleter(Global) -> + spawn_link(fun() -> loop_the_deleter(Global) end). + +loop_the_deleter(Global) -> + Deletions = collect_deletions(Global, []), + ?trace({loop_the_deleter, self(), {deletions,Deletions}, + {names,get_names()}}), + %% trans_all_known is called rather than trans/3 with nodes() as + %% third argument. The reason is that known gets updated by + %% new_nodes when the lock is still set. nodes() on the other hand + %% could be updated later (if in_sync is received after the lock + %% is gone). It is not likely that in_sync would be received after + %% the lock has been taken here, but using trans_all_known makes it + %% even less likely. + trans_all_known( + fun(Known) -> + lists:map( + fun({Name,Pid}) -> + gen_server:abcast(Known, global_name_server, + {async_del_name, Name, Pid}) + end, Deletions) + end), + loop_the_deleter(Global). + +collect_deletions(Global, Deletions) -> + receive + {delete_name, Global, Name, Pid} -> + collect_deletions(Global, [{Name,Pid} | Deletions]); + Other -> + unexpected_message(Other, deleter), + collect_deletions(Global, Deletions) + after case Deletions of + [] -> infinity; + _ -> 0 + end -> + lists:reverse(Deletions) + end. + +%% The registrar is a helper process that registers and unregisters +%% names. Since it never dies it assures that names are registered and +%% unregistered on all known nodes. It is started by and linked to +%% global_name_server. + +start_the_registrar() -> + spawn_link(fun() -> loop_the_registrar() end). + +loop_the_registrar() -> + receive + {trans_all_known, Fun, From} -> + ?trace({loop_the_registrar, self(), Fun, From}), + gen_server:reply(From, trans_all_known(Fun)); + Other -> + unexpected_message(Other, register) + end, + loop_the_registrar(). + +unexpected_message({'EXIT', _Pid, _Reason}, _What) -> + %% global_name_server died + ok; +unexpected_message(Message, What) -> + error_logger:warning_msg("The global_name_server ~w process " + "received an unexpected message:\n~p\n", + [What, Message]). + +%%% Utilities + +%% When/if erlang:monitor() returns before trying to connect to the +%% other node this function can be removed. +do_monitor(Pid) -> + case (node(Pid) =:= node()) orelse lists:member(node(Pid), nodes()) of + true -> + %% Assume the node is still up + {Pid, erlang:monitor(process, Pid)}; + false -> + F = fun() -> + Ref = erlang:monitor(process, Pid), + receive + {'DOWN', Ref, process, Pid, _Info} -> + exit(normal) + end + end, + erlang:spawn_monitor(F) + end. + +intersection(_, []) -> + []; +intersection(L1, L2) -> + L1 -- (L1 -- L2). |