Merge branch 'stable' into rabbitmq-server-1054

7 files changed, 38 insertions, 31 deletions

diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl
index dd64c6f1c8..daf2c167fa 100644
--- a/src/rabbit_alarm.erl
+++ b/src/rabbit_alarm.erl
@@ -16,7 +16,7 @@
 %% There are two types of alarms handled by this module:
 %%
 %% * per-node resource (disk, memory) alarms for the whole cluster. If any node
-%%   has an alarm, then all publishing should be disabled througout the
+%%   has an alarm, then all publishing should be disabled across the
 %%   cluster until all alarms clear. When a node sets such an alarm,
 %%   this information is automatically propagated throughout the cluster.
 %%   `#alarms.alarmed_nodes' is being used to track this type of alarms.
diff --git a/src/rabbit_cli.erl b/src/rabbit_cli.erl
index c0e5c93247..1feda43b6e 100644
--- a/src/rabbit_cli.erl
+++ b/src/rabbit_cli.erl
@@ -154,7 +154,7 @@ start_distribution_anon(TriesLeft, _) ->
             start_distribution_anon(TriesLeft - 1, Reason)
     end.
 
-%% Tries to start distribution with random name choosen from limited list of candidates - to
+%% Tries to start distribution with random name chosen from limited list of candidates - to
 %% prevent atom table pollution on target nodes.
 start_distribution() ->
     rabbit_nodes:ensure_epmd(),
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 6bb7e093f6..c420838312 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -858,7 +858,7 @@ check_rabbit_consistency(Remote) ->
 %% that a `reset' would leave it in.  We cannot simply check if the
 %% mnesia tables aren't there because restarted RAM nodes won't have
 %% tables while still being non-virgin.  What we do instead is to
-%% check if the mnesia directory is non existant or empty, with the
+%% check if the mnesia directory is non existent or empty, with the
 %% exception of the cluster status files, which will be there thanks to
 %% `rabbit_node_monitor:prepare_cluster_status_file/0'.
 is_virgin_node() ->
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index d3ff077c8b..5e2e43a73d 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -265,7 +265,7 @@
 %% updated.
 %%
 %% On non-clean startup, we scan the files we discover, dealing with
-%% the possibilites of a crash having occured during a compaction
+%% the possibilites of a crash having occurred during a compaction
 %% (this consists of tidyup - the compaction is deliberately designed
 %% such that data is duplicated on disk rather than risking it being
 %% lost), and rebuild the file summary and index ETS table.
@@ -310,7 +310,7 @@
 %% From this reasoning, we do have a bound on the number of times the
 %% message is rewritten. From when it is inserted, there can be no
 %% files inserted between it and the head of the queue, and the worst
-%% case is that everytime it is rewritten, it moves one position lower
+%% case is that every time it is rewritten, it moves one position lower
 %% in the file (for it to stay at the same position requires that
 %% there are no holes beneath it, which means truncate would be used
 %% and so it would not be rewritten at all). Thus this seems to
@@ -352,7 +352,7 @@
 %% because in the event of the same message being sent to several
 %% different queues, there is the possibility of one queue writing and
 %% removing the message before other queues write it at all. Thus
-%% accomodating 0-reference counts allows us to avoid unnecessary
+%% accommodating 0-reference counts allows us to avoid unnecessary
 %% writes here. Of course, there are complications: the file to which
 %% the message has already been written could be locked pending
 %% deletion or GC, which means we have to rewrite the message as the
diff --git a/src/rabbit_plugins.erl b/src/rabbit_plugins.erl
index bde4ff5388..b408b7e613 100644
--- a/src/rabbit_plugins.erl
+++ b/src/rabbit_plugins.erl
@@ -15,7 +15,7 @@
 %%
 
 -module(rabbit_plugins).
--include("rabbit.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([setup/0, active/0, read_enabled/1, list/1, list/2, dependencies/3]).
 -export([ensure/1]).
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index 6a14854882..6560d61625 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -123,7 +123,7 @@
 -define(SEGMENT_EXTENSION, ".idx").
 
 %% TODO: The segment size would be configurable, but deriving all the
-%% other values is quite hairy and quite possibly noticably less
+%% other values is quite hairy and quite possibly noticeably less
 %% efficient, depending on how clever the compiler is when it comes to
 %% binary generation/matching with constant vs variable lengths.
 
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index dd92256146..ae4603493c 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -432,21 +432,28 @@
 %% rabbit_amqqueue_process need fairly fresh rates.
 -define(MSGS_PER_RATE_CALC, 100).
 
-
 %% we define the garbage collector threshold
-%% it needs to tune the GC calls inside `reduce_memory_use`
-%% see: rabbitmq-server-973 and `maybe_execute_gc` function
--define(DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD, 250).
--define(EXPLICIT_GC_RUN_OP_THRESHOLD,
+%% it needs to tune the `reduce_memory_use` calls. Thus, the garbage collection.
+%% see: rabbitmq-server-973 and rabbitmq-server-964
+-define(DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD, 1000).
+-define(EXPLICIT_GC_RUN_OP_THRESHOLD(Mode),
     case get(explicit_gc_run_operation_threshold) of
         undefined ->
-            Val = rabbit_misc:get_env(rabbit, lazy_queue_explicit_gc_run_operation_threshold,
-                ?DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD),
+            Val = explicit_gc_run_operation_threshold_for_mode(Mode),
             put(explicit_gc_run_operation_threshold, Val),
             Val;
         Val       -> Val
     end).
 
+explicit_gc_run_operation_threshold_for_mode(Mode) ->
+    {Key, Fallback} = case Mode of
+        lazy -> {lazy_queue_explicit_gc_run_operation_threshold,
+                 ?DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD};
+        _    -> {queue_explicit_gc_run_operation_threshold,
+                 ?DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD}
+    end,
+    rabbit_misc:get_env(rabbit, Key, Fallback).
+
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
@@ -586,27 +593,27 @@ publish(Msg, MsgProps, IsDelivered, ChPid, Flow, State) ->
         publish1(Msg, MsgProps, IsDelivered, ChPid, Flow,
                  fun maybe_write_to_disk/4,
                  State),
-    a(reduce_memory_use(maybe_update_rates(State1))).
+    a(maybe_reduce_memory_use(maybe_update_rates(State1))).
 
 batch_publish(Publishes, ChPid, Flow, State) ->
     {ChPid, Flow, State1} =
         lists:foldl(fun batch_publish1/2, {ChPid, Flow, State}, Publishes),
     State2 = ui(State1),
-    a(reduce_memory_use(maybe_update_rates(State2))).
+    a(maybe_reduce_memory_use(maybe_update_rates(State2))).
 
 publish_delivered(Msg, MsgProps, ChPid, Flow, State) ->
     {SeqId, State1} =
         publish_delivered1(Msg, MsgProps, ChPid, Flow,
                            fun maybe_write_to_disk/4,
                            State),
-    {SeqId, a(reduce_memory_use(maybe_update_rates(State1)))}.
+    {SeqId, a(maybe_reduce_memory_use(maybe_update_rates(State1)))}.
 
 batch_publish_delivered(Publishes, ChPid, Flow, State) ->
     {ChPid, Flow, SeqIds, State1} =
         lists:foldl(fun batch_publish_delivered1/2,
                     {ChPid, Flow, [], State}, Publishes),
     State2 = ui(State1),
-    {lists:reverse(SeqIds), a(reduce_memory_use(maybe_update_rates(State2)))}.
+    {lists:reverse(SeqIds), a(maybe_reduce_memory_use(maybe_update_rates(State2)))}.
 
 discard(_MsgId, _ChPid, _Flow, State) -> State.
 
@@ -710,7 +717,7 @@ requeue(AckTags, #vqstate { mode       = default,
     {Delta1, MsgIds2, State3}       = delta_merge(SeqIds1, Delta, MsgIds1,
                                                   State2),
     MsgCount = length(MsgIds2),
-    {MsgIds2, a(reduce_memory_use(
+    {MsgIds2, a(maybe_reduce_memory_use(
                   maybe_update_rates(ui(
                     State3 #vqstate { delta      = Delta1,
                                       q3         = Q3a,
@@ -728,7 +735,7 @@ requeue(AckTags, #vqstate { mode       = lazy,
     {Delta1, MsgIds1, State2}     = delta_merge(SeqIds, Delta, MsgIds,
                                                 State1),
     MsgCount = length(MsgIds1),
-    {MsgIds1, a(reduce_memory_use(
+    {MsgIds1, a(maybe_reduce_memory_use(
                   maybe_update_rates(ui(
                     State2 #vqstate { delta      = Delta1,
                                       q3         = Q3a,
@@ -778,7 +785,7 @@ set_ram_duration_target(
           (TargetRamCount =/= infinity andalso
            TargetRamCount1 >= TargetRamCount) of
           true  -> State1;
-          false -> reduce_memory_use(State1)
+          false -> maybe_reduce_memory_use(State1)
       end).
 
 maybe_update_rates(State = #vqstate{ in_counter  = InCount,
@@ -860,7 +867,7 @@ timeout(State = #vqstate { index_state = IndexState }) ->
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
 
-resume(State) -> a(reduce_memory_use(State)).
+resume(State) -> a(maybe_reduce_memory_use(State)).
 
 msg_rates(#vqstate { rates = #rates { in  = AvgIngressRate,
                                       out = AvgEgressRate } }) ->
@@ -2310,12 +2317,12 @@ ifold(Fun, Acc, Its, State) ->
 %% Phase changes
 %%----------------------------------------------------------------------------
 
-maybe_execute_gc(State = #vqstate {memory_reduction_run_count = MRedRunCount}) ->
-    case MRedRunCount >= ?EXPLICIT_GC_RUN_OP_THRESHOLD of
-	true -> garbage_collect(),
-		 State#vqstate{memory_reduction_run_count =  0};
-        false ->    State#vqstate{memory_reduction_run_count =  MRedRunCount + 1}
-
+maybe_reduce_memory_use(State = #vqstate {memory_reduction_run_count = MRedRunCount,
+                                          mode = Mode}) ->
+    case MRedRunCount >= ?EXPLICIT_GC_RUN_OP_THRESHOLD(Mode) of
+	true -> State1 = reduce_memory_use(State),
+                State1#vqstate{memory_reduction_run_count =  0};
+        false -> State#vqstate{memory_reduction_run_count =  MRedRunCount + 1}
     end.
 
 reduce_memory_use(State = #vqstate { target_ram_count = infinity }) ->
@@ -2330,7 +2337,6 @@ reduce_memory_use(State = #vqstate {
                                                 out     = AvgEgress,
                                                 ack_in  = AvgAckIngress,
                                                 ack_out = AvgAckEgress } }) ->
-
     State1 = #vqstate { q2 = Q2, q3 = Q3 } =
         case chunk_size(RamMsgCount + gb_trees:size(RPA), TargetRamCount) of
             0  -> State;
@@ -2390,7 +2396,8 @@ reduce_memory_use(State = #vqstate {
             S2 ->
                 push_betas_to_deltas(S2, State1)
         end,
-    maybe_execute_gc(State3).
+    garbage_collect(),
+    State3.
 
 limit_ram_acks(0, State) ->
     {0, ui(State)};