Skip to content

Fix memory leak when using mandatory flag in CQs (backport #3560) #3572

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 2 commits into from
Oct 13, 2021
Merged

Fix memory leak when using mandatory flag in CQs (backport #3560) #3572

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fe82b1d
Fix memory leak when using mandatory flag in CQs
Oct 12, 2021
347a813
Merge branch 'v3.9.x' into mergify/bp/v3.9.x/pr-3560
michaelklishin Oct 13, 2021
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
14 changes: 7 additions & 7 deletions deps/rabbit/src/rabbit_classic_queue.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,9 +299,9 @@ settlement_action(Type, QRef, MsgSeqs, Acc) ->
deliver(Qs0, #delivery{flow = Flow,
msg_seq_no = MsgNo,
message = #basic_message{exchange_name = _Ex},
confirm = _Confirm} = Delivery) ->
confirm = Confirm} = Delivery) ->
%% TODO: record master and slaves for confirm processing
{MPids, SPids, Qs, Actions} = qpids(Qs0, MsgNo),
{MPids, SPids, Qs, Actions} = qpids(Qs0, Confirm, MsgNo),
QPids = MPids ++ SPids,
case Flow of
%% Here we are tracking messages sent by the rabbit_channel
Expand Down Expand Up @@ -361,22 +361,22 @@ purge(Q) when ?is_amqqueue(Q) ->
QPid = amqqueue:get_pid(Q),
delegate:invoke(QPid, {gen_server2, call, [purge, infinity]}).

qpids(Qs, MsgNo) ->
qpids(Qs, Confirm, MsgNo) ->
lists:foldl(
fun ({Q, S0}, {MPidAcc, SPidAcc, Qs0, Actions0}) ->
QPid = amqqueue:get_pid(Q),
SPids = amqqueue:get_slave_pids(Q),
QRef = amqqueue:get_name(Q),
Actions = [{monitor, QPid, QRef}
| [{monitor, P, QRef} || P <- SPids]] ++ Actions0,
%% confirm record only if MsgNo isn't undefined
%% confirm record only if necessary
S = case S0 of
#?STATE{unconfirmed = U0} ->
Rec = [QPid | SPids],
U = case MsgNo of
undefined ->
U = case Confirm of
false ->
U0;
_ ->
true ->
U0#{MsgNo => #msg_status{pending = Rec}}
end,
S0#?STATE{pid = QPid,
Expand Down
4 changes: 2 additions & 2 deletions release-notes/3.9.8.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,12 @@ Contributors are encouraged to update them together with their changes. This hel
* When the mandatory flag was used when publishing to classic queues,
but publisher confirms were not, channels memory usage would grow indefinitely.

GitHub issue: [#3560](https://github.com/rabbitmq/rabbitmq-server/issues/3560)
GitHub issue: [#3560](https://github.com/rabbitmq/rabbitmq-server/issues/3560)

* `rabbitmq-diagnostics memory_breakdown` failed to read memory of connection
reader, writer and channel processes.

GitHub issue: [#3570](https://github.com/rabbitmq/rabbitmq-server/issues/3570)
GitHub issue: [#3570](https://github.com/rabbitmq/rabbitmq-server/issues/3570)


## Dependency Upgrades
Expand Down