diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit.erl | 14 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_config.erl | 38 | ||||
| -rw-r--r-- | src/rabbit_disk_monitor.erl | 49 | ||||
| -rw-r--r-- | src/rabbit_mirror_queue_master.erl | 28 | ||||
| -rw-r--r-- | src/rabbit_mirror_queue_slave.erl | 56 | ||||
| -rw-r--r-- | src/rabbit_mnesia.erl | 13 | ||||
| -rw-r--r-- | src/rabbit_msg_store.erl | 8 | ||||
| -rw-r--r-- | src/rabbit_msg_store_gc.erl | 8 | ||||
| -rw-r--r-- | src/rabbit_peer_discovery.erl | 119 | ||||
| -rw-r--r-- | src/rabbit_peer_discovery_classic_config.erl | 13 | ||||
| -rw-r--r-- | src/rabbit_peer_discovery_dns.erl | 36 | ||||
| -rw-r--r-- | src/rabbit_queue_consumers.erl | 5 | ||||
| -rw-r--r-- | src/rabbit_queue_index.erl | 22 |
14 files changed, 312 insertions, 99 deletions
diff --git a/src/rabbit.erl b/src/rabbit.erl index b63d624293..8e6c9ead26 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -22,7 +22,7 @@ stop_and_halt/0, await_startup/0, status/0, is_running/0, alarms/0, is_running/1, environment/0, rotate_logs/0, force_event_refresh/1, start_fhc/0]). --export([start/2, stop/1]). +-export([start/2, stop/1, prep_stop/1]). -export([start_apps/1, stop_apps/1]). -export([log_locations/0, config_files/0, decrypt_config/2]). %% for testing and mgmt-agent @@ -327,7 +327,11 @@ broker_start() -> ToBeLoaded = Plugins ++ ?APPS, start_apps(ToBeLoaded), maybe_sd_notify(), - ok = log_broker_started(rabbit_plugins:strictly_plugins(rabbit_plugins:active())). + ok = log_broker_started(rabbit_plugins:strictly_plugins(rabbit_plugins:active())), + %% See rabbitmq/rabbitmq-server#1202 for details. + rabbit_peer_discovery:maybe_inject_randomized_delay(), + rabbit_peer_discovery:maybe_register(), + ok. %% Try to send systemd ready notification if it makes sense in the %% current environment. standard_error is used intentionally in all @@ -471,6 +475,8 @@ stop() -> end, rabbit_log:info("RabbitMQ is asked to stop...~n", []), Apps = ?APPS ++ rabbit_plugins:active(), + %% this will also perform unregistration with the peer discovery backend + %% as needed stop_apps(app_utils:app_dependency_order(Apps, true)), rabbit_log:info("Successfully stopped RabbitMQ and its dependencies~n", []). @@ -759,6 +765,10 @@ start(normal, []) -> Error end. +prep_stop(State) -> + rabbit_peer_discovery:maybe_unregister(), + State. + stop(_State) -> ok = rabbit_alarm:stop(), ok = case rabbit_mnesia:is_clustered() of diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index e9f91041e7..c52d329392 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -99,7 +99,7 @@ -spec info_keys() -> rabbit_types:info_keys(). -spec init_with_backing_queue_state (rabbit_types:amqqueue(), atom(), tuple(), any(), - [rabbit_types:delivery()], pmon:pmon(), dict:dict()) -> + [rabbit_types:delivery()], pmon:pmon(), gb_trees:tree()) -> #q{}. %%---------------------------------------------------------------------------- diff --git a/src/rabbit_config.erl b/src/rabbit_config.erl index 67e7523ec0..9e70898c82 100644 --- a/src/rabbit_config.erl +++ b/src/rabbit_config.erl @@ -33,10 +33,10 @@ prepare_and_use_config() -> legacy_erlang_term_config_used() -> case init:get_argument(config) of error -> false; - {ok, [Config | _]} -> + {ok, [Config | _]} -> ConfigFile = Config ++ ".config", - rabbit_file:is_file(ConfigFile) - andalso + rabbit_file:is_file(ConfigFile) + andalso get_advanced_config() == none end. @@ -66,10 +66,34 @@ prepare_config(Configs) -> end. update_app_config(ConfigFile) -> - ok = application_controller:change_application_data([], [ConfigFile]). + RunningApps = [ App || {App, _, _} <- application:which_applications() ], + LoadedApps = [ App || {App, _, _} <- application:loaded_applications() ], + {ok, [Config]} = file:consult(ConfigFile), + %% For application config to be updated, applications should + %% be unloaded first. + %% If an application is already running, print an error. + lists:foreach(fun({App, _Config}) -> + case lists:member(App, RunningApps) of + true -> + io:format(standard_error, + "~nUnable to update config for app ~p from *.conf file." + " App is already running. Use advanced.config instead.~n", + [App]); + false -> + case lists:member(App, LoadedApps) of + true -> application:unload(App); + false -> ok + end + end + end, + Config), + ok = application_controller:change_application_data([], [ConfigFile]), + %% Make sure to load all the applications we're unloaded + lists:foreach(fun(App) -> application:load(App) end, LoadedApps), + ok. generate_config_file(ConfFiles, ConfDir, ScriptDir) -> - generate_config_file(ConfFiles, ConfDir, ScriptDir, + generate_config_file(ConfFiles, ConfDir, ScriptDir, schema_dir(), get_advanced_config()). @@ -145,8 +169,8 @@ config_files() -> {ok, Files} -> [Abs(File, ".config") || [File] <- Files]; error -> case config_setting() of none -> []; - File -> [Abs(File, ".config") - ++ + File -> [Abs(File, ".config") + ++ " (not found)"] end end; diff --git a/src/rabbit_disk_monitor.erl b/src/rabbit_disk_monitor.erl index b2548cb61a..868fc1a4aa 100644 --- a/src/rabbit_disk_monitor.erl +++ b/src/rabbit_disk_monitor.erl @@ -65,12 +65,17 @@ alarmed, %% is monitoring enabled? false on unsupported %% platforms - enabled + enabled, + %% number of retries to enable monitoring if it fails + %% on start-up + retries, + %% Interval between retries + interval }). %%---------------------------------------------------------------------------- --type disk_free_limit() :: (integer() | string() | {'mem_relative', float()}). +-type disk_free_limit() :: (integer() | string() | {'mem_relative', float() | integer()}). -spec start_link(disk_free_limit()) -> rabbit_types:ok_pid_or_error(). -spec get_disk_free_limit() -> integer(). -spec set_disk_free_limit(disk_free_limit()) -> 'ok'. @@ -114,20 +119,17 @@ start_link(Args) -> init([Limit]) -> Dir = dir(), + {ok, Retries} = application:get_env(rabbit, disk_monitor_failure_retries), + {ok, Interval} = application:get_env(rabbit, disk_monitor_failure_retry_interval), State = #state{dir = Dir, min_interval = ?DEFAULT_MIN_DISK_CHECK_INTERVAL, max_interval = ?DEFAULT_MAX_DISK_CHECK_INTERVAL, alarmed = false, - enabled = true}, - case {catch get_disk_free(Dir), - vm_memory_monitor:get_total_memory()} of - {N1, N2} when is_integer(N1), is_integer(N2) -> - {ok, start_timer(set_disk_limits(State, Limit))}; - Err -> - rabbit_log:info("Disabling disk free space monitoring " - "on unsupported platform:~n~p~n", [Err]), - {ok, State#state{enabled = false}} - end. + enabled = true, + limit = Limit, + retries = Retries, + interval = Interval}, + {ok, enable(State)}. handle_call(get_disk_free_limit, _From, State = #state{limit = Limit}) -> {reply, Limit, State}; @@ -161,6 +163,8 @@ handle_call(_Request, _From, State) -> handle_cast(_Request, State) -> {noreply, State}. +handle_info(try_enable, #state{retries = Retries} = State) -> + {noreply, enable(State#state{retries = Retries - 1})}; handle_info(update, State) -> {noreply, start_timer(internal_update(State))}; @@ -233,7 +237,7 @@ parse_free_win32(CommandResult) -> list_to_integer(lists:reverse(Free)). interpret_limit({mem_relative, Relative}) - when is_float(Relative) -> + when is_number(Relative) -> round(Relative * vm_memory_monitor:get_total_memory()); interpret_limit(Absolute) -> case rabbit_resource_monitor_misc:parse_information_unit(Absolute) of @@ -246,7 +250,7 @@ interpret_limit(Absolute) -> emit_update_info(StateStr, CurrentFree, Limit) -> rabbit_log:info( - "Disk free space ~s. Free bytes:~p Limit:~p~n", + "Free disk space is ~s. Free bytes: ~p. Limit: ~p~n", [StateStr, CurrentFree, Limit]). start_timer(State) -> @@ -261,3 +265,20 @@ interval(#state{limit = Limit, max_interval = MaxInterval}) -> IdealInterval = 2 * (Actual - Limit) / ?FAST_RATE, trunc(erlang:max(MinInterval, erlang:min(MaxInterval, IdealInterval))). + +enable(#state{retries = 0} = State) -> + State; +enable(#state{dir = Dir, interval = Interval, limit = Limit, retries = Retries} + = State) -> + case {catch get_disk_free(Dir), + vm_memory_monitor:get_total_memory()} of + {N1, N2} when is_integer(N1), is_integer(N2) -> + rabbit_log:info("Enabling free disk space monitoring~n", []), + start_timer(set_disk_limits(State, Limit)); + Err -> + rabbit_log:info("Free disk space monitor encountered an error " + "(e.g. failed to parse output from OS tools): ~p, retries left: ~s~n", + [Err, Retries]), + timer:send_after(Interval, self(), try_enable), + State#state{enabled = false} + end. diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl index 94710aed43..fefa0de1c9 100644 --- a/src/rabbit_mirror_queue_master.erl +++ b/src/rabbit_mirror_queue_master.erl @@ -57,13 +57,13 @@ coordinator :: pid(), backing_queue :: atom(), backing_queue_state :: any(), - seen_status :: dict:dict(), + seen_status :: map(), confirmed :: [rabbit_guid:guid()], known_senders :: sets:set() }. -spec promote_backing_queue_state (rabbit_amqqueue:name(), pid(), atom(), any(), pid(), [any()], - dict:dict(), [pid()]) -> + map(), [pid()]) -> master_state(). -spec sender_death_fun() -> death_fun(). @@ -127,7 +127,7 @@ init_with_existing_bq(Q = #amqqueue{name = QName}, BQ, BQS) -> coordinator = CPid, backing_queue = BQ, backing_queue_state = BQS, - seen_status = dict:new(), + seen_status = #{}, confirmed = [], known_senders = sets:new(), wait_timeout = rabbit_misc:get_env(rabbit, slave_wait_timeout, 15000) }; @@ -266,7 +266,7 @@ publish(Msg = #basic_message { id = MsgId }, MsgProps, IsDelivered, ChPid, Flow, seen_status = SS, backing_queue = BQ, backing_queue_state = BQS }) -> - false = dict:is_key(MsgId, SS), %% ASSERTION + false = maps:is_key(MsgId, SS), %% ASSERTION ok = gm:broadcast(GM, {publish, ChPid, Flow, MsgProps, Msg}, rabbit_basic:msg_size(Msg)), BQS1 = BQ:publish(Msg, MsgProps, IsDelivered, ChPid, Flow, BQS), @@ -281,7 +281,7 @@ batch_publish(Publishes, ChPid, Flow, lists:foldl(fun ({Msg = #basic_message { id = MsgId }, MsgProps, _IsDelivered}, {Pubs, false, Sizes}) -> {[{Msg, MsgProps, true} | Pubs], %% [0] - false = dict:is_key(MsgId, SS), %% ASSERTION + false = maps:is_key(MsgId, SS), %% ASSERTION Sizes + rabbit_basic:msg_size(Msg)} end, {[], false, 0}, Publishes), Publishes2 = lists:reverse(Publishes1), @@ -298,7 +298,7 @@ publish_delivered(Msg = #basic_message { id = MsgId }, MsgProps, seen_status = SS, backing_queue = BQ, backing_queue_state = BQS }) -> - false = dict:is_key(MsgId, SS), %% ASSERTION + false = maps:is_key(MsgId, SS), %% ASSERTION ok = gm:broadcast(GM, {publish_delivered, ChPid, Flow, MsgProps, Msg}, rabbit_basic:msg_size(Msg)), {AckTag, BQS1} = BQ:publish_delivered(Msg, MsgProps, ChPid, Flow, BQS), @@ -313,7 +313,7 @@ batch_publish_delivered(Publishes, ChPid, Flow, {false, MsgSizes} = lists:foldl(fun ({Msg = #basic_message { id = MsgId }, _MsgProps}, {false, Sizes}) -> - {false = dict:is_key(MsgId, SS), %% ASSERTION + {false = maps:is_key(MsgId, SS), %% ASSERTION Sizes + rabbit_basic:msg_size(Msg)} end, {false, 0}, Publishes), ok = gm:broadcast(GM, {batch_publish_delivered, ChPid, Flow, Publishes}, @@ -326,7 +326,7 @@ discard(MsgId, ChPid, Flow, State = #state { gm = GM, backing_queue = BQ, backing_queue_state = BQS, seen_status = SS }) -> - false = dict:is_key(MsgId, SS), %% ASSERTION + false = maps:is_key(MsgId, SS), %% ASSERTION ok = gm:broadcast(GM, {discard, ChPid, Flow, MsgId}), ensure_monitoring(ChPid, State #state { backing_queue_state = @@ -353,7 +353,7 @@ drain_confirmed(State = #state { backing_queue = BQ, lists:foldl( fun (MsgId, {MsgIdsN, SSN}) -> %% We will never see 'discarded' here - case dict:find(MsgId, SSN) of + case maps:find(MsgId, SSN) of error -> {[MsgId | MsgIdsN], SSN}; {ok, published} -> @@ -364,7 +364,7 @@ drain_confirmed(State = #state { backing_queue = BQ, %% consequently we need to filter out the %% confirm here. We will issue the confirm %% when we see the publish from the channel. - {MsgIdsN, dict:store(MsgId, confirmed, SSN)}; + {MsgIdsN, maps:put(MsgId, confirmed, SSN)}; {ok, confirmed} -> %% Well, confirms are racy by definition. {[MsgId | MsgIdsN], SSN} @@ -457,7 +457,7 @@ msg_rates(#state { backing_queue = BQ, backing_queue_state = BQS }) -> info(backing_queue_status, State = #state { backing_queue = BQ, backing_queue_state = BQS }) -> BQ:info(backing_queue_status, BQS) ++ - [ {mirror_seen, dict:size(State #state.seen_status)}, + [ {mirror_seen, maps:size(State #state.seen_status)}, {mirror_senders, sets:size(State #state.known_senders)} ]; info(Item, #state { backing_queue = BQ, backing_queue_state = BQS }) -> BQ:info(Item, BQS). @@ -480,7 +480,7 @@ is_duplicate(Message = #basic_message { id = MsgId }, %% it. %% We will never see {published, ChPid, MsgSeqNo} here. - case dict:find(MsgId, SS) of + case maps:find(MsgId, SS) of error -> %% We permit the underlying BQ to have a peek at it, but %% only if we ourselves are not filtering out the msg. @@ -494,7 +494,7 @@ is_duplicate(Message = #basic_message { id = MsgId }, %% immediately after calling is_duplicate). The msg is %% invalid. We will not see this again, nor will we be %% further involved in confirming this message, so erase. - {true, State #state { seen_status = dict:erase(MsgId, SS) }}; + {true, State #state { seen_status = maps:remove(MsgId, SS) }}; {ok, Disposition} when Disposition =:= confirmed %% It got published when we were a slave via gm, and @@ -509,7 +509,7 @@ is_duplicate(Message = #basic_message { id = MsgId }, %% Message was discarded while we were a slave. Confirm now. %% As above, amqqueue_process will have the entry for the %% msg_id_to_channel mapping. - {true, State #state { seen_status = dict:erase(MsgId, SS), + {true, State #state { seen_status = maps:remove(MsgId, SS), confirmed = [MsgId | Confirmed] }} end. diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl index 5c1af7f4aa..ee697be501 100644 --- a/src/rabbit_mirror_queue_slave.erl +++ b/src/rabbit_mirror_queue_slave.erl @@ -129,10 +129,10 @@ handle_go(Q = #amqqueue{name = QName}) -> rate_timer_ref = undefined, sync_timer_ref = undefined, - sender_queues = dict:new(), - msg_id_ack = dict:new(), + sender_queues = #{}, + msg_id_ack = #{}, - msg_id_status = dict:new(), + msg_id_status = #{}, known_senders = pmon:new(delegate), depth_delta = undefined @@ -312,7 +312,7 @@ handle_cast({sync_start, Ref, Syncer}, State1 = #state{rate_timer_ref = TRef} = ensure_rate_timer(State), S = fun({MA, TRefN, BQSN}) -> State1#state{depth_delta = undefined, - msg_id_ack = dict:from_list(MA), + msg_id_ack = maps:from_list(MA), rate_timer_ref = TRefN, backing_queue_state = BQSN} end, @@ -548,7 +548,7 @@ send_or_record_confirm(published, #delivery { sender = ChPid, id = MsgId, is_persistent = true } }, MS, #state { q = #amqqueue { durable = true } }) -> - dict:store(MsgId, {published, ChPid, MsgSeqNo} , MS); + maps:put(MsgId, {published, ChPid, MsgSeqNo} , MS); send_or_record_confirm(_Status, #delivery { sender = ChPid, confirm = true, msg_seq_no = MsgSeqNo }, @@ -561,7 +561,7 @@ confirm_messages(MsgIds, State = #state { msg_id_status = MS }) -> lists:foldl( fun (MsgId, {CMsN, MSN} = Acc) -> %% We will never see 'discarded' here - case dict:find(MsgId, MSN) of + case maps:find(MsgId, MSN) of error -> %% If it needed confirming, it'll have %% already been done. @@ -569,12 +569,12 @@ confirm_messages(MsgIds, State = #state { msg_id_status = MS }) -> {ok, published} -> %% Still not seen it from the channel, just %% record that it's been confirmed. - {CMsN, dict:store(MsgId, confirmed, MSN)}; + {CMsN, maps:put(MsgId, confirmed, MSN)}; {ok, {published, ChPid, MsgSeqNo}} -> %% Seen from both GM and Channel. Can now %% confirm. {rabbit_misc:gb_trees_cons(ChPid, MsgSeqNo, CMsN), - dict:erase(MsgId, MSN)}; + maps:remove(MsgId, MSN)}; {ok, confirmed} -> %% It's already been confirmed. This is %% probably it's been both sync'd to disk @@ -674,21 +674,21 @@ promote_me(From, #state { q = Q = #amqqueue { name = QName }, %% Master, or MTC in queue_process. St = [published, confirmed, discarded], - SS = dict:filter(fun (_MsgId, Status) -> lists:member(Status, St) end, MS), - AckTags = [AckTag || {_MsgId, AckTag} <- dict:to_list(MA)], + SS = maps:filter(fun (_MsgId, Status) -> lists:member(Status, St) end, MS), + AckTags = [AckTag || {_MsgId, AckTag} <- maps:to_list(MA)], MasterState = rabbit_mirror_queue_master:promote_backing_queue_state( QName, CPid, BQ, BQS, GM, AckTags, SS, MPids), - MTC = dict:fold(fun (MsgId, {published, ChPid, MsgSeqNo}, MTC0) -> + MTC = maps:fold(fun (MsgId, {published, ChPid, MsgSeqNo}, MTC0) -> gb_trees:insert(MsgId, {ChPid, MsgSeqNo}, MTC0); (_Msgid, _Status, MTC0) -> MTC0 end, gb_trees:empty(), MS), Deliveries = [promote_delivery(Delivery) || - {_ChPid, {PubQ, _PendCh, _ChState}} <- dict:to_list(SQ), + {_ChPid, {PubQ, _PendCh, _ChState}} <- maps:to_list(SQ), Delivery <- queue:to_list(PubQ)], - AwaitGmDown = [ChPid || {ChPid, {_, _, down_from_ch}} <- dict:to_list(SQ)], + AwaitGmDown = [ChPid || {ChPid, {_, _, down_from_ch}} <- maps:to_list(SQ)], KS1 = lists:foldl(fun (ChPid0, KS0) -> pmon:demonitor(ChPid0, KS0) end, KS, AwaitGmDown), @@ -800,20 +800,20 @@ forget_sender(Down1, Down2) when Down1 =/= Down2 -> true. maybe_forget_sender(ChPid, ChState, State = #state { sender_queues = SQ, msg_id_status = MS, known_senders = KS }) -> - case dict:find(ChPid, SQ) of + case maps:find(ChPid, SQ) of error -> State; {ok, {MQ, PendCh, ChStateRecord}} -> case forget_sender(ChState, ChStateRecord) of true -> credit_flow:peer_down(ChPid), - State #state { sender_queues = dict:erase(ChPid, SQ), + State #state { sender_queues = maps:remove(ChPid, SQ), msg_id_status = lists:foldl( - fun dict:erase/2, + fun maps:remove/2, MS, sets:to_list(PendCh)), known_senders = pmon:demonitor(ChPid, KS) }; false -> - SQ1 = dict:store(ChPid, {MQ, PendCh, ChState}, SQ), + SQ1 = maps:put(ChPid, {MQ, PendCh, ChState}, SQ), State #state { sender_queues = SQ1 } end end. @@ -825,32 +825,32 @@ maybe_enqueue_message( send_mandatory(Delivery), %% must do this before confirms State1 = ensure_monitoring(ChPid, State), %% We will never see {published, ChPid, MsgSeqNo} here. - case dict:find(MsgId, MS) of + case maps:find(MsgId, MS) of error -> {MQ, PendingCh, ChState} = get_sender_queue(ChPid, SQ), MQ1 = queue:in(Delivery, MQ), - SQ1 = dict:store(ChPid, {MQ1, PendingCh, ChState}, SQ), + SQ1 = maps:put(ChPid, {MQ1, PendingCh, ChState}, SQ), State1 #state { sender_queues = SQ1 }; {ok, Status} -> MS1 = send_or_record_confirm( - Status, Delivery, dict:erase(MsgId, MS), State1), + Status, Delivery, maps:remove(MsgId, MS), State1), SQ1 = remove_from_pending_ch(MsgId, ChPid, SQ), State1 #state { msg_id_status = MS1, sender_queues = SQ1 } end. get_sender_queue(ChPid, SQ) -> - case dict:find(ChPid, SQ) of + case maps:find(ChPid, SQ) of error -> {queue:new(), sets:new(), running}; {ok, Val} -> Val end. remove_from_pending_ch(MsgId, ChPid, SQ) -> - case dict:find(ChPid, SQ) of + case maps:find(ChPid, SQ) of error -> SQ; {ok, {MQ, PendingCh, ChState}} -> - dict:store(ChPid, {MQ, sets:del_element(MsgId, PendingCh), ChState}, + maps:put(ChPid, {MQ, sets:del_element(MsgId, PendingCh), ChState}, SQ) end. @@ -867,7 +867,7 @@ publish_or_discard(Status, ChPid, MsgId, case queue:out(MQ) of {empty, _MQ2} -> {MQ, sets:add_element(MsgId, PendingCh), - dict:store(MsgId, Status, MS)}; + maps:put(MsgId, Status, MS)}; {{value, Delivery = #delivery { message = #basic_message { id = MsgId } }}, MQ2} -> {MQ2, PendingCh, @@ -882,7 +882,7 @@ publish_or_discard(Status, ChPid, MsgId, %% expecting any confirms from us. {MQ, PendingCh, MS} end, - SQ1 = dict:store(ChPid, {MQ1, PendingCh1, ChState}, SQ), + SQ1 = maps:put(ChPid, {MQ1, PendingCh1, ChState}, SQ), State1 #state { sender_queues = SQ1, msg_id_status = MS1 }. @@ -1004,9 +1004,9 @@ msg_ids_to_acktags(MsgIds, MA) -> {AckTags, MA1} = lists:foldl( fun (MsgId, {Acc, MAN}) -> - case dict:find(MsgId, MA) of + case maps:find(MsgId, MA) of error -> {Acc, MAN}; - {ok, AckTag} -> {[AckTag | Acc], dict:erase(MsgId, MAN)} + {ok, AckTag} -> {[AckTag | Acc], maps:remove(MsgId, MAN)} end end, {[], MA}, MsgIds), {lists:reverse(AckTags), MA1}. @@ -1014,7 +1014,7 @@ msg_ids_to_acktags(MsgIds, MA) -> maybe_store_ack(false, _MsgId, _AckTag, State) -> State; maybe_store_ack(true, MsgId, AckTag, State = #state { msg_id_ack = MA }) -> - State #state { msg_id_ack = dict:store(MsgId, AckTag, MA) }. + State #state { msg_id_ack = maps:put(MsgId, AckTag, MA) }. set_delta(0, State = #state { depth_delta = undefined }) -> ok = record_synchronised(State#state.q), diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index 6d3b47a405..e48eb2b91f 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -119,6 +119,8 @@ init_from_config() -> (Name, BadNames) when is_atom(Name) -> BadNames; (Name, BadNames) -> [Name | BadNames] end, + %% See rabbitmq/rabbitmq-server#1202 for details. + rabbit_peer_discovery:maybe_inject_randomized_delay(), {DiscoveredNodes, NodeType} = case rabbit_peer_discovery:discover_cluster_nodes() of {ok, {Nodes, Type} = Config} @@ -139,14 +141,17 @@ init_from_config() -> {ok, _} -> e(invalid_cluster_nodes_conf) end, - case DiscoveredNodes of + rabbit_log:info("All discovered existing cluster peers: ~p~n", + [rabbit_peer_discovery:format_discovered_nodes(DiscoveredNodes)]), + Peers = nodes_excl_me(DiscoveredNodes), + case Peers of [] -> rabbit_log:info("Discovered no peer nodes to cluster with"), init_db_and_upgrade([node()], disc, false, _Retry = true); _ -> - rabbit_log:info("Discovered peer nodes: ~s~n", - [rabbit_peer_discovery:format_discovered_nodes(DiscoveredNodes)]), - join_discovered_peers(DiscoveredNodes, NodeType) + rabbit_log:info("Peer nodes we can cluster with: ~s~n", + [rabbit_peer_discovery:format_discovered_nodes(Peers)]), + join_discovered_peers(Peers, NodeType) end. %% Attempts to join discovered, diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl index 891cdf0236..be85091948 100644 --- a/src/rabbit_msg_store.erl +++ b/src/rabbit_msg_store.erl @@ -1337,9 +1337,11 @@ update_pending_confirms(Fun, CRef, record_pending_confirm(CRef, MsgId, State) -> update_pending_confirms( fun (_MsgOnDiskFun, CTM) -> - maps:update_with(CRef, - fun (MsgIds) -> gb_sets:add(MsgId, MsgIds) end, - gb_sets:singleton(MsgId), CTM) + NewMsgIds = case maps:find(CRef, CTM) of + error -> gb_sets:singleton(MsgId); + {ok, MsgIds} -> gb_sets:add(MsgId, MsgIds) + end, + maps:put(CRef, NewMsgIds, CTM) end, CRef, State). client_confirm(CRef, MsgIds, ActionTaken, State) -> diff --git a/src/rabbit_msg_store_gc.erl b/src/rabbit_msg_store_gc.erl index 6179ef95bb..728c9652d0 100644 --- a/src/rabbit_msg_store_gc.erl +++ b/src/rabbit_msg_store_gc.erl @@ -70,7 +70,7 @@ set_maximum_since_use(Pid, Age) -> init([MsgStoreState]) -> ok = file_handle_cache:register_callback(?MODULE, set_maximum_since_use, [self()]), - {ok, #state { pending_no_readers = dict:new(), + {ok, #state { pending_no_readers = #{}, on_action = [], msg_store_state = MsgStoreState }, hibernate, {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. @@ -89,11 +89,11 @@ handle_cast({delete, File}, State) -> handle_cast({no_readers, File}, State = #state { pending_no_readers = Pending }) -> - {noreply, case dict:find(File, Pending) of + {noreply, case maps:find(File, Pending) of error -> State; {ok, {Action, Files}} -> - Pending1 = dict:erase(File, Pending), + Pending1 = maps:remove(File, Pending), attempt_action( Action, Files, State #state { pending_no_readers = Pending1 }) @@ -123,7 +123,7 @@ attempt_action(Action, Files, fun (Thunk) -> not Thunk() end, [do_action(Action, Files, MsgStoreState) | Thunks]) }; - [File | _] -> Pending1 = dict:store(File, {Action, Files}, Pending), + [File | _] -> Pending1 = maps:put(File, {Action, Files}, Pending), State #state { pending_no_readers = Pending1 } end. diff --git a/src/rabbit_peer_discovery.erl b/src/rabbit_peer_discovery.erl index 7cdb35f6f4..088f1c88fd 100644 --- a/src/rabbit_peer_discovery.erl +++ b/src/rabbit_peer_discovery.erl @@ -21,7 +21,9 @@ %% -export([discover_cluster_nodes/0, backend/0, node_type/0, - normalize/1, format_discovered_nodes/1, log_configured_backend/0]). + normalize/1, format_discovered_nodes/1, log_configured_backend/0, + register/0, unregister/0, maybe_register/0, maybe_unregister/0, + maybe_inject_randomized_delay/0]). -export([append_node_prefix/1, node_prefix/0]). -define(DEFAULT_BACKEND, rabbit_peer_discovery_classic_config). @@ -30,6 +32,9 @@ -define(DEFAULT_NODE_TYPE, disc). %% default node prefix to attach to discovered hostnames -define(DEFAULT_PREFIX, "rabbit"). +%% default randomized delay range, in seconds +-define(DEFAULT_STARTUP_RANDOMIZED_DELAY, {5, 60}). + -define(NODENAME_PART_SEPARATOR, "@"). @@ -72,6 +77,117 @@ discover_cluster_nodes() -> normalize(Backend:list_nodes()). +-spec maybe_register() -> ok. + +maybe_register() -> + Backend = backend(), + case Backend:supports_registration() of + true -> + register(), + Backend:post_registration(); + false -> + rabbit_log:info("Peer discovery backend ~s does not support registration, skipping registration.", [Backend]), + ok + end. + + +-spec maybe_unregister() -> ok. + +maybe_unregister() -> + Backend = backend(), + case Backend:supports_registration() of + true -> + unregister(); + false -> + rabbit_log:info("Peer discovery backend ~s does not support registration, skipping unregistration.", [Backend]), + ok + end. + + +-spec maybe_inject_randomized_delay() -> ok. +maybe_inject_randomized_delay() -> + Backend = backend(), + case Backend:supports_registration() of + true -> + rabbit_log:info("Peer discovery backend ~s supports registration.", [Backend]), + inject_randomized_delay(); + false -> + rabbit_log:info("Peer discovery backend ~s does not support registration, skipping randomized startup delay.", [Backend]), + ok + end. + +-spec inject_randomized_delay() -> ok. + +inject_randomized_delay() -> + {Min, Max} = case randomized_delay_range_in_ms() of + {A, B} -> {A, B}; + [A, B] -> {A, B} + end, + case {Min, Max} of + %% When the max value is set to 0, consider the delay to be disabled. + %% In addition, `rand:uniform/1` will fail with a "no function clause" + %% when the argument is 0. + {_, 0} -> + rabbit_log:info("Randomized delay range's upper bound is set to 0. Considering it disabled."), + ok; + {_, N} when is_number(N) -> + rand:seed(exsplus), + RandomVal = rand:uniform(round(N)), + rabbit_log:debug("Randomized startup delay: configured range is from ~p to ~p milliseconds, PRNG pick: ~p...", + [Min, Max, RandomVal]), + Effective = case RandomVal < Min of + true -> Min; + false -> RandomVal + end, + rabbit_log:info("Will wait for ~p milliseconds before proceeding with regitration...", [Effective]), + timer:sleep(Effective), + ok + end. + +-spec randomized_delay_range_in_ms() -> {integer(), integer()}. + +randomized_delay_range_in_ms() -> + {Min, Max} = case application:get_env(rabbit, autocluster) of + {ok, Proplist} -> + proplists:get_value(randomized_startup_delay_range, Proplist, ?DEFAULT_STARTUP_RANDOMIZED_DELAY); + undefined -> + ?DEFAULT_STARTUP_RANDOMIZED_DELAY + end, + {Min * 1000, Max * 1000}. + + +-spec register() -> ok. + +register() -> + Backend = backend(), + rabbit_log:info("Will register with peer discovery backend ~s", [Backend]), + case Backend:register() of + ok -> ok; + {error, Error} -> + rabbit_log:error("Failed to register with peer discovery backend ~s: ~p", + [Backend, Error]), + ok + end. + + +-spec unregister() -> ok. + +unregister() -> + Backend = backend(), + rabbit_log:info("Will unregister with peer discovery backend ~s", [Backend]), + case Backend:unregister() of + ok -> ok; + {error, Error} -> + rabbit_log:error("Failed to unregister with peer discovery backend ~s: ~p", + [Backend, Error]), + ok + end. + + +%% +%% Implementation +%% + -spec normalize(Nodes :: list() | {Nodes :: list(), NodeType :: rabbit_types:node_type()} | {ok, Nodes :: list()} | @@ -90,7 +206,6 @@ normalize({ok, {Nodes, NodeType}}) when is_list(Nodes) andalso is_atom(NodeType) normalize({error, Reason}) -> {error, Reason}. - -spec format_discovered_nodes(Nodes :: list()) -> string(). format_discovered_nodes(Nodes) -> diff --git a/src/rabbit_peer_discovery_classic_config.erl b/src/rabbit_peer_discovery_classic_config.erl index 95e5532548..da5fb6c971 100644 --- a/src/rabbit_peer_discovery_classic_config.erl +++ b/src/rabbit_peer_discovery_classic_config.erl @@ -19,7 +19,8 @@ -include("rabbit.hrl"). --export([list_nodes/0, register/0, unregister/0]). +-export([list_nodes/0, supports_registration/0, register/0, unregister/0, + post_registration/0]). %% %% API @@ -34,6 +35,11 @@ list_nodes() -> undefined -> {[], disc} end. +-spec supports_registration() -> boolean(). + +supports_registration() -> + false. + -spec register() -> ok. register() -> @@ -43,3 +49,8 @@ register() -> unregister() -> ok. + +-spec post_registration() -> ok. + +post_registration() -> + ok. diff --git a/src/rabbit_peer_discovery_dns.erl b/src/rabbit_peer_discovery_dns.erl index c8f6a7f39a..e4215cd048 100644 --- a/src/rabbit_peer_discovery_dns.erl +++ b/src/rabbit_peer_discovery_dns.erl @@ -19,7 +19,8 @@ -include("rabbit.hrl"). --export([list_nodes/0, register/0, unregister/0]). +-export([list_nodes/0, supports_registration/0, register/0, unregister/0, + post_registration/0]). %% for tests -export([discover_nodes/2, discover_hostnames/2]). @@ -48,6 +49,13 @@ list_nodes() -> end end. + +-spec supports_registration() -> boolean(). + +supports_registration() -> + false. + + -spec register() -> ok. register() -> @@ -58,6 +66,11 @@ register() -> unregister() -> ok. +-spec post_registration() -> ok. + +post_registration() -> + ok. + %% %% Implementation @@ -68,14 +81,25 @@ discover_nodes(SeedHostname, LongNamesUsed) -> H <- discover_hostnames(SeedHostname, LongNamesUsed)]. discover_hostnames(SeedHostname, LongNamesUsed) -> - %% TODO: IPv6 support - IPs = inet_res:lookup(SeedHostname, in, a), - rabbit_log:info("Addresses discovered via A records of ~s: ~s", - [SeedHostname, string:join([inet_parse:ntoa(IP) || IP <- IPs], ", ")]), + lookup(SeedHostname, LongNamesUsed, ipv4) ++ + lookup(SeedHostname, LongNamesUsed, ipv6). + +decode_record(ipv4) -> + a; +decode_record(ipv6) -> + aaaa. + +lookup(SeedHostname, LongNamesUsed, IPv) -> + IPs = inet_res:lookup(SeedHostname, in, decode_record(IPv)), + rabbit_log:info("Addresses discovered via ~s records of ~s: ~s", + [string:to_upper(atom_to_list(decode_record(IPv))), + SeedHostname, + string:join([inet_parse:ntoa(IP) || IP <- IPs], ", ")]), Hosts = [extract_host(inet:gethostbyaddr(A), LongNamesUsed, A) || - A <- IPs], + A <- IPs], lists:filter(fun(E) -> E =/= error end, Hosts). + %% long node names are used extract_host({ok, {hostent, FQDN, _, _, _, _}}, true, _Address) -> FQDN; diff --git a/src/rabbit_queue_consumers.erl b/src/rabbit_queue_consumers.erl index 4e9715fba2..f13a46fcf3 100644 --- a/src/rabbit_queue_consumers.erl +++ b/src/rabbit_queue_consumers.erl @@ -64,7 +64,8 @@ -spec max_active_priority(state()) -> integer() | 'infinity' | 'empty'. -spec inactive(state()) -> boolean(). -spec all(state()) -> [{ch(), rabbit_types:ctag(), boolean(), - non_neg_integer(), rabbit_framing:amqp_table()}]. + non_neg_integer(), rabbit_framing:amqp_table(), + rabbit_types:username()}]. -spec count() -> non_neg_integer(). -spec unacknowledged_message_count() -> non_neg_integer(). -spec add(ch(), rabbit_types:ctag(), boolean(), pid(), boolean(), @@ -280,7 +281,7 @@ subtract_acks([T | TL] = AckTags, Prefix, CTagCounts, AckQ) -> orddict:update_counter(CTag, 1, CTagCounts), QTail); {{value, V}, QTail} -> subtract_acks(AckTags, [V | Prefix], CTagCounts, QTail); - {empty, _} -> + {empty, _} -> subtract_acks([], Prefix, CTagCounts, AckQ) end. diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl index 3c4f23a416..a71eaf1ff4 100644 --- a/src/rabbit_queue_index.erl +++ b/src/rabbit_queue_index.erl @@ -83,7 +83,7 @@ %% contains a mapping from segment numbers to state-per-segment (this %% state is held for all segments which have been "seen": thus a %% segment which has been read but has no pending entries in the -%% journal is still held in this mapping. Also note that a dict is +%% journal is still held in this mapping. Also note that a map is %% used for this mapping, not an array because with an array, you will %% always have entries from 0). Actions are stored directly in this %% state. Thus at the point of flushing the journal, firstly no @@ -236,10 +236,10 @@ unacked :: non_neg_integer() }). -type seq_id() :: integer(). --type seg_dict() :: {dict:dict(), [segment()]}. +-type seg_map() :: {map(), [segment()]}. -type on_sync_fun() :: fun ((gb_sets:set()) -> ok). -type qistate() :: #qistate { dir :: file:filename(), - segments :: 'undefined' | seg_dict(), + segments :: 'undefined' | seg_map(), journal_handle :: hdl(), dirty_count :: integer(), max_journal_entries :: non_neg_integer(), @@ -1027,7 +1027,7 @@ segment_find(Seg, {_Segments, [Segment = #segment { num = Seg } |_]}) -> segment_find(Seg, {_Segments, [_, Segment = #segment { num = Seg }]}) -> {ok, Segment}; %% 2, matches tail segment_find(Seg, {Segments, _}) -> %% no match - dict:find(Seg, Segments). + maps:find(Seg, Segments). segment_store(Segment = #segment { num = Seg }, %% 1 or (2, matches head) {Segments, [#segment { num = Seg } | Tail]}) -> @@ -1036,28 +1036,28 @@ segment_store(Segment = #segment { num = Seg }, %% 2, matches tail {Segments, [SegmentA, #segment { num = Seg }]}) -> {Segments, [Segment, SegmentA]}; segment_store(Segment = #segment { num = Seg }, {Segments, []}) -> - {dict:erase(Seg, Segments), [Segment]}; + {maps:remove(Seg, Segments), [Segment]}; segment_store(Segment = #segment { num = Seg }, {Segments, [SegmentA]}) -> - {dict:erase(Seg, Segments), [Segment, SegmentA]}; + {maps:remove(Seg, Segments), [Segment, SegmentA]}; segment_store(Segment = #segment { num = Seg }, {Segments, [SegmentA, SegmentB]}) -> - {dict:store(SegmentB#segment.num, SegmentB, dict:erase(Seg, Segments)), + {maps:put(SegmentB#segment.num, SegmentB, maps:remove(Seg, Segments)), [Segment, SegmentA]}. segment_fold(Fun, Acc, {Segments, CachedSegments}) -> - dict:fold(fun (_Seg, Segment, Acc1) -> Fun(Segment, Acc1) end, + maps:fold(fun (_Seg, Segment, Acc1) -> Fun(Segment, Acc1) end, lists:foldl(Fun, Acc, CachedSegments), Segments). segment_map(Fun, {Segments, CachedSegments}) -> - {dict:map(fun (_Seg, Segment) -> Fun(Segment) end, Segments), + {maps:map(fun (_Seg, Segment) -> Fun(Segment) end, Segments), lists:map(Fun, CachedSegments)}. segment_nums({Segments, CachedSegments}) -> lists:map(fun (#segment { num = Num }) -> Num end, CachedSegments) ++ - dict:fetch_keys(Segments). + maps:keys(Segments). segments_new() -> - {dict:new(), []}. + {#{}, []}. entry_to_segment(_RelSeq, {?PUB, del, ack}, Initial) -> Initial; |