merge default into bug19684

and adapt the new rabbit_control functions to use Inform

16 files changed, 355 insertions, 113 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index a33c5b7bcb..195dd729bd 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -57,8 +57,8 @@
 -spec(rotate_logs/1 :: (file_suffix()) -> 'ok' | {'error', any()}).
 -spec(status/0 :: () ->
              [{running_applications, [{atom(), string(), string()}]} |
-              {nodes, [node()]} |
-              {running_nodes, [node()]}]).
+              {nodes, [erlang_node()]} |
+              {running_nodes, [erlang_node()]}]).
 -spec(log_location/1 :: ('sasl' | 'kernel') -> log_location()).
 
 -endif.
diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl
index d9c1c45042..6d65b3a407 100644
--- a/src/rabbit_alarm.erl
+++ b/src/rabbit_alarm.erl
@@ -50,22 +50,57 @@
 %%----------------------------------------------------------------------------
 
 start() ->
+    ok = alarm_handler:add_alarm_handler(?MODULE),
+    case whereis(memsup) of
+        undefined ->
+            Mod = case os:type() of 
+                      %% memsup doesn't take account of buffers or
+                      %% cache when considering "free" memory -
+                      %% therefore on Linux we can get memory alarms
+                      %% very easily without any pressure existing on
+                      %% memory at all. Therefore we need to use our
+                      %% own simple memory monitor.
+                      %%
+                      {unix, linux} -> rabbit_memsup_linux;
+
+                      %% Start memsup programmatically rather than via
+                      %% the rabbitmq-server script. This is not quite
+                      %% the right thing to do as os_mon checks to see
+                      %% if memsup is available before starting it,
+                      %% but as memsup is available everywhere (even
+                      %% on VXWorks) it should be ok.
+                      %%
+                      %% One benefit of the programmatic startup is
+                      %% that we can add our alarm_handler before
+                      %% memsup is running, thus ensuring that we
+                      %% notice memory alarms that go off on startup.
+                      %%
+                      _             -> memsup
+                  end,
+            %% This is based on os_mon:childspec(memsup, true)
+            {ok, _} = supervisor:start_child(
+                        os_mon_sup,
+                        {memsup, {Mod, start_link, []},
+                         permanent, 2000, worker, [Mod]}),
+            ok;
+        _ ->
+            ok
+    end,
     %% 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
+    %% 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).
+    ok = os_mon:call(memsup,
+                     {set_check_interval, ?MEMSUP_CHECK_INTERVAL},
+                     infinity).
 
 stop() ->
     ok = alarm_handler:delete_alarm_handler(?MODULE).
@@ -118,7 +153,7 @@ 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]),
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index e2174e1100..694e525e07 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -96,7 +96,7 @@
 -spec(basic_cancel/4 :: (amqqueue(), pid(), ctag(), any()) -> 'ok').
 -spec(notify_sent/2 :: (pid(), pid()) -> 'ok').
 -spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()).
--spec(on_node_down/1 :: (node()) -> 'ok').
+-spec(on_node_down/1 :: (erlang_node()) -> 'ok').
 -spec(pseudo_queue/2 :: (binary(), pid()) -> amqqueue()).
 
 -endif.
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index a9addd5293..186f43898a 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -749,7 +749,9 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) ->
     handle_ch_down(DownPid, State);
 
 handle_info(timeout, State) ->
-    {noreply, State, hibernate};
+    %% TODO: Once we drop support for R11B-5, we can change this to
+    %% {noreply, State, hibernate};
+    proc_lib:hibernate(gen_server, enter_loop, [?MODULE, [], State]);
 
 handle_info(Info, State) ->
     ?LOGDEBUG("Info in queue: ~p~n", [Info]),
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index d4906d8f1e..ae346236e1 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -26,43 +26,67 @@
 -module(rabbit_control).
 -include("rabbit.hrl").
 
--export([start/0, stop/0, action/3]).
+-export([start/0, stop/0, action/4]).
+
+-record(params, {quiet, node, command, args}).
 
 -define(RPC_TIMEOUT, 30000).
 
 start() ->
-    case init:get_plain_arguments() of
-        [] ->
+    FullCommand = init:get_plain_arguments(),
+    #params{quiet = Quiet, node = Node, command = Command, args = Args} = 
+        parse_args(FullCommand, #params{quiet = false,
+                                        node = rabbit_misc:localnode(rabbit)}),
+    Inform = case Quiet of
+                 true  -> fun(_Format, _Args1) -> ok end;
+                 false -> fun(Format, Args1) ->
+                                  io:format(Format ++ " ...~n", Args1)
+                         end
+             end,
+    %% The reason we don't use a try/catch here is that rpc:call turns
+    %% thrown errors into normal return values
+    case catch action(Command, Node, Args, Inform) of
+        ok ->
+            case Quiet of
+                true  -> ok;
+                false -> io:format("...done.~n")
+            end,
+            init:stop();
+        {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
+            error("invalid command '~s'",
+                  [lists:flatten(
+                     rabbit_misc:intersperse(
+                       " ", [atom_to_list(Command) | Args]))]),
             usage();
-        FullCommand ->
-            {Node, Command, Args} = parse_args(FullCommand),
-            case catch action(Command, Node, Args) of
-                ok ->
-                    io:format("done.~n"),
-                    init:stop();
-                {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
-                    io:format("Invalid command ~p~n", [FullCommand]),
-                    usage();
-                Other ->
-                    io:format("~nrabbit_control action ~p failed:~n~p~n", [Command, Other]),
-                    halt(2)
-            end
+        {error, Reason} ->
+            error("~p", [Reason]),
+            halt(2);
+        Other ->
+            error("~p", [Other]),
+            halt(2)
     end.
 
-parse_args(["-n", NodeS, Command | Args]) ->
+error(Format, Args) ->
+    rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args).
+
+parse_args(["-n", NodeS | Args], Params) ->
     Node = case lists:member($@, NodeS) of
                true  -> list_to_atom(NodeS);
                false -> rabbit_misc:localnode(list_to_atom(NodeS))
            end,
-    {Node, list_to_atom(Command), Args};
-parse_args([Command | Args]) ->
-    {rabbit_misc:localnode(rabbit), list_to_atom(Command), Args}.
+    parse_args(Args, Params#params{node = Node});
+parse_args(["-q" | Args], Params) ->
+    parse_args(Args, Params#params{quiet = true});
+parse_args([Command | Args], Params) ->
+    Params#params{command = list_to_atom(Command), args = Args};
+parse_args([], _) ->
+    usage().
 
 stop() ->
     ok.
 
 usage() ->
-    io:format("Usage: rabbitmqctl [-n <node>] <command> [<arg> ...]
+    io:format("Usage: rabbitmqctl [-q] [-n <node>] <command> [<arg> ...]
 
 Available commands:
 
@@ -94,6 +118,9 @@ Available commands:
   list_bindings
   list_connections <ConnectionInfoItem> [<ConnectionInfoItem> ...]
 
+Quiet output mode is selected with the \"-q\" flag. Informational messages
+are suppressed when quiet mode is in effect.
+
 <node> should be the name of the master node of the RabbitMQ cluster. It
 defaults to the node named \"rabbit\" on the local host. On a host named
 \"server.example.com\", the master node will usually be rabbit@server (unless
@@ -120,101 +147,101 @@ user, peer_address and peer_port.
 "),
     halt(1).
 
-action(stop, Node, []) ->
-    io:format("Stopping and halting node ~p ...", [Node]),
+action(stop, Node, [], Inform) ->
+    Inform("Stopping and halting node ~p", [Node]),
     call(Node, {rabbit, stop_and_halt, []});
 
-action(stop_app, Node, []) ->
-    io:format("Stopping node ~p ...", [Node]),
+action(stop_app, Node, [], Inform) ->
+    Inform("Stopping node ~p", [Node]),
     call(Node, {rabbit, stop, []});
 
-action(start_app, Node, []) ->
-    io:format("Starting node ~p ...", [Node]),
+action(start_app, Node, [], Inform) ->
+    Inform("Starting node ~p", [Node]),
     call(Node, {rabbit, start, []});
 
-action(reset, Node, []) ->
-    io:format("Resetting node ~p ...", [Node]),
+action(reset, Node, [], Inform) ->
+    Inform("Resetting node ~p", [Node]),
     call(Node, {rabbit_mnesia, reset, []});
 
-action(force_reset, Node, []) ->
-    io:format("Forcefully resetting node ~p ...", [Node]),
+action(force_reset, Node, [], Inform) ->
+    Inform("Forcefully resetting node ~p", [Node]),
     call(Node, {rabbit_mnesia, force_reset, []});
 
-action(cluster, Node, ClusterNodeSs) ->
+action(cluster, Node, ClusterNodeSs, Inform) ->
     ClusterNodes = lists:map(fun list_to_atom/1, ClusterNodeSs),
-    io:format("Clustering node ~p with ~p ...",
+    Inform("Clustering node ~p with ~p",
               [Node, ClusterNodes]),
     rpc_call(Node, rabbit_mnesia, cluster, [ClusterNodes]);
 
-action(status, Node, []) ->
-    io:format("Status of node ~p ...", [Node]),
+action(status, Node, [], Inform) ->
+    Inform("Status of node ~p", [Node]),
     Res = call(Node, {rabbit, status, []}),
-    io:format("~n~p~n", [Res]),
+    io:format("~p~n", [Res]),
     ok;
 
-action(rotate_logs, Node, []) ->
-    io:format("Reopening logs for node ~p ...", [Node]),
+action(rotate_logs, Node, [], Inform) ->
+    Inform("Reopening logs for node ~p", [Node]),
     call(Node, {rabbit, rotate_logs, [""]});
-action(rotate_logs, Node, Args = [Suffix]) ->
-    io:format("Rotating logs to files with suffix ~p ...", [Suffix]),
+action(rotate_logs, Node, Args = [Suffix], Inform) ->
+    Inform("Rotating logs to files with suffix ~p", [Suffix]),
     call(Node, {rabbit, rotate_logs, Args});
 
-action(add_user, Node, Args = [Username, _Password]) ->
-    io:format("Creating user ~p ...", [Username]),
+action(add_user, Node, Args = [Username, _Password], Inform) ->
+    Inform("Creating user ~p", [Username]),
     call(Node, {rabbit_access_control, add_user, Args});
 
-action(delete_user, Node, Args = [_Username]) ->
-    io:format("Deleting user ~p ...", Args),
+action(delete_user, Node, Args = [_Username], Inform) ->
+    Inform("Deleting user ~p", Args),
     call(Node, {rabbit_access_control, delete_user, Args});
 
-action(change_password, Node, Args = [Username, _Newpassword]) ->
-    io:format("Changing password for user ~p ...", [Username]),
+action(change_password, Node, Args = [Username, _Newpassword], Inform) ->
+    Inform("Changing password for user ~p", [Username]),
     call(Node, {rabbit_access_control, change_password, Args});
 
-action(list_users, Node, []) ->
-    io:format("Listing users ..."),
+action(list_users, Node, [], Inform) ->
+    Inform("Listing users", []),
     display_list(call(Node, {rabbit_access_control, list_users, []}));
 
-action(add_vhost, Node, Args = [_VHostPath]) ->
-    io:format("Creating vhost ~p ...", Args),
+action(add_vhost, Node, Args = [_VHostPath], Inform) ->
+    Inform("Creating vhost ~p", Args),
     call(Node, {rabbit_access_control, add_vhost, Args});
 
-action(delete_vhost, Node, Args = [_VHostPath]) ->
-    io:format("Deleting vhost ~p ...", Args),
+action(delete_vhost, Node, Args = [_VHostPath], Inform) ->
+    Inform("Deleting vhost ~p", Args),
     call(Node, {rabbit_access_control, delete_vhost, Args});
 
-action(list_vhosts, Node, []) ->
-    io:format("Listing vhosts ..."),
+action(list_vhosts, Node, [], Inform) ->
+    Inform("Listing vhosts", []),
     display_list(call(Node, {rabbit_access_control, list_vhosts, []}));
 
-action(map_user_vhost, Node, Args = [_Username, _VHostPath]) ->
-    io:format("Mapping user ~p to vhost ~p ...", Args),
+action(map_user_vhost, Node, Args = [_Username, _VHostPath], Inform) ->
+    Inform("Mapping user ~p to vhost ~p", Args),
     call(Node, {rabbit_access_control, map_user_vhost, Args});
 
-action(unmap_user_vhost, Node, Args = [_Username, _VHostPath]) ->
-    io:format("Unmapping user ~p from vhost ~p ...", Args),
+action(unmap_user_vhost, Node, Args = [_Username, _VHostPath], Inform) ->
+    Inform("Unmapping user ~p from vhost ~p", Args),
     call(Node, {rabbit_access_control, unmap_user_vhost, Args});
 
-action(list_user_vhosts, Node, Args = [_Username]) ->
-    io:format("Listing vhosts for user ~p...", Args),
+action(list_user_vhosts, Node, Args = [_Username], Inform) ->
+    Inform("Listing vhosts for user ~p", Args),
     display_list(call(Node, {rabbit_access_control, list_user_vhosts, Args}));
 
-action(list_vhost_users, Node, Args = [_VHostPath]) ->
-    io:format("Listing users for vhosts ~p...", Args),
+action(list_vhost_users, Node, Args = [_VHostPath], Inform) ->
+    Inform("Listing users for vhosts ~p", Args),
     display_list(call(Node, {rabbit_access_control, list_vhost_users, Args}));
 
-action(list_queues, Node, Args) ->
-    io:format("Listing queues ...~n"),
+action(list_queues, Node, Args, Inform) ->
+    Inform("Listing queues", []),
     ArgAtoms = [list_to_atom(X) || X <- default_if_empty(Args, ["name", "messages"])],
     display_info_list(rpc_call(Node, rabbit_amqqueue, info_all, [ArgAtoms]), ArgAtoms);
 
-action(list_exchanges, Node, Args) ->
-    io:format("Listing exchanges ...~n"),
+action(list_exchanges, Node, Args, Inform) ->
+    Inform("Listing exchanges", []),
     ArgAtoms = [list_to_atom(X) || X <- default_if_empty(Args, ["name", "type"])],
     display_info_list(rpc_call(Node, rabbit_exchange, info_all, [ArgAtoms]), ArgAtoms);
 
-action(list_bindings, Node, []) ->
-    io:format("Listing bindings ...~n"),
+action(list_bindings, Node, [], Inform) ->
+    Inform("Listing bindings", []),
     lists:map(
         fun({#resource{name = ExchangeName, virtual_host = VirtualHost}, 
              #resource{name = QueueName, virtual_host = VirtualHost},
@@ -225,8 +252,8 @@ action(list_bindings, Node, []) ->
         rpc_call(Node, rabbit_exchange, list_bindings, [])),
     ok;
 
-action(list_connections, Node, Args) ->
-    io:format("Listing connections ...~n"),
+action(list_connections, Node, Args, Inform) ->
+    Inform("Listing connections", []),
     ArgAtoms = [list_to_atom(X) || X <- default_if_empty(Args, ["user", "peer_address", "peer_port"])],
     display_info_list(rpc_call(Node, rabbit_networking, connection_info_all, [ArgAtoms]), ArgAtoms).
 
@@ -261,10 +288,10 @@ display_info_list(Results, InfoItemArgs) when is_list(Results) ->
 
 display_list(L) when is_list(L) ->
     lists:foreach(fun (I) ->
-                          io:format("~n~s", [binary_to_list(I)])
+                          io:format("~s~n", [binary_to_list(I)])
                   end,
                   lists:sort(L)),
-    io:nl();
+    ok;
 display_list(Other) -> Other.
 
 call(Node, {Mod, Fun, Args}) ->
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 273cf3d380..9a3caad107 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -302,9 +302,17 @@ exchanges_for_queue(QueueName) ->
 
 has_bindings(ExchangeName) ->
     MatchHead = #route{binding = #binding{exchange_name = ExchangeName,
-                                          queue_name = '$1',
                                           _ = '_'}},
-    continue(mnesia:select(route, [{MatchHead, [], ['$1']}], 1, read)).
+    try
+        continue(mnesia:select(route, [{MatchHead, [], ['$_']}], 1, read))
+    catch exit:{aborted, {badarg, _}} ->
+            %% work around OTP-7025, which was fixed in R12B-1, by
+            %% falling back on a less efficient method
+            case mnesia:match_object(MatchHead) of
+                []    -> false;
+                [_|_] -> true
+            end
+    end.
 
 continue('$end_of_table')    -> false;
 continue({[_|_], _})         -> true;
diff --git a/src/rabbit_heartbeat.erl b/src/rabbit_heartbeat.erl
index 6e825578d3..b05a4655e5 100644
--- a/src/rabbit_heartbeat.erl
+++ b/src/rabbit_heartbeat.erl
@@ -67,7 +67,7 @@ heartbeater(Sock, TimeoutMillisec, StatName, Threshold, Handler, MonitorRef) ->
                             {'DOWN', MonitorRef, process, _Object, _Info} -> ok;
                             Other -> exit({unexpected_message, Other})
                         after TimeoutMillisec ->
-                                case prim_inet:getstat(Sock, [StatName]) of
+                                case inet:getstat(Sock, [StatName]) of
                                     {ok, [{StatName, NewStatVal}]} ->
                                         if NewStatVal =/= StatVal ->
                                                 F({NewStatVal, 0});
diff --git a/src/rabbit_load.erl b/src/rabbit_load.erl
index 8deec8ebee..2154bf72ae 100644
--- a/src/rabbit_load.erl
+++ b/src/rabbit_load.erl
@@ -34,11 +34,11 @@
 
 -ifdef(use_specs).
 
--type(node() :: atom()).
--type(load() :: {{non_neg_integer(), float()}, node()}).
+-type(erlang_node() :: atom()).
+-type(load() :: {{non_neg_integer(), float()}, erlang_node()}).
 -spec(local_load/0 :: () -> load()).
 -spec(remote_loads/0 :: () -> [load()]).
--spec(pick/0 :: () -> node()).
+-spec(pick/0 :: () -> erlang_node()).
 
 -endif.
 
diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl
index a8f839f0a6..b4729230e6 100644
--- a/src/rabbit_log.erl
+++ b/src/rabbit_log.erl
@@ -32,7 +32,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--export([debug/1, debug/2, info/1, info/2,
+-export([debug/1, debug/2, message/4, info/1, info/2,
          warning/1, warning/2, error/1, error/2]).
 
 -import(io).
@@ -67,6 +67,10 @@ debug(Fmt) ->
 debug(Fmt, Args) when is_list(Args) ->
     gen_server:cast(?SERVER, {debug, Fmt, Args}).
 
+message(Direction, Channel, MethodRecord, Content) ->
+    gen_server:cast(?SERVER,
+		    {message, Direction, Channel, MethodRecord, Content}).
+
 info(Fmt) ->
     gen_server:cast(?SERVER, {info, Fmt}).
 
@@ -100,6 +104,14 @@ handle_cast({debug, Fmt, Args}, State) ->
     io:format("debug:: "), io:format(Fmt, Args),
     error_logger:info_msg("debug:: " ++ Fmt, Args),
     {noreply, State};
+handle_cast({message, Direction, Channel, MethodRecord, Content}, State) ->
+    io:format("~s ch~p ~p~n",
+	      [case Direction of
+		   in -> "-->";
+		   out -> "<--" end,
+	       Channel,
+	       {MethodRecord, Content}]),
+    {noreply, State};
 handle_cast({info, Fmt}, State) ->
     error_logger:info_msg(Fmt),
     {noreply, State};
diff --git a/src/rabbit_memsup_linux.erl b/src/rabbit_memsup_linux.erl
new file mode 100644
index 0000000000..b77ffcabb4
--- /dev/null
+++ b/src/rabbit_memsup_linux.erl
@@ -0,0 +1,150 @@
+%%   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_memsup_linux).
+
+-behaviour(gen_server).
+
+-export([start_link/0]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-export([update/0]).
+
+-define(SERVER, memsup). %% must be the same as the standard memsup
+
+-define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000).
+
+-record(state, {memory_fraction, alarmed, timeout, timer}).   
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}).
+-spec(update/0 :: () -> 'ok').
+     
+-endif.
+
+%%----------------------------------------------------------------------------
+
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+
+update() ->
+    gen_server:cast(?SERVER, update).
+
+%%----------------------------------------------------------------------------
+
+init(_Args) -> 
+    Fraction = os_mon:get_env(memsup, system_memory_high_watermark),
+    TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL),
+    {ok, #state{alarmed = false, 
+                memory_fraction = Fraction, 
+                timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL,
+                timer = TRef}}.
+
+start_timer(Timeout) ->
+    {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []),
+    TRef.
+
+%% Export the same API as the real memsup. Note that
+%% get_sysmem_high_watermark gives an int in the range 0 - 100, while
+%% set_sysmem_high_watermark takes a float in the range 0.0 - 1.0.
+handle_call(get_sysmem_high_watermark, _From, State) ->
+    {reply, trunc(100 * State#state.memory_fraction), State};
+
+handle_call({set_sysmem_high_watermark, Float}, _From, State) ->
+    {reply, ok, State#state{memory_fraction = Float}};
+
+handle_call(get_check_interval, _From, State) ->
+    {reply, State#state.timeout, State};
+
+handle_call({set_check_interval, Timeout}, _From, State) ->
+    {ok, cancel} = timer:cancel(State#state.timer),
+    {reply, ok, State#state{timeout = Timeout, timer = start_timer(Timeout)}};
+
+handle_call(_Request, _From, State) -> 
+    {noreply, State}.
+
+handle_cast(update, State = #state{alarmed = Alarmed,
+                                   memory_fraction = MemoryFraction}) -> 
+    File = read_proc_file("/proc/meminfo"),
+    Lines = string:tokens(File, "\n"),
+    Dict = dict:from_list(lists:map(fun parse_line/1, Lines)),
+    MemTotal = dict:fetch('MemTotal', Dict),
+    MemUsed = MemTotal
+        - dict:fetch('MemFree', Dict)
+        - dict:fetch('Buffers', Dict)
+        - dict:fetch('Cached', Dict),
+    NewAlarmed = MemUsed / MemTotal > MemoryFraction,
+    case {Alarmed, NewAlarmed} of
+        {false, true} ->
+            alarm_handler:set_alarm({system_memory_high_watermark, []});
+        {true, false} ->
+            alarm_handler:clear_alarm(system_memory_high_watermark);
+        _ ->
+            ok
+    end,
+    {noreply,  State#state{alarmed = NewAlarmed}};
+
+handle_cast(_Request, State) -> 
+    {noreply, State}.
+
+handle_info(_Info, State) -> 
+    {noreply, State}.
+
+terminate(_Reason, _State) -> 
+    ok.
+
+code_change(_OldVsn, State, _Extra) -> 
+    {ok, State}.
+
+%%----------------------------------------------------------------------------
+
+-define(BUFFER_SIZE, 1024).
+
+%% file:read_file does not work on files in /proc as it seems to get
+%% the size of the file first and then read that many bytes. But files
+%% in /proc always have length 0, we just have to read until we get
+%% eof.
+read_proc_file(File) ->
+    {ok, IoDevice} = file:open(File, [read, raw]),
+    Res = read_proc_file(IoDevice, []),
+    file:close(IoDevice),
+    lists:flatten(lists:reverse(Res)).
+    
+read_proc_file(IoDevice, Acc) ->
+    case file:read(IoDevice, ?BUFFER_SIZE) of
+        {ok, Res} -> read_proc_file(IoDevice, [Res | Acc]);
+        eof       -> Acc
+    end.
+
+%% A line looks like "FooBar: 123456 kB"
+parse_line(Line) ->
+    [Name, Value | _] = string:tokens(Line, ": "),   
+    {list_to_atom(Name), list_to_integer(Value)}.
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 3c8da720ff..18570171f7 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -43,6 +43,7 @@
 -export([guid/0, string_guid/1, binstring_guid/1]).
 -export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]).
 -export([append_file/2]).
+-export([format_stderr/2]).
 
 -import(mnesia).
 -import(lists).
@@ -86,7 +87,7 @@
 -spec(with_user_and_vhost/3 :: (username(), vhost(), thunk(A)) -> A).
 -spec(execute_mnesia_transaction/1 :: (thunk(A)) -> A).
 -spec(ensure_ok/2 :: ('ok' | {'error', any()}, atom()) -> 'ok').
--spec(localnode/1 :: (atom()) -> node()).
+-spec(localnode/1 :: (atom()) -> erlang_node()).
 -spec(tcp_name/3 :: (atom(), ip_address(), ip_port()) -> atom()).
 -spec(intersperse/2 :: (A, [A]) -> [A]).
 -spec(upmap/2 :: (fun ((A) -> B), [A]) -> [B]).
@@ -99,6 +100,7 @@
              'ok' | 'aborted').
 -spec(dirty_dump_log/1 :: (string()) -> 'ok' | {'error', any()}).
 -spec(append_file/2 :: (string(), string()) -> 'ok' | {'error', any()}).
+-spec(format_stderr/2 :: (string(), [any()]) -> 'true').
 
 -endif.
 
@@ -377,3 +379,8 @@ append_file(File, _, Suffix) ->
         {ok, Data} -> file:write_file([File, Suffix], Data, [append]);
         Error      -> Error
     end.
+
+format_stderr(Fmt, Args) ->
+    Port = open_port({fd, 0, 2}, [out]),
+    port_command(Port, io_lib:format(Fmt, Args)),
+    port_close(Port).
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 9b67135def..8d34d2855b 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -40,11 +40,11 @@
 
 -ifdef(use_specs).
 
--spec(status/0 :: () -> [{'nodes' | 'running_nodes', [node()]}]).
+-spec(status/0 :: () -> [{'nodes' | 'running_nodes', [erlang_node()]}]).
 -spec(ensure_mnesia_dir/0 :: () -> 'ok').
 -spec(init/0 :: () -> 'ok').
 -spec(is_db_empty/0 :: () -> bool()).
--spec(cluster/1 :: ([node()]) -> 'ok').
+-spec(cluster/1 :: ([erlang_node()]) -> 'ok').
 -spec(reset/0 :: () -> 'ok').
 -spec(force_reset/0 :: () -> 'ok').
 -spec(create_tables/0 :: () -> 'ok').
diff --git a/src/rabbit_multi.erl b/src/rabbit_multi.erl
index c6a7e9200d..b99dfbc1c1 100644
--- a/src/rabbit_multi.erl
+++ b/src/rabbit_multi.erl
@@ -46,18 +46,22 @@ start() ->
                     io:format("done.~n"),
                     init:stop();
                 {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
-                    io:format("Invalid command ~p~n", [FullCommand]),
+                    error("invalid command '~s'",
+                          [lists:flatten(
+                             rabbit_misc:intersperse(" ", FullCommand))]),
                     usage();
                 timeout ->
-                    io:format("timeout starting some nodes.~n"),
+                    error("timeout starting some nodes.", []),
                     halt(1);
                 Other ->
-                    io:format("~nrabbit_multi action ~p failed:~n~p~n",
-                              [Command, Other]),
+                    error("~p", [Other]),
                     halt(2)
             end
     end.
 
+error(Format, Args) ->
+    rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args).
+
 parse_args([Command | Args]) ->
     {list_to_atom(Command), Args}.
 
@@ -223,8 +227,7 @@ write_pids_file(Pids) ->
                  {ok, Device} ->
                      Device;
                  {error, Reason} ->
-                     throw({error, {cannot_create_pids_file,
-                                    FileName, Reason}})
+                     throw({cannot_create_pids_file, FileName, Reason})
              end,
     try
         ok = io:write(Handle, Pids),
@@ -233,8 +236,7 @@ write_pids_file(Pids) ->
         case file:close(Handle) of
             ok -> ok;
             {error, Reason1} ->
-                throw({error, {cannot_create_pids_file,
-                               FileName, Reason1}})
+                throw({cannot_create_pids_file, FileName, Reason1})
         end
     end,
     ok.
@@ -244,8 +246,7 @@ delete_pids_file() ->
     case file:delete(FileName) of
         ok              -> ok;
         {error, enoent} -> ok;
-        {error, Reason} -> throw({error, {cannot_delete_pids_file,
-                                          FileName, Reason}})
+        {error, Reason} -> throw({cannot_delete_pids_file, FileName, Reason})
     end.
 
 read_pids_file() ->
@@ -253,8 +254,7 @@ read_pids_file() ->
     case file:consult(FileName) of
         {ok, [Pids]}    -> Pids;
         {error, enoent} -> [];
-        {error, Reason} -> throw({error, {cannot_read_pids_file,
-                                          FileName, Reason}})
+        {error, Reason} -> throw({cannot_read_pids_file, FileName, Reason})
     end.
 
 kill_wait(Pid, TimeLeft, Forceful) when TimeLeft < 0 ->
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 2ed57d7783..04d4c7d453 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -48,13 +48,13 @@
 -spec(start_tcp_listener/2 :: (host(), ip_port()) -> 'ok').
 -spec(stop_tcp_listener/2 :: (host(), ip_port()) -> 'ok').
 -spec(active_listeners/0 :: () -> [listener()]).
--spec(node_listeners/1 :: (node()) -> [listener()]).
+-spec(node_listeners/1 :: (erlang_node()) -> [listener()]).
 -spec(connections/0 :: () -> [connection()]).
 -spec(connection_info/1 :: (connection()) -> [info()]).
 -spec(connection_info/2 :: (connection(), [info_key()]) -> [info()]).
 -spec(connection_info_all/0 :: () -> [[info()]]).
 -spec(connection_info_all/1 :: ([info_key()]) -> [[info()]]).
--spec(on_node_down/1 :: (node()) -> 'ok').
+-spec(on_node_down/1 :: (erlang_node()) -> 'ok').
 -spec(check_tcp_listener_address/3 :: (atom(), host(), ip_port()) ->
              {ip_address(), atom()}).
 
diff --git a/src/rabbit_sasl_report_file_h.erl b/src/rabbit_sasl_report_file_h.erl
index 3374d63d0b..10e2ee6eb9 100644
--- a/src/rabbit_sasl_report_file_h.erl
+++ b/src/rabbit_sasl_report_file_h.erl
@@ -71,8 +71,9 @@ handle_call(Event, State) ->
 terminate(Reason, State) ->
     sasl_report_file_h:terminate(Reason, State).
 
-code_change(OldVsn, State, Extra) ->
-    sasl_report_file_h:code_change(OldVsn, State, Extra).
+code_change(_OldVsn, State, _Extra) ->
+    %% There is no sasl_report_file_h:code_change/3
+    {ok, State}.
 
 %%----------------------------------------------------------------------
 
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index fff02d7363..db78bbcc50 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -465,7 +465,7 @@ test_user_management() ->
 control_action(Command, Args) -> control_action(Command, node(), Args).
 
 control_action(Command, Node, Args) ->
-    case catch rabbit_control:action(Command, Node, Args) of
+    case catch rabbit_control:action(Command, Node, Args, fun io:format/2) of
         ok ->
             io:format("done.~n"),
             ok;