diff options
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 420 | ||||
| -rw-r--r-- | src/rabbit_limiter.erl | 10 | ||||
| -rw-r--r-- | src/rabbit_queue_consumers.erl | 426 |
3 files changed, 503 insertions, 353 deletions
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index bc105e3d26..281aecb93c 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -20,7 +20,6 @@ -behaviour(gen_server2). --define(UNSENT_MESSAGE_LIMIT, 200). -define(SYNC_INTERVAL, 25). %% milliseconds -define(RAM_DURATION_UPDATE_INTERVAL, 5000). @@ -38,8 +37,7 @@ has_had_consumers, backing_queue, backing_queue_state, - active_consumers, - consumer_use, + consumers, expires, sync_timer_ref, rate_timer_ref, @@ -57,21 +55,6 @@ status }). --record(consumer, {tag, ack_required, args}). - -%% These are held in our process dictionary --record(cr, {ch_pid, - monitor_ref, - acktags, - consumer_count, - %% Queue of {ChPid, #consumer{}} for consumers which have - %% been blocked for any reason - blocked_consumers, - %% The limiter itself - limiter, - %% Internal flow control for queue -> writer - unsent_message_count}). - %%---------------------------------------------------------------------------- -ifdef(use_specs). @@ -150,8 +133,7 @@ init_state(Q) -> State = #q{q = Q, exclusive_consumer = none, has_had_consumers = false, - active_consumers = priority_queue:new(), - consumer_use = {inactive, now_micros(), 0, 0.0}, + consumers = rabbit_queue_consumers:new(), senders = pmon:new(delegate), msg_id_to_channel = gb_trees:empty(), status = running, @@ -235,10 +217,10 @@ notify_decorators(Event, Props, State) when Event =:= startup; Event =:= shutdown -> decorator_callback(qname(State), Event, Props); -notify_decorators(Event, Props, State = #q{active_consumers = ACs, +notify_decorators(Event, Props, State = #q{consumers = Consumers, backing_queue = BQ, backing_queue_state = BQS}) -> - P = priority_queue:highest(ACs), + P = rabbit_queue_consumers:max_active_priority(Consumers), decorator_callback(qname(State), notify, [Event, [{max_active_consumer_priority, P}, {is_empty, BQ:is_empty(BQS)} | @@ -316,7 +298,7 @@ init_max_length(MaxLen, State) -> State1. terminate_shutdown(Fun, State) -> - State1 = #q{backing_queue_state = BQS} = + State1 = #q{backing_queue_state = BQS, consumers = Consumers} = lists:foldl(fun (F, S) -> F(S) end, State, [fun stop_sync_timer/1, fun stop_rate_timer/1, @@ -328,7 +310,8 @@ terminate_shutdown(Fun, State) -> QName = qname(State), notify_decorators(shutdown, [], State), [emit_consumer_deleted(Ch, CTag, QName) || - {Ch, CTag, _} <- consumers(State1)], + {Ch, CTag, _, _} <- + rabbit_queue_consumers:all(Consumers)], State1#q{backing_queue_state = Fun(BQS)} end. @@ -411,153 +394,27 @@ stop_ttl_timer(State) -> rabbit_misc:stop_timer(State, #q.ttl_timer_ref). ensure_stats_timer(State) -> rabbit_event:ensure_stats_timer(State, #q.stats_timer, emit_stats). -assert_invariant(State = #q{active_consumers = AC}) -> - true = (priority_queue:is_empty(AC) orelse is_empty(State)). +assert_invariant(State = #q{consumers = Consumers}) -> + true = (rabbit_queue_consumers:inactive(Consumers) orelse is_empty(State)). is_empty(#q{backing_queue = BQ, backing_queue_state = BQS}) -> BQ:is_empty(BQS). -lookup_ch(ChPid) -> - case get({ch, ChPid}) of - undefined -> not_found; - C -> C - end. - -ch_record(ChPid, LimiterPid) -> - Key = {ch, ChPid}, - case get(Key) of - undefined -> MonitorRef = erlang:monitor(process, ChPid), - Limiter = rabbit_limiter:client(LimiterPid), - C = #cr{ch_pid = ChPid, - monitor_ref = MonitorRef, - acktags = queue:new(), - consumer_count = 0, - blocked_consumers = priority_queue:new(), - limiter = Limiter, - unsent_message_count = 0}, - put(Key, C), - C; - C = #cr{} -> C - end. - -update_ch_record(C = #cr{consumer_count = ConsumerCount, - acktags = ChAckTags, - unsent_message_count = UnsentMessageCount}) -> - case {queue:is_empty(ChAckTags), ConsumerCount, UnsentMessageCount} of - {true, 0, 0} -> ok = erase_ch_record(C); - _ -> ok = store_ch_record(C) - end, - C. - -store_ch_record(C = #cr{ch_pid = ChPid}) -> - put({ch, ChPid}, C), - ok. - -erase_ch_record(#cr{ch_pid = ChPid, monitor_ref = MonitorRef}) -> - erlang:demonitor(MonitorRef), - erase({ch, ChPid}), - ok. - -all_ch_record() -> [C || {{ch, _}, C} <- get()]. - -block_consumer(C = #cr{blocked_consumers = Blocked}, - {_ChPid, #consumer{tag = CTag}} = QEntry, State) -> - update_ch_record(C#cr{blocked_consumers = add_consumer(QEntry, Blocked)}), - notify_decorators(consumer_blocked, [{consumer_tag, CTag}], State). - -is_ch_blocked(#cr{unsent_message_count = Count, limiter = Limiter}) -> - Count >= ?UNSENT_MESSAGE_LIMIT orelse rabbit_limiter:is_suspended(Limiter). - maybe_send_drained(WasEmpty, State) -> case (not WasEmpty) andalso is_empty(State) of true -> notify_decorators(queue_empty, [], State), - [send_drained(C) || C <- all_ch_record()]; + rabbit_queue_consumers:send_drained(); false -> ok end, State. -send_drained(C = #cr{ch_pid = ChPid, limiter = Limiter}) -> - case rabbit_limiter:drained(Limiter) of - {[], Limiter} -> ok; - {CTagCredit, Limiter2} -> rabbit_channel:send_drained( - ChPid, CTagCredit), - update_ch_record(C#cr{limiter = Limiter2}) - end. - -deliver_msgs_to_consumers(_DeliverFun, true, State) -> - {true, State}; -deliver_msgs_to_consumers(DeliverFun, false, - State = #q{active_consumers = ActiveConsumers}) -> - case priority_queue:out_p(ActiveConsumers) of - {empty, _} -> - {false, update_consumer_use(State, inactive)}; - {{value, QEntry, Priority}, Tail} -> - {Stop, State1} = deliver_msg_to_consumer( - DeliverFun, QEntry, Priority, - State#q{active_consumers = Tail}), - deliver_msgs_to_consumers(DeliverFun, Stop, State1) - end. - -deliver_msg_to_consumer(DeliverFun, E = {ChPid, Consumer}, Priority, State) -> - C = lookup_ch(ChPid), - case is_ch_blocked(C) of - true -> block_consumer(C, E, State), - {false, State}; - false -> case rabbit_limiter:can_send(C#cr.limiter, - Consumer#consumer.ack_required, - Consumer#consumer.tag) of - {suspend, Limiter} -> - block_consumer(C#cr{limiter = Limiter}, E, State), - {false, State}; - {continue, Limiter} -> - AC1 = priority_queue:in(E, Priority, - State#q.active_consumers), - deliver_msg_to_consumer0( - DeliverFun, Consumer, C#cr{limiter = Limiter}, - State#q{active_consumers = AC1}) - end - end. - -deliver_msg_to_consumer0(DeliverFun, - #consumer{tag = ConsumerTag, - ack_required = AckRequired}, - C = #cr{ch_pid = ChPid, - acktags = ChAckTags, - unsent_message_count = Count}, - State = #q{q = #amqqueue{name = QName}}) -> - {{Message, IsDelivered, AckTag}, Stop, State1} = - DeliverFun(AckRequired, State), - rabbit_channel:deliver(ChPid, ConsumerTag, AckRequired, - {QName, self(), AckTag, IsDelivered, Message}), - ChAckTags1 = case AckRequired of - true -> queue:in(AckTag, ChAckTags); - false -> ChAckTags - end, - update_ch_record(C#cr{acktags = ChAckTags1, - unsent_message_count = Count + 1}), - {Stop, State1}. - -update_consumer_use(State = #q{consumer_use = CUInfo}, Use) -> - State#q{consumer_use = update_consumer_use1(CUInfo, Use)}. - -update_consumer_use1({inactive, _, _, _} = CUInfo, inactive) -> - CUInfo; -update_consumer_use1({active, _, _} = CUInfo, active) -> - CUInfo; -update_consumer_use1({active, Since, Avg}, inactive) -> - Now = now_micros(), - {inactive, Now, Now - Since, Avg}; -update_consumer_use1({inactive, Since, Active, Avg}, active) -> - Now = now_micros(), - {active, Now, consumer_use_avg(Active, Now - Since, Avg)}. - -consumer_use_avg(Active, Inactive, Avg) -> - Time = Inactive + Active, - Ratio = Active / Time, - Weight = erlang:min(1, Time / 1000000), - case Avg of - undefined -> Ratio; - _ -> Ratio * Weight + Avg * (1 - Weight) - end. +deliver_msgs_to_consumers(FetchFun, Stop, State = #q{consumers = Consumers}) -> + {Active, Blocked, State1, Consumers1} = + rabbit_queue_consumers:deliver(FetchFun, Stop, qname(State), State, + Consumers), + State2 = State1#q{consumers = Consumers1}, + [notify_decorators(consumer_blocked, [{consumer_tag, CTag}], State2) || + {_ChPid, CTag} <- Blocked], + {Active, State2}. confirm_messages([], State) -> State; @@ -612,13 +469,6 @@ run_message_queue(State) -> end, is_empty(State), State), State3. -add_consumer({ChPid, Consumer = #consumer{args = Args}}, ActiveConsumers) -> - Priority = case rabbit_misc:table_lookup(Args, <<"x-priority">>) of - {_, P} -> P; - _ -> 0 - end, - priority_queue:in({ChPid, Consumer}, Priority, ActiveConsumers). - attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message}, Props, Delivered, State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> @@ -714,54 +564,14 @@ requeue(AckTags, ChPid, State) -> subtract_acks(ChPid, AckTags, State, fun (State1) -> requeue_and_run(AckTags, State1) end). -remove_consumer(ChPid, ConsumerTag, Queue) -> - priority_queue:filter(fun ({CP, #consumer{tag = CTag}}) -> - (CP /= ChPid) or (CTag /= ConsumerTag) - end, Queue). - -remove_consumers(ChPid, Queue, QName) -> - priority_queue:filter(fun ({CP, #consumer{tag = CTag}}) when CP =:= ChPid -> - emit_consumer_deleted(ChPid, CTag, QName), - false; - (_) -> - true - end, Queue). - -channel_consumers(ChPid, Queue) -> - priority_queue:fold( - fun ({CP, #consumer{tag = CTag}}, _, Acc) when CP =:= ChPid -> - [CTag | Acc] - end, [], Queue). - -possibly_unblock(State, ChPid, Update) -> - case lookup_ch(ChPid) of - not_found -> State; - C -> C1 = Update(C), - case is_ch_blocked(C) andalso not is_ch_blocked(C1) of - false -> update_ch_record(C1), - State; - true -> unblock(State, C1) - end - end. - -unblock(State, C = #cr{limiter = Limiter}) -> - case lists:partition( - fun({_P, {_ChPid, #consumer{tag = CTag}}}) -> - rabbit_limiter:is_consumer_blocked(Limiter, CTag) - end, priority_queue:to_list(C#cr.blocked_consumers)) of - {_, []} -> - update_ch_record(C), +possibly_unblock(Update, ChPid, State = #q{consumers = Consumers}) -> + case rabbit_queue_consumers:possibly_unblock(Update, ChPid, Consumers) of + unchanged -> State; - {Blocked, Unblocked} -> - BlockedQ = priority_queue:from_list(Blocked), - UnblockedQ = priority_queue:from_list(Unblocked), - update_ch_record(C#cr{blocked_consumers = BlockedQ}), - AC1 = priority_queue:join(State#q.active_consumers, UnblockedQ), - State1 = update_consumer_use(State#q{active_consumers = AC1}, - active), - [notify_decorators( - consumer_unblocked, [{consumer_tag, CTag}], State1) || - {_P, {_ChPid, #consumer{tag = CTag}}} <- Unblocked], + {unblocked, UnblockedCTags, Consumers1} -> + State1 = State#q{consumers = Consumers1}, + [notify_decorators(consumer_unblocked, [{consumer_tag, CTag}], + State1) || CTag <- UnblockedCTags], run_message_queue(State1) end. @@ -769,7 +579,7 @@ should_auto_delete(#q{q = #amqqueue{auto_delete = false}}) -> false; should_auto_delete(#q{has_had_consumers = false}) -> false; should_auto_delete(State) -> is_unused(State). -handle_ch_down(DownPid, State = #q{active_consumers = AC, +handle_ch_down(DownPid, State = #q{consumers = Consumers, exclusive_consumer = Holder, senders = Senders}) -> State1 = State#q{senders = case pmon:is_monitored(DownPid, Senders) of @@ -777,29 +587,23 @@ handle_ch_down(DownPid, State = #q{active_consumers = AC, true -> credit_flow:peer_down(DownPid), pmon:demonitor(DownPid, Senders) end}, - case lookup_ch(DownPid) of + case rabbit_queue_consumers:erase_ch(DownPid, Consumers) of not_found -> {ok, State1}; - C = #cr{ch_pid = ChPid, - acktags = ChAckTags, - blocked_consumers = Blocked} -> - QName = qname(State), - AC1 = remove_consumers(ChPid, AC, QName), - _ = remove_consumers(ChPid, Blocked, QName), %% for stats emission - ok = erase_ch_record(C), + {ChAckTags, ChCTags, Consumers1} -> + QName = qname(State1), + [emit_consumer_deleted(DownPid, CTag, QName) || CTag <- ChCTags], Holder1 = case Holder of {DownPid, _} -> none; Other -> Other end, - State2 = State1#q{active_consumers = AC1, + State2 = State1#q{consumers = Consumers1, exclusive_consumer = Holder1}, [notify_decorators(basic_cancel, [{consumer_tag, CTag}], State2) || - CTag <- channel_consumers(ChPid, AC)], - [notify_decorators(basic_cancel, [{consumer_tag, CTag}], State2) || - CTag <- channel_consumers(ChPid, Blocked)], + CTag <- ChCTags], case should_auto_delete(State2) of true -> {stop, State2}; - false -> {ok, requeue_and_run(queue:to_list(ChAckTags), + false -> {ok, requeue_and_run(ChAckTags, ensure_expiry_timer(State2))} end end. @@ -814,10 +618,7 @@ check_exclusive_access(none, true, State) -> false -> in_use end. -consumer_count() -> - lists:sum([Count || #cr{consumer_count = Count} <- all_ch_record()]). - -is_unused(_State) -> consumer_count() == 0. +is_unused(_State) -> rabbit_queue_consumers:count() == 0. maybe_send_reply(_ChPid, undefined) -> ok; maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg). @@ -829,23 +630,9 @@ backing_queue_timeout(State = #q{backing_queue = BQ, State#q{backing_queue_state = BQ:timeout(BQS)}. subtract_acks(ChPid, AckTags, State, Fun) -> - case lookup_ch(ChPid) of - not_found -> - State; - C = #cr{acktags = ChAckTags} -> - update_ch_record( - C#cr{acktags = subtract_acks(AckTags, [], ChAckTags)}), - Fun(State) - end. - -subtract_acks([], [], AckQ) -> - AckQ; -subtract_acks([], Prefix, AckQ) -> - queue:join(queue:from_list(lists:reverse(Prefix)), AckQ); -subtract_acks([T | TL] = AckTags, Prefix, AckQ) -> - case queue:out(AckQ) of - {{value, T}, QTail} -> subtract_acks(TL, Prefix, QTail); - {{value, AT}, QTail} -> subtract_acks(AckTags, [AT | Prefix], QTail) + case rabbit_queue_consumers:subtract_acks(ChPid, AckTags) of + not_found -> State; + ok -> Fun(State) end. message_properties(Message, Confirm, #q{ttl = TTL}) -> @@ -1066,21 +853,16 @@ i(exclusive_consumer_tag, #q{exclusive_consumer = {_ChPid, ConsumerTag}}) -> i(messages_ready, #q{backing_queue_state = BQS, backing_queue = BQ}) -> BQ:len(BQS); i(messages_unacknowledged, _) -> - lists:sum([queue:len(C#cr.acktags) || C <- all_ch_record()]); + rabbit_queue_consumers:unacknowledged_message_count(); i(messages, State) -> lists:sum([i(Item, State) || Item <- [messages_ready, messages_unacknowledged]]); i(consumers, _) -> - consumer_count(); -i(consumer_utilisation, #q{consumer_use = ConsumerUse}) -> - case consumer_count() of + rabbit_queue_consumers:count(); +i(consumer_utilisation, #q{consumers = Consumers}) -> + case rabbit_queue_consumers:count() of 0 -> ''; - _ -> case ConsumerUse of - {active, Since, Avg} -> - consumer_use_avg(now_micros() - Since, 0, Avg); - {inactive, Since, Active, Avg} -> - consumer_use_avg(Active, now_micros() - Since, Avg) - end + _ -> rabbit_queue_consumers:utilisation(Consumers) end; i(memory, _) -> {memory, M} = process_info(self(), memory), @@ -1106,17 +888,6 @@ i(backing_queue_status, #q{backing_queue_state = BQS, backing_queue = BQ}) -> i(Item, _) -> throw({bad_argument, Item}). -consumers(#q{active_consumers = ActiveConsumers}) -> - lists:foldl(fun (C, Acc) -> consumers(C#cr.blocked_consumers, Acc) end, - consumers(ActiveConsumers, []), all_ch_record()). - -consumers(Consumers, Acc) -> - priority_queue:fold( - fun ({ChPid, Consumer}, _P, Acc1) -> - #consumer{tag = CTag, ack_required = Ack, args = Args} = Consumer, - [{ChPid, CTag, Ack, Args} | Acc1] - end, Acc, Consumers). - emit_stats(State) -> emit_stats(State, []). @@ -1205,8 +976,8 @@ handle_call({info, Items}, _From, State) -> catch Error -> reply({error, Error}, State) end; -handle_call(consumers, _From, State) -> - reply(consumers(State), State); +handle_call(consumers, _From, State = #q{consumers = Consumers}) -> + reply(rabbit_queue_consumers:all(Consumers), State); handle_call({deliver, Delivery, Delivered}, From, State) -> %% Synchronous, "mandatory" deliver mode. @@ -1234,10 +1005,8 @@ handle_call({basic_get, ChPid, NoAck, LimiterPid}, _From, {{Message, IsDelivered, AckTag}, #q{backing_queue = BQ, backing_queue_state = BQS} = State2} -> case AckRequired of - true -> C = #cr{acktags = ChAckTags} = - ch_record(ChPid, LimiterPid), - ChAckTags1 = queue:in(AckTag, ChAckTags), - update_ch_record(C#cr{acktags = ChAckTags1}); + true -> ok = rabbit_queue_consumers:record_ack( + ChPid, LimiterPid, AckTag); false -> ok end, Msg = {QName, self(), AckTag, IsDelivered, Message}, @@ -1246,39 +1015,21 @@ handle_call({basic_get, ChPid, NoAck, LimiterPid}, _From, handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive, ConsumerTag, ExclusiveConsume, CreditArgs, OtherArgs, OkMsg}, - _From, State = #q{active_consumers = AC, + _From, State = #q{consumers = Consumers, exclusive_consumer = Holder}) -> case check_exclusive_access(Holder, ExclusiveConsume, State) of in_use -> reply({error, exclusive_consume_unavailable}, State); - ok -> C = #cr{consumer_count = Count, - limiter = Limiter} = - ch_record(ChPid, LimiterPid), - Limiter1 = case LimiterActive of - true -> rabbit_limiter:activate(Limiter); - false -> Limiter - end, - Limiter2 = case CreditArgs of - none -> Limiter1; - {Crd, Drain} -> rabbit_limiter:credit( - Limiter1, ConsumerTag, - Crd, Drain) - end, - C1 = update_ch_record(C#cr{consumer_count = Count + 1, - limiter = Limiter2}), - case is_empty(State) of - true -> send_drained(C1); - false -> ok - end, - Consumer = #consumer{tag = ConsumerTag, - ack_required = not NoAck, - args = OtherArgs}, - AC1 = add_consumer({ChPid, Consumer}, AC), + ok -> Consumers1 = rabbit_queue_consumers:add( + ChPid, ConsumerTag, NoAck, + LimiterPid, LimiterActive, + CreditArgs, OtherArgs, + is_empty(State), Consumers), ExclusiveConsumer = if ExclusiveConsume -> {ChPid, ConsumerTag}; true -> Holder end, - State1 = State#q{active_consumers = AC1, - has_had_consumers = true, + State1 = State#q{consumers = Consumers1, + has_had_consumers = true, exclusive_consumer = ExclusiveConsumer}, ok = maybe_send_reply(ChPid, OkMsg), emit_consumer_created(ChPid, ConsumerTag, ExclusiveConsume, @@ -1289,30 +1040,18 @@ handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive, end; handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From, - State = #q{active_consumers = AC, + State = #q{consumers = Consumers, exclusive_consumer = Holder}) -> ok = maybe_send_reply(ChPid, OkMsg), - case lookup_ch(ChPid) of + case rabbit_queue_consumers:remove(ChPid, ConsumerTag, Consumers) of not_found -> reply(ok, State); - C = #cr{consumer_count = Count, - limiter = Limiter, - blocked_consumers = Blocked} -> - AC1 = remove_consumer(ChPid, ConsumerTag, AC), - Blocked1 = remove_consumer(ChPid, ConsumerTag, Blocked), - Limiter1 = case Count of - 1 -> rabbit_limiter:deactivate(Limiter); - _ -> Limiter - end, - Limiter2 = rabbit_limiter:forget_consumer(Limiter1, ConsumerTag), - update_ch_record(C#cr{consumer_count = Count - 1, - limiter = Limiter2, - blocked_consumers = Blocked1}), + Consumers1 -> Holder1 = case Holder of {ChPid, ConsumerTag} -> none; _ -> Holder end, - State1 = State#q{active_consumers = AC1, + State1 = State#q{consumers = Consumers1, exclusive_consumer = Holder1}, emit_consumer_deleted(ChPid, ConsumerTag, qname(State1)), notify_decorators( @@ -1326,7 +1065,7 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From, handle_call(stat, _From, State) -> State1 = #q{backing_queue = BQ, backing_queue_state = BQS} = ensure_expiry_timer(State), - reply({ok, BQ:len(BQS), consumer_count()}, State1); + reply({ok, BQ:len(BQS), rabbit_queue_consumers:count()}, State1); handle_call({delete, IfUnused, IfEmpty}, _From, State = #q{backing_queue_state = BQS, backing_queue = BQ}) -> @@ -1378,10 +1117,11 @@ handle_call(cancel_sync_mirrors, _From, State) -> reply({ok, not_syncing}, State); handle_call(force_event_refresh, _From, - State = #q{exclusive_consumer = Exclusive}) -> + State = #q{consumers = Consumers, + exclusive_consumer = Exclusive}) -> rabbit_event:notify(queue_created, infos(?CREATION_EVENT_KEYS, State)), QName = qname(State), - AllConsumers = consumers(State), + AllConsumers = rabbit_queue_consumers:all(Consumers), case Exclusive of none -> [emit_consumer_created( Ch, CTag, false, AckRequired, QName, Args) || @@ -1427,25 +1167,16 @@ handle_cast(delete_immediately, State) -> stop(State); handle_cast({resume, ChPid}, State) -> - noreply( - possibly_unblock(State, ChPid, - fun (C = #cr{limiter = Limiter}) -> - C#cr{limiter = rabbit_limiter:resume(Limiter)} - end)); + noreply(possibly_unblock(rabbit_queue_consumers:resume_fun(), + ChPid, State)); handle_cast({notify_sent, ChPid, Credit}, State) -> - noreply( - possibly_unblock(State, ChPid, - fun (C = #cr{unsent_message_count = Count}) -> - C#cr{unsent_message_count = Count - Credit} - end)); + noreply(possibly_unblock(rabbit_queue_consumers:notify_sent_fun(Credit), + ChPid, State)); handle_cast({activate_limit, ChPid}, State) -> - noreply( - possibly_unblock(State, ChPid, - fun (C = #cr{limiter = Limiter}) -> - C#cr{limiter = rabbit_limiter:activate(Limiter)} - end)); + noreply(possibly_unblock(rabbit_queue_consumers:activate_limit_fun(), + ChPid, State)); handle_cast({flush, ChPid}, State) -> ok = rabbit_channel:flushed(ChPid, self()), @@ -1482,18 +1213,9 @@ handle_cast({credit, ChPid, CTag, Credit, Drain}, backing_queue_state = BQS}) -> Len = BQ:len(BQS), rabbit_channel:send_credit_reply(ChPid, Len), - C = #cr{limiter = Limiter} = lookup_ch(ChPid), - C1 = C#cr{limiter = rabbit_limiter:credit(Limiter, CTag, Credit, Drain)}, - noreply(case Drain andalso Len == 0 of - true -> update_ch_record(C1), - send_drained(C1), - State; - false -> case is_ch_blocked(C1) of - true -> update_ch_record(C1), - State; - false -> unblock(State, C1) - end - end); + noreply(possibly_unblock(rabbit_queue_consumers:credit_fun( + Len == 0, Credit, Drain, CTag), + ChPid, State)); handle_cast(notify_decorators, State) -> notify_decorators(refresh, [], State), diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl index 22da465b88..4e1e299c54 100644 --- a/src/rabbit_limiter.erl +++ b/src/rabbit_limiter.erl @@ -126,7 +126,7 @@ get_prefetch_limit/1, ack/2, pid/1]). %% queue API -export([client/1, activate/1, can_send/3, resume/1, deactivate/1, - is_suspended/1, is_consumer_blocked/2, credit/4, drained/1, + is_suspended/1, is_consumer_blocked/2, credit/5, drained/1, forget_consumer/2]). %% callbacks -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2, @@ -168,8 +168,8 @@ -spec(deactivate/1 :: (qstate()) -> qstate()). -spec(is_suspended/1 :: (qstate()) -> boolean()). -spec(is_consumer_blocked/2 :: (qstate(), rabbit_types:ctag()) -> boolean()). --spec(credit/4 :: (qstate(), rabbit_types:ctag(), non_neg_integer(), boolean()) - -> qstate()). +-spec(credit/5 :: (qstate(), rabbit_types:ctag(), non_neg_integer(), boolean(), + boolean()) -> qstate()). -spec(drained/1 :: (qstate()) -> {[{rabbit_types:ctag(), non_neg_integer()}], qstate()}). -spec(forget_consumer/2 :: (qstate(), rabbit_types:ctag()) -> qstate()). @@ -276,7 +276,9 @@ is_consumer_blocked(#qstate{credits = Credits}, CTag) -> none -> false end. -credit(Limiter = #qstate{credits = Credits}, CTag, Credit, Drain) -> +credit(Limiter = #qstate{credits = Credits}, CTag, _Credit, true, true) -> + Limiter#qstate{credits = update_credit(CTag, 0, true, Credits)}; +credit(Limiter = #qstate{credits = Credits}, CTag, Credit, false, Drain) -> Limiter#qstate{credits = update_credit(CTag, Credit, Drain, Credits)}. drained(Limiter = #qstate{credits = Credits}) -> diff --git a/src/rabbit_queue_consumers.erl b/src/rabbit_queue_consumers.erl new file mode 100644 index 0000000000..702091dca9 --- /dev/null +++ b/src/rabbit_queue_consumers.erl @@ -0,0 +1,426 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License +%% at http://www.mozilla.org/MPL/ +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and +%% limitations under the License. +%% +%% The Original Code is RabbitMQ. +%% +%% The Initial Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2013 GoPivotal, Inc. All rights reserved. +%% + +-module(rabbit_queue_consumers). + +-export([new/0, max_active_priority/1, inactive/1, all/1, count/0, + unacknowledged_message_count/0, add/9, remove/3, erase_ch/2, + send_drained/0, deliver/5, record_ack/3, subtract_acks/2, + possibly_unblock/3, + resume_fun/0, notify_sent_fun/1, activate_limit_fun/0, credit_fun/4, + utilisation/1]). + +%%---------------------------------------------------------------------------- + +-define(UNSENT_MESSAGE_LIMIT, 200). + +-record(state, {consumers, use}). + +-record(consumer, {tag, ack_required, args}). + +%% These are held in our process dictionary +-record(cr, {ch_pid, + monitor_ref, + acktags, + consumer_count, + %% Queue of {ChPid, #consumer{}} for consumers which have + %% been blocked for any reason + blocked_consumers, + %% The limiter itself + limiter, + %% Internal flow control for queue -> writer + unsent_message_count}). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-type time_micros() :: non_neg_integer(). +-type ratio() :: float(). +-type state() :: #state{consumers ::priority_queue:q(), + use :: {'inactive', + time_micros(), time_micros(), ratio()} | + {'active', time_micros(), ratio()}}. +-type ch() :: pid(). +-type ack() :: non_neg_integer(). +-type cr_fun() :: fun ((#cr{}) -> #cr{}). +-type credit_args() :: {non_neg_integer(), boolean()} | 'none'. +-type fetch_result() :: {rabbit_types:basic_message(), boolean(), ack()}. + +-spec new() -> state(). +-spec max_active_priority(state()) -> integer() | 'infinity' | 'empty'. +-spec inactive(state()) -> boolean(). +-spec all(state()) -> [{ch(), rabbit_types:ctag(), boolean(), + rabbit_framing:amqp_table()}]. +-spec count() -> non_neg_integer(). +-spec unacknowledged_message_count() -> non_neg_integer(). +-spec add(ch(), rabbit_types:ctag(), boolean(), pid(), boolean(), + credit_args(), rabbit_framing:amqp_table(), boolean(), + state()) -> state(). +-spec remove(ch(), rabbit_types:ctag(), state()) -> + 'not_found' | state(). +-spec erase_ch(ch(), state()) -> + 'not_found' | {[ack()], [rabbit_types:ctag()], + state()}. +-spec send_drained() -> 'ok'. +-spec deliver(fun ((boolean(), T) -> {fetch_result(), boolean(), T}), + boolean(), rabbit_amqqueue:name(), T, state()) -> + {boolean(), [{ch(), rabbit_types:ctag()}], T, state()}. +-spec record_ack(ch(), pid(), ack()) -> 'ok'. +-spec subtract_acks(ch(), [ack()]) -> 'not_found' | 'ok'. +-spec possibly_unblock(cr_fun(), ch(), state()) -> + 'unchanged' | + {'unblocked', [rabbit_types:ctag()], state()}. +-spec resume_fun() -> cr_fun(). +-spec notify_sent_fun(non_neg_integer()) -> cr_fun(). +-spec activate_limit_fun() -> cr_fun(). +-spec credit_fun(boolean(), non_neg_integer(), boolean(), + rabbit_types:ctag()) -> cr_fun(). +-spec utilisation(state()) -> ratio(). + +-endif. + +%%---------------------------------------------------------------------------- + +new() -> #state{consumers = priority_queue:new(), + use = {inactive, now_micros(), 0, 0.0}}. + +max_active_priority(#state{consumers = Consumers}) -> + priority_queue:highest(Consumers). + +inactive(#state{consumers = Consumers}) -> + priority_queue:is_empty(Consumers). + +all(#state{consumers = Consumers}) -> + lists:foldl(fun (C, Acc) -> consumers(C#cr.blocked_consumers, Acc) end, + consumers(Consumers, []), all_ch_record()). + +consumers(Consumers, Acc) -> + priority_queue:fold( + fun ({ChPid, Consumer}, _P, Acc1) -> + #consumer{tag = CTag, ack_required = Ack, args = Args} = Consumer, + [{ChPid, CTag, Ack, Args} | Acc1] + end, Acc, Consumers). + +count() -> lists:sum([Count || #cr{consumer_count = Count} <- all_ch_record()]). + +unacknowledged_message_count() -> + lists:sum([queue:len(C#cr.acktags) || C <- all_ch_record()]). + +add(ChPid, ConsumerTag, NoAck, LimiterPid, LimiterActive, CreditArgs, OtherArgs, + IsEmpty, State = #state{consumers = Consumers}) -> + C = #cr{consumer_count = Count, + limiter = Limiter} = ch_record(ChPid, LimiterPid), + Limiter1 = case LimiterActive of + true -> rabbit_limiter:activate(Limiter); + false -> Limiter + end, + Limiter2 = case CreditArgs of + none -> Limiter1; + {Crd, Drain} -> rabbit_limiter:credit( + Limiter1, ConsumerTag, Crd, IsEmpty, Drain) + end, + C1 = C#cr{consumer_count = Count + 1, + limiter = Limiter2}, + update_ch_record(case IsEmpty of + true -> send_drained(C1); + false -> C1 + end), + Consumer = #consumer{tag = ConsumerTag, + ack_required = not NoAck, + args = OtherArgs}, + State#state{consumers = add_consumer({ChPid, Consumer}, Consumers)}. + +remove(ChPid, ConsumerTag, State = #state{consumers = Consumers}) -> + case lookup_ch(ChPid) of + not_found -> + not_found; + C = #cr{consumer_count = Count, + limiter = Limiter, + blocked_consumers = Blocked} -> + Blocked1 = remove_consumer(ChPid, ConsumerTag, Blocked), + Limiter1 = case Count of + 1 -> rabbit_limiter:deactivate(Limiter); + _ -> Limiter + end, + Limiter2 = rabbit_limiter:forget_consumer(Limiter1, ConsumerTag), + update_ch_record(C#cr{consumer_count = Count - 1, + limiter = Limiter2, + blocked_consumers = Blocked1}), + State#state{consumers = + remove_consumer(ChPid, ConsumerTag, Consumers)} + end. + +erase_ch(ChPid, State = #state{consumers = Consumers}) -> + case lookup_ch(ChPid) of + not_found -> + not_found; + C = #cr{ch_pid = ChPid, + acktags = ChAckTags, + blocked_consumers = BlockedQ} -> + AllConsumers = priority_queue:join(Consumers, BlockedQ), + ok = erase_ch_record(C), + {queue:to_list(ChAckTags), + tags(priority_queue:to_list(AllConsumers)), + State#state{consumers = remove_consumers(ChPid, Consumers)}} + end. + +send_drained() -> [update_ch_record(send_drained(C)) || C <- all_ch_record()], + ok. + +deliver(FetchFun, Stop, QName, S, State) -> + deliver(FetchFun, Stop, QName, [], S, State). + +deliver(_FetchFun, true, _QName, Blocked, S, State) -> + {true, Blocked, S, State}; +deliver( FetchFun, false, QName, Blocked, S, + State = #state{consumers = Consumers, use = Use}) -> + case priority_queue:out_p(Consumers) of + {empty, _} -> + {false, Blocked, S, State#state{use = update_use(Use, inactive)}}; + {{value, QEntry, Priority}, Tail} -> + {Stop, Blocked1, S1, Consumers1} = + deliver_to_consumer(FetchFun, QEntry, Priority, QName, + Blocked, S, Tail), + deliver(FetchFun, Stop, QName, Blocked1, S1, + State#state{consumers = Consumers1}) + end. + +deliver_to_consumer(FetchFun, E = {ChPid, Consumer}, Priority, QName, + Blocked, S, Consumers) -> + C = lookup_ch(ChPid), + case is_ch_blocked(C) of + true -> block_consumer(C, E), + Blocked1 = [{ChPid, Consumer#consumer.tag} | Blocked], + {false, Blocked1, S, Consumers}; + false -> case rabbit_limiter:can_send(C#cr.limiter, + Consumer#consumer.ack_required, + Consumer#consumer.tag) of + {suspend, Limiter} -> + block_consumer(C#cr{limiter = Limiter}, E), + Blocked1 = [{ChPid, Consumer#consumer.tag} | Blocked], + {false, Blocked1, S, Consumers}; + {continue, Limiter} -> + {Stop, S1} = deliver_to_consumer( + FetchFun, Consumer, + C#cr{limiter = Limiter}, QName, S), + {Stop, Blocked, S1, + priority_queue:in(E, Priority, Consumers)} + end + end. + +deliver_to_consumer(FetchFun, + #consumer{tag = ConsumerTag, + ack_required = AckRequired}, + C = #cr{ch_pid = ChPid, + acktags = ChAckTags, + unsent_message_count = Count}, + QName, S) -> + {{Message, IsDelivered, AckTag}, Stop, S1} = FetchFun(AckRequired, S), + rabbit_channel:deliver(ChPid, ConsumerTag, AckRequired, + {QName, self(), AckTag, IsDelivered, Message}), + ChAckTags1 = case AckRequired of + true -> queue:in(AckTag, ChAckTags); + false -> ChAckTags + end, + update_ch_record(C#cr{acktags = ChAckTags1, + unsent_message_count = Count + 1}), + {Stop, S1}. + +record_ack(ChPid, LimiterPid, AckTag) -> + C = #cr{acktags = ChAckTags} = ch_record(ChPid, LimiterPid), + update_ch_record(C#cr{acktags = queue:in(AckTag, ChAckTags)}), + ok. + +subtract_acks(ChPid, AckTags) -> + case lookup_ch(ChPid) of + not_found -> + not_found; + C = #cr{acktags = ChAckTags} -> + update_ch_record( + C#cr{acktags = subtract_acks(AckTags, [], ChAckTags)}), + ok + end. + +subtract_acks([], [], AckQ) -> + AckQ; +subtract_acks([], Prefix, AckQ) -> + queue:join(queue:from_list(lists:reverse(Prefix)), AckQ); +subtract_acks([T | TL] = AckTags, Prefix, AckQ) -> + case queue:out(AckQ) of + {{value, T}, QTail} -> subtract_acks(TL, Prefix, QTail); + {{value, AT}, QTail} -> subtract_acks(AckTags, [AT | Prefix], QTail) + end. + +possibly_unblock(Update, ChPid, State) -> + case lookup_ch(ChPid) of + not_found -> unchanged; + C -> C1 = Update(C), + case is_ch_blocked(C) andalso not is_ch_blocked(C1) of + false -> update_ch_record(C1), + unchanged; + true -> unblock(C1, State) + end + end. + +unblock(C = #cr{blocked_consumers = BlockedQ, limiter = Limiter}, + State = #state{consumers = Consumers, use = Use}) -> + case lists:partition( + fun({_P, {_ChPid, #consumer{tag = CTag}}}) -> + rabbit_limiter:is_consumer_blocked(Limiter, CTag) + end, priority_queue:to_list(BlockedQ)) of + {_, []} -> + update_ch_record(C), + unchanged; + {Blocked, Unblocked} -> + BlockedQ1 = priority_queue:from_list(Blocked), + UnblockedQ = priority_queue:from_list(Unblocked), + update_ch_record(C#cr{blocked_consumers = BlockedQ1}), + {unblocked, + tags(Unblocked), + State#state{consumers = priority_queue:join(Consumers, UnblockedQ), + use = update_use(Use, active)}} + end. + +resume_fun() -> + fun (C = #cr{limiter = Limiter}) -> + C#cr{limiter = rabbit_limiter:resume(Limiter)} + end. + +notify_sent_fun(Credit) -> + fun (C = #cr{unsent_message_count = Count}) -> + C#cr{unsent_message_count = Count - Credit} + end. + +activate_limit_fun() -> + fun (C = #cr{limiter = Limiter}) -> + C#cr{limiter = rabbit_limiter:activate(Limiter)} + end. + +credit_fun(IsEmpty, Credit, Drain, CTag) -> + fun (C = #cr{limiter = Limiter}) -> + C1 = C#cr{limiter = rabbit_limiter:credit( + Limiter, CTag, Credit, IsEmpty, Drain)}, + case Drain andalso IsEmpty of + true -> send_drained(C1); + false -> C1 + end + end. + +utilisation(#state{use = {active, Since, Avg}}) -> + use_avg(now_micros() - Since, 0, Avg); +utilisation(#state{use = {inactive, Since, Active, Avg}}) -> + use_avg(Active, now_micros() - Since, Avg). + +%%---------------------------------------------------------------------------- + +lookup_ch(ChPid) -> + case get({ch, ChPid}) of + undefined -> not_found; + C -> C + end. + +ch_record(ChPid, LimiterPid) -> + Key = {ch, ChPid}, + case get(Key) of + undefined -> MonitorRef = erlang:monitor(process, ChPid), + Limiter = rabbit_limiter:client(LimiterPid), + C = #cr{ch_pid = ChPid, + monitor_ref = MonitorRef, + acktags = queue:new(), + consumer_count = 0, + blocked_consumers = priority_queue:new(), + limiter = Limiter, + unsent_message_count = 0}, + put(Key, C), + C; + C = #cr{} -> C + end. + +update_ch_record(C = #cr{consumer_count = ConsumerCount, + acktags = ChAckTags, + unsent_message_count = UnsentMessageCount}) -> + case {queue:is_empty(ChAckTags), ConsumerCount, UnsentMessageCount} of + {true, 0, 0} -> ok = erase_ch_record(C); + _ -> ok = store_ch_record(C) + end, + C. + +store_ch_record(C = #cr{ch_pid = ChPid}) -> + put({ch, ChPid}, C), + ok. + +erase_ch_record(#cr{ch_pid = ChPid, monitor_ref = MonitorRef}) -> + erlang:demonitor(MonitorRef), + erase({ch, ChPid}), + ok. + +all_ch_record() -> [C || {{ch, _}, C} <- get()]. + +block_consumer(C = #cr{blocked_consumers = Blocked}, QEntry) -> + update_ch_record(C#cr{blocked_consumers = add_consumer(QEntry, Blocked)}). + +is_ch_blocked(#cr{unsent_message_count = Count, limiter = Limiter}) -> + Count >= ?UNSENT_MESSAGE_LIMIT orelse rabbit_limiter:is_suspended(Limiter). + +send_drained(C = #cr{ch_pid = ChPid, limiter = Limiter}) -> + case rabbit_limiter:drained(Limiter) of + {[], Limiter} -> C; + {CTagCredit, Limiter2} -> rabbit_channel:send_drained( + ChPid, CTagCredit), + C#cr{limiter = Limiter2} + end. + +tags(CList) -> [CTag || {_P, {_ChPid, #consumer{tag = CTag}}} <- CList]. + +add_consumer({ChPid, Consumer = #consumer{args = Args}}, Queue) -> + Priority = case rabbit_misc:table_lookup(Args, <<"x-priority">>) of + {_, P} -> P; + _ -> 0 + end, + priority_queue:in({ChPid, Consumer}, Priority, Queue). + +remove_consumer(ChPid, ConsumerTag, Queue) -> + priority_queue:filter(fun ({CP, #consumer{tag = CTag}}) -> + (CP /= ChPid) or (CTag /= ConsumerTag) + end, Queue). + +remove_consumers(ChPid, Queue) -> + priority_queue:filter(fun ({CP, _Consumer}) when CP =:= ChPid -> false; + (_) -> true + end, Queue). + +update_use({inactive, _, _, _} = CUInfo, inactive) -> + CUInfo; +update_use({active, _, _} = CUInfo, active) -> + CUInfo; +update_use({active, Since, Avg}, inactive) -> + Now = now_micros(), + {inactive, Now, Now - Since, Avg}; +update_use({inactive, Since, Active, Avg}, active) -> + Now = now_micros(), + {active, Now, use_avg(Active, Now - Since, Avg)}. + +use_avg(Active, Inactive, Avg) -> + Time = Inactive + Active, + Ratio = Active / Time, + Weight = erlang:min(1, Time / 1000000), + case Avg of + undefined -> Ratio; + _ -> Ratio * Weight + Avg * (1 - Weight) + end. + +now_micros() -> timer:now_diff(now(), {0,0,0}). |