diff options
| -rw-r--r-- | src/mirrored_supervisor.erl | 188 | ||||
| -rw-r--r-- | src/pg2_fixed.erl | 388 | ||||
| -rw-r--r-- | src/rabbit_mnesia.erl | 3 | ||||
| -rw-r--r-- | src/rabbit_upgrade_functions.erl | 5 |
4 files changed, 533 insertions, 51 deletions
diff --git a/src/mirrored_supervisor.erl b/src/mirrored_supervisor.erl index 91b4bd74d9..b7e11464c0 100644 --- a/src/mirrored_supervisor.erl +++ b/src/mirrored_supervisor.erl @@ -16,15 +16,24 @@ -module(mirrored_supervisor). --define(SUPERVISOR, supervisor2). --define(GEN_SERVER, gen_server2). --define(TABLE, ?MODULE). %% TODO documentation %% We need a thing like a supervisor, except that it joins something %% like a process group, and if a child process dies it can be %% restarted under another supervisor (probably on another node). +-define(SUPERVISOR, supervisor2). +-define(GEN_SERVER, gen_server2). +-define(ETS_TABLE, ?MODULE). +-define(ID, ?MODULE). + +-define(MNESIA_TABLE_NAME, mirrored_sup_childspec). +-define(MNESIA_TABLE, + {?MNESIA_TABLE_NAME, + [{record_name, mirrored_sup_childspec}, + {attributes, record_info(fields, mirrored_sup_childspec)}]}). +-define(MNESIA_TABLE_MATCH, {match, #mirrored_sup_childspec{ _ = '_' }}). + -export([start_link/2,start_link/3, start_child/2, restart_child/2, delete_child/2, terminate_child/2, @@ -39,13 +48,15 @@ handle_cast/2]). -export([start_internal/2]). +-export([create_tables/0, table_definitions/0]). --record(state, {}). +-record(mirrored_sup_childspec, {id, sup_pid, childspec}). -%%---------------------------------------------------------------------------- +-record(state, {name}). --define(ID, ?MODULE). +%%---------------------------------------------------------------------------- +%% TODO this is going to make testing awkward. Maybe we need a local name and a group name? start_link(_Mod, _Args) -> exit(mirrored_supervisors_must_be_locally_named). @@ -59,49 +70,97 @@ start_link({local, SupName}, Mod, Args) -> start_link({_, _SupName}, _Mod, _Args) -> exit(mirrored_supervisors_must_be_locally_named). -start_child(Sup, ChildSpec) -> call(Sup, {start_child, [Sup, ChildSpec]}). -restart_child(Sup, Name) -> call(Sup, {restart_child, [Sup, Name]}). -delete_child(Sup, Name) -> call(Sup, {delete_child, [Sup, Name]}). -terminate_child(Sup, Name) -> call(Sup, {terminate_child, [Sup, Name]}). -which_children(Sup) -> call(Sup, {which_children, [Sup]}). -find_child(Sup, Name) -> call(Sup, {find_child, [Sup, Name]}). +start_child(Sup, ChildSpec) -> call(Sup, {start_child, ChildSpec}). +delete_child(Sup, Name) -> call(Sup, {delete_child, Name}). +restart_child(Sup, Name) -> call(Sup, {msg, restart_child, [Sup, Name]}). +terminate_child(Sup, Name) -> call(Sup, {msg, terminate_child, [Sup, Name]}). +which_children(Sup) -> call(Sup, {msg, which_children, [Sup]}). +find_child(Sup, Name) -> call(Sup, {msg, find_child, [Sup, Name]}). check_childspecs(ChildSpecs) -> ?SUPERVISOR:check_childspecs(ChildSpecs). behaviour_info(callbacks) -> [{init,1}]; behaviour_info(_Other) -> undefined. call(SupName, Msg) -> - [{SupName, Pid}] = ets:lookup(?TABLE, SupName), - ?GEN_SERVER:call(Pid, {sup_msg, Msg}, infinity). + [{SupName, Pid}] = ets:lookup(?ETS_TABLE, SupName), + ?GEN_SERVER:call(Pid, Msg, infinity). %%---------------------------------------------------------------------------- start_internal(SupName, Args) -> - {ok, Pid} = ?GEN_SERVER:start_link(?MODULE, Args, [{timeout, infinity}]), - Ins = fun() -> true = ets:insert(?TABLE, {SupName, Pid}) end, + {ok, Pid} = ?GEN_SERVER:start_link(?MODULE, {SupName, Args}, + [{timeout, infinity}]), + Ins = fun() -> true = ets:insert(?ETS_TABLE, {SupName, Pid}) end, try Ins() - catch error:badarg -> ets:new(?TABLE, [named_table]), + catch error:badarg -> ets:new(?ETS_TABLE, [named_table]), Ins() end, {ok, Pid}. %%---------------------------------------------------------------------------- -init(_Args) -> - {ok, #state{}}. - -handle_call({sup_msg, {F, A}}, _From, State) -> +init({SupName, _Args}) -> + pg2_fixed:create(SupName), + [begin + io:format("Announce to ~p~n", [Pid]), + gen_server2:call(Pid, {hello, self()}, infinity), + erlang:monitor(process, Pid) + end + || Pid <- pg2_fixed:get_members(SupName)], + ok = pg2_fixed:join(SupName, self()), + {ok, #state{name = SupName}}. + +handle_call({start_child, ChildSpec}, _From, State = #state{name = SupName}) -> + {reply, case mnesia:transaction(fun() -> check_start(ChildSpec) end) of + {atomic, start} -> io:format("Start ~p~n", [id(ChildSpec)]), + apply(?SUPERVISOR, + start_child, [SupName, ChildSpec]); + {atomic, already} -> io:format("Already ~p~n", [id(ChildSpec)]), + {ok, already} + end, State}; + +handle_call({delete_child, ChildSpec}, _From, + State = #state{name = SupName}) -> + {atomic, ok} = mnesia:transaction(fun() -> delete(ChildSpec) end), + {reply, apply(?SUPERVISOR, delete_child, [SupName, id(ChildSpec)]), State}; + +handle_call({msg, F, A}, _From, State) -> {reply, apply(?SUPERVISOR, F, A), State}; -handle_call(Msg, _From, State) -> - {reply, {unexpected_call, Msg}, State}. +handle_call({hello, Pid}, _From, State) -> + io:format("Hello from ~p~n", [Pid]), + erlang:monitor(process, Pid), + {reply, ok, State}; -handle_cast(_Msg, State) -> - {noreply, State}. +handle_call(alive, _From, State) -> + {reply, true, State}; -handle_info(_Info, State) -> - {noreply, State}. +handle_call(Msg, _From, State) -> + {stop, {unexpected_call, Msg}, State}. + +handle_cast(Msg, State) -> + {stop, {unexpected_cast, Msg}, State}. + +handle_info({'DOWN', _Ref, process, Pid, _Reason}, + State = #state{name = SupName}) -> + io:format("Pid ~p down!~n", [Pid]), + %% TODO load balance this + Self = self(), + case lists:sort(pg2_fixed:get_members(SupName)) of + [Self | _] -> {atomic, ChildSpecs} = + mnesia:transaction(fun() -> restart_all(Pid) end), + [begin + apply(?SUPERVISOR, start_child, + [SupName, ChildSpec]), + io:format("Restarted ~p~n", [id(ChildSpec)]) + end || ChildSpec <- ChildSpecs]; + _ -> ok + end, + {noreply, State}; + +handle_info(Info, State) -> + {stop, {unexpected_info, Info}, State}. terminate(_Reason, _State) -> ok. @@ -109,32 +168,61 @@ terminate(_Reason, _State) -> code_change(_OldVsn, State, _Extra) -> {ok, State}. +%%---------------------------------------------------------------------------- +check_start(ChildSpec) -> + case mnesia:wread({?MNESIA_TABLE_NAME, id(ChildSpec)}) of + [] -> write(ChildSpec), + start; + [S] -> #mirrored_sup_childspec{sup_pid = Pid} = S, + case alive(Pid) of + true -> already; %% TODO return real pid? + false -> delete(ChildSpec), + write(ChildSpec), + start + end + end. + +alive(Pid) -> + gen_server:call(Pid, alive, infinity). + +write(ChildSpec) -> + ok = mnesia:write(#mirrored_sup_childspec{id = id(ChildSpec), + sup_pid = self(), + childspec = ChildSpec}). + +delete(ChildSpec) -> + ok = mnesia:delete({?MNESIA_TABLE_NAME, id(ChildSpec)}). + +id({Id, _, _, _, _, _}) -> Id. + +restart_all(OldPid) -> + MatchHead = #mirrored_sup_childspec{sup_pid = OldPid, + childspec = '$1', + _ = '_'}, + [begin + delete(ChildSpec), + write(ChildSpec), + ChildSpec + end || ChildSpec <- + mnesia:select(?MNESIA_TABLE_NAME, [{MatchHead, [], ['$1']}])]. +%%---------------------------------------------------------------------------- -%%-export([create_tables/0, table_definitions/0]). - -%% -define(TABLE, {?GROUP_TABLE, [{record_name, gm_group}, -%% {attributes, record_info(fields, gm_group)}]}). -%% -define(TABLE_MATCH, {match, #gm_group { _ = '_' }}). - -%% -define(GROUP_TABLE, gm_group). - - -%% create_tables() -> -%% create_tables([?TABLE]). - -%% create_tables([]) -> -%% ok; -%% create_tables([{Table, Attributes} | Tables]) -> -%% case mnesia:create_table(Table, Attributes) of -%% {atomic, ok} -> create_tables(Tables); -%% {aborted, {already_exists, gm_group}} -> create_tables(Tables); -%% Err -> Err -%% end. +create_tables() -> + create_tables([?MNESIA_TABLE]). -%% table_definitions() -> -%% {Name, Attributes} = ?TABLE, -%% [{Name, [?TABLE_MATCH | Attributes]}]. +create_tables([]) -> + ok; +create_tables([{Table, Attributes} | Ts]) -> + case mnesia:create_table(Table, Attributes) of + {atomic, ok} -> create_tables(Ts); + {aborted, {already_exists, ?MNESIA_TABLE_NAME}} -> create_tables(Ts); + Err -> Err + end. +table_definitions() -> + {Name, Attributes} = ?MNESIA_TABLE, + [{Name, [?MNESIA_TABLE_MATCH | Attributes]}]. +%%---------------------------------------------------------------------------- diff --git a/src/pg2_fixed.erl b/src/pg2_fixed.erl new file mode 100644 index 0000000000..224715eb70 --- /dev/null +++ b/src/pg2_fixed.erl @@ -0,0 +1,388 @@ +%% This is the version of pg2 from R14B02, which contains the fix +%% described at +%% http://erlang.2086793.n4.nabble.com/pg2-still-busted-in-R13B04-td2230601.html. +%% The only changes are a search-and-replace to rename the module and +%% avoid clashes with other versions of pg2. + + +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1997-2010. 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(pg2_fixed). + +-export([create/1, delete/1, join/2, leave/2]). +-export([get_members/1, get_local_members/1]). +-export([get_closest_pid/1, which_groups/0]). +-export([start/0,start_link/0,init/1,handle_call/3,handle_cast/2,handle_info/2, + terminate/2]). + +%%% As of R13B03 monitors are used instead of links. + +%%% +%%% Exported functions +%%% + +-spec start_link() -> {'ok', pid()} | {'error', term()}. + +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +-spec start() -> {'ok', pid()} | {'error', term()}. + +start() -> + ensure_started(). + +-spec create(term()) -> 'ok'. + +create(Name) -> + ensure_started(), + case ets:member(pg2_fixed_table, {group, Name}) of + false -> + global:trans({{?MODULE, Name}, self()}, + fun() -> + gen_server:multi_call(?MODULE, {create, Name}) + end), + ok; + true -> + ok + end. + +-type name() :: term(). + +-spec delete(name()) -> 'ok'. + +delete(Name) -> + ensure_started(), + global:trans({{?MODULE, Name}, self()}, + fun() -> + gen_server:multi_call(?MODULE, {delete, Name}) + end), + ok. + +-spec join(name(), pid()) -> 'ok' | {'error', {'no_such_group', term()}}. + +join(Name, Pid) when is_pid(Pid) -> + ensure_started(), + case ets:member(pg2_fixed_table, {group, Name}) of + false -> + {error, {no_such_group, Name}}; + true -> + global:trans({{?MODULE, Name}, self()}, + fun() -> + gen_server:multi_call(?MODULE, + {join, Name, Pid}) + end), + ok + end. + +-spec leave(name(), pid()) -> 'ok' | {'error', {'no_such_group', name()}}. + +leave(Name, Pid) when is_pid(Pid) -> + ensure_started(), + case ets:member(pg2_fixed_table, {group, Name}) of + false -> + {error, {no_such_group, Name}}; + true -> + global:trans({{?MODULE, Name}, self()}, + fun() -> + gen_server:multi_call(?MODULE, + {leave, Name, Pid}) + end), + ok + end. + +-type get_members_ret() :: [pid()] | {'error', {'no_such_group', name()}}. + +-spec get_members(name()) -> get_members_ret(). + +get_members(Name) -> + ensure_started(), + case ets:member(pg2_fixed_table, {group, Name}) of + true -> + group_members(Name); + false -> + {error, {no_such_group, Name}} + end. + +-spec get_local_members(name()) -> get_members_ret(). + +get_local_members(Name) -> + ensure_started(), + case ets:member(pg2_fixed_table, {group, Name}) of + true -> + local_group_members(Name); + false -> + {error, {no_such_group, Name}} + end. + +-spec which_groups() -> [name()]. + +which_groups() -> + ensure_started(), + all_groups(). + +-type gcp_error_reason() :: {'no_process', term()} | {'no_such_group', term()}. + +-spec get_closest_pid(term()) -> pid() | {'error', gcp_error_reason()}. + +get_closest_pid(Name) -> + case get_local_members(Name) of + [Pid] -> + Pid; + [] -> + {_,_,X} = erlang:now(), + case get_members(Name) of + [] -> {error, {no_process, Name}}; + Members -> + lists:nth((X rem length(Members))+1, Members) + end; + Members when is_list(Members) -> + {_,_,X} = erlang:now(), + lists:nth((X rem length(Members))+1, Members); + Else -> + Else + end. + +%%% +%%% Callback functions from gen_server +%%% + +-record(state, {}). + +-spec init([]) -> {'ok', #state{}}. + +init([]) -> + Ns = nodes(), + net_kernel:monitor_nodes(true), + lists:foreach(fun(N) -> + {?MODULE, N} ! {new_pg2_fixed, node()}, + self() ! {nodeup, N} + end, Ns), + pg2_fixed_table = ets:new(pg2_fixed_table, [ordered_set, protected, named_table]), + {ok, #state{}}. + +-type call() :: {'create', name()} + | {'delete', name()} + | {'join', name(), pid()} + | {'leave', name(), pid()}. + +-spec handle_call(call(), _, #state{}) -> + {'reply', 'ok', #state{}}. + +handle_call({create, Name}, _From, S) -> + assure_group(Name), + {reply, ok, S}; +handle_call({join, Name, Pid}, _From, S) -> + ets:member(pg2_fixed_table, {group, Name}) andalso join_group(Name, Pid), + {reply, ok, S}; +handle_call({leave, Name, Pid}, _From, S) -> + ets:member(pg2_fixed_table, {group, Name}) andalso leave_group(Name, Pid), + {reply, ok, S}; +handle_call({delete, Name}, _From, S) -> + delete_group(Name), + {reply, ok, S}; +handle_call(Request, From, S) -> + error_logger:warning_msg("The pg2_fixed server received an unexpected message:\n" + "handle_call(~p, ~p, _)\n", + [Request, From]), + {noreply, S}. + +-type all_members() :: [[name(),...]]. +-type cast() :: {'exchange', node(), all_members()} + | {'del_member', name(), pid()}. + +-spec handle_cast(cast(), #state{}) -> {'noreply', #state{}}. + +handle_cast({exchange, _Node, List}, S) -> + store(List), + {noreply, S}; +handle_cast(_, S) -> + %% Ignore {del_member, Name, Pid}. + {noreply, S}. + +-spec handle_info(tuple(), #state{}) -> {'noreply', #state{}}. + +handle_info({'DOWN', MonitorRef, process, _Pid, _Info}, S) -> + member_died(MonitorRef), + {noreply, S}; +handle_info({nodeup, Node}, S) -> + gen_server:cast({?MODULE, Node}, {exchange, node(), all_members()}), + {noreply, S}; +handle_info({new_pg2_fixed, Node}, S) -> + gen_server:cast({?MODULE, Node}, {exchange, node(), all_members()}), + {noreply, S}; +handle_info(_, S) -> + {noreply, S}. + +-spec terminate(term(), #state{}) -> 'ok'. + +terminate(_Reason, _S) -> + true = ets:delete(pg2_fixed_table), + ok. + +%%% +%%% Local functions +%%% + +%%% One ETS table, pg2_fixed_table, is used for bookkeeping. The type of the +%%% table is ordered_set, and the fast matching of partially +%%% instantiated keys is used extensively. +%%% +%%% {{group, Name}} +%%% Process group Name. +%%% {{ref, Pid}, RPid, MonitorRef, Counter} +%%% {{ref, MonitorRef}, Pid} +%%% Each process has one monitor. Sometimes a process is spawned to +%%% monitor the pid (RPid). Counter is incremented when the Pid joins +%%% some group. +%%% {{member, Name, Pid}, GroupCounter} +%%% {{local_member, Name, Pid}} +%%% Pid is a member of group Name, GroupCounter is incremented when the +%%% Pid joins the group Name. +%%% {{pid, Pid, Name}} +%%% Pid is a member of group Name. + +store(List) -> + _ = [(assure_group(Name) + andalso + [join_group(Name, P) || P <- Members -- group_members(Name)]) || + [Name, Members] <- List], + ok. + +assure_group(Name) -> + Key = {group, Name}, + ets:member(pg2_fixed_table, Key) orelse true =:= ets:insert(pg2_fixed_table, {Key}). + +delete_group(Name) -> + _ = [leave_group(Name, Pid) || Pid <- group_members(Name)], + true = ets:delete(pg2_fixed_table, {group, Name}), + ok. + +member_died(Ref) -> + [{{ref, Ref}, Pid}] = ets:lookup(pg2_fixed_table, {ref, Ref}), + Names = member_groups(Pid), + _ = [leave_group(Name, P) || + Name <- Names, + P <- member_in_group(Pid, Name)], + %% Kept for backward compatibility with links. Can be removed, eventually. + _ = [gen_server:abcast(nodes(), ?MODULE, {del_member, Name, Pid}) || + Name <- Names], + ok. + +join_group(Name, Pid) -> + Ref_Pid = {ref, Pid}, + try _ = ets:update_counter(pg2_fixed_table, Ref_Pid, {4, +1}) + catch _:_ -> + {RPid, Ref} = do_monitor(Pid), + true = ets:insert(pg2_fixed_table, {Ref_Pid, RPid, Ref, 1}), + true = ets:insert(pg2_fixed_table, {{ref, Ref}, Pid}) + end, + Member_Name_Pid = {member, Name, Pid}, + try _ = ets:update_counter(pg2_fixed_table, Member_Name_Pid, {2, +1, 1, 1}) + catch _:_ -> + true = ets:insert(pg2_fixed_table, {Member_Name_Pid, 1}), + _ = [ets:insert(pg2_fixed_table, {{local_member, Name, Pid}}) || + node(Pid) =:= node()], + true = ets:insert(pg2_fixed_table, {{pid, Pid, Name}}) + end. + +leave_group(Name, Pid) -> + Member_Name_Pid = {member, Name, Pid}, + try ets:update_counter(pg2_fixed_table, Member_Name_Pid, {2, -1, 0, 0}) of + N -> + if + N =:= 0 -> + true = ets:delete(pg2_fixed_table, {pid, Pid, Name}), + _ = [ets:delete(pg2_fixed_table, {local_member, Name, Pid}) || + node(Pid) =:= node()], + true = ets:delete(pg2_fixed_table, Member_Name_Pid); + true -> + ok + end, + Ref_Pid = {ref, Pid}, + case ets:update_counter(pg2_fixed_table, Ref_Pid, {4, -1}) of + 0 -> + [{Ref_Pid,RPid,Ref,0}] = ets:lookup(pg2_fixed_table, Ref_Pid), + true = ets:delete(pg2_fixed_table, {ref, Ref}), + true = ets:delete(pg2_fixed_table, Ref_Pid), + true = erlang:demonitor(Ref, [flush]), + kill_monitor_proc(RPid, Pid); + _ -> + ok + end + catch _:_ -> + ok + end. + +all_members() -> + [[G, group_members(G)] || G <- all_groups()]. + +group_members(Name) -> + [P || + [P, N] <- ets:match(pg2_fixed_table, {{member, Name, '$1'},'$2'}), + _ <- lists:seq(1, N)]. + +local_group_members(Name) -> + [P || + [Pid] <- ets:match(pg2_fixed_table, {{local_member, Name, '$1'}}), + P <- member_in_group(Pid, Name)]. + +member_in_group(Pid, Name) -> + case ets:lookup(pg2_fixed_table, {member, Name, Pid}) of + [] -> []; + [{{member, Name, Pid}, N}] -> + lists:duplicate(N, Pid) + end. + +member_groups(Pid) -> + [Name || [Name] <- ets:match(pg2_fixed_table, {{pid, Pid, '$1'}})]. + +all_groups() -> + [N || [N] <- ets:match(pg2_fixed_table, {{group,'$1'}})]. + +ensure_started() -> + case whereis(?MODULE) of + undefined -> + C = {pg2_fixed, {?MODULE, start_link, []}, permanent, + 1000, worker, [?MODULE]}, + supervisor:start_child(kernel_safe_sup, C); + Pg2_FixedPid -> + {ok, Pg2_FixedPid} + end. + + +kill_monitor_proc(RPid, Pid) -> + RPid =:= Pid orelse exit(RPid, kill). + +%% 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. diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index 8d5c86464c..463da98c60 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -242,7 +242,8 @@ table_definitions() -> [{record_name, amqqueue}, {attributes, record_info(fields, amqqueue)}, {match, #amqqueue{name = queue_name_match(), _='_'}}]}] - ++ gm:table_definitions(). + ++ gm:table_definitions() + ++ mirrored_supervisor:table_definitions(). binding_match() -> #binding{source = exchange_name_match(), diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl index 0f7a781043..15887af1d7 100644 --- a/src/rabbit_upgrade_functions.erl +++ b/src/rabbit_upgrade_functions.erl @@ -32,6 +32,7 @@ -rabbit_upgrade({user_admin_to_tags, mnesia, [user_to_internal_user]}). -rabbit_upgrade({ha_mirrors, mnesia, []}). -rabbit_upgrade({gm, mnesia, []}). +-rabbit_upgrade({mirrored_supervisor, mnesia, []}). %% ------------------------------------------------------------------- @@ -155,6 +156,10 @@ gm() -> create(gm_group, [{record_name, gm_group}, {attributes, [name, version, members]}]). +mirrored_supervisor() -> + create(mirrored_sup_childspec, [{record_name, mirrored_sup_childspec}, + {attributes, [id, sup_pid, childspec]}]). + %%-------------------------------------------------------------------- transform(TableName, Fun, FieldList) -> |