Merge branch 'stable'

Conflicts:
	src/rabbit_channel.erl
	src/rabbit_control_main.erl

Author: michaelklishin

src/rabbit_channel.erl

@@ -183,10 +183,21 @@
 
 -define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]).
 
--define(INCR_STATS(Incs, Measure, State),
+-define(INCR_STATS(Type, Key, Inc, Measure, State),
         case rabbit_event:stats_level(State, #ch.stats_timer) of
-            fine -> incr_stats(Incs, Measure);
-            _    -> ok
+            fine ->
+                rabbit_core_metrics:channel_stats(Type, Measure, {self(), Key}, Inc),
+                %% Keys in the process dictionary are used to clean up the core metrics
+                put({Type, Key}, none);
+            _ ->
+                ok
+        end).
+
+-define(INCR_STATS(Type, Key, Inc, Measure),
+        begin
+            rabbit_core_metrics:channel_stats(Type, Measure, {self(), Key}, Inc),
+            %% Keys in the process dictionary are used to clean up the core metrics
+            put({Type, Key}, none)
         end).
 
 %%----------------------------------------------------------------------------
@@ -1611,7 +1622,7 @@ basic_return(#basic_message{exchange_name = ExchangeName,
                             content       = Content},
              State = #ch{protocol = Protocol, writer_pid = WriterPid},
              Reason) ->
-    ?INCR_STATS([{exchange_stats, ExchangeName, 1}], return_unroutable, State),
+    ?INCR_STATS(exchange_stats, ExchangeName, 1, return_unroutable, State),
     {_Close, ReplyCode, ReplyText} = Protocol:lookup_amqp_exception(Reason),
     ok = rabbit_writer:send_command(
            WriterPid,
@@ -1648,14 +1659,14 @@ record_sent(ConsumerTag, AckRequired,
                         user              = #user{username = Username},
                         conn_name         = ConnName,
                         channel           = ChannelNum}) ->
-    ?INCR_STATS([{queue_stats, QName, 1}], case {ConsumerTag, AckRequired} of
-                                               {none,  true} -> get;
-                                               {none, false} -> get_no_ack;
-                                               {_   ,  true} -> deliver;
-                                               {_   , false} -> deliver_no_ack
-                                           end, State),
+    ?INCR_STATS(queue_stats, QName, 1, case {ConsumerTag, AckRequired} of
+                                           {none,  true} -> get;
+                                           {none, false} -> get_no_ack;
+                                           {_   ,  true} -> deliver;
+                                           {_   , false} -> deliver_no_ack
+                                       end, State),
     case Redelivered of
-        true  -> ?INCR_STATS([{queue_stats, QName, 1}], redeliver, State);
+        true  -> ?INCR_STATS(queue_stats, QName, 1, redeliver, State);
         false -> ok
     end,
     rabbit_trace:tap_out(Msg, ConnName, ChannelNum, Username, TraceState),
@@ -1700,11 +1711,11 @@ ack(Acked, State = #ch{queue_names = QNames}) ->
     foreach_per_queue(
       fun (QPid, MsgIds) ->
               ok = rabbit_amqqueue:ack(QPid, MsgIds, self()),
-              ?INCR_STATS(case maps:find(QPid, QNames) of
-                              {ok, QName} -> Count = length(MsgIds),
-                                             [{queue_stats, QName, Count}];
-                              error       -> []
-                          end, ack, State)
+              case maps:find(QPid, QNames) of
+                  {ok, QName} -> Count = length(MsgIds),
+                                 ?INCR_STATS(queue_stats, QName, Count, ack, State);
+                  error       -> ok
+              end
       end, Acked),
     ok = notify_limiter(State#ch.limiter, Acked).
 
@@ -1769,7 +1780,7 @@ deliver_to_queues({#delivery{message   = #basic_message{exchange_name = XName},
                              confirm   = false,
                              mandatory = false},
                    []}, State) -> %% optimisation
-    ?INCR_STATS([{exchange_stats, XName, 1}], publish, State),
+    ?INCR_STATS(exchange_stats, XName, 1, publish, State),
     State;
 deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{
                                                        exchange_name = XName},
@@ -1806,11 +1817,15 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{
                                        Message, State1),
     State3 = process_routing_confirm(  Confirm,   DeliveredQPids, MsgSeqNo,
                                        XName,   State2),
-    ?INCR_STATS([{exchange_stats, XName, 1} |
-                 [{queue_exchange_stats, {QName, XName}, 1} ||
-                     QPid        <- DeliveredQPids,
-                     {ok, QName} <- [maps:find(QPid, QNames1)]]],
-                publish, State3),
+    case rabbit_event:stats_level(State3, #ch.stats_timer) of
+        fine ->
+            ?INCR_STATS(exchange_stats, XName, 1, publish),
+            [?INCR_STATS(queue_exchange_stats, {QName, XName}, 1, publish) ||
+                QPid        <- DeliveredQPids,
+                {ok, QName} <- [maps:find(QPid, QNames1)]];
+        _ ->
+            ok
+    end,
     State3.
 
 process_routing_mandatory(false,     _, _MsgSeqNo, _Msg, State) ->
@@ -1853,7 +1868,7 @@ send_confirms(State = #ch{tx = none, confirmed = C}) ->
         ok      -> MsgSeqNos =
                        lists:foldl(
                          fun ({MsgSeqNo, XName}, MSNs) ->
-                                 ?INCR_STATS([{exchange_stats, XName, 1}],
+                                 ?INCR_STATS(exchange_stats, XName, 1,
                                              confirm, State),
                                  [MsgSeqNo | MSNs]
                          end, [], lists:append(C)),
@@ -1953,17 +1968,14 @@ i(Item, _) ->
 name(#ch{conn_name = ConnName, channel = Channel}) ->
     list_to_binary(rabbit_misc:format("~s (~p)", [ConnName, Channel])).
 
-incr_stats(Incs, Measure) ->
-    [begin
-         rabbit_core_metrics:channel_stats(Type, Measure, {self(), Key}, Inc),
-         %% Keys in the process dictionary are used to clean up the core metrics
-         put({Type, Key}, none)
-     end || {Type, Key, Inc} <- Incs].
-
 emit_stats(State) -> emit_stats(State, []).
 
 emit_stats(State, Extra) ->
     [{reductions, Red} | Coarse0] = infos(?STATISTICS_KEYS, State),
+    %% First metric must be `idle_since` (if available), as expected by
+    %% `rabbit_mgmt_format:format_channel_stats`. This is a performance
+    %% optimisation that avoids traversing the whole list when only
+    %% one element has to be formatted.
     rabbit_core_metrics:channel_stats(self(), Extra ++ Coarse0),
     rabbit_core_metrics:channel_stats(reductions, self(), Red).