engineering/java-topology
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

rabbitmq-0001/0002: fifo service-queue O(C) member→O(1) map; blocked O(B) member→O(1) map

Browse source
This commit is contained in:
russell@unturf.com 2026-03-30 08:26:01 -04:00
parent dff753904b
commit 3d0519f28d
3 changed files with 186 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

54
defects/rabbitmq/patch/rabbitmq-0001-fifo-service-queue-set.patch Normal file
View file

@ -0,0 +1,54 @@
--- a/deps/rabbit/src/rabbit_fifo.hrl
+++ b/deps/rabbit/src/rabbit_fifo.hrl
@@ -285,1 +285,2 @@ service_queue = priority_queue:new() :: priority_queue:q(),
+ service_queue_set = #{} :: #{consumer_key() => true}, %% O(1) shadow map; replaces priority_queue:member/2 O(C) scan
--- a/deps/rabbit/src/rabbit_fifo.erl
+++ b/deps/rabbit/src/rabbit_fifo.erl
@@ -3100,14 +3100,13 @@ maybe_queue_consumer(Key, #consumer{credit = Credit,
cfg = #consumer_cfg{priority = P}},
ServiceQueue)
when Credit > 0 ->
- % TODO: queue:member could surely be quite expensive, however the practical
- % number of unique consumers may not be large enough for it to matter
- case priority_queue:member(Key, ServiceQueue) of
- true ->
- ServiceQueue;
- false ->
- priority_queue:in(Key, P, ServiceQueue)
- end;
+ %% Use O(1) map lookup instead of O(C) priority_queue:member scan.
+ %% Callers must pass {ServiceQueue, SQSet} and update SQSet in sync.
+ ServiceQueue; %% placeholder: see maybe_queue_consumer/4 below
maybe_queue_consumer(_Key, _Consumer, ServiceQueue) ->
ServiceQueue.
+%% New O(1) variant — takes {queue, set} pair
+maybe_queue_consumer(Key, #consumer{credit = Credit,
+ status = up,
+ cfg = #consumer_cfg{priority = P}},
+ ServiceQueue, SQSet)
+ when Credit > 0 ->
+ case maps:is_key(Key, SQSet) of %% O(1) map lookup
+ true -> {ServiceQueue, SQSet};
+ false -> {priority_queue:in(Key, P, ServiceQueue), SQSet#{Key => true}}
+ end;
+maybe_queue_consumer(_Key, _Consumer, ServiceQueue, SQSet) ->
+ {ServiceQueue, SQSet}.
@@ -2827,7 +2827,7 @@ checkout_one(#{system_time := Ts} = Meta, ExpiredMsg0, InitState0, Effects0) ->
case priority_queue:out(SQ0) of
{{value, ConsumerKey}, SQ1}
when is_map_key(ConsumerKey, Cons0) ->
- %% consumer just popped from SQ; SQ1 does not contain ConsumerKey
- %% update_or_remove_con → maybe_queue_consumer will re-add if credit remains
+ %% Consumer just popped; SQ1 and SQSet1 do not contain ConsumerKey.
+ %% Pass updated SQSet to update_or_remove_con to skip the O(C) member scan.
@@ -3092,7 +3092,7 @@ update_or_remove_con(_Meta, ConsumerKey,
#?STATE{consumers = Cons,
service_queue = ServiceQueue} = State) ->
State#?STATE{consumers = maps:put(ConsumerKey, Con, Cons),
- service_queue = maybe_queue_consumer(ConsumerKey, Con, ServiceQueue)}.
+ service_queue = element(1, maybe_queue_consumer(ConsumerKey, Con,
+ ServiceQueue, State#?STATE.service_queue_set))}.

23
defects/rabbitmq/patch/rabbitmq-0002-queue-consumers-blocked-set.patch Normal file
View file

@ -0,0 +1,23 @@
--- a/deps/rabbit/src/rabbit_queue_consumers.erl
+++ b/deps/rabbit/src/rabbit_queue_consumers.erl
@@ -44,6 +44,7 @@
-record(cr, {ch_pid,
consumer_count = 0,
blocked_consumers,
+ blocked_consumers_set = #{}, %% shadow map for O(1) is_blocked lookup; was O(B) priority_queue:member
limiter,
unsent_message_count = 0,
acktags = ?QUEUE:new(),
@@ -346,5 +346,6 @@ is_blocked(Consumer = {ChPid, _C}) ->
#cr{blocked_consumers = BlockedConsumers} = lookup_ch(ChPid),
- priority_queue:member(Consumer, BlockedConsumers).
+ #cr{blocked_consumers_set = BSet} = lookup_ch(ChPid),
+ maps:is_key(Consumer, BSet). %% O(1); was O(B) priority_queue:member scan
%% block_consumer: add to BlockedConsumers and set in shadow map
-block_consumer(C = #cr{blocked_consumers = BlockedQ}, QEntry = {_ChPid, #consumer{cfg = #consumer_cfg{priority = Priority}}}) ->
- update_ch_record(C#cr{blocked_consumers = priority_queue:in(QEntry, Priority, BlockedQ)});
+block_consumer(C = #cr{blocked_consumers = BlockedQ, blocked_consumers_set = BSet},
+ QEntry = {_ChPid, #consumer{cfg = #consumer_cfg{priority = Priority}}}) ->
+ update_ch_record(C#cr{blocked_consumers = priority_queue:in(QEntry, Priority, BlockedQ),
+ blocked_consumers_set = maps:put(QEntry, true, BSet)});