diff options
author | Matthias Radestock <matthias@rabbitmq.com> | 2014-01-07 20:08:22 +0000 |
---|---|---|
committer | Matthias Radestock <matthias@rabbitmq.com> | 2014-01-07 20:08:22 +0000 |
commit | 5efc04587e24205fe1fd0e2c31d501c7a2bf83e8 (patch) | |
tree | 0c35c5d554cc0583b3e8fe9f3d0ba19275638aeb | |
parent | 02a1457f8bef8adce0d16de0b0e976f065daceb2 (diff) | |
download | rabbitmq-server-5efc04587e24205fe1fd0e2c31d501c7a2bf83e8.tar.gz |
make rabbit_queue_consumers:deliver deliver at most one message
...which makes the control flow more obvious
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 106 | ||||
-rw-r--r-- | src/rabbit_queue_consumers.erl | 57 |
2 files changed, 88 insertions, 75 deletions
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 281aecb9..97c7a346 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -407,15 +407,6 @@ maybe_send_drained(WasEmpty, State) -> end, State. -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; confirm_messages(MsgIds, State = #q{msg_id_to_channel = MTC}) -> @@ -461,49 +452,72 @@ discard(#delivery{sender = SenderPid, BQS1 = BQ:discard(MsgId, SenderPid, BQS), State1#q{backing_queue_state = BQS1}. -run_message_queue(State) -> - {_Active, State3} = deliver_msgs_to_consumers( - fun(AckRequired, State1) -> - {Result, State2} = fetch(AckRequired, State1), - {Result, is_empty(State2), State2} - end, is_empty(State), State), - State3. +run_message_queue(State) -> run_message_queue([], State). + +run_message_queue(Blocked, State) -> + case is_empty(State) of + true -> notify_decorators(lists:append(Blocked), State); + false -> case rabbit_queue_consumers:deliver( + fun(AckRequired, State1) -> + fetch(AckRequired, State1) + end, qname(State), State, State#q.consumers) of + {delivered, MoreBlocked, State2, Consumers} -> + run_message_queue([MoreBlocked | Blocked], + State2#q{consumers = Consumers}); + {undelivered, MoreBlocked, Consumers} -> + notify_decorators( + lists:append([MoreBlocked | Blocked]), + State#q{consumers = Consumers}) + end + end. attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message}, - Props, Delivered, State = #q{backing_queue = BQ, - backing_queue_state = BQS}) -> - {IsDuplicate, BQS1} = BQ:is_duplicate(Message, BQS), - State1 = State#q{backing_queue_state = BQS1}, - case IsDuplicate of - false -> deliver_msgs_to_consumers( - fun (true, State2 = #q{backing_queue_state = BQS2}) -> - true = BQ:is_empty(BQS2), - {AckTag, BQS3} = BQ:publish_delivered( - Message, Props, SenderPid, BQS2), - {{Message, Delivered, AckTag}, - true, State2#q{backing_queue_state = BQS3}}; - (false, State2) -> - {{Message, Delivered, undefined}, - true, discard(Delivery, State2)} - end, false, State1); - true -> {true, State1} + Props, Delivered, State) -> + case rabbit_queue_consumers:deliver( + fun (true, State2 = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> + true = BQ:is_empty(BQS), + {AckTag, BQS1} = BQ:publish_delivered( + Message, Props, SenderPid, BQS), + {{Message, Delivered, AckTag}, + State2#q{backing_queue_state = BQS1}}; + (false, State2) -> + {{Message, Delivered, undefined}, discard(Delivery, State2)} + end, qname(State), State, State#q.consumers) of + {delivered, Blocked, State1, Consumers} -> + {delivered, notify_decorators( + Blocked, State1#q{consumers = Consumers})}; + {undelivered, Blocked, Consumers} -> + {undelivered, notify_decorators( + Blocked, State#q{consumers = Consumers})} end. +notify_decorators(Blocked, State) -> + [notify_decorators(consumer_blocked, [{consumer_tag, CTag}], State) || + {_ChPid, CTag} <- Blocked], + State. + deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid}, - Delivered, State) -> + Delivered, State = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> {Confirm, State1} = send_or_record_confirm(Delivery, State), Props = message_properties(Message, Confirm, State), - case attempt_delivery(Delivery, Props, Delivered, State1) of - {true, State2} -> + {IsDuplicate, BQS1} = BQ:is_duplicate(Message, BQS), + State2 = State1#q{backing_queue_state = BQS1}, + case IsDuplicate orelse attempt_delivery(Delivery, Props, Delivered, + State2) of + true -> State2; + {delivered, State3} -> + State3; %% The next one is an optimisation - {false, State2 = #q{ttl = 0, dlx = undefined}} -> - discard(Delivery, State2); - {false, State2 = #q{backing_queue = BQ, backing_queue_state = BQS}} -> - BQS1 = BQ:publish(Message, Props, Delivered, SenderPid, BQS), - {Dropped, State3 = #q{backing_queue_state = BQS2}} = - maybe_drop_head(State2#q{backing_queue_state = BQS1}), - QLen = BQ:len(BQS2), + {undelivered, State3 = #q{ttl = 0, dlx = undefined}} -> + discard(Delivery, State3); + {undelivered, State3 = #q{backing_queue_state = BQS2}} -> + BQS3 = BQ:publish(Message, Props, Delivered, SenderPid, BQS2), + {Dropped, State4 = #q{backing_queue_state = BQS4}} = + maybe_drop_head(State3#q{backing_queue_state = BQS3}), + QLen = BQ:len(BQS4), %% optimisation: it would be perfectly safe to always %% invoke drop_expired_msgs here, but that is expensive so %% we only do that if a new message that might have an @@ -512,9 +526,9 @@ deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid}, %% has no expiry and becomes the head of the queue then %% the call is unnecessary. case {Dropped > 0, QLen =:= 1, Props#message_properties.expiry} of - {false, false, _} -> State3; - {true, true, undefined} -> State3; - {_, _, _} -> drop_expired_msgs(State3) + {false, false, _} -> State4; + {true, true, undefined} -> State4; + {_, _, _} -> drop_expired_msgs(State4) end end. diff --git a/src/rabbit_queue_consumers.erl b/src/rabbit_queue_consumers.erl index 702091dc..38b1bad9 100644 --- a/src/rabbit_queue_consumers.erl +++ b/src/rabbit_queue_consumers.erl @@ -18,7 +18,7 @@ -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, + send_drained/0, deliver/4, 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]). @@ -77,8 +77,9 @@ 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()}. + rabbit_amqqueue:name(), T, state()) -> + {'delivered', [{ch(), rabbit_types:ctag()}], T, state()} | + {'undelivered', [{ch(), rabbit_types:ctag()}], state()}. -spec record_ack(ch(), pid(), ack()) -> 'ok'. -spec subtract_acks(ch(), [ack()]) -> 'not_found' | 'ok'. -spec possibly_unblock(cr_fun(), ch(), state()) -> @@ -181,44 +182,42 @@ erase_ch(ChPid, State = #state{consumers = Consumers}) -> 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, QName, S, State) -> + deliver(FetchFun, 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}) -> +deliver(FetchFun, QName, Blocked, S, State = #state{consumers = Consumers}) -> 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}) + {undelivered, Blocked, + State#state{use = update_use(State#state.use, inactive)}}; + {{value, QEntry = {ChPid, Consumer}, Priority}, Tail} -> + case deliver_to_consumer(FetchFun, QEntry, QName, S) of + {delivered, S1} -> + {delivered, Blocked, S1, + State#state{consumers = priority_queue:in(QEntry, Priority, + Tail)}}; + undelivered -> + deliver(FetchFun, QName, + [{ChPid, Consumer#consumer.tag} | Blocked], S, + State#state{consumers = Tail}) + end end. -deliver_to_consumer(FetchFun, E = {ChPid, Consumer}, Priority, QName, - Blocked, S, Consumers) -> +deliver_to_consumer(FetchFun, E = {ChPid, Consumer}, QName, S) -> 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}; + undelivered; 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}; + undelivered; {continue, Limiter} -> - {Stop, S1} = deliver_to_consumer( - FetchFun, Consumer, - C#cr{limiter = Limiter}, QName, S), - {Stop, Blocked, S1, - priority_queue:in(E, Priority, Consumers)} + {delivered, deliver_to_consumer( + FetchFun, Consumer, + C#cr{limiter = Limiter}, QName, S)} end end. @@ -229,7 +228,7 @@ deliver_to_consumer(FetchFun, acktags = ChAckTags, unsent_message_count = Count}, QName, S) -> - {{Message, IsDelivered, AckTag}, Stop, S1} = FetchFun(AckRequired, S), + {{Message, IsDelivered, AckTag}, S1} = FetchFun(AckRequired, S), rabbit_channel:deliver(ChPid, ConsumerTag, AckRequired, {QName, self(), AckTag, IsDelivered, Message}), ChAckTags1 = case AckRequired of @@ -238,7 +237,7 @@ deliver_to_consumer(FetchFun, end, update_ch_record(C#cr{acktags = ChAckTags1, unsent_message_count = Count + 1}), - {Stop, S1}. + S1. record_ack(ChPid, LimiterPid, AckTag) -> C = #cr{acktags = ChAckTags} = ch_record(ChPid, LimiterPid), |