Skip to content

Commit b74999d

Browse files
committed
rabbit_quorum_queue: Shrink batches of QQs in parallel
Shrinking a member node off of a QQ can be parallelized. The operation involves * removing the node from the QQ's cluster membership (appending a command to the log and committing it) with `ra:remove_member/3` * updating the metadata store to remove the member from the QQ type state with `rabbit_amqqueue:update/2` * deleting the queue data from the node with `ra:force_delete_server/2` if the node can be reached All of these operations are I/O bound. Updating the cluster membership and metadata store involves appending commands to those logs and replicating them. Writing commands to Ra synchronously in serial is fairly slow - sending many commands in parallel is much more efficient. By parallelizing these steps we can write larger chunks of commands to WAL(s). `ra:force_delete_server/2` benefits from parallelizing if the node being shrunk off is no longer reachable, for example in some hardware failures. The underlying `rpc:call/4` will attempt to auto-connect to the node and this can take some time to time out. By parallelizing this, each `rpc:call/4` reuses the same underlying distribution entry and all calls fail together once the connection fails to establish.
1 parent 2311d30 commit b74999d

1 file changed

Lines changed: 64 additions & 33 deletions

File tree

deps/rabbit/src/rabbit_quorum_queue.erl

Lines changed: 64 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1491,39 +1491,70 @@ delete_member(Q, Node) when ?amqqueue_is_quorum(Q) ->
14911491
{ok, pos_integer()} | {error, pos_integer(), term()}}].
14921492
shrink_all(Node) ->
14931493
?LOG_INFO("Asked to remove all quorum queue replicas from node ~ts", [Node]),
1494-
[begin
1495-
QName = amqqueue:get_name(Q),
1496-
?LOG_INFO("~ts: removing member (replica) on node ~w",
1497-
[rabbit_misc:rs(QName), Node]),
1498-
Size = length(get_nodes(Q)),
1499-
case delete_member(Q, Node) of
1500-
ok ->
1501-
{QName, {ok, Size-1}};
1502-
{error, cluster_change_not_permitted} ->
1503-
%% this could be timing related and due to a new leader just being
1504-
%% elected but it's noop command not been committed yet.
1505-
%% lets sleep and retry once
1506-
?LOG_INFO("~ts: failed to remove member (replica) on node ~w "
1507-
"as cluster change is not permitted. "
1508-
"retrying once in 500ms",
1509-
[rabbit_misc:rs(QName), Node]),
1510-
timer:sleep(500),
1511-
case delete_member(Q, Node) of
1512-
ok ->
1513-
{QName, {ok, Size-1}};
1514-
{error, Err} ->
1515-
?LOG_WARNING("~ts: failed to remove member (replica) on node ~w, error: ~w",
1516-
[rabbit_misc:rs(QName), Node, Err]),
1517-
{QName, {error, Size, Err}}
1518-
end;
1519-
{error, Err} ->
1520-
?LOG_WARNING("~ts: failed to remove member (replica) on node ~w, error: ~w",
1521-
[rabbit_misc:rs(QName), Node, Err]),
1522-
{QName, {error, Size, Err}}
1523-
end
1524-
end || Q <- rabbit_amqqueue:list(),
1525-
amqqueue:get_type(Q) == ?MODULE,
1526-
lists:member(Node, get_nodes(Q))].
1494+
%% This operation is bound by I/O so this default is set high.
1495+
Size = application:get_env(rabbit, quorum_queue_shrink_batch_size, 64),
1496+
Chunks = ra_lib:lists_chunk(Size, [Q || Q <- rabbit_amqqueue:list(),
1497+
amqqueue:get_type(Q) == ?MODULE,
1498+
lists:member(Node, get_nodes(Q))]),
1499+
Parent = self(),
1500+
lists:flatten([begin
1501+
Tasks = [{spawn_monitor(fun() ->
1502+
Res = shrink(Node, Q),
1503+
Parent ! {self(), Res}
1504+
end), amqqueue:get_name(Q)}
1505+
|| Q <- Chunk],
1506+
[receive
1507+
{Pid, Res} ->
1508+
erlang:demonitor(MRef, [flush]),
1509+
{QName, Res};
1510+
{'DOWN', MRef, process, Pid, Reason} ->
1511+
?LOG_WARNING("~ts: failed to remove member "
1512+
"(replica) on node ~w, error: ~w",
1513+
[rabbit_misc:rs(QName), Node,
1514+
Reason]),
1515+
[]
1516+
after
1517+
15_000 ->
1518+
?LOG_WARNING("~ts: failed to remove member "
1519+
"(replica) on node ~w within 15 "
1520+
"seconds",
1521+
[rabbit_misc:rs(QName), Node]),
1522+
[]
1523+
end || {{Pid, MRef}, QName} <- Tasks]
1524+
end || Chunk <- Chunks]).
1525+
1526+
-spec shrink(node(), amqqueue:amqqueue()) ->
1527+
{ok, Size} | {error, Size, term()} when Size :: pos_integer().
1528+
shrink(Node, Q) ->
1529+
QName = amqqueue:get_name(Q),
1530+
?LOG_INFO("~ts: removing member (replica) on node ~w",
1531+
[rabbit_misc:rs(QName), Node]),
1532+
Size = length(get_nodes(Q)),
1533+
case delete_member(Q, Node) of
1534+
ok ->
1535+
{ok, Size-1};
1536+
{error, cluster_change_not_permitted} ->
1537+
%% this could be timing related and due to a new leader just being
1538+
%% elected but it's noop command not been committed yet.
1539+
%% lets sleep and retry once
1540+
?LOG_INFO("~ts: failed to remove member (replica) on node ~w "
1541+
"as cluster change is not permitted. "
1542+
"retrying once in 500ms",
1543+
[rabbit_misc:rs(QName), Node]),
1544+
timer:sleep(500),
1545+
case delete_member(Q, Node) of
1546+
ok ->
1547+
{ok, Size-1};
1548+
{error, Err} ->
1549+
?LOG_WARNING("~ts: failed to remove member (replica) on node ~w, error: ~w",
1550+
[rabbit_misc:rs(QName), Node, Err]),
1551+
{error, Size, Err}
1552+
end;
1553+
{error, Err} ->
1554+
?LOG_WARNING("~ts: failed to remove member (replica) on node ~w, error: ~w",
1555+
[rabbit_misc:rs(QName), Node, Err]),
1556+
{error, Size, Err}
1557+
end.
15271558

15281559
-spec grow(node() | integer(), binary(), binary(), all | even) ->
15291560
[{rabbit_amqqueue:name(),

0 commit comments

Comments
 (0)