Skip to content

4.x: Additional CQv2 message store optimisations #11112

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

Merged
merged 4 commits into from
Jun 17, 2024
Merged

4.x: Additional CQv2 message store optimisations #11112

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
18acc01
CQ: Make dirty recovery of shared store more efficient
May 6, 2024
d45fbc3
CQ: Write large messages into their own files
May 14, 2024
41ce4da
CQ: Remove ability to change shared store index module
lhoguin Jun 11, 2024
cc73b86
CQ: Remove rabbit_msg_store.hrl
lhoguin Jun 11, 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
1 change: 0 additions & 1 deletion deps/rabbit/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,6 @@ _APP_ENV = """[
{vm_memory_calculation_strategy, rss},
{memory_monitor_interval, 2500},
{disk_free_limit, 50000000}, %% 50MB
{msg_store_index_module, rabbit_msg_store_ets_index},
{backing_queue_module, rabbit_variable_queue},
%% 0 ("no limit") would make a better default, but that
%% breaks the QPid Java client
Expand Down
1 change: 0 additions & 1 deletion deps/rabbit/Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ define PROJECT_ENV
{vm_memory_calculation_strategy, rss},
{memory_monitor_interval, 2500},
{disk_free_limit, 50000000}, %% 50MB
{msg_store_index_module, rabbit_msg_store_ets_index},
{backing_queue_module, rabbit_variable_queue},
%% 0 ("no limit") would make a better default, but that
%% breaks the QPid Java client
Expand Down
3 changes: 0 additions & 3 deletions deps/rabbit/app.bzl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,6 @@ def all_beam_files(name = "all_beam_files"):
"src/rabbit_mirror_queue_misc.erl",
"src/rabbit_mnesia.erl",
"src/rabbit_msg_store.erl",
"src/rabbit_msg_store_ets_index.erl",
"src/rabbit_msg_store_gc.erl",
"src/rabbit_networking.erl",
"src/rabbit_networking_store.erl",
Expand Down Expand Up @@ -431,7 +430,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
"src/rabbit_mirror_queue_misc.erl",
"src/rabbit_mnesia.erl",
"src/rabbit_msg_store.erl",
"src/rabbit_msg_store_ets_index.erl",
"src/rabbit_msg_store_gc.erl",
"src/rabbit_networking.erl",
"src/rabbit_networking_store.erl",
Expand Down Expand Up @@ -711,7 +709,6 @@ def all_srcs(name = "all_srcs"):
"src/rabbit_mirror_queue_misc.erl",
"src/rabbit_mnesia.erl",
"src/rabbit_msg_store.erl",
"src/rabbit_msg_store_ets_index.erl",
"src/rabbit_msg_store_gc.erl",
"src/rabbit_networking.erl",
"src/rabbit_networking_store.erl",
Expand Down
20 changes: 13 additions & 7 deletions deps/rabbit/src/rabbit_classic_queue_index_v2.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1125,8 +1125,11 @@ queue_index_walker({next, Gatherer}) when is_pid(Gatherer) ->
empty ->
ok = gatherer:stop(Gatherer),
finished;
%% From v1 index walker. @todo Remove when no longer possible to convert from v1.
{value, {MsgId, Count}} ->
{MsgId, Count, {next, Gatherer}}
{MsgId, Count, {next, Gatherer}};
{value, MsgIds} ->
{MsgIds, {next, Gatherer}}
end.

queue_index_walker_reader(#resource{ virtual_host = VHost } = Name, Gatherer) ->
Expand All @@ -1153,27 +1156,30 @@ queue_index_walker_segment(F, Gatherer) ->
{ok, <<?MAGIC:32,?VERSION:8,
FromSeqId:64/unsigned,ToSeqId:64/unsigned,
_/bits>>} ->
queue_index_walker_segment(Fd, Gatherer, 0, ToSeqId - FromSeqId);
queue_index_walker_segment(Fd, Gatherer, 0, ToSeqId - FromSeqId, []);
_ ->
%% Invalid segment file. Skip.
ok
end,
ok = file:close(Fd).

queue_index_walker_segment(_, _, N, N) ->
queue_index_walker_segment(_, Gatherer, N, N, Acc) ->
%% We reached the end of the segment file.
gatherer:sync_in(Gatherer, Acc),
ok;
queue_index_walker_segment(Fd, Gatherer, N, Total) ->
queue_index_walker_segment(Fd, Gatherer, N, Total, Acc) ->
case file:read(Fd, ?ENTRY_SIZE) of
%% We found a non-ack persistent entry. Gather it.
{ok, <<1,_:7,1:1,_,1,Id:16/binary,_/bits>>} ->
gatherer:sync_in(Gatherer, {Id, 1}),
queue_index_walker_segment(Fd, Gatherer, N + 1, Total);
queue_index_walker_segment(Fd, Gatherer, N + 1, Total, [Id|Acc]);
%% We found an ack, a transient entry or a non-entry. Skip it.
{ok, _} ->
queue_index_walker_segment(Fd, Gatherer, N + 1, Total);
queue_index_walker_segment(Fd, Gatherer, N + 1, Total, Acc);
%% We reached the end of a partial segment file.
eof when Acc =:= [] ->
ok;
eof ->
gatherer:sync_in(Gatherer, Acc),
ok
end.

Expand Down
Loading