only trigger stats emission from writers in the serverbug23225

(and not the Erlang client)

The server (and *only* the server) starts channels with
rabbit_channel_sup:start_link({tcp, ...}). Writers are created as part
of that. We pass an extra argument to rabbit_writer:start_link in
this case, indicating that the reader wants be told by the writer to
emit stats.

Internally, the writer handles the stats/no_stats distinction by
initialising the stats timer differently. We introduce a new function
to create a disabled stats timer, thus taking advantage of all the
existing logic in rabbit_event that suppresses timer/stats operations
when stats are disabled.

This approach requires inclusion of rabbit_event in rabbit_common.

3 files changed, 39 insertions, 16 deletions

diff --git a/src/rabbit_channel_sup.erl b/src/rabbit_channel_sup.erl
index bcb83851..42459833 100644
--- a/src/rabbit_channel_sup.erl
+++ b/src/rabbit_channel_sup.erl
@@ -83,7 +83,7 @@ init(Type) ->
 
 child_specs({tcp, Sock, Channel, FrameMax, ReaderPid, Protocol}) ->
     [{writer, {rabbit_writer, start_link,
-               [Sock, Channel, FrameMax, Protocol, ReaderPid]},
+               [Sock, Channel, FrameMax, Protocol, ReaderPid, true]},
       intrinsic, ?MAX_WAIT, worker, [rabbit_writer]} | child_specs(direct)];
 child_specs(direct) ->
     [{limiter, {rabbit_limiter, start_link, []},
diff --git a/src/rabbit_event.erl b/src/rabbit_event.erl
index 3f1b20fe..7d91b6fa 100644
--- a/src/rabbit_event.erl
+++ b/src/rabbit_event.erl
@@ -19,8 +19,8 @@
 -include("rabbit.hrl").
 
 -export([start_link/0]).
--export([init_stats_timer/2, ensure_stats_timer/3, stop_stats_timer/2]).
--export([reset_stats_timer/2]).
+-export([init_stats_timer/2, init_disabled_stats_timer/2,
+         ensure_stats_timer/3, stop_stats_timer/2, reset_stats_timer/2]).
 -export([stats_level/2, if_enabled/3]).
 -export([notify/2, notify_if/3]).
 
@@ -51,6 +51,7 @@
 
 -spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
 -spec(init_stats_timer/2 :: (container(), pos()) -> container()).
+-spec(init_disabled_stats_timer/2 :: (container(), pos()) -> container()).
 -spec(ensure_stats_timer/3 :: (container(), pos(), term()) -> container()).
 -spec(stop_stats_timer/2 :: (container(), pos()) -> container()).
 -spec(reset_stats_timer/2 :: (container(), pos()) -> container()).
@@ -90,10 +91,13 @@ start_link() ->
 
 init_stats_timer(C, P) ->
     {ok, StatsLevel} = application:get_env(rabbit, collect_statistics),
-    {ok, Interval} = application:get_env(rabbit, collect_statistics_interval),
+    {ok, Interval}   = application:get_env(rabbit, collect_statistics_interval),
     setelement(P, C, #state{level = StatsLevel, interval = Interval,
                             timer = undefined}).
 
+init_disabled_stats_timer(C, P) ->
+    setelement(P, C, #state{level = none, interval = 0, timer = undefined}).
+
 ensure_stats_timer(C, P, Msg) ->
     case element(P, C) of
         #state{level = Level, interval = Interval, timer = undefined} = State
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index c23f46f9..a7ea3d99 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -18,7 +18,7 @@
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 
--export([start/5, start_link/5]).
+-export([start/5, start_link/5, start/6, start_link/6]).
 -export([send_command/2, send_command/3,
          send_command_sync/2, send_command_sync/3,
          send_command_and_notify/4, send_command_and_notify/5]).
@@ -44,6 +44,14 @@
         (rabbit_net:socket(), rabbit_channel:channel_number(),
          non_neg_integer(), rabbit_types:protocol(), pid())
         -> rabbit_types:ok(pid())).
+-spec(start/6 ::
+        (rabbit_net:socket(), rabbit_channel:channel_number(),
+         non_neg_integer(), rabbit_types:protocol(), pid(), boolean())
+        -> rabbit_types:ok(pid())).
+-spec(start_link/6 ::
+        (rabbit_net:socket(), rabbit_channel:channel_number(),
+         non_neg_integer(), rabbit_types:protocol(), pid(), boolean())
+        -> rabbit_types:ok(pid())).
 -spec(send_command/2 ::
         (pid(), rabbit_framing:amqp_method_record()) -> 'ok').
 -spec(send_command/3 ::
@@ -76,21 +84,32 @@
 %%---------------------------------------------------------------------------
 
 start(Sock, Channel, FrameMax, Protocol, ReaderPid) ->
-    State = initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid),
-    {ok, proc_lib:spawn(?MODULE, mainloop, [State])}.
+    start(Sock, Channel, FrameMax, Protocol, ReaderPid, false).
 
 start_link(Sock, Channel, FrameMax, Protocol, ReaderPid) ->
-    State = initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid),
+    start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, false).
+
+start(Sock, Channel, FrameMax, Protocol, ReaderPid, ReaderWantsStats) ->
+    State = initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid,
+                          ReaderWantsStats),
+    {ok, proc_lib:spawn(?MODULE, mainloop, [State])}.
+
+start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, ReaderWantsStats) ->
+    State = initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid,
+                          ReaderWantsStats),
     {ok, proc_lib:spawn_link(?MODULE, mainloop, [State])}.
 
-initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid) ->
-    rabbit_event:init_stats_timer(#wstate{sock      = Sock,
-                                          channel   = Channel,
-                                          frame_max = FrameMax,
-                                          protocol  = Protocol,
-                                          reader    = ReaderPid,
-                                          pending   = []},
-                                  #wstate.stats_timer).
+initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid, ReaderWantsStats) ->
+    (case ReaderWantsStats of
+         true  -> fun rabbit_event:init_stats_timer/2;
+         false -> fun rabbit_event:init_disabled_stats_timer/2
+     end)(#wstate{sock      = Sock,
+                  channel   = Channel,
+                  frame_max = FrameMax,
+                  protocol  = Protocol,
+                  reader    = ReaderPid,
+                  pending   = []},
+          #wstate.stats_timer).
 
 mainloop(State) ->
     try