Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Ra partial read #12805

Draft
wants to merge 4 commits into
base: main
Choose a base branch
from
+71 −28
Draft

Ra partial read #12805

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3aa01e5
wip
kjnilsson Nov 13, 2024
ef9260e
try
kjnilsson Nov 19, 2024
1b973d2
wip
kjnilsson Nov 21, 2024
a751df9
wip
kjnilsson Nov 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion deps/rabbit/src/rabbit_channel.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -564,8 +564,10 @@ prioritise_cast(Msg, _Len, _State) ->
case Msg of
{confirm, _MsgSeqNos, _QPid} -> 5;
{reject_publish, _MsgSeqNos, _QPid} -> 5;
{queue_event, _, {confirm, _MsgSeqNos, _QPid}} -> 5;
% {queue_event, _, {confirm, _MsgSeqNos, _QPid}} -> 5;
{queue_event, _, {reject_publish, _MsgSeqNos, _QPid}} -> 5;
{method, #'basic.ack'{}, _Content, _Flow} -> 5;
% {queue_event, _, {delivery, _, _}} -> 0;
_ -> 0
end.

Expand Down
76 changes: 54 additions & 22 deletions deps/rabbit/src/rabbit_fifo.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,10 @@
make_purge/0,
make_purge_nodes/1,
make_update_config/1,
make_garbage_collection/0
make_garbage_collection/0,

exec_read/3

]).

-ifdef(TEST).
Expand Down Expand Up @@ -2058,31 +2061,52 @@ delivery_effect(ConsumerKey, [{MsgId, ?MSG(Idx, Header)}],
{send_msg, CPid, {delivery, CTag, [{MsgId, {Header, RawMsg}}]},
?DELIVERY_SEND_MSG_OPTS};
delivery_effect(ConsumerKey, Msgs,
#?STATE{cfg = #cfg{resource = QR}} = State) ->
#?STATE{cfg = #cfg{resource = _QR}} = State) ->
{CTag, CPid} = consumer_id(ConsumerKey, State),
{RaftIdxs, Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
{[I | Acc], N+1}
end, {[], 0}, Msgs),
{log, RaftIdxs,
fun (Commands)
when length(Commands) < Num ->
%% the mandatory length/1 guard is a bit :(
rabbit_log:info("~ts: requested read consumer tag '~ts' of ~b "
"indexes ~w but only ~b were returned. "
"This is most likely a stale read request "
"and can be ignored",
[rabbit_misc:rs(QR), CTag, Num, RaftIdxs,
length(Commands)]),
[];
(Commands) ->
DelMsgs = lists:zipwith(
fun (Cmd, {MsgId, ?MSG(_Idx, Header)}) ->
{MsgId, {Header, get_msg(Cmd)}}
end, Commands, Msgs),
[{send_msg, CPid, {delivery, CTag, DelMsgs},
{RaftIdxs, _Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
{[I | Acc], N+1}
end, {[], 0}, Msgs),
rabbit_log:debug("DELIVERY EFFECT created for ~s ~p!", [CTag, RaftIdxs]),
{log_ext, RaftIdxs,
fun
% (Commands)
% when length(Commands) < Num ->
% %% the mandatory length/1 guard is a bit :(
% rabbit_log:info("~ts: requested read consumer tag '~ts' of ~b "
% "indexes ~w but only ~b were returned. "
% "This is most likely a stale read request "
% "and can be ignored",
% [rabbit_misc:rs(QR), CTag, Num, RaftIdxs,
% length(Commands)]),
% [];
(ReadPlan) ->
rabbit_log:debug("READPLAN created for ~s ~p!", [CTag, RaftIdxs]),
%% TODO: check if CPid is local or not
%% TODO: could consider introducing a leader local proxy process
[{send_msg, CPid, {delivery, CTag, ReadPlan, Msgs},
?DELIVERY_SEND_MSG_OPTS}]
end,
{local, node(CPid)}}.
% {log, RaftIdxs,
% fun (Commands)
% when length(Commands) < Num ->
% %% the mandatory length/1 guard is a bit :(
% rabbit_log:info("~ts: requested read consumer tag '~ts' of ~b "
% "indexes ~w but only ~b were returned. "
% "This is most likely a stale read request "
% "and can be ignored",
% [rabbit_misc:rs(QR), CTag, Num, RaftIdxs,
% length(Commands)]),
% [];
% (Commands) ->
% DelMsgs = lists:zipwith(
% fun (Cmd, {MsgId, ?MSG(_Idx, Header)}) ->
% {MsgId, {Header, get_msg(Cmd)}}
% end, Commands, Msgs),
% [{send_msg, CPid, {delivery, CTag, DelMsgs},
% ?DELIVERY_SEND_MSG_OPTS}]
% end,
% {local, node(CPid)}}.

reply_log_effect(RaftIdx, MsgId, Header, Ready, From) ->
{log, [RaftIdx],
Expand Down Expand Up @@ -2995,3 +3019,11 @@ incr_msg(Msg0, DelFailed, Anns) ->
false ->
Msg2
end.

exec_read(Flru0, ReadPlan, Msgs) ->
{Entries, Flru} = ra_log_read_plan:execute(ReadPlan, Flru0),
%% return a list in original order
{lists:map(fun ({MsgId, ?MSG(Idx, Header)}) ->
Cmd = maps:get(Idx, Entries),
{MsgId, {Header, get_msg(Cmd)}}
end, Msgs), Flru}.
15 changes: 11 additions & 4 deletions deps/rabbit/src/rabbit_fifo_client.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,8 @@
pending = #{} :: #{seq() =>
{term(), rabbit_fifo:command()}},
consumers = #{} :: #{rabbit_types:ctag() => #consumer{}},
timer_state :: term()
timer_state :: term(),
cached_segments :: undefined | ra_flru:state()
}).

-opaque state() :: #state{}.
Expand Down Expand Up @@ -167,7 +168,7 @@ enqueue(QName, Correlation, Msg,
%% @param QueueName Name of the queue.
%% @param Msg an arbitrary erlang term representing the message.
%% @param State the current {@module} state.
%% @returns
%% @return's
%% `{ok, State, Actions}' if the command was successfully sent.
%% {@module} assigns a sequence number to every raft command it issues. The
%% SequenceNumber can be correlated to the applied sequence numbers returned
Expand Down Expand Up @@ -633,7 +634,8 @@ handle_ra_event(QName, From, {applied, Seqs},
_ ->
{ok, State1, Actions}
end;
handle_ra_event(QName, From, {machine, {delivery, _ConsumerTag, _} = Del}, State0) ->
handle_ra_event(QName, From, {machine, Del}, State0)
when element(1, Del) == delivery ->
handle_delivery(QName, From, Del, State0);
handle_ra_event(_QName, _From, {machine, Action}, State)
when element(1, Action) =:= credit_reply orelse
Expand Down Expand Up @@ -835,7 +837,12 @@ handle_delivery(_QName, _Leader, {delivery, Tag, [_ | _] = IdMsgs},
%% we should return all messages.
MsgIntIds = [Id || {Id, _} <- IdMsgs],
{State1, Deliveries} = return(Tag, MsgIntIds, State0),
{ok, State1, Deliveries}.
{ok, State1, Deliveries};
handle_delivery(QName, Leader, {delivery, Tag, ReadState, Msgs},
#state{cached_segments = Cached0} = State) ->
{MsgIds, Cached} = rabbit_fifo:exec_read(Cached0, ReadState, Msgs),
handle_delivery(QName, Leader, {delivery, Tag, MsgIds},
State#state{cached_segments = Cached}).

transform_msgs(QName, QRef, Msgs) ->
lists:map(
Expand Down
1 change: 1 addition & 0 deletions deps/rabbit/test/quorum_queue_SUITE.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3632,6 +3632,7 @@ receive_and_ack(Ch) ->
amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag,
multiple = false})
after 5000 ->
flush(1),
ct:fail("receive_and_ack timed out", [])
end.

Expand Down
3 changes: 2 additions & 1 deletion rabbitmq-components.mk
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ dep_khepri = hex 0.16.0
dep_khepri_mnesia_migration = hex 0.7.0
dep_osiris = git https://github.com/rabbitmq/osiris v1.8.4
dep_prometheus = hex 4.11.0
dep_ra = hex 2.14.0
dep_ra = git https://github.com/rabbitmq/ra partial-read-api
# dep_ra = hex 2.14.0
dep_ranch = hex 2.1.0
dep_recon = hex 2.5.6
dep_redbug = hex 2.0.7
Expand Down
Loading