55 lines
2.8 KiB
Diff
55 lines
2.8 KiB
Diff
# UNDF: UNDF-2026-000000239
|
|
--- 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))}.
|