Merged 19468 into default

6 files changed, 203 insertions, 35 deletions

diff --git a/src/buffering_proxy.erl b/src/buffering_proxy.erl
index dc1686081f..fcb7b412e2 100644
--- a/src/buffering_proxy.erl
+++ b/src/buffering_proxy.erl
@@ -32,6 +32,8 @@
 -export([mainloop/4, drain/2]).
 -export([proxy_loop/3]).
 
+-define(HIBERNATE_AFTER, 5000).
+
 %%----------------------------------------------------------------------------
 
 start_link(M, A) ->
@@ -59,6 +61,9 @@ mainloop(ProxyPid, Ref, M, State) ->
                 ProxyPid ! Ref,
                 NewSt;
             Msg -> M:handle_message(Msg, State)
+        after ?HIBERNATE_AFTER ->
+                erlang:hibernate(?MODULE, mainloop,
+                                 [ProxyPid, Ref, M, State])
         end,
     ?MODULE:mainloop(ProxyPid, Ref, M, NewState).
 
@@ -92,4 +97,6 @@ proxy_loop(Ref, Pid, State) ->
                    waiting  -> Pid ! {Ref, [Msg]}, empty;
                    Messages -> [Msg | Messages]
                end)
+    after ?HIBERNATE_AFTER ->
+            erlang:hibernate(?MODULE, proxy_loop, [Ref, Pid, State])
     end.
diff --git a/src/rabbit.erl b/src/rabbit.erl
index c6ef1749f2..a33c5b7bcb 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -157,6 +157,8 @@ start(normal, []) ->
 
                 ok = rabbit_amqqueue:start(),
 
+                ok = rabbit_alarm:start(),
+
                 ok = rabbit_binary_generator:
                     check_empty_content_body_frame_size(),
 
@@ -198,6 +200,7 @@ start(normal, []) ->
 
 stop(_State) ->
     terminated_ok = error_logger:delete_report_handler(rabbit_error_logger),
+    ok = rabbit_alarm:stop(),
     ok.
 
 %---------------------------------------------------------------------------
diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl
new file mode 100644
index 0000000000..346f5361e4
--- /dev/null
+++ b/src/rabbit_alarm.erl
@@ -0,0 +1,125 @@
+%%   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 Developers of the Original Code are LShift Ltd.,
+%%   Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd., Cohesive Financial Technologies
+%%   LLC., and Rabbit Technologies Ltd. are Copyright (C) 2007-2008
+%%   LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit
+%%   Technologies Ltd.;
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_alarm).
+
+-behaviour(gen_event).
+
+-export([start/0, stop/0, register/2]).
+
+-export([init/1, handle_call/2, handle_event/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-define(MEMSUP_CHECK_INTERVAL, 1000).
+
+-record(alarms, {alertees, system_memory_high_watermark = false}).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(start/0 :: () -> 'ok').
+-spec(stop/0 :: () -> 'ok').
+-spec(register/2 :: (pid(), mfa()) -> 'ok').
+             
+-endif.
+
+%%----------------------------------------------------------------------------
+
+start() ->
+    %% The default memsup check interval is 1 minute, which is way too
+    %% long - rabbit can gobble up all memory in a matter of
+    %% seconds. Unfortunately the memory_check_interval configuration
+    %% parameter and memsup:set_check_interval/1 function only provide
+    %% a granularity of minutes. So we have to peel off one layer of
+    %% the API to get to the underlying layer which operates at the
+    %% granularity of milliseconds.
+    %%
+    %% Note that the new setting will only take effect after the first
+    %% check has completed, i.e. after one minute. So if rabbit eats
+    %% all the memory within the first minute after startup then we
+    %% are out of luck.
+    ok = os_mon:call(memsup, {set_check_interval, ?MEMSUP_CHECK_INTERVAL},
+                     infinity),
+
+    ok = alarm_handler:add_alarm_handler(?MODULE).
+
+stop() ->
+    ok = alarm_handler:delete_alarm_handler(?MODULE).
+
+register(Pid, HighMemMFA) ->
+    ok = gen_event:call(alarm_handler, ?MODULE,
+                        {register, Pid, HighMemMFA}).
+
+%%----------------------------------------------------------------------------
+
+init([]) ->
+    {ok, #alarms{alertees = dict:new()}}.
+
+handle_call({register, Pid, HighMemMFA},
+            State = #alarms{alertees = Alertess}) ->
+    _MRef = erlang:monitor(process, Pid),
+    case State#alarms.system_memory_high_watermark of
+        true  -> {M, F, A} = HighMemMFA,
+                 ok = erlang:apply(M, F, A ++ [Pid, true]);
+        false -> ok
+    end,
+    NewAlertees = dict:store(Pid, HighMemMFA, Alertess),
+    {ok, ok, State#alarms{alertees = NewAlertees}};
+    
+handle_call(_Request, State) ->
+    {ok, not_understood, State}.
+
+handle_event({set_alarm, {system_memory_high_watermark, []}}, State) ->
+    ok = alert(true, State#alarms.alertees),
+    {ok, State#alarms{system_memory_high_watermark = true}};
+
+handle_event({clear_alarm, system_memory_high_watermark}, State) ->
+    ok = alert(false, State#alarms.alertees),
+    {ok, State#alarms{system_memory_high_watermark = false}};
+
+handle_event(_Event, State) ->
+    {ok, State}.
+
+handle_info({'DOWN', _MRef, process, Pid, _Reason},
+            State = #alarms{alertees = Alertess}) ->
+    {ok, State#alarms{alertees = dict:erase(Pid, Alertess)}};
+
+handle_info(_Info, State) ->
+    {ok, State}.
+
+terminate(_Arg, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%----------------------------------------------------------------------------
+    
+alert(Alert, Alertees) ->
+    dict:fold(fun (Pid, {M, F, A}, Acc) ->
+                      ok = erlang:apply(M, F, A ++ [Pid, Alert]),
+                      Acc
+              end, ok, Alertees).
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 7716ef1646..e687df846a 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -30,6 +30,7 @@
 -behaviour(gen_server).
 
 -define(UNSENT_MESSAGE_LIMIT, 100).
+-define(HIBERNATE_AFTER, 1000).
 
 -export([start_link/1]).
 
@@ -75,7 +76,7 @@ init(Q) ->
             has_had_consumers = false,
             next_msg_id = 1,
             message_buffer = queue:new(),
-            round_robin = queue:new()}}.
+            round_robin = queue:new()}, ?HIBERNATE_AFTER}.
 
 terminate(_Reason, State) ->
     %% FIXME: How do we cancel active subscriptions?
@@ -90,6 +91,10 @@ code_change(_OldVsn, State, _Extra) ->
 
 %%----------------------------------------------------------------------------
 
+reply(Reply, NewState) -> {reply, Reply, NewState, ?HIBERNATE_AFTER}.
+
+noreply(NewState) -> {noreply, NewState, ?HIBERNATE_AFTER}.
+
 lookup_ch(ChPid) ->
     case get({ch, ChPid}) of
         undefined -> not_found;
@@ -254,7 +259,7 @@ check_auto_delete(State = #q{round_robin = RoundRobin}) ->
 handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder,
                                    round_robin = ActiveConsumers}) ->
     case lookup_ch(DownPid) of
-        not_found -> {noreply, State};
+        not_found -> noreply(State);
         #cr{monitor_ref = MonitorRef, ch_pid = ChPid, unacked_messages = UAM} ->
             NewActive = block_consumers(ChPid, ActiveConsumers),
             erlang:demonitor(MonitorRef),
@@ -270,7 +275,7 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder,
                                             end,
                        round_robin = NewActive})) of
                 {continue, NewState} ->
-                    {noreply, NewState};
+                    noreply(NewState);
                 {stop, NewState} ->
                     {stop, normal, NewState}
             end
@@ -470,12 +475,12 @@ handle_call({deliver_immediately, Txn, Message}, _From, State) ->
     %% queues discarding the message?
     %%
     {Delivered, NewState} = attempt_delivery(Txn, Message, State),
-    {reply, Delivered, NewState};
+    reply(Delivered, NewState);
 
 handle_call({deliver, Txn, Message}, _From, State) ->
     %% Synchronous, "mandatory" delivery mode
     {Delivered, NewState} = deliver_or_enqueue(Txn, Message, State),
-    {reply, Delivered, NewState};
+    reply(Delivered, NewState);
 
 handle_call({commit, Txn}, From, State) ->
     ok = commit_work(Txn, qname(State)),
@@ -483,7 +488,7 @@ handle_call({commit, Txn}, From, State) ->
     gen_server:reply(From, ok),
     NewState = process_pending(Txn, State),
     erase_tx(Txn),
-    {noreply, NewState};
+    noreply(NewState);
 
 handle_call({notify_down, ChPid}, From, State) ->
     %% optimisation: we reply straight away so the sender can continue
@@ -507,10 +512,11 @@ handle_call({basic_get, ChPid, NoAck}, _From,
                     persist_auto_ack(QName, Message)
             end,
             Msg = {QName, self(), NextId, Delivered, Message},
-            {reply, {ok, queue:len(BufferTail), Msg},
-             State#q{message_buffer = BufferTail, next_msg_id = NextId + 1}};
+            reply({ok, queue:len(BufferTail), Msg},
+                  State#q{message_buffer = BufferTail,
+                          next_msg_id = NextId + 1});
         {empty, _} ->
-            {reply, empty, State}
+            reply(empty, State)
     end;
 
 handle_call({basic_consume, NoAck, ReaderPid, ChPid, ConsumerTag,
@@ -520,11 +526,11 @@ handle_call({basic_consume, NoAck, ReaderPid, ChPid, ConsumerTag,
                               round_robin = RoundRobin}) ->
     case check_queue_owner(Owner, ReaderPid) of
         mismatch ->
-            {reply, {error, queue_owned_by_another_connection}, State};
+            reply({error, queue_owned_by_another_connection}, State);
         ok ->
             case check_exclusive_access(ExistingHolder, ExclusiveConsume) of
                 in_use ->
-                    {reply, {error, exclusive_consume_unavailable}, State};
+                    reply({error, exclusive_consume_unavailable}, State);
                 ok ->
                     C = #cr{consumers = Consumers} = ch_record(ChPid),
                     Consumer = #consumer{tag = ConsumerTag, ack_required = not(NoAck)},
@@ -538,7 +544,7 @@ handle_call({basic_consume, NoAck, ReaderPid, ChPid, ConsumerTag,
                                        end,
                                      round_robin = queue:in({ChPid, Consumer}, RoundRobin)},
                     ok = maybe_send_reply(ChPid, OkMsg),
-                    {reply, ok, run_poke_burst(State1)}
+                    reply(ok, run_poke_burst(State1))
             end
     end;
 
@@ -548,7 +554,7 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
     case lookup_ch(ChPid) of
         not_found ->
             ok = maybe_send_reply(ChPid, OkMsg),
-            {reply, ok, State};
+            reply(ok, State);
         C = #cr{consumers = Consumers} ->
             NewConsumers = lists:filter
                              (fun (#consumer{tag = CT}) -> CT /= ConsumerTag end,
@@ -564,7 +570,7 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
                                                         ConsumerTag,
                                                         RoundRobin)}) of
                 {continue, State1} ->
-                    {reply, ok, State1};
+                    reply(ok, State1);
                 {stop, State1} ->
                     {stop, normal, ok, State1}
             end
@@ -573,7 +579,7 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
 handle_call(stat, _From, State = #q{q = #amqqueue{name = Name},
                                     message_buffer = MessageBuffer,
                                     round_robin = RoundRobin}) ->
-    {reply, {ok, Name, queue:len(MessageBuffer), queue:len(RoundRobin)}, State};
+    reply({ok, Name, queue:len(MessageBuffer), queue:len(RoundRobin)}, State);
 
 handle_call({delete, IfUnused, IfEmpty}, _From,
             State = #q{message_buffer = MessageBuffer}) ->
@@ -581,16 +587,17 @@ handle_call({delete, IfUnused, IfEmpty}, _From,
     IsUnused = is_unused(),
     if
         IfEmpty and not(IsEmpty) ->
-            {reply, {error, not_empty}, State};
+            reply({error, not_empty}, State);
         IfUnused and not(IsUnused) ->
-            {reply, {error, in_use}, State};
+            reply({error, in_use}, State);
         true ->
             {stop, normal, {ok, queue:len(MessageBuffer)}, State}
     end;
 
 handle_call(purge, _From, State = #q{message_buffer = MessageBuffer}) ->
     ok = purge_message_buffer(qname(State), MessageBuffer),
-    {reply, {ok, queue:len(MessageBuffer)}, State#q{message_buffer = queue:new()}};
+    reply({ok, queue:len(MessageBuffer)},
+          State#q{message_buffer = queue:new()});
 
 handle_call({claim_queue, ReaderPid}, _From, State = #q{owner = Owner,
                                                         exclusive_consumer = Holder}) ->
@@ -604,25 +611,25 @@ handle_call({claim_queue, ReaderPid}, _From, State = #q{owner = Owner,
                     %% to check, we'd need to hold not just the ch
                     %% pid for each consumer, but also its reader
                     %% pid...
-                    {reply, locked, State};
+                    reply(locked, State);
                 ok ->
-                    {reply, ok, State#q{owner = {ReaderPid, erlang:monitor(process, ReaderPid)}}}
+                    reply(ok, State#q{owner = {ReaderPid, erlang:monitor(process, ReaderPid)}})
             end;
         {ReaderPid, _MonitorRef} ->
-            {reply, ok, State};
+            reply(ok, State);
         _ ->
-            {reply, locked, State}
+            reply(locked, State)
     end.
 
 handle_cast({deliver, Txn, Message}, State) ->
     %% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
     {_Delivered, NewState} = deliver_or_enqueue(Txn, Message, State),
-    {noreply, NewState};
+    noreply(NewState);
 
 handle_cast({ack, Txn, MsgIds, ChPid}, State) ->
     case lookup_ch(ChPid) of
         not_found ->
-            {noreply, State};
+            noreply(State);
         C = #cr{unacked_messages = UAM} ->
             {Acked, Remaining} = collect_messages(MsgIds, UAM),
             persist_acks(Txn, qname(State), Acked),
@@ -632,37 +639,37 @@ handle_cast({ack, Txn, MsgIds, ChPid}, State) ->
                 _  ->
                     record_pending_acks(Txn, ChPid, MsgIds)
             end,
-            {noreply, State}
+            noreply(State)
     end;
 
 handle_cast({rollback, Txn}, State) ->
     ok = rollback_work(Txn, qname(State)),
     erase_tx(Txn),
-    {noreply, State};
+    noreply(State);
 
 handle_cast({redeliver, Messages}, State) ->
-    {noreply, deliver_or_enqueue_n(Messages, State)};
+    noreply(deliver_or_enqueue_n(Messages, State));
 
 handle_cast({requeue, MsgIds, ChPid}, State) ->
     case lookup_ch(ChPid) of
         not_found ->
             rabbit_log:warning("Ignoring requeue from unknown ch: ~p~n",
                                [ChPid]),
-            {noreply, State};
+            noreply(State);
         C = #cr{unacked_messages = UAM} ->
             {Messages, NewUAM} = collect_messages(MsgIds, UAM),
             store_ch_record(C#cr{unacked_messages = NewUAM}),
-            {noreply, deliver_or_enqueue_n(
-                        [{Message, true} || Message <- Messages], State)}
+            noreply(deliver_or_enqueue_n(
+                      [{Message, true} || Message <- Messages], State))
     end;
 
 handle_cast({notify_sent, ChPid}, State) ->
     case lookup_ch(ChPid) of
-        not_found -> {noreply, State};
+        not_found -> noreply(State);
         T = #cr{unsent_message_count =Count} ->
-            {noreply, possibly_unblock(
-                        T#cr{unsent_message_count = Count - 1},
-                        State)}
+            noreply(possibly_unblock(
+                      T#cr{unsent_message_count = Count - 1},
+                      State))
     end.
 
 handle_info({'DOWN', MonitorRef, process, DownPid, _Reason},
@@ -681,6 +688,9 @@ handle_info({'DOWN', MonitorRef, process, DownPid, _Reason},
 handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) ->
     handle_ch_down(DownPid, State);
 
+handle_info(timeout, State) ->
+    {noreply, State, hibernate};
+
 handle_info(Info, State) ->
     ?LOGDEBUG("Info in queue: ~p~n", [Info]),
     {stop, {unhandled_info, Info}, State}.
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 0544d32e6a..b4e0fbab37 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -28,7 +28,7 @@
 -include("rabbit.hrl").
 
 -export([start_link/4, do/2, do/3, shutdown/1]).
--export([send_command/2, deliver/4]).
+-export([send_command/2, deliver/4, conserve_memory/2]).
 
 %% callbacks
 -export([init/2, handle_message/2]).
@@ -49,6 +49,7 @@
 -spec(shutdown/1 :: (pid()) -> 'ok').
 -spec(send_command/2 :: (pid(), amqp_method()) -> 'ok').
 -spec(deliver/4 :: (pid(), ctag(), bool(), msg()) -> 'ok').
+-spec(conserve_memory/2 :: (pid(), bool()) -> 'ok').
 
 -endif.
 
@@ -77,11 +78,18 @@ deliver(Pid, ConsumerTag, AckRequired, Msg) ->
     Pid ! {deliver, ConsumerTag, AckRequired, Msg},
     ok.
 
+conserve_memory(Pid, Conserve) ->
+    Pid ! {conserve_memory, Conserve},
+    ok.
+
 %%---------------------------------------------------------------------------
 
 init(ProxyPid, [ReaderPid, WriterPid, Username, VHost]) ->
     process_flag(trap_exit, true),
     link(WriterPid),
+    %% this is bypassing the proxy so alarms can "jump the queue" and
+    %% be handled promptly
+    rabbit_alarm:register(self(), {?MODULE, conserve_memory, []}),
     #ch{state                   = starting,
         proxy_pid               = ProxyPid,
         reader_pid              = ReaderPid,
@@ -129,6 +137,11 @@ handle_message({deliver, ConsumerTag, AckRequired, Msg},
                           true, ConsumerTag, DeliveryTag, Msg),
     State1#ch{next_tag = DeliveryTag + 1};
 
+handle_message({conserve_memory, Conserve}, State) ->
+    ok = rabbit_writer:send_command(
+           State#ch.writer_pid, #'channel.flow'{active = not(Conserve)}),
+    State;
+
 handle_message({'EXIT', _Pid, Reason}, State) ->
     terminate(Reason, State);
 
@@ -619,6 +632,12 @@ handle_method(#'channel.flow'{active = _}, _, State) ->
     %% FIXME: implement
     {reply, #'channel.flow_ok'{active = true}, State};
 
+handle_method(#'channel.flow_ok'{active = _}, _, State) ->
+    %% TODO: We may want to correlate this to channel.flow messages we
+    %% have sent, and complain if we get an unsolicited
+    %% channel.flow_ok, or the client refuses our flow request.
+    {noreply, State};
+
 handle_method(_MethodRecord, _Content, _State) ->
     rabbit_misc:protocol_error(
       command_invalid, "unimplemented method", []).
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 2c7fa2ab90..a2688625be 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -36,6 +36,8 @@
 
 -record(wstate, {sock, channel, frame_max}).
 
+-define(HIBERNATE_AFTER, 5000).
+
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
@@ -63,6 +65,8 @@ start(Sock, Channel, FrameMax) ->
 mainloop(State) ->
     receive
         Message -> ?MODULE:mainloop(handle_message(Message, State))
+    after ?HIBERNATE_AFTER ->
+            erlang:hibernate(?MODULE, mainloop, [State])
     end.
 
 handle_message({send_command, MethodRecord},