merge bug25817 into default

41 files changed, 1737 insertions, 1307 deletions

diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index d2a3f7c7..d19acd00 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -1559,14 +1559,6 @@
                 <term>prefetch_count</term>
                 <listitem><para>QoS prefetch count limit in force, 0 if unlimited.</para></listitem>
               </varlistentry>
-              <varlistentry>
-                <term>client_flow_blocked</term>
-                <listitem><para>True if the client issued a
-                    <command>channel.flow{active=false}</command>
-                    command, blocking the server from delivering
-                    messages to the channel's consumers.
-                    </para></listitem>
-              </varlistentry>
             </variablelist>
             <para>
               If no <command>channelinfoitem</command>s are specified then pid,
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index bd7a0eed..6d117e3d 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -70,7 +70,7 @@
                         is_persistent}).
 
 -record(ssl_socket, {tcp, ssl}).
--record(delivery, {mandatory, sender, message, msg_seq_no}).
+-record(delivery, {mandatory, confirm, sender, message, msg_seq_no}).
 -record(amqp_error, {name, explanation = "", method = none}).
 
 -record(event, {type, props, timestamp}).
@@ -111,3 +111,12 @@
 -define(INVALID_HEADERS_KEY, <<"x-invalid-headers">>).
 -define(ROUTING_HEADERS, [<<"CC">>, <<"BCC">>]).
 -define(DELETED_HEADER, <<"BCC">>).
+
+%% Trying to send a term across a cluster larger than 2^31 bytes will
+%% cause the VM to exit with "Absurdly large distribution output data
+%% buffer". So we limit the max message size to 2^31 - 10^6 bytes (1MB
+%% to allow plenty of leeway for the #basic_message{} and #content{}
+%% wrapping the message body).
+-define(MAX_MSG_SIZE, 2147383648).
+
+-define(store_proc_name(N), rabbit_misc:store_proc_name(?MODULE, N)).
diff --git a/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec
index f53eea95..a96ccb35 100644
--- a/packaging/RPMS/Fedora/rabbitmq-server.spec
+++ b/packaging/RPMS/Fedora/rabbitmq-server.spec
@@ -130,6 +130,9 @@ done
 rm -rf %{buildroot}
 
 %changelog
+* Thu Jan 23 2014 emile@rabbitmq.com 3.2.3-1
+- New Upstream Release
+
 * Tue Dec 10 2013 emile@rabbitmq.com 3.2.2-1
 - New Upstream Release
 
diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog
index f1e1c66b..7138409c 100644
--- a/packaging/debs/Debian/debian/changelog
+++ b/packaging/debs/Debian/debian/changelog
@@ -1,3 +1,9 @@
+rabbitmq-server (3.2.3-1) unstable; urgency=low
+
+  * New Upstream Release
+
+ -- Emile Joubert <emile@rabbitmq.com>  Thu, 23 Jan 2014 14:46:37 +0000
+
 rabbitmq-server (3.2.2-1) unstable; urgency=low
 
   * New Upstream Release
diff --git a/src/dtree.erl b/src/dtree.erl
index 5ff36bd9..72abe248 100644
--- a/src/dtree.erl
+++ b/src/dtree.erl
@@ -32,7 +32,7 @@
 
 -module(dtree).
 
--export([empty/0, insert/4, take/3, take/2, take_all/2,
+-export([empty/0, insert/4, take/3, take/2, take_all/2, drop/2,
          is_defined/2, is_empty/1, smallest/1, size/1]).
 
 %%----------------------------------------------------------------------------
@@ -53,6 +53,7 @@
 -spec(take/3       :: ([pk()], sk(), ?MODULE()) -> {[kv()], ?MODULE()}).
 -spec(take/2       :: (sk(), ?MODULE()) -> {[kv()], ?MODULE()}).
 -spec(take_all/2   :: (sk(), ?MODULE()) -> {[kv()], ?MODULE()}).
+-spec(drop/2       :: (pk(), ?MODULE()) -> ?MODULE()).
 -spec(is_defined/2 :: (sk(), ?MODULE()) -> boolean()).
 -spec(is_empty/1   :: (?MODULE()) -> boolean()).
 -spec(smallest/1   :: (?MODULE()) -> kv()).
@@ -120,6 +121,14 @@ take_all(SK, {P, S}) ->
                         {KVs, {P1, prune(SKS, PKS, S)}}
     end.
 
+%% Drop all entries for the given primary key (which does not have to exist).
+drop(PK, {P, S}) ->
+    case gb_trees:lookup(PK, P) of
+        none               -> {P, S};
+        {value, {SKS, _V}} -> {gb_trees:delete(PK, P),
+                               prune(SKS, gb_sets:singleton(PK), S)}
+    end.
+
 is_defined(SK, {_P, S}) -> gb_trees:is_defined(SK, S).
 
 is_empty({P, _S}) -> gb_trees:is_empty(P).
diff --git a/src/gm.erl b/src/gm.erl
index 098d84fa..5a82950a 100644
--- a/src/gm.erl
+++ b/src/gm.erl
@@ -382,7 +382,7 @@
 
 -behaviour(gen_server2).
 
--export([create_tables/0, start_link/4, leave/1, broadcast/2,
+-export([create_tables/0, start_link/4, leave/1, broadcast/2, broadcast/3,
          confirmed_broadcast/2, info/1, validate_members/2, forget_group/1]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@@ -395,6 +395,7 @@
 -export([table_definitions/0]).
 
 -define(GROUP_TABLE, gm_group).
+-define(MAX_BUFFER_SIZE, 100000000). %% 100MB
 -define(HIBERNATE_AFTER_MIN, 1000).
 -define(DESIRED_HIBERNATE, 10000).
 -define(BROADCAST_TIMER, 25).
@@ -414,6 +415,7 @@
           callback_args,
           confirms,
           broadcast_buffer,
+          broadcast_buffer_sz,
           broadcast_timer,
           txn_executor
         }).
@@ -522,8 +524,10 @@ start_link(GroupName, Module, Args, TxnFun) ->
 leave(Server) ->
     gen_server2:cast(Server, leave).
 
-broadcast(Server, Msg) ->
-    gen_server2:cast(Server, {broadcast, Msg}).
+broadcast(Server, Msg) -> broadcast(Server, Msg, 0).
+
+broadcast(Server, Msg, SizeHint) ->
+    gen_server2:cast(Server, {broadcast, Msg, SizeHint}).
 
 confirmed_broadcast(Server, Msg) ->
     gen_server2:call(Server, {confirmed_broadcast, Msg}, infinity).
@@ -542,23 +546,25 @@ forget_group(GroupName) ->
     ok.
 
 init([GroupName, Module, Args, TxnFun]) ->
+    put(process_name, {?MODULE, GroupName}),
     {MegaSecs, Secs, MicroSecs} = now(),
     random:seed(MegaSecs, Secs, MicroSecs),
     Self = make_member(GroupName),
     gen_server2:cast(self(), join),
-    {ok, #state { self             = Self,
-                  left             = {Self, undefined},
-                  right            = {Self, undefined},
-                  group_name       = GroupName,
-                  module           = Module,
-                  view             = undefined,
-                  pub_count        = -1,
-                  members_state    = undefined,
-                  callback_args    = Args,
-                  confirms         = queue:new(),
-                  broadcast_buffer = [],
-                  broadcast_timer  = undefined,
-                  txn_executor     = TxnFun }, hibernate,
+    {ok, #state { self                = Self,
+                  left                = {Self, undefined},
+                  right               = {Self, undefined},
+                  group_name          = GroupName,
+                  module              = Module,
+                  view                = undefined,
+                  pub_count           = -1,
+                  members_state       = undefined,
+                  callback_args       = Args,
+                  confirms            = queue:new(),
+                  broadcast_buffer    = [],
+                  broadcast_buffer_sz = 0,
+                  broadcast_timer     = undefined,
+                  txn_executor        = TxnFun }, hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
 
@@ -575,7 +581,7 @@ handle_call({confirmed_broadcast, Msg}, _From,
                            ok, State});
 
 handle_call({confirmed_broadcast, Msg}, From, State) ->
-    internal_broadcast(Msg, From, State);
+    internal_broadcast(Msg, From, 0, State);
 
 handle_call(info, _From,
             State = #state { members_state = undefined }) ->
@@ -638,10 +644,11 @@ handle_cast({?TAG, ReqVer, Msg},
       if_callback_success(
         Result, fun handle_msg_true/3, fun handle_msg_false/3, Msg, State1));
 
-handle_cast({broadcast, _Msg}, State = #state { members_state = undefined }) ->
+handle_cast({broadcast, _Msg, _SizeHint},
+            State = #state { members_state = undefined }) ->
     noreply(State);
 
-handle_cast({broadcast, Msg},
+handle_cast({broadcast, Msg, _SizeHint},
             State = #state { self          = Self,
                              right         = {Self, undefined},
                              module        = Module,
@@ -649,8 +656,8 @@ handle_cast({broadcast, Msg},
     handle_callback_result({Module:handle_msg(Args, get_pid(Self), Msg),
                             State});
 
-handle_cast({broadcast, Msg}, State) ->
-    internal_broadcast(Msg, none, State);
+handle_cast({broadcast, Msg, SizeHint}, State) ->
+    internal_broadcast(Msg, none, SizeHint, State);
 
 handle_cast(join, State = #state { self          = Self,
                                    group_name    = GroupName,
@@ -882,12 +889,14 @@ ensure_broadcast_timer(State = #state { broadcast_timer = undefined }) ->
 ensure_broadcast_timer(State) ->
     State.
 
-internal_broadcast(Msg, From, State = #state { self             = Self,
-                                               pub_count        = PubCount,
-                                               module           = Module,
-                                               confirms         = Confirms,
-                                               callback_args    = Args,
-                                               broadcast_buffer = Buffer }) ->
+internal_broadcast(Msg, From, SizeHint,
+                   State = #state { self                = Self,
+                                    pub_count           = PubCount,
+                                    module              = Module,
+                                    confirms            = Confirms,
+                                    callback_args       = Args,
+                                    broadcast_buffer    = Buffer,
+                                    broadcast_buffer_sz = BufferSize }) ->
     PubCount1 = PubCount + 1,
     Result = Module:handle_msg(Args, get_pid(Self), Msg),
     Buffer1 = [{PubCount1, Msg} | Buffer],
@@ -895,15 +904,37 @@ internal_broadcast(Msg, From, State = #state { self             = Self,
                     none -> Confirms;
                     _    -> queue:in({PubCount1, From}, Confirms)
                 end,
-    State1 = State #state { pub_count        = PubCount1,
-                            confirms         = Confirms1,
-                            broadcast_buffer = Buffer1 },
-    case From =/= none of
-        true ->
-            handle_callback_result({Result, flush_broadcast_buffer(State1)});
-        false ->
-            handle_callback_result(
-              {Result, State1 #state { broadcast_buffer = Buffer1 }})
+    State1 = State #state { pub_count           = PubCount1,
+                            confirms            = Confirms1,
+                            broadcast_buffer    = Buffer1,
+                            broadcast_buffer_sz = BufferSize + SizeHint},
+    handle_callback_result(
+        {Result, case From of
+                     none -> maybe_flush_broadcast_buffer(State1);
+                     _    -> flush_broadcast_buffer(State1)
+                 end}).
+
+%% The Erlang distribution mechanism has an interesting quirk - it
+%% will kill the VM cold with "Absurdly large distribution output data
+%% buffer" if you attempt to send a message which serialises out to
+%% more than 2^31 bytes in size. It's therefore a very good idea to
+%% make sure that we don't exceed that size!
+%%
+%% Now, we could figure out the size of messages as they come in using
+%% size(term_to_binary(Msg)) or similar. The trouble is, that requires
+%% us to serialise the message only to throw the serialised form
+%% away. Hard to believe that's a sensible thing to do. So instead we
+%% accept a size hint from the application, via broadcast/3. This size
+%% hint can be the size of anything in the message which we expect
+%% could be large, and we just ignore the size of any small bits of
+%% the message term. Therefore MAX_BUFFER_SIZE is set somewhat
+%% conservatively at 100MB - but the buffer is only to allow us to
+%% buffer tiny messages anyway, so 100MB is plenty.
+
+maybe_flush_broadcast_buffer(State = #state{broadcast_buffer_sz = Size}) ->
+    case Size > ?MAX_BUFFER_SIZE of
+        true  -> flush_broadcast_buffer(State);
+        false -> State
     end.
 
 flush_broadcast_buffer(State = #state { broadcast_buffer = [] }) ->
@@ -922,8 +953,9 @@ flush_broadcast_buffer(State = #state { self             = Self,
                               Member #member { pending_ack = PA1,
                                                last_pub = PubCount }
                       end, Self, MembersState),
-    State #state { members_state    = MembersState1,
-                   broadcast_buffer = [] }.
+    State #state { members_state       = MembersState1,
+                   broadcast_buffer    = [],
+                   broadcast_buffer_sz = 0}.
 
 
 %% ---------------------------------------------------------------------------
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index d54c2a8d..19171659 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -52,18 +52,31 @@ check_user_pass_login(Username, Password) ->
 check_user_login(Username, AuthProps) ->
     {ok, Modules} = application:get_env(rabbit, auth_backends),
     lists:foldl(
-      fun(Module, {refused, _, _}) ->
-              case Module:check_user_login(Username, AuthProps) of
-                  {error, E} ->
-                      {refused, "~s failed authenticating ~s: ~p~n",
-                       [Module, Username, E]};
-                  Else ->
-                      Else
+      fun ({ModN, ModZ}, {refused, _, _}) ->
+              %% Different modules for authN vs authZ. So authenticate
+              %% with authN module, then if that succeeds do
+              %% passwordless (i.e pre-authenticated) login with authZ
+              %% module, and use the #user{} the latter gives us.
+              case try_login(ModN, Username, AuthProps) of
+                  {ok, _} -> try_login(ModZ, Username, []);
+                  Else    -> Else
               end;
-         (_, {ok, User}) ->
+          (Mod, {refused, _, _}) ->
+              %% Same module for authN and authZ. Just take the result
+              %% it gives us
+              try_login(Mod, Username, AuthProps);
+          (_, {ok, User}) ->
+              %% We've successfully authenticated. Skip to the end...
               {ok, User}
       end, {refused, "No modules checked '~s'", [Username]}, Modules).
 
+try_login(Module, Username, AuthProps) ->
+    case Module:check_user_login(Username, AuthProps) of
+        {error, E} -> {refused, "~s failed authenticating ~s: ~p~n",
+                       [Module, Username, E]};
+        Else       -> Else
+    end.
+
 check_vhost_access(User = #user{ username     = Username,
                                  auth_backend = Module }, VHostPath) ->
     check_access(
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 6b1e00b7..eeb0e0bf 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -26,8 +26,8 @@
 -export([list/0, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2]).
 -export([force_event_refresh/0, notify_policy_changed/1]).
 -export([consumers/1, consumers_all/1, consumer_info_keys/0]).
--export([basic_get/4, basic_consume/10, basic_cancel/4, notify_decorators/1]).
--export([notify_sent/2, notify_sent_queue_down/1, resume/2, flush_all/2]).
+-export([basic_get/4, basic_consume/9, basic_cancel/4, notify_decorators/1]).
+-export([notify_sent/2, notify_sent_queue_down/1, resume/2]).
 -export([notify_down_all/2, activate_limit_all/2, credit/5]).
 -export([on_node_down/1]).
 -export([update/2, store_queue/1, policy_changed/2]).
@@ -51,7 +51,7 @@
 
 -ifdef(use_specs).
 
--export_type([name/0, qmsg/0, routing_result/0]).
+-export_type([name/0, qmsg/0]).
 
 -type(name() :: rabbit_types:r('queue')).
 -type(qpids() :: [pid()]).
@@ -61,7 +61,6 @@
 -type(msg_id() :: non_neg_integer()).
 -type(ok_or_errors() ::
         'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}).
--type(routing_result() :: 'routed' | 'unroutable').
 -type(queue_or_absent() :: rabbit_types:amqqueue() |
                            {'absent', rabbit_types:amqqueue()}).
 -type(not_found_or_absent() :: 'not_found' |
@@ -113,10 +112,9 @@
                     -> [rabbit_types:infos()]).
 -spec(force_event_refresh/0 :: () -> 'ok').
 -spec(notify_policy_changed/1 :: (rabbit_types:amqqueue()) -> 'ok').
--spec(consumers/1 ::
-        (rabbit_types:amqqueue())
-        -> [{pid(), rabbit_types:ctag(), boolean(),
-             rabbit_framing:amqp_table()}]).
+-spec(consumers/1 :: (rabbit_types:amqqueue())
+                     -> [{pid(), rabbit_types:ctag(), boolean(),
+                          rabbit_framing:amqp_table()}]).
 -spec(consumer_info_keys/0 :: () -> rabbit_types:info_keys()).
 -spec(consumers_all/1 ::
         (rabbit_types:vhost())
@@ -139,9 +137,9 @@
 -spec(purge/1 :: (rabbit_types:amqqueue()) -> qlen()).
 -spec(forget_all_durable/1 :: (node()) -> 'ok').
 -spec(deliver/2 :: ([rabbit_types:amqqueue()], rabbit_types:delivery()) ->
-                        {routing_result(), qpids()}).
+                        qpids()).
 -spec(deliver_flow/2 :: ([rabbit_types:amqqueue()], rabbit_types:delivery()) ->
-                             {routing_result(), qpids()}).
+                             qpids()).
 -spec(requeue/3 :: (pid(), [msg_id()],  pid()) -> 'ok').
 -spec(ack/3 :: (pid(), [msg_id()], pid()) -> 'ok').
 -spec(reject/4 :: (pid(), [msg_id()], boolean(), pid()) -> 'ok').
@@ -151,9 +149,9 @@
                           {'ok', non_neg_integer(), qmsg()} | 'empty').
 -spec(credit/5 :: (rabbit_types:amqqueue(), pid(), rabbit_types:ctag(),
                    non_neg_integer(), boolean()) -> 'ok').
--spec(basic_consume/10 ::
+-spec(basic_consume/9 ::
         (rabbit_types:amqqueue(), boolean(), pid(), pid(), boolean(),
-         rabbit_types:ctag(), boolean(), {non_neg_integer(), boolean()} | 'none', any(), any())
+         rabbit_types:ctag(), boolean(), rabbit_framing:amqp_table(), any())
         -> rabbit_types:ok_or_error('exclusive_consume_unavailable')).
 -spec(basic_cancel/4 ::
         (rabbit_types:amqqueue(), pid(), rabbit_types:ctag(), any()) -> 'ok').
@@ -161,7 +159,6 @@
 -spec(notify_sent/2 :: (pid(), pid()) -> 'ok').
 -spec(notify_sent_queue_down/1 :: (pid()) -> 'ok').
 -spec(resume/2 :: (pid(), pid()) -> 'ok').
--spec(flush_all/2 :: (qpids(), pid()) -> 'ok').
 -spec(internal_delete/1 ::
         (name()) -> rabbit_types:ok_or_error('not_found') |
                     rabbit_types:connection_exit() |
@@ -195,13 +192,18 @@ recover() ->
     on_node_down(node()),
     DurableQueues = find_durable_queues(),
     {ok, BQ} = application:get_env(rabbit, backing_queue_module),
-    ok = BQ:start([QName || #amqqueue{name = QName} <- DurableQueues]),
+
+    %% We rely on BQ:start/1 returning the recovery terms in the same
+    %% order as the supplied queue names, so that we can zip them together
+    %% for further processing in recover_durable_queues.
+    {ok, OrderedRecoveryTerms} =
+        BQ:start([QName || #amqqueue{name = QName} <- DurableQueues]),
     {ok,_} = supervisor:start_child(
                rabbit_sup,
                {rabbit_amqqueue_sup,
                 {rabbit_amqqueue_sup, start_link, []},
                 transient, infinity, supervisor, [rabbit_amqqueue_sup]}),
-    recover_durable_queues(DurableQueues).
+    recover_durable_queues(lists:zip(DurableQueues, OrderedRecoveryTerms)).
 
 stop() ->
     ok = supervisor:terminate_child(rabbit_sup, rabbit_amqqueue_sup),
@@ -229,10 +231,11 @@ find_durable_queues() ->
                                 node(Pid) == Node]))
       end).
 
-recover_durable_queues(DurableQueues) ->
-    Qs = [start_queue_process(node(), Q) || Q <- DurableQueues],
-    [Q || Q = #amqqueue{pid = Pid} <- Qs,
-          gen_server2:call(Pid, {init, self()}, infinity) == {new, Q}].
+recover_durable_queues(QueuesAndRecoveryTerms) ->
+    Qs = [{start_queue_process(node(), Q), Terms} ||
+             {Q, Terms} <- QueuesAndRecoveryTerms],
+    [Q || {Q = #amqqueue{ pid = Pid }, Terms} <- Qs,
+          gen_server2:call(Pid, {init, {self(), Terms}}, infinity) == {new, Q}].
 
 declare(QueueName, Durable, AutoDelete, Args, Owner) ->
     ok = check_declare_arguments(QueueName, Args),
@@ -427,7 +430,7 @@ declare_args() ->
     [{<<"x-expires">>,                 fun check_expires_arg/2},
      {<<"x-message-ttl">>,             fun check_message_ttl_arg/2},
      {<<"x-dead-letter-routing-key">>, fun check_dlxrk_arg/2},
-     {<<"x-max-length">>,              fun check_max_length_arg/2}].
+     {<<"x-max-length">>,              fun check_non_neg_int_arg/2}].
 
 consume_args() -> [{<<"x-priority">>, fun check_int_arg/2}].
 
@@ -437,7 +440,7 @@ check_int_arg({Type, _}, _) ->
         false -> {error, {unacceptable_type, Type}}
     end.
 
-check_max_length_arg({Type, Val}, Args) ->
+check_non_neg_int_arg({Type, Val}, Args) ->
     case check_int_arg({Type, Val}, Args) of
         ok when Val >= 0 -> ok;
         ok               -> {error, {value_negative, Val}};
@@ -550,8 +553,8 @@ requeue(QPid, MsgIds, ChPid) -> delegate:call(QPid, {requeue, MsgIds, ChPid}).
 
 ack(QPid, MsgIds, ChPid) -> delegate:cast(QPid, {ack, MsgIds, ChPid}).
 
-reject(QPid, MsgIds, Requeue, ChPid) ->
-    delegate:cast(QPid, {reject, MsgIds, Requeue, ChPid}).
+reject(QPid, Requeue, MsgIds, ChPid) ->
+    delegate:cast(QPid, {reject, Requeue, MsgIds, ChPid}).
 
 notify_down_all(QPids, ChPid) ->
     {_, Bads} = delegate:call(QPids, {notify_down, ChPid}),
@@ -572,13 +575,11 @@ credit(#amqqueue{pid = QPid}, ChPid, CTag, Credit, Drain) ->
 basic_get(#amqqueue{pid = QPid}, ChPid, NoAck, LimiterPid) ->
     delegate:call(QPid, {basic_get, ChPid, NoAck, LimiterPid}).
 
-basic_consume(#amqqueue{pid = QPid, name = QName}, NoAck, ChPid,
-              LimiterPid, LimiterActive,
-              ConsumerTag, ExclusiveConsume, CreditArgs, OtherArgs, OkMsg) ->
-    ok = check_consume_arguments(QName, OtherArgs),
+basic_consume(#amqqueue{pid = QPid, name = QName}, NoAck, ChPid, LimiterPid,
+              LimiterActive, ConsumerTag, ExclusiveConsume, Args, OkMsg) ->
+    ok = check_consume_arguments(QName, Args),
     delegate:call(QPid, {basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
-                         ConsumerTag, ExclusiveConsume, CreditArgs, OtherArgs,
-                         OkMsg}).
+                         ConsumerTag, ExclusiveConsume, Args, OkMsg}).
 
 basic_cancel(#amqqueue{pid = QPid}, ChPid, ConsumerTag, OkMsg) ->
     delegate:call(QPid, {basic_cancel, ChPid, ConsumerTag, OkMsg}).
@@ -605,8 +606,6 @@ notify_sent_queue_down(QPid) ->
 
 resume(QPid, ChPid) -> delegate:cast(QPid, {resume, ChPid}).
 
-flush_all(QPids, ChPid) -> delegate:cast(QPids, {flush, ChPid}).
-
 internal_delete1(QueueName) ->
     ok = mnesia:delete({rabbit_queue, QueueName}),
     %% this 'guarded' delete prevents unnecessary writes to the mnesia
@@ -704,17 +703,11 @@ pseudo_queue(QueueName, Pid) ->
               pid          = Pid,
               slave_pids   = []}.
 
-deliver([], #delivery{mandatory = false}, _Flow) ->
+deliver([], _Delivery, _Flow) ->
     %% /dev/null optimisation
-    {routed, []};
-
-deliver(Qs, Delivery = #delivery{mandatory = false}, Flow) ->
-    %% optimisation: when Mandatory = false, rabbit_amqqueue:deliver
-    %% will deliver the message to the queue process asynchronously,
-    %% and return true, which means all the QPids will always be
-    %% returned. It is therefore safe to use a fire-and-forget cast
-    %% here and return the QPids - the semantics is preserved. This
-    %% scales much better than the case below.
+    [];
+
+deliver(Qs, Delivery, Flow) ->
     {MPids, SPids} = qpids(Qs),
     QPids = MPids ++ SPids,
     case Flow of
@@ -731,19 +724,7 @@ deliver(Qs, Delivery = #delivery{mandatory = false}, Flow) ->
     SMsg = {deliver, Delivery, true,  Flow},
     delegate:cast(MPids, MMsg),
     delegate:cast(SPids, SMsg),
-    {routed, QPids};
-
-deliver(Qs, Delivery, _Flow) ->
-    {MPids, SPids} = qpids(Qs),
-    %% see comment above
-    MMsg = {deliver, Delivery, false},
-    SMsg = {deliver, Delivery, true},
-    {MRouted, _} = delegate:call(MPids, MMsg),
-    {SRouted, _} = delegate:call(SPids, SMsg),
-    case MRouted ++ SRouted of
-        [] -> {unroutable, []};
-        R  -> {routed,     [QPid || {QPid, ok} <- R]}
-    end.
+    QPids.
 
 qpids([]) -> {[], []}; %% optimisation
 qpids([#amqqueue{pid = QPid, slave_pids = SPids}]) -> {[QPid], SPids}; %% opt
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index cb59edd9..9bc5a775 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -20,9 +20,9 @@
 
 -behaviour(gen_server2).
 
--define(UNSENT_MESSAGE_LIMIT,          200).
 -define(SYNC_INTERVAL,                 25). %% milliseconds
 -define(RAM_DURATION_UPDATE_INTERVAL,  5000).
+-define(CONSUMER_BIAS_RATIO,           1.1). %% i.e. consume 10% faster
 
 -export([start_link/1, info_keys/0]).
 
@@ -38,8 +38,7 @@
             has_had_consumers,
             backing_queue,
             backing_queue_state,
-            active_consumers,
-            consumer_use,
+            consumers,
             expires,
             sync_timer_ref,
             rate_timer_ref,
@@ -57,21 +56,6 @@
             status
            }).
 
--record(consumer, {tag, ack_required, args}).
-
-%% These are held in our process dictionary
--record(cr, {ch_pid,
-             monitor_ref,
-             acktags,
-             consumer_count,
-             %% Queue of {ChPid, #consumer{}} for consumers which have
-             %% been blocked for any reason
-             blocked_consumers,
-             %% The limiter itself
-             limiter,
-             %% Internal flow control for queue -> writer
-             unsent_message_count}).
-
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
@@ -124,6 +108,7 @@ info_keys() -> ?INFO_KEYS.
 
 init(Q) ->
     process_flag(trap_exit, true),
+    ?store_proc_name(Q#amqqueue.name),
     {ok, init_state(Q#amqqueue{pid = self()}), hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
@@ -143,15 +128,14 @@ init_with_backing_queue_state(Q = #amqqueue{exclusive_owner = Owner}, BQ, BQS,
     State3 = lists:foldl(fun (Delivery, StateN) ->
                                  deliver_or_enqueue(Delivery, true, StateN)
                          end, State2, Deliveries),
-    notify_decorators(startup, [], State3),
+    notify_decorators(startup, State3),
     State3.
 
 init_state(Q) ->
     State = #q{q                   = Q,
                exclusive_consumer  = none,
                has_had_consumers   = false,
-               active_consumers    = priority_queue:new(),
-               consumer_use        = {inactive, now_micros(), 0, 0.0},
+               consumers           = rabbit_queue_consumers:new(),
                senders             = pmon:new(delegate),
                msg_id_to_channel   = gb_trees:empty(),
                status              = running,
@@ -190,9 +174,10 @@ code_change(_OldVsn, State, _Extra) ->
 declare(Recover, From, State = #q{q                   = Q,
                                   backing_queue       = undefined,
                                   backing_queue_state = undefined}) ->
-    case rabbit_amqqueue:internal_declare(Q, Recover =/= new) of
+    {Recovery, TermsOrNew} = recovery_status(Recover),
+    case rabbit_amqqueue:internal_declare(Q, Recovery /= new) of
         #amqqueue{} = Q1 ->
-            case matches(Recover, Q, Q1) of
+            case matches(Recovery, Q, Q1) of
                 true ->
                     gen_server2:reply(From, {new, Q}),
                     ok = file_handle_cache:register_callback(
@@ -201,12 +186,12 @@ declare(Recover, From, State = #q{q                   = Q,
                            self(), {rabbit_amqqueue,
                                     set_ram_duration_target, [self()]}),
                     BQ = backing_queue_module(Q1),
-                    BQS = bq_init(BQ, Q, Recover),
-                    recovery_barrier(Recover),
+                    BQS = bq_init(BQ, Q, TermsOrNew),
+                    recovery_barrier(Recovery),
                     State1 = process_args_policy(
                                State#q{backing_queue       = BQ,
                                        backing_queue_state = BQS}),
-                    notify_decorators(startup, [], State),
+                    notify_decorators(startup, State),
                     rabbit_event:notify(queue_created,
                                         infos(?CREATION_EVENT_KEYS, State1)),
                     rabbit_event:if_enabled(State1, #q.stats_timer,
@@ -219,6 +204,9 @@ declare(Recover, From, State = #q{q                   = Q,
             {stop, normal, Err, State}
     end.
 
+recovery_status(new)              -> {new,     new};
+recovery_status({Recover, Terms}) -> {Recover, Terms}.
+
 matches(new, Q1, Q2) ->
     %% i.e. not policy
     Q1#amqqueue.name            =:= Q2#amqqueue.name            andalso
@@ -231,18 +219,17 @@ matches(new, Q1, Q2) ->
 matches(_,  Q,   Q) -> true;
 matches(_, _Q, _Q1) -> false.
 
-notify_decorators(Event, Props, State) when Event =:= startup;
-                                            Event =:= shutdown ->
-    decorator_callback(qname(State), Event, Props);
+maybe_notify_decorators(false, State) -> State;
+maybe_notify_decorators(true,  State) -> notify_decorators(State), State.
+
+notify_decorators(Event, State) -> decorator_callback(qname(State), Event, []).
 
-notify_decorators(Event, Props, State = #q{active_consumers    = ACs,
-                                           backing_queue       = BQ,
-                                           backing_queue_state = BQS}) ->
-    P = priority_queue:highest(ACs),
-    decorator_callback(qname(State), notify,
-                       [Event, [{max_active_consumer_priority, P},
-                                {is_empty, BQ:is_empty(BQS)} |
-                                Props]]).
+notify_decorators(State = #q{consumers           = Consumers,
+                             backing_queue       = BQ,
+                             backing_queue_state = BQS}) ->
+    P = rabbit_queue_consumers:max_active_priority(Consumers),
+    decorator_callback(qname(State), consumer_state_changed,
+                       [P, BQ:is_empty(BQS)]).
 
 decorator_callback(QName, F, A) ->
     %% Look up again in case policy and hence decorators have changed
@@ -255,7 +242,7 @@ decorator_callback(QName, F, A) ->
 
 bq_init(BQ, Q, Recover) ->
     Self = self(),
-    BQ:init(Q, Recover =/= new,
+    BQ:init(Q, Recover,
             fun (Mod, Fun) ->
                     rabbit_amqqueue:run_backing_queue(Self, Mod, Fun)
             end).
@@ -316,7 +303,7 @@ init_max_length(MaxLen, State) ->
     State1.
 
 terminate_shutdown(Fun, State) ->
-    State1 = #q{backing_queue_state = BQS} =
+    State1 = #q{backing_queue_state = BQS, consumers = Consumers} =
         lists:foldl(fun (F, S) -> F(S) end, State,
                     [fun stop_sync_timer/1,
                      fun stop_rate_timer/1,
@@ -326,9 +313,10 @@ terminate_shutdown(Fun, State) ->
         undefined -> State1;
         _         -> ok = rabbit_memory_monitor:deregister(self()),
                      QName = qname(State),
-                     notify_decorators(shutdown, [], State),
+                     notify_decorators(shutdown, State),
                      [emit_consumer_deleted(Ch, CTag, QName) ||
-                         {Ch, CTag, _} <- consumers(State1)],
+                         {Ch, CTag, _, _} <-
+                             rabbit_queue_consumers:all(Consumers)],
                      State1#q{backing_queue_state = Fun(BQS)}
     end.
 
@@ -411,154 +399,19 @@ stop_ttl_timer(State) -> rabbit_misc:stop_timer(State, #q.ttl_timer_ref).
 ensure_stats_timer(State) ->
     rabbit_event:ensure_stats_timer(State, #q.stats_timer, emit_stats).
 
-assert_invariant(State = #q{active_consumers = AC}) ->
-    true = (priority_queue:is_empty(AC) orelse is_empty(State)).
+assert_invariant(State = #q{consumers = Consumers}) ->
+    true = (rabbit_queue_consumers:inactive(Consumers) orelse is_empty(State)).
 
 is_empty(#q{backing_queue = BQ, backing_queue_state = BQS}) -> BQ:is_empty(BQS).
 
-lookup_ch(ChPid) ->
-    case get({ch, ChPid}) of
-        undefined -> not_found;
-        C         -> C
-    end.
-
-ch_record(ChPid, LimiterPid) ->
-    Key = {ch, ChPid},
-    case get(Key) of
-        undefined -> MonitorRef = erlang:monitor(process, ChPid),
-                     Limiter = rabbit_limiter:client(LimiterPid),
-                     C = #cr{ch_pid               = ChPid,
-                             monitor_ref          = MonitorRef,
-                             acktags              = queue:new(),
-                             consumer_count       = 0,
-                             blocked_consumers    = priority_queue:new(),
-                             limiter              = Limiter,
-                             unsent_message_count = 0},
-                     put(Key, C),
-                     C;
-        C = #cr{} -> C
-    end.
-
-update_ch_record(C = #cr{consumer_count       = ConsumerCount,
-                         acktags              = ChAckTags,
-                         unsent_message_count = UnsentMessageCount}) ->
-    case {queue:is_empty(ChAckTags), ConsumerCount, UnsentMessageCount} of
-        {true, 0, 0} -> ok = erase_ch_record(C);
-        _            -> ok = store_ch_record(C)
-    end,
-    C.
-
-store_ch_record(C = #cr{ch_pid = ChPid}) ->
-    put({ch, ChPid}, C),
-    ok.
-
-erase_ch_record(#cr{ch_pid = ChPid, monitor_ref = MonitorRef}) ->
-    erlang:demonitor(MonitorRef),
-    erase({ch, ChPid}),
-    ok.
-
-all_ch_record() -> [C || {{ch, _}, C} <- get()].
-
-block_consumer(C = #cr{blocked_consumers = Blocked},
-               {_ChPid, #consumer{tag = CTag}} = QEntry, State) ->
-    update_ch_record(C#cr{blocked_consumers = add_consumer(QEntry, Blocked)}),
-    notify_decorators(consumer_blocked, [{consumer_tag, CTag}], State).
-
-is_ch_blocked(#cr{unsent_message_count = Count, limiter = Limiter}) ->
-    Count >= ?UNSENT_MESSAGE_LIMIT orelse rabbit_limiter:is_suspended(Limiter).
-
 maybe_send_drained(WasEmpty, State) ->
     case (not WasEmpty) andalso is_empty(State) of
-        true  -> notify_decorators(queue_empty, [], State),
-                 [send_drained(C) || C <- all_ch_record()];
+        true  -> notify_decorators(State),
+                 rabbit_queue_consumers:send_drained();
         false -> ok
     end,
     State.
 
-send_drained(C = #cr{ch_pid = ChPid, limiter = Limiter}) ->
-    case rabbit_limiter:drained(Limiter) of
-        {[], Limiter}          -> ok;
-        {CTagCredit, Limiter2} -> rabbit_channel:send_drained(
-                                    ChPid, CTagCredit),
-                                  update_ch_record(C#cr{limiter = Limiter2})
-    end.
-
-deliver_msgs_to_consumers(_DeliverFun, true, State) ->
-    {true, State};
-deliver_msgs_to_consumers(DeliverFun, false,
-                          State = #q{active_consumers = ActiveConsumers}) ->
-    case priority_queue:out_p(ActiveConsumers) of
-        {empty, _} ->
-            {false, update_consumer_use(State, inactive)};
-        {{value, QEntry, Priority}, Tail} ->
-            {Stop, State1} = deliver_msg_to_consumer(
-                               DeliverFun, QEntry, Priority,
-                               State#q{active_consumers = Tail}),
-            deliver_msgs_to_consumers(DeliverFun, Stop, State1)
-    end.
-
-deliver_msg_to_consumer(DeliverFun, E = {ChPid, Consumer}, Priority, State) ->
-    C = lookup_ch(ChPid),
-    case is_ch_blocked(C) of
-        true  -> block_consumer(C, E, State),
-                 {false, State};
-        false -> case rabbit_limiter:can_send(C#cr.limiter,
-                                              Consumer#consumer.ack_required,
-                                              Consumer#consumer.tag) of
-                     {suspend, Limiter} ->
-                         block_consumer(C#cr{limiter = Limiter}, E, State),
-                         {false, State};
-                     {continue, Limiter} ->
-                         AC1 = priority_queue:in(E, Priority,
-                                                 State#q.active_consumers),
-                         deliver_msg_to_consumer0(
-                           DeliverFun, Consumer, C#cr{limiter = Limiter},
-                           State#q{active_consumers = AC1})
-                 end
-    end.
-
-deliver_msg_to_consumer0(DeliverFun,
-                         #consumer{tag          = ConsumerTag,
-                                   ack_required = AckRequired},
-                         C = #cr{ch_pid               = ChPid,
-                                 acktags              = ChAckTags,
-                                 unsent_message_count = Count},
-                         State = #q{q = #amqqueue{name = QName}}) ->
-    {{Message, IsDelivered, AckTag}, Stop, State1} =
-        DeliverFun(AckRequired, State),
-    rabbit_channel:deliver(ChPid, ConsumerTag, AckRequired,
-                           {QName, self(), AckTag, IsDelivered, Message}),
-    ChAckTags1 = case AckRequired of
-                     true  -> queue:in(AckTag, ChAckTags);
-                     false -> ChAckTags
-                 end,
-    update_ch_record(C#cr{acktags              = ChAckTags1,
-                          unsent_message_count = Count + 1}),
-    {Stop, State1}.
-
-update_consumer_use(State = #q{consumer_use = CUInfo}, Use) ->
-    State#q{consumer_use = update_consumer_use1(CUInfo, Use)}.
-
-update_consumer_use1({inactive, _, _, _}   = CUInfo, inactive) ->
-    CUInfo;
-update_consumer_use1({active,   _, _}      = CUInfo,   active) ->
-    CUInfo;
-update_consumer_use1({active,   Since,         Avg}, inactive) ->
-    Now = now_micros(),
-    {inactive, Now, Now - Since, Avg};
-update_consumer_use1({inactive, Since, Active, Avg},   active) ->
-    Now = now_micros(),
-    {active, Now, consumer_use_avg(Active, Now - Since, Avg)}.
-
-consumer_use_avg(Active, Inactive, Avg) ->
-    Time = Inactive + Active,
-    Ratio = Active / Time,
-    Weight = erlang:min(1, Time / 1000000),
-    case Avg of
-        undefined -> Ratio;
-        _         -> Ratio * Weight + Avg * (1 - Weight)
-    end.
-
 confirm_messages([], State) ->
     State;
 confirm_messages(MsgIds, State = #q{msg_id_to_channel = MTC}) ->
@@ -577,9 +430,10 @@ confirm_messages(MsgIds, State = #q{msg_id_to_channel = MTC}) ->
     rabbit_misc:gb_trees_foreach(fun rabbit_misc:confirm_to_sender/2, CMs),
     State#q{msg_id_to_channel = MTC1}.
 
-send_or_record_confirm(#delivery{msg_seq_no = undefined}, State) ->
+send_or_record_confirm(#delivery{confirm    = false}, State) ->
     {never, State};
-send_or_record_confirm(#delivery{sender     = SenderPid,
+send_or_record_confirm(#delivery{confirm    = true,
+                                 sender     = SenderPid,
                                  msg_seq_no = MsgSeqNo,
                                  message    = #basic_message {
                                    is_persistent = true,
@@ -588,11 +442,19 @@ send_or_record_confirm(#delivery{sender     = SenderPid,
                                   msg_id_to_channel = MTC}) ->
     MTC1 = gb_trees:insert(MsgId, {SenderPid, MsgSeqNo}, MTC),
     {eventually, State#q{msg_id_to_channel = MTC1}};
-send_or_record_confirm(#delivery{sender     = SenderPid,
+send_or_record_confirm(#delivery{confirm    = true,
+                                 sender     = SenderPid,
                                  msg_seq_no = MsgSeqNo}, State) ->
     rabbit_misc:confirm_to_sender(SenderPid, [MsgSeqNo]),
     {immediately, State}.
 
+send_mandatory(#delivery{mandatory = false}) ->
+    ok;
+send_mandatory(#delivery{mandatory  = true,
+                         sender     = SenderPid,
+                         msg_seq_no = MsgSeqNo}) ->
+    gen_server2:cast(SenderPid, {mandatory_received, MsgSeqNo}).
+
 discard(#delivery{sender     = SenderPid,
                   msg_seq_no = MsgSeqNo,
                   message    = #basic_message{id = MsgId}}, State) ->
@@ -604,56 +466,69 @@ discard(#delivery{sender     = SenderPid,
     BQS1 = BQ:discard(MsgId, SenderPid, BQS),
     State1#q{backing_queue_state = BQS1}.
 
-run_message_queue(State) ->
-    {_Active, State3} = deliver_msgs_to_consumers(
-                          fun(AckRequired, State1) ->
-                                  {Result, State2} = fetch(AckRequired, State1),
-                                  {Result, is_empty(State2), State2}
-                          end, is_empty(State), State),
-    State3.
+run_message_queue(State) -> run_message_queue(false, State).
 
-add_consumer({ChPid, Consumer = #consumer{args = Args}}, ActiveConsumers) ->
-    Priority = case rabbit_misc:table_lookup(Args, <<"x-priority">>) of
-                   {_, P} -> P;
-                   _      -> 0
-               end,
-    priority_queue:in({ChPid, Consumer}, Priority, ActiveConsumers).
+run_message_queue(ActiveConsumersChanged, State) ->
+    case is_empty(State) of
+        true  -> maybe_notify_decorators(ActiveConsumersChanged, State);
+        false -> case rabbit_queue_consumers:deliver(
+                        fun(AckRequired) -> fetch(AckRequired, State) end,
+                        qname(State), State#q.consumers) of
+                     {delivered, ActiveConsumersChanged1, State1, Consumers} ->
+                         run_message_queue(
+                           ActiveConsumersChanged or ActiveConsumersChanged1,
+                           State1#q{consumers = Consumers});
+                     {undelivered, ActiveConsumersChanged1, Consumers} ->
+                         maybe_notify_decorators(
+                           ActiveConsumersChanged or ActiveConsumersChanged1,
+                           State#q{consumers = Consumers})
+                 end
+    end.
 
 attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message},
                  Props, Delivered, State = #q{backing_queue       = BQ,
                                               backing_queue_state = BQS}) ->
-    {IsDuplicate, BQS1} = BQ:is_duplicate(Message, BQS),
-    State1 = State#q{backing_queue_state = BQS1},
-    case IsDuplicate of
-        false -> deliver_msgs_to_consumers(
-                   fun (true, State2 = #q{backing_queue_state = BQS2}) ->
-                           true = BQ:is_empty(BQS2),
-                           {AckTag, BQS3} = BQ:publish_delivered(
-                                              Message, Props, SenderPid, BQS2),
-                           {{Message, Delivered, AckTag},
-                            true, State2#q{backing_queue_state = BQS3}};
-                       (false, State2) ->
-                           {{Message, Delivered, undefined},
-                            true, discard(Delivery, State2)}
-                   end, false, State1);
-        true  -> {true, State1}
+    case rabbit_queue_consumers:deliver(
+           fun (true)  -> true = BQ:is_empty(BQS),
+                          {AckTag, BQS1} = BQ:publish_delivered(
+                                             Message, Props, SenderPid, BQS),
+                          {{Message, Delivered, AckTag},
+                           State#q{backing_queue_state = BQS1}};
+               (false) -> {{Message, Delivered, undefined},
+                           discard(Delivery, State)}
+           end, qname(State), State#q.consumers) of
+        {delivered, ActiveConsumersChanged, State1, Consumers} ->
+            {delivered,   maybe_notify_decorators(
+                            ActiveConsumersChanged,
+                            State1#q{consumers = Consumers})};
+        {undelivered, ActiveConsumersChanged, Consumers} ->
+            {undelivered, maybe_notify_decorators(
+                            ActiveConsumersChanged,
+                            State#q{consumers = Consumers})}
     end.
 
 deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid},
-                   Delivered, State) ->
+                   Delivered, State = #q{backing_queue       = BQ,
+                                         backing_queue_state = BQS}) ->
+    send_mandatory(Delivery), %% must do this before confirms
     {Confirm, State1} = send_or_record_confirm(Delivery, State),
     Props = message_properties(Message, Confirm, State),
-    case attempt_delivery(Delivery, Props, Delivered, State1) of
-        {true, State2} ->
+    {IsDuplicate, BQS1} = BQ:is_duplicate(Message, BQS),
+    State2 = State1#q{backing_queue_state = BQS1},
+    case IsDuplicate orelse attempt_delivery(Delivery, Props, Delivered,
+                                             State2) of
+        true ->
             State2;
+        {delivered, State3} ->
+            State3;
         %% The next one is an optimisation
-        {false, State2 = #q{ttl = 0, dlx = undefined}} ->
-            discard(Delivery, State2);
-        {false, State2 = #q{backing_queue = BQ, backing_queue_state = BQS}} ->
-            BQS1 = BQ:publish(Message, Props, Delivered, SenderPid, BQS),
-            {Dropped, State3 = #q{backing_queue_state = BQS2}} =
-                maybe_drop_head(State2#q{backing_queue_state = BQS1}),
-            QLen = BQ:len(BQS2),
+        {undelivered, State3 = #q{ttl = 0, dlx = undefined}} ->
+            discard(Delivery, State3);
+        {undelivered, State3 = #q{backing_queue_state = BQS2}} ->
+            BQS3 = BQ:publish(Message, Props, Delivered, SenderPid, BQS2),
+            {Dropped, State4 = #q{backing_queue_state = BQS4}} =
+                maybe_drop_head(State3#q{backing_queue_state = BQS3}),
+            QLen = BQ:len(BQS4),
             %% optimisation: it would be perfectly safe to always
             %% invoke drop_expired_msgs here, but that is expensive so
             %% we only do that if a new message that might have an
@@ -662,9 +537,9 @@ deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid},
             %% has no expiry and becomes the head of the queue then
             %% the call is unnecessary.
             case {Dropped > 0, QLen =:= 1, Props#message_properties.expiry} of
-                {false, false,         _} -> State3;
-                {true,  true,  undefined} -> State3;
-                {_,     _,             _} -> drop_expired_msgs(State3)
+                {false, false,         _} -> State4;
+                {true,  true,  undefined} -> State4;
+                {_,     _,             _} -> drop_expired_msgs(State4)
             end
     end.
 
@@ -714,56 +589,18 @@ requeue(AckTags, ChPid, State) ->
     subtract_acks(ChPid, AckTags, State,
                   fun (State1) -> requeue_and_run(AckTags, State1) end).
 
-remove_consumer(ChPid, ConsumerTag, Queue) ->
-    priority_queue:filter(fun ({CP, #consumer{tag = CTag}}) ->
-                                  (CP /= ChPid) or (CTag /= ConsumerTag)
-                          end, Queue).
-
-remove_consumers(ChPid, Queue, QName) ->
-    priority_queue:filter(fun ({CP, #consumer{tag = CTag}}) when CP =:= ChPid ->
-                                  emit_consumer_deleted(ChPid, CTag, QName),
-                                  false;
-                              (_) ->
-                                  true
-                          end, Queue).
-
-possibly_unblock(State, ChPid, Update) ->
-    case lookup_ch(ChPid) of
-        not_found -> State;
-        C         -> C1 = Update(C),
-                     case is_ch_blocked(C) andalso not is_ch_blocked(C1) of
-                         false -> update_ch_record(C1),
-                                  State;
-                         true  -> unblock(State, C1)
-                     end
-    end.
-
-unblock(State, C = #cr{limiter = Limiter}) ->
-    case lists:partition(
-           fun({_P, {_ChPid, #consumer{tag = CTag}}}) ->
-                   rabbit_limiter:is_consumer_blocked(Limiter, CTag)
-           end, priority_queue:to_list(C#cr.blocked_consumers)) of
-        {_, []} ->
-            update_ch_record(C),
-            State;
-        {Blocked, Unblocked} ->
-            BlockedQ   = priority_queue:from_list(Blocked),
-            UnblockedQ = priority_queue:from_list(Unblocked),
-            update_ch_record(C#cr{blocked_consumers = BlockedQ}),
-            AC1 = priority_queue:join(State#q.active_consumers, UnblockedQ),
-            State1 = update_consumer_use(State#q{active_consumers = AC1},
-                                         active),
-            [notify_decorators(
-               consumer_unblocked, [{consumer_tag, CTag}], State1) ||
-                {_P, {_ChPid, #consumer{tag = CTag}}} <- Unblocked],
-            run_message_queue(State1)
+possibly_unblock(Update, ChPid, State = #q{consumers = Consumers}) ->
+    case rabbit_queue_consumers:possibly_unblock(Update, ChPid, Consumers) of
+        unchanged               -> State;
+        {unblocked, Consumers1} -> State1 = State#q{consumers = Consumers1},
+                                   run_message_queue(true, State1)
     end.
 
 should_auto_delete(#q{q = #amqqueue{auto_delete = false}}) -> false;
 should_auto_delete(#q{has_had_consumers = false}) -> false;
 should_auto_delete(State) -> is_unused(State).
 
-handle_ch_down(DownPid, State = #q{active_consumers   = AC,
+handle_ch_down(DownPid, State = #q{consumers          = Consumers,
                                    exclusive_consumer = Holder,
                                    senders            = Senders}) ->
     State1 = State#q{senders = case pmon:is_monitored(DownPid, Senders) of
@@ -771,25 +608,22 @@ handle_ch_down(DownPid, State = #q{active_consumers   = AC,
                                    true  -> credit_flow:peer_down(DownPid),
                                             pmon:demonitor(DownPid, Senders)
                                end},
-    case lookup_ch(DownPid) of
+    case rabbit_queue_consumers:erase_ch(DownPid, Consumers) of
         not_found ->
             {ok, State1};
-        C = #cr{ch_pid            = ChPid,
-                acktags           = ChAckTags,
-                blocked_consumers = Blocked} ->
-            QName = qname(State),
-            AC1 = remove_consumers(ChPid, AC,      QName),
-            _   = remove_consumers(ChPid, Blocked, QName), %% for stats emission
-            ok = erase_ch_record(C),
+        {ChAckTags, ChCTags, Consumers1} ->
+            QName = qname(State1),
+            [emit_consumer_deleted(DownPid, CTag, QName) || CTag <- ChCTags],
             Holder1 = case Holder of
                           {DownPid, _} -> none;
                           Other        -> Other
                       end,
-            State2 = State1#q{active_consumers   = AC1,
+            State2 = State1#q{consumers          = Consumers1,
                               exclusive_consumer = Holder1},
+            notify_decorators(State2),
             case should_auto_delete(State2) of
                 true  -> {stop, State2};
-                false -> {ok, requeue_and_run(queue:to_list(ChAckTags),
+                false -> {ok, requeue_and_run(ChAckTags,
                                               ensure_expiry_timer(State2))}
             end
     end.
@@ -804,10 +638,7 @@ check_exclusive_access(none, true, State) ->
         false -> in_use
     end.
 
-consumer_count() ->
-    lists:sum([Count || #cr{consumer_count = Count} <- all_ch_record()]).
-
-is_unused(_State) -> consumer_count() == 0.
+is_unused(_State) -> rabbit_queue_consumers:count() == 0.
 
 maybe_send_reply(_ChPid, undefined) -> ok;
 maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg).
@@ -819,23 +650,9 @@ backing_queue_timeout(State = #q{backing_queue       = BQ,
     State#q{backing_queue_state = BQ:timeout(BQS)}.
 
 subtract_acks(ChPid, AckTags, State, Fun) ->
-    case lookup_ch(ChPid) of
-        not_found ->
-            State;
-        C = #cr{acktags = ChAckTags} ->
-            update_ch_record(
-              C#cr{acktags = subtract_acks(AckTags, [], ChAckTags)}),
-            Fun(State)
-    end.
-
-subtract_acks([], [], AckQ) ->
-    AckQ;
-subtract_acks([], Prefix, AckQ) ->
-    queue:join(queue:from_list(lists:reverse(Prefix)), AckQ);
-subtract_acks([T | TL] = AckTags, Prefix, AckQ) ->
-    case queue:out(AckQ) of
-        {{value,  T}, QTail} -> subtract_acks(TL,             Prefix, QTail);
-        {{value, AT}, QTail} -> subtract_acks(AckTags, [AT | Prefix], QTail)
+    case rabbit_queue_consumers:subtract_acks(ChPid, AckTags) of
+        not_found -> State;
+        ok        -> Fun(State)
     end.
 
 message_properties(Message, Confirm, #q{ttl = TTL}) ->
@@ -915,117 +732,17 @@ dead_letter_msgs(Fun, Reason, X, State = #q{dlx_routing_key     = RK,
     QName = qname(State),
     {Res, Acks1, BQS1} =
         Fun(fun (Msg, AckTag, Acks) ->
-                    dead_letter_publish(Msg, Reason, X, RK, QName),
+                    rabbit_dead_letter:publish(Msg, Reason, X, RK, QName),
                     [AckTag | Acks]
             end, [], BQS),
     {_Guids, BQS2} = BQ:ack(Acks1, BQS1),
     {Res, State#q{backing_queue_state = BQS2}}.
 
-dead_letter_publish(Msg, Reason, X, RK, QName) ->
-    DLMsg = make_dead_letter_msg(Msg, Reason, X#exchange.name, RK, QName),
-    Delivery = rabbit_basic:delivery(false, DLMsg, undefined),
-    {Queues, Cycles} = detect_dead_letter_cycles(
-                         Reason, DLMsg, rabbit_exchange:route(X, Delivery)),
-    lists:foreach(fun log_cycle_once/1, Cycles),
-    rabbit_amqqueue:deliver( rabbit_amqqueue:lookup(Queues), Delivery),
-    ok.
-
 stop(State) -> stop(noreply, State).
 
 stop(noreply, State) -> {stop, normal, State};
 stop(Reply,   State) -> {stop, normal, Reply, State}.
 
-
-detect_dead_letter_cycles(expired,
-                          #basic_message{content = Content}, Queues) ->
-    #content{properties = #'P_basic'{headers = Headers}} =
-        rabbit_binary_parser:ensure_content_decoded(Content),
-    NoCycles = {Queues, []},
-    case Headers of
-        undefined ->
-            NoCycles;
-        _ ->
-            case rabbit_misc:table_lookup(Headers, <<"x-death">>) of
-                {array, Deaths} ->
-                    {Cycling, NotCycling} =
-                        lists:partition(
-                          fun (#resource{name = Queue}) ->
-                                  is_dead_letter_cycle(Queue, Deaths)
-                          end, Queues),
-                    OldQueues = [rabbit_misc:table_lookup(D, <<"queue">>) ||
-                                    {table, D} <- Deaths],
-                    OldQueues1 = [QName || {longstr, QName} <- OldQueues],
-                    {NotCycling, [[QName | OldQueues1] ||
-                                     #resource{name = QName} <- Cycling]};
-                _ ->
-                    NoCycles
-            end
-    end;
-detect_dead_letter_cycles(_Reason, _Msg, Queues) ->
-    {Queues, []}.
-
-is_dead_letter_cycle(Queue, Deaths) ->
-    {Cycle, Rest} =
-        lists:splitwith(
-          fun ({table, D}) ->
-                  {longstr, Queue} =/= rabbit_misc:table_lookup(D, <<"queue">>);
-              (_) ->
-                  true
-          end, Deaths),
-    %% Is there a cycle, and if so, is it entirely due to expiry?
-    case Rest of
-        []    -> false;
-        [H|_] -> lists:all(
-                   fun ({table, D}) ->
-                           {longstr, <<"expired">>} =:=
-                               rabbit_misc:table_lookup(D, <<"reason">>);
-                       (_) ->
-                           false
-                   end, Cycle ++ [H])
-    end.
-
-make_dead_letter_msg(Msg = #basic_message{content       = Content,
-                                          exchange_name = Exchange,
-                                          routing_keys  = RoutingKeys},
-                     Reason, DLX, RK, #resource{name = QName}) ->
-    {DeathRoutingKeys, HeadersFun1} =
-        case RK of
-            undefined -> {RoutingKeys, fun (H) -> H end};
-            _         -> {[RK], fun (H) -> lists:keydelete(<<"CC">>, 1, H) end}
-        end,
-    ReasonBin = list_to_binary(atom_to_list(Reason)),
-    TimeSec = rabbit_misc:now_ms() div 1000,
-    PerMsgTTL = per_msg_ttl_header(Content#content.properties),
-    HeadersFun2 =
-        fun (Headers) ->
-                %% The first routing key is the one specified in the
-                %% basic.publish; all others are CC or BCC keys.
-                RKs  = [hd(RoutingKeys) | rabbit_basic:header_routes(Headers)],
-                RKs1 = [{longstr, Key} || Key <- RKs],
-                Info = [{<<"reason">>,       longstr,   ReasonBin},
-                        {<<"queue">>,        longstr,   QName},
-                        {<<"time">>,         timestamp, TimeSec},
-                        {<<"exchange">>,     longstr,   Exchange#resource.name},
-                        {<<"routing-keys">>, array,     RKs1}] ++ PerMsgTTL,
-                HeadersFun1(rabbit_basic:prepend_table_header(<<"x-death">>,
-                                                              Info, Headers))
-        end,
-    Content1 = #content{properties = Props} =
-        rabbit_basic:map_headers(HeadersFun2, Content),
-    Content2 = Content1#content{properties =
-                                    Props#'P_basic'{expiration = undefined}},
-    Msg#basic_message{exchange_name = DLX,
-                      id            = rabbit_guid:gen(),
-                      routing_keys  = DeathRoutingKeys,
-                      content       = Content2}.
-
-per_msg_ttl_header(#'P_basic'{expiration = undefined}) ->
-    [];
-per_msg_ttl_header(#'P_basic'{expiration = Expiration}) ->
-    [{<<"original-expiration">>, longstr, Expiration}];
-per_msg_ttl_header(_) ->
-    [].
-
 now_micros() -> timer:now_diff(now(), {0,0,0}).
 
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
@@ -1056,21 +773,16 @@ i(exclusive_consumer_tag, #q{exclusive_consumer = {_ChPid, ConsumerTag}}) ->
 i(messages_ready, #q{backing_queue_state = BQS, backing_queue = BQ}) ->
     BQ:len(BQS);
 i(messages_unacknowledged, _) ->
-    lists:sum([queue:len(C#cr.acktags) || C <- all_ch_record()]);
+    rabbit_queue_consumers:unacknowledged_message_count();
 i(messages, State) ->
     lists:sum([i(Item, State) || Item <- [messages_ready,
                                           messages_unacknowledged]]);
 i(consumers, _) ->
-    consumer_count();
-i(consumer_utilisation, #q{consumer_use = ConsumerUse}) ->
-    case consumer_count() of
+    rabbit_queue_consumers:count();
+i(consumer_utilisation, #q{consumers = Consumers}) ->
+    case rabbit_queue_consumers:count() of
         0 -> '';
-        _ -> case ConsumerUse of
-                 {active, Since, Avg} ->
-                     consumer_use_avg(now_micros() - Since, 0, Avg);
-                 {inactive, Since, Active, Avg} ->
-                     consumer_use_avg(Active, now_micros() - Since, Avg)
-             end
+        _ -> rabbit_queue_consumers:utilisation(Consumers)
     end;
 i(memory, _) ->
     {memory, M} = process_info(self(), memory),
@@ -1096,17 +808,6 @@ i(backing_queue_status, #q{backing_queue_state = BQS, backing_queue = BQ}) ->
 i(Item, _) ->
     throw({bad_argument, Item}).
 
-consumers(#q{active_consumers = ActiveConsumers}) ->
-    lists:foldl(fun (C, Acc) -> consumers(C#cr.blocked_consumers, Acc) end,
-                consumers(ActiveConsumers, []), all_ch_record()).
-
-consumers(Consumers, Acc) ->
-    priority_queue:fold(
-      fun ({ChPid, Consumer}, _P, Acc1) ->
-              #consumer{tag = CTag, ack_required = Ack, args = Args} = Consumer,
-              [{ChPid, CTag, Ack, Args} | Acc1]
-      end, Acc, Consumers).
-
 emit_stats(State) ->
     emit_stats(State, []).
 
@@ -1133,24 +834,36 @@ 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;
-        _                                    -> 0
+        info                                       -> 9;
+        {info, _Items}                             -> 9;
+        consumers                                  -> 9;
+        stat                                       -> 7;
+        {basic_consume, _, _, _, _, _, _, _, _, _} -> consumer_bias(State);
+        {basic_cancel, _, _, _}                    -> consumer_bias(State);
+        _                                          -> 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}              -> consumer_bias(State);
+        {notify_sent, _ChPid, _Credit}       -> consumer_bias(State);
+        {resume, _ChPid}                     -> consumer_bias(State);
         _                                    -> 0
     end.
 
+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;
@@ -1169,7 +882,7 @@ handle_call({init, Recover}, From,
 %% You used to be able to declare an exclusive durable queue. Sadly we
 %% need to still tidy up after that case, there could be the remnants
 %% of one left over from an upgrade. So that's why we don't enforce
-%% Recover = false here.
+%% Recover = new here.
 handle_call({init, Recover}, From,
             State = #q{q = #amqqueue{exclusive_owner = Owner}}) ->
     case rabbit_misc:is_process_alive(Owner) of
@@ -1180,7 +893,8 @@ handle_call({init, Recover}, From,
                     q                   = Q} = State,
                  gen_server2:reply(From, {owner_died, Q}),
                  BQ = backing_queue_module(Q),
-                 BQS = bq_init(BQ, Q, Recover),
+                 {_, Terms} = recovery_status(Recover),
+                 BQS = bq_init(BQ, Q, Terms),
                  %% Rely on terminate to delete the queue.
                  {stop, {shutdown, missing_owner},
                   State#q{backing_queue = BQ, backing_queue_state = BQS}}
@@ -1195,13 +909,8 @@ handle_call({info, Items}, _From, State) ->
     catch Error -> reply({error, Error}, State)
     end;
 
-handle_call(consumers, _From, State) ->
-    reply(consumers(State), State);
-
-handle_call({deliver, Delivery, Delivered}, From, State) ->
-    %% Synchronous, "mandatory" deliver mode.
-    gen_server2:reply(From, ok),
-    noreply(deliver_or_enqueue(Delivery, Delivered, State));
+handle_call(consumers, _From, State = #q{consumers = Consumers}) ->
+    reply(rabbit_queue_consumers:all(Consumers), State);
 
 handle_call({notify_down, ChPid}, _From, State) ->
     %% we want to do this synchronously, so that auto_deleted queues
@@ -1224,10 +933,8 @@ handle_call({basic_get, ChPid, NoAck, LimiterPid}, _From,
         {{Message, IsDelivered, AckTag},
          #q{backing_queue = BQ, backing_queue_state = BQS} = State2} ->
             case AckRequired of
-                true  -> C = #cr{acktags = ChAckTags} =
-                             ch_record(ChPid, LimiterPid),
-                         ChAckTags1 = queue:in(AckTag, ChAckTags),
-                         update_ch_record(C#cr{acktags = ChAckTags1});
+                true  -> ok = rabbit_queue_consumers:record_ack(
+                                ChPid, LimiterPid, AckTag);
                 false -> ok
             end,
             Msg = {QName, self(), AckTag, IsDelivered, Message},
@@ -1235,78 +942,45 @@ handle_call({basic_get, ChPid, NoAck, LimiterPid}, _From,
     end;
 
 handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
-             ConsumerTag, ExclusiveConsume, CreditArgs, OtherArgs, OkMsg},
-            _From, State = #q{active_consumers   = AC,
+             ConsumerTag, ExclusiveConsume, Args, OkMsg},
+            _From, State = #q{consumers          = Consumers,
                               exclusive_consumer = Holder}) ->
     case check_exclusive_access(Holder, ExclusiveConsume, State) of
         in_use -> reply({error, exclusive_consume_unavailable}, State);
-        ok     -> C = #cr{consumer_count = Count,
-                          limiter        = Limiter} =
-                      ch_record(ChPid, LimiterPid),
-                  Limiter1 = case LimiterActive of
-                                 true  -> rabbit_limiter:activate(Limiter);
-                                 false -> Limiter
-                             end,
-                  Limiter2 = case CreditArgs of
-                                 none         -> Limiter1;
-                                 {Crd, Drain} -> rabbit_limiter:credit(
-                                                   Limiter1, ConsumerTag,
-                                                   Crd, Drain)
-                             end,
-                  C1 = update_ch_record(C#cr{consumer_count = Count + 1,
-                                             limiter        = Limiter2}),
-                  case is_empty(State) of
-                      true  -> send_drained(C1);
-                      false -> ok
-                  end,
-                  Consumer = #consumer{tag          = ConsumerTag,
-                                       ack_required = not NoAck,
-                                       args         = OtherArgs},
-                  AC1 = add_consumer({ChPid, Consumer}, AC),
+        ok     -> Consumers1 = rabbit_queue_consumers:add(
+                                 ChPid, ConsumerTag, NoAck,
+                                 LimiterPid, LimiterActive,
+                                 Args, is_empty(State), Consumers),
                   ExclusiveConsumer =
                       if ExclusiveConsume -> {ChPid, ConsumerTag};
                          true             -> Holder
                       end,
-                  State1 = State#q{active_consumers   = AC1,
-                                   has_had_consumers = true,
+                  State1 = State#q{consumers          = Consumers1,
+                                   has_had_consumers  = true,
                                    exclusive_consumer = ExclusiveConsumer},
                   ok = maybe_send_reply(ChPid, OkMsg),
                   emit_consumer_created(ChPid, ConsumerTag, ExclusiveConsume,
-                                        not NoAck, qname(State1), OtherArgs),
-                  notify_decorators(
-                    basic_consume, [{consumer_tag, ConsumerTag}], State1),
+                                        not NoAck, qname(State1), Args),
+                  notify_decorators(State1),
                   reply(ok, run_message_queue(State1))
     end;
 
 handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
-            State = #q{active_consumers   = AC,
+            State = #q{consumers          = Consumers,
                        exclusive_consumer = Holder}) ->
     ok = maybe_send_reply(ChPid, OkMsg),
-    case lookup_ch(ChPid) of
+    case rabbit_queue_consumers:remove(ChPid, ConsumerTag, Consumers) of
         not_found ->
             reply(ok, State);
-        C = #cr{consumer_count    = Count,
-                limiter           = Limiter,
-                blocked_consumers = Blocked} ->
-            AC1      = remove_consumer(ChPid, ConsumerTag, AC),
-            Blocked1 = remove_consumer(ChPid, ConsumerTag, Blocked),
-            Limiter1 = case Count of
-                           1 -> rabbit_limiter:deactivate(Limiter);
-                           _ -> Limiter
-                       end,
-            Limiter2 = rabbit_limiter:forget_consumer(Limiter1, ConsumerTag),
-            update_ch_record(C#cr{consumer_count    = Count - 1,
-                                  limiter           = Limiter2,
-                                  blocked_consumers = Blocked1}),
+        Consumers1 ->
             Holder1 = case Holder of
                           {ChPid, ConsumerTag} -> none;
                           _                    -> Holder
                       end,
-            State1 = State#q{active_consumers   = AC1,
+            State1 = State#q{consumers          = Consumers1,
                              exclusive_consumer = Holder1},
             emit_consumer_deleted(ChPid, ConsumerTag, qname(State1)),
-            notify_decorators(
-              basic_cancel, [{consumer_tag, ConsumerTag}], State1),
+            notify_decorators(State1),
             case should_auto_delete(State1) of
                 false -> reply(ok, ensure_expiry_timer(State1));
                 true  -> stop(ok, State1)
@@ -1316,7 +990,7 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
 handle_call(stat, _From, State) ->
     State1 = #q{backing_queue = BQ, backing_queue_state = BQS} =
         ensure_expiry_timer(State),
-    reply({ok, BQ:len(BQS), consumer_count()}, State1);
+    reply({ok, BQ:len(BQS), rabbit_queue_consumers:count()}, State1);
 
 handle_call({delete, IfUnused, IfEmpty}, _From,
             State = #q{backing_queue_state = BQS, backing_queue = BQ}) ->
@@ -1368,10 +1042,11 @@ handle_call(cancel_sync_mirrors, _From, State) ->
     reply({ok, not_syncing}, State);
 
 handle_call(force_event_refresh, _From,
-            State = #q{exclusive_consumer = Exclusive}) ->
+            State = #q{consumers          = Consumers,
+                       exclusive_consumer = Exclusive}) ->
     rabbit_event:notify(queue_created, infos(?CREATION_EVENT_KEYS, State)),
     QName = qname(State),
-    AllConsumers = consumers(State),
+    AllConsumers = rabbit_queue_consumers:all(Consumers),
     case Exclusive of
         none       -> [emit_consumer_created(
                          Ch, CTag, false, AckRequired, QName, Args) ||
@@ -1388,7 +1063,6 @@ handle_cast({run_backing_queue, Mod, Fun},
 
 handle_cast({deliver, Delivery = #delivery{sender = Sender}, Delivered, Flow},
             State = #q{senders = Senders}) ->
-    %% Asynchronous, non-"mandatory" deliver mode.
     Senders1 = case Flow of
                    flow   -> credit_flow:ack(Sender),
                              pmon:monitor(Sender, Senders);
@@ -1400,10 +1074,10 @@ handle_cast({deliver, Delivery = #delivery{sender = Sender}, Delivered, Flow},
 handle_cast({ack, AckTags, ChPid}, State) ->
     noreply(ack(AckTags, ChPid, State));
 
-handle_cast({reject, AckTags, true, ChPid}, State) ->
+handle_cast({reject, true,  AckTags, ChPid}, State) ->
     noreply(requeue(AckTags, ChPid, State));
 
-handle_cast({reject, AckTags, false, ChPid}, State) ->
+handle_cast({reject, false, AckTags, ChPid}, State) ->
     noreply(with_dlx(
               State#q.dlx,
               fun (X) -> subtract_acks(ChPid, AckTags, State,
@@ -1417,29 +1091,16 @@ handle_cast(delete_immediately, State) ->
     stop(State);
 
 handle_cast({resume, ChPid}, State) ->
-    noreply(
-      possibly_unblock(State, ChPid,
-                       fun (C = #cr{limiter = Limiter}) ->
-                               C#cr{limiter = rabbit_limiter:resume(Limiter)}
-                       end));
+    noreply(possibly_unblock(rabbit_queue_consumers:resume_fun(),
+                             ChPid, State));
 
 handle_cast({notify_sent, ChPid, Credit}, State) ->
-    noreply(
-      possibly_unblock(State, ChPid,
-                       fun (C = #cr{unsent_message_count = Count}) ->
-                               C#cr{unsent_message_count = Count - Credit}
-                       end));
+    noreply(possibly_unblock(rabbit_queue_consumers:notify_sent_fun(Credit),
+                             ChPid, State));
 
 handle_cast({activate_limit, ChPid}, State) ->
-    noreply(
-      possibly_unblock(State, ChPid,
-                       fun (C = #cr{limiter = Limiter}) ->
-                               C#cr{limiter = rabbit_limiter:activate(Limiter)}
-                       end));
-
-handle_cast({flush, ChPid}, State) ->
-    ok = rabbit_channel:flushed(ChPid, self()),
-    noreply(State);
+    noreply(possibly_unblock(rabbit_queue_consumers:activate_limit_fun(),
+                             ChPid, State));
 
 handle_cast({set_ram_duration_target, Duration},
             State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
@@ -1468,25 +1129,21 @@ handle_cast(stop_mirroring, State = #q{backing_queue       = BQ,
                     backing_queue_state = BQS1});
 
 handle_cast({credit, ChPid, CTag, Credit, Drain},
-            State = #q{backing_queue       = BQ,
+            State = #q{consumers           = Consumers,
+                       backing_queue       = BQ,
                        backing_queue_state = BQS}) ->
     Len = BQ:len(BQS),
     rabbit_channel:send_credit_reply(ChPid, Len),
-    C = #cr{limiter = Limiter} = lookup_ch(ChPid),
-    C1 = C#cr{limiter = rabbit_limiter:credit(Limiter, CTag, Credit, Drain)},
-    noreply(case Drain andalso Len == 0 of
-                true  -> update_ch_record(C1),
-                         send_drained(C1),
-                         State;
-                false -> case is_ch_blocked(C1) of
-                             true  -> update_ch_record(C1),
-                                      State;
-                             false -> unblock(State, C1)
-                         end
-            end);
+    noreply(
+      case rabbit_queue_consumers:credit(Len == 0, Credit, Drain, ChPid, CTag,
+                                         Consumers) of
+          unchanged               -> State;
+          {unblocked, Consumers1} -> State1 = State#q{consumers = Consumers1},
+                                     run_message_queue(true, State1)
+      end);
 
 handle_cast(notify_decorators, State) ->
-    notify_decorators(refresh, [], State),
+    notify_decorators(State),
     noreply(State);
 
 handle_cast(policy_changed, State = #q{q = #amqqueue{name = Name}}) ->
@@ -1584,14 +1241,3 @@ handle_pre_hibernate(State = #q{backing_queue = BQ,
     {hibernate, stop_rate_timer(State1)}.
 
 format_message_queue(Opt, MQ) -> rabbit_misc:format_message_queue(Opt, MQ).
-
-log_cycle_once(Queues) ->
-    Key = {queue_cycle, Queues},
-    case get(Key) of
-        true      -> ok;
-        undefined -> rabbit_log:warning(
-                       "Message dropped. Dead-letter queues cycle detected" ++
-                       ": ~p~nThis cycle will NOT be reported again.~n",
-                       [Queues]),
-                     put(Key, true)
-    end.
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index 61b504bc..3d88be7a 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -27,7 +27,8 @@
         ('empty' | {rabbit_types:basic_message(), boolean(), Ack})).
 -type(drop_result(Ack) ::
         ('empty' | {rabbit_types:msg_id(), Ack})).
--type(attempt_recovery() :: boolean()).
+-type(recovery_terms() :: [term()] | 'non_clean_shutdown').
+-type(recovery_info() :: 'new' | recovery_terms()).
 -type(purged_msg_count() :: non_neg_integer()).
 -type(async_callback() ::
         fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok')).
@@ -40,7 +41,10 @@
 %% aren't being started at this point, but this call allows the
 %% backing queue to perform any checking necessary for the consistency
 %% of those queues, or initialise any other shared resources.
--callback start([rabbit_amqqueue:name()]) -> 'ok'.
+%%
+%% The list of queue recovery terms returned as {ok, Terms} must be given
+%% in the same order as the list of queue names supplied.
+-callback start([rabbit_amqqueue:name()]) -> rabbit_types:ok(recovery_terms()).
 
 %% Called to tear down any state/resources. NB: Implementations should
 %% not depend on this function being called on shutdown and instead
@@ -51,15 +55,17 @@
 %%
 %% Takes
 %% 1. the amqqueue record
-%% 2. a boolean indicating whether the queue is an existing queue that
-%%    should be recovered
+%% 2. a term indicating whether the queue is an existing queue that
+%%    should be recovered or not. When 'new' is given, no recovery is
+%%    taking place, otherwise a list of recovery terms is given, or
+%%    the atom 'non_clean_shutdown' if no recovery terms are available.
 %% 3. an asynchronous callback which accepts a function of type
 %%    backing-queue-state to backing-queue-state. This callback
 %%    function can be safely invoked from any process, which makes it
 %%    useful for passing messages back into the backing queue,
 %%    especially as the backing queue does not have control of its own
 %%    mailbox.
--callback init(rabbit_types:amqqueue(), attempt_recovery(),
+-callback init(rabbit_types:amqqueue(), recovery_info(),
                async_callback()) -> state().
 
 %% Called on queue shutdown when queue isn't being deleted.
@@ -203,6 +209,10 @@
 %% Called immediately before the queue hibernates.
 -callback handle_pre_hibernate(state()) -> state().
 
+%% Used to help prioritisation in rabbit_amqqueue_process. The rate of
+%% inbound messages and outbound messages at the moment.
+-callback msg_rates(state()) -> {float(), float()}.
+
 %% Exists for debugging purposes, to be able to expose state via
 %% rabbitmqctl list_queues backing_queue_status
 -callback status(state()) -> [{atom(), any()}].
@@ -230,7 +240,8 @@ behaviour_info(callbacks) ->
      {fetch, 2}, {ack, 2}, {requeue, 2}, {ackfold, 4}, {fold, 3}, {len, 1},
      {is_empty, 1}, {depth, 1}, {set_ram_duration_target, 2},
      {ram_duration, 1}, {needs_timeout, 1}, {timeout, 1},
-     {handle_pre_hibernate, 1}, {status, 1}, {invoke, 3}, {is_duplicate, 2}] ;
+     {handle_pre_hibernate, 1}, {msg_rates, 1}, {status, 1},
+     {invoke, 3}, {is_duplicate, 2}] ;
 behaviour_info(_Other) ->
     undefined.
 
diff --git a/src/rabbit_backing_queue_qc.erl b/src/rabbit_backing_queue_qc.erl
index e2bc3247..b0545915 100644
--- a/src/rabbit_backing_queue_qc.erl
+++ b/src/rabbit_backing_queue_qc.erl
@@ -373,7 +373,7 @@ qc_default_exchange() ->
 
 qc_variable_queue_init(Q) ->
     {call, ?BQMOD, init,
-     [Q, false, function(2, ok)]}.
+     [Q, new, function(2, {ok, []})]}.
 
 qc_test_q() -> {call, rabbit_misc, r, [<<"/">>, queue, noshrink(binary(16))]}.
 
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index 2e825536..a5dc6eb2 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -20,9 +20,9 @@
 
 -export([publish/4, publish/5, publish/1,
          message/3, message/4, properties/1, prepend_table_header/3,
-         extract_headers/1, map_headers/2, delivery/3, header_routes/1,
+         extract_headers/1, map_headers/2, delivery/4, header_routes/1,
          parse_expiration/1]).
--export([build_content/2, from_content/1]).
+-export([build_content/2, from_content/1, msg_size/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -31,8 +31,7 @@
 -type(properties_input() ::
         (rabbit_framing:amqp_property_record() | [{atom(), any()}])).
 -type(publish_result() ::
-        ({ok, rabbit_amqqueue:routing_result(), [pid()]}
-         | rabbit_types:error('not_found'))).
+        ({ok, [pid()]} | rabbit_types:error('not_found'))).
 -type(headers() :: rabbit_framing:amqp_table() | 'undefined').
 
 -type(exchange_input() :: (rabbit_types:exchange() | rabbit_exchange:name())).
@@ -46,8 +45,8 @@
          properties_input(), body_input()) -> publish_result()).
 -spec(publish/1 ::
         (rabbit_types:delivery()) -> publish_result()).
--spec(delivery/3 ::
-        (boolean(), rabbit_types:message(), undefined | integer()) ->
+-spec(delivery/4 ::
+        (boolean(), boolean(), rabbit_types:message(), undefined | integer()) ->
                          rabbit_types:delivery()).
 -spec(message/4 ::
         (rabbit_exchange:name(), rabbit_router:routing_key(),
@@ -77,6 +76,9 @@
         (rabbit_framing:amqp_property_record())
         -> rabbit_types:ok_or_error2('undefined' | non_neg_integer(), any())).
 
+-spec(msg_size/1 :: (rabbit_types:content() | rabbit_types:message()) ->
+                         non_neg_integer()).
+
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -90,10 +92,10 @@ publish(Exchange, RoutingKeyBin, Properties, Body) ->
 %% erlang distributed network.
 publish(X = #exchange{name = XName}, RKey, Mandatory, Props, Body) ->
     Message = message(XName, RKey, properties(Props), Body),
-    publish(X, delivery(Mandatory, Message, undefined));
+    publish(X, delivery(Mandatory, false, Message, undefined));
 publish(XName, RKey, Mandatory, Props, Body) ->
     Message = message(XName, RKey, properties(Props), Body),
-    publish(delivery(Mandatory, Message, undefined)).
+    publish(delivery(Mandatory, false, Message, undefined)).
 
 publish(Delivery = #delivery{
           message = #basic_message{exchange_name = XName}}) ->
@@ -104,11 +106,11 @@ publish(Delivery = #delivery{
 
 publish(X, Delivery) ->
     Qs = rabbit_amqqueue:lookup(rabbit_exchange:route(X, Delivery)),
-    {RoutingRes, DeliveredQPids} = rabbit_amqqueue:deliver(Qs, Delivery),
-    {ok, RoutingRes, DeliveredQPids}.
+    DeliveredQPids = rabbit_amqqueue:deliver(Qs, Delivery),
+    {ok, DeliveredQPids}.
 
-delivery(Mandatory, Message, MsgSeqNo) ->
-    #delivery{mandatory = Mandatory, sender = self(),
+delivery(Mandatory, Confirm, Message, MsgSeqNo) ->
+    #delivery{mandatory = Mandatory, confirm = Confirm, sender = self(),
               message = Message, msg_seq_no = MsgSeqNo}.
 
 build_content(Properties, BodyBin) when is_binary(BodyBin) ->
@@ -274,3 +276,5 @@ parse_expiration(#'P_basic'{expiration = Expiration}) ->
             {error, {leftover_string, S}}
     end.
 
+msg_size(#content{payload_fragments_rev = PFR}) -> iolist_size(PFR);
+msg_size(#basic_message{content = Content})     -> msg_size(Content).
diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl
index ae5bbf51..83f68ed3 100644
--- a/src/rabbit_binary_generator.erl
+++ b/src/rabbit_binary_generator.erl
@@ -119,52 +119,51 @@ create_frame(TypeInt, ChannelInt, Payload) ->
 table_field_to_binary({FName, T, V}) ->
     [short_string_to_binary(FName) | field_value_to_binary(T, V)].
 
-field_value_to_binary(longstr,   V) -> ["S", long_string_to_binary(V)];
-field_value_to_binary(signedint, V) -> ["I", <<V:32/signed>>];
+field_value_to_binary(longstr,   V) -> [$S | long_string_to_binary(V)];
+field_value_to_binary(signedint, V) -> [$I, <<V:32/signed>>];
 field_value_to_binary(decimal,   V) -> {Before, After} = V,
-                                       ["D", Before, <<After:32>>];
-field_value_to_binary(timestamp, V) -> ["T", <<V:64>>];
-field_value_to_binary(table,     V) -> ["F", table_to_binary(V)];
-field_value_to_binary(array,     V) -> ["A", array_to_binary(V)];
-field_value_to_binary(byte,      V) -> ["b", <<V:8/unsigned>>];
-field_value_to_binary(double,    V) -> ["d", <<V:64/float>>];
-field_value_to_binary(float,     V) -> ["f", <<V:32/float>>];
-field_value_to_binary(long,      V) -> ["l", <<V:64/signed>>];
-field_value_to_binary(short,     V) -> ["s", <<V:16/signed>>];
-field_value_to_binary(bool,      V) -> ["t", if V -> 1; true -> 0 end];
-field_value_to_binary(binary,    V) -> ["x", long_string_to_binary(V)];
-field_value_to_binary(void,     _V) -> ["V"].
+                                       [$D, Before, <<After:32>>];
+field_value_to_binary(timestamp, V) -> [$T, <<V:64>>];
+field_value_to_binary(table,     V) -> [$F | table_to_binary(V)];
+field_value_to_binary(array,     V) -> [$A | array_to_binary(V)];
+field_value_to_binary(byte,      V) -> [$b, <<V:8/unsigned>>];
+field_value_to_binary(double,    V) -> [$d, <<V:64/float>>];
+field_value_to_binary(float,     V) -> [$f, <<V:32/float>>];
+field_value_to_binary(long,      V) -> [$l, <<V:64/signed>>];
+field_value_to_binary(short,     V) -> [$s, <<V:16/signed>>];
+field_value_to_binary(bool,      V) -> [$t, if V -> 1; true -> 0 end];
+field_value_to_binary(binary,    V) -> [$x | long_string_to_binary(V)];
+field_value_to_binary(void,     _V) -> [$V].
 
 table_to_binary(Table) when is_list(Table) ->
-    BinTable = generate_table(Table),
-    [<<(size(BinTable)):32>>, BinTable].
+    BinTable = generate_table_iolist(Table),
+    [<<(iolist_size(BinTable)):32>> | BinTable].
 
 array_to_binary(Array) when is_list(Array) ->
-    BinArray = generate_array(Array),
-    [<<(size(BinArray)):32>>, BinArray].
+    BinArray = generate_array_iolist(Array),
+    [<<(iolist_size(BinArray)):32>> | BinArray].
 
 generate_table(Table) when is_list(Table) ->
-    list_to_binary(lists:map(fun table_field_to_binary/1, Table)).
+    list_to_binary(generate_table_iolist(Table)).
 
-generate_array(Array) when is_list(Array) ->
-    list_to_binary(lists:map(fun ({T, V}) -> field_value_to_binary(T, V) end,
-                             Array)).
+generate_table_iolist(Table) ->
+    lists:map(fun table_field_to_binary/1, Table).
+
+generate_array_iolist(Array) ->
+    lists:map(fun ({T, V}) -> field_value_to_binary(T, V) end, Array).
 
-short_string_to_binary(String) when is_binary(String) ->
-    Len = size(String),
-    if Len < 256 -> [<<Len:8>>, String];
-       true      -> exit(content_properties_shortstr_overflow)
-    end;
 short_string_to_binary(String) ->
-    Len = length(String),
+    Len = string_length(String),
     if Len < 256 -> [<<Len:8>>, String];
        true      -> exit(content_properties_shortstr_overflow)
     end.
 
-long_string_to_binary(String) when is_binary(String) ->
-    [<<(size(String)):32>>, String];
 long_string_to_binary(String) ->
-    [<<(length(String)):32>>, String].
+    Len = string_length(String),
+    [<<Len:32>>, String].
+
+string_length(String) when is_binary(String) ->   size(String);
+string_length(String)                        -> length(String).
 
 check_empty_frame_size() ->
     %% Intended to ensure that EMPTY_FRAME_SIZE is defined correctly.
diff --git a/src/rabbit_binary_parser.erl b/src/rabbit_binary_parser.erl
index 088ad0e5..f65d8ea7 100644
--- a/src/rabbit_binary_parser.erl
+++ b/src/rabbit_binary_parser.erl
@@ -53,35 +53,35 @@ parse_array(<<ValueAndRest/binary>>) ->
     {Type, Value, Rest} = parse_field_value(ValueAndRest),
     [{Type, Value} | parse_array(Rest)].
 
-parse_field_value(<<"S", VLen:32/unsigned, V:VLen/binary, R/binary>>) ->
+parse_field_value(<<$S, VLen:32/unsigned, V:VLen/binary, R/binary>>) ->
     {longstr, V, R};
 
-parse_field_value(<<"I", V:32/signed, R/binary>>) ->
+parse_field_value(<<$I, V:32/signed, R/binary>>) ->
     {signedint, V, R};
 
-parse_field_value(<<"D", Before:8/unsigned, After:32/unsigned, R/binary>>) ->
+parse_field_value(<<$D, Before:8/unsigned, After:32/unsigned, R/binary>>) ->
     {decimal, {Before, After}, R};
 
-parse_field_value(<<"T", V:64/unsigned, R/binary>>) ->
+parse_field_value(<<$T, V:64/unsigned, R/binary>>) ->
     {timestamp, V, R};
 
-parse_field_value(<<"F", VLen:32/unsigned, Table:VLen/binary, R/binary>>) ->
+parse_field_value(<<$F, VLen:32/unsigned, Table:VLen/binary, R/binary>>) ->
     {table, parse_table(Table), R};
 
-parse_field_value(<<"A", VLen:32/unsigned, Array:VLen/binary, R/binary>>) ->
+parse_field_value(<<$A, VLen:32/unsigned, Array:VLen/binary, R/binary>>) ->
     {array, parse_array(Array), R};
 
-parse_field_value(<<"b", V:8/unsigned, R/binary>>) -> {byte,        V, R};
-parse_field_value(<<"d", V:64/float,   R/binary>>) -> {double,      V, R};
-parse_field_value(<<"f", V:32/float,   R/binary>>) -> {float,       V, R};
-parse_field_value(<<"l", V:64/signed,  R/binary>>) -> {long,        V, R};
-parse_field_value(<<"s", V:16/signed,  R/binary>>) -> {short,       V, R};
-parse_field_value(<<"t", V:8/unsigned, R/binary>>) -> {bool, (V /= 0), R};
+parse_field_value(<<$b, V:8/unsigned, R/binary>>) -> {byte,        V, R};
+parse_field_value(<<$d, V:64/float,   R/binary>>) -> {double,      V, R};
+parse_field_value(<<$f, V:32/float,   R/binary>>) -> {float,       V, R};
+parse_field_value(<<$l, V:64/signed,  R/binary>>) -> {long,        V, R};
+parse_field_value(<<$s, V:16/signed,  R/binary>>) -> {short,       V, R};
+parse_field_value(<<$t, V:8/unsigned, R/binary>>) -> {bool, (V /= 0), R};
 
-parse_field_value(<<"x", VLen:32/unsigned, V:VLen/binary, R/binary>>) ->
+parse_field_value(<<$x, VLen:32/unsigned, V:VLen/binary, R/binary>>) ->
     {binary, V, R};
 
-parse_field_value(<<"V", R/binary>>) ->
+parse_field_value(<<$V, R/binary>>) ->
     {void, undefined, R}.
 
 ensure_content_decoded(Content = #content{properties = Props})
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 0832f742..4d866908 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -21,8 +21,7 @@
 -behaviour(gen_server2).
 
 -export([start_link/11, do/2, do/3, do_flow/3, flush/1, shutdown/1]).
--export([send_command/2, deliver/4, send_credit_reply/2, send_drained/2,
-         flushed/2]).
+-export([send_command/2, deliver/4, send_credit_reply/2, send_drained/2]).
 -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]).
 -export([refresh_config_local/0, ready_for_close/1]).
 -export([force_event_refresh/0]).
@@ -37,9 +36,9 @@
              conn_name, limiter, tx, next_tag, unacked_message_q, user,
              virtual_host, most_recently_declared_queue,
              queue_names, queue_monitors, consumer_mapping,
-             blocking, queue_consumers, delivering_queues,
+             queue_consumers, delivering_queues,
              queue_collector_pid, stats_timer, confirm_enabled, publish_seqno,
-             unconfirmed, confirmed, capabilities, trace_state}).
+             unconfirmed, confirmed, mandatory, capabilities, trace_state}).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 
@@ -53,7 +52,6 @@
          messages_uncommitted,
          acks_uncommitted,
          prefetch_count,
-         client_flow_blocked,
          state]).
 
 -define(CREATION_EVENT_KEYS,
@@ -99,7 +97,6 @@
 -spec(send_credit_reply/2 :: (pid(), non_neg_integer()) -> 'ok').
 -spec(send_drained/2 :: (pid(), [{rabbit_types:ctag(), non_neg_integer()}])
                         -> 'ok').
--spec(flushed/2 :: (pid(), pid()) -> 'ok').
 -spec(list/0 :: () -> [pid()]).
 -spec(list_local/0 :: () -> [pid()]).
 -spec(info_keys/0 :: () -> rabbit_types:info_keys()).
@@ -149,9 +146,6 @@ send_credit_reply(Pid, Len) ->
 send_drained(Pid, CTagCredit) ->
     gen_server2:cast(Pid, {send_drained, CTagCredit}).
 
-flushed(Pid, QPid) ->
-    gen_server2:cast(Pid, {flushed, QPid}).
-
 list() ->
     rabbit_misc:append_rpc_all_nodes(rabbit_mnesia:cluster_nodes(running),
                                      rabbit_channel, list_local, []).
@@ -194,6 +188,7 @@ force_event_refresh() ->
 init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
       Capabilities, CollectorPid, LimiterPid]) ->
     process_flag(trap_exit, true),
+    ?store_proc_name({ConnName, Channel}),
     ok = pg_local:join(rabbit_channels, self()),
     State = #ch{state                   = starting,
                 protocol                = Protocol,
@@ -212,7 +207,6 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                 queue_names             = dict:new(),
                 queue_monitors          = pmon:new(),
                 consumer_mapping        = dict:new(),
-                blocking                = sets:new(),
                 queue_consumers         = dict:new(),
                 delivering_queues       = sets:new(),
                 queue_collector_pid     = CollectorPid,
@@ -220,6 +214,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                 publish_seqno           = 1,
                 unconfirmed             = dtree:empty(),
                 confirmed               = [],
+                mandatory               = dtree:empty(),
                 capabilities            = Capabilities,
                 trace_state             = rabbit_trace:init(VHost)},
     State1 = rabbit_event:init_stats_timer(State, #ch.stats_timer),
@@ -238,8 +233,9 @@ prioritise_call(Msg, _From, _Len, _State) ->
 
 prioritise_cast(Msg, _Len, _State) ->
     case Msg of
-        {confirm, _MsgSeqNos, _QPid} -> 5;
-        _                            -> 0
+        {confirm,            _MsgSeqNos, _QPid} -> 5;
+        {mandatory_received, _MsgSeqNo,  _QPid} -> 5;
+        _                                       -> 0
     end.
 
 prioritise_info(Msg, _Len, _State) ->
@@ -291,9 +287,6 @@ handle_cast({method, Method, Content, Flow},
             {stop, {Reason, erlang:get_stacktrace()}, State}
     end;
 
-handle_cast({flushed, QPid}, State) ->
-    {noreply, queue_blocked(QPid, State), hibernate};
-
 handle_cast(ready_for_close, State = #ch{state      = closing,
                                          writer_pid = WriterPid}) ->
     ok = rabbit_writer:send_command_sync(WriterPid, #'channel.close_ok'{}),
@@ -346,11 +339,14 @@ handle_cast(force_event_refresh, State) ->
     rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State)),
     noreply(State);
 
-handle_cast({confirm, MsgSeqNos, From}, State) ->
-    State1 = #ch{confirmed = C} = confirm(MsgSeqNos, From, State),
-    Timeout = case C of [] -> hibernate; _ -> 0 end,
+handle_cast({mandatory_received, MsgSeqNo}, State = #ch{mandatory = Mand}) ->
+    %% NB: don't call noreply/1 since we don't want to send confirms.
+    noreply_coalesce(State#ch{mandatory = dtree:drop(MsgSeqNo, Mand)});
+
+handle_cast({confirm, MsgSeqNos, QPid}, State = #ch{unconfirmed = UC}) ->
+    {MXs, UC1} = dtree:take(MsgSeqNos, QPid, UC),
     %% NB: don't call noreply/1 since we don't want to send confirms.
-    {noreply, ensure_stats_timer(State1), Timeout}.
+    noreply_coalesce(record_confirms(MXs, State#ch{unconfirmed = UC1})).
 
 handle_info({bump_credit, Msg}, State) ->
     credit_flow:handle_bump_msg(Msg),
@@ -368,8 +364,7 @@ handle_info(emit_stats, State) ->
 
 handle_info({'DOWN', _MRef, process, QPid, Reason}, State) ->
     State1 = handle_publishing_queue_down(QPid, Reason, State),
-    State2 = queue_blocked(QPid, State1),
-    State3 = handle_consuming_queue_down(QPid, State2),
+    State3 = handle_consuming_queue_down(QPid, State1),
     State4 = handle_delivering_queue_down(QPid, State3),
     credit_flow:peer_down(QPid),
     #ch{queue_names = QNames, queue_monitors = QMons} = State4,
@@ -416,6 +411,10 @@ noreply(NewState) -> {noreply, next_state(NewState), hibernate}.
 
 next_state(State) -> ensure_stats_timer(send_confirms(State)).
 
+noreply_coalesce(State = #ch{confirmed = C}) ->
+    Timeout = case C of [] -> hibernate; _ -> 0 end,
+    {noreply, ensure_stats_timer(State), Timeout}.
+
 ensure_stats_timer(State) ->
     rabbit_event:ensure_stats_timer(State, #ch.stats_timer, emit_stats).
 
@@ -480,9 +479,8 @@ check_resource_access(User, Resource, Perm) ->
                  put(permission_cache, [V | CacheTail])
     end.
 
-clear_permission_cache() ->
-    erase(permission_cache),
-    ok.
+clear_permission_cache() -> erase(permission_cache),
+                            ok.
 
 check_configure_permitted(Resource, #ch{user = User}) ->
     check_resource_access(User, Resource, configure).
@@ -522,15 +520,22 @@ check_internal_exchange(#exchange{name = Name, internal = true}) ->
 check_internal_exchange(_) ->
     ok.
 
-qbin_to_resource(QueueNameBin, #ch{virtual_host = VHostPath}) ->
-    rabbit_misc:r(VHostPath, queue, QueueNameBin).
+check_msg_size(Content) ->
+    Size = rabbit_basic:msg_size(Content),
+    case Size > ?MAX_MSG_SIZE of
+        true  -> precondition_failed("message size ~B larger than max size ~B",
+                                     [Size, ?MAX_MSG_SIZE]);
+        false -> ok
+    end.
+
+qbin_to_resource(QueueNameBin, State) ->
+    name_to_resource(queue, QueueNameBin, State).
 
 name_to_resource(Type, NameBin, #ch{virtual_host = VHostPath}) ->
     rabbit_misc:r(VHostPath, Type, NameBin).
 
 expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = <<>>}) ->
-    rabbit_misc:protocol_error(
-      not_found, "no previously declared queue", []);
+    rabbit_misc:protocol_error(not_found, "no previously declared queue", []);
 expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = MRDQ}) ->
     MRDQ;
 expand_queue_name_shortcut(QueueNameBin, _) ->
@@ -538,8 +543,7 @@ expand_queue_name_shortcut(QueueNameBin, _) ->
 
 expand_routing_key_shortcut(<<>>, <<>>,
                             #ch{most_recently_declared_queue = <<>>}) ->
-    rabbit_misc:protocol_error(
-      not_found, "no previously declared queue", []);
+    rabbit_misc:protocol_error(not_found, "no previously declared queue", []);
 expand_routing_key_shortcut(<<>>, <<>>,
                             #ch{most_recently_declared_queue = MRDQ}) ->
     MRDQ;
@@ -593,31 +597,11 @@ check_name(Kind, NameBin = <<"amq.", _/binary>>) ->
 check_name(_Kind, NameBin) ->
     NameBin.
 
-queue_blocked(QPid, State = #ch{blocking = Blocking}) ->
-    case sets:is_element(QPid, Blocking) of
-        false -> State;
-        true  -> maybe_send_flow_ok(
-                   State#ch{blocking = sets:del_element(QPid, Blocking)})
-    end.
-
-maybe_send_flow_ok(State = #ch{blocking = Blocking}) ->
-    case sets:size(Blocking) of
-        0 -> ok = send(#'channel.flow_ok'{active = false}, State);
-        _ -> ok
-    end,
-    State.
-
 record_confirms([], State) ->
     State;
 record_confirms(MXs, State = #ch{confirmed = C}) ->
     State#ch{confirmed = [MXs | C]}.
 
-confirm([], _QPid, State) ->
-    State;
-confirm(MsgSeqNos, QPid, State = #ch{unconfirmed = UC}) ->
-    {MXs, UC1} = dtree:take(MsgSeqNos, QPid, UC),
-    record_confirms(MXs, State#ch{unconfirmed = UC1}).
-
 handle_method(#'channel.open'{}, _, State = #ch{state = starting}) ->
     %% Don't leave "starting" as the state for 5s. TODO is this TRTTD?
     State1 = State#ch{state = running},
@@ -679,6 +663,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                    tx              = Tx,
                                    confirm_enabled = ConfirmEnabled,
                                    trace_state     = TraceState}) ->
+    check_msg_size(Content),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, State),
     Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
@@ -689,16 +674,18 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
         rabbit_binary_parser:ensure_content_decoded(Content),
     check_user_id_header(Props, State),
     check_expiration_header(Props),
+    DoConfirm = Tx =/= none orelse ConfirmEnabled,
     {MsgSeqNo, State1} =
-        case {Tx, ConfirmEnabled} of
-            {none, false} -> {undefined, State};
-            {_, _}        -> SeqNo = State#ch.publish_seqno,
-                             {SeqNo, State#ch{publish_seqno = SeqNo + 1}}
+        case DoConfirm orelse Mandatory of
+            false -> {undefined, State};
+            true  -> SeqNo = State#ch.publish_seqno,
+                     {SeqNo, State#ch{publish_seqno = SeqNo + 1}}
         end,
     case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of
         {ok, Message} ->
             rabbit_trace:tap_in(Message, TraceState),
-            Delivery = rabbit_basic:delivery(Mandatory, Message, MsgSeqNo),
+            Delivery = rabbit_basic:delivery(
+                         Mandatory, DoConfirm, Message, MsgSeqNo),
             QNames = rabbit_exchange:route(Exchange, Delivery),
             DQ = {Delivery, QNames},
             {noreply, case Tx of
@@ -712,12 +699,11 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
 
 handle_method(#'basic.nack'{delivery_tag = DeliveryTag,
                             multiple     = Multiple,
-                            requeue      = Requeue},
-              _, State) ->
+                            requeue      = Requeue}, _, State) ->
     reject(DeliveryTag, Requeue, Multiple, State);
 
 handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
-                           multiple = Multiple},
+                           multiple     = Multiple},
               _, State = #ch{unacked_message_q = UAMQ, tx = Tx}) ->
     {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, Multiple),
     State1 = State#ch{unacked_message_q = Remaining},
@@ -728,8 +714,7 @@ handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
                                   State1#ch{tx = {Msgs, Acks1}}
               end};
 
-handle_method(#'basic.get'{queue = QueueNameBin,
-                           no_ack = NoAck},
+handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck},
               _, State = #ch{writer_pid = WriterPid,
                              conn_pid   = ConnPid,
                              limiter    = Limiter,
@@ -767,9 +752,9 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
                                exclusive    = ExclusiveConsume,
                                nowait       = NoWait,
                                arguments    = Args},
-              _, State = #ch{conn_pid          = ConnPid,
-                             limiter           = Limiter,
-                             consumer_mapping  = ConsumerMapping}) ->
+              _, State = #ch{conn_pid         = ConnPid,
+                             limiter          = Limiter,
+                             consumer_mapping = ConsumerMapping}) ->
     case dict:find(ConsumerTag, ConsumerMapping) of
         error ->
             QueueName = qbin_to_resource(QueueNameBin, State),
@@ -787,13 +772,11 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
             case rabbit_amqqueue:with_exclusive_access_or_die(
                    QueueName, ConnPid,
                    fun (Q) ->
-                           {CreditArgs, OtherArgs} = parse_credit_args(Args),
                            {rabbit_amqqueue:basic_consume(
                               Q, NoAck, self(),
                               rabbit_limiter:pid(Limiter),
                               rabbit_limiter:is_active(Limiter),
-                              ActualConsumerTag, ExclusiveConsume,
-                              CreditArgs, OtherArgs,
+                              ActualConsumerTag, ExclusiveConsume, Args,
                               ok_msg(NoWait, #'basic.consume_ok'{
                                        consumer_tag = ActualConsumerTag})),
                             Q}
@@ -819,8 +802,7 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
               not_allowed, "attempt to reuse consumer tag '~s'", [ConsumerTag])
     end;
 
-handle_method(#'basic.cancel'{consumer_tag = ConsumerTag,
-                              nowait = NoWait},
+handle_method(#'basic.cancel'{consumer_tag = ConsumerTag, nowait = NoWait},
               _, State = #ch{consumer_mapping = ConsumerMapping,
                              queue_consumers  = QCons}) ->
     OkMsg = #'basic.cancel_ok'{consumer_tag = ConsumerTag},
@@ -867,23 +849,27 @@ handle_method(#'basic.qos'{prefetch_size = Size}, _, _State) when Size /= 0 ->
     rabbit_misc:protocol_error(not_implemented,
                                "prefetch_size!=0 (~w)", [Size]);
 
-handle_method(#'basic.qos'{prefetch_count = 0}, _,
-              State = #ch{limiter = Limiter}) ->
+handle_method(#'basic.qos'{prefetch_count = 0},
+              _, State = #ch{limiter = Limiter}) ->
     Limiter1 = rabbit_limiter:unlimit_prefetch(Limiter),
     {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}};
 
-handle_method(#'basic.qos'{prefetch_count = PrefetchCount}, _,
-              State = #ch{limiter = Limiter, unacked_message_q = UAMQ}) ->
+handle_method(#'basic.qos'{prefetch_count = PrefetchCount},
+              _, State = #ch{limiter = Limiter, unacked_message_q = UAMQ}) ->
     %% TODO queue:len(UAMQ) is not strictly right since that counts
     %% unacked messages from basic.get too. Pretty obscure though.
     Limiter1 = rabbit_limiter:limit_prefetch(Limiter,
                                              PrefetchCount, queue:len(UAMQ)),
-    {reply, #'basic.qos_ok'{},
-     maybe_limit_queues(Limiter, Limiter1, State#ch{limiter = Limiter1})};
+    case ((not rabbit_limiter:is_active(Limiter)) andalso
+          rabbit_limiter:is_active(Limiter1)) of
+        true  -> rabbit_amqqueue:activate_limit_all(
+                   consumer_queues(State#ch.consumer_mapping), self());
+        false -> ok
+    end,
+    {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}};
 
 handle_method(#'basic.recover_async'{requeue = true},
-              _, State = #ch{unacked_message_q = UAMQ,
-                             limiter = Limiter}) ->
+              _, State = #ch{unacked_message_q = UAMQ, limiter = Limiter}) ->
     OkFun = fun () -> ok end,
     UAMQL = queue:to_list(UAMQ),
     foreach_per_queue(
@@ -905,19 +891,18 @@ handle_method(#'basic.recover'{requeue = Requeue}, Content, State) ->
                                       Content, State),
     {reply, #'basic.recover_ok'{}, State1};
 
-handle_method(#'basic.reject'{delivery_tag = DeliveryTag,
-                              requeue = Requeue},
+handle_method(#'basic.reject'{delivery_tag = DeliveryTag, requeue = Requeue},
               _, State) ->
     reject(DeliveryTag, Requeue, false, State);
 
-handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
-                                  type = TypeNameBin,
-                                  passive = false,
-                                  durable = Durable,
+handle_method(#'exchange.declare'{exchange    = ExchangeNameBin,
+                                  type        = TypeNameBin,
+                                  passive     = false,
+                                  durable     = Durable,
                                   auto_delete = AutoDelete,
-                                  internal = Internal,
-                                  nowait = NoWait,
-                                  arguments = Args},
+                                  internal    = Internal,
+                                  nowait      = NoWait,
+                                  arguments   = Args},
               _, State = #ch{virtual_host = VHostPath}) ->
     CheckedType = rabbit_exchange:check_type(TypeNameBin),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
@@ -950,17 +935,17 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
     return_ok(State, NoWait, #'exchange.declare_ok'{});
 
 handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
-                                  passive = true,
-                                  nowait = NoWait},
+                                  passive  = true,
+                                  nowait   = NoWait},
               _, State = #ch{virtual_host = VHostPath}) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_not_default_exchange(ExchangeName),
     _ = rabbit_exchange:lookup_or_die(ExchangeName),
     return_ok(State, NoWait, #'exchange.declare_ok'{});
 
-handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
+handle_method(#'exchange.delete'{exchange  = ExchangeNameBin,
                                  if_unused = IfUnused,
-                                 nowait = NoWait},
+                                 nowait    = NoWait},
               _, State = #ch{virtual_host = VHostPath}) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_not_default_exchange(ExchangeName),
@@ -976,19 +961,19 @@ handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
     end;
 
 handle_method(#'exchange.bind'{destination = DestinationNameBin,
-                               source = SourceNameBin,
+                               source      = SourceNameBin,
                                routing_key = RoutingKey,
-                               nowait = NoWait,
-                               arguments = Arguments}, _, State) ->
+                               nowait      = NoWait,
+                               arguments   = Arguments}, _, State) ->
     binding_action(fun rabbit_binding:add/2,
                    SourceNameBin, exchange, DestinationNameBin, RoutingKey,
                    Arguments, #'exchange.bind_ok'{}, NoWait, State);
 
 handle_method(#'exchange.unbind'{destination = DestinationNameBin,
-                                 source = SourceNameBin,
+                                 source      = SourceNameBin,
                                  routing_key = RoutingKey,
-                                 nowait = NoWait,
-                                 arguments = Arguments}, _, State) ->
+                                 nowait      = NoWait,
+                                 arguments   = Arguments}, _, State) ->
     binding_action(fun rabbit_binding:remove/2,
                    SourceNameBin, exchange, DestinationNameBin, RoutingKey,
                    Arguments, #'exchange.unbind_ok'{}, NoWait, State);
@@ -1080,10 +1065,10 @@ handle_method(#'queue.declare'{queue   = QueueNameBin,
     return_queue_declare_ok(QueueName, NoWait, MessageCount, ConsumerCount,
                             State);
 
-handle_method(#'queue.delete'{queue = QueueNameBin,
+handle_method(#'queue.delete'{queue     = QueueNameBin,
                               if_unused = IfUnused,
-                              if_empty = IfEmpty,
-                              nowait = NoWait},
+                              if_empty  = IfEmpty,
+                              nowait    = NoWait},
               _, State = #ch{conn_pid = ConnPid}) ->
     QueueName = qbin_to_resource(QueueNameBin, State),
     check_configure_permitted(QueueName, State),
@@ -1105,25 +1090,24 @@ handle_method(#'queue.delete'{queue = QueueNameBin,
                       #'queue.delete_ok'{message_count = PurgedMessageCount})
     end;
 
-handle_method(#'queue.bind'{queue = QueueNameBin,
-                            exchange = ExchangeNameBin,
+handle_method(#'queue.bind'{queue       = QueueNameBin,
+                            exchange    = ExchangeNameBin,
                             routing_key = RoutingKey,
-                            nowait = NoWait,
-                            arguments = Arguments}, _, State) ->
+                            nowait      = NoWait,
+                            arguments   = Arguments}, _, State) ->
     binding_action(fun rabbit_binding:add/2,
                    ExchangeNameBin, queue, QueueNameBin, RoutingKey, Arguments,
                    #'queue.bind_ok'{}, NoWait, State);
 
-handle_method(#'queue.unbind'{queue = QueueNameBin,
-                              exchange = ExchangeNameBin,
+handle_method(#'queue.unbind'{queue       = QueueNameBin,
+                              exchange    = ExchangeNameBin,
                               routing_key = RoutingKey,
-                              arguments = Arguments}, _, State) ->
+                              arguments   = Arguments}, _, State) ->
     binding_action(fun rabbit_binding:remove/2,
                    ExchangeNameBin, queue, QueueNameBin, RoutingKey, Arguments,
                    #'queue.unbind_ok'{}, false, State);
 
-handle_method(#'queue.purge'{queue = QueueNameBin,
-                             nowait = NoWait},
+handle_method(#'queue.purge'{queue = QueueNameBin, nowait = NoWait},
               _, State = #ch{conn_pid = ConnPid}) ->
     QueueName = qbin_to_resource(QueueNameBin, State),
     check_read_permitted(QueueName, State),
@@ -1171,41 +1155,16 @@ handle_method(#'confirm.select'{nowait = NoWait}, _, State) ->
     return_ok(State#ch{confirm_enabled = true},
               NoWait, #'confirm.select_ok'{});
 
-handle_method(#'channel.flow'{active = true}, _,
-              State = #ch{limiter = Limiter}) ->
-    Limiter1 = rabbit_limiter:unblock(Limiter),
-    {reply, #'channel.flow_ok'{active = true},
-     maybe_limit_queues(Limiter, Limiter1, State#ch{limiter = Limiter1})};
-
-handle_method(#'channel.flow'{active = false}, _,
-              State = #ch{consumer_mapping = Consumers,
-                          limiter          = Limiter}) ->
-    case rabbit_limiter:is_blocked(Limiter) of
-        true  -> {noreply, maybe_send_flow_ok(State)};
-        false -> Limiter1 = rabbit_limiter:block(Limiter),
-                 State1 = maybe_limit_queues(Limiter, Limiter1,
-                                             State#ch{limiter = Limiter1}),
-                 %% The semantics of channel.flow{active=false}
-                 %% require that no messages are delivered after the
-                 %% channel.flow_ok has been sent. We accomplish that
-                 %% by "flushing" all messages in flight from the
-                 %% consumer queues to us. To do this we tell all the
-                 %% queues to invoke rabbit_channel:flushed/2, which
-                 %% will send us a {flushed, ...} message that appears
-                 %% *after* all the {deliver, ...} messages. We keep
-                 %% track of all the QPids thus asked, and once all of
-                 %% them have responded (or died) we send the
-                 %% channel.flow_ok.
-                 QPids = consumer_queues(Consumers),
-                 ok = rabbit_amqqueue:flush_all(QPids, self()),
-                 {noreply, maybe_send_flow_ok(
-                             State1#ch{blocking = sets:from_list(QPids)})}
-    end;
+handle_method(#'channel.flow'{active = true}, _, State) ->
+    {reply, #'channel.flow_ok'{active = true}, State};
+
+handle_method(#'channel.flow'{active = false}, _, _State) ->
+    rabbit_misc:protocol_error(not_implemented, "active=false", []);
 
 handle_method(#'basic.credit'{consumer_tag = CTag,
                               credit       = Credit,
-                              drain        = Drain}, _,
-              State = #ch{consumer_mapping = Consumers}) ->
+                              drain        = Drain},
+              _, State = #ch{consumer_mapping = Consumers}) ->
     case dict:find(CTag, Consumers) of
         {ok, Q} -> ok = rabbit_amqqueue:credit(
                           Q, self(), CTag, Credit, Drain),
@@ -1251,12 +1210,17 @@ monitor_delivering_queue(NoAck, QPid, QName,
                                      false -> sets:add_element(QPid, DQ)
                                  end}.
 
-handle_publishing_queue_down(QPid, Reason, State = #ch{unconfirmed = UC}) ->
+handle_publishing_queue_down(QPid, Reason, State = #ch{unconfirmed = UC,
+                                                       mandatory   = Mand}) ->
+    {MMsgs, Mand1} = dtree:take(QPid, Mand),
+    [basic_return(Msg, State, no_route) || {_, Msg} <- MMsgs],
+    State1 = State#ch{mandatory = Mand1},
     case rabbit_misc:is_abnormal_exit(Reason) of
         true  -> {MXs, UC1} = dtree:take_all(QPid, UC),
-                 send_nacks(MXs, State#ch{unconfirmed = UC1});
+                 send_nacks(MXs, State1#ch{unconfirmed = UC1});
         false -> {MXs, UC1} = dtree:take(QPid, UC),
-                 record_confirms(MXs, State#ch{unconfirmed = UC1})
+                              record_confirms(MXs, State1#ch{unconfirmed = UC1})
+
     end.
 
 handle_consuming_queue_down(QPid,
@@ -1285,16 +1249,6 @@ handle_consuming_queue_down(QPid,
 handle_delivering_queue_down(QPid, State = #ch{delivering_queues = DQ}) ->
     State#ch{delivering_queues = sets:del_element(QPid, DQ)}.
 
-parse_credit_args(Arguments) ->
-    case rabbit_misc:table_lookup(Arguments, <<"x-credit">>) of
-        {table, T} -> {case {rabbit_misc:table_lookup(T, <<"credit">>),
-                             rabbit_misc:table_lookup(T, <<"drain">>)} of
-                           {{long, Credit}, {bool, Drain}} -> {Credit, Drain};
-                           _                               -> none
-                       end, lists:keydelete(<<"x-credit">>, 1, Arguments)};
-        undefined  -> {none, Arguments}
-    end.
-
 binding_action(Fun, ExchangeNameBin, DestinationType, DestinationNameBin,
                RoutingKey, Arguments, ReturnMethod, NoWait,
                State = #ch{virtual_host = VHostPath,
@@ -1334,7 +1288,9 @@ binding_action(Fun, ExchangeNameBin, DestinationType, DestinationNameBin,
 basic_return(#basic_message{exchange_name = ExchangeName,
                             routing_keys  = [RoutingKey | _CcRoutes],
                             content       = Content},
-             #ch{protocol = Protocol, writer_pid = WriterPid}, Reason) ->
+             State = #ch{protocol = Protocol, writer_pid = WriterPid},
+             Reason) ->
+    ?INCR_STATS([{exchange_stats, ExchangeName, 1}], return_unroutable, State),
     {_Close, ReplyCode, ReplyText} = Protocol:lookup_amqp_exception(Reason),
     ok = rabbit_writer:send_command(
            WriterPid,
@@ -1359,7 +1315,7 @@ reject(DeliveryTag, Requeue, Multiple,
 reject(Requeue, Acked, Limiter) ->
     foreach_per_queue(
       fun (QPid, MsgIds) ->
-              rabbit_amqqueue:reject(QPid, MsgIds, Requeue, self())
+              rabbit_amqqueue:reject(QPid, Requeue, MsgIds, self())
       end, Acked),
     ok = notify_limiter(Limiter, Acked).
 
@@ -1462,15 +1418,6 @@ foreach_per_queue(F, UAL) ->
                     end, gb_trees:empty(), UAL),
     rabbit_misc:gb_trees_foreach(F, T).
 
-maybe_limit_queues(OldLimiter, NewLimiter, State) ->
-    case ((not rabbit_limiter:is_active(OldLimiter)) andalso
-          rabbit_limiter:is_active(NewLimiter)) of
-        true  -> Queues = consumer_queues(State#ch.consumer_mapping),
-                 rabbit_amqqueue:activate_limit_all(Queues, self());
-        false -> ok
-    end,
-    State.
-
 consumer_queues(Consumers) ->
     lists:usort([QPid ||
                     {_Key, #amqqueue{pid = QPid}} <- dict:to_list(Consumers)]).
@@ -1482,7 +1429,7 @@ consumer_queues(Consumers) ->
 notify_limiter(Limiter, Acked) ->
     %% optimisation: avoid the potentially expensive 'foldl' in the
     %% common case.
-     case rabbit_limiter:is_prefetch_limited(Limiter) of
+     case rabbit_limiter:is_active(Limiter) of
         false -> ok;
         true  -> case lists:foldl(fun ({_, none, _}, Acc) -> Acc;
                                       ({_,    _, _}, Acc) -> Acc + 1
@@ -1493,18 +1440,19 @@ notify_limiter(Limiter, Acked) ->
     end.
 
 deliver_to_queues({#delivery{message    = #basic_message{exchange_name = XName},
-                             msg_seq_no = undefined,
                              mandatory  = false},
                    []}, State) -> %% optimisation
     ?INCR_STATS([{exchange_stats, XName, 1}], publish, State),
     State;
 deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{
                                                        exchange_name = XName},
+                                        mandatory  = Mandatory,
+                                        confirm    = Confirm,
                                         msg_seq_no = MsgSeqNo},
                    DelQNames}, State = #ch{queue_names    = QNames,
                                            queue_monitors = QMons}) ->
     Qs = rabbit_amqqueue:lookup(DelQNames),
-    {RoutingRes, DeliveredQPids} = rabbit_amqqueue:deliver_flow(Qs, Delivery),
+    DeliveredQPids = rabbit_amqqueue:deliver_flow(Qs, Delivery),
     %% The pmon:monitor_all/2 monitors all queues to which we
     %% delivered. But we want to monitor even queues we didn't deliver
     %% to, since we need their 'DOWN' messages to clean
@@ -1523,31 +1471,37 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{
                                  false -> dict:store(QPid, QName, QNames0)
                              end, pmon:monitor(QPid, QMons0)}
                     end, {QNames, pmon:monitor_all(DeliveredQPids, QMons)}, Qs),
-    State1 = process_routing_result(RoutingRes, DeliveredQPids,
-                                    XName, MsgSeqNo, Message,
-                                    State#ch{queue_names    = QNames1,
-                                             queue_monitors = QMons1}),
+    State1 = State#ch{queue_names    = QNames1,
+                      queue_monitors = QMons1},
+    %% NB: the order here is important since basic.returns must be
+    %% sent before confirms.
+    State2 = process_routing_mandatory(Mandatory, DeliveredQPids, MsgSeqNo,
+                                       Message, State1),
+    State3 = process_routing_confirm(  Confirm,   DeliveredQPids, MsgSeqNo,
+                                       XName,   State2),
     ?INCR_STATS([{exchange_stats, XName, 1} |
                  [{queue_exchange_stats, {QName, XName}, 1} ||
                      QPid        <- DeliveredQPids,
                      {ok, QName} <- [dict:find(QPid, QNames1)]]],
-                publish, State1),
-    State1.
+                publish, State3),
+    State3.
 
-process_routing_result(routed,     _,     _, undefined,   _, State) ->
+process_routing_mandatory(false,     _, _MsgSeqNo, _Msg, State) ->
     State;
-process_routing_result(routed,    [], XName,  MsgSeqNo,   _, State) ->
+process_routing_mandatory(true,     [], _MsgSeqNo,  Msg, State) ->
+    ok = basic_return(Msg, State, no_route),
+    State;
+process_routing_mandatory(true,  QPids,  MsgSeqNo,  Msg, State) ->
+    State#ch{mandatory = dtree:insert(MsgSeqNo, QPids, Msg,
+                                      State#ch.mandatory)}.
+
+process_routing_confirm(false,    _, _MsgSeqNo, _XName, State) ->
+    State;
+process_routing_confirm(true,    [],  MsgSeqNo,  XName, State) ->
     record_confirms([{MsgSeqNo, XName}], State);
-process_routing_result(routed, QPids, XName,  MsgSeqNo,   _, State) ->
+process_routing_confirm(true, QPids,  MsgSeqNo,  XName, State) ->
     State#ch{unconfirmed = dtree:insert(MsgSeqNo, QPids, XName,
-                                        State#ch.unconfirmed)};
-process_routing_result(unroutable, _, XName,  MsgSeqNo, Msg, State) ->
-    ok = basic_return(Msg, State, no_route),
-    ?INCR_STATS([{exchange_stats, XName, 1}], return_unroutable, State),
-    case MsgSeqNo of
-        undefined -> State;
-        _         -> record_confirms([{MsgSeqNo, XName}], State)
-    end.
+                                        State#ch.unconfirmed)}.
 
 send_nacks([], State) ->
     State;
@@ -1650,8 +1604,6 @@ i(state,                   #ch{state = running})         -> credit_flow:state();
 i(state,                   #ch{state = State})            -> State;
 i(prefetch_count, #ch{limiter = Limiter}) ->
     rabbit_limiter:get_prefetch_limit(Limiter);
-i(client_flow_blocked, #ch{limiter = Limiter}) ->
-    rabbit_limiter:is_blocked(Limiter);
 i(Item, _) ->
     throw({bad_argument, Item}).
 
@@ -1672,8 +1624,7 @@ update_measures(Type, Key, Inc, Measure) ->
           end,
     put({Type, Key}, orddict:store(Measure, Cur + Inc, Measures)).
 
-emit_stats(State) ->
-    emit_stats(State, []).
+emit_stats(State) -> emit_stats(State, []).
 
 emit_stats(State, Extra) ->
     Coarse = infos(?STATISTICS_KEYS, State),
diff --git a/src/rabbit_channel_interceptor.erl b/src/rabbit_channel_interceptor.erl
index a19d4b18..49f7e388 100644
--- a/src/rabbit_channel_interceptor.erl
+++ b/src/rabbit_channel_interceptor.erl
@@ -90,5 +90,7 @@ select(Method)  ->
 validate_method(M, M2) ->
     rabbit_misc:method_record_type(M) =:= rabbit_misc:method_record_type(M2).
 
+%% keep dialyzer happy
+-spec internal_error(string(), [any()]) -> no_return().
 internal_error(Format, Args) ->
     rabbit_misc:protocol_error(internal_error, Format, Args).
diff --git a/src/rabbit_channel_sup.erl b/src/rabbit_channel_sup.erl
index df2e80ca..26f9700e 100644
--- a/src/rabbit_channel_sup.erl
+++ b/src/rabbit_channel_sup.erl
@@ -47,9 +47,9 @@
 
 start_link({tcp, Sock, Channel, FrameMax, ReaderPid, ConnName, Protocol, User,
             VHost, Capabilities, Collector}) ->
-    {ok, SupPid} = supervisor2:start_link(?MODULE,
-                                          {tcp, Sock, Channel, FrameMax,
-                                           ReaderPid, Protocol}),
+    {ok, SupPid} = supervisor2:start_link(
+                     ?MODULE, {tcp, Sock, Channel, FrameMax,
+                               ReaderPid, Protocol, {ConnName, Channel}}),
     [LimiterPid] = supervisor2:find_child(SupPid, limiter),
     [WriterPid] = supervisor2:find_child(SupPid, writer),
     {ok, ChannelPid} =
@@ -64,7 +64,8 @@ start_link({tcp, Sock, Channel, FrameMax, ReaderPid, ConnName, Protocol, User,
     {ok, SupPid, {ChannelPid, AState}};
 start_link({direct, Channel, ClientChannelPid, ConnPid, ConnName, Protocol,
             User, VHost, Capabilities, Collector}) ->
-    {ok, SupPid} = supervisor2:start_link(?MODULE, direct),
+    {ok, SupPid} = supervisor2:start_link(
+                     ?MODULE, {direct, {ConnName, Channel}}),
     [LimiterPid] = supervisor2:find_child(SupPid, limiter),
     {ok, ChannelPid} =
         supervisor2:start_child(
@@ -81,10 +82,11 @@ start_link({direct, Channel, ClientChannelPid, ConnPid, ConnName, Protocol,
 init(Type) ->
     {ok, {{one_for_all, 0, 1}, child_specs(Type)}}.
 
-child_specs({tcp, Sock, Channel, FrameMax, ReaderPid, Protocol}) ->
+child_specs({tcp, Sock, Channel, FrameMax, ReaderPid, Protocol, Identity}) ->
     [{writer, {rabbit_writer, start_link,
-               [Sock, Channel, FrameMax, Protocol, ReaderPid, true]},
-      intrinsic, ?MAX_WAIT, worker, [rabbit_writer]} | child_specs(direct)];
-child_specs(direct) ->
-    [{limiter, {rabbit_limiter, start_link, []},
+               [Sock, Channel, FrameMax, Protocol, ReaderPid, Identity, true]},
+      intrinsic, ?MAX_WAIT, worker, [rabbit_writer]}
+     | child_specs({direct, Identity})];
+child_specs({direct, Identity}) ->
+    [{limiter, {rabbit_limiter, start_link, [Identity]},
       transient, ?MAX_WAIT, worker, [rabbit_limiter]}].
diff --git a/src/rabbit_connection_helper_sup.erl b/src/rabbit_connection_helper_sup.erl
index e51615e8..f268d8d6 100644
--- a/src/rabbit_connection_helper_sup.erl
+++ b/src/rabbit_connection_helper_sup.erl
@@ -20,7 +20,7 @@
 
 -export([start_link/0]).
 -export([start_channel_sup_sup/1,
-         start_queue_collector/1]).
+         start_queue_collector/2]).
 
 -export([init/1]).
 
@@ -31,7 +31,8 @@
 -ifdef(use_specs).
 -spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
 -spec(start_channel_sup_sup/1 :: (pid()) -> rabbit_types:ok_pid_or_error()).
--spec(start_queue_collector/1 :: (pid()) -> rabbit_types:ok_pid_or_error()).
+-spec(start_queue_collector/2 :: (pid(), rabbit_types:proc_name()) ->
+                                      rabbit_types:ok_pid_or_error()).
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -45,10 +46,10 @@ start_channel_sup_sup(SupPid) ->
           {channel_sup_sup, {rabbit_channel_sup_sup, start_link, []},
            intrinsic, infinity, supervisor, [rabbit_channel_sup_sup]}).
 
-start_queue_collector(SupPid) ->
+start_queue_collector(SupPid, Identity) ->
     supervisor2:start_child(
       SupPid,
-      {collector, {rabbit_queue_collector, start_link, []},
+      {collector, {rabbit_queue_collector, start_link, [Identity]},
        intrinsic, ?MAX_WAIT, worker, [rabbit_queue_collector]}).
 
 %%----------------------------------------------------------------------------
diff --git a/src/rabbit_dead_letter.erl b/src/rabbit_dead_letter.erl
new file mode 100644
index 00000000..b8a2cc9c
--- /dev/null
+++ b/src/rabbit_dead_letter.erl
@@ -0,0 +1,141 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2013 GoPivotal, Inc.  All rights reserved.
+%%
+
+-module(rabbit_dead_letter).
+
+-export([publish/5]).
+
+-include("rabbit.hrl").
+-include("rabbit_framing.hrl").
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec publish(rabbit_types:message(), atom(), rabbit_types:exchange(),
+              'undefined' | binary(), rabbit_amqqueue:name()) -> 'ok'.
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+publish(Msg, Reason, X, RK, QName) ->
+    DLMsg = make_msg(Msg, Reason, X#exchange.name, RK, QName),
+    Delivery = rabbit_basic:delivery(false, false, DLMsg, undefined),
+    {Queues, Cycles} = detect_cycles(Reason, DLMsg,
+                                     rabbit_exchange:route(X, Delivery)),
+    lists:foreach(fun log_cycle_once/1, Cycles),
+    rabbit_amqqueue:deliver(rabbit_amqqueue:lookup(Queues), Delivery),
+    ok.
+
+make_msg(Msg = #basic_message{content       = Content,
+                              exchange_name = Exchange,
+                              routing_keys  = RoutingKeys},
+         Reason, DLX, RK, #resource{name = QName}) ->
+    {DeathRoutingKeys, HeadersFun1} =
+        case RK of
+            undefined -> {RoutingKeys, fun (H) -> H end};
+            _         -> {[RK], fun (H) -> lists:keydelete(<<"CC">>, 1, H) end}
+        end,
+    ReasonBin = list_to_binary(atom_to_list(Reason)),
+    TimeSec = rabbit_misc:now_ms() div 1000,
+    PerMsgTTL = per_msg_ttl_header(Content#content.properties),
+    HeadersFun2 =
+        fun (Headers) ->
+                %% The first routing key is the one specified in the
+                %% basic.publish; all others are CC or BCC keys.
+                RKs  = [hd(RoutingKeys) | rabbit_basic:header_routes(Headers)],
+                RKs1 = [{longstr, Key} || Key <- RKs],
+                Info = [{<<"reason">>,       longstr,   ReasonBin},
+                        {<<"queue">>,        longstr,   QName},
+                        {<<"time">>,         timestamp, TimeSec},
+                        {<<"exchange">>,     longstr,   Exchange#resource.name},
+                        {<<"routing-keys">>, array,     RKs1}] ++ PerMsgTTL,
+                HeadersFun1(rabbit_basic:prepend_table_header(<<"x-death">>,
+                                                              Info, Headers))
+        end,
+    Content1 = #content{properties = Props} =
+        rabbit_basic:map_headers(HeadersFun2, Content),
+    Content2 = Content1#content{properties =
+                                    Props#'P_basic'{expiration = undefined}},
+    Msg#basic_message{exchange_name = DLX,
+                      id            = rabbit_guid:gen(),
+                      routing_keys  = DeathRoutingKeys,
+                      content       = Content2}.
+
+per_msg_ttl_header(#'P_basic'{expiration = undefined}) ->
+    [];
+per_msg_ttl_header(#'P_basic'{expiration = Expiration}) ->
+    [{<<"original-expiration">>, longstr, Expiration}];
+per_msg_ttl_header(_) ->
+    [].
+
+detect_cycles(expired, #basic_message{content = Content}, Queues) ->
+    #content{properties = #'P_basic'{headers = Headers}} =
+        rabbit_binary_parser:ensure_content_decoded(Content),
+    NoCycles = {Queues, []},
+    case Headers of
+        undefined ->
+            NoCycles;
+        _ ->
+            case rabbit_misc:table_lookup(Headers, <<"x-death">>) of
+                {array, Deaths} ->
+                    {Cycling, NotCycling} =
+                        lists:partition(fun (#resource{name = Queue}) ->
+                                                is_cycle(Queue, Deaths)
+                                        end, Queues),
+                    OldQueues = [rabbit_misc:table_lookup(D, <<"queue">>) ||
+                                    {table, D} <- Deaths],
+                    OldQueues1 = [QName || {longstr, QName} <- OldQueues],
+                    {NotCycling, [[QName | OldQueues1] ||
+                                     #resource{name = QName} <- Cycling]};
+                _ ->
+                    NoCycles
+            end
+    end;
+detect_cycles(_Reason, _Msg, Queues) ->
+    {Queues, []}.
+
+is_cycle(Queue, Deaths) ->
+    {Cycle, Rest} =
+        lists:splitwith(
+          fun ({table, D}) ->
+                  {longstr, Queue} =/= rabbit_misc:table_lookup(D, <<"queue">>);
+              (_) ->
+                  true
+          end, Deaths),
+    %% Is there a cycle, and if so, is it entirely due to expiry?
+    case Rest of
+        []    -> false;
+        [H|_] -> lists:all(
+                   fun ({table, D}) ->
+                           {longstr, <<"expired">>} =:=
+                               rabbit_misc:table_lookup(D, <<"reason">>);
+                       (_) ->
+                           false
+                   end, Cycle ++ [H])
+    end.
+
+log_cycle_once(Queues) ->
+    Key = {queue_cycle, Queues},
+    case get(Key) of
+        true      -> ok;
+        undefined -> rabbit_log:warning(
+                       "Message dropped. Dead-letter queues cycle detected" ++
+                       ": ~p~nThis cycle will NOT be reported again.~n",
+                       [Queues]),
+                     put(Key, true)
+    end.
diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl
index ab8c62fe..447cd893 100644
--- a/src/rabbit_error_logger.erl
+++ b/src/rabbit_error_logger.erl
@@ -87,7 +87,7 @@ publish1(RoutingKey, Format, Data, LogExch) ->
     %% 0-9-1 says the timestamp is a "64 bit POSIX timestamp". That's
     %% second resolution, not millisecond.
     Timestamp = rabbit_misc:now_ms() div 1000,
-    {ok, _RoutingRes, _DeliveredQPids} =
+    {ok, _DeliveredQPids} =
         rabbit_basic:publish(LogExch, RoutingKey,
                              #'P_basic'{content_type = <<"text/plain">>,
                                         timestamp    = Timestamp},
diff --git a/src/rabbit_error_logger_file_h.erl b/src/rabbit_error_logger_file_h.erl
index d59641b0..9421b52e 100644
--- a/src/rabbit_error_logger_file_h.erl
+++ b/src/rabbit_error_logger_file_h.erl
@@ -79,6 +79,25 @@ init_file(File, PrevHandler) ->
 %% filter out "application: foo; exited: stopped; type: temporary"
 handle_event({info_report, _, {_, std_info, _}}, State) ->
     {ok, State};
+%% When a node restarts quickly it is possible the rest of the cluster
+%% will not have had the chance to remove its queues from
+%% Mnesia. That's why rabbit_amqqueue:recover/0 invokes
+%% on_node_down(node()). But before we get there we can receive lots
+%% of messages intended for the old version of the node. The emulator
+%% logs an event for every one of those messages; in extremis this can
+%% bring the server to its knees just logging "Discarding..."
+%% again and again. So just log the first one, then go silent.
+handle_event(Event = {error, _, {emulator, _, ["Discarding message" ++ _]}},
+             State) ->
+    case get(discarding_message_seen) of
+        true      -> {ok, State};
+        undefined -> put(discarding_message_seen, true),
+                     error_logger_file_h:handle_event(Event, State)
+    end;
+%% Clear this state if we log anything else (but not a progress report).
+handle_event(Event = {info_msg, _, _}, State) ->
+    erase(discarding_message_seen),
+    error_logger_file_h:handle_event(Event, State);
 handle_event(Event, State) ->
     error_logger_file_h:handle_event(Event, State).
 
diff --git a/src/rabbit_heartbeat.erl b/src/rabbit_heartbeat.erl
index ca67254b..ff9de67a 100644
--- a/src/rabbit_heartbeat.erl
+++ b/src/rabbit_heartbeat.erl
@@ -16,8 +16,8 @@
 
 -module(rabbit_heartbeat).
 
--export([start/6]).
--export([start_heartbeat_sender/3, start_heartbeat_receiver/3,
+-export([start/6, start/7]).
+-export([start_heartbeat_sender/4, start_heartbeat_receiver/4,
          pause_monitor/1, resume_monitor/1]).
 
 -export([system_continue/3, system_terminate/4, system_code_change/4]).
@@ -39,12 +39,17 @@
          non_neg_integer(), heartbeat_callback(),
          non_neg_integer(), heartbeat_callback()) -> heartbeaters()).
 
--spec(start_heartbeat_sender/3 ::
-        (rabbit_net:socket(), non_neg_integer(), heartbeat_callback()) ->
-                                       rabbit_types:ok(pid())).
--spec(start_heartbeat_receiver/3 ::
-        (rabbit_net:socket(), non_neg_integer(), heartbeat_callback()) ->
-                                         rabbit_types:ok(pid())).
+-spec(start/7 ::
+        (pid(), rabbit_net:socket(), rabbit_types:proc_name(),
+         non_neg_integer(), heartbeat_callback(),
+         non_neg_integer(), heartbeat_callback()) -> heartbeaters()).
+
+-spec(start_heartbeat_sender/4 ::
+        (rabbit_net:socket(), non_neg_integer(), heartbeat_callback(),
+         rabbit_types:proc_type_and_name()) -> rabbit_types:ok(pid())).
+-spec(start_heartbeat_receiver/4 ::
+        (rabbit_net:socket(), non_neg_integer(), heartbeat_callback(),
+         rabbit_types:proc_type_and_name()) -> rabbit_types:ok(pid())).
 
 -spec(pause_monitor/1 :: (heartbeaters()) -> 'ok').
 -spec(resume_monitor/1 :: (heartbeaters()) -> 'ok').
@@ -56,31 +61,35 @@
 -endif.
 
 %%----------------------------------------------------------------------------
-
 start(SupPid, Sock, SendTimeoutSec, SendFun, ReceiveTimeoutSec, ReceiveFun) ->
+    start(SupPid, Sock, unknown,
+          SendTimeoutSec, SendFun, ReceiveTimeoutSec, ReceiveFun).
+
+start(SupPid, Sock, Identity,
+      SendTimeoutSec, SendFun, ReceiveTimeoutSec, ReceiveFun) ->
     {ok, Sender} =
         start_heartbeater(SendTimeoutSec, SupPid, Sock,
                           SendFun, heartbeat_sender,
-                          start_heartbeat_sender),
+                          start_heartbeat_sender, Identity),
     {ok, Receiver} =
         start_heartbeater(ReceiveTimeoutSec, SupPid, Sock,
                           ReceiveFun, heartbeat_receiver,
-                          start_heartbeat_receiver),
+                          start_heartbeat_receiver, Identity),
     {Sender, Receiver}.
 
-start_heartbeat_sender(Sock, TimeoutSec, SendFun) ->
+start_heartbeat_sender(Sock, TimeoutSec, SendFun, Identity) ->
     %% the 'div 2' is there so that we don't end up waiting for nearly
     %% 2 * TimeoutSec before sending a heartbeat in the boundary case
     %% where the last message was sent just after a heartbeat.
     heartbeater({Sock, TimeoutSec * 1000 div 2, send_oct, 0,
-                 fun () -> SendFun(), continue end}).
+                 fun () -> SendFun(), continue end}, Identity).
 
-start_heartbeat_receiver(Sock, TimeoutSec, ReceiveFun) ->
+start_heartbeat_receiver(Sock, TimeoutSec, ReceiveFun, Identity) ->
     %% we check for incoming data every interval, and time out after
     %% two checks with no change. As a result we will time out between
     %% 2 and 3 intervals after the last data has been received.
     heartbeater({Sock, TimeoutSec * 1000, recv_oct, 1,
-                 fun () -> ReceiveFun(), stop end}).
+                 fun () -> ReceiveFun(), stop end}, Identity).
 
 pause_monitor({_Sender,     none}) -> ok;
 pause_monitor({_Sender, Receiver}) -> Receiver ! pause, ok.
@@ -98,17 +107,23 @@ system_code_change(Misc, _Module, _OldVsn, _Extra) ->
     {ok, Misc}.
 
 %%----------------------------------------------------------------------------
-start_heartbeater(0, _SupPid, _Sock, _TimeoutFun, _Name, _Callback) ->
+start_heartbeater(0, _SupPid, _Sock, _TimeoutFun, _Name, _Callback,
+                  _Identity) ->
     {ok, none};
-start_heartbeater(TimeoutSec, SupPid, Sock, TimeoutFun, Name, Callback) ->
+start_heartbeater(TimeoutSec, SupPid, Sock, TimeoutFun, Name, Callback,
+                  Identity) ->
     supervisor2:start_child(
       SupPid, {Name,
-               {rabbit_heartbeat, Callback, [Sock, TimeoutSec, TimeoutFun]},
+               {rabbit_heartbeat, Callback,
+                [Sock, TimeoutSec, TimeoutFun, {Name, Identity}]},
                transient, ?MAX_WAIT, worker, [rabbit_heartbeat]}).
 
-heartbeater(Params) ->
+heartbeater(Params, Identity) ->
     Deb = sys:debug_options([]),
-    {ok, proc_lib:spawn_link(fun () -> heartbeater(Params, Deb, {0, 0}) end)}.
+    {ok, proc_lib:spawn_link(fun () ->
+                                     rabbit_misc:store_proc_name(Identity),
+                                     heartbeater(Params, Deb, {0, 0})
+                             end)}.
 
 heartbeater({Sock, TimeoutMillisec, StatName, Threshold, Handler} = Params,
             Deb, {StatVal, SameCount} = State) ->
diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl
index 22da465b..d37b356c 100644
--- a/src/rabbit_limiter.erl
+++ b/src/rabbit_limiter.erl
@@ -17,8 +17,7 @@
 %% The purpose of the limiter is to stem the flow of messages from
 %% queues to channels, in order to act upon various protocol-level
 %% flow control mechanisms, specifically AMQP 0-9-1's basic.qos
-%% prefetch_count and channel.flow, and AMQP 1.0's link (aka consumer)
-%% credit mechanism.
+%% prefetch_count and AMQP 1.0's link (aka consumer) credit mechanism.
 %%
 %% Each channel has an associated limiter process, created with
 %% start_link/1, which it passes to queues on consumer creation with
@@ -65,11 +64,9 @@
 %%
 %% 1. Channels tell the limiter about basic.qos prefetch counts -
 %%    that's what the limit_prefetch/3, unlimit_prefetch/1,
-%%    is_prefetch_limited/1, get_prefetch_limit/1 API functions are
-%%    about - and channel.flow blocking - that's what block/1,
-%%    unblock/1 and is_blocked/1 are for. They also tell the limiter
-%%    queue state (via the queue) about consumer credit changes -
-%%    that's what credit/4 is for.
+%%    get_prefetch_limit/1 API functions are about. They also tell the
+%%    limiter queue state (via the queue) about consumer credit
+%%    changes - that's what credit/5 is for.
 %%
 %% 2. Queues also tell the limiter queue state about the queue
 %%    becoming empty (via drained/1) and consumers leaving (via
@@ -83,12 +80,11 @@
 %%
 %% 5. Queues ask the limiter for permission (with can_send/3) whenever
 %%    they want to deliver a message to a channel. The limiter checks
-%%    whether a) the channel isn't blocked by channel.flow, b) the
-%%    volume has not yet reached the prefetch limit, and c) whether
-%%    the consumer has enough credit. If so it increments the volume
-%%    and tells the queue to proceed. Otherwise it marks the queue as
-%%    requiring notification (see below) and tells the queue not to
-%%    proceed.
+%%    whether a) the volume has not yet reached the prefetch limit,
+%%    and b) whether the consumer has enough credit. If so it
+%%    increments the volume and tells the queue to proceed. Otherwise
+%%    it marks the queue as requiring notification (see below) and
+%%    tells the queue not to proceed.
 %%
 %% 6. A queue that has been told to proceed (by the return value of
 %%    can_send/3) sends the message to the channel. Conversely, a
@@ -117,16 +113,17 @@
 
 -module(rabbit_limiter).
 
+-include("rabbit.hrl").
+
 -behaviour(gen_server2).
 
--export([start_link/0]).
+-export([start_link/1]).
 %% channel API
--export([new/1, limit_prefetch/3, unlimit_prefetch/1, block/1, unblock/1,
-         is_prefetch_limited/1, is_blocked/1, is_active/1,
+-export([new/1, limit_prefetch/3, unlimit_prefetch/1, is_active/1,
          get_prefetch_limit/1, ack/2, pid/1]).
 %% queue API
 -export([client/1, activate/1, can_send/3, resume/1, deactivate/1,
-         is_suspended/1, is_consumer_blocked/2, credit/4, drained/1,
+         is_suspended/1, is_consumer_blocked/2, credit/5, drained/1,
          forget_consumer/2]).
 %% callbacks
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
@@ -134,27 +131,23 @@
 
 %%----------------------------------------------------------------------------
 
--record(lstate, {pid, prefetch_limited, blocked}).
+-record(lstate, {pid, prefetch_limited}).
 -record(qstate, {pid, state, credits}).
 
 -ifdef(use_specs).
 
 -type(lstate() :: #lstate{pid              :: pid(),
-                          prefetch_limited :: boolean(),
-                          blocked          :: boolean()}).
+                          prefetch_limited :: boolean()}).
 -type(qstate() :: #qstate{pid :: pid(),
                           state :: 'dormant' | 'active' | 'suspended'}).
 
--spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
+-spec(start_link/1 :: (rabbit_types:proc_name()) ->
+                           rabbit_types:ok_pid_or_error()).
 -spec(new/1 :: (pid()) -> lstate()).
 
 -spec(limit_prefetch/3      :: (lstate(), non_neg_integer(), non_neg_integer())
                                -> lstate()).
 -spec(unlimit_prefetch/1    :: (lstate()) -> lstate()).
--spec(block/1               :: (lstate()) -> lstate()).
--spec(unblock/1             :: (lstate()) -> lstate()).
--spec(is_prefetch_limited/1 :: (lstate()) -> boolean()).
--spec(is_blocked/1          :: (lstate()) -> boolean()).
 -spec(is_active/1           :: (lstate()) -> boolean()).
 -spec(get_prefetch_limit/1  :: (lstate()) -> non_neg_integer()).
 -spec(ack/2                 :: (lstate(), non_neg_integer()) -> 'ok').
@@ -168,8 +161,8 @@
 -spec(deactivate/1   :: (qstate()) -> qstate()).
 -spec(is_suspended/1 :: (qstate()) -> boolean()).
 -spec(is_consumer_blocked/2 :: (qstate(), rabbit_types:ctag()) -> boolean()).
--spec(credit/4 :: (qstate(), rabbit_types:ctag(), non_neg_integer(), boolean())
-                  -> qstate()).
+-spec(credit/5 :: (qstate(), rabbit_types:ctag(), non_neg_integer(), boolean(),
+                   boolean()) -> {boolean(), qstate()}).
 -spec(drained/1 :: (qstate())
                    -> {[{rabbit_types:ctag(), non_neg_integer()}], qstate()}).
 -spec(forget_consumer/2 :: (qstate(), rabbit_types:ctag()) -> qstate()).
@@ -180,7 +173,6 @@
 
 -record(lim, {prefetch_count = 0,
               ch_pid,
-              blocked = false,
               queues = orddict:new(), % QPid -> {MonitorRef, Notify}
               volume = 0}).
 %% 'Notify' is a boolean that indicates whether a queue should be
@@ -193,12 +185,12 @@
 %% API
 %%----------------------------------------------------------------------------
 
-start_link() -> gen_server2:start_link(?MODULE, [], []).
+start_link(ProcName) -> gen_server2:start_link(?MODULE, [ProcName], []).
 
 new(Pid) ->
     %% this a 'call' to ensure that it is invoked at most once.
     ok = gen_server:call(Pid, {new, self()}, infinity),
-    #lstate{pid = Pid, prefetch_limited = false, blocked = false}.
+    #lstate{pid = Pid, prefetch_limited = false}.
 
 limit_prefetch(L, PrefetchCount, UnackedCount) when PrefetchCount > 0 ->
     ok = gen_server:call(
@@ -210,19 +202,7 @@ unlimit_prefetch(L) ->
     ok = gen_server:call(L#lstate.pid, unlimit_prefetch, infinity),
     L#lstate{prefetch_limited = false}.
 
-block(L) ->
-    ok = gen_server:call(L#lstate.pid, block, infinity),
-    L#lstate{blocked = true}.
-
-unblock(L) ->
-    ok = gen_server:call(L#lstate.pid, unblock, infinity),
-    L#lstate{blocked = false}.
-
-is_prefetch_limited(#lstate{prefetch_limited = Limited}) -> Limited.
-
-is_blocked(#lstate{blocked = Blocked}) -> Blocked.
-
-is_active(L) -> is_prefetch_limited(L) orelse is_blocked(L).
+is_active(#lstate{prefetch_limited = Limited}) -> Limited.
 
 get_prefetch_limit(#lstate{prefetch_limited = false}) -> 0;
 get_prefetch_limit(L) ->
@@ -245,9 +225,9 @@ can_send(L = #qstate{pid = Pid, state = State, credits = Credits},
     case is_consumer_blocked(L, CTag) of
         false -> case (State =/= active orelse
                        safe_call(Pid, {can_send, self(), AckRequired}, true)) of
-                     true  -> {continue, L#qstate{
-                                credits = record_send_q(CTag, Credits)}};
-                     false -> {suspend, L#qstate{state = suspended}}
+                     true  -> Credits1 = decrement_credit(CTag, Credits),
+                              {continue, L#qstate{credits = Credits1}};
+                     false -> {suspend,  L#qstate{state = suspended}}
                  end;
         true  -> {suspend, L}
     end.
@@ -271,13 +251,17 @@ is_suspended(#qstate{})                  -> false.
 
 is_consumer_blocked(#qstate{credits = Credits}, CTag) ->
     case gb_trees:lookup(CTag, Credits) of
+        none                                    -> false;
         {value, #credit{credit = C}} when C > 0 -> false;
-        {value, #credit{}}                      -> true;
-        none                                    -> false
+        {value, #credit{}}                      -> true
     end.
 
-credit(Limiter = #qstate{credits = Credits}, CTag, Credit, Drain) ->
-    Limiter#qstate{credits = update_credit(CTag, Credit, Drain, Credits)}.
+credit(Limiter = #qstate{credits = Credits}, CTag, Credit, Drain, IsEmpty) ->
+    {Res, Cr} = case IsEmpty andalso Drain of
+                    true  -> {true,  make_credit(0,      false)};
+                    false -> {false, make_credit(Credit, Drain)}
+                end,
+    {Res, Limiter#qstate{credits = gb_trees:enter(CTag, Cr, Credits)}}.
 
 drained(Limiter = #qstate{credits = Credits}) ->
     {CTagCredits, Credits2} =
@@ -303,7 +287,11 @@ forget_consumer(Limiter = #qstate{credits = Credits}, CTag) ->
 %% state for us (#qstate.credits), and maintain a fiction that the
 %% limiter is making the decisions...
 
-record_send_q(CTag, Credits) ->
+make_credit(Credit, Drain) ->
+    %% Using up all credit implies no need to send a 'drained' event
+    #credit{credit = Credit, drain = Drain andalso Credit > 0}.
+
+decrement_credit(CTag, Credits) ->
     case gb_trees:lookup(CTag, Credits) of
         {value, #credit{credit = Credit, drain = Drain}} ->
             update_credit(CTag, Credit - 1, Drain, Credits);
@@ -312,15 +300,14 @@ record_send_q(CTag, Credits) ->
     end.
 
 update_credit(CTag, Credit, Drain, Credits) ->
-    %% Using up all credit implies no need to send a 'drained' event
-    Drain1 = Drain andalso Credit > 0,
-    gb_trees:enter(CTag, #credit{credit = Credit, drain = Drain1}, Credits).
+    gb_trees:update(CTag, make_credit(Credit, Drain), Credits).
 
 %%----------------------------------------------------------------------------
 %% gen_server callbacks
 %%----------------------------------------------------------------------------
 
-init([]) -> {ok, #lim{}}.
+init([ProcName]) -> ?store_proc_name(ProcName),
+                    {ok, #lim{}}.
 
 prioritise_call(get_prefetch_limit, _From, _Len, _State) -> 9;
 prioritise_call(_Msg,               _From, _Len, _State) -> 0.
@@ -339,19 +326,10 @@ handle_call(unlimit_prefetch, _From, State) ->
     {reply, ok, maybe_notify(State, State#lim{prefetch_count = 0,
                                               volume         = 0})};
 
-handle_call(block, _From, State) ->
-    {reply, ok, State#lim{blocked = true}};
-
-handle_call(unblock, _From, State) ->
-    {reply, ok, maybe_notify(State, State#lim{blocked = false})};
-
 handle_call(get_prefetch_limit, _From,
             State = #lim{prefetch_count = PrefetchCount}) ->
     {reply, PrefetchCount, State};
 
-handle_call({can_send, QPid, _AckRequired}, _From,
-            State = #lim{blocked = true}) ->
-    {reply, false, limit_queue(QPid, State)};
 handle_call({can_send, QPid, AckRequired}, _From,
             State = #lim{volume = Volume}) ->
     case prefetch_limit_reached(State) of
@@ -387,8 +365,8 @@ code_change(_, State, _) ->
 %%----------------------------------------------------------------------------
 
 maybe_notify(OldState, NewState) ->
-    case (prefetch_limit_reached(OldState) orelse blocked(OldState)) andalso
-        not (prefetch_limit_reached(NewState) orelse blocked(NewState)) of
+    case prefetch_limit_reached(OldState) andalso
+        not prefetch_limit_reached(NewState) of
         true  -> notify_queues(NewState);
         false -> NewState
     end.
@@ -396,8 +374,6 @@ maybe_notify(OldState, NewState) ->
 prefetch_limit_reached(#lim{prefetch_count = Limit, volume = Volume}) ->
     Limit =/= 0 andalso Volume >= Limit.
 
-blocked(#lim{blocked = Blocked}) -> Blocked.
-
 remember_queue(QPid, State = #lim{queues = Queues}) ->
     case orddict:is_key(QPid, Queues) of
         false -> MRef = erlang:monitor(process, QPid),
diff --git a/src/rabbit_mirror_queue_coordinator.erl b/src/rabbit_mirror_queue_coordinator.erl
index a0e8bcc6..6661408c 100644
--- a/src/rabbit_mirror_queue_coordinator.erl
+++ b/src/rabbit_mirror_queue_coordinator.erl
@@ -323,6 +323,7 @@ ensure_monitoring(CPid, Pids) ->
 %% ---------------------------------------------------------------------------
 
 init([#amqqueue { name = QueueName } = Q, GM, DeathFun, DepthFun]) ->
+    ?store_proc_name(QueueName),
     GM1 = case GM of
               undefined ->
                   {ok, GM2} = gm:start_link(
diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl
index d9cef642..b272c64f 100644
--- a/src/rabbit_mirror_queue_master.erl
+++ b/src/rabbit_mirror_queue_master.erl
@@ -22,7 +22,7 @@
          len/1, is_empty/1, depth/1, drain_confirmed/1,
          dropwhile/2, fetchwhile/4, set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1,
-         status/1, invoke/3, is_duplicate/2]).
+         msg_rates/1, status/1, invoke/3, is_duplicate/2]).
 
 -export([start/1, stop/0]).
 
@@ -145,7 +145,7 @@ sync_mirrors(HandleInfo, EmitStats,
     Log("~p messages to synchronise", [BQ:len(BQS)]),
     {ok, #amqqueue{slave_pids = SPids}} = rabbit_amqqueue:lookup(QName),
     Ref = make_ref(),
-    Syncer = rabbit_mirror_queue_sync:master_prepare(Ref, Log, SPids),
+    Syncer = rabbit_mirror_queue_sync:master_prepare(Ref, QName, Log, SPids),
     gm:broadcast(GM, {sync_start, Ref, Syncer, SPids}),
     S = fun(BQSN) -> State#state{backing_queue_state = BQSN} end,
     case rabbit_mirror_queue_sync:master_go(
@@ -212,7 +212,8 @@ publish(Msg = #basic_message { id = MsgId }, MsgProps, IsDelivered, ChPid,
                          backing_queue       = BQ,
                          backing_queue_state = BQS }) ->
     false = dict:is_key(MsgId, SS), %% ASSERTION
-    ok = gm:broadcast(GM, {publish, ChPid, MsgProps, Msg}),
+    ok = gm:broadcast(GM, {publish, ChPid, MsgProps, Msg},
+                      rabbit_basic:msg_size(Msg)),
     BQS1 = BQ:publish(Msg, MsgProps, IsDelivered, ChPid, BQS),
     ensure_monitoring(ChPid, State #state { backing_queue_state = BQS1 }).
 
@@ -222,7 +223,8 @@ publish_delivered(Msg = #basic_message { id = MsgId }, MsgProps,
                                           backing_queue       = BQ,
                                           backing_queue_state = BQS }) ->
     false = dict:is_key(MsgId, SS), %% ASSERTION
-    ok = gm:broadcast(GM, {publish_delivered, ChPid, MsgProps, Msg}),
+    ok = gm:broadcast(GM, {publish_delivered, ChPid, MsgProps, Msg},
+                      rabbit_basic:msg_size(Msg)),
     {AckTag, BQS1} = BQ:publish_delivered(Msg, MsgProps, ChPid, BQS),
     State1 = State #state { backing_queue_state = BQS1 },
     {AckTag, ensure_monitoring(ChPid, State1)}.
@@ -351,6 +353,9 @@ handle_pre_hibernate(State = #state { backing_queue       = BQ,
                                       backing_queue_state = BQS }) ->
     State #state { backing_queue_state = BQ:handle_pre_hibernate(BQS) }.
 
+msg_rates(#state { backing_queue = BQ, backing_queue_state = BQS }) ->
+    BQ:msg_rates(BQS).
+
 status(State = #state { backing_queue = BQ, backing_queue_state = BQS }) ->
     BQ:status(BQS) ++
         [ {mirror_seen,    dict:size(State #state.seen_status)},
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index 96f89ecc..1f31b5c8 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -79,6 +79,7 @@ set_maximum_since_use(QPid, Age) ->
 info(QPid) -> gen_server2:call(QPid, info, infinity).
 
 init(Q) ->
+    ?store_proc_name(Q#amqqueue.name),
     {ok, {not_started, Q}, hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN,
       ?DESIRED_HIBERNATE}}.
@@ -114,7 +115,7 @@ handle_go(Q = #amqqueue{name = QName}) ->
                    Self, {rabbit_amqqueue, set_ram_duration_target, [Self]}),
             {ok, BQ} = application:get_env(backing_queue_module),
             Q1 = Q #amqqueue { pid = QPid },
-            BQS = bq_init(BQ, Q1, false),
+            BQS = bq_init(BQ, Q1, new),
             State = #state { q                   = Q1,
                              gm                  = GM,
                              backing_queue       = BQ,
@@ -616,6 +617,7 @@ promote_me(From, #state { q                   = Q = #amqqueue { name = QName },
     KS1 = lists:foldl(fun (ChPid0, KS0) ->
                               pmon:demonitor(ChPid0, KS0)
                       end, KS, AwaitGmDown),
+    rabbit_misc:store_proc_name(rabbit_amqqueue_process, QName),
     rabbit_amqqueue_process:init_with_backing_queue_state(
       Q1, rabbit_mirror_queue_master, MasterState, RateTRef, Deliveries, KS1,
       MTC).
diff --git a/src/rabbit_mirror_queue_sync.erl b/src/rabbit_mirror_queue_sync.erl
index 61e90105..e3fae4c0 100644
--- a/src/rabbit_mirror_queue_sync.erl
+++ b/src/rabbit_mirror_queue_sync.erl
@@ -18,7 +18,7 @@
 
 -include("rabbit.hrl").
 
--export([master_prepare/3, master_go/7, slave/7]).
+-export([master_prepare/4, master_go/7, slave/7]).
 
 -define(SYNC_PROGRESS_INTERVAL, 1000000).
 
@@ -61,7 +61,8 @@
 -type(slave_sync_state() :: {[{rabbit_types:msg_id(), ack()}], timer:tref(),
                              bqs()}).
 
--spec(master_prepare/3 :: (reference(), log_fun(), [pid()]) -> pid()).
+-spec(master_prepare/4 :: (reference(), rabbit_amqqueue:name(),
+                               log_fun(), [pid()]) -> pid()).
 -spec(master_go/7 :: (pid(), reference(), log_fun(),
                       rabbit_mirror_queue_master:stats_fun(),
                       rabbit_mirror_queue_master:stats_fun(),
@@ -80,9 +81,12 @@
 %% ---------------------------------------------------------------------------
 %% Master
 
-master_prepare(Ref, Log, SPids) ->
+master_prepare(Ref, QName, Log, SPids) ->
     MPid = self(),
-    spawn_link(fun () -> syncer(Ref, Log, MPid, SPids) end).
+    spawn_link(fun () ->
+                       ?store_proc_name(QName),
+                       syncer(Ref, Log, MPid, SPids)
+               end).
 
 master_go(Syncer, Ref, Log, HandleInfo, EmitStats, BQ, BQS) ->
     Args = {Syncer, Ref, Log, HandleInfo, EmitStats, rabbit_misc:get_parent()},
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 00c4eaf3..1c63980e 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -70,6 +70,8 @@
 -export([interval_operation/4]).
 -export([ensure_timer/4, stop_timer/2]).
 -export([get_parent/0]).
+-export([store_proc_name/1, store_proc_name/2]).
+-export([moving_average/4]).
 
 %% Horrible macro to use in guards
 -define(IS_BENIGN_EXIT(R),
@@ -248,6 +250,10 @@
 -spec(ensure_timer/4 :: (A, non_neg_integer(), non_neg_integer(), any()) -> A).
 -spec(stop_timer/2 :: (A, non_neg_integer()) -> A).
 -spec(get_parent/0 :: () -> pid()).
+-spec(store_proc_name/2 :: (atom(), rabbit_types:proc_name()) -> ok).
+-spec(store_proc_name/1 :: (rabbit_types:proc_type_and_name()) -> ok).
+-spec(moving_average/4 :: (float(), float(), float(), float() | 'undefined')
+                          -> float()).
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -1082,6 +1088,15 @@ stop_timer(State, Idx) ->
                      end
     end.
 
+store_proc_name(Type, ProcName) -> store_proc_name({Type, ProcName}).
+store_proc_name(TypeProcName)   -> put(process_name, TypeProcName).
+
+moving_average(_Time, _HalfLife, Next, undefined) ->
+    Next;
+moving_average(Time,  HalfLife,  Next, Current) ->
+    Weight = math:exp(Time * math:log(0.5) / HalfLife),
+    Next * (1 - Weight) + Current * Weight.
+
 %% -------------------------------------------------------------------------
 %% Begin copypasta from gen_server2.erl
 
diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl
index e8c96818..401b8ab1 100644
--- a/src/rabbit_net.erl
+++ b/src/rabbit_net.erl
@@ -222,10 +222,9 @@ maybe_ntoab(Addr) when is_tuple(Addr) -> rabbit_misc:ntoab(Addr);
 maybe_ntoab(Host)                     -> Host.
 
 rdns(Addr) ->
-    {ok, Lookup} = application:get_env(rabbit, reverse_dns_lookups),
-    case Lookup of
-        true -> list_to_binary(rabbit_networking:tcp_host(Addr));
-        _    -> Addr
+    case application:get_env(rabbit, reverse_dns_lookups) of
+        {ok, true} -> list_to_binary(rabbit_networking:tcp_host(Addr));
+        _          -> Addr
     end.
 
 sock_funs(inbound)  -> {fun peername/1, fun sockname/1};
diff --git a/src/rabbit_nodes.erl b/src/rabbit_nodes.erl
index b54fdd2e..5a1613a7 100644
--- a/src/rabbit_nodes.erl
+++ b/src/rabbit_nodes.erl
@@ -17,7 +17,9 @@
 -module(rabbit_nodes).
 
 -export([names/1, diagnostics/1, make/1, parts/1, cookie_hash/0,
-         is_running/2, is_process_running/2]).
+         is_running/2, is_process_running/2, fqdn_nodename/0]).
+
+-include_lib("kernel/include/inet.hrl").
 
 -define(EPMD_TIMEOUT, 30000).
 
@@ -35,6 +37,7 @@
 -spec(cookie_hash/0 :: () -> string()).
 -spec(is_running/2 :: (node(), atom()) -> boolean()).
 -spec(is_process_running/2 :: (node(), atom()) -> boolean()).
+-spec(fqdn_nodename/0 :: () -> binary()).
 
 -endif.
 
@@ -107,3 +110,9 @@ is_process_running(Node, Process) ->
         undefined        -> false;
         P when is_pid(P) -> true
     end.
+
+fqdn_nodename() ->
+    {ID, _} = rabbit_nodes:parts(node()),
+    {ok, Host} = inet:gethostname(),
+    {ok, #hostent{h_name = FQDN}} = inet:gethostbyname(Host),
+    list_to_binary(atom_to_list(rabbit_nodes:make({ID, FQDN}))).
diff --git a/src/rabbit_queue_collector.erl b/src/rabbit_queue_collector.erl
index 6406f7e9..855c7995 100644
--- a/src/rabbit_queue_collector.erl
+++ b/src/rabbit_queue_collector.erl
@@ -18,7 +18,7 @@
 
 -behaviour(gen_server).
 
--export([start_link/0, register/2, delete_all/1]).
+-export([start_link/1, register/2, delete_all/1]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
@@ -31,7 +31,8 @@
 
 -ifdef(use_specs).
 
--spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
+-spec(start_link/1 :: (rabbit_types:proc_name()) ->
+                           rabbit_types:ok_pid_or_error()).
 -spec(register/2 :: (pid(), pid()) -> 'ok').
 -spec(delete_all/1 :: (pid()) -> 'ok').
 
@@ -39,8 +40,8 @@
 
 %%----------------------------------------------------------------------------
 
-start_link() ->
-    gen_server:start_link(?MODULE, [], []).
+start_link(ProcName) ->
+    gen_server:start_link(?MODULE, [ProcName], []).
 
 register(CollectorPid, Q) ->
     gen_server:call(CollectorPid, {register, Q}, infinity).
@@ -50,7 +51,8 @@ delete_all(CollectorPid) ->
 
 %%----------------------------------------------------------------------------
 
-init([]) ->
+init([ProcName]) ->
+    ?store_proc_name(ProcName),
     {ok, #state{monitors = pmon:new(), delete_from = undefined}}.
 
 %%--------------------------------------------------------------------------
diff --git a/src/rabbit_queue_consumers.erl b/src/rabbit_queue_consumers.erl
new file mode 100644
index 00000000..c9540da8
--- /dev/null
+++ b/src/rabbit_queue_consumers.erl
@@ -0,0 +1,438 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2013 GoPivotal, Inc.  All rights reserved.
+%%
+
+-module(rabbit_queue_consumers).
+
+-export([new/0, max_active_priority/1, inactive/1, all/1, count/0,
+         unacknowledged_message_count/0, add/8, remove/3, erase_ch/2,
+         send_drained/0, deliver/3, record_ack/3, subtract_acks/2,
+         possibly_unblock/3,
+         resume_fun/0, notify_sent_fun/1, activate_limit_fun/0,
+         credit/6, utilisation/1]).
+
+%%----------------------------------------------------------------------------
+
+-define(UNSENT_MESSAGE_LIMIT,          200).
+
+%% Utilisation average calculations are all in μs.
+-define(USE_AVG_HALF_LIFE, 1000000.0).
+
+-record(state, {consumers, use}).
+
+-record(consumer, {tag, ack_required, args}).
+
+%% These are held in our process dictionary
+-record(cr, {ch_pid,
+             monitor_ref,
+             acktags,
+             consumer_count,
+             %% Queue of {ChPid, #consumer{}} for consumers which have
+             %% been blocked for any reason
+             blocked_consumers,
+             %% The limiter itself
+             limiter,
+             %% Internal flow control for queue -> writer
+             unsent_message_count}).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-type time_micros() :: non_neg_integer().
+-type ratio() :: float().
+-type state() :: #state{consumers ::priority_queue:q(),
+                        use       :: {'inactive',
+                                      time_micros(), time_micros(), ratio()} |
+                                     {'active', time_micros(), ratio()}}.
+-type ch() :: pid().
+-type ack() :: non_neg_integer().
+-type cr_fun() :: fun ((#cr{}) -> #cr{}).
+-type fetch_result() :: {rabbit_types:basic_message(), boolean(), ack()}.
+
+-spec new() -> state().
+-spec max_active_priority(state()) -> integer() | 'infinity' | 'empty'.
+-spec inactive(state()) -> boolean().
+-spec all(state()) -> [{ch(), rabbit_types:ctag(), boolean(),
+                        rabbit_framing:amqp_table()}].
+-spec count() -> non_neg_integer().
+-spec unacknowledged_message_count() -> non_neg_integer().
+-spec add(ch(), rabbit_types:ctag(), boolean(), pid(), boolean(),
+          rabbit_framing:amqp_table(), boolean(), state()) -> state().
+-spec remove(ch(), rabbit_types:ctag(), state()) ->
+                    'not_found' | state().
+-spec erase_ch(ch(), state()) ->
+                      'not_found' | {[ack()], [rabbit_types:ctag()],
+                                     state()}.
+-spec send_drained() -> 'ok'.
+-spec deliver(fun ((boolean()) -> {fetch_result(), T}),
+              rabbit_amqqueue:name(), state()) ->
+                     {'delivered',   boolean(), T, state()} |
+                     {'undelivered', boolean(), state()}.
+-spec record_ack(ch(), pid(), ack()) -> 'ok'.
+-spec subtract_acks(ch(), [ack()]) -> 'not_found' | 'ok'.
+-spec possibly_unblock(cr_fun(), ch(), state()) ->
+                              'unchanged' | {'unblocked', state()}.
+-spec resume_fun()                       -> cr_fun().
+-spec notify_sent_fun(non_neg_integer()) -> cr_fun().
+-spec activate_limit_fun()               -> cr_fun().
+-spec credit(boolean(), integer(), boolean(), ch(), rabbit_types:ctag(),
+             state()) -> 'unchanged' | {'unblocked', state()}.
+-spec utilisation(state()) -> ratio().
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+new() -> #state{consumers = priority_queue:new(),
+                use       = {inactive, now_micros(), 0, 0.0}}.
+
+max_active_priority(#state{consumers = Consumers}) ->
+    priority_queue:highest(Consumers).
+
+inactive(#state{consumers = Consumers}) ->
+    priority_queue:is_empty(Consumers).
+
+all(#state{consumers = Consumers}) ->
+    lists:foldl(fun (C, Acc) -> consumers(C#cr.blocked_consumers, Acc) end,
+                consumers(Consumers, []), all_ch_record()).
+
+consumers(Consumers, Acc) ->
+    priority_queue:fold(
+      fun ({ChPid, Consumer}, _P, Acc1) ->
+              #consumer{tag = CTag, ack_required = Ack, args = Args} = Consumer,
+              [{ChPid, CTag, Ack, Args} | Acc1]
+      end, Acc, Consumers).
+
+count() -> lists:sum([Count || #cr{consumer_count = Count} <- all_ch_record()]).
+
+unacknowledged_message_count() ->
+    lists:sum([queue:len(C#cr.acktags) || C <- all_ch_record()]).
+
+add(ChPid, CTag, NoAck, LimiterPid, LimiterActive, Args, IsEmpty,
+    State = #state{consumers = Consumers}) ->
+    C = #cr{consumer_count = Count,
+            limiter        = Limiter} = ch_record(ChPid, LimiterPid),
+    Limiter1 = case LimiterActive of
+                   true  -> rabbit_limiter:activate(Limiter);
+                   false -> Limiter
+               end,
+    C1 = C#cr{consumer_count = Count + 1, limiter = Limiter1},
+    update_ch_record(case parse_credit_args(Args) of
+                         none         -> C1;
+                         {Crd, Drain} -> credit_and_drain(
+                                           C1, CTag, Crd, Drain, IsEmpty)
+                     end),
+    Consumer = #consumer{tag          = CTag,
+                         ack_required = not NoAck,
+                         args         = Args},
+    State#state{consumers = add_consumer({ChPid, Consumer}, Consumers)}.
+
+remove(ChPid, CTag, State = #state{consumers = Consumers}) ->
+    case lookup_ch(ChPid) of
+        not_found ->
+            not_found;
+        C = #cr{consumer_count    = Count,
+                limiter           = Limiter,
+                blocked_consumers = Blocked} ->
+            Blocked1 = remove_consumer(ChPid, CTag, Blocked),
+            Limiter1 = case Count of
+                           1 -> rabbit_limiter:deactivate(Limiter);
+                           _ -> Limiter
+                       end,
+            Limiter2 = rabbit_limiter:forget_consumer(Limiter1, CTag),
+            update_ch_record(C#cr{consumer_count    = Count - 1,
+                                  limiter           = Limiter2,
+                                  blocked_consumers = Blocked1}),
+            State#state{consumers =
+                            remove_consumer(ChPid, CTag, Consumers)}
+    end.
+
+erase_ch(ChPid, State = #state{consumers = Consumers}) ->
+    case lookup_ch(ChPid) of
+        not_found ->
+            not_found;
+        C = #cr{ch_pid            = ChPid,
+                acktags           = ChAckTags,
+                blocked_consumers = BlockedQ} ->
+            AllConsumers = priority_queue:join(Consumers, BlockedQ),
+            ok = erase_ch_record(C),
+            {queue:to_list(ChAckTags),
+             tags(priority_queue:to_list(AllConsumers)),
+             State#state{consumers = remove_consumers(ChPid, Consumers)}}
+    end.
+
+send_drained() -> [update_ch_record(send_drained(C)) || C <- all_ch_record()],
+                  ok.
+
+deliver(FetchFun, QName, State) -> deliver(FetchFun, QName, false, State).
+
+deliver(FetchFun, QName, ConsumersChanged,
+        State = #state{consumers = Consumers}) ->
+    case priority_queue:out_p(Consumers) of
+        {empty, _} ->
+            {undelivered, ConsumersChanged,
+             State#state{use = update_use(State#state.use, inactive)}};
+        {{value, QEntry, Priority}, Tail} ->
+            case deliver_to_consumer(FetchFun, QEntry, QName) of
+                {delivered, R} ->
+                    {delivered, ConsumersChanged, R,
+                     State#state{consumers = priority_queue:in(QEntry, Priority,
+                                                               Tail)}};
+                undelivered ->
+                    deliver(FetchFun, QName, true,
+                            State#state{consumers = Tail})
+            end
+    end.
+
+deliver_to_consumer(FetchFun, E = {ChPid, Consumer}, QName) ->
+    C = lookup_ch(ChPid),
+    case is_ch_blocked(C) of
+        true  -> block_consumer(C, E),
+                 undelivered;
+        false -> case rabbit_limiter:can_send(C#cr.limiter,
+                                              Consumer#consumer.ack_required,
+                                              Consumer#consumer.tag) of
+                     {suspend, Limiter} ->
+                         block_consumer(C#cr{limiter = Limiter}, E),
+                         undelivered;
+                     {continue, Limiter} ->
+                         {delivered, deliver_to_consumer(
+                                       FetchFun, Consumer,
+                                       C#cr{limiter = Limiter}, QName)}
+                 end
+    end.
+
+deliver_to_consumer(FetchFun,
+                    #consumer{tag          = CTag,
+                              ack_required = AckRequired},
+                    C = #cr{ch_pid               = ChPid,
+                            acktags              = ChAckTags,
+                            unsent_message_count = Count},
+                    QName) ->
+    {{Message, IsDelivered, AckTag}, R} = FetchFun(AckRequired),
+    rabbit_channel:deliver(ChPid, CTag, AckRequired,
+                           {QName, self(), AckTag, IsDelivered, Message}),
+    ChAckTags1 = case AckRequired of
+                     true  -> queue:in(AckTag, ChAckTags);
+                     false -> ChAckTags
+                 end,
+    update_ch_record(C#cr{acktags              = ChAckTags1,
+                          unsent_message_count = Count + 1}),
+    R.
+
+record_ack(ChPid, LimiterPid, AckTag) ->
+    C = #cr{acktags = ChAckTags} = ch_record(ChPid, LimiterPid),
+    update_ch_record(C#cr{acktags = queue:in(AckTag, ChAckTags)}),
+    ok.
+
+subtract_acks(ChPid, AckTags) ->
+    case lookup_ch(ChPid) of
+        not_found ->
+            not_found;
+        C = #cr{acktags = ChAckTags} ->
+            update_ch_record(
+              C#cr{acktags = subtract_acks(AckTags, [], ChAckTags)}),
+            ok
+    end.
+
+subtract_acks([], [], AckQ) ->
+    AckQ;
+subtract_acks([], Prefix, AckQ) ->
+    queue:join(queue:from_list(lists:reverse(Prefix)), AckQ);
+subtract_acks([T | TL] = AckTags, Prefix, AckQ) ->
+    case queue:out(AckQ) of
+        {{value,  T}, QTail} -> subtract_acks(TL,             Prefix, QTail);
+        {{value, AT}, QTail} -> subtract_acks(AckTags, [AT | Prefix], QTail)
+    end.
+
+possibly_unblock(Update, ChPid, State) ->
+    case lookup_ch(ChPid) of
+        not_found -> unchanged;
+        C         -> C1 = Update(C),
+                     case is_ch_blocked(C) andalso not is_ch_blocked(C1) of
+                         false -> update_ch_record(C1),
+                                  unchanged;
+                         true  -> unblock(C1, State)
+                     end
+    end.
+
+unblock(C = #cr{blocked_consumers = BlockedQ, limiter = Limiter},
+        State = #state{consumers = Consumers, use = Use}) ->
+    case lists:partition(
+           fun({_P, {_ChPid, #consumer{tag = CTag}}}) ->
+                   rabbit_limiter:is_consumer_blocked(Limiter, CTag)
+           end, priority_queue:to_list(BlockedQ)) of
+        {_, []} ->
+            update_ch_record(C),
+            unchanged;
+        {Blocked, Unblocked} ->
+            BlockedQ1  = priority_queue:from_list(Blocked),
+            UnblockedQ = priority_queue:from_list(Unblocked),
+            update_ch_record(C#cr{blocked_consumers = BlockedQ1}),
+            {unblocked,
+             State#state{consumers = priority_queue:join(Consumers, UnblockedQ),
+                         use       = update_use(Use, active)}}
+    end.
+
+resume_fun() ->
+    fun (C = #cr{limiter = Limiter}) ->
+            C#cr{limiter = rabbit_limiter:resume(Limiter)}
+    end.
+
+notify_sent_fun(Credit) ->
+    fun (C = #cr{unsent_message_count = Count}) ->
+            C#cr{unsent_message_count = Count - Credit}
+    end.
+
+activate_limit_fun() ->
+    fun (C = #cr{limiter = Limiter}) ->
+            C#cr{limiter = rabbit_limiter:activate(Limiter)}
+    end.
+
+credit(IsEmpty, Credit, Drain, ChPid, CTag, State) ->
+    case lookup_ch(ChPid) of
+        not_found ->
+            unchanged;
+        #cr{limiter = Limiter} = C ->
+            C1 = #cr{limiter = Limiter1} =
+                credit_and_drain(C, CTag, Credit, Drain, IsEmpty),
+            case is_ch_blocked(C1) orelse
+                (not rabbit_limiter:is_consumer_blocked(Limiter, CTag)) orelse
+                rabbit_limiter:is_consumer_blocked(Limiter1, CTag) of
+                true  -> update_ch_record(C1),
+                         unchanged;
+                false -> unblock(C1, State)
+            end
+    end.
+
+utilisation(#state{use = {active, Since, Avg}}) ->
+    use_avg(now_micros() - Since, 0, Avg);
+utilisation(#state{use = {inactive, Since, Active, Avg}}) ->
+    use_avg(Active, now_micros() - Since, Avg).
+
+%%----------------------------------------------------------------------------
+
+parse_credit_args(Args) ->
+    case rabbit_misc:table_lookup(Args, <<"x-credit">>) of
+        {table, T} -> case {rabbit_misc:table_lookup(T, <<"credit">>),
+                            rabbit_misc:table_lookup(T, <<"drain">>)} of
+                          {{long, Credit}, {bool, Drain}} -> {Credit, Drain};
+                          _                               -> none
+                      end;
+        undefined  -> none
+    end.
+
+lookup_ch(ChPid) ->
+    case get({ch, ChPid}) of
+        undefined -> not_found;
+        C         -> C
+    end.
+
+ch_record(ChPid, LimiterPid) ->
+    Key = {ch, ChPid},
+    case get(Key) of
+        undefined -> MonitorRef = erlang:monitor(process, ChPid),
+                     Limiter = rabbit_limiter:client(LimiterPid),
+                     C = #cr{ch_pid               = ChPid,
+                             monitor_ref          = MonitorRef,
+                             acktags              = queue:new(),
+                             consumer_count       = 0,
+                             blocked_consumers    = priority_queue:new(),
+                             limiter              = Limiter,
+                             unsent_message_count = 0},
+                     put(Key, C),
+                     C;
+        C = #cr{} -> C
+    end.
+
+update_ch_record(C = #cr{consumer_count       = ConsumerCount,
+                         acktags              = ChAckTags,
+                         unsent_message_count = UnsentMessageCount}) ->
+    case {queue:is_empty(ChAckTags), ConsumerCount, UnsentMessageCount} of
+        {true, 0, 0} -> ok = erase_ch_record(C);
+        _            -> ok = store_ch_record(C)
+    end,
+    C.
+
+store_ch_record(C = #cr{ch_pid = ChPid}) ->
+    put({ch, ChPid}, C),
+    ok.
+
+erase_ch_record(#cr{ch_pid = ChPid, monitor_ref = MonitorRef}) ->
+    erlang:demonitor(MonitorRef),
+    erase({ch, ChPid}),
+    ok.
+
+all_ch_record() -> [C || {{ch, _}, C} <- get()].
+
+block_consumer(C = #cr{blocked_consumers = Blocked}, QEntry) ->
+    update_ch_record(C#cr{blocked_consumers = add_consumer(QEntry, Blocked)}).
+
+is_ch_blocked(#cr{unsent_message_count = Count, limiter = Limiter}) ->
+    Count >= ?UNSENT_MESSAGE_LIMIT orelse rabbit_limiter:is_suspended(Limiter).
+
+send_drained(C = #cr{ch_pid = ChPid, limiter = Limiter}) ->
+    case rabbit_limiter:drained(Limiter) of
+        {[],         Limiter}  -> C;
+        {CTagCredit, Limiter2} -> rabbit_channel:send_drained(
+                                    ChPid, CTagCredit),
+                                  C#cr{limiter = Limiter2}
+    end.
+
+credit_and_drain(C = #cr{ch_pid = ChPid, limiter = Limiter},
+                 CTag, Credit, Drain, IsEmpty) ->
+    case rabbit_limiter:credit(Limiter, CTag, Credit, Drain, IsEmpty) of
+        {true,  Limiter1} -> rabbit_channel:send_drained(ChPid,
+                                                         [{CTag, Credit}]),
+                             C#cr{limiter = Limiter1};
+        {false, Limiter1} -> C#cr{limiter = Limiter1}
+    end.
+
+tags(CList) -> [CTag || {_P, {_ChPid, #consumer{tag = CTag}}} <- CList].
+
+add_consumer({ChPid, Consumer = #consumer{args = Args}}, Queue) ->
+    Priority = case rabbit_misc:table_lookup(Args, <<"x-priority">>) of
+                   {_, P} -> P;
+                   _      -> 0
+               end,
+    priority_queue:in({ChPid, Consumer}, Priority, Queue).
+
+remove_consumer(ChPid, CTag, Queue) ->
+    priority_queue:filter(fun ({CP, #consumer{tag = CT}}) ->
+                                  (CP /= ChPid) or (CT /= CTag)
+                          end, Queue).
+
+remove_consumers(ChPid, Queue) ->
+    priority_queue:filter(fun ({CP, _Consumer}) when CP =:= ChPid -> false;
+                              (_)                                 -> true
+                          end, Queue).
+
+update_use({inactive, _, _, _}   = CUInfo, inactive) ->
+    CUInfo;
+update_use({active,   _, _}      = CUInfo,   active) ->
+    CUInfo;
+update_use({active,   Since,         Avg}, inactive) ->
+    Now = now_micros(),
+    {inactive, Now, Now - Since, Avg};
+update_use({inactive, Since, Active, Avg},   active) ->
+    Now = now_micros(),
+    {active, Now, use_avg(Active, Now - Since, Avg)}.
+
+use_avg(Active, Inactive, Avg) ->
+    Time = Inactive + Active,
+    rabbit_misc:moving_average(Time, ?USE_AVG_HALF_LIFE, Active / Time, Avg).
+
+now_micros() -> timer:now_diff(now(), {0,0,0}).
diff --git a/src/rabbit_queue_decorator.erl b/src/rabbit_queue_decorator.erl
index 8f6375a5..6205e2dc 100644
--- a/src/rabbit_queue_decorator.erl
+++ b/src/rabbit_queue_decorator.erl
@@ -8,13 +8,6 @@
 
 -ifdef(use_specs).
 
--type(notify_event() :: 'consumer_blocked'   |
-                        'consumer_unblocked' |
-                        'queue_empty'        |
-                        'basic_consume'      |
-                        'basic_cancel'       |
-                        'refresh').
-
 -callback startup(rabbit_types:amqqueue()) -> 'ok'.
 
 -callback shutdown(rabbit_types:amqqueue()) -> 'ok'.
@@ -24,7 +17,9 @@
 
 -callback active_for(rabbit_types:amqqueue()) -> boolean().
 
--callback notify(rabbit_types:amqqueue(), notify_event(), any()) -> 'ok'.
+%% called with Queue, MaxActivePriority, IsEmpty
+-callback consumer_state_changed(
+            rabbit_types:amqqueue(), integer(), boolean()) -> 'ok'.
 
 -else.
 
@@ -32,7 +27,7 @@
 
 behaviour_info(callbacks) ->
     [{description, 0}, {startup, 1}, {shutdown, 1}, {policy_changed, 2},
-     {active_for, 1}, {notify, 3}];
+     {active_for, 1}, {consumer_state_changed, 3}];
 behaviour_info(_Other) ->
     undefined.
 
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index f69d8355..919b7376 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -16,12 +16,10 @@
 
 -module(rabbit_queue_index).
 
--export([init/2, shutdown_terms/1, recover/5,
+-export([init/2, recover/5,
          terminate/2, delete_and_terminate/1,
          publish/5, deliver/2, ack/2, sync/1, needs_sync/1, flush/1,
-         read/3, next_segment_boundary/1, bounds/1, recover/1]).
-
--export([scan/3]).
+         read/3, next_segment_boundary/1, bounds/1, start/1, stop/0]).
 
 -export([add_queue_ttl/0, avoid_zeroes/0]).
 
@@ -196,10 +194,9 @@
 -type(contains_predicate() :: fun ((rabbit_types:msg_id()) -> boolean())).
 -type(walker(A) :: fun ((A) -> 'finished' |
                                {rabbit_types:msg_id(), non_neg_integer(), A})).
--type(shutdown_terms() :: [any()]).
+-type(shutdown_terms() :: [term()] | 'non_clean_shutdown').
 
 -spec(init/2 :: (rabbit_amqqueue:name(), on_sync_fun()) -> qistate()).
--spec(shutdown_terms/1 :: (rabbit_amqqueue:name()) -> shutdown_terms()).
 -spec(recover/5 :: (rabbit_amqqueue:name(), shutdown_terms(), boolean(),
                     contains_predicate(), on_sync_fun()) ->
                         {'undefined' | non_neg_integer(), qistate()}).
@@ -220,13 +217,7 @@
 -spec(next_segment_boundary/1 :: (seq_id()) -> seq_id()).
 -spec(bounds/1 :: (qistate()) ->
                        {non_neg_integer(), non_neg_integer(), qistate()}).
--spec(recover/1 :: ([rabbit_amqqueue:name()]) -> {[[any()]], {walker(A), A}}).
-
--spec(scan/3 :: (file:filename(),
-                 fun ((seq_id(), rabbit_types:msg_id(),
-                       rabbit_types:message_properties(), boolean(),
-                       ('del' | 'no_del'), ('ack' | 'no_ack'), A) -> A),
-                     A) -> A).
+-spec(start/1 :: ([rabbit_amqqueue:name()]) -> {[[any()]], {walker(A), A}}).
 
 -spec(add_queue_ttl/0 :: () -> 'ok').
 
@@ -242,26 +233,20 @@ init(Name, OnSyncFun) ->
     false = rabbit_file:is_file(Dir), %% is_file == is file or dir
     State #qistate { on_sync = OnSyncFun }.
 
-shutdown_terms(Name) ->
-    #qistate { dir = Dir } = blank_state(Name),
-    case read_shutdown_terms(Dir) of
-        {error, _}   -> [];
-        {ok, Terms1} -> Terms1
-    end.
-
 recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun, OnSyncFun) ->
-    State = #qistate { dir = Dir } = blank_state(Name),
+    State = blank_state(Name),
     State1 = State #qistate { on_sync = OnSyncFun },
-    CleanShutdown = detect_clean_shutdown(Dir),
+    CleanShutdown = Terms /= non_clean_shutdown,
     case CleanShutdown andalso MsgStoreRecovered of
         true  -> RecoveredCounts = proplists:get_value(segments, Terms, []),
                  init_clean(RecoveredCounts, State1);
         false -> init_dirty(CleanShutdown, ContainsCheckFun, State1)
     end.
 
-terminate(Terms, State) ->
-    {SegmentCounts, State1 = #qistate { dir = Dir }} = terminate(State),
-    store_clean_shutdown([{segments, SegmentCounts} | Terms], Dir),
+terminate(Terms, State = #qistate { dir = Dir }) ->
+    {SegmentCounts, State1} = terminate(State),
+    rabbit_recovery_terms:store(filename:basename(Dir),
+                                [{segments, SegmentCounts} | Terms]),
     State1.
 
 delete_and_terminate(State) ->
@@ -357,37 +342,40 @@ bounds(State = #qistate { segments = Segments }) ->
         end,
     {LowSeqId, NextSeqId, State}.
 
-recover(DurableQueues) ->
-    DurableDict = dict:from_list([ {queue_name_to_dir_name(Queue), Queue} ||
-                                     Queue <- DurableQueues ]),
-    QueuesDir = queues_dir(),
-    QueueDirNames = all_queue_directory_names(QueuesDir),
-    DurableDirectories = sets:from_list(dict:fetch_keys(DurableDict)),
-    {DurableQueueNames, DurableTerms} =
+start(DurableQueueNames) ->
+    ok = rabbit_recovery_terms:start(),
+    {DurableTerms, DurableDirectories} =
         lists:foldl(
-          fun (QueueDirName, {DurableAcc, TermsAcc}) ->
-                  QueueDirPath = filename:join(QueuesDir, QueueDirName),
-                  case sets:is_element(QueueDirName, DurableDirectories) of
-                      true ->
-                          TermsAcc1 =
-                              case read_shutdown_terms(QueueDirPath) of
-                                  {error, _}  -> TermsAcc;
-                                  {ok, Terms} -> [Terms | TermsAcc]
-                              end,
-                          {[dict:fetch(QueueDirName, DurableDict) | DurableAcc],
-                           TermsAcc1};
-                      false ->
-                          ok = rabbit_file:recursive_delete([QueueDirPath]),
-                          {DurableAcc, TermsAcc}
-                  end
-          end, {[], []}, QueueDirNames),
-    {DurableTerms, {fun queue_index_walker/1, {start, DurableQueueNames}}}.
+          fun(QName, {RecoveryTerms, ValidDirectories}) ->
+                  DirName = queue_name_to_dir_name(QName),
+                  RecoveryInfo = case rabbit_recovery_terms:read(DirName) of
+                                     {error, _}  -> non_clean_shutdown;
+                                     {ok, Terms} -> Terms
+                                 end,
+                  {[RecoveryInfo | RecoveryTerms],
+                   sets:add_element(DirName, ValidDirectories)}
+          end, {[], sets:new()}, DurableQueueNames),
+
+    %% Any queue directory we've not been asked to recover is considered garbage
+    QueuesDir = queues_dir(),
+    rabbit_file:recursive_delete(
+      [filename:join(QueuesDir, DirName) ||
+          DirName <- all_queue_directory_names(QueuesDir),
+          not sets:is_element(DirName, DurableDirectories)]),
+
+    rabbit_recovery_terms:clear(),
+
+    %% The backing queue interface requires that the queue recovery terms
+    %% which come back from start/1 are in the same order as DurableQueueNames
+    OrderedTerms = lists:reverse(DurableTerms),
+    {OrderedTerms, {fun queue_index_walker/1, {start, DurableQueueNames}}}.
+
+stop() -> rabbit_recovery_terms:stop().
 
 all_queue_directory_names(Dir) ->
     case rabbit_file:list_dir(Dir) of
-        {ok, Entries}   -> [ Entry || Entry <- Entries,
-                                      rabbit_file:is_dir(
-                                        filename:join(Dir, Entry)) ];
+        {ok, Entries}   -> [E || E <- Entries,
+                                 rabbit_file:is_dir(filename:join(Dir, E))];
         {error, enoent} -> []
     end.
 
@@ -410,22 +398,6 @@ blank_state_dir(Dir) ->
                on_sync             = fun (_) -> ok end,
                unconfirmed         = gb_sets:new() }.
 
-clean_filename(Dir) -> filename:join(Dir, ?CLEAN_FILENAME).
-
-detect_clean_shutdown(Dir) ->
-    case rabbit_file:delete(clean_filename(Dir)) of
-        ok              -> true;
-        {error, enoent} -> false
-    end.
-
-read_shutdown_terms(Dir) ->
-    rabbit_file:read_term_file(clean_filename(Dir)).
-
-store_clean_shutdown(Terms, Dir) ->
-    CleanFileName = clean_filename(Dir),
-    ok = rabbit_file:ensure_dir(CleanFileName),
-    rabbit_file:write_term_file(CleanFileName, Terms).
-
 init_clean(RecoveredCounts, State) ->
     %% Load the journal. Since this is a clean recovery this (almost)
     %% gets us back to where we were on shutdown.
@@ -554,9 +526,6 @@ queue_index_walker_reader(QueueName, Gatherer) ->
            end, ok, State),
     ok = gatherer:finish(Gatherer).
 
-scan(Dir, Fun, Acc) ->
-    scan_segments(Fun, Acc, blank_state_dir(Dir)).
-
 scan_segments(Fun, Acc, State) ->
     State1 = #qistate { segments = Segments, dir = Dir } =
         recover_journal(State),
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 0aa44d9f..122eb305 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -23,7 +23,7 @@
 
 -export([system_continue/3, system_terminate/4, system_code_change/4]).
 
--export([init/2, mainloop/2, recvloop/2]).
+-export([init/2, mainloop/4, recvloop/4]).
 
 -export([conserve_resources/3, server_properties/1]).
 
@@ -38,8 +38,7 @@
 
 -record(v1, {parent, sock, connection, callback, recv_len, pending_recv,
              connection_state, helper_sup, queue_collector, heartbeater,
-             stats_timer, channel_sup_sup_pid, buf, buf_len, channel_count,
-             throttle}).
+             stats_timer, channel_sup_sup_pid, channel_count, throttle}).
 
 -record(connection, {name, host, peer_host, port, peer_port,
                      protocol, user, timeout_sec, frame_max, channel_max, vhost,
@@ -92,9 +91,10 @@
                      rabbit_types:ok_or_error2(
                        rabbit_net:socket(), any()))) -> no_return()).
 
--spec(mainloop/2 :: (_,#v1{}) -> any()).
+-spec(mainloop/4 :: (_,[binary()], non_neg_integer(), #v1{}) -> any()).
 -spec(system_code_change/4 :: (_,_,_,_) -> {'ok',_}).
--spec(system_continue/3 :: (_,_,#v1{}) -> any()).
+-spec(system_continue/3 :: (_,_,{[binary()], non_neg_integer(), #v1{}}) ->
+                                any()).
 -spec(system_terminate/4 :: (_,_,_,_) -> none()).
 
 -endif.
@@ -114,8 +114,8 @@ init(Parent, HelperSup) ->
             start_connection(Parent, HelperSup, Deb, Sock, SockTransform)
     end.
 
-system_continue(Parent, Deb, State) ->
-    mainloop(Deb, State#v1{parent = Parent}).
+system_continue(Parent, Deb, {Buf, BufLen, State}) ->
+    mainloop(Deb, Buf, BufLen, State#v1{parent = Parent}).
 
 system_terminate(Reason, _Parent, _Deb, _State) ->
     exit(Reason).
@@ -214,6 +214,7 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) ->
     erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(), handshake_timeout),
     {PeerHost, PeerPort, Host, Port} =
         socket_op(Sock, fun (S) -> rabbit_net:socket_ends(S, inbound) end),
+    ?store_proc_name(list_to_binary(Name)),
     State = #v1{parent              = Parent,
                 sock                = ClientSock,
                 connection          = #connection{
@@ -239,8 +240,6 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) ->
                 helper_sup          = HelperSup,
                 heartbeater         = none,
                 channel_sup_sup_pid = none,
-                buf                 = [],
-                buf_len             = 0,
                 channel_count       = 0,
                 throttle            = #throttle{
                                          alarmed_by      = [],
@@ -249,9 +248,9 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) ->
                                          blocked_sent    = false}},
     try
         run({?MODULE, recvloop,
-             [Deb, switch_callback(rabbit_event:init_stats_timer(
-                                     State, #v1.stats_timer),
-                                   handshake, 8)]}),
+             [Deb, [], 0, switch_callback(rabbit_event:init_stats_timer(
+                                            State, #v1.stats_timer),
+                                          handshake, 8)]}),
         log(info, "closing AMQP connection ~p (~s)~n", [self(), Name])
     catch
         Ex -> log(case Ex of
@@ -278,47 +277,62 @@ run({M, F, A}) ->
     catch {become, MFA} -> run(MFA)
     end.
 
-recvloop(Deb, State = #v1{pending_recv = true}) ->
-    mainloop(Deb, State);
-recvloop(Deb, State = #v1{connection_state = blocked}) ->
-    mainloop(Deb, State);
-recvloop(Deb, State = #v1{sock = Sock, recv_len = RecvLen, buf_len = BufLen})
+recvloop(Deb, Buf, BufLen, State = #v1{pending_recv = true}) ->
+    mainloop(Deb, Buf, BufLen, State);
+recvloop(Deb, Buf, BufLen, State = #v1{connection_state = blocked}) ->
+    mainloop(Deb, Buf, BufLen, State);
+recvloop(Deb, Buf, BufLen, State = #v1{connection_state = {become, F}}) ->
+    throw({become, F(Deb, Buf, BufLen, State)});
+recvloop(Deb, Buf, BufLen, State = #v1{sock = Sock, recv_len = RecvLen})
   when BufLen < RecvLen ->
-    ok = rabbit_net:setopts(Sock, [{active, once}]),
-    mainloop(Deb, State#v1{pending_recv = true});
-recvloop(Deb, State = #v1{recv_len = RecvLen, buf = Buf, buf_len = BufLen}) ->
-    {Data, Rest} = split_binary(case Buf of
-                                    [B] -> B;
-                                    _   -> list_to_binary(lists:reverse(Buf))
-                                end, RecvLen),
-    recvloop(Deb, handle_input(State#v1.callback, Data,
-                               State#v1{buf = [Rest],
-                                        buf_len = BufLen - RecvLen})).
-
-mainloop(Deb, State = #v1{sock = Sock, buf = Buf, buf_len = BufLen}) ->
+    case rabbit_net:setopts(Sock, [{active, once}]) of
+        ok              -> mainloop(Deb, Buf, BufLen,
+                                    State#v1{pending_recv = true});
+        {error, Reason} -> stop(Reason, State)
+    end;
+recvloop(Deb, [B], _BufLen, State) ->
+    {Rest, State1} = handle_input(State#v1.callback, B, State),
+    recvloop(Deb, [Rest], size(Rest), State1);
+recvloop(Deb, Buf, BufLen, State = #v1{recv_len = RecvLen}) ->
+    {DataLRev, RestLRev} = binlist_split(BufLen - RecvLen, Buf, []),
+    Data = list_to_binary(lists:reverse(DataLRev)),
+    {<<>>, State1} = handle_input(State#v1.callback, Data, State),
+    recvloop(Deb, lists:reverse(RestLRev), BufLen - RecvLen, State1).
+
+binlist_split(0, L, Acc) ->
+    {L, Acc};
+binlist_split(Len, L, [Acc0|Acc]) when Len < 0 ->
+    {H, T} = split_binary(Acc0, -Len),
+    {[H|L], [T|Acc]};
+binlist_split(Len, [H|T], Acc) ->
+    binlist_split(Len - size(H), T, [H|Acc]).
+
+mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock}) ->
     case rabbit_net:recv(Sock) of
         {data, Data} ->
-            recvloop(Deb, State#v1{buf = [Data | Buf],
-                                   buf_len = BufLen + size(Data),
-                                   pending_recv = false});
+            recvloop(Deb, [Data | Buf], BufLen + size(Data),
+                     State#v1{pending_recv = false});
         closed when State#v1.connection_state =:= closed ->
             ok;
         closed ->
-            maybe_emit_stats(State),
-            throw(connection_closed_abruptly);
+            stop(closed, State);
         {error, Reason} ->
-            maybe_emit_stats(State),
-            throw({inet_error, Reason});
+            stop(Reason, State);
         {other, {system, From, Request}} ->
             sys:handle_system_msg(Request, From, State#v1.parent,
-                                  ?MODULE, Deb, State);
+                                  ?MODULE, Deb, {Buf, BufLen, State});
         {other, Other}  ->
             case handle_other(Other, State) of
                 stop     -> ok;
-                NewState -> recvloop(Deb, NewState)
+                NewState -> recvloop(Deb, Buf, BufLen, NewState)
             end
     end.
 
+stop(closed, State) -> maybe_emit_stats(State),
+                       throw(connection_closed_abruptly);
+stop(Reason, State) -> maybe_emit_stats(State),
+                       throw({inet_error, Reason}).
+
 handle_other({conserve_resources, Source, Conserve},
              State = #v1{throttle = Throttle =
                              #throttle{alarmed_by = CR}}) ->
@@ -715,32 +729,36 @@ post_process_frame(_Frame, _ChPid, State) ->
 %% a few get it wrong - off-by 1 or 8 (empty frame size) are typical.
 -define(FRAME_SIZE_FUDGE, ?EMPTY_FRAME_SIZE).
 
-handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32>>,
+handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32, _/binary>>,
              State = #v1{connection = #connection{frame_max = FrameMax}})
   when FrameMax /= 0 andalso
        PayloadSize > FrameMax - ?EMPTY_FRAME_SIZE + ?FRAME_SIZE_FUDGE ->
     fatal_frame_error(
       {frame_too_large, PayloadSize, FrameMax - ?EMPTY_FRAME_SIZE},
       Type, Channel, <<>>, State);
-handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32>>, State) ->
-    ensure_stats_timer(
-      switch_callback(State, {frame_payload, Type, Channel, PayloadSize},
-                      PayloadSize + 1));
-
+handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32,
+                             Payload:PayloadSize/binary, ?FRAME_END,
+                             Rest/binary>>,
+             State) ->
+    {Rest, ensure_stats_timer(handle_frame(Type, Channel, Payload, State))};
+handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32, Rest/binary>>,
+             State) ->
+    {Rest, ensure_stats_timer(
+             switch_callback(State,
+                             {frame_payload, Type, Channel, PayloadSize},
+                             PayloadSize + 1))};
 handle_input({frame_payload, Type, Channel, PayloadSize}, Data, State) ->
-    <<Payload:PayloadSize/binary, EndMarker>> = Data,
+    <<Payload:PayloadSize/binary, EndMarker, Rest/binary>> = Data,
     case EndMarker of
         ?FRAME_END -> State1 = handle_frame(Type, Channel, Payload, State),
-                      switch_callback(State1, frame_header, 7);
+                      {Rest, switch_callback(State1, frame_header, 7)};
         _          -> fatal_frame_error({invalid_frame_end_marker, EndMarker},
                                         Type, Channel, Payload, State)
     end;
-
-handle_input(handshake, <<"AMQP", A, B, C, D>>, State) ->
-    handshake({A, B, C, D}, State);
-handle_input(handshake, Other, #v1{sock = Sock}) ->
+handle_input(handshake, <<"AMQP", A, B, C, D, Rest/binary>>, State) ->
+    {Rest, handshake({A, B, C, D}, State)};
+handle_input(handshake, <<Other:8/binary, _/binary>>, #v1{sock = Sock}) ->
     refuse_connection(Sock, {bad_header, Other});
-
 handle_input(Callback, Data, _State) ->
     throw({bad_input, Callback, Data}).
 
@@ -820,7 +838,7 @@ handle_method0(MethodName, FieldsBin,
     try
         handle_method0(Protocol:decode_method_fields(MethodName, FieldsBin),
                        State)
-    catch throw:{writer_inet_error, closed} ->
+    catch throw:{inet_error, closed} ->
             maybe_emit_stats(State),
             throw(connection_closed_abruptly);
           exit:#amqp_error{method = none} = Reason ->
@@ -866,15 +884,15 @@ handle_method0(#'connection.tune_ok'{frame_max   = FrameMax,
            frame_max,   ?FRAME_MIN_SIZE, FrameMax),
     ok = validate_negotiated_integer_value(
            channel_max, ?CHANNEL_MIN,    ChannelMax),
-    {ok, Collector} =
-        rabbit_connection_helper_sup:start_queue_collector(SupPid),
+    {ok, Collector} = rabbit_connection_helper_sup:start_queue_collector(
+                        SupPid, Connection#connection.name),
     Frame = rabbit_binary_generator:build_heartbeat_frame(),
     SendFun = fun() -> catch rabbit_net:send(Sock, Frame) end,
     Parent = self(),
     ReceiveFun = fun() -> Parent ! heartbeat_timeout end,
-    Heartbeater =
-        rabbit_heartbeat:start(SupPid, Sock, ClientHeartbeat,
-                               SendFun, ClientHeartbeat, ReceiveFun),
+    Heartbeater = rabbit_heartbeat:start(
+                    SupPid, Sock, Connection#connection.name,
+                    ClientHeartbeat, SendFun, ClientHeartbeat, ReceiveFun),
     State#v1{connection_state = opening,
              connection = Connection#connection{
                             frame_max   = FrameMax,
@@ -939,6 +957,9 @@ validate_negotiated_integer_value(Field, Min, ClientValue) ->
             ok
     end.
 
+%% keep dialyzer happy
+-spec fail_negotiation(atom(), 'min' | 'max', integer(), integer()) ->
+                              no_return().
 fail_negotiation(Field, MinOrMax, ServerValue, ClientValue) ->
     {S1, S2} = case MinOrMax of
                    min -> {lower,  minimum};
@@ -1128,15 +1149,16 @@ become_1_0(Id, State = #v1{sock = Sock}) ->
                                    Sock, {unsupported_amqp1_0_protocol_id, Id},
                                    {3, 1, 0, 0})
                         end,
-                 throw({become, {rabbit_amqp1_0_reader, init,
-                                 [Mode, pack_for_1_0(State)]}})
+                 F = fun (_Deb, Buf, BufLen, S) ->
+                             {rabbit_amqp1_0_reader, init,
+                              [Mode, pack_for_1_0(Buf, BufLen, S)]}
+                     end,
+                 State = #v1{connection_state = {become, F}}
     end.
 
-pack_for_1_0(#v1{parent              = Parent,
-                 sock                = Sock,
-                 recv_len            = RecvLen,
-                 pending_recv        = PendingRecv,
-                 helper_sup          = SupPid,
-                 buf                 = Buf,
-                 buf_len             = BufLen}) ->
+pack_for_1_0(Buf, BufLen, #v1{parent       = Parent,
+                              sock         = Sock,
+                              recv_len     = RecvLen,
+                              pending_recv = PendingRecv,
+                              helper_sup   = SupPid}) ->
     {Parent, Sock, RecvLen, PendingRecv, SupPid, Buf, BufLen}.
diff --git a/src/rabbit_recovery_terms.erl b/src/rabbit_recovery_terms.erl
new file mode 100644
index 00000000..efb94b81
--- /dev/null
+++ b/src/rabbit_recovery_terms.erl
@@ -0,0 +1,121 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2013 GoPivotal, Inc.  All rights reserved.
+%%
+
+%% We use a gen_server simply so that during the terminate/2 call
+%% (i.e., during shutdown), we can sync/flush the dets table to disk.
+
+-module(rabbit_recovery_terms).
+
+-behaviour(gen_server).
+
+-export([start/0, stop/0, store/2, read/1, clear/0]).
+
+-export([upgrade_recovery_terms/0, start_link/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-rabbit_upgrade({upgrade_recovery_terms, local, []}).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(start() -> rabbit_types:ok_or_error(term())).
+-spec(stop() -> rabbit_types:ok_or_error(term())).
+-spec(store(file:filename(), term()) -> rabbit_types:ok_or_error(term())).
+-spec(read(file:filename()) -> rabbit_types:ok_or_error2(term(), not_found)).
+-spec(clear() -> 'ok').
+
+-endif. % use_specs
+
+%%----------------------------------------------------------------------------
+
+-define(SERVER, ?MODULE).
+
+start() -> rabbit_sup:start_child(?MODULE).
+
+stop() -> rabbit_sup:stop_child(?MODULE).
+
+store(DirBaseName, Terms) -> dets:insert(?MODULE, {DirBaseName, Terms}).
+
+read(DirBaseName) ->
+    case dets:lookup(?MODULE, DirBaseName) of
+        [{_, Terms}] -> {ok, Terms};
+        _            -> {error, not_found}
+    end.
+
+clear() ->
+    dets:delete_all_objects(?MODULE),
+    flush().
+
+%%----------------------------------------------------------------------------
+
+upgrade_recovery_terms() ->
+    open_table(),
+    try
+        QueuesDir = filename:join(rabbit_mnesia:dir(), "queues"),
+        Dirs = case rabbit_file:list_dir(QueuesDir) of
+                   {ok, Entries} -> Entries;
+                   {error, _}    -> []
+               end,
+        [begin
+             File = filename:join([QueuesDir, Dir, "clean.dot"]),
+             case rabbit_file:read_term_file(File) of
+                 {ok, Terms} -> ok  = store(Dir, Terms);
+                 {error, _}  -> ok
+             end,
+             file:delete(File)
+         end || Dir <- Dirs],
+        ok
+    after
+        close_table()
+    end.
+
+start_link() -> gen_server:start_link(?MODULE, [], []).
+
+%%----------------------------------------------------------------------------
+
+init(_) ->
+    process_flag(trap_exit, true),
+    open_table(),
+    {ok, undefined}.
+
+handle_call(Msg, _, State) -> {stop, {unexpected_call, Msg}, State}.
+
+handle_cast(Msg, State) -> {stop, {unexpected_cast, Msg}, State}.
+
+handle_info(_Info, State) -> {noreply, State}.
+
+terminate(_Reason, _State) ->
+    close_table().
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%----------------------------------------------------------------------------
+
+open_table() ->
+    File = filename:join(rabbit_mnesia:dir(), "recovery.dets"),
+    {ok, _} = dets:open_file(?MODULE, [{file,      File},
+                                       {ram_file,  true},
+                                       {auto_save, infinity}]).
+
+flush() -> dets:sync(?MODULE).
+
+close_table() ->
+    ok = flush(),
+    ok = dets:close(?MODULE).
+
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 5fe319d3..2d6ff73b 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -1172,7 +1172,7 @@ test_server_status() ->
                                rabbit_misc:r(<<"/">>, queue, Name),
                                false, false, [], none)]],
     ok = rabbit_amqqueue:basic_consume(
-           Q, true, Ch, Limiter, false, <<"ctag">>, true, none, [], undefined),
+           Q, true, Ch, Limiter, false, <<"ctag">>, true, [], undefined),
 
     %% list queues
     ok = info_action(list_queues, rabbit_amqqueue:info_keys(), true),
@@ -1262,7 +1262,7 @@ test_writer(Pid) ->
 test_channel() ->
     Me = self(),
     Writer = spawn(fun () -> test_writer(Me) end),
-    {ok, Limiter} = rabbit_limiter:start_link(),
+    {ok, Limiter} = rabbit_limiter:start_link(no_id),
     {ok, Ch} = rabbit_channel:start_link(
                  1, Me, Writer, Me, "", rabbit_framing_amqp_0_9_1,
                  user(<<"guest">>), <<"/">>, [], Me, Limiter),
@@ -2129,11 +2129,10 @@ test_queue() ->
 
 init_test_queue() ->
     TestQueue = test_queue(),
-    Terms = rabbit_queue_index:shutdown_terms(TestQueue),
-    PRef = proplists:get_value(persistent_ref, Terms, rabbit_guid:gen()),
+    PRef = rabbit_guid:gen(),
     PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE, PRef),
     Res = rabbit_queue_index:recover(
-            TestQueue, Terms, false,
+            TestQueue, [], false,
             fun (MsgId) ->
                     rabbit_msg_store:contains(MsgId, PersistentClient)
             end,
@@ -2144,12 +2143,12 @@ init_test_queue() ->
 restart_test_queue(Qi) ->
     _ = rabbit_queue_index:terminate([], Qi),
     ok = rabbit_variable_queue:stop(),
-    ok = rabbit_variable_queue:start([test_queue()]),
+    {ok, _} = rabbit_variable_queue:start([test_queue()]),
     init_test_queue().
 
 empty_test_queue() ->
     ok = rabbit_variable_queue:stop(),
-    ok = rabbit_variable_queue:start([]),
+    {ok, _} = rabbit_variable_queue:start([]),
     {0, Qi} = init_test_queue(),
     _ = rabbit_queue_index:delete_and_terminate(Qi),
     ok.
@@ -2205,7 +2204,7 @@ test_queue_index_props() ->
       end),
 
     ok = rabbit_variable_queue:stop(),
-    ok = rabbit_variable_queue:start([]),
+    {ok, _} = rabbit_variable_queue:start([]),
 
     passed.
 
@@ -2329,13 +2328,16 @@ test_queue_index() ->
       end),
 
     ok = rabbit_variable_queue:stop(),
-    ok = rabbit_variable_queue:start([]),
+    {ok, _} = rabbit_variable_queue:start([]),
 
     passed.
 
 variable_queue_init(Q, Recover) ->
     rabbit_variable_queue:init(
-      Q, Recover, fun nop/2, fun nop/2, fun nop/1).
+      Q, case Recover of
+             true  -> non_clean_shutdown;
+             false -> new
+         end, fun nop/2, fun nop/2, fun nop/1).
 
 variable_queue_publish(IsPersistent, Count, VQ) ->
     variable_queue_publish(IsPersistent, Count, fun (_N, P) -> P end, VQ).
@@ -2408,8 +2410,8 @@ publish_and_confirm(Q, Payload, Count) ->
                                     <<>>, #'P_basic'{delivery_mode = 2},
                                     Payload),
          Delivery = #delivery{mandatory = false, sender = self(),
-                              message = Msg, msg_seq_no = Seq},
-         {routed, _} = rabbit_amqqueue:deliver([Q], Delivery)
+                              confirm = true, message = Msg, msg_seq_no = Seq},
+          _QPids = rabbit_amqqueue:deliver([Q], Delivery)
      end || Seq <- Seqs],
     wait_for_confirms(gb_sets:from_list(Seqs)).
 
@@ -2815,7 +2817,7 @@ test_queue_recover() ->
     end,
     rabbit_amqqueue:stop(),
     rabbit_amqqueue:start(rabbit_amqqueue:recover()),
-    {ok, Limiter} = rabbit_limiter:start_link(),
+    {ok, Limiter} = rabbit_limiter:start_link(no_id),
     rabbit_amqqueue:with_or_die(
       QName,
       fun (Q1 = #amqqueue { pid = QPid1 }) ->
@@ -2842,7 +2844,7 @@ test_variable_queue_delete_msg_store_files_callback() ->
 
     rabbit_amqqueue:set_ram_duration_target(QPid, 0),
 
-    {ok, Limiter} = rabbit_limiter:start_link(),
+    {ok, Limiter} = rabbit_limiter:start_link(no_id),
 
     CountMinusOne = Count - 1,
     {ok, CountMinusOne, {QName, QPid, _AckTag, false, _Msg}} =
diff --git a/src/rabbit_trace.erl b/src/rabbit_trace.erl
index d0dcaa71..b08a9a1c 100644
--- a/src/rabbit_trace.erl
+++ b/src/rabbit_trace.erl
@@ -88,9 +88,9 @@ trace(#exchange{name = Name}, #basic_message{exchange_name = Name},
     ok;
 trace(X, Msg = #basic_message{content = #content{payload_fragments_rev = PFR}},
       RKPrefix, RKSuffix, Extra) ->
-    {ok, _, _} = rabbit_basic:publish(
-                   X, <<RKPrefix/binary, ".", RKSuffix/binary>>,
-                   #'P_basic'{headers = msg_to_table(Msg) ++ Extra}, PFR),
+    {ok, _} = rabbit_basic:publish(
+                X, <<RKPrefix/binary, ".", RKSuffix/binary>>,
+                #'P_basic'{headers = msg_to_table(Msg) ++ Extra}, PFR),
     ok.
 
 msg_to_table(#basic_message{exchange_name = #resource{name = XName},
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index a36613db..0edebff1 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -30,7 +30,8 @@
               connection/0, protocol/0, user/0, internal_user/0,
               username/0, password/0, password_hash/0,
               ok/1, error/1, ok_or_error/1, ok_or_error2/2, ok_pid_or_error/0,
-              channel_exit/0, connection_exit/0, mfargs/0]).
+              channel_exit/0, connection_exit/0, mfargs/0, proc_name/0,
+              proc_type_and_name/0]).
 
 -type(maybe(T) :: T | 'none').
 -type(vhost() :: binary()).
@@ -156,4 +157,7 @@
 
 -type(mfargs() :: {atom(), atom(), [any()]}).
 
+-type(proc_name() :: term()).
+-type(proc_type_and_name() :: {atom(), proc_name()}).
+
 -endif. % use_specs
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index ac2b9f52..020b5b33 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -21,8 +21,8 @@
          dropwhile/2, fetchwhile/4,
          fetch/2, drop/2, ack/2, requeue/2, ackfold/4, fold/3, len/1,
          is_empty/1, depth/1, set_ram_duration_target/2, ram_duration/1,
-         needs_timeout/1, timeout/1, handle_pre_hibernate/1, status/1, invoke/3,
-         is_duplicate/2, multiple_routing_keys/0]).
+         needs_timeout/1, timeout/1, handle_pre_hibernate/1, msg_rates/1,
+         status/1, invoke/3, is_duplicate/2, multiple_routing_keys/0]).
 
 -export([start/1, stop/0]).
 
@@ -277,11 +277,10 @@
           unconfirmed,
           confirmed,
           ack_out_counter,
-          ack_in_counter,
-          ack_rates
+          ack_in_counter
         }).
 
--record(rates, { egress, ingress, avg_egress, avg_ingress, timestamp }).
+-record(rates, { in, out, ack_in, ack_out, timestamp }).
 
 -record(msg_status,
         { seq_id,
@@ -322,11 +321,11 @@
 -type(timestamp() :: {non_neg_integer(), non_neg_integer(), non_neg_integer()}).
 -type(seq_id()  :: non_neg_integer()).
 
--type(rates() :: #rates { egress      :: {timestamp(), non_neg_integer()},
-                          ingress     :: {timestamp(), non_neg_integer()},
-                          avg_egress  :: float(),
-                          avg_ingress :: float(),
-                          timestamp   :: timestamp() }).
+-type(rates() :: #rates { in        :: float(),
+                          out       :: float(),
+                          ack_in    :: float(),
+                          ack_out   :: float(),
+                          timestamp :: timestamp()}).
 
 -type(delta() :: #delta { start_seq_id :: non_neg_integer(),
                           count        :: non_neg_integer(),
@@ -368,8 +367,7 @@
              unconfirmed           :: gb_set(),
              confirmed             :: gb_set(),
              ack_out_counter       :: non_neg_integer(),
-             ack_in_counter        :: non_neg_integer(),
-             ack_rates             :: rates() }).
+             ack_in_counter        :: non_neg_integer() }).
 %% Duplicated from rabbit_backing_queue
 -spec(ack/2 :: ([ack()], state()) -> {[rabbit_guid:guid()], state()}).
 
@@ -384,21 +382,37 @@
                                          count        = 0,
                                          end_seq_id   = Z }).
 
+-define(MICROS_PER_SECOND, 1000000.0).
+
+%% We're sampling every 5s for RAM duration; a half life that is of
+%% the same order of magnitude is probably about right.
+-define(RATE_AVG_HALF_LIFE, 5.0).
+
+%% We will recalculate the #rates{} every time we get asked for our
+%% RAM duration, or every N messages published, whichever is
+%% sooner. We do this since the priority calculations in
+%% rabbit_amqqueue_process need fairly fresh rates.
+-define(MSGS_PER_RATE_CALC, 100).
+
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
 
 start(DurableQueues) ->
-    {AllTerms, StartFunState} = rabbit_queue_index:recover(DurableQueues),
+    {AllTerms, StartFunState} = rabbit_queue_index:start(DurableQueues),
     start_msg_store(
       [Ref || Terms <- AllTerms,
+              Terms /= non_clean_shutdown,
               begin
                   Ref = proplists:get_value(persistent_ref, Terms),
                   Ref =/= undefined
               end],
-      StartFunState).
+      StartFunState),
+    {ok, AllTerms}.
 
-stop() -> stop_msg_store().
+stop() ->
+    ok = stop_msg_store(),
+    ok = rabbit_queue_index:stop().
 
 start_msg_store(Refs, StartFunState) ->
     ok = rabbit_sup:start_child(?TRANSIENT_MSG_STORE, rabbit_msg_store,
@@ -419,7 +433,7 @@ init(Queue, Recover, AsyncCallback) ->
          end,
          fun (MsgIds) -> msg_indices_written_to_disk(AsyncCallback, MsgIds) end).
 
-init(#amqqueue { name = QueueName, durable = IsDurable }, false,
+init(#amqqueue { name = QueueName, durable = IsDurable }, new,
      AsyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) ->
     IndexState = rabbit_queue_index:init(QueueName, MsgIdxOnDiskFun),
     init(IsDurable, IndexState, 0, [],
@@ -430,29 +444,32 @@ init(#amqqueue { name = QueueName, durable = IsDurable }, false,
          end,
          msg_store_client_init(?TRANSIENT_MSG_STORE, undefined, AsyncCallback));
 
-init(#amqqueue { name = QueueName, durable = true }, true,
+init(#amqqueue { name = QueueName, durable = true }, Terms,
      AsyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) ->
-    Terms = rabbit_queue_index:shutdown_terms(QueueName),
-    {PRef, Terms1} =
-        case proplists:get_value(persistent_ref, Terms) of
-            undefined -> {rabbit_guid:gen(), []};
-            PRef1     -> {PRef1, Terms}
-        end,
+    {PRef, RecoveryTerms} = process_recovery_terms(Terms),
     PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE, PRef,
                                              MsgOnDiskFun, AsyncCallback),
     TransientClient  = msg_store_client_init(?TRANSIENT_MSG_STORE,
                                              undefined, AsyncCallback),
     {DeltaCount, IndexState} =
         rabbit_queue_index:recover(
-          QueueName, Terms1,
+          QueueName, RecoveryTerms,
           rabbit_msg_store:successfully_recovered_state(?PERSISTENT_MSG_STORE),
           fun (MsgId) ->
                   rabbit_msg_store:contains(MsgId, PersistentClient)
           end,
           MsgIdxOnDiskFun),
-    init(true, IndexState, DeltaCount, Terms1,
+    init(true, IndexState, DeltaCount, RecoveryTerms,
          PersistentClient, TransientClient).
 
+process_recovery_terms(Terms=non_clean_shutdown) ->
+    {rabbit_guid:gen(), Terms};
+process_recovery_terms(Terms) ->
+    case proplists:get_value(persistent_ref, Terms) of
+        undefined -> {rabbit_guid:gen(), []};
+        PRef      -> {PRef, Terms}
+    end.
+
 terminate(_Reason, State) ->
     State1 = #vqstate { persistent_count  = PCount,
                         index_state       = IndexState,
@@ -533,14 +550,18 @@ publish(Msg = #basic_message { is_persistent = IsPersistent, id = MsgId },
                  false -> State1 #vqstate { q1 = ?QUEUE:in(m(MsgStatus1), Q1) };
                  true  -> State1 #vqstate { q4 = ?QUEUE:in(m(MsgStatus1), Q4) }
              end,
-    PCount1 = PCount + one_if(IsPersistent1),
+    InCount1 = InCount + 1,
+    PCount1  = PCount  + one_if(IsPersistent1),
     UC1 = gb_sets_maybe_insert(NeedsConfirming, MsgId, UC),
-    a(reduce_memory_use(
-        inc_ram_msg_count(State2 #vqstate { next_seq_id      = SeqId   + 1,
-                                            len              = Len     + 1,
-                                            in_counter       = InCount + 1,
-                                            persistent_count = PCount1,
-                                            unconfirmed      = UC1 }))).
+    State3 = inc_ram_msg_count(State2 #vqstate { next_seq_id      = SeqId + 1,
+                                                 len              = Len   + 1,
+                                                 in_counter       = InCount1,
+                                                 persistent_count = PCount1,
+                                                 unconfirmed      = UC1 }),
+    a(reduce_memory_use(case InCount1 > ?MSGS_PER_RATE_CALC of
+                            true  -> update_rates(State3);
+                            false -> State3
+                        end)).
 
 publish_delivered(Msg = #basic_message { is_persistent = IsPersistent,
                                          id = MsgId },
@@ -689,10 +710,10 @@ depth(State = #vqstate { ram_pending_ack = RPA, disk_pending_ack = DPA }) ->
 
 set_ram_duration_target(
   DurationTarget, State = #vqstate {
-                    rates     = #rates { avg_egress  = AvgEgressRate,
-                                         avg_ingress = AvgIngressRate },
-                    ack_rates = #rates { avg_egress  = AvgAckEgressRate,
-                                         avg_ingress = AvgAckIngressRate },
+                    rates = #rates { in      = AvgIngressRate,
+                                     out     = AvgEgressRate,
+                                     ack_in  = AvgAckIngressRate,
+                                     ack_out = AvgAckEgressRate },
                     target_ram_count = TargetRamCount }) ->
     Rate =
         AvgEgressRate + AvgIngressRate + AvgAckEgressRate + AvgAckIngressRate,
@@ -709,29 +730,43 @@ set_ram_duration_target(
           false -> reduce_memory_use(State1)
       end).
 
-ram_duration(State = #vqstate {
-               rates              = #rates { timestamp = Timestamp,
-                                             egress    = Egress,
-                                             ingress   = Ingress } = Rates,
-               ack_rates          = #rates { timestamp = AckTimestamp,
-                                             egress    = AckEgress,
-                                             ingress   = AckIngress } = ARates,
-               in_counter         = InCount,
-               out_counter        = OutCount,
-               ack_in_counter     = AckInCount,
-               ack_out_counter    = AckOutCount,
-               ram_msg_count      = RamMsgCount,
-               ram_msg_count_prev = RamMsgCountPrev,
-               ram_pending_ack    = RPA,
-               ram_ack_count_prev = RamAckCountPrev }) ->
-    Now = now(),
-    {AvgEgressRate,   Egress1} = update_rate(Now, Timestamp, OutCount, Egress),
-    {AvgIngressRate, Ingress1} = update_rate(Now, Timestamp, InCount, Ingress),
-
-    {AvgAckEgressRate,   AckEgress1} =
-        update_rate(Now, AckTimestamp, AckOutCount, AckEgress),
-    {AvgAckIngressRate, AckIngress1} =
-        update_rate(Now, AckTimestamp, AckInCount, AckIngress),
+update_rates(State = #vqstate{ in_counter      = InCount,
+                               out_counter     = OutCount,
+                               ack_in_counter  = AckInCount,
+                               ack_out_counter = AckOutCount,
+                               rates = #rates{ in        = InRate,
+                                               out       = OutRate,
+                                               ack_in    = AckInRate,
+                                               ack_out   = AckOutRate,
+                                               timestamp = TS }}) ->
+    Now = erlang:now(),
+
+    Rates = #rates { in        = update_rate(Now, TS, InCount,     InRate),
+                     out       = update_rate(Now, TS, OutCount,    OutRate),
+                     ack_in    = update_rate(Now, TS, AckInCount,  AckInRate),
+                     ack_out   = update_rate(Now, TS, AckOutCount, AckOutRate),
+                     timestamp = Now },
+
+    State#vqstate{ in_counter      = 0,
+                   out_counter     = 0,
+                   ack_in_counter  = 0,
+                   ack_out_counter = 0,
+                   rates           = Rates }.
+
+update_rate(Now, TS, Count, Rate) ->
+    Time = timer:now_diff(Now, TS) / ?MICROS_PER_SECOND,
+    rabbit_misc:moving_average(Time, ?RATE_AVG_HALF_LIFE, Count / Time, Rate).
+
+ram_duration(State) ->
+    State1 = #vqstate { rates = #rates { in      = AvgIngressRate,
+                                         out     = AvgEgressRate,
+                                         ack_in  = AvgAckIngressRate,
+                                         ack_out = AvgAckEgressRate },
+                        ram_msg_count      = RamMsgCount,
+                        ram_msg_count_prev = RamMsgCountPrev,
+                        ram_pending_ack    = RPA,
+                        ram_ack_count_prev = RamAckCountPrev } =
+        update_rates(State),
 
     RamAckCount = gb_trees:size(RPA),
 
@@ -745,25 +780,7 @@ ram_duration(State = #vqstate {
                                    AvgAckEgressRate + AvgAckIngressRate))
         end,
 
-    {Duration, State #vqstate {
-                 rates              = Rates #rates {
-                                        egress      = Egress1,
-                                        ingress     = Ingress1,
-                                        avg_egress  = AvgEgressRate,
-                                        avg_ingress = AvgIngressRate,
-                                        timestamp   = Now },
-                 ack_rates          = ARates #rates {
-                                        egress      = AckEgress1,
-                                        ingress     = AckIngress1,
-                                        avg_egress  = AvgAckEgressRate,
-                                        avg_ingress = AvgAckIngressRate,
-                                        timestamp   = Now },
-                 in_counter         = 0,
-                 out_counter        = 0,
-                 ack_in_counter     = 0,
-                 ack_out_counter    = 0,
-                 ram_msg_count_prev = RamMsgCount,
-                 ram_ack_count_prev = RamAckCount }}.
+    {Duration, State1}.
 
 needs_timeout(State = #vqstate { index_state      = IndexState,
                                  target_ram_count = TargetRamCount }) ->
@@ -789,6 +806,10 @@ timeout(State = #vqstate { index_state = IndexState }) ->
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
 
+msg_rates(#vqstate { rates = #rates { in  = AvgIngressRate,
+                                      out = AvgEgressRate } }) ->
+    {AvgIngressRate, AvgEgressRate}.
+
 status(#vqstate {
           q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
           len              = Len,
@@ -798,10 +819,11 @@ status(#vqstate {
           ram_msg_count    = RamMsgCount,
           next_seq_id      = NextSeqId,
           persistent_count = PersistentCount,
-          rates            = #rates { avg_egress  = AvgEgressRate,
-                                      avg_ingress = AvgIngressRate },
-          ack_rates        = #rates { avg_egress  = AvgAckEgressRate,
-                                      avg_ingress = AvgAckIngressRate } }) ->
+          rates            = #rates { in      = AvgIngressRate,
+                                      out     = AvgEgressRate,
+                                      ack_in  = AvgAckIngressRate,
+                                      ack_out = AvgAckEgressRate }}) ->
+
     [ {q1                  , ?QUEUE:len(Q1)},
       {q2                  , ?QUEUE:len(Q2)},
       {delta               , Delta},
@@ -991,10 +1013,6 @@ expand_delta(SeqId, #delta { count        = Count,
 expand_delta(_SeqId, #delta { count       = Count } = Delta) ->
     d(Delta #delta { count = Count + 1 }).
 
-update_rate(Now, Then, Count, {OThen, OCount}) ->
-    %% avg over the current period and the previous
-    {1000000.0 * (Count + OCount) / timer:now_diff(Now, OThen), {Then, Count}}.
-
 %%----------------------------------------------------------------------------
 %% Internal major helpers for Public API
 %%----------------------------------------------------------------------------
@@ -1003,7 +1021,12 @@ init(IsDurable, IndexState, DeltaCount, Terms,
      PersistentClient, TransientClient) ->
     {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState),
 
-    DeltaCount1 = proplists:get_value(persistent_count, Terms, DeltaCount),
+    DeltaCount1 =
+        case Terms of
+            non_clean_shutdown -> DeltaCount;
+            _                  -> proplists:get_value(persistent_count,
+                                                      Terms, DeltaCount)
+        end,
     Delta = case DeltaCount1 == 0 andalso DeltaCount /= undefined of
                 true  -> ?BLANK_DELTA;
                 false -> d(#delta { start_seq_id = LowSeqId,
@@ -1034,22 +1057,21 @@ init(IsDurable, IndexState, DeltaCount, Terms,
       ram_ack_count_prev  = 0,
       out_counter         = 0,
       in_counter          = 0,
-      rates               = blank_rate(Now, DeltaCount1),
+      rates               = blank_rates(Now),
       msgs_on_disk        = gb_sets:new(),
       msg_indices_on_disk = gb_sets:new(),
       unconfirmed         = gb_sets:new(),
       confirmed           = gb_sets:new(),
       ack_out_counter     = 0,
-      ack_in_counter      = 0,
-      ack_rates           = blank_rate(Now, 0) },
+      ack_in_counter      = 0 },
     a(maybe_deltas_to_betas(State)).
 
-blank_rate(Timestamp, IngressLength) ->
-    #rates { egress      = {Timestamp, 0},
-             ingress     = {Timestamp, IngressLength},
-             avg_egress  = 0.0,
-             avg_ingress = 0.0,
-             timestamp   = Timestamp }.
+blank_rates(Now) ->
+    #rates { in        = 0.0,
+             out       = 0.0,
+             ack_in    = 0.0,
+             ack_out   = 0.0,
+             timestamp = Now}.
 
 in_r(MsgStatus = #msg_status { msg = undefined },
      State = #vqstate { q3 = Q3, q4 = Q4 }) ->
@@ -1523,11 +1545,10 @@ reduce_memory_use(AlphaBetaFun, BetaDeltaFun, AckFun,
                     ram_pending_ack  = RPA,
                     ram_msg_count    = RamMsgCount,
                     target_ram_count = TargetRamCount,
-                    rates            = #rates { avg_ingress = AvgIngress,
-                                                avg_egress  = AvgEgress },
-                    ack_rates        = #rates { avg_ingress = AvgAckIngress,
-                                                avg_egress  = AvgAckEgress }
-                   }) ->
+                    rates            = #rates { in      = AvgIngress,
+                                                out     = AvgEgress,
+                                                ack_in  = AvgAckIngress,
+                                                ack_out = AvgAckEgress } }) ->
 
     {Reduce, State1 = #vqstate { q2 = Q2, q3 = Q3 }} =
         case chunk_size(RamMsgCount + gb_trees:size(RPA), TargetRamCount) of
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 92d48e63..3571692b 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, start/6, start_link/6]).
+-export([start/6, start_link/6, start/7, start_link/7]).
 
 -export([system_continue/3, system_terminate/4, system_code_change/4]).
 
@@ -30,7 +30,7 @@
 -export([internal_send_command/4, internal_send_command/6]).
 
 %% internal
--export([mainloop/2, mainloop1/2]).
+-export([enter_mainloop/2, mainloop/2, mainloop1/2]).
 
 -record(wstate, {sock, channel, frame_max, protocol, reader,
                  stats_timer, pending}).
@@ -41,21 +41,25 @@
 
 -ifdef(use_specs).
 
--spec(start/5 ::
+-spec(start/6 ::
         (rabbit_net:socket(), rabbit_channel:channel_number(),
-         non_neg_integer(), rabbit_types:protocol(), pid())
+         non_neg_integer(), rabbit_types:protocol(), pid(),
+         rabbit_types:proc_name())
         -> rabbit_types:ok(pid())).
--spec(start_link/5 ::
+-spec(start_link/6 ::
         (rabbit_net:socket(), rabbit_channel:channel_number(),
-         non_neg_integer(), rabbit_types:protocol(), pid())
+         non_neg_integer(), rabbit_types:protocol(), pid(),
+         rabbit_types:proc_name())
         -> rabbit_types:ok(pid())).
--spec(start/6 ::
+-spec(start/7 ::
         (rabbit_net:socket(), rabbit_channel:channel_number(),
-         non_neg_integer(), rabbit_types:protocol(), pid(), boolean())
+         non_neg_integer(), rabbit_types:protocol(), pid(),
+         rabbit_types:proc_name(), boolean())
         -> rabbit_types:ok(pid())).
--spec(start_link/6 ::
+-spec(start_link/7 ::
         (rabbit_net:socket(), rabbit_channel:channel_number(),
-         non_neg_integer(), rabbit_types:protocol(), pid(), boolean())
+         non_neg_integer(), rabbit_types:protocol(), pid(),
+         rabbit_types:proc_name(), boolean())
         -> rabbit_types:ok(pid())).
 
 -spec(system_code_change/4 :: (_,_,_,_) -> {'ok',_}).
@@ -99,23 +103,23 @@
 
 %%---------------------------------------------------------------------------
 
-start(Sock, Channel, FrameMax, Protocol, ReaderPid) ->
-    start(Sock, Channel, FrameMax, Protocol, ReaderPid, false).
+start(Sock, Channel, FrameMax, Protocol, ReaderPid, Identity) ->
+    start(Sock, Channel, FrameMax, Protocol, ReaderPid, Identity, false).
 
-start_link(Sock, Channel, FrameMax, Protocol, ReaderPid) ->
-    start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, false).
+start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, Identity) ->
+    start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, Identity, false).
 
-start(Sock, Channel, FrameMax, Protocol, ReaderPid, ReaderWantsStats) ->
+start(Sock, Channel, FrameMax, Protocol, ReaderPid, Identity,
+      ReaderWantsStats) ->
     State = initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid,
                           ReaderWantsStats),
-    Deb = sys:debug_options([]),
-    {ok, proc_lib:spawn(?MODULE, mainloop, [Deb, State])}.
+    {ok, proc_lib:spawn(?MODULE, enter_mainloop, [Identity, State])}.
 
-start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, ReaderWantsStats) ->
+start_link(Sock, Channel, FrameMax, Protocol, ReaderPid, Identity,
+           ReaderWantsStats) ->
     State = initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid,
                           ReaderWantsStats),
-    Deb = sys:debug_options([]),
-    {ok, proc_lib:spawn_link(?MODULE, mainloop, [Deb, State])}.
+    {ok, proc_lib:spawn_link(?MODULE, enter_mainloop, [Identity, State])}.
 
 initial_state(Sock, Channel, FrameMax, Protocol, ReaderPid, ReaderWantsStats) ->
     (case ReaderWantsStats of
@@ -138,6 +142,11 @@ system_terminate(Reason, _Parent, _Deb, _State) ->
 system_code_change(Misc, _Module, _OldVsn, _Extra) ->
     {ok, Misc}.
 
+enter_mainloop(Identity, State) ->
+    Deb = sys:debug_options([]),
+    ?store_proc_name(Identity),
+    mainloop(Deb, State).
+
 mainloop(Deb, State) ->
     try
         mainloop1(Deb, State)
@@ -272,7 +281,7 @@ assemble_frames(Channel, MethodRecord, Content, FrameMax, Protocol) ->
     [MethodFrame | ContentFrames].
 
 tcp_send(Sock, Data) ->
-    rabbit_misc:throw_on_error(writer_inet_error,
+    rabbit_misc:throw_on_error(inet_error,
                                fun () -> rabbit_net:send(Sock, Data) end).
 
 internal_send_command(Sock, Channel, MethodRecord, Protocol) ->