Skip to content

Commit ea57c83

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 dd96339 commit ea57c83

File tree

1 file changed

+64
-33
lines changed

1 file changed

+64
-33
lines changed

deps/rabbit/src/rabbit_quorum_queue.erl

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

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

0 commit comments

Comments
 (0)