Force GM garbage collection every 250ms

In high throughput scenarios, e.g. basic.reject or basic.nack, messages
which belong to a mirrored queue and replicated in a GM group are
promoted to the old heap quickly. This means that garbage collection
happens only when the VM is under memory pressure, which might be too
late. When the system is under pressure, garbage collection slows it
down even further, to the point that RabbitMQ nodes can run out of
memory and crash. We want the GM process to garbage collect binaries a
lot more often, i.e. every 250ms. There is precedence for explicit GC in
the variable queue, which we do for the same reason.

Initially, we wanted to use the number of messages as the trigger for
garbage collection, but we soon discovered that different workloads
(e.g. small vs large messages) would result in unpredictable and
sub-optimal garbage collections.

Before setting fullsweep_after to 0, memory usage was 2x higher (400MB
vs 200MB) and throughput was 0.1x lower (45k vs 50k). With this setting,
there is a single heap generation resulting in a quicker GC.

The RabbitMQ deployment that this fix was tested against is:
* AWS, c4.2xlarge
* 3 nodes of RabbitMQ 3.6.11 with OTP 20.0.1
* 3 durable & auto-delete queues with 3 replicas each
* each queue master was defined on a different RabbitMQ node
* 1 consumer per queue with QOS 100
* 100 durable messages @ 1KiB each
* basic.reject operations

| Node | Message throughput | Memory usage |
|------|--------------------|--------------|
| rmq0 | 20K msg/s          | 700 MB       |
| rmq1 | 14K msg/s          | 900 MB       |
| rmq2 | 15K msg/s          | 800 MB       |

[#148892851]

Signed-off-by: Gerhard Lazu <[email protected]>

Author: dumbbell

src/gm.erl

@@ -395,9 +395,8 @@
 
 -define(GROUP_TABLE, gm_group).
 -define(MAX_BUFFER_SIZE, 100000000). %% 100MB
--define(HIBERNATE_AFTER_MIN, 1000).
--define(DESIRED_HIBERNATE, 10000).
 -define(BROADCAST_TIMER, 25).
+-define(FORCE_GC_TIMER, 250).
 -define(VERSION_START, 0).
 -define(SETS, ordsets).
 -define(DICT, orddict).
@@ -416,6 +415,7 @@
           broadcast_buffer,
           broadcast_buffer_sz,
           broadcast_timer,
+          force_gc_timer,
           txn_executor,
           shutting_down
         }).
@@ -508,7 +508,8 @@ table_definitions() ->
     [{Name, [?TABLE_MATCH | Attributes]}].
 
 start_link(GroupName, Module, Args, TxnFun) ->
-    gen_server2:start_link(?MODULE, [GroupName, Module, Args, TxnFun], []).
+    gen_server2:start_link(?MODULE, [GroupName, Module, Args, TxnFun],
+                       [{spawn_opt, [{fullsweep_after, 0}]}]).
 
 leave(Server) ->
     gen_server2:cast(Server, leave).
@@ -551,6 +552,7 @@ init([GroupName, Module, Args, TxnFun]) ->
                   broadcast_buffer    = [],
                   broadcast_buffer_sz = 0,
                   broadcast_timer     = undefined,
+                  force_gc_timer      = undefined,
                   txn_executor        = TxnFun,
                   shutting_down       = false }}.
 
@@ -707,6 +709,10 @@ handle_cast(leave, State) ->
     {stop, normal, State}.
 
 
+handle_info(force_gc, State) ->
+    garbage_collect(),
+    noreply(State #state { force_gc_timer = undefined });
+
 handle_info(flush, State) ->
     noreply(
       flush_broadcast_buffer(State #state { broadcast_timer = undefined }));
@@ -882,14 +888,24 @@ handle_msg({activity, _NotLeft, _Activity}, State) ->
 
 
 noreply(State) ->
-    {noreply, ensure_broadcast_timer(State), flush_timeout(State)}.
+    {noreply, ensure_timers(State), flush_timeout(State)}.
 
 reply(Reply, State) ->
-    {reply, Reply, ensure_broadcast_timer(State), flush_timeout(State)}.
+    {reply, Reply, ensure_timers(State), flush_timeout(State)}.
+
+ensure_timers(State) ->
+    ensure_force_gc_timer(ensure_broadcast_timer(State)).
 
 flush_timeout(#state{broadcast_buffer = []}) -> infinity;
 flush_timeout(_)                             -> 0.
 
+ensure_force_gc_timer(State = #state { force_gc_timer = TRef })
+  when TRef =/= undefined ->
+    State;
+ensure_force_gc_timer(State = #state { force_gc_timer = undefined }) ->
+    TRef = erlang:send_after(?FORCE_GC_TIMER, self(), force_gc),
+    State #state { force_gc_timer = TRef }.
+
 ensure_broadcast_timer(State = #state { broadcast_buffer = [],
                                         broadcast_timer  = undefined }) ->
     State;
@@ -957,8 +973,7 @@ flush_broadcast_buffer(State = #state { self             = Self,
                       end, Self, MembersState),
     State #state { members_state       = MembersState1,
                    broadcast_buffer    = [],
-                   broadcast_buffer_sz = 0}.
-
+                   broadcast_buffer_sz = 0 }.
 
 %% ---------------------------------------------------------------------------
 %% View construction and inspection