Skip to content

Commit

Permalink
Merge branch 'Ayanda-D-rabbitmq-server-914' into stable
Browse files Browse the repository at this point in the history
  • Loading branch information
dcorbacho committed Aug 19, 2016
2 parents c633856 + 6fc561f commit d4020f5
Show file tree
Hide file tree
Showing 5 changed files with 74 additions and 8 deletions.
7 changes: 6 additions & 1 deletion src/gm.erl
Original file line number Diff line number Diff line change
Expand Up @@ -760,6 +760,9 @@ handle_info({'DOWN', MRef, process, _Pid, Reason},
end;
handle_info(_, State) ->
%% Discard any unexpected messages, such as late replies from neighbour_call/2
%% TODO: For #gm_group{} related info messages, it could be worthwhile to
%% change_view/2, as this might reflect an alteration in the gm group, meaning
%% we now need to update our state. see rabbitmq-server#914.
noreply(State).

terminate(Reason, #state { module = Module, callback_args = Args }) ->
Expand Down Expand Up @@ -1596,7 +1599,9 @@ check_membership(Self, #gm_group{members = M} = Group) ->
Group;
false ->
throw(lost_membership)
end.
end;
check_membership(_Self, {error, not_found}) ->
throw(lost_membership).

check_membership(GroupName) ->
case dirty_read_group(GroupName) of
Expand Down
9 changes: 9 additions & 0 deletions src/rabbit_mirror_queue_coordinator.erl
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,15 @@ handle_cast({gm_deaths, DeadGMPids},
DeadPids),
rabbit_mirror_queue_misc:add_mirrors(QueueName, ExtraNodes, async),
noreply(State);
{ok, _MPid0, DeadPids, _ExtraNodes} ->
%% see rabbitmq-server#914;
%% Different slave is now master, stop current coordinator normally.
%% Initiating queue is now slave and the least we could do is report
%% deaths which we 'think' we saw.
%% NOTE: Reported deaths here, could be inconsistant.
rabbit_mirror_queue_misc:report_deaths(MPid, false, QueueName,
DeadPids),
{stop, normal, State};
{error, not_found} ->
{stop, normal, State}
end;
Expand Down
16 changes: 13 additions & 3 deletions src/rabbit_mirror_queue_misc.erl
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
rabbit_misc:execute_mnesia_transaction(
fun () ->
%% Someone else could have deleted the queue before we
%% get here.
%% get here. Or, gm group could've altered. see rabbitmq-server#914
case mnesia:read({rabbit_queue, QueueName}) of
[] -> {error, not_found};
[Q = #amqqueue { pid = QPid,
Expand All @@ -90,15 +90,25 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
AlivePids = [Pid || {_GM, Pid} <- AliveGM],
Alive = [Pid || Pid <- [QPid | SPids],
lists:member(Pid, AlivePids)],
{QPid1, SPids1} = promote_slave(Alive),
{QPid1, SPids1} = case Alive of
[] ->
%% GM altered, & if all pids are
%% perceived as dead, rather do
%% do nothing here, & trust the
%% promoted slave to have updated
%% mnesia during the alteration.
{QPid, SPids};
_ -> promote_slave(Alive)
end,
Extra =
case {{QPid, SPids}, {QPid1, SPids1}} of
{Same, Same} ->
[];
_ when QPid =:= QPid1 orelse QPid1 =:= Self ->
%% Either master hasn't changed, so
%% we're ok to update mnesia; or we have
%% become the master.
%% become the master. If gm altered,
%% we have no choice but to proceed.
Q1 = Q#amqqueue{pid = QPid1,
slave_pids = SPids1,
gm_pids = AliveGM},
Expand Down
12 changes: 9 additions & 3 deletions src/rabbit_mirror_queue_slave.erl
Original file line number Diff line number Diff line change
Expand Up @@ -225,9 +225,15 @@ handle_call({gm_deaths, DeadGMPids}, From,
_ ->
%% master has changed to not us
gen_server2:reply(From, ok),
%% assertion, we don't need to add_mirrors/2 in this
%% branch, see last clause in remove_from_queue/2
[] = ExtraNodes,
%% see rabbitmq-server#914;
%% It's not always guaranteed that we won't have ExtraNodes.
%% If gm alters, master can change to not us with extra nodes,
%% in which case we attempt to add mirrors on those nodes.
case ExtraNodes of
[] -> void;
_ -> rabbit_mirror_queue_misc:add_mirrors(
QName, ExtraNodes, async)
end,
%% Since GM is by nature lazy we need to make sure
%% there is some traffic when a master dies, to
%% make sure all slaves get informed of the
Expand Down
38 changes: 37 additions & 1 deletion test/gm_SUITE.erl
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ all() ->
confirmed_broadcast,
member_death,
receive_in_order,
unexpected_msg
unexpected_msg,
down_in_members_change
].

init_per_suite(Config) ->
Expand Down Expand Up @@ -123,6 +124,41 @@ unexpected_msg(_Config) ->
passed
end).

down_in_members_change(_Config) ->
%% Setup
ok = gm:create_tables(),
{ok, Pid} = gm:start_link(?MODULE, ?MODULE, self(),
fun rabbit_misc:execute_mnesia_transaction/1),
passed = receive_joined(Pid, [Pid], timeout_joining_gm_group_1),
{ok, Pid2} = gm:start_link(?MODULE, ?MODULE, self(),
fun rabbit_misc:execute_mnesia_transaction/1),
passed = receive_joined(Pid2, [Pid, Pid2], timeout_joining_gm_group_2),
passed = receive_birth(Pid, Pid2, timeout_waiting_for_birth_2),

%% Test. Simulate that the gm group is deleted (forget_group) while
%% processing the 'DOWN' message from the neighbour
process_flag(trap_exit, true),
ok = meck:new(mnesia, [passthrough]),
ok = meck:expect(mnesia, read, fun({gm_group, ?MODULE}) ->
[];
(Key) ->
meck:passthrough([Key])
end),
gm:leave(Pid2),
Passed = receive
{'EXIT', Pid, normal} ->
passed;
{'EXIT', Pid, _} ->
crashed
after 15000 ->
timeout
end,
%% Cleanup
meck:unload(mnesia),
process_flag(trap_exit, false),
passed = Passed.


do_broadcast(Fun) ->
with_two_members(broadcast_fun(Fun)).

Expand Down

0 comments on commit d4020f5

Please sign in to comment.