1. A bugfix in MM, in internal_update. 2. In MM only enforce limits if we're using more than half the available RAM. 3. In VQ, sum the ingress and egress rates - this makes things much smoother when queues go from being flooded to being consumed, and vice versa.

2 files changed, 13 insertions, 18 deletions

diff --git a/src/rabbit_memory_monitor.erl b/src/rabbit_memory_monitor.erl
index ec08d475f0..1a879b7c5a 100644
--- a/src/rabbit_memory_monitor.erl
+++ b/src/rabbit_memory_monitor.erl
@@ -230,14 +230,14 @@ internal_update(State = #state{memory_limit = Limit,
                                callbacks = Callbacks}) ->
     %% available memory / used memory
     MemoryRatio = Limit / erlang:memory(total),
-    AvgDuration = case Count of
-                      0 -> infinity;
-                      _ -> Sum / Count
+    AvgDuration = case Count == 0 of
+                      true  -> infinity;
+                      false -> Sum / Count
                   end,
     DesiredDurationAvg1 =
-        case AvgDuration of
-            infinity -> infinity;
-            AvgQueueDuration -> lists:max([0, AvgQueueDuration * MemoryRatio])
+        case AvgDuration == infinity orelse MemoryRatio > 2 of
+            true  -> infinity;
+            false -> lists:max([0, AvgDuration * MemoryRatio])
         end,
     State1 = State#state{memory_ratio = MemoryRatio,
                          desired_duration = DesiredDurationAvg1},
@@ -245,7 +245,8 @@ internal_update(State = #state{memory_limit = Limit,
     %% only inform queues immediately if the desired duration has
     %% decreased
     case (DesiredDurationAvg == infinity andalso DesiredDurationAvg /= infinity)
-        orelse (DesiredDurationAvg1 < DesiredDurationAvg) of
+        orelse (DesiredDurationAvg1 /= infinity andalso
+                DesiredDurationAvg1 < DesiredDurationAvg) of
         true ->
             %% If we have pessimistic information, we need to inform
             %% queues to reduce it's memory usage when needed. This
@@ -261,7 +262,7 @@ internal_update(State = #state{memory_limit = Limit,
                                      ets:insert(Durations,
                                                 {Pid, QueueDuration,
                                                  DesiredDurationAvg1});
-                                 _ -> true
+                                 false -> true
                              end
                      end, true, Durations);
         false -> ok
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 2ee57ba7df..461d311014 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -224,10 +224,7 @@ set_queue_ram_duration_target(
                                      avg_ingress_rate = AvgIngressRate,
                                      target_ram_msg_count = TargetRamMsgCount
                                    }) ->
-    Rate = case 0 == AvgEgressRate of
-               true -> AvgIngressRate;
-               false -> AvgEgressRate
-           end,
+    Rate = AvgEgressRate + AvgIngressRate,
     TargetRamMsgCount1 =
         case DurationTarget of
             infinity -> undefined;
@@ -265,12 +262,9 @@ ram_duration(#vqstate { avg_egress_rate = AvgEgressRate,
                         avg_ingress_rate = AvgIngressRate,
                         ram_msg_count = RamMsgCount }) ->
     %% msgs / (msgs/sec) == sec
-    case AvgEgressRate == 0 of
-        true  -> case AvgIngressRate == 0 of
-                     true -> infinity;
-                     false -> RamMsgCount / AvgIngressRate
-                 end;
-        false -> RamMsgCount / AvgEgressRate
+    case AvgEgressRate == 0 andalso AvgIngressRate == 0 of
+        true  -> infinity;
+        false -> RamMsgCount / (AvgEgressRate + AvgIngressRate)
     end.
 
 fetch(State =