diff options
author | Matthias Radestock <matthias@lshift.net> | 2009-12-17 11:16:21 +0000 |
---|---|---|
committer | Matthias Radestock <matthias@lshift.net> | 2009-12-17 11:16:21 +0000 |
commit | bb173a12eb269eeca191369b8ac71dffce8292a1 (patch) | |
tree | 3d9b7b242bcc9468f30f0dd5cfbbcb94ccb06741 | |
parent | 5482a87d0ae83fb39df473931b15c63c62677b60 (diff) | |
parent | 937626a88075d2936cb1cc49e84841a409b6cc5a (diff) | |
download | rabbitmq-server-bb173a12eb269eeca191369b8ac71dffce8292a1.tar.gz |
merge bug22013 into default
-rw-r--r-- | Makefile | 13 | ||||
-rw-r--r-- | codegen.py | 25 | ||||
-rw-r--r-- | docs/rabbitmqctl.1.pod | 11 | ||||
-rw-r--r-- | include/rabbit.hrl | 2 | ||||
-rw-r--r-- | src/rabbit.erl | 1 | ||||
-rw-r--r-- | src/rabbit_control.erl | 46 | ||||
-rw-r--r-- | src/rabbit_exchange.erl | 18 | ||||
-rw-r--r-- | src/rabbit_exchange_type.erl | 107 | ||||
-rw-r--r-- | src/rabbit_networking.erl | 48 | ||||
-rw-r--r-- | src/rabbit_plugin_activator.erl | 24 | ||||
-rw-r--r-- | src/rabbit_reader.erl | 50 | ||||
-rw-r--r-- | src/vm_memory_monitor.erl | 83 |
12 files changed, 282 insertions, 146 deletions
@@ -15,7 +15,20 @@ TARGETS=$(EBIN_DIR)/rabbit.app $(BEAM_TARGETS) WEB_URL=http://stage.rabbitmq.com/ MANPAGES=$(patsubst %.pod, %.gz, $(wildcard docs/*.[0-9].pod)) +ifeq ($(shell python -c 'import simplejson' 2>/dev/null && echo yes),yes) PYTHON=python +else +ifeq ($(shell python2.6 -c 'import simplejson' 2>/dev/null && echo yes),yes) +PYTHON=python2.6 +else +ifeq ($(shell python2.5 -c 'import simplejson' 2>/dev/null && echo yes),yes) +PYTHON=python2.5 +else +# Hmm. Missing simplejson? +PYTHON=python +endif +endif +endif BASIC_PLT=basic.plt RABBIT_PLT=rabbit.plt @@ -244,19 +244,10 @@ def genErl(spec): print 'lookup_amqp_exception(%s) -> {%s, ?%s, <<"%s">>};' % \ (n.lower(), hardErrorBoolStr, n, n) - def genIsAmqpHardErrorCode(c,v,cls): - mCls = messageConstantClass(cls) - if mCls == 'SOFT_ERROR' : genIsAmqpHardErrorCode1(c,'false') - elif mCls == 'HARD_ERROR' : genIsAmqpHardErrorCode1(c,'true') - elif mCls == '' : pass - else: raise 'Unkown constant class', cls - - def genIsAmqpHardErrorCode1(c,hardErrorBoolStr): + def genAmqpException(c,v,cls): n = erlangConstantName(c) - print 'is_amqp_hard_error_code(?%s) -> %s;' % \ - (n, hardErrorBoolStr) - print 'is_amqp_hard_error_code(%s) -> %s;' % \ - (n.lower(), hardErrorBoolStr) + print 'amqp_exception(?%s) -> %s;' % \ + (n, n.lower()) methods = spec.allMethods() @@ -274,7 +265,7 @@ def genErl(spec): -export([encode_method_fields/1]). -export([encode_properties/1]). -export([lookup_amqp_exception/1]). --export([is_amqp_hard_error_code/1]). +-export([amqp_exception/1]). bitvalue(true) -> 1; bitvalue(false) -> 0; @@ -313,12 +304,8 @@ bitvalue(undefined) -> 0. print " rabbit_log:warning(\"Unknown AMQP error code '~p'~n\", [Code])," print " {true, ?INTERNAL_ERROR, <<\"INTERNAL_ERROR\">>}." - for(c,v,cls) in spec.constants: genIsAmqpHardErrorCode(c,v,cls) - print "is_amqp_hard_error_code(Code) when is_integer(Code) ->" - print " true;" - print "is_amqp_hard_error_code(Code) ->" - print " rabbit_log:warning(\"Unknown AMQP error code '~p'~n\", [Code])," - print " true." + for(c,v,cls) in spec.constants: genAmqpException(c,v,cls) + print "amqp_exception(_Code) -> undefined." def genHrl(spec): def erlType(domain): diff --git a/docs/rabbitmqctl.1.pod b/docs/rabbitmqctl.1.pod index 6b420872..5255be28 100644 --- a/docs/rabbitmqctl.1.pod +++ b/docs/rabbitmqctl.1.pod @@ -198,9 +198,9 @@ whether the queue will be deleted when no longer used queue arguments -=item node +=item pid -node on which the process associated with the queue resides +id of the Erlang process associated with the queue =item messages_ready @@ -297,7 +297,7 @@ I<user>, I<peer_address>, I<peer_port> and I<state> are assumed. =item node -node on which the process associated with the connection resides +id of the Erlang process associated with the connection =item address @@ -340,6 +340,11 @@ connection timeout maximum frame size (bytes) +=item client_properties + +informational properties transmitted by the client during connection +establishment + =item recv_oct octets received diff --git a/include/rabbit.hrl b/include/rabbit.hrl index 5703d0d6..4b157cbc 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -36,7 +36,7 @@ -record(vhost, {virtual_host, dummy}). --record(connection, {user, timeout_sec, frame_max, vhost}). +-record(connection, {user, timeout_sec, frame_max, vhost, client_properties}). -record(content, {class_id, diff --git a/src/rabbit.erl b/src/rabbit.erl index c6dde385..3293927a 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -134,6 +134,7 @@ start(normal, []) -> fun () -> ok = rabbit_mnesia:init() end}, {"core processes", fun () -> + ok = start_child(rabbit_exchange_type), ok = start_child(rabbit_log), ok = rabbit_hooks:start(), diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl index 19579729..ddd0c002 100644 --- a/src/rabbit_control.erl +++ b/src/rabbit_control.erl @@ -173,7 +173,7 @@ The list_queues, list_exchanges and list_bindings commands accept an optional virtual host parameter for which to display results. The default value is \"/\". <QueueInfoItem> must be a member of the list [name, durable, auto_delete, -arguments, node, messages_ready, messages_unacknowledged, messages_uncommitted, +arguments, pid, messages_ready, messages_unacknowledged, messages_uncommitted, messages, acks_uncommitted, consumers, transactions, memory]. The default is to display name and (number of) messages. @@ -183,10 +183,10 @@ auto_delete, arguments]. The default is to display name and type. The output format for \"list_bindings\" is a list of rows containing exchange name, queue name, routing key and arguments, in that order. -<ConnectionInfoItem> must be a member of the list [node, address, port, +<ConnectionInfoItem> must be a member of the list [pid, address, port, peer_address, peer_port, state, channels, user, vhost, timeout, frame_max, -recv_oct, recv_cnt, send_oct, send_cnt, send_pend]. The default is to display -user, peer_address, peer_port and state. +client_properties, recv_oct, recv_cnt, send_oct, send_cnt, send_pend]. +The default is to display user, peer_address, peer_port and state. "), halt(1). @@ -268,8 +268,7 @@ action(list_user_permissions, Node, Args = [_Username], Inform) -> action(list_queues, Node, Args, Inform) -> Inform("Listing queues", []), {VHostArg, RemainingArgs} = parse_vhost_flag_bin(Args), - ArgAtoms = list_replace(node, pid, - default_if_empty(RemainingArgs, [name, messages])), + ArgAtoms = default_if_empty(RemainingArgs, [name, messages]), display_info_list(rpc_call(Node, rabbit_amqqueue, info_all, [VHostArg, ArgAtoms]), ArgAtoms); @@ -294,9 +293,7 @@ action(list_bindings, Node, Args, Inform) -> action(list_connections, Node, Args, Inform) -> Inform("Listing connections", []), - ArgAtoms = list_replace(node, pid, - default_if_empty(Args, [user, peer_address, - peer_port, state])), + ArgAtoms = default_if_empty(Args, [user, peer_address, peer_port, state]), display_info_list(rpc_call(Node, rabbit_networking, connection_info_all, [ArgAtoms]), ArgAtoms); @@ -358,12 +355,15 @@ format_info_item(Key, Items) -> is_tuple(Value) -> inet_parse:ntoa(Value); Value when is_pid(Value) -> - atom_to_list(node(Value)); + pid_to_string(Value); Value when is_binary(Value) -> escape(Value); Value when is_atom(Value) -> - escape(atom_to_list(Value)); - Value -> + escape(atom_to_list(Value)); + Value = [{TableEntryKey, TableEntryType, _TableEntryValue} | _] + when is_binary(TableEntryKey) andalso is_atom(TableEntryType) -> + io_lib:format("~1000000000000p", [prettify_amqp_table(Value)]); + Value -> io_lib:format("~w", [Value]) end. @@ -388,14 +388,14 @@ rpc_call(Node, Mod, Fun, Args) -> %% characters. We don't escape characters above 127, since they may %% form part of UTF-8 strings. -escape(Bin) when binary(Bin) -> +escape(Bin) when is_binary(Bin) -> escape(binary_to_list(Bin)); escape(L) when is_list(L) -> escape_char(lists:reverse(L), []). escape_char([$\\ | T], Acc) -> escape_char(T, [$\\, $\\ | Acc]); -escape_char([X | T], Acc) when X > 32, X /= 127 -> +escape_char([X | T], Acc) when X >= 32, X /= 127 -> escape_char(T, [X | Acc]); escape_char([X | T], Acc) -> escape_char(T, [$\\, $0 + (X bsr 6), $0 + (X band 8#070 bsr 3), @@ -403,6 +403,20 @@ escape_char([X | T], Acc) -> escape_char([], Acc) -> Acc. -list_replace(Find, Replace, List) -> - [case X of Find -> Replace; _ -> X end || X <- List]. +prettify_amqp_table(Table) -> + [{escape(K), prettify_typed_amqp_value(T, V)} || {K, T, V} <- Table]. +prettify_typed_amqp_value(Type, Value) -> + case Type of + longstr -> escape(Value); + table -> prettify_amqp_table(Value); + array -> [prettify_typed_amqp_value(T, V) || {T, V} <- Value]; + _ -> Value + end. + +%% see http://erlang.org/doc/apps/erts/erl_ext_dist.html (8.10 and 8.7) +pid_to_string(Pid) -> + <<131,103,100,NodeLen:16,NodeBin:NodeLen/binary,Id:32,Ser:32,_Cre:8>> + = term_to_binary(Pid), + Node = binary_to_term(<<131,100,NodeLen:16,NodeBin:NodeLen/binary>>), + lists:flatten(io_lib:format("<~w.~B.~B>", [Node, Id, Ser])). diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl index 09ea1e96..be73e818 100644 --- a/src/rabbit_exchange.erl +++ b/src/rabbit_exchange.erl @@ -134,18 +134,18 @@ declare(ExchangeName, Type, Durable, AutoDelete, Args) -> end end). -typename_to_plugin_module(T) when is_binary(T) -> - case catch list_to_existing_atom("rabbit_exchange_type_" ++ binary_to_list(T)) of - {'EXIT', {badarg, _}} -> +typename_to_plugin_module(T) -> + case rabbit_exchange_type:lookup_module(T) of + {ok, Module} -> + Module; + {error, not_found} -> rabbit_misc:protocol_error( - command_invalid, "invalid exchange type '~s'", [T]); - Module -> - Module + command_invalid, "invalid exchange type '~s'", [T]) end. -plugin_module_to_typename(M) when is_atom(M) -> - "rabbit_exchange_type_" ++ S = atom_to_list(M), - list_to_binary(S). +plugin_module_to_typename(M) -> + {ok, TypeName} = rabbit_exchange_type:lookup_name(M), + TypeName. check_type(T) -> Module = typename_to_plugin_module(T), diff --git a/src/rabbit_exchange_type.erl b/src/rabbit_exchange_type.erl new file mode 100644 index 00000000..58dcfbb6 --- /dev/null +++ b/src/rabbit_exchange_type.erl @@ -0,0 +1,107 @@ +%% 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_exchange_type). + +-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([register/2, lookup_module/1, lookup_name/1]). + +-define(SERVER, ?MODULE). + +%%--------------------------------------------------------------------------- + +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). + +%%--------------------------------------------------------------------------- + +register(TypeName, ModuleName) -> + gen_server:call(?SERVER, {register, TypeName, ModuleName}). + +lookup_module(T) when is_binary(T) -> + case ets:lookup(rabbit_exchange_type_modules, T) of + [{_, Module}] -> + {ok, Module}; + [] -> + {error, not_found} + end. + +lookup_name(M) when is_atom(M) -> + [{_, TypeName}] = ets:lookup(rabbit_exchange_type_names, M), + {ok, TypeName}. + +%%--------------------------------------------------------------------------- + +internal_register(TypeName, ModuleName) + when is_binary(TypeName), is_atom(ModuleName) -> + true = ets:insert(rabbit_exchange_type_modules, {TypeName, ModuleName}), + true = ets:insert(rabbit_exchange_type_names, {ModuleName, TypeName}), + ok. + +%%--------------------------------------------------------------------------- + +init([]) -> + rabbit_exchange_type_modules = + ets:new(rabbit_exchange_type_modules, [protected, set, named_table]), + rabbit_exchange_type_names = + ets:new(rabbit_exchange_type_names, [protected, set, named_table]), + + %% TODO: split out into separate boot startup steps. + ok = internal_register(<<"direct">>, rabbit_exchange_type_direct), + ok = internal_register(<<"fanout">>, rabbit_exchange_type_fanout), + ok = internal_register(<<"headers">>, rabbit_exchange_type_headers), + ok = internal_register(<<"topic">>, rabbit_exchange_type_topic), + + {ok, none}. + +handle_call({register, TypeName, ModuleName}, _From, State) -> + ok = internal_register(TypeName, ModuleName), + {reply, ok, State}; +handle_call(Request, _From, State) -> + {stop, {unhandled_call, Request}, State}. + +handle_cast(Request, State) -> + {stop, {unhandled_cast, Request}, State}. + +handle_info(Message, State) -> + {stop, {unhandled_info, Message}, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl index 1bc17a32..3a0f9240 100644 --- a/src/rabbit_networking.erl +++ b/src/rabbit_networking.erl @@ -53,6 +53,9 @@ %% {delay_send, true}, {exit_on_close, false} ]). + +-define(SSL_TIMEOUT, 5). %% seconds + %%---------------------------------------------------------------------------- -ifdef(use_specs). @@ -160,36 +163,31 @@ node_listeners(Node) -> on_node_down(Node) -> ok = mnesia:dirty_delete(rabbit_listener, Node). -start_client(Sock) -> +start_client(Sock, SockTransform) -> {ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []), ok = rabbit_net:controlling_process(Sock, Child), - Child ! {go, Sock}, + Child ! {go, Sock, SockTransform}, Child. +start_client(Sock) -> + start_client(Sock, fun (S) -> {ok, S} end). + start_ssl_client(SslOpts, Sock) -> - case rabbit_net:peername(Sock) of - {ok, {PeerAddress, PeerPort}} -> - PeerIp = inet_parse:ntoa(PeerAddress), - case ssl:ssl_accept(Sock, SslOpts) of - {ok, SslSock} -> - rabbit_log:info("upgraded TCP connection " - "from ~s:~p to SSL~n", - [PeerIp, PeerPort]), - RabbitSslSock = #ssl_socket{tcp = Sock, ssl = SslSock}, - start_client(RabbitSslSock); - {error, Reason} -> - gen_tcp:close(Sock), - rabbit_log:error("failed to upgrade TCP connection " - "from ~s:~p to SSL: ~n~p~n", - [PeerIp, PeerPort, Reason]), - {error, Reason} - end; - {error, Reason} -> - gen_tcp:close(Sock), - rabbit_log:error("failed to upgrade TCP connection to SSL: ~p~n", - [Reason]), - {error, Reason} - end. + start_client( + Sock, + fun (Sock1) -> + case catch ssl:ssl_accept(Sock1, SslOpts, ?SSL_TIMEOUT * 1000) of + {ok, SslSock} -> + rabbit_log:info("upgraded TCP connection ~p to SSL~n", + [self()]), + {ok, #ssl_socket{tcp = Sock1, ssl = SslSock}}; + {error, Reason} -> + {error, {ssl_upgrade_error, Reason}}; + {'EXIT', Reason} -> + {error, {ssl_upgrade_failure, Reason}} + + end + end). connections() -> [Pid || {_, Pid, _, _} <- supervisor:which_children( diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_plugin_activator.erl index a2ac74ef..9f787920 100644 --- a/src/rabbit_plugin_activator.erl +++ b/src/rabbit_plugin_activator.erl @@ -44,7 +44,7 @@ -ifdef(use_specs). --spec(start/0 :: () -> no_return()). +-spec(start/0 :: () -> no_return()). -spec(stop/0 :: () -> 'ok'). -endif. @@ -73,7 +73,7 @@ start() -> %% Build the entire set of dependencies - this will load the %% applications along the way AllApps = case catch sets:to_list(expand_dependencies(RequiredApps)) of - {failed_to_load_app, App, Err} -> + {failed_to_load_app, App, Err} -> error("failed to load application ~s:~n~p", [App, Err]); AppList -> AppList @@ -82,8 +82,8 @@ start() -> {rabbit, RabbitVersion} = proplists:lookup(rabbit, AppVersions), %% Build the overall release descriptor - RDesc = {release, - {"rabbit", RabbitVersion}, + RDesc = {release, + {"rabbit", RabbitVersion}, {erts, erlang:system_info(version)}, AppVersions}, @@ -93,15 +93,15 @@ start() -> %% Compile the script ScriptFile = RootName ++ ".script", case systools:make_script(RootName, [local, silent]) of - {ok, Module, Warnings} -> + {ok, Module, Warnings} -> %% This gets lots of spurious no-source warnings when we %% have .ez files, so we want to supress them to prevent %% hiding real issues. WarningStr = Module:format_warning( - [W || W <- Warnings, - case W of - {warning, {source_not_found, _}} -> false; - _ -> true + [W || W <- Warnings, + case W of + {warning, {source_not_found, _}} -> false; + _ -> true end]), case length(WarningStr) of 0 -> ok; @@ -136,8 +136,8 @@ get_env(Key, Default) -> end. determine_version(App) -> - application:load(App), - {ok, Vsn} = application:get_key(App, vsn), + application:load(App), + {ok, Vsn} = application:get_key(App, vsn), {App, Vsn}. assert_dir(Dir) -> @@ -236,7 +236,7 @@ post_process_script(ScriptFile) -> {error, {failed_to_load_script, Reason}} end. -process_entries([]) -> +process_entries([]) -> []; process_entries([Entry = {apply,{application,start_boot,[stdlib,permanent]}} | Rest]) -> diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index e21485b5..e78d889d 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -58,7 +58,7 @@ -define(INFO_KEYS, [pid, address, port, peer_address, peer_port, recv_oct, recv_cnt, send_oct, send_cnt, send_pend, - state, channels, user, vhost, timeout, frame_max]). + state, channels, user, vhost, timeout, frame_max, client_properties]). %% connection lifecycle %% @@ -142,7 +142,8 @@ start_link() -> init(Parent) -> Deb = sys:debug_options([]), receive - {go, Sock} -> start_connection(Parent, Deb, Sock) + {go, Sock, SockTransform} -> + start_connection(Parent, Deb, Sock, SockTransform) end. system_continue(Parent, Deb, State) -> @@ -192,34 +193,35 @@ teardown_profiling(Value) -> inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F). -peername(Sock) -> - try - {Address, Port} = inet_op(fun () -> rabbit_net:peername(Sock) end), - AddressS = inet_parse:ntoa(Address), - {AddressS, Port} - catch - Ex -> rabbit_log:error("error on TCP connection ~p:~p~n", - [self(), Ex]), - rabbit_log:info("closing TCP connection ~p", [self()]), - exit(normal) +socket_op(Sock, Fun) -> + case Fun(Sock) of + {ok, Res} -> Res; + {error, Reason} -> rabbit_log:error("error on TCP connection ~p:~p~n", + [self(), Reason]), + rabbit_log:info("closing TCP connection ~p~n", + [self()]), + exit(normal) end. -start_connection(Parent, Deb, ClientSock) -> +start_connection(Parent, Deb, Sock, SockTransform) -> process_flag(trap_exit, true), - {PeerAddressS, PeerPort} = peername(ClientSock), + {PeerAddress, PeerPort} = socket_op(Sock, fun rabbit_net:peername/1), + PeerAddressS = inet_parse:ntoa(PeerAddress), + rabbit_log:info("starting TCP connection ~p from ~s:~p~n", + [self(), PeerAddressS, PeerPort]), + ClientSock = socket_op(Sock, SockTransform), + erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(), + handshake_timeout), ProfilingValue = setup_profiling(), try - rabbit_log:info("starting TCP connection ~p from ~s:~p~n", - [self(), PeerAddressS, PeerPort]), - erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(), - handshake_timeout), mainloop(Parent, Deb, switch_callback( #v1{sock = ClientSock, connection = #connection{ user = none, timeout_sec = ?HANDSHAKE_TIMEOUT, frame_max = ?FRAME_MIN_SIZE, - vhost = none}, + vhost = none, + client_properties = none}, callback = uninitialized_callback, recv_ref = none, connection_state = pre_init}, @@ -558,7 +560,8 @@ handle_method0(MethodName, FieldsBin, State) -> end. handle_method0(#'connection.start_ok'{mechanism = Mechanism, - response = Response}, + response = Response, + client_properties = ClientProperties}, State = #v1{connection_state = starting, connection = Connection, sock = Sock}) -> @@ -570,7 +573,9 @@ handle_method0(#'connection.start_ok'{mechanism = Mechanism, frame_max = 131072, heartbeat = 0}), State#v1{connection_state = tuning, - connection = Connection#connection{user = User}}; + connection = Connection#connection{ + user = User, + client_properties = ClientProperties}}; handle_method0(#'connection.tune_ok'{channel_max = _ChannelMax, frame_max = FrameMax, heartbeat = ClientHeartbeat}, @@ -689,6 +694,9 @@ i(timeout, #v1{connection = #connection{timeout_sec = Timeout}}) -> Timeout; i(frame_max, #v1{connection = #connection{frame_max = FrameMax}}) -> FrameMax; +i(client_properties, #v1{connection = #connection{ + client_properties = ClientProperties}}) -> + ClientProperties; i(Item, #v1{}) -> throw({bad_argument, Item}). diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl index 6da47933..8be28f52 100644 --- a/src/vm_memory_monitor.erl +++ b/src/vm_memory_monitor.erl @@ -34,7 +34,7 @@ %% 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. %% @@ -42,7 +42,7 @@ -module(vm_memory_monitor). --behaviour(gen_server2). +-behaviour(gen_server). -export([start_link/1]). @@ -73,9 +73,10 @@ -ifdef(use_specs). --spec(start_link/1 :: (float()) -> ('ignore' | {error, any()} | {'ok', pid()})). +-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_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()). @@ -85,25 +86,48 @@ %%---------------------------------------------------------------------------- +%% Public API +%%---------------------------------------------------------------------------- + +update() -> + gen_server:cast(?SERVER, update). + +get_total_memory() -> + get_total_memory(os:type()). + +get_check_interval() -> + gen_server:call(?MODULE, get_check_interval). + +set_check_interval(Fraction) -> + gen_server:call(?MODULE, {set_check_interval, Fraction}). + +get_vm_memory_high_watermark() -> + gen_server:call(?MODULE, get_vm_memory_high_watermark). + +set_vm_memory_high_watermark(Fraction) -> + gen_server:call(?MODULE, {set_vm_memory_high_watermark, Fraction}). + +%%---------------------------------------------------------------------------- %% gen_server callbacks %%---------------------------------------------------------------------------- start_link(Args) -> - gen_server2:start_link({local, ?SERVER}, ?MODULE, [Args], []). + gen_server:start_link({local, ?SERVER}, ?MODULE, [Args], []). -init([MemFraction]) -> +init([MemFraction]) -> TotalMemory = case get_total_memory() of unknown -> - rabbit_log:warning( + error_logger:warning_msg( "Unknown total memory size for your OS ~p. " - "Assuming memory size is ~pMB.~n", + "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)]), + error_logger:info_msg("Memory limit set to ~pMB.~n", + [trunc(MemLimit/1048576)]), TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL), State = #state { total_memory = TotalMemory, memory_limit = MemLimit, @@ -117,8 +141,8 @@ handle_call(get_vm_memory_high_watermark, _From, 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]), + error_logger:info_msg("Memory alarm changed to ~p, ~p bytes.~n", + [MemFraction, MemLimit]), {reply, ok, State#state{memory_limit = MemLimit}}; handle_call(get_check_interval, _From, State) -> @@ -134,41 +158,19 @@ handle_call(_Request, _From, State) -> handle_cast(update, State) -> {noreply, internal_update(State)}; -handle_cast(_Request, State) -> +handle_cast(_Request, State) -> {noreply, State}. -handle_info(_Info, State) -> +handle_info(_Info, State) -> {noreply, State}. -terminate(_Reason, _State) -> +terminate(_Reason, _State) -> ok. -code_change(_OldVsn, State, _Extra) -> +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 %%---------------------------------------------------------------------------- @@ -189,8 +191,9 @@ internal_update(State = #state { memory_limit = MemLimit, 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]). + error_logger:info_msg( + "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, []), @@ -202,7 +205,7 @@ 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 + %%http://en.wikipedia.org/wiki/X86-64#Virtual_address_space_details end. get_mem_limit(MemFraction, TotalMemory) -> |