make rabbit_queue_consumers:deliver deliver at most one message

...which makes the control flow more obvious

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),