merge default into bug25817

4 files changed, 162 insertions, 154 deletions

diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 7002fd367c..cb59edd9b1 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -238,10 +238,11 @@ notify_decorators(Event, Props, State) when Event =:= startup;
 notify_decorators(Event, Props, State = #q{active_consumers    = ACs,
                                            backing_queue       = BQ,
                                            backing_queue_state = BQS}) ->
-    decorator_callback(
-      qname(State), notify,
-      [Event, [{max_active_consumer_priority, priority_queue:highest(ACs)},
-               {is_empty,                     BQ:is_empty(BQS)} | Props]]).
+    P = priority_queue:highest(ACs),
+    decorator_callback(qname(State), notify,
+                       [Event, [{max_active_consumer_priority, P},
+                                {is_empty, BQ:is_empty(BQS)} |
+                                Props]]).
 
 decorator_callback(QName, F, A) ->
     %% Look up again in case policy and hence decorators have changed
@@ -326,8 +327,8 @@ terminate_shutdown(Fun, State) ->
         _         -> ok = rabbit_memory_monitor:deregister(self()),
                      QName = qname(State),
                      notify_decorators(shutdown, [], State),
-                     [emit_consumer_deleted(Ch, CTag, QName)
-                      || {Ch, CTag, _} <- consumers(State1)],
+                     [emit_consumer_deleted(Ch, CTag, QName) ||
+                         {Ch, CTag, _} <- consumers(State1)],
                      State1#q{backing_queue_state = Fun(BQS)}
     end.
 
@@ -485,12 +486,10 @@ send_drained(C = #cr{ch_pid = ChPid, limiter = Limiter}) ->
 deliver_msgs_to_consumers(_DeliverFun, true, State) ->
     {true, State};
 deliver_msgs_to_consumers(DeliverFun, false,
-                          State = #q{active_consumers  = ActiveConsumers,
-                                     consumer_use      = CUInfo}) ->
+                          State = #q{active_consumers = ActiveConsumers}) ->
     case priority_queue:out_p(ActiveConsumers) of
         {empty, _} ->
-            {false,
-             State#q{consumer_use = update_consumer_use(CUInfo, inactive)}};
+            {false, update_consumer_use(State, inactive)};
         {{value, QEntry, Priority}, Tail} ->
             {Stop, State1} = deliver_msg_to_consumer(
                                DeliverFun, QEntry, Priority,
@@ -537,18 +536,17 @@ deliver_msg_to_consumer0(DeliverFun,
                           unsent_message_count = Count + 1}),
     {Stop, State1}.
 
-deliver_from_queue_deliver(AckRequired, State) ->
-    {Result, State1} = fetch(AckRequired, State),
-    {Result, is_empty(State1), State1}.
+update_consumer_use(State = #q{consumer_use = CUInfo}, Use) ->
+    State#q{consumer_use = update_consumer_use1(CUInfo, Use)}.
 
-update_consumer_use({inactive, _, _, _}   = CUInfo, inactive) ->
+update_consumer_use1({inactive, _, _, _}   = CUInfo, inactive) ->
     CUInfo;
-update_consumer_use({active,   _, _}      = CUInfo, active) ->
+update_consumer_use1({active,   _, _}      = CUInfo,   active) ->
     CUInfo;
-update_consumer_use({active,   Since,         Avg}, inactive) ->
+update_consumer_use1({active,   Since,         Avg}, inactive) ->
     Now = now_micros(),
     {inactive, Now, Now - Since, Avg};
-update_consumer_use({inactive, Since, Active, Avg}, active) ->
+update_consumer_use1({inactive, Since, Active, Avg},   active) ->
     Now = now_micros(),
     {active, Now, consumer_use_avg(Active, Now - Since, Avg)}.
 
@@ -607,10 +605,12 @@ discard(#delivery{sender     = SenderPid,
     State1#q{backing_queue_state = BQS1}.
 
 run_message_queue(State) ->
-    {_IsEmpty1, State1} = deliver_msgs_to_consumers(
-                            fun deliver_from_queue_deliver/2,
-                            is_empty(State), State),
-    State1.
+    {_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.
 
 add_consumer({ChPid, Consumer = #consumer{args = Args}}, ActiveConsumers) ->
     Priority = case rabbit_misc:table_lookup(Args, <<"x-priority">>) of
@@ -622,21 +622,21 @@ add_consumer({ChPid, Consumer = #consumer{args = Args}}, ActiveConsumers) ->
 attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message},
                  Props, Delivered, State = #q{backing_queue       = BQ,
                                               backing_queue_state = BQS}) ->
-    case BQ:is_duplicate(Message, BQS) of
-        {false, BQS1} ->
-            deliver_msgs_to_consumers(
-              fun (true, State1 = #q{backing_queue_state = BQS2}) ->
-                      true = BQ:is_empty(BQS2),
-                      {AckTag, BQS3} = BQ:publish_delivered(
-                                         Message, Props, SenderPid, BQS2),
-                      {{Message, Delivered, AckTag},
-                       true, State1#q{backing_queue_state = BQS3}};
-                  (false, State1) ->
-                      {{Message, Delivered, undefined},
-                       true, discard(Delivery, State1)}
-              end, false, State#q{backing_queue_state = BQS1});
-        {true, BQS1} ->
-            {true, State#q{backing_queue_state = BQS1}}
+    {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}
     end.
 
 deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid},
@@ -652,7 +652,7 @@ deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid},
         {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}),
+                maybe_drop_head(State2#q{backing_queue_state = BQS1}),
             QLen = BQ:len(BQS2),
             %% optimisation: it would be perfectly safe to always
             %% invoke drop_expired_msgs here, but that is expensive so
@@ -738,7 +738,7 @@ possibly_unblock(State, ChPid, Update) ->
                      end
     end.
 
-unblock(State = #q{consumer_use = CUInfo}, C = #cr{limiter = Limiter}) ->
+unblock(State, C = #cr{limiter = Limiter}) ->
     case lists:partition(
            fun({_P, {_ChPid, #consumer{tag = CTag}}}) ->
                    rabbit_limiter:is_consumer_blocked(Limiter, CTag)
@@ -750,49 +750,47 @@ unblock(State = #q{consumer_use = CUInfo}, C = #cr{limiter = Limiter}) ->
             BlockedQ   = priority_queue:from_list(Blocked),
             UnblockedQ = priority_queue:from_list(Unblocked),
             update_ch_record(C#cr{blocked_consumers = BlockedQ}),
-            State1 = State#q{consumer_use =
-                                 update_consumer_use(CUInfo, active)},
-            AC1 = priority_queue:join(State1#q.active_consumers, UnblockedQ),
-            State2 = State1#q{active_consumers = AC1},
+            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}], State2) ||
+               consumer_unblocked, [{consumer_tag, CTag}], State1) ||
                 {_P, {_ChPid, #consumer{tag = CTag}}} <- Unblocked],
-            run_message_queue(State2)
+            run_message_queue(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{exclusive_consumer = Holder,
+handle_ch_down(DownPid, State = #q{active_consumers   = AC,
+                                   exclusive_consumer = Holder,
                                    senders            = Senders}) ->
-    Senders1 = case pmon:is_monitored(DownPid, Senders) of
-                   false -> Senders;
-                   true  -> credit_flow:peer_down(DownPid),
-                            pmon:demonitor(DownPid, Senders)
-               end,
+    State1 = State#q{senders = case pmon:is_monitored(DownPid, Senders) of
+                                   false -> Senders;
+                                   true  -> credit_flow:peer_down(DownPid),
+                                            pmon:demonitor(DownPid, Senders)
+                               end},
     case lookup_ch(DownPid) of
         not_found ->
-            {ok, State#q{senders = Senders1}};
+            {ok, State1};
         C = #cr{ch_pid            = ChPid,
                 acktags           = ChAckTags,
                 blocked_consumers = Blocked} ->
             QName = qname(State),
-            _ = remove_consumers(ChPid, Blocked, QName), %% for stats emission
+            AC1 = remove_consumers(ChPid, AC,      QName),
+            _   = remove_consumers(ChPid, Blocked, QName), %% for stats emission
             ok = erase_ch_record(C),
-            State1 = State#q{
-                       exclusive_consumer = case Holder of
-                                                {ChPid, _} -> none;
-                                                Other      -> Other
-                                            end,
-                       active_consumers = remove_consumers(
-                                            ChPid, State#q.active_consumers,
-                                            QName),
-                       senders          = Senders1},
-            case should_auto_delete(State1) of
-                true  -> {stop, State1};
+            Holder1 = case Holder of
+                          {DownPid, _} -> none;
+                          Other        -> Other
+                      end,
+            State2 = State1#q{active_consumers   = AC1,
+                              exclusive_consumer = Holder1},
+            case should_auto_delete(State2) of
+                true  -> {stop, State2};
                 false -> {ok, requeue_and_run(queue:to_list(ChAckTags),
-                                              ensure_expiry_timer(State1))}
+                                              ensure_expiry_timer(State2))}
             end
     end.
 
@@ -1223,64 +1221,66 @@ handle_call({basic_get, ChPid, NoAck, LimiterPid}, _From,
     case fetch(AckRequired, State1) of
         {empty, State2} ->
             reply(empty, State2);
-        {{Message, IsDelivered, AckTag}, State2} ->
-            State3 = #q{backing_queue = BQ, backing_queue_state = BQS} =
-                case AckRequired of
-                    true  -> C = #cr{acktags = ChAckTags} =
-                                 ch_record(ChPid, LimiterPid),
-                             ChAckTags1 = queue:in(AckTag, ChAckTags),
-                             update_ch_record(C#cr{acktags = ChAckTags1}),
-                             State2;
-                    false -> State2
-                end,
+        {{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});
+                false -> ok
+            end,
             Msg = {QName, self(), AckTag, IsDelivered, Message},
-            reply({ok, BQ:len(BQS), Msg}, State3)
+            reply({ok, BQ:len(BQS), Msg}, State2)
     end;
 
 handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
              ConsumerTag, ExclusiveConsume, CreditArgs, OtherArgs, OkMsg},
-            _From, State = #q{exclusive_consumer = Holder}) ->
+            _From, State = #q{active_consumers   = AC,
+                              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},
-            ExclusiveConsumer = if ExclusiveConsume -> {ChPid, ConsumerTag};
-                                   true             -> Holder
-                                end,
-            State1 = State#q{has_had_consumers = true,
-                             exclusive_consumer = ExclusiveConsumer},
-            ok = maybe_send_reply(ChPid, OkMsg),
-            emit_consumer_created(ChPid, ConsumerTag, ExclusiveConsume,
-                                  not NoAck, qname(State1), OtherArgs),
-            AC1 = add_consumer({ChPid, Consumer}, State1#q.active_consumers),
-            State2 = State1#q{active_consumers = AC1},
-            notify_decorators(
-              basic_consume, [{consumer_tag, ConsumerTag}], State2),
-            reply(ok, run_message_queue(State2))
+        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),
+                  ExclusiveConsumer =
+                      if ExclusiveConsume -> {ChPid, ConsumerTag};
+                         true             -> Holder
+                      end,
+                  State1 = State#q{active_consumers   = AC1,
+                                   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),
+                  reply(ok, run_message_queue(State1))
     end;
 
 handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
-            State = #q{exclusive_consumer = Holder}) ->
+            State = #q{active_consumers   = AC,
+                       exclusive_consumer = Holder}) ->
     ok = maybe_send_reply(ChPid, OkMsg),
     case lookup_ch(ChPid) of
         not_found ->
@@ -1288,7 +1288,7 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
         C = #cr{consumer_count    = Count,
                 limiter           = Limiter,
                 blocked_consumers = Blocked} ->
-            emit_consumer_deleted(ChPid, ConsumerTag, qname(State)),
+            AC1      = remove_consumer(ChPid, ConsumerTag, AC),
             Blocked1 = remove_consumer(ChPid, ConsumerTag, Blocked),
             Limiter1 = case Count of
                            1 -> rabbit_limiter:deactivate(Limiter);
@@ -1298,14 +1298,13 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
             update_ch_record(C#cr{consumer_count    = Count - 1,
                                   limiter           = Limiter2,
                                   blocked_consumers = Blocked1}),
-            State1 = State#q{
-                       exclusive_consumer = case Holder of
-                                                {ChPid, ConsumerTag} -> none;
-                                                _                    -> Holder
-                                            end,
-                       active_consumers   = remove_consumer(
-                                              ChPid, ConsumerTag,
-                                              State#q.active_consumers)},
+            Holder1 = case Holder of
+                          {ChPid, ConsumerTag} -> none;
+                          _                    -> Holder
+                      end,
+            State1 = State#q{active_consumers   = AC1,
+                             exclusive_consumer = Holder1},
+            emit_consumer_deleted(ChPid, ConsumerTag, qname(State1)),
             notify_decorators(
               basic_cancel, [{consumer_tag, ConsumerTag}], State1),
             case should_auto_delete(State1) of
@@ -1372,11 +1371,12 @@ handle_call(force_event_refresh, _From,
             State = #q{exclusive_consumer = Exclusive}) ->
     rabbit_event:notify(queue_created, infos(?CREATION_EVENT_KEYS, State)),
     QName = qname(State),
+    AllConsumers = consumers(State),
     case Exclusive of
         none       -> [emit_consumer_created(
                          Ch, CTag, false, AckRequired, QName, Args) ||
-                          {Ch, CTag, AckRequired, Args} <- consumers(State)];
-        {Ch, CTag} -> [{Ch, CTag, AckRequired, Args}] = consumers(State),
+                          {Ch, CTag, AckRequired, Args} <- AllConsumers];
+        {Ch, CTag} -> [{Ch, CTag, AckRequired, Args}] = AllConsumers,
                       emit_consumer_created(
                         Ch, CTag, true, AckRequired, QName, Args)
     end,
@@ -1451,21 +1451,21 @@ handle_cast({set_maximum_since_use, Age}, State) ->
     noreply(State);
 
 handle_cast(start_mirroring, State = #q{backing_queue       = BQ,
-					backing_queue_state = BQS}) ->
+                                        backing_queue_state = BQS}) ->
     %% lookup again to get policy for init_with_existing_bq
     {ok, Q} = rabbit_amqqueue:lookup(qname(State)),
     true = BQ =/= rabbit_mirror_queue_master, %% assertion
     BQ1 = rabbit_mirror_queue_master,
     BQS1 = BQ1:init_with_existing_bq(Q, BQ, BQS),
     noreply(State#q{backing_queue       = BQ1,
-		    backing_queue_state = BQS1});
+                    backing_queue_state = BQS1});
 
 handle_cast(stop_mirroring, State = #q{backing_queue       = BQ,
-				       backing_queue_state = BQS}) ->
+                                       backing_queue_state = BQS}) ->
     BQ = rabbit_mirror_queue_master, %% assertion
     {BQ1, BQS1} = BQ:stop_mirroring(BQS),
     noreply(State#q{backing_queue       = BQ1,
-		    backing_queue_state = BQS1});
+                    backing_queue_state = BQS1});
 
 handle_cast({credit, ChPid, CTag, Credit, Drain},
             State = #q{backing_queue       = BQ,
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index da22e09266..0aa44d9f47 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -736,6 +736,14 @@ handle_input({frame_payload, Type, Channel, PayloadSize}, Data, State) ->
                                         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}) ->
+    refuse_connection(Sock, {bad_header, Other});
+
+handle_input(Callback, Data, _State) ->
+    throw({bad_input, Callback, Data}).
+
 %% The two rules pertaining to version negotiation:
 %%
 %% * If the server cannot support the protocol specified in the
@@ -744,37 +752,31 @@ handle_input({frame_payload, Type, Channel, PayloadSize}, Data, State) ->
 %%
 %% * The server MUST provide a protocol version that is lower than or
 %% equal to that requested by the client in the protocol header.
-handle_input(handshake, <<"AMQP", 0, 0, 9, 1>>, State) ->
+handshake({0, 0, 9, 1}, State) ->
     start_connection({0, 9, 1}, rabbit_framing_amqp_0_9_1, State);
 
 %% This is the protocol header for 0-9, which we can safely treat as
 %% though it were 0-9-1.
-handle_input(handshake, <<"AMQP", 1, 1, 0, 9>>, State) ->
+handshake({1, 1, 0, 9}, State) ->
     start_connection({0, 9, 0}, rabbit_framing_amqp_0_9_1, State);
 
 %% This is what most clients send for 0-8.  The 0-8 spec, confusingly,
 %% defines the version as 8-0.
-handle_input(handshake, <<"AMQP", 1, 1, 8, 0>>, State) ->
+handshake({1, 1, 8, 0}, State) ->
     start_connection({8, 0, 0}, rabbit_framing_amqp_0_8, State);
 
 %% The 0-8 spec as on the AMQP web site actually has this as the
 %% protocol header; some libraries e.g., py-amqplib, send it when they
 %% want 0-8.
-handle_input(handshake, <<"AMQP", 1, 1, 9, 1>>, State) ->
+handshake({1, 1, 9, 1}, State) ->
     start_connection({8, 0, 0}, rabbit_framing_amqp_0_8, State);
 
-%% ... and finally, the 1.0 spec is crystal clear!  Note that the
-handle_input(handshake, <<"AMQP", Id, 1, 0, 0>>, State) ->
+%% ... and finally, the 1.0 spec is crystal clear!
+handshake({Id, 1, 0, 0}, State) ->
     become_1_0(Id, State);
 
-handle_input(handshake, <<"AMQP", A, B, C, D>>, #v1{sock = Sock}) ->
-    refuse_connection(Sock, {bad_version, {A, B, C, D}});
-
-handle_input(handshake, Other, #v1{sock = Sock}) ->
-    refuse_connection(Sock, {bad_header, Other});
-
-handle_input(Callback, Data, _State) ->
-    throw({bad_input, Callback, Data}).
+handshake(Vsn, #v1{sock = Sock}) ->
+    refuse_connection(Sock, {bad_version, Vsn}).
 
 %% Offer a protocol version to the client.  Connection.start only
 %% includes a major and minor version number, Luckily 0-9 and 0-9-1
@@ -818,7 +820,10 @@ handle_method0(MethodName, FieldsBin,
     try
         handle_method0(Protocol:decode_method_fields(MethodName, FieldsBin),
                        State)
-    catch exit:#amqp_error{method = none} = Reason ->
+    catch throw:{writer_inet_error, closed} ->
+            maybe_emit_stats(State),
+            throw(connection_closed_abruptly);
+          exit:#amqp_error{method = none} = Reason ->
             handle_exception(State, 0, Reason#amqp_error{method = MethodName});
           Type:Reason ->
             Stack = erlang:get_stacktrace(),
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 34dd3d3b35..92d48e633f 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -272,7 +272,7 @@ assemble_frames(Channel, MethodRecord, Content, FrameMax, Protocol) ->
     [MethodFrame | ContentFrames].
 
 tcp_send(Sock, Data) ->
-    rabbit_misc:throw_on_error(inet_error,
+    rabbit_misc:throw_on_error(writer_inet_error,
                                fun () -> rabbit_net:send(Sock, Data) end).
 
 internal_send_command(Sock, Channel, MethodRecord, Protocol) ->
diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl
index 369ec65596..fc4353dcb4 100644
--- a/src/vm_memory_monitor.erl
+++ b/src/vm_memory_monitor.erl
@@ -173,16 +173,19 @@ set_mem_limits(State, MemFraction) ->
                 ?MEMORY_SIZE_FOR_UNKNOWN_OS;
             M -> M
         end,
-    UsableMemory = case get_vm_limit() of
-                       Limit when Limit < TotalMemory ->
-                           error_logger:warning_msg(
-                             "Only ~pMB of ~pMB memory usable due to "
-                             "limited address space.~n",
-                             [trunc(V/?ONE_MB) || V <- [Limit, TotalMemory]]),
-                           Limit;
-                       _ ->
-                           TotalMemory
-                   end,
+    UsableMemory =
+        case get_vm_limit() of
+            Limit when Limit < TotalMemory ->
+                error_logger:warning_msg(
+                  "Only ~pMB of ~pMB memory usable due to "
+                  "limited address space.~n"
+                  "Crashes due to memory exhaustion are possible - see~n"
+                  "http://www.rabbitmq.com/memory.html#address-space~n",
+                  [trunc(V/?ONE_MB) || V <- [Limit, TotalMemory]]),
+                Limit;
+            _ ->
+                TotalMemory
+        end,
     MemLim = trunc(MemFraction * UsableMemory),
     error_logger:info_msg("Memory limit set to ~pMB of ~pMB total.~n",
                           [trunc(MemLim/?ONE_MB), trunc(TotalMemory/?ONE_MB)]),