Merge branch 'stable' into rabbitmq-server-1379

Author: michaelklishin

src/rabbit_amqqueue_process.erl

@@ -22,7 +22,6 @@
 
 -define(SYNC_INTERVAL,                 200). %% milliseconds
 -define(RAM_DURATION_UPDATE_INTERVAL, 5000).
--define(CONSUMER_BIAS_RATIO,           1.1). %% i.e. consume 10% faster
 
 -export([info_keys/0]).
 
@@ -969,26 +968,26 @@ emit_consumer_deleted(ChPid, ConsumerTag, QName) ->
 
 %%----------------------------------------------------------------------------
 
-prioritise_call(Msg, _From, _Len, State) ->
+prioritise_call(Msg, _From, _Len, _State) ->
     case Msg of
         info                                       -> 9;
         {info, _Items}                             -> 9;
         consumers                                  -> 9;
         stat                                       -> 7;
-        {basic_consume, _, _, _, _, _, _, _, _, _} -> consumer_bias(State);
-        {basic_cancel, _, _, _}                    -> consumer_bias(State);
+        {basic_consume, _, _, _, _, _, _, _, _, _} -> 1;
+        {basic_cancel, _, _, _}                    -> 1;
         _                                          -> 0
     end.
 
-prioritise_cast(Msg, _Len, State) ->
+prioritise_cast(Msg, _Len, _State) ->
     case Msg of
         delete_immediately                   -> 8;
         {set_ram_duration_target, _Duration} -> 8;
         {set_maximum_since_use, _Age}        -> 8;
         {run_backing_queue, _Mod, _Fun}      -> 6;
         {ack, _AckTags, _ChPid}              -> 3; %% [1]
         {resume, _ChPid}                     -> 2;
-        {notify_sent, _ChPid, _Credit}       -> consumer_bias(State);
+        {notify_sent, _ChPid, _Credit}       -> 1;
         _                                    -> 0
     end.
 
@@ -1001,13 +1000,6 @@ prioritise_cast(Msg, _Len, State) ->
 %% about. Finally, we prioritise ack over resume since it should
 %% always reduce memory use.
 
-consumer_bias(#q{backing_queue = BQ, backing_queue_state = BQS}) ->
-    case BQ:msg_rates(BQS) of
-        {0.0,          _} -> 0;
-        {Ingress, Egress} when Egress / Ingress < ?CONSUMER_BIAS_RATIO -> 1;
-        {_,            _} -> 0
-    end.
-
 prioritise_info(Msg, _Len, #q{q = #amqqueue{exclusive_owner = DownPid}}) ->
     case Msg of
         {'DOWN', _, process, DownPid, _}     -> 8;

src/rabbit_cli.erl

@@ -122,15 +122,12 @@ main(ParseFun, DoFun, UsageMod) ->
                 _ ->
                     print_error("unable to connect to node ~w: ~w", [Node, Reason]),
                     print_badrpc_diagnostics([Node]),
-                    case Command of
-                        stop -> rabbit_misc:quit(?EX_OK);
-                        _    -> rabbit_misc:quit(?EX_UNAVAILABLE)
-                    end
+                    exit_badrpc(Command)
             end;
         {badrpc_multi, Reason, Nodes} ->
             print_error("unable to connect to nodes ~p: ~w", [Nodes, Reason]),
             print_badrpc_diagnostics(Nodes),
-            rabbit_misc:quit(?EX_UNAVAILABLE);
+            exit_badrpc(Command);
         function_clause ->
             print_error("operation ~w used with invalid parameter: ~p",
                         [Command, Args]),
@@ -143,6 +140,11 @@ main(ParseFun, DoFun, UsageMod) ->
             rabbit_misc:quit(?EX_SOFTWARE)
     end.
 
+exit_badrpc(stop) ->
+    rabbit_misc:quit(?EX_OK);
+exit_badrpc(_) ->
+    rabbit_misc:quit(?EX_UNAVAILABLE).
+
 start_distribution_anon(0, LastError) ->
     {error, LastError};
 start_distribution_anon(TriesLeft, _) ->