diff options
author | Ben Hood <0x6e6562@gmail.com> | 2009-11-05 14:43:08 +0000 |
---|---|---|
committer | Ben Hood <0x6e6562@gmail.com> | 2009-11-05 14:43:08 +0000 |
commit | 1dd1714e10a70ac0765f3062e79d23e00ade2000 (patch) | |
tree | 700eab5983fbcdde0e3807f11b9aa069233d0e73 | |
parent | d79eb60d0c74d3591a2b470a44eca0b09e540b2f (diff) | |
parent | 3e86a1b9881957309d432c8b2e5f295a07f8f920 (diff) | |
download | rabbitmq-server-1dd1714e10a70ac0765f3062e79d23e00ade2000.tar.gz |
Merged default into bug21775bug21775
-rw-r--r-- | Makefile | 2 | ||||
-rw-r--r-- | ebin/rabbit_app.in | 3 | ||||
-rw-r--r-- | generate_app | 6 | ||||
-rw-r--r-- | scripts/rabbitmq-activate-plugins.bat | 4 | ||||
-rw-r--r-- | scripts/rabbitmq-deactivate-plugins.bat | 4 | ||||
-rwxr-xr-x | scripts/rabbitmq-multi.bat | 12 | ||||
-rwxr-xr-x | scripts/rabbitmq-server | 4 | ||||
-rwxr-xr-x | scripts/rabbitmq-server.bat | 8 | ||||
-rwxr-xr-x | scripts/rabbitmq-service.bat | 14 | ||||
-rwxr-xr-x | scripts/rabbitmqctl.bat | 4 | ||||
-rw-r--r-- | src/gen_server2.erl | 14 | ||||
-rw-r--r-- | src/rabbit.erl | 21 | ||||
-rw-r--r-- | src/rabbit_alarm.erl | 110 | ||||
-rw-r--r-- | src/rabbit_memsup.erl | 142 | ||||
-rw-r--r-- | src/rabbit_memsup_darwin.erl | 88 | ||||
-rw-r--r-- | src/rabbit_memsup_linux.erl | 101 | ||||
-rw-r--r-- | src/rabbit_mnesia.erl | 20 | ||||
-rw-r--r-- | src/rabbit_plugin_activator.erl | 11 | ||||
-rw-r--r-- | src/vm_memory_monitor.erl | 322 |
19 files changed, 443 insertions, 447 deletions
@@ -46,7 +46,7 @@ ERL_EBIN=erl -noinput -pa $(EBIN_DIR) all: $(TARGETS) $(EBIN_DIR)/rabbit.app: $(EBIN_DIR)/rabbit_app.in $(BEAM_TARGETS) generate_app - escript generate_app $(EBIN_DIR) < $< > $@ + escript generate_app $(EBIN_DIR) $@ < $< $(EBIN_DIR)/gen_server2.beam: $(SOURCE_DIR)/gen_server2.erl erlc $(ERLC_OPTS) $< diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in index dd907d1a..39f98cbe 100644 --- a/ebin/rabbit_app.in +++ b/ebin/rabbit_app.in @@ -20,5 +20,4 @@ {default_user, <<"guest">>}, {default_pass, <<"guest">>}, {default_vhost, <<"/">>}, - {default_permissions, [<<".*">>, <<".*">>, <<".*">>]}, - {memory_alarms, auto}]}]}. + {default_permissions, [<<".*">>, <<".*">>, <<".*">>]}]}]}. diff --git a/generate_app b/generate_app index 62301292..576b485e 100644 --- a/generate_app +++ b/generate_app @@ -1,10 +1,12 @@ #!/usr/bin/env escript %% -*- erlang -*- -main([BeamDir]) -> +main([BeamDir, TargetFile]) -> Modules = [list_to_atom(filename:basename(F, ".beam")) || F <- filelib:wildcard("*.beam", BeamDir)], {ok, {application, Application, Properties}} = io:read(''), NewProperties = lists:keyreplace(modules, 1, Properties, {modules, Modules}), - io:format("~p.", [{application, Application, NewProperties}]). + file:write_file( + TargetFile, + io_lib:format("~p.~n", [{application, Application, NewProperties}])). diff --git a/scripts/rabbitmq-activate-plugins.bat b/scripts/rabbitmq-activate-plugins.bat index d639ff0e..e7aa7095 100644 --- a/scripts/rabbitmq-activate-plugins.bat +++ b/scripts/rabbitmq-activate-plugins.bat @@ -30,6 +30,8 @@ REM REM Contributor(s): ______________________________________.
REM
+setlocal
+
if not exist "%ERLANG_HOME%\bin\erl.exe" (
echo.
echo ******************************
@@ -54,3 +56,5 @@ set RABBITMQ_EBIN_DIR=%~dp0..\ebin -rabbit plugins_expand_dir \""%RABBITMQ_PLUGINS_EXPAND_DIR:\=/%"\" ^
-rabbit rabbit_ebin \""%RABBITMQ_EBIN_DIR:\=/%"\" ^
-extra %*
+
+endlocal
diff --git a/scripts/rabbitmq-deactivate-plugins.bat b/scripts/rabbitmq-deactivate-plugins.bat index 71e60605..40155183 100644 --- a/scripts/rabbitmq-deactivate-plugins.bat +++ b/scripts/rabbitmq-deactivate-plugins.bat @@ -30,6 +30,10 @@ REM REM Contributor(s): ______________________________________.
REM
+setlocal
+
set RABBITMQ_EBIN_DIR=%~dp0..\ebin
del /f "%RABBITMQ_EBIN_DIR%"\rabbit.rel "%RABBITMQ_EBIN_DIR%"\rabbit.script "%RABBITMQ_EBIN_DIR%"\rabbit.boot
+
+endlocal
diff --git a/scripts/rabbitmq-multi.bat b/scripts/rabbitmq-multi.bat index 8abf13f1..8de18405 100755 --- a/scripts/rabbitmq-multi.bat +++ b/scripts/rabbitmq-multi.bat @@ -30,6 +30,8 @@ REM REM Contributor(s): ______________________________________.
REM
+setlocal
+
if "%RABBITMQ_BASE%"=="" (
set RABBITMQ_BASE=%APPDATA%\RabbitMQ
)
@@ -61,5 +63,13 @@ if not exist "%ERLANG_HOME%\bin\erl.exe" ( exit /B
)
-"%ERLANG_HOME%\bin\erl.exe" -pa "%~dp0..\ebin" -noinput -hidden %RABBITMQ_MULTI_ERL_ARGS% -sname rabbitmq_multi -s rabbit_multi %RABBITMQ_MULTI_START_ARGS% -extra %*
+"%ERLANG_HOME%\bin\erl.exe" ^
+-pa "%~dp0..\ebin" ^
+-noinput -hidden ^
+%RABBITMQ_MULTI_ERL_ARGS% ^
+-sname rabbitmq_multi ^
+-s rabbit_multi ^
+%RABBITMQ_MULTI_START_ARGS% ^
+-extra %*
+endlocal
diff --git a/scripts/rabbitmq-server b/scripts/rabbitmq-server index 67768c0e..34904850 100755 --- a/scripts/rabbitmq-server +++ b/scripts/rabbitmq-server @@ -109,9 +109,7 @@ exec erl \ -os_mon start_cpu_sup true \ -os_mon start_disksup false \ -os_mon start_memsup false \ - -os_mon start_os_sup false \ - -os_mon memsup_system_only true \ - -os_mon system_memory_high_watermark 0.95 \ + -os_mon vm_memory_high_watermark 0.4 \ -mnesia dir "\"${RABBITMQ_MNESIA_DIR}\"" \ ${RABBITMQ_CLUSTER_CONFIG_OPTION} \ ${RABBITMQ_SERVER_START_ARGS} \ diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat index b45a2713..5b82ec15 100755 --- a/scripts/rabbitmq-server.bat +++ b/scripts/rabbitmq-server.bat @@ -30,6 +30,8 @@ REM REM Contributor(s): ______________________________________.
REM
+setlocal
+
if "%RABBITMQ_BASE%"=="" (
set RABBITMQ_BASE=%APPDATA%\RabbitMQ
)
@@ -138,10 +140,10 @@ if exist "%RABBITMQ_CONFIG_FILE%.config" ( -os_mon start_cpu_sup true ^
-os_mon start_disksup false ^
-os_mon start_memsup false ^
--os_mon start_os_sup false ^
--os_mon memsup_system_only true ^
--os_mon system_memory_high_watermark 0.95 ^
+-os_mon vm_memory_high_watermark 0.4 ^
-mnesia dir \""%RABBITMQ_MNESIA_DIR%"\" ^
%CLUSTER_CONFIG% ^
%RABBITMQ_SERVER_START_ARGS% ^
%*
+
+endlocal
diff --git a/scripts/rabbitmq-service.bat b/scripts/rabbitmq-service.bat index 3a30ea26..0cc7bfa9 100755 --- a/scripts/rabbitmq-service.bat +++ b/scripts/rabbitmq-service.bat @@ -30,6 +30,8 @@ REM REM Contributor(s): ______________________________________.
REM
+setlocal
+
if "%RABBITMQ_SERVICENAME%"=="" (
set RABBITMQ_SERVICENAME=RabbitMQ
)
@@ -51,7 +53,7 @@ if "%RABBITMQ_NODE_PORT%"=="" ( )
if "%ERLANG_SERVICE_MANAGER_PATH%"=="" (
- set ERLANG_SERVICE_MANAGER_PATH=C:\Program Files\erl5.5.5\erts-5.5.5\bin
+ set ERLANG_SERVICE_MANAGER_PATH=C:\Program Files\erl5.6.5\erts-5.6.5\bin
)
set CONSOLE_FLAG=
@@ -69,7 +71,7 @@ if not exist "%ERLANG_SERVICE_MANAGER_PATH%\erlsrv.exe" ( echo ERLANG_SERVICE_MANAGER_PATH not set correctly.
echo **********************************************
echo.
- echo %ERLANG_SERVICE_MANAGER_PATH%\erlsrv.exe not found!
+ echo "%ERLANG_SERVICE_MANAGER_PATH%\erlsrv.exe" not found!
echo Please set ERLANG_SERVICE_MANAGER_PATH to the folder containing "erlsrv.exe".
echo.
exit /B 1
@@ -100,7 +102,7 @@ set SASL_LOGS_BACKUP=%RABBITMQ_BASE%\log\%RABBITMQ_NODENAME%-sasl.log%BACKUP_EXT if exist "%LOGS%" (
type "%LOGS%" >> "%LOGS_BACKUP%"
)
-if exist %SASL_LOGS% (
+if exist "%SASL_LOGS%" (
type "%SASL_LOGS%" >> "%SASL_LOGS_BACKUP%"
)
@@ -173,9 +175,7 @@ set ERLANG_SERVICE_ARGUMENTS= ^ -os_mon start_cpu_sup true ^
-os_mon start_disksup false ^
-os_mon start_memsup false ^
--os_mon start_os_sup false ^
--os_mon memsup_system_only true ^
--os_mon system_memory_high_watermark 0.95 ^
+-os_mon vm_memory_high_watermark 0.4 ^
-mnesia dir \""%RABBITMQ_MNESIA_DIR%"\" ^
%CLUSTER_CONFIG% ^
%RABBITMQ_SERVER_START_ARGS% ^
@@ -202,3 +202,5 @@ goto END :END
+
+endlocal
diff --git a/scripts/rabbitmqctl.bat b/scripts/rabbitmqctl.bat index 8a4e5445..512e8587 100755 --- a/scripts/rabbitmqctl.bat +++ b/scripts/rabbitmqctl.bat @@ -30,6 +30,8 @@ REM REM Contributor(s): ______________________________________.
REM
+setlocal
+
if "%RABBITMQ_NODENAME%"=="" (
set RABBITMQ_NODENAME=rabbit
)
@@ -47,3 +49,5 @@ if not exist "%ERLANG_HOME%\bin\erl.exe" ( )
"%ERLANG_HOME%\bin\erl.exe" -pa "%~dp0..\ebin" -noinput -hidden %RABBITMQ_CTL_ERL_ARGS% -sname rabbitmqctl -s rabbit_control -nodename %RABBITMQ_NODENAME% -extra %*
+
+endlocal
diff --git a/src/gen_server2.erl b/src/gen_server2.erl index a2d9350c..53edf8de 100644 --- a/src/gen_server2.erl +++ b/src/gen_server2.erl @@ -180,6 +180,20 @@ -import(error_logger, [format/2]). %%%========================================================================= +%%% Specs. These exist only to shut up dialyzer's warnings +%%%========================================================================= + +-ifdef(use_specs). + +-spec(handle_common_termination/6 :: + (any(), any(), any(), atom(), any(), any()) -> no_return()). + +-spec(hibernate/7 :: + (pid(), any(), any(), atom(), any(), queue(), any()) -> no_return()). + +-endif. + +%%%========================================================================= %%% API %%%========================================================================= diff --git a/src/rabbit.erl b/src/rabbit.erl index 18fd1b17..29407e4e 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -140,8 +140,16 @@ start(normal, []) -> ok = rabbit_binary_generator: check_empty_content_body_frame_size(), - {ok, MemoryAlarms} = application:get_env(memory_alarms), - ok = rabbit_alarm:start(MemoryAlarms), + ok = rabbit_alarm:start(), + MemoryWatermark = + application:get_env(os_mon, vm_memory_high_watermark), + ok = case MemoryWatermark of + {ok, Float} when Float == 0 -> ok; + {ok, Float} -> start_child(vm_memory_monitor, [Float]); + undefined -> + throw({undefined, os_mon, + vm_memory_high_watermark, settings}) + end, ok = rabbit_amqqueue:start(), @@ -202,6 +210,10 @@ start(normal, []) -> stop(_State) -> terminated_ok = error_logger:delete_report_handler(rabbit_error_logger), ok = rabbit_alarm:stop(), + ok = case rabbit_mnesia:is_clustered() of + true -> rabbit_amqqueue:on_node_down(node()); + false -> rabbit_mnesia:empty_ram_only_tables() + end, ok. %--------------------------------------------------------------------------- @@ -264,8 +276,11 @@ print_banner() -> io:nl(). start_child(Mod) -> + start_child(Mod, []). + +start_child(Mod, Args) -> {ok,_} = supervisor:start_child(rabbit_sup, - {Mod, {Mod, start_link, []}, + {Mod, {Mod, start_link, Args}, transient, 100, worker, [Mod]}), ok. diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl index 7a2fbcb8..9a639ed4 100644 --- a/src/rabbit_alarm.erl +++ b/src/rabbit_alarm.erl @@ -33,24 +33,19 @@ -behaviour(gen_event). --export([start/1, stop/0, register/2]). +-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). - -%% OSes on which we know memory alarms to be trustworthy --define(SUPPORTED_OS, [{unix, linux}, {unix, darwin}]). - --record(alarms, {alertees, system_memory_high_watermark = false}). +-record(alarms, {alertees, vm_memory_high_watermark = false}). %%---------------------------------------------------------------------------- -ifdef(use_specs). -type(mfa_tuple() :: {atom(), atom(), list()}). --spec(start/1 :: (boolean() | 'auto') -> 'ok'). +-spec(start/0 :: () -> 'ok'). -spec(stop/0 :: () -> 'ok'). -spec(register/2 :: (pid(), mfa_tuple()) -> 'ok'). @@ -58,20 +53,8 @@ %%---------------------------------------------------------------------------- -start(MemoryAlarms) -> - EnableAlarms = case MemoryAlarms of - true -> true; - false -> false; - auto -> lists:member(os:type(), ?SUPPORTED_OS) - end, - ok = alarm_handler:add_alarm_handler(?MODULE, [EnableAlarms]), - case whereis(memsup) of - undefined -> if EnableAlarms -> ok = start_memsup(), - ok = adjust_memsup_interval(); - true -> ok - end; - _ -> ok = adjust_memsup_interval() - end. +start() -> + ok = alarm_handler:add_alarm_handler(?MODULE, []). stop() -> ok = alarm_handler:delete_alarm_handler(?MODULE). @@ -83,43 +66,33 @@ register(Pid, HighMemMFA) -> %%---------------------------------------------------------------------------- -init([MemoryAlarms]) -> - {ok, #alarms{alertees = case MemoryAlarms of - true -> dict:new(); - false -> undefined - end}}. +init([]) -> + {ok, #alarms{alertees = dict:new()}}. -handle_call({register, _Pid, _HighMemMFA}, - State = #alarms{alertees = undefined}) -> - {ok, ok, State}; -handle_call({register, Pid, HighMemMFA}, +handle_call({register, Pid, {M, F, A} = 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, + ok = case State#alarms.vm_memory_high_watermark of + true -> 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) -> +handle_event({set_alarm, {vm_memory_high_watermark, []}}, State) -> ok = alert(true, State#alarms.alertees), - {ok, State#alarms{system_memory_high_watermark = true}}; + {ok, State#alarms{vm_memory_high_watermark = true}}; -handle_event({clear_alarm, system_memory_high_watermark}, State) -> +handle_event({clear_alarm, vm_memory_high_watermark}, State) -> ok = alert(false, State#alarms.alertees), - {ok, State#alarms{system_memory_high_watermark = false}}; + {ok, State#alarms{vm_memory_high_watermark = false}}; handle_event(_Event, State) -> {ok, State}. -handle_info({'DOWN', _MRef, process, _Pid, _Reason}, - State = #alarms{alertees = undefined}) -> - {ok, State}; handle_info({'DOWN', _MRef, process, Pid, _Reason}, State = #alarms{alertees = Alertess}) -> {ok, State#alarms{alertees = dict:erase(Pid, Alertess)}}; @@ -134,57 +107,6 @@ code_change(_OldVsn, State, _Extra) -> {ok, State}. %%---------------------------------------------------------------------------- - -start_memsup() -> - {Mod, Args} = - 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, [rabbit_memsup_linux]}; - {unix, darwin} -> {rabbit_memsup, [rabbit_memsup_darwin]}; - - %% 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, Args}, - permanent, 2000, worker, [Mod]}), - ok. - -adjust_memsup_interval() -> - %% 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). - alert(_Alert, undefined) -> ok; alert(Alert, Alertees) -> diff --git a/src/rabbit_memsup.erl b/src/rabbit_memsup.erl deleted file mode 100644 index b0d57cb2..00000000 --- a/src/rabbit_memsup.erl +++ /dev/null @@ -1,142 +0,0 @@ -%% 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 before 22-Nov-2008 00:00:00 GMT by LShift Ltd, -%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd -%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial -%% Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift -%% Ltd. Portions created by Cohesive Financial Technologies LLC are -%% Copyright (C) 2007-2009 Cohesive Financial Technologies -%% LLC. Portions created by Rabbit Technologies Ltd are Copyright -%% (C) 2007-2009 Rabbit Technologies Ltd. -%% -%% All Rights Reserved. -%% -%% Contributor(s): ______________________________________. -%% - --module(rabbit_memsup). - --behaviour(gen_server). - --export([start_link/1]). - --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - --export([update/0]). - --record(state, {memory_fraction, - timeout, - timer, - mod, - mod_state, - alarmed - }). - --define(SERVER, memsup). %% must be the same as the standard memsup - --define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000). - -%%---------------------------------------------------------------------------- - --ifdef(use_specs). - --spec(start_link/1 :: (atom()) -> {'ok', pid()} | 'ignore' | {'error', any()}). --spec(update/0 :: () -> 'ok'). - --endif. - -%%---------------------------------------------------------------------------- - -start_link(Args) -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [Args], []). - -update() -> - gen_server:cast(?SERVER, update). - -%%---------------------------------------------------------------------------- - -init([Mod]) -> - Fraction = os_mon:get_env(memsup, system_memory_high_watermark), - TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL), - InitState = Mod:init(), - State = #state { memory_fraction = Fraction, - timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL, - timer = TRef, - mod = Mod, - mod_state = InitState, - alarmed = false }, - {ok, internal_update(State)}. - -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(get_memory_data, _From, - State = #state { mod = Mod, mod_state = ModState }) -> - {reply, Mod:get_memory_data(ModState), State}; - -handle_call(_Request, _From, State) -> - {noreply, State}. - -handle_cast(update, State) -> - {noreply, internal_update(State)}; - -handle_cast(_Request, State) -> - {noreply, State}. - -handle_info(_Info, State) -> - {noreply, State}. - -terminate(_Reason, _State) -> - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -internal_update(State = #state { memory_fraction = MemoryFraction, - alarmed = Alarmed, - mod = Mod, mod_state = ModState }) -> - ModState1 = Mod:update(ModState), - {MemTotal, MemUsed, _BigProc} = Mod:get_memory_data(ModState1), - 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, - State #state { mod_state = ModState1, alarmed = NewAlarmed }. diff --git a/src/rabbit_memsup_darwin.erl b/src/rabbit_memsup_darwin.erl deleted file mode 100644 index 3de2d843..00000000 --- a/src/rabbit_memsup_darwin.erl +++ /dev/null @@ -1,88 +0,0 @@ -%% 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 before 22-Nov-2008 00:00:00 GMT by LShift Ltd, -%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd -%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial -%% Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift -%% Ltd. Portions created by Cohesive Financial Technologies LLC are -%% Copyright (C) 2007-2009 Cohesive Financial Technologies -%% LLC. Portions created by Rabbit Technologies Ltd are Copyright -%% (C) 2007-2009 Rabbit Technologies Ltd. -%% -%% All Rights Reserved. -%% -%% Contributor(s): ______________________________________. -%% - --module(rabbit_memsup_darwin). - --export([init/0, update/1, get_memory_data/1]). - --record(state, {total_memory, - allocated_memory}). - -%%---------------------------------------------------------------------------- - --ifdef(use_specs). - --type(state() :: #state { total_memory :: ('undefined' | non_neg_integer()), - allocated_memory :: ('undefined' | non_neg_integer()) - }). - --spec(init/0 :: () -> state()). --spec(update/1 :: (state()) -> state()). --spec(get_memory_data/1 :: (state()) -> {non_neg_integer(), non_neg_integer(), - ('undefined' | pid())}). - --endif. - -%%---------------------------------------------------------------------------- - -init() -> - #state{total_memory = undefined, - allocated_memory = undefined}. - -update(State) -> - File = os:cmd("/usr/bin/vm_stat"), - Lines = string:tokens(File, "\n"), - Dict = dict:from_list(lists:map(fun parse_line/1, Lines)), - [PageSize, Inactive, Active, Free, Wired] = - [dict:fetch(Key, Dict) || - Key <- [page_size, 'Pages inactive', 'Pages active', 'Pages free', - 'Pages wired down']], - MemTotal = PageSize * (Inactive + Active + Free + Wired), - MemUsed = PageSize * (Active + Wired), - State#state{total_memory = MemTotal, allocated_memory = MemUsed}. - -get_memory_data(State) -> - {State#state.total_memory, State#state.allocated_memory, undefined}. - -%%---------------------------------------------------------------------------- - -%% A line looks like "Foo bar: 123456." -parse_line(Line) -> - [Name, RHS | _Rest] = string:tokens(Line, ":"), - case Name of - "Mach Virtual Memory Statistics" -> - ["(page", "size", "of", PageSize, "bytes)"] = - string:tokens(RHS, " "), - {page_size, list_to_integer(PageSize)}; - _ -> - [Value | _Rest1] = string:tokens(RHS, " ."), - {list_to_atom(Name), list_to_integer(Value)} - end. diff --git a/src/rabbit_memsup_linux.erl b/src/rabbit_memsup_linux.erl deleted file mode 100644 index ca942d7c..00000000 --- a/src/rabbit_memsup_linux.erl +++ /dev/null @@ -1,101 +0,0 @@ -%% 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 before 22-Nov-2008 00:00:00 GMT by LShift Ltd, -%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd -%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial -%% Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift -%% Ltd. Portions created by Cohesive Financial Technologies LLC are -%% Copyright (C) 2007-2009 Cohesive Financial Technologies -%% LLC. Portions created by Rabbit Technologies Ltd are Copyright -%% (C) 2007-2009 Rabbit Technologies Ltd. -%% -%% All Rights Reserved. -%% -%% Contributor(s): ______________________________________. -%% - --module(rabbit_memsup_linux). - --export([init/0, update/1, get_memory_data/1]). - --record(state, {total_memory, - allocated_memory}). - -%%---------------------------------------------------------------------------- - --ifdef(use_specs). - --type(state() :: #state { total_memory :: ('undefined' | non_neg_integer()), - allocated_memory :: ('undefined' | non_neg_integer()) - }). - --spec(init/0 :: () -> state()). --spec(update/1 :: (state()) -> state()). --spec(get_memory_data/1 :: (state()) -> {non_neg_integer(), non_neg_integer(), - ('undefined' | pid())}). - --endif. - -%%---------------------------------------------------------------------------- - -init() -> - #state{total_memory = undefined, - allocated_memory = undefined}. - -update(State) -> - File = read_proc_file("/proc/meminfo"), - Lines = string:tokens(File, "\n"), - Dict = dict:from_list(lists:map(fun parse_line/1, Lines)), - [MemTotal, MemFree, Buffers, Cached] = - [dict:fetch(Key, Dict) || - Key <- ['MemTotal', 'MemFree', 'Buffers', 'Cached']], - MemUsed = MemTotal - MemFree - Buffers - Cached, - State#state{total_memory = MemTotal, allocated_memory = MemUsed}. - -get_memory_data(State) -> - {State#state.total_memory, State#state.allocated_memory, undefined}. - -%%---------------------------------------------------------------------------- - --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, RHS | _Rest] = string:tokens(Line, ":"), - [Value | UnitsRest] = string:tokens(RHS, " "), - Value1 = case UnitsRest of - [] -> list_to_integer(Value); %% no units - ["kB"] -> list_to_integer(Value) * 1024 - end, - {list_to_atom(Name), Value1}. diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index c4d5aac6..749038db 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -32,7 +32,8 @@ -module(rabbit_mnesia). -export([ensure_mnesia_dir/0, dir/0, status/0, init/0, is_db_empty/0, - cluster/1, reset/0, force_reset/0]). + cluster/1, reset/0, force_reset/0, is_clustered/0, + empty_ram_only_tables/0]). -export([table_names/0]). @@ -54,6 +55,8 @@ -spec(cluster/1 :: ([erlang_node()]) -> 'ok'). -spec(reset/0 :: () -> 'ok'). -spec(force_reset/0 :: () -> 'ok'). +-spec(is_clustered/0 :: () -> boolean()). +-spec(empty_ram_only_tables/0 :: () -> 'ok'). -spec(create_tables/0 :: () -> 'ok'). -endif. @@ -98,6 +101,21 @@ cluster(ClusterNodes) -> reset() -> reset(false). force_reset() -> reset(true). +is_clustered() -> + RunningNodes = mnesia:system_info(running_db_nodes), + [node()] /= RunningNodes andalso [] /= RunningNodes. + +empty_ram_only_tables() -> + Node = node(), + lists:foreach( + fun (TabName) -> + case lists:member(Node, mnesia:table_info(TabName, ram_copies)) of + true -> {atomic, ok} = mnesia:clear_table(TabName); + false -> ok + end + end, table_names()), + ok. + %%-------------------------------------------------------------------- table_definitions() -> diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_plugin_activator.erl index e22d844f..a2ac74ef 100644 --- a/src/rabbit_plugin_activator.erl +++ b/src/rabbit_plugin_activator.erl @@ -39,6 +39,17 @@ -define(BaseApps, [rabbit]). %%---------------------------------------------------------------------------- +%% Specs +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-spec(start/0 :: () -> no_return()). +-spec(stop/0 :: () -> 'ok'). + +-endif. + +%%---------------------------------------------------------------------------- start() -> %% Ensure Rabbit is loaded so we can access it's environment diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl new file mode 100644 index 00000000..6da47933 --- /dev/null +++ b/src/vm_memory_monitor.erl @@ -0,0 +1,322 @@ +%% 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 before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +%% In practice Erlang shouldn't be allowed to grow to more than a half +%% of available memory. The pessimistic scenario is when the Erlang VM +%% has a single process that's consuming all memory. In such a case, +%% during garbage collection, Erlang tries to allocate a huge chunk of +%% continuous memory, which can result in a crash or heavy swapping. +%% +%% This module tries to warn Rabbit before such situations occur, so +%% that it has a higher chance to avoid running out of memory. +%% +%% This code depends on Erlang os_mon application. + +-module(vm_memory_monitor). + +-behaviour(gen_server2). + +-export([start_link/1]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-export([update/0, get_total_memory/0, + get_check_interval/0, set_check_interval/1, + get_vm_memory_high_watermark/0, set_vm_memory_high_watermark/1]). + + +-define(SERVER, ?MODULE). +-define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000). + +%% For an unknown OS, we assume that we have 1GB of memory. It'll be +%% wrong. Scale by vm_memory_high_watermark in configuration to get a +%% sensible value. +-define(MEMORY_SIZE_FOR_UNKNOWN_OS, 1073741824). + +-record(state, {total_memory, + memory_limit, + timeout, + timer, + alarmed + }). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-spec(start_link/1 :: (float()) -> ('ignore' | {error, any()} | {'ok', pid()})). +-spec(update/0 :: () -> 'ok'). +-spec(get_total_memory/0 :: () -> (non_neg_integer() | unknown)). +-spec(get_check_interval/0 :: () -> non_neg_integer()). +-spec(set_check_interval/1 :: (non_neg_integer()) -> 'ok'). +-spec(get_vm_memory_high_watermark/0 :: () -> float()). +-spec(set_vm_memory_high_watermark/1 :: (float()) -> 'ok'). + +-endif. + + +%%---------------------------------------------------------------------------- +%% gen_server callbacks +%%---------------------------------------------------------------------------- + +start_link(Args) -> + gen_server2:start_link({local, ?SERVER}, ?MODULE, [Args], []). + +init([MemFraction]) -> + TotalMemory = + case get_total_memory() of + unknown -> + rabbit_log:warning( + "Unknown total memory size for your OS ~p. " + "Assuming memory size is ~pMB.~n", + [os:type(), trunc(?MEMORY_SIZE_FOR_UNKNOWN_OS/1048576)]), + ?MEMORY_SIZE_FOR_UNKNOWN_OS; + M -> M + end, + MemLimit = get_mem_limit(MemFraction, TotalMemory), + rabbit_log:info("Memory limit set to ~pMB.~n", [trunc(MemLimit/1048576)]), + TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL), + State = #state { total_memory = TotalMemory, + memory_limit = MemLimit, + timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL, + timer = TRef, + alarmed = false}, + {ok, internal_update(State)}. + +handle_call(get_vm_memory_high_watermark, _From, State) -> + {reply, State#state.memory_limit / State#state.total_memory, State}; + +handle_call({set_vm_memory_high_watermark, MemFraction}, _From, State) -> + MemLimit = get_mem_limit(MemFraction, State#state.total_memory), + rabbit_log:info("Memory alarm changed to ~p, ~p bytes.~n", + [MemFraction, MemLimit]), + {reply, ok, State#state{memory_limit = MemLimit}}; + +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) -> + {noreply, internal_update(State)}; + +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info(_Info, State) -> + {noreply, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%---------------------------------------------------------------------------- +%% Public API +%%---------------------------------------------------------------------------- + +update() -> + gen_server2:cast(?SERVER, update). + +get_total_memory() -> + get_total_memory(os:type()). + +get_check_interval() -> + gen_server2:call(?MODULE, get_check_interval). + +set_check_interval(Fraction) -> + gen_server2:call(?MODULE, {set_check_interval, Fraction}). + +get_vm_memory_high_watermark() -> + gen_server2:call(?MODULE, get_vm_memory_high_watermark). + +set_vm_memory_high_watermark(Fraction) -> + gen_server2:call(?MODULE, {set_vm_memory_high_watermark, Fraction}). + +%%---------------------------------------------------------------------------- +%% Server Internals +%%---------------------------------------------------------------------------- + +internal_update(State = #state { memory_limit = MemLimit, + alarmed = Alarmed}) -> + MemUsed = erlang:memory(total), + NewAlarmed = MemUsed > MemLimit, + case {Alarmed, NewAlarmed} of + {false, true} -> + emit_update_info(set, MemUsed, MemLimit), + alarm_handler:set_alarm({vm_memory_high_watermark, []}); + {true, false} -> + emit_update_info(clear, MemUsed, MemLimit), + alarm_handler:clear_alarm(vm_memory_high_watermark); + _ -> + ok + end, + State #state {alarmed = NewAlarmed}. + +emit_update_info(State, MemUsed, MemLimit) -> + rabbit_log:info("vm_memory_high_watermark ~p. Memory used:~p allowed:~p~n", + [State, MemUsed, MemLimit]). + +start_timer(Timeout) -> + {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []), + TRef. + +%% On a 32-bit machine, if you're using more than 2 gigs of RAM you're +%% in big trouble anyway. +get_vm_limit() -> + case erlang:system_info(wordsize) of + 4 -> 4294967296; %% 4 GB for 32 bits 2^32 + 8 -> 281474976710656 %% 256 TB for 64 bits 2^48 + %% http://en.wikipedia.org/wiki/X86-64#Virtual_address_space_details + end. + +get_mem_limit(MemFraction, TotalMemory) -> + lists:min([trunc(TotalMemory * MemFraction), get_vm_limit()]). + +%%---------------------------------------------------------------------------- +%% Internal Helpers +%%---------------------------------------------------------------------------- +cmd(Command) -> + Exec = hd(string:tokens(Command, " ")), + case os:find_executable(Exec) of + false -> throw({command_not_found, Exec}); + _ -> os:cmd(Command) + end. + +%% get_total_memory(OS) -> Total +%% Windows and Freebsd code based on: memsup:get_memory_usage/1 +%% Original code was part of OTP and released under "Erlang Public License". + +get_total_memory({unix,darwin}) -> + File = cmd("/usr/bin/vm_stat"), + Lines = string:tokens(File, "\n"), + Dict = dict:from_list(lists:map(fun parse_line_mach/1, Lines)), + [PageSize, Inactive, Active, Free, Wired] = + [dict:fetch(Key, Dict) || + Key <- [page_size, 'Pages inactive', 'Pages active', 'Pages free', + 'Pages wired down']], + PageSize * (Inactive + Active + Free + Wired); + +get_total_memory({unix,freebsd}) -> + PageSize = freebsd_sysctl("vm.stats.vm.v_page_size"), + PageCount = freebsd_sysctl("vm.stats.vm.v_page_count"), + PageCount * PageSize; + +get_total_memory({win32,_OSname}) -> + [Result|_] = os_mon_sysinfo:get_mem_info(), + {ok, [_MemLoad, TotPhys, _AvailPhys, + _TotPage, _AvailPage, _TotV, _AvailV], _RestStr} = + io_lib:fread("~d~d~d~d~d~d~d", Result), + TotPhys; + +get_total_memory({unix, linux}) -> + File = read_proc_file("/proc/meminfo"), + Lines = string:tokens(File, "\n"), + Dict = dict:from_list(lists:map(fun parse_line_linux/1, Lines)), + dict:fetch('MemTotal', Dict); + +get_total_memory({unix, sunos}) -> + File = cmd("/usr/sbin/prtconf"), + Lines = string:tokens(File, "\n"), + Dict = dict:from_list(lists:map(fun parse_line_sunos/1, Lines)), + dict:fetch('Memory size', Dict); + +get_total_memory(_OsType) -> + unknown. + +%% A line looks like "Foo bar: 123456." +parse_line_mach(Line) -> + [Name, RHS | _Rest] = string:tokens(Line, ":"), + case Name of + "Mach Virtual Memory Statistics" -> + ["(page", "size", "of", PageSize, "bytes)"] = + string:tokens(RHS, " "), + {page_size, list_to_integer(PageSize)}; + _ -> + [Value | _Rest1] = string:tokens(RHS, " ."), + {list_to_atom(Name), list_to_integer(Value)} + end. + +%% A line looks like "FooBar: 123456 kB" +parse_line_linux(Line) -> + [Name, RHS | _Rest] = string:tokens(Line, ":"), + [Value | UnitsRest] = string:tokens(RHS, " "), + Value1 = case UnitsRest of + [] -> list_to_integer(Value); %% no units + ["kB"] -> list_to_integer(Value) * 1024 + end, + {list_to_atom(Name), Value1}. + +%% A line looks like "Memory size: 1024 Megabytes" +parse_line_sunos(Line) -> + case string:tokens(Line, ":") of + [Name, RHS | _Rest] -> + [Value1 | UnitsRest] = string:tokens(RHS, " "), + Value2 = case UnitsRest of + ["Gigabytes"] -> + list_to_integer(Value1) * 1024 * 1024 * 1024; + ["Megabytes"] -> + list_to_integer(Value1) * 1024 * 1024; + ["Kilobytes"] -> + list_to_integer(Value1) * 1024; + _ -> + Value1 ++ UnitsRest %% no known units + end, + {list_to_atom(Name), Value2}; + [Name] -> {list_to_atom(Name), none} + end. + +freebsd_sysctl(Def) -> + list_to_integer(cmd("/sbin/sysctl -n " ++ Def) -- "\n"). + +%% 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)). + +-define(BUFFER_SIZE, 1024). +read_proc_file(IoDevice, Acc) -> + case file:read(IoDevice, ?BUFFER_SIZE) of + {ok, Res} -> read_proc_file(IoDevice, [Res | Acc]); + eof -> Acc + end. |