diff options
author | Simon MacMullen <simon@rabbitmq.com> | 2014-06-24 16:19:01 +0100 |
---|---|---|
committer | Simon MacMullen <simon@rabbitmq.com> | 2014-06-24 16:19:01 +0100 |
commit | a61daae11b1b259317d89518faf26276d5b970d5 (patch) | |
tree | b4211999c851b2b1efb9814f1378dd36d21084dc | |
parent | fc6737d5450fea892694ae25546cf791150207c2 (diff) | |
parent | 61dd82a5f1e3c918aa1a108dcf868a06bbc84bed (diff) | |
download | rabbitmq-server-a61daae11b1b259317d89518faf26276d5b970d5.tar.gz |
Merge bug26148
-rw-r--r-- | docs/rabbitmq.config.example | 9 | ||||
-rw-r--r-- | ebin/rabbit_app.in | 2 | ||||
-rw-r--r-- | include/rabbit.hrl | 1 | ||||
-rw-r--r-- | packaging/RPMS/Fedora/rabbitmq-server.spec | 3 | ||||
-rw-r--r-- | packaging/debs/Debian/debian/changelog | 6 | ||||
-rwxr-xr-x | scripts/rabbitmq-env | 5 | ||||
-rwxr-xr-x | scripts/rabbitmq-server | 3 | ||||
-rwxr-xr-x | scripts/rabbitmq-server.bat | 1 | ||||
-rwxr-xr-x | scripts/rabbitmq-service.bat | 1 | ||||
-rw-r--r-- | src/rabbit.erl | 27 | ||||
-rw-r--r-- | src/rabbit_amqqueue.erl | 67 | ||||
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 9 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_misc.erl | 32 | ||||
-rw-r--r-- | src/rabbit_misc.erl | 7 | ||||
-rw-r--r-- | src/rabbit_networking.erl | 11 | ||||
-rw-r--r-- | src/rabbit_node_monitor.erl | 1 | ||||
-rw-r--r-- | src/rabbit_prelaunch.erl | 4 | ||||
-rw-r--r-- | src/rabbit_reader.erl | 6 | ||||
-rw-r--r-- | src/rabbit_upgrade_functions.erl | 18 |
19 files changed, 170 insertions, 43 deletions
diff --git a/docs/rabbitmq.config.example b/docs/rabbitmq.config.example index 4fad1542..e8b56660 100644 --- a/docs/rabbitmq.config.example +++ b/docs/rabbitmq.config.example @@ -27,6 +27,11 @@ %% %% {ssl_listeners, [5671]}, + %% Maximum time for AMQP 0-8/0-9/0-9-1 handshake (after socket connection + %% and SSL handshake), in milliseconds. + %% + %% {handshake_timeout, 10000}, + %% Log levels (currently just used for connection logging). %% One of 'info', 'warning', 'error' or 'none', in decreasing order %% of verbosity. Defaults to 'info'. @@ -103,6 +108,10 @@ %% %% {ssl_cert_login_from, common_name}, + %% SSL handshake timeout, in milliseconds. + %% + %% {ssl_handshake_timeout, 5000}, + %% %% Default User / VHost %% ==================== diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in index 3647c04a..f26e0f77 100644 --- a/ebin/rabbit_app.in +++ b/ebin/rabbit_app.in @@ -46,6 +46,8 @@ {trace_vhosts, []}, {log_levels, [{connection, info}]}, {ssl_cert_login_from, distinguished_name}, + {ssl_handshake_timeout, 5000}, + {handshake_timeout, 10000}, {reverse_dns_lookups, false}, {cluster_partition_handling, ignore}, {tcp_listen_options, [binary, diff --git a/include/rabbit.hrl b/include/rabbit.hrl index c1386803..7a40f9eb 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -52,6 +52,7 @@ arguments, %% immutable pid, %% durable (just so we know home node) slave_pids, sync_slave_pids, %% transient + down_slave_nodes, %% durable policy, %% durable, implicit update as above gm_pids, %% transient decorators}). %% transient, recalculated as above diff --git a/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec index 682c10ee..32404057 100644 --- a/packaging/RPMS/Fedora/rabbitmq-server.spec +++ b/packaging/RPMS/Fedora/rabbitmq-server.spec @@ -130,6 +130,9 @@ done rm -rf %{buildroot} %changelog +* Tue Jun 24 2014 simon@rabbitmq.com 3.3.4-1 +- New Upstream Release + * Mon Jun 16 2014 simon@rabbitmq.com 3.3.3-1 - New Upstream Release diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog index 84e2afe7..d26991e4 100644 --- a/packaging/debs/Debian/debian/changelog +++ b/packaging/debs/Debian/debian/changelog @@ -1,3 +1,9 @@ +rabbitmq-server (3.3.4-1) unstable; urgency=low + + * New Upstream Release + + -- Simon MacMullen <simon@rabbitmq.com> Tue, 24 Jun 2014 12:50:29 +0100 + rabbitmq-server (3.3.3-1) unstable; urgency=low * New Upstream Release diff --git a/scripts/rabbitmq-env b/scripts/rabbitmq-env index b7741670..69d5a9c9 100755 --- a/scripts/rabbitmq-env +++ b/scripts/rabbitmq-env @@ -15,10 +15,14 @@ ## Copyright (c) 2007-2014 GoPivotal, Inc. All rights reserved. ## +# We set +e here since since our test for "readlink -f" below needs to +# be able to fail. +set +e # Determine where this script is really located (if this script is # invoked from another script, this is the location of the caller) SCRIPT_PATH="$0" while [ -h "$SCRIPT_PATH" ] ; do + # Determine if readlink -f is supported at all. TODO clean this up. FULL_PATH=`readlink -f $SCRIPT_PATH 2>/dev/null` if [ "$?" != "0" ]; then REL_PATH=`readlink $SCRIPT_PATH` @@ -31,6 +35,7 @@ while [ -h "$SCRIPT_PATH" ] ; do SCRIPT_PATH=$FULL_PATH fi done +set -e SCRIPT_DIR=`dirname $SCRIPT_PATH` RABBITMQ_HOME="${SCRIPT_DIR}/.." diff --git a/scripts/rabbitmq-server b/scripts/rabbitmq-server index bd397441..2dbda427 100755 --- a/scripts/rabbitmq-server +++ b/scripts/rabbitmq-server @@ -39,7 +39,7 @@ DEFAULT_NODE_PORT=5672 [ "x" = "x$RABBITMQ_LOG_BASE" ] && RABBITMQ_LOG_BASE=${LOG_BASE} [ "x" = "x$RABBITMQ_MNESIA_BASE" ] && RABBITMQ_MNESIA_BASE=${MNESIA_BASE} [ "x" = "x$RABBITMQ_SERVER_START_ARGS" ] && RABBITMQ_SERVER_START_ARGS=${SERVER_START_ARGS} - +[ "x" = "x$RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS" ] && RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=${SERVER_ADDITIONAL_ERL_ARGS} [ "x" = "x$RABBITMQ_MNESIA_DIR" ] && RABBITMQ_MNESIA_DIR=${MNESIA_DIR} [ "x" = "x$RABBITMQ_MNESIA_DIR" ] && RABBITMQ_MNESIA_DIR=${RABBITMQ_MNESIA_BASE}/${RABBITMQ_NODENAME} @@ -131,6 +131,7 @@ exec ${ERL_DIR}erl \ ${RABBITMQ_CONFIG_ARG} \ +W w \ ${RABBITMQ_SERVER_ERL_ARGS} \ + ${RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS} \ ${RABBITMQ_LISTEN_ARG} \ -sasl errlog_type error \ -sasl sasl_error_logger false \ diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat index 043204fa..e2312406 100755 --- a/scripts/rabbitmq-server.bat +++ b/scripts/rabbitmq-server.bat @@ -147,6 +147,7 @@ if not "!RABBITMQ_NODE_IP_ADDRESS!"=="" ( -kernel inet_default_connect_options "[{nodelay, true}]" ^
!RABBITMQ_LISTEN_ARG! ^
!RABBITMQ_SERVER_ERL_ARGS! ^
+!RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS! ^
-sasl errlog_type error ^
-sasl sasl_error_logger false ^
-rabbit error_logger {file,\""!LOGS:\=/!"\"} ^
diff --git a/scripts/rabbitmq-service.bat b/scripts/rabbitmq-service.bat index 895561d4..fb2703f2 100755 --- a/scripts/rabbitmq-service.bat +++ b/scripts/rabbitmq-service.bat @@ -235,6 +235,7 @@ set ERLANG_SERVICE_ARGUMENTS= ^ -os_mon start_memsup false ^
-mnesia dir \""!RABBITMQ_MNESIA_DIR:\=/!"\" ^
!RABBITMQ_SERVER_START_ARGS! ^
+!RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS! ^
!RABBITMQ_DIST_ARG! ^
!STARVAR!
diff --git a/src/rabbit.erl b/src/rabbit.erl index 4901ea17..4b7a9a1f 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -201,6 +201,7 @@ %% practice 2 processes seems just as fast as any other number > 1, %% and keeps the progress bar realistic-ish. -define(HIPE_PROCESSES, 2). +-define(ASYNC_THREADS_WARNING_THRESHOLD, 8). %%---------------------------------------------------------------------------- @@ -491,6 +492,7 @@ start(normal, []) -> true = register(rabbit, self()), print_banner(), log_banner(), + warn_if_kernel_config_dubious(), run_boot_steps(), {ok, SupPid}; Error -> @@ -816,6 +818,31 @@ log_banner() -> end || S <- Settings]), error_logger:info_msg("~s", [Banner]). +warn_if_kernel_config_dubious() -> + case erlang:system_info(kernel_poll) of + true -> ok; + false -> error_logger:warning_msg( + "Kernel poll (epoll, kqueue, etc) is disabled. Throughput " + "and CPU utilization may worsen.~n") + end, + AsyncThreads = erlang:system_info(thread_pool_size), + case AsyncThreads < ?ASYNC_THREADS_WARNING_THRESHOLD of + true -> error_logger:warning_msg( + "Erlang VM is running with ~b I/O threads, " + "file I/O performance may worsen~n", [AsyncThreads]); + false -> ok + end, + IDCOpts = case application:get_env(kernel, inet_default_connect_options) of + undefined -> []; + {ok, Val} -> Val + end, + case proplists:get_value(nodelay, IDCOpts, false) of + false -> error_logger:warning_msg( + "Nagle's algorithm is enabled for sockets, " + "network I/O latency will be higher~n"); + true -> ok + end. + home_dir() -> case init:get_argument(home) of {ok, [[Home]]} -> Home; diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 8a1d162a..4e23dbd2 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -29,7 +29,7 @@ -export([basic_get/4, basic_consume/10, basic_cancel/4, notify_decorators/1]). -export([notify_sent/2, notify_sent_queue_down/1, resume/2]). -export([notify_down_all/2, activate_limit_all/2, credit/5]). --export([on_node_down/1]). +-export([on_node_up/1, on_node_down/1]). -export([update/2, store_queue/1, update_decorators/1, policy_changed/2]). -export([start_mirroring/1, stop_mirroring/1, sync_mirrors/1, cancel_sync_mirrors/1]). @@ -174,6 +174,7 @@ (fun ((atom(), A) -> {[rabbit_types:msg_id()], A}))) -> 'ok'). -spec(set_ram_duration_target/2 :: (pid(), number() | 'infinity') -> 'ok'). -spec(set_maximum_since_use/2 :: (pid(), non_neg_integer()) -> 'ok'). +-spec(on_node_up/1 :: (node()) -> 'ok'). -spec(on_node_down/1 :: (node()) -> 'ok'). -spec(pseudo_queue/2 :: (name(), pid()) -> rabbit_types:amqqueue()). -spec(immutable/1 :: (rabbit_types:amqqueue()) -> rabbit_types:amqqueue()). @@ -257,15 +258,16 @@ declare(QueueName, Durable, AutoDelete, Args, Owner) -> declare(QueueName, Durable, AutoDelete, Args, Owner, Node) -> ok = check_declare_arguments(QueueName, Args), Q = rabbit_queue_decorator:set( - rabbit_policy:set(#amqqueue{name = QueueName, - durable = Durable, - auto_delete = AutoDelete, - arguments = Args, - exclusive_owner = Owner, - pid = none, - slave_pids = [], - sync_slave_pids = [], - gm_pids = []})), + rabbit_policy:set(#amqqueue{name = QueueName, + durable = Durable, + auto_delete = AutoDelete, + arguments = Args, + exclusive_owner = Owner, + pid = none, + slave_pids = [], + sync_slave_pids = [], + down_slave_nodes = [], + gm_pids = []})), Node = rabbit_mirror_queue_misc:initial_queue_node(Q, Node), gen_server2:call(start_queue_process(Node, Q), {init, new}, infinity). @@ -665,15 +667,23 @@ forget_all_durable(Node) -> fun () -> Qs = mnesia:match_object(rabbit_durable_queue, #amqqueue{_ = '_'}, write), - [rabbit_binding:process_deletions( - internal_delete1(Name)) || - #amqqueue{name = Name, pid = Pid} = Q <- Qs, - node(Pid) =:= Node, - rabbit_policy:get(<<"ha-mode">>, Q) =:= undefined], + [forget_node_for_queue(Q) || #amqqueue{pid = Pid} = Q <- Qs, + node(Pid) =:= Node], ok end), ok. +forget_node_for_queue(#amqqueue{name = Name, + down_slave_nodes = []}) -> + %% No slaves to recover from, queue is gone + rabbit_binding:process_deletions(internal_delete1(Name)); + +forget_node_for_queue(Q = #amqqueue{down_slave_nodes = [H|T]}) -> + %% Promote a slave while down - it'll happily recover as a master + Q1 = Q#amqqueue{pid = rabbit_misc:node_to_fake_pid(H), + down_slave_nodes = T}, + ok = mnesia:write(rabbit_durable_queue, Q1, write). + run_backing_queue(QPid, Mod, Fun) -> gen_server2:cast(QPid, {run_backing_queue, Mod, Fun}). @@ -689,6 +699,20 @@ stop_mirroring(QPid) -> ok = delegate:cast(QPid, stop_mirroring). sync_mirrors(QPid) -> delegate:call(QPid, sync_mirrors). cancel_sync_mirrors(QPid) -> delegate:call(QPid, cancel_sync_mirrors). +on_node_up(Node) -> + ok = rabbit_misc:execute_mnesia_transaction( + fun () -> + Qs = mnesia:match_object(rabbit_queue, + #amqqueue{_ = '_'}, write), + [case lists:member(Node, DSNs) of + true -> DSNs1 = DSNs -- [Node], + store_queue( + Q#amqqueue{down_slave_nodes = DSNs1}); + false -> ok + end || #amqqueue{down_slave_nodes = DSNs} = Q <- Qs], + ok + end). + on_node_down(Node) -> rabbit_misc:execute_mnesia_tx_with_tail( fun () -> QsDels = @@ -724,12 +748,13 @@ pseudo_queue(QueueName, Pid) -> pid = Pid, slave_pids = []}. -immutable(Q) -> Q#amqqueue{pid = none, - slave_pids = none, - sync_slave_pids = none, - gm_pids = none, - policy = none, - decorators = none}. +immutable(Q) -> Q#amqqueue{pid = none, + slave_pids = none, + sync_slave_pids = none, + down_slave_nodes = none, + gm_pids = none, + policy = none, + decorators = none}. deliver([], _Delivery, _Flow) -> %% /dev/null optimisation diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 97206df3..4082c53d 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -84,6 +84,7 @@ memory, slave_pids, synchronised_slave_pids, + down_slave_nodes, backing_queue_status, state ]). @@ -810,6 +811,14 @@ i(synchronised_slave_pids, #q{q = #amqqueue{name = Name}}) -> false -> ''; true -> SSPids end; +i(down_slave_nodes, #q{q = #amqqueue{name = Name, + durable = Durable}}) -> + {ok, Q = #amqqueue{down_slave_nodes = Nodes}} = + rabbit_amqqueue:lookup(Name), + case Durable andalso rabbit_mirror_queue_misc:is_mirrored(Q) of + false -> ''; + true -> Nodes + end; i(state, #q{status = running}) -> credit_flow:state(); i(state, #q{status = State}) -> State; i(backing_queue_status, #q{backing_queue_state = BQS, backing_queue = BQ}) -> diff --git a/src/rabbit_mirror_queue_misc.erl b/src/rabbit_mirror_queue_misc.erl index 7aec1ac8..9e8c4a18 100644 --- a/src/rabbit_mirror_queue_misc.erl +++ b/src/rabbit_mirror_queue_misc.erl @@ -78,9 +78,10 @@ remove_from_queue(QueueName, Self, DeadGMPids) -> %% get here. case mnesia:read({rabbit_queue, QueueName}) of [] -> {error, not_found}; - [Q = #amqqueue { pid = QPid, - slave_pids = SPids, - gm_pids = GMPids }] -> + [Q = #amqqueue { pid = QPid, + slave_pids = SPids, + gm_pids = GMPids, + down_slave_nodes = DSNs}] -> {DeadGM, AliveGM} = lists:partition( fun ({GM, _}) -> lists:member(GM, DeadGMPids) @@ -89,6 +90,9 @@ remove_from_queue(QueueName, Self, DeadGMPids) -> AlivePids = [Pid || {_GM, Pid} <- AliveGM], Alive = [Pid || Pid <- [QPid | SPids], lists:member(Pid, AlivePids)], + DSNs1 = [node(Pid) || + Pid <- SPids, + not lists:member(Pid, AlivePids)] ++ DSNs, {QPid1, SPids1} = promote_slave(Alive), case {{QPid, SPids}, {QPid1, SPids1}} of {Same, Same} -> @@ -97,9 +101,10 @@ remove_from_queue(QueueName, Self, DeadGMPids) -> %% Either master hasn't changed, so %% we're ok to update mnesia; or we have %% become the master. - Q1 = Q#amqqueue{pid = QPid1, - slave_pids = SPids1, - gm_pids = AliveGM}, + Q1 = Q#amqqueue{pid = QPid1, + slave_pids = SPids1, + gm_pids = AliveGM, + down_slave_nodes = DSNs1}, store_updated_slaves(Q1), %% If we add and remove nodes at the same time we %% might tell the old master we need to sync and @@ -109,8 +114,9 @@ remove_from_queue(QueueName, Self, DeadGMPids) -> _ -> %% Master has changed, and we're not it. %% [1]. - Q1 = Q#amqqueue{slave_pids = Alive, - gm_pids = AliveGM}, + Q1 = Q#amqqueue{slave_pids = Alive, + gm_pids = AliveGM, + down_slave_nodes = DSNs1}, store_updated_slaves(Q1) end, {ok, QPid1, DeadPids} @@ -239,12 +245,16 @@ log(Level, QName, Fmt, Args) -> rabbit_log:log(mirroring, Level, "Mirrored ~s: " ++ Fmt, [rabbit_misc:rs(QName) | Args]). -store_updated_slaves(Q = #amqqueue{slave_pids = SPids, - sync_slave_pids = SSPids}) -> +store_updated_slaves(Q = #amqqueue{pid = MPid, + slave_pids = SPids, + sync_slave_pids = SSPids, + down_slave_nodes = DSNs}) -> %% TODO now that we clear sync_slave_pids in rabbit_durable_queue, %% do we still need this filtering? SSPids1 = [SSPid || SSPid <- SSPids, lists:member(SSPid, SPids)], - Q1 = Q#amqqueue{sync_slave_pids = SSPids1}, + DSNs1 = DSNs -- [node(P) || P <- [MPid | SPids]], + Q1 = Q#amqqueue{sync_slave_pids = SSPids1, + down_slave_nodes = DSNs1}, ok = rabbit_amqqueue:store_queue(Q1), %% Wake it up so that we emit a stats event rabbit_amqqueue:notify_policy_changed(Q1), diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl index 6f353da5..fd4b7b11 100644 --- a/src/rabbit_misc.erl +++ b/src/rabbit_misc.erl @@ -45,7 +45,7 @@ -export([with_local_io/1, local_info_msg/2]). -export([unfold/2, ceil/1, queue_fold/3]). -export([sort_field_table/1]). --export([pid_to_string/1, string_to_pid/1]). +-export([pid_to_string/1, string_to_pid/1, node_to_fake_pid/1]). -export([version_compare/2, version_compare/3]). -export([version_minor_equivalent/2]). -export([dict_cons/3, orddict_cons/3, gb_trees_cons/3]). @@ -193,6 +193,7 @@ (rabbit_framing:amqp_table()) -> rabbit_framing:amqp_table()). -spec(pid_to_string/1 :: (pid()) -> string()). -spec(string_to_pid/1 :: (string()) -> pid()). +-spec(node_to_fake_pid/1 :: (atom()) -> pid()). -spec(version_compare/2 :: (string(), string()) -> 'lt' | 'eq' | 'gt'). -spec(version_compare/3 :: (string(), string(), ('lt' | 'lte' | 'eq' | 'gte' | 'gt')) @@ -709,6 +710,10 @@ string_to_pid(Str) -> throw(Err) end. +%% node(node_to_fake_pid(Node)) =:= Node. +node_to_fake_pid(Node) -> + string_to_pid(format("<~s.0.0.0>", [Node])). + version_compare(A, B, lte) -> case version_compare(A, B) of eq -> true; diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl index 9082dbd3..96448f32 100644 --- a/src/rabbit_networking.erl +++ b/src/rabbit_networking.erl @@ -37,8 +37,6 @@ -include("rabbit.hrl"). -include_lib("kernel/include/inet.hrl"). --define(SSL_TIMEOUT, 5). %% seconds - -define(FIRST_TEST_BIND_PORT, 10000). %%---------------------------------------------------------------------------- @@ -168,9 +166,14 @@ ensure_ssl() -> end end. +ssl_timeout() -> + {ok, Val} = application:get_env(rabbit, ssl_handshake_timeout), + Val. + ssl_transform_fun(SslOpts) -> fun (Sock) -> - case catch ssl:ssl_accept(Sock, SslOpts, ?SSL_TIMEOUT * 1000) of + Timeout = ssl_timeout(), + case catch ssl:ssl_accept(Sock, SslOpts, Timeout) of {ok, SslSock} -> {ok, #ssl_socket{tcp = Sock, ssl = SslSock}}; {error, timeout} -> @@ -185,7 +188,7 @@ ssl_transform_fun(SslOpts) -> %% form, according to the TLS spec). So we give %% the ssl_connection a little bit of time to send %% such alerts. - timer:sleep(?SSL_TIMEOUT * 1000), + timer:sleep(Timeout), {error, {ssl_upgrade_error, Reason}}; {'EXIT', Reason} -> {error, {ssl_upgrade_failure, Reason}} diff --git a/src/rabbit_node_monitor.erl b/src/rabbit_node_monitor.erl index 14961478..1c971c1d 100644 --- a/src/rabbit_node_monitor.erl +++ b/src/rabbit_node_monitor.erl @@ -415,6 +415,7 @@ ensure_ping_timer(State) -> State, #state.down_ping_timer, ?RABBIT_DOWN_PING_INTERVAL, ping_nodes). handle_live_rabbit(Node) -> + ok = rabbit_amqqueue:on_node_up(Node), ok = rabbit_alarm:on_node_up(Node), ok = rabbit_mnesia:on_node_up(Node). diff --git a/src/rabbit_prelaunch.erl b/src/rabbit_prelaunch.erl index 4cc9cd12..6a6a4ee6 100644 --- a/src/rabbit_prelaunch.erl +++ b/src/rabbit_prelaunch.erl @@ -90,9 +90,9 @@ dist_port_set_check() -> {none, none} -> ok; _ -> rabbit_misc:quit(?DIST_PORT_CONFIGURED) end; + {ok, _} -> + ok; {error, _} -> - %% TODO can we present errors more nicely here - %% than after -config has failed? ok end end. diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index c121f10e..db6d1eb0 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -27,7 +27,6 @@ -export([conserve_resources/3, server_properties/1]). --define(HANDSHAKE_TIMEOUT, 10). -define(NORMAL_TIMEOUT, 3). -define(CLOSING_TIMEOUT, 30). -define(CHANNEL_TERMINATION_TIMEOUT, 3). @@ -216,8 +215,9 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) -> exit(normal) end, log(info, "accepting AMQP connection ~p (~s)~n", [self(), Name]), + {ok, HandshakeTimeout} = application:get_env(rabbit, handshake_timeout), ClientSock = socket_op(Sock, SockTransform), - erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(), handshake_timeout), + erlang:send_after(HandshakeTimeout, self(), handshake_timeout), {PeerHost, PeerPort, Host, Port} = socket_op(Sock, fun (S) -> rabbit_net:socket_ends(S, inbound) end), ?store_proc_name(list_to_binary(Name)), @@ -231,7 +231,7 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) -> peer_port = PeerPort, protocol = none, user = none, - timeout_sec = ?HANDSHAKE_TIMEOUT, + timeout_sec = (HandshakeTimeout / 1000), frame_max = ?FRAME_MIN_SIZE, vhost = none, client_properties = none, diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl index b6d37852..1104f373 100644 --- a/src/rabbit_upgrade_functions.erl +++ b/src/rabbit_upgrade_functions.erl @@ -48,6 +48,7 @@ -rabbit_upgrade({queue_decorators, mnesia, [gm_pids]}). -rabbit_upgrade({internal_system_x, mnesia, [exchange_decorators]}). -rabbit_upgrade({cluster_name, mnesia, [runtime_parameters]}). +-rabbit_upgrade({down_slave_nodes, mnesia, [queue_decorators]}). %% ------------------------------------------------------------------- @@ -77,6 +78,8 @@ -spec(policy_apply_to/0 :: () -> 'ok'). -spec(queue_decorators/0 :: () -> 'ok'). -spec(internal_system_x/0 :: () -> 'ok'). +-spec(cluster_name/0 :: () -> 'ok'). +-spec(down_slave_nodes/0 :: () -> 'ok'). -endif. @@ -382,6 +385,21 @@ cluster_name_tx() -> [mnesia:delete(T, K, write) || K <- Ks], ok. +down_slave_nodes() -> + ok = down_slave_nodes(rabbit_queue), + ok = down_slave_nodes(rabbit_durable_queue). + +down_slave_nodes(Table) -> + transform( + Table, + fun ({amqqueue, Name, Durable, AutoDelete, ExclusiveOwner, Arguments, + Pid, SlavePids, SyncSlavePids, Policy, GmPids, Decorators}) -> + {amqqueue, Name, Durable, AutoDelete, ExclusiveOwner, Arguments, + Pid, SlavePids, SyncSlavePids, [], Policy, GmPids, Decorators} + end, + [name, durable, auto_delete, exclusive_owner, arguments, pid, slave_pids, + sync_slave_pids, down_slave_nodes, policy, gm_pids, decorators]). + %%-------------------------------------------------------------------- transform(TableName, Fun, FieldList) -> |