diff options
| -rw-r--r-- | include/rabbit_log.hrl | 1 | ||||
| -rw-r--r-- | src/mnesia_sync.erl | 73 | ||||
| -rw-r--r-- | src/rabbit_amqqueue.erl | 1045 | ||||
| -rw-r--r-- | src/rabbit_auth_backend_internal.erl | 604 | ||||
| -rw-r--r-- | src/rabbit_basic.erl | 302 | ||||
| -rw-r--r-- | src/rabbit_channel.erl | 2150 | ||||
| -rw-r--r-- | src/rabbit_channel_interceptor.erl | 113 | ||||
| -rw-r--r-- | src/rabbit_exchange_decorator.erl | 114 | ||||
| -rw-r--r-- | src/rabbit_health_check.erl | 95 | ||||
| -rw-r--r-- | src/rabbit_log.erl | 128 | ||||
| -rw-r--r-- | src/rabbit_networking.erl | 481 | ||||
| -rw-r--r-- | src/rabbit_nodes.erl | 232 | ||||
| -rw-r--r-- | src/rabbit_queue_collector.erl | 89 | ||||
| -rw-r--r-- | src/rabbit_queue_decorator.erl | 73 | ||||
| -rw-r--r-- | src/rabbit_reader.erl | 1652 | ||||
| -rw-r--r-- | src/rabbit_resource_monitor_misc.erl | 47 | ||||
| -rw-r--r-- | src/rabbit_vhost.erl | 7 | ||||
| -rw-r--r-- | src/vm_memory_monitor.erl | 540 | ||||
| -rw-r--r-- | src/worker_pool.erl | 172 | ||||
| -rw-r--r-- | src/worker_pool_sup.erl | 56 | ||||
| -rw-r--r-- | src/worker_pool_worker.erl | 193 |
21 files changed, 6956 insertions, 1211 deletions
diff --git a/include/rabbit_log.hrl b/include/rabbit_log.hrl deleted file mode 100644 index fcb72e2e31..0000000000 --- a/include/rabbit_log.hrl +++ /dev/null @@ -1 +0,0 @@ --define(LAGER_SINK, rabbit_log_lager_event). diff --git a/src/mnesia_sync.erl b/src/mnesia_sync.erl deleted file mode 100644 index a263a4f576..0000000000 --- a/src/mnesia_sync.erl +++ /dev/null @@ -1,73 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - --module(mnesia_sync). - -%% mnesia:sync_transaction/3 fails to guarantee that the log is flushed to disk -%% at commit. This module is an attempt to minimise the risk of data loss by -%% performing a coalesced log fsync. Unfortunately this is performed regardless -%% of whether or not the log was appended to. - --behaviour(gen_server). - --export([sync/0]). - --export([start_link/0, init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - --define(SERVER, ?MODULE). - --record(state, {waiting, disc_node}). - -%%---------------------------------------------------------------------------- - --spec sync() -> 'ok'. - -%%---------------------------------------------------------------------------- - -start_link() -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). - -sync() -> - gen_server:call(?SERVER, sync, infinity). - -%%---------------------------------------------------------------------------- - -init([]) -> - {ok, #state{disc_node = mnesia:system_info(use_dir), waiting = []}}. - -handle_call(sync, _From, #state{disc_node = false} = State) -> - {reply, ok, State}; -handle_call(sync, From, #state{waiting = Waiting} = State) -> - {noreply, State#state{waiting = [From | Waiting]}, 0}; -handle_call(Request, _From, State) -> - {stop, {unhandled_call, Request}, State}. - -handle_cast(Request, State) -> - {stop, {unhandled_cast, Request}, State}. - -handle_info(timeout, #state{waiting = Waiting} = State) -> - ok = disk_log:sync(latest_log), - _ = [gen_server:reply(From, ok) || From <- Waiting], - {noreply, State#state{waiting = []}}; -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_amqqueue.erl b/src/rabbit_amqqueue.erl new file mode 100644 index 0000000000..3fb76be5e8 --- /dev/null +++ b/src/rabbit_amqqueue.erl @@ -0,0 +1,1045 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_amqqueue). + +-export([warn_file_limit/0]). +-export([recover/1, stop/1, start/1, declare/6, declare/7, + delete_immediately/1, delete_exclusive/2, delete/4, purge/1, + forget_all_durable/1, delete_crashed/1, delete_crashed/2, + delete_crashed_internal/2]). +-export([pseudo_queue/2, immutable/1]). +-export([lookup/1, not_found_or_absent/1, with/2, with/3, with_or_die/2, + assert_equivalence/5, + check_exclusive_access/2, with_exclusive_access_or_die/3, + stat/1, deliver/2, requeue/3, ack/3, reject/4]). +-export([list/0, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2, + emit_info_all/5, list_local/1, info_local/1, + emit_info_local/4, emit_info_down/4]). +-export([list_down/1, count/1, list_names/0]). +-export([force_event_refresh/1, notify_policy_changed/1]). +-export([consumers/1, consumers_all/1, emit_consumers_all/4, consumer_info_keys/0]). +-export([basic_get/4, basic_consume/11, basic_cancel/5, notify_decorators/1]). +-export([notify_sent/2, notify_sent_queue_down/1, resume/2]). +-export([notify_down_all/2, notify_down_all/3, activate_limit_all/2, credit/5]). +-export([on_node_up/1, on_node_down/1]). +-export([update/2, store_queue/1, update_decorators/1, policy_changed/2]). +-export([update_mirroring/1, sync_mirrors/1, cancel_sync_mirrors/1, is_mirrored/1]). + +-export([pid_of/1, pid_of/2]). + +%% internal +-export([internal_declare/2, internal_delete/2, run_backing_queue/3, + set_ram_duration_target/2, set_maximum_since_use/2, + emit_consumers_local/3]). + +-include("rabbit.hrl"). +-include_lib("stdlib/include/qlc.hrl"). + +-define(INTEGER_ARG_TYPES, [byte, short, signedint, long, + unsignedbyte, unsignedshort, unsignedint]). + +-define(MORE_CONSUMER_CREDIT_AFTER, 50). + +%%---------------------------------------------------------------------------- + +-export_type([name/0, qmsg/0, absent_reason/0]). + +-type name() :: rabbit_types:r('queue'). +-type qpids() :: [pid()]. +-type qlen() :: rabbit_types:ok(non_neg_integer()). +-type qfun(A) :: fun ((rabbit_types:amqqueue()) -> A | no_return()). +-type qmsg() :: {name(), pid(), msg_id(), boolean(), rabbit_types:message()}. +-type msg_id() :: non_neg_integer(). +-type ok_or_errors() :: + 'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}. +-type absent_reason() :: 'nodedown' | 'crashed'. +-type queue_or_absent() :: rabbit_types:amqqueue() | + {'absent', rabbit_types:amqqueue(),absent_reason()}. +-type not_found_or_absent() :: + 'not_found' | {'absent', rabbit_types:amqqueue(), absent_reason()}. +-spec recover(rabbit_types:vhost()) -> [rabbit_types:amqqueue()]. +-spec stop(rabbit_types:vhost()) -> 'ok'. +-spec start([rabbit_types:amqqueue()]) -> 'ok'. +-spec declare + (name(), boolean(), boolean(), rabbit_framing:amqp_table(), + rabbit_types:maybe(pid()), rabbit_types:username()) -> + {'new' | 'existing' | 'absent' | 'owner_died', + rabbit_types:amqqueue()} | + rabbit_types:channel_exit(). +-spec declare + (name(), boolean(), boolean(), rabbit_framing:amqp_table(), + rabbit_types:maybe(pid()), rabbit_types:username(), node()) -> + {'new' | 'existing' | 'owner_died', rabbit_types:amqqueue()} | + {'absent', rabbit_types:amqqueue(), absent_reason()} | + rabbit_types:channel_exit(). +-spec internal_declare(rabbit_types:amqqueue(), boolean()) -> + queue_or_absent() | rabbit_misc:thunk(queue_or_absent()). +-spec update + (name(), fun((rabbit_types:amqqueue()) -> rabbit_types:amqqueue())) -> + 'not_found' | rabbit_types:amqqueue(). +-spec lookup + (name()) -> + rabbit_types:ok(rabbit_types:amqqueue()) | + rabbit_types:error('not_found'); + ([name()]) -> + [rabbit_types:amqqueue()]. +-spec not_found_or_absent(name()) -> not_found_or_absent(). +-spec with(name(), qfun(A)) -> + A | rabbit_types:error(not_found_or_absent()). +-spec with(name(), qfun(A), fun((not_found_or_absent()) -> B)) -> A | B. +-spec with_or_die(name(), qfun(A)) -> A | rabbit_types:channel_exit(). +-spec assert_equivalence + (rabbit_types:amqqueue(), boolean(), boolean(), + rabbit_framing:amqp_table(), rabbit_types:maybe(pid())) -> + 'ok' | rabbit_types:channel_exit() | rabbit_types:connection_exit(). +-spec check_exclusive_access(rabbit_types:amqqueue(), pid()) -> + 'ok' | rabbit_types:channel_exit(). +-spec with_exclusive_access_or_die(name(), pid(), qfun(A)) -> + A | rabbit_types:channel_exit(). +-spec list() -> [rabbit_types:amqqueue()]. +-spec list(rabbit_types:vhost()) -> [rabbit_types:amqqueue()]. +-spec list_names() -> [rabbit_amqqueue:name()]. +-spec list_down(rabbit_types:vhost()) -> [rabbit_types:amqqueue()]. +-spec info_keys() -> rabbit_types:info_keys(). +-spec info(rabbit_types:amqqueue()) -> rabbit_types:infos(). +-spec info(rabbit_types:amqqueue(), rabbit_types:info_keys()) -> + rabbit_types:infos(). +-spec info_all(rabbit_types:vhost()) -> [rabbit_types:infos()]. +-spec info_all(rabbit_types:vhost(), rabbit_types:info_keys()) -> + [rabbit_types:infos()]. +-spec force_event_refresh(reference()) -> 'ok'. +-spec notify_policy_changed(rabbit_types:amqqueue()) -> 'ok'. +-spec consumers(rabbit_types:amqqueue()) -> + [{pid(), rabbit_types:ctag(), boolean(), non_neg_integer(), + rabbit_framing:amqp_table()}]. +-spec consumer_info_keys() -> rabbit_types:info_keys(). +-spec consumers_all(rabbit_types:vhost()) -> + [{name(), pid(), rabbit_types:ctag(), boolean(), + non_neg_integer(), rabbit_framing:amqp_table()}]. +-spec stat(rabbit_types:amqqueue()) -> + {'ok', non_neg_integer(), non_neg_integer()}. +-spec delete_immediately(qpids()) -> 'ok'. +-spec delete_exclusive(qpids(), pid()) -> 'ok'. +-spec delete + (rabbit_types:amqqueue(), 'false', 'false', rabbit_types:username()) -> + qlen(); + (rabbit_types:amqqueue(), 'true' , 'false', rabbit_types:username()) -> + qlen() | rabbit_types:error('in_use'); + (rabbit_types:amqqueue(), 'false', 'true', rabbit_types:username()) -> + qlen() | rabbit_types:error('not_empty'); + (rabbit_types:amqqueue(), 'true' , 'true', rabbit_types:username()) -> + qlen() | + rabbit_types:error('in_use') | + rabbit_types:error('not_empty'). +-spec delete_crashed(rabbit_types:amqqueue()) -> 'ok'. +-spec delete_crashed_internal(rabbit_types:amqqueue(), rabbit_types:username()) -> 'ok'. +-spec purge(rabbit_types:amqqueue()) -> qlen(). +-spec forget_all_durable(node()) -> 'ok'. +-spec deliver([rabbit_types:amqqueue()], rabbit_types:delivery()) -> + qpids(). +-spec requeue(pid(), [msg_id()], pid()) -> 'ok'. +-spec ack(pid(), [msg_id()], pid()) -> 'ok'. +-spec reject(pid(), [msg_id()], boolean(), pid()) -> 'ok'. +-spec notify_down_all(qpids(), pid()) -> ok_or_errors(). +-spec notify_down_all(qpids(), pid(), non_neg_integer()) -> + ok_or_errors(). +-spec activate_limit_all(qpids(), pid()) -> ok_or_errors(). +-spec basic_get(rabbit_types:amqqueue(), pid(), boolean(), pid()) -> + {'ok', non_neg_integer(), qmsg()} | 'empty'. +-spec credit + (rabbit_types:amqqueue(), pid(), rabbit_types:ctag(), non_neg_integer(), + boolean()) -> + 'ok'. +-spec basic_consume + (rabbit_types:amqqueue(), boolean(), pid(), pid(), boolean(), + non_neg_integer(), rabbit_types:ctag(), boolean(), + rabbit_framing:amqp_table(), any(), rabbit_types:username()) -> + rabbit_types:ok_or_error('exclusive_consume_unavailable'). +-spec basic_cancel + (rabbit_types:amqqueue(), pid(), rabbit_types:ctag(), any(), + rabbit_types:username()) -> 'ok'. +-spec notify_decorators(rabbit_types:amqqueue()) -> 'ok'. +-spec resume(pid(), pid()) -> 'ok'. +-spec internal_delete(name(), rabbit_types:username()) -> + rabbit_types:ok_or_error('not_found') | + rabbit_types:connection_exit() | + fun (() -> + rabbit_types:ok_or_error('not_found') | + rabbit_types:connection_exit()). +-spec run_backing_queue + (pid(), atom(), (fun ((atom(), A) -> {[rabbit_types:msg_id()], A}))) -> + 'ok'. +-spec set_ram_duration_target(pid(), number() | 'infinity') -> 'ok'. +-spec set_maximum_since_use(pid(), non_neg_integer()) -> 'ok'. +-spec on_node_up(node()) -> 'ok'. +-spec on_node_down(node()) -> 'ok'. +-spec pseudo_queue(name(), pid()) -> rabbit_types:amqqueue(). +-spec immutable(rabbit_types:amqqueue()) -> rabbit_types:amqqueue(). +-spec store_queue(rabbit_types:amqqueue()) -> 'ok'. +-spec update_decorators(name()) -> 'ok'. +-spec policy_changed(rabbit_types:amqqueue(), rabbit_types:amqqueue()) -> + 'ok'. +-spec update_mirroring(pid()) -> 'ok'. +-spec sync_mirrors(rabbit_types:amqqueue() | pid()) -> + 'ok' | rabbit_types:error('not_mirrored'). +-spec cancel_sync_mirrors(rabbit_types:amqqueue() | pid()) -> + 'ok' | {'ok', 'not_syncing'}. +-spec is_mirrored(rabbit_types:amqqueue()) -> boolean(). + +-spec pid_of(rabbit_types:amqqueue()) -> + {'ok', pid()} | rabbit_types:error('not_found'). +-spec pid_of(rabbit_types:vhost(), rabbit_misc:resource_name()) -> + {'ok', pid()} | rabbit_types:error('not_found'). + +%%---------------------------------------------------------------------------- + +-define(CONSUMER_INFO_KEYS, + [queue_name, channel_pid, consumer_tag, ack_required, prefetch_count, + arguments]). + +warn_file_limit() -> + DurableQueues = find_durable_queues(), + L = length(DurableQueues), + + %% if there are not enough file handles, the server might hang + %% when trying to recover queues, warn the user: + case file_handle_cache:get_limit() < L of + true -> + rabbit_log:warning( + "Recovering ~p queues, available file handles: ~p. Please increase max open file handles limit to at least ~p!~n", + [L, file_handle_cache:get_limit(), L]); + false -> + ok + end. + +recover(VHost) -> + Queues = find_durable_queues(VHost), + {ok, BQ} = application:get_env(rabbit, backing_queue_module), + %% We rely on BQ:start/1 returning the recovery terms in the same + %% order as the supplied queue names, so that we can zip them together + %% for further processing in recover_durable_queues. + {ok, OrderedRecoveryTerms} = + BQ:start(VHost, [QName || #amqqueue{name = QName} <- Queues]), + {ok, _} = rabbit_amqqueue_sup_sup:start_for_vhost(VHost), + recover_durable_queues(lists:zip(Queues, OrderedRecoveryTerms)). + +stop(VHost) -> + ok = rabbit_amqqueue_sup_sup:stop_for_vhost(VHost), + {ok, BQ} = application:get_env(rabbit, backing_queue_module), + ok = BQ:stop(VHost). + +start(Qs) -> + %% At this point all recovered queues and their bindings are + %% visible to routing, so now it is safe for them to complete + %% their initialisation (which may involve interacting with other + %% queues). + [Pid ! {self(), go} || #amqqueue{pid = Pid} <- Qs], + ok. + +find_durable_queues(VHost) -> + Node = node(), + mnesia:async_dirty( + fun () -> + qlc:e(qlc:q([Q || Q = #amqqueue{name = Name, + vhost = VH, + pid = Pid} + <- mnesia:table(rabbit_durable_queue), + VH =:= VHost, + node(Pid) == Node andalso + %% Terminations on node down will not remove the rabbit_queue + %% record if it is a mirrored queue (such info is now obtained from + %% the policy). Thus, we must check if the local pid is alive + %% - if the record is present - in order to restart. + (mnesia:read(rabbit_queue, Name, read) =:= [] + orelse not erlang:is_process_alive(Pid))])) + end). + +find_durable_queues() -> + Node = node(), + mnesia:async_dirty( + fun () -> + qlc:e(qlc:q([Q || Q = #amqqueue{name = Name, + pid = Pid} + <- mnesia:table(rabbit_durable_queue), + node(Pid) == Node andalso + %% Terminations on node down will not remove the rabbit_queue + %% record if it is a mirrored queue (such info is now obtained from + %% the policy). Thus, we must check if the local pid is alive + %% - if the record is present - in order to restart. + (mnesia:read(rabbit_queue, Name, read) =:= [] + orelse not erlang:is_process_alive(Pid))])) + end). + +recover_durable_queues(QueuesAndRecoveryTerms) -> + {Results, Failures} = + gen_server2:mcall( + [{rabbit_amqqueue_sup_sup:start_queue_process(node(), Q, recovery), + {init, {self(), Terms}}} || {Q, Terms} <- QueuesAndRecoveryTerms]), + [rabbit_log:error("Queue ~p failed to initialise: ~p~n", + [Pid, Error]) || {Pid, Error} <- Failures], + [Q || {_, {new, Q}} <- Results]. + +declare(QueueName, Durable, AutoDelete, Args, Owner, ActingUser) -> + declare(QueueName, Durable, AutoDelete, Args, Owner, ActingUser, node()). + + +%% The Node argument suggests where the queue (master if mirrored) +%% should be. Note that in some cases (e.g. with "nodes" policy in +%% effect) this might not be possible to satisfy. +declare(QueueName = #resource{virtual_host = VHost}, Durable, AutoDelete, Args, + Owner, ActingUser, 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 = [], + recoverable_slaves = [], + gm_pids = [], + state = live, + policy_version = 0, + slave_pids_pending_shutdown = [], + vhost = VHost, + options = #{user => ActingUser}})), + + Node1 = case rabbit_queue_master_location_misc:get_location(Q) of + {ok, Node0} -> Node0; + {error, _} -> Node + end, + + Node1 = rabbit_mirror_queue_misc:initial_queue_node(Q, Node1), + gen_server2:call( + rabbit_amqqueue_sup_sup:start_queue_process(Node1, Q, declare), + {init, new}, infinity). + +internal_declare(Q, true) -> + rabbit_misc:execute_mnesia_tx_with_tail( + fun () -> + ok = store_queue(Q#amqqueue{state = live}), + rabbit_misc:const(Q) + end); +internal_declare(Q = #amqqueue{name = QueueName}, false) -> + rabbit_misc:execute_mnesia_tx_with_tail( + fun () -> + case mnesia:wread({rabbit_queue, QueueName}) of + [] -> + case not_found_or_absent(QueueName) of + not_found -> Q1 = rabbit_policy:set(Q), + Q2 = Q1#amqqueue{state = live}, + ok = store_queue(Q2), + B = add_default_binding(Q1), + fun () -> B(), Q1 end; + {absent, _Q, _} = R -> rabbit_misc:const(R) + end; + [ExistingQ] -> + rabbit_misc:const(ExistingQ) + end + end). + +update(Name, Fun) -> + case mnesia:wread({rabbit_queue, Name}) of + [Q = #amqqueue{durable = Durable}] -> + Q1 = Fun(Q), + ok = mnesia:write(rabbit_queue, Q1, write), + case Durable of + true -> ok = mnesia:write(rabbit_durable_queue, Q1, write); + _ -> ok + end, + Q1; + [] -> + not_found + end. + +store_queue(Q = #amqqueue{durable = true}) -> + ok = mnesia:write(rabbit_durable_queue, + Q#amqqueue{slave_pids = [], + sync_slave_pids = [], + gm_pids = [], + decorators = undefined}, write), + store_queue_ram(Q); +store_queue(Q = #amqqueue{durable = false}) -> + store_queue_ram(Q). + +store_queue_ram(Q) -> + ok = mnesia:write(rabbit_queue, rabbit_queue_decorator:set(Q), write). + +update_decorators(Name) -> + rabbit_misc:execute_mnesia_transaction( + fun() -> + case mnesia:wread({rabbit_queue, Name}) of + [Q] -> store_queue_ram(Q), + ok; + [] -> ok + end + end). + +policy_changed(Q1 = #amqqueue{decorators = Decorators1}, + Q2 = #amqqueue{decorators = Decorators2}) -> + rabbit_mirror_queue_misc:update_mirrors(Q1, Q2), + D1 = rabbit_queue_decorator:select(Decorators1), + D2 = rabbit_queue_decorator:select(Decorators2), + [ok = M:policy_changed(Q1, Q2) || M <- lists:usort(D1 ++ D2)], + %% Make sure we emit a stats event even if nothing + %% mirroring-related has changed - the policy may have changed anyway. + notify_policy_changed(Q1). + +add_default_binding(#amqqueue{name = QueueName}) -> + ExchangeName = rabbit_misc:r(QueueName, exchange, <<>>), + RoutingKey = QueueName#resource.name, + rabbit_binding:add(#binding{source = ExchangeName, + destination = QueueName, + key = RoutingKey, + args = []}, + ?INTERNAL_USER). + +lookup([]) -> []; %% optimisation +lookup([Name]) -> ets:lookup(rabbit_queue, Name); %% optimisation +lookup(Names) when is_list(Names) -> + %% Normally we'd call mnesia:dirty_read/1 here, but that is quite + %% expensive for reasons explained in rabbit_misc:dirty_read/1. + lists:append([ets:lookup(rabbit_queue, Name) || Name <- Names]); +lookup(Name) -> + rabbit_misc:dirty_read({rabbit_queue, Name}). + +not_found_or_absent(Name) -> + %% NB: we assume that the caller has already performed a lookup on + %% rabbit_queue and not found anything + case mnesia:read({rabbit_durable_queue, Name}) of + [] -> not_found; + [Q] -> {absent, Q, nodedown} %% Q exists on stopped node + end. + +not_found_or_absent_dirty(Name) -> + %% We should read from both tables inside a tx, to get a + %% consistent view. But the chances of an inconsistency are small, + %% and only affect the error kind. + case rabbit_misc:dirty_read({rabbit_durable_queue, Name}) of + {error, not_found} -> not_found; + {ok, Q} -> {absent, Q, nodedown} + end. + +with(Name, F, E) -> + with(Name, F, E, 2000). + +with(Name, F, E, RetriesLeft) -> + case lookup(Name) of + {ok, Q = #amqqueue{}} when RetriesLeft =:= 0 -> + %% Something bad happened to that queue, we are bailing out + %% on processing current request. + E({absent, Q, timeout}); + {ok, Q = #amqqueue{state = crashed}} -> + E({absent, Q, crashed}); + {ok, Q = #amqqueue{pid = QPid}} -> + %% We check is_process_alive(QPid) in case we receive a + %% nodedown (for example) in F() that has nothing to do + %% with the QPid. F() should be written s.t. that this + %% cannot happen, so we bail if it does since that + %% indicates a code bug and we don't want to get stuck in + %% the retry loop. + rabbit_misc:with_exit_handler( + fun () -> false = rabbit_mnesia:is_process_alive(QPid), + timer:sleep(30), + with(Name, F, E, RetriesLeft - 1) + end, fun () -> F(Q) end); + {error, not_found} -> + E(not_found_or_absent_dirty(Name)) + end. + +with(Name, F) -> with(Name, F, fun (E) -> {error, E} end). + +with_or_die(Name, F) -> + with(Name, F, fun (not_found) -> rabbit_misc:not_found(Name); + ({absent, Q, Reason}) -> rabbit_misc:absent(Q, Reason) + end). + +assert_equivalence(#amqqueue{name = QName, + durable = Durable, + auto_delete = AD} = Q, + Durable1, AD1, Args1, Owner) -> + rabbit_misc:assert_field_equivalence(Durable, Durable1, QName, durable), + rabbit_misc:assert_field_equivalence(AD, AD1, QName, auto_delete), + assert_args_equivalence(Q, Args1), + check_exclusive_access(Q, Owner, strict). + +check_exclusive_access(Q, Owner) -> check_exclusive_access(Q, Owner, lax). + +check_exclusive_access(#amqqueue{exclusive_owner = Owner}, Owner, _MatchType) -> + ok; +check_exclusive_access(#amqqueue{exclusive_owner = none}, _ReaderPid, lax) -> + ok; +check_exclusive_access(#amqqueue{name = QueueName}, _ReaderPid, _MatchType) -> + rabbit_misc:protocol_error( + resource_locked, + "cannot obtain exclusive access to locked ~s", + [rabbit_misc:rs(QueueName)]). + +with_exclusive_access_or_die(Name, ReaderPid, F) -> + with_or_die(Name, + fun (Q) -> check_exclusive_access(Q, ReaderPid), F(Q) end). + +assert_args_equivalence(#amqqueue{name = QueueName, arguments = Args}, + RequiredArgs) -> + rabbit_misc:assert_args_equivalence(Args, RequiredArgs, QueueName, + [Key || {Key, _Fun} <- declare_args()]). + +check_declare_arguments(QueueName, Args) -> + check_arguments(QueueName, Args, declare_args()). + +check_consume_arguments(QueueName, Args) -> + check_arguments(QueueName, Args, consume_args()). + +check_arguments(QueueName, Args, Validators) -> + [case rabbit_misc:table_lookup(Args, Key) of + undefined -> ok; + TypeVal -> case Fun(TypeVal, Args) of + ok -> ok; + {error, Error} -> rabbit_misc:protocol_error( + precondition_failed, + "invalid arg '~s' for ~s: ~255p", + [Key, rabbit_misc:rs(QueueName), + Error]) + end + end || {Key, Fun} <- Validators], + ok. + +declare_args() -> + [{<<"x-expires">>, fun check_expires_arg/2}, + {<<"x-message-ttl">>, fun check_message_ttl_arg/2}, + {<<"x-dead-letter-exchange">>, fun check_dlxname_arg/2}, + {<<"x-dead-letter-routing-key">>, fun check_dlxrk_arg/2}, + {<<"x-max-length">>, fun check_non_neg_int_arg/2}, + {<<"x-max-length-bytes">>, fun check_non_neg_int_arg/2}, + {<<"x-max-priority">>, fun check_non_neg_int_arg/2}, + {<<"x-queue-mode">>, fun check_queue_mode/2}]. + +consume_args() -> [{<<"x-priority">>, fun check_int_arg/2}, + {<<"x-cancel-on-ha-failover">>, fun check_bool_arg/2}]. + +check_int_arg({Type, _}, _) -> + case lists:member(Type, ?INTEGER_ARG_TYPES) of + true -> ok; + false -> {error, {unacceptable_type, Type}} + end. + +check_bool_arg({bool, _}, _) -> ok; +check_bool_arg({Type, _}, _) -> {error, {unacceptable_type, Type}}. + +check_non_neg_int_arg({Type, Val}, Args) -> + case check_int_arg({Type, Val}, Args) of + ok when Val >= 0 -> ok; + ok -> {error, {value_negative, Val}}; + Error -> Error + end. + +check_expires_arg({Type, Val}, Args) -> + case check_int_arg({Type, Val}, Args) of + ok when Val == 0 -> {error, {value_zero, Val}}; + ok -> rabbit_misc:check_expiry(Val); + Error -> Error + end. + +check_message_ttl_arg({Type, Val}, Args) -> + case check_int_arg({Type, Val}, Args) of + ok -> rabbit_misc:check_expiry(Val); + Error -> Error + end. + +%% Note that the validity of x-dead-letter-exchange is already verified +%% by rabbit_channel's queue.declare handler. +check_dlxname_arg({longstr, _}, _) -> ok; +check_dlxname_arg({Type, _}, _) -> {error, {unacceptable_type, Type}}. + +check_dlxrk_arg({longstr, _}, Args) -> + case rabbit_misc:table_lookup(Args, <<"x-dead-letter-exchange">>) of + undefined -> {error, routing_key_but_no_dlx_defined}; + _ -> ok + end; +check_dlxrk_arg({Type, _}, _Args) -> + {error, {unacceptable_type, Type}}. + +check_queue_mode({longstr, Val}, _Args) -> + case lists:member(Val, [<<"default">>, <<"lazy">>]) of + true -> ok; + false -> {error, invalid_queue_mode} + end; +check_queue_mode({Type, _}, _Args) -> + {error, {unacceptable_type, Type}}. + +list() -> mnesia:dirty_match_object(rabbit_queue, #amqqueue{_ = '_'}). + +list_names() -> mnesia:dirty_all_keys(rabbit_queue). + +list(VHostPath) -> list(VHostPath, rabbit_queue). + +%% Not dirty_match_object since that would not be transactional when used in a +%% tx context +list(VHostPath, TableName) -> + mnesia:async_dirty( + fun () -> + mnesia:match_object( + TableName, + #amqqueue{name = rabbit_misc:r(VHostPath, queue), _ = '_'}, + read) + end). + +list_down(VHostPath) -> + Present = list(VHostPath), + Durable = list(VHostPath, rabbit_durable_queue), + PresentS = sets:from_list([N || #amqqueue{name = N} <- Present]), + sets:to_list(sets:filter(fun (#amqqueue{name = N}) -> + not sets:is_element(N, PresentS) + end, sets:from_list(Durable))). + +count(VHost) -> + try + %% this is certainly suboptimal but there is no way to count + %% things using a secondary index in Mnesia. Our counter-table-per-node + %% won't work here because with master migration of mirrored queues + %% the "ownership" of queues by nodes becomes a non-trivial problem + %% that requires a proper consensus algorithm. + length(mnesia:dirty_index_read(rabbit_queue, VHost, #amqqueue.vhost)) + catch _:Err -> + rabbit_log:error("Failed to fetch number of queues in vhost ~p:~n~p~n", + [VHost, Err]), + 0 + end. + +info_keys() -> rabbit_amqqueue_process:info_keys(). + +map(Qs, F) -> rabbit_misc:filter_exit_map(F, Qs). + +info(Q = #amqqueue{ state = crashed }) -> info_down(Q, crashed); +info(#amqqueue{ pid = QPid }) -> delegate:call(QPid, info). + +info(Q = #amqqueue{ state = crashed }, Items) -> + info_down(Q, Items, crashed); +info(#amqqueue{ pid = QPid }, Items) -> + case delegate:call(QPid, {info, Items}) of + {ok, Res} -> Res; + {error, Error} -> throw(Error) + end. + +info_down(Q, DownReason) -> + info_down(Q, rabbit_amqqueue_process:info_keys(), DownReason). + +info_down(Q, Items, DownReason) -> + [{Item, i_down(Item, Q, DownReason)} || Item <- Items]. + +i_down(name, #amqqueue{name = Name}, _) -> Name; +i_down(durable, #amqqueue{durable = Dur}, _) -> Dur; +i_down(auto_delete, #amqqueue{auto_delete = AD}, _) -> AD; +i_down(arguments, #amqqueue{arguments = Args}, _) -> Args; +i_down(pid, #amqqueue{pid = QPid}, _) -> QPid; +i_down(recoverable_slaves, #amqqueue{recoverable_slaves = RS}, _) -> RS; +i_down(state, _Q, DownReason) -> DownReason; +i_down(K, _Q, _DownReason) -> + case lists:member(K, rabbit_amqqueue_process:info_keys()) of + true -> ''; + false -> throw({bad_argument, K}) + end. + +info_all(VHostPath) -> + map(list(VHostPath), fun (Q) -> info(Q) end) ++ + map(list_down(VHostPath), fun (Q) -> info_down(Q, down) end). + +info_all(VHostPath, Items) -> + map(list(VHostPath), fun (Q) -> info(Q, Items) end) ++ + map(list_down(VHostPath), fun (Q) -> info_down(Q, Items, down) end). + +emit_info_local(VHostPath, Items, Ref, AggregatorPid) -> + rabbit_control_misc:emitting_map_with_exit_handler( + AggregatorPid, Ref, fun(Q) -> info(Q, Items) end, list_local(VHostPath)). + +emit_info_all(Nodes, VHostPath, Items, Ref, AggregatorPid) -> + Pids = [ spawn_link(Node, rabbit_amqqueue, emit_info_local, [VHostPath, Items, Ref, AggregatorPid]) || Node <- Nodes ], + rabbit_control_misc:await_emitters_termination(Pids). + +emit_info_down(VHostPath, Items, Ref, AggregatorPid) -> + rabbit_control_misc:emitting_map_with_exit_handler( + AggregatorPid, Ref, fun(Q) -> info_down(Q, Items, down) end, + list_down(VHostPath)). + +info_local(VHostPath) -> + map(list_local(VHostPath), fun (Q) -> info(Q, [name]) end). + +list_local(VHostPath) -> + [ Q || #amqqueue{state = State, pid=QPid} = Q <- list(VHostPath), + State =/= crashed, + node() =:= node(QPid) ]. + +force_event_refresh(Ref) -> + [gen_server2:cast(Q#amqqueue.pid, + {force_event_refresh, Ref}) || Q <- list()], + ok. + +notify_policy_changed(#amqqueue{pid = QPid}) -> + gen_server2:cast(QPid, policy_changed). + +consumers(#amqqueue{ pid = QPid }) -> delegate:call(QPid, consumers). + +consumer_info_keys() -> ?CONSUMER_INFO_KEYS. + +consumers_all(VHostPath) -> + ConsumerInfoKeys = consumer_info_keys(), + lists:append( + map(list(VHostPath), + fun(Q) -> get_queue_consumer_info(Q, ConsumerInfoKeys) end)). + +emit_consumers_all(Nodes, VHostPath, Ref, AggregatorPid) -> + Pids = [ spawn_link(Node, rabbit_amqqueue, emit_consumers_local, [VHostPath, Ref, AggregatorPid]) || Node <- Nodes ], + rabbit_control_misc:await_emitters_termination(Pids), + ok. + +emit_consumers_local(VHostPath, Ref, AggregatorPid) -> + ConsumerInfoKeys = consumer_info_keys(), + rabbit_control_misc:emitting_map( + AggregatorPid, Ref, + fun(Q) -> get_queue_consumer_info(Q, ConsumerInfoKeys) end, + list_local(VHostPath)). + +get_queue_consumer_info(Q, ConsumerInfoKeys) -> + [lists:zip(ConsumerInfoKeys, + [Q#amqqueue.name, ChPid, CTag, + AckRequired, Prefetch, Args]) || + {ChPid, CTag, AckRequired, Prefetch, Args, _} <- consumers(Q)]. + +stat(#amqqueue{pid = QPid}) -> delegate:call(QPid, stat). + +pid_of(#amqqueue{pid = Pid}) -> Pid. +pid_of(VHost, QueueName) -> + case lookup(rabbit_misc:r(VHost, queue, QueueName)) of + {ok, Q} -> pid_of(Q); + {error, not_found} = E -> E + end. + +delete_exclusive(QPids, ConnId) -> + [gen_server2:cast(QPid, {delete_exclusive, ConnId}) || QPid <- QPids], + ok. + +delete_immediately(QPids) -> + [gen_server2:cast(QPid, delete_immediately) || QPid <- QPids], + ok. + +delete(#amqqueue{ pid = QPid }, IfUnused, IfEmpty, ActingUser) -> + delegate:call(QPid, {delete, IfUnused, IfEmpty, ActingUser}). + +delete_crashed(Q) -> + delete_crashed(Q, ?INTERNAL_USER). + +delete_crashed(#amqqueue{ pid = QPid } = Q, ActingUser) -> + ok = rpc:call(node(QPid), ?MODULE, delete_crashed_internal, [Q, ActingUser]). + +delete_crashed_internal(Q = #amqqueue{ name = QName }, ActingUser) -> + {ok, BQ} = application:get_env(rabbit, backing_queue_module), + BQ:delete_crashed(Q), + ok = internal_delete(QName, ActingUser). + +purge(#amqqueue{ pid = QPid }) -> delegate:call(QPid, purge). + +requeue(QPid, MsgIds, ChPid) -> delegate:call(QPid, {requeue, MsgIds, ChPid}). + +ack(QPid, MsgIds, ChPid) -> delegate:cast(QPid, {ack, MsgIds, ChPid}). + +reject(QPid, Requeue, MsgIds, ChPid) -> + delegate:cast(QPid, {reject, Requeue, MsgIds, ChPid}). + +notify_down_all(QPids, ChPid) -> + notify_down_all(QPids, ChPid, ?CHANNEL_OPERATION_TIMEOUT). + +notify_down_all(QPids, ChPid, Timeout) -> + case rpc:call(node(), delegate, call, + [QPids, {notify_down, ChPid}], Timeout) of + {badrpc, timeout} -> {error, {channel_operation_timeout, Timeout}}; + {badrpc, Reason} -> {error, Reason}; + {_, Bads} -> + case lists:filter( + fun ({_Pid, {exit, {R, _}, _}}) -> + rabbit_misc:is_abnormal_exit(R); + ({_Pid, _}) -> false + end, Bads) of + [] -> ok; + Bads1 -> {error, Bads1} + end; + Error -> {error, Error} + end. + +activate_limit_all(QPids, ChPid) -> + delegate:cast(QPids, {activate_limit, ChPid}). + +credit(#amqqueue{pid = QPid}, ChPid, CTag, Credit, Drain) -> + delegate:cast(QPid, {credit, ChPid, CTag, Credit, Drain}). + +basic_get(#amqqueue{pid = QPid}, ChPid, NoAck, LimiterPid) -> + delegate:call(QPid, {basic_get, ChPid, NoAck, LimiterPid}). + +basic_consume(#amqqueue{pid = QPid, name = QName}, NoAck, ChPid, LimiterPid, + LimiterActive, ConsumerPrefetchCount, ConsumerTag, + ExclusiveConsume, Args, OkMsg, ActingUser) -> + ok = check_consume_arguments(QName, Args), + delegate:call(QPid, {basic_consume, NoAck, ChPid, LimiterPid, LimiterActive, + ConsumerPrefetchCount, ConsumerTag, ExclusiveConsume, + Args, OkMsg, ActingUser}). + +basic_cancel(#amqqueue{pid = QPid}, ChPid, ConsumerTag, OkMsg, ActingUser) -> + delegate:call(QPid, {basic_cancel, ChPid, ConsumerTag, OkMsg, ActingUser}). + +notify_decorators(#amqqueue{pid = QPid}) -> + delegate:cast(QPid, notify_decorators). + +notify_sent(QPid, ChPid) -> + rabbit_amqqueue_common:notify_sent(QPid, ChPid). + +notify_sent_queue_down(QPid) -> + rabbit_amqqueue_common:notify_sent_queue_down(QPid). + +resume(QPid, ChPid) -> delegate:cast(QPid, {resume, ChPid}). + +internal_delete1(QueueName, OnlyDurable) -> + ok = mnesia:delete({rabbit_queue, QueueName}), + %% this 'guarded' delete prevents unnecessary writes to the mnesia + %% disk log + case mnesia:wread({rabbit_durable_queue, QueueName}) of + [] -> ok; + [_] -> ok = mnesia:delete({rabbit_durable_queue, QueueName}) + end, + %% we want to execute some things, as decided by rabbit_exchange, + %% after the transaction. + rabbit_binding:remove_for_destination(QueueName, OnlyDurable). + +internal_delete(QueueName, ActingUser) -> + rabbit_misc:execute_mnesia_tx_with_tail( + fun () -> + case {mnesia:wread({rabbit_queue, QueueName}), + mnesia:wread({rabbit_durable_queue, QueueName})} of + {[], []} -> + rabbit_misc:const({error, not_found}); + _ -> + Deletions = internal_delete1(QueueName, false), + T = rabbit_binding:process_deletions(Deletions, + ?INTERNAL_USER), + fun() -> + ok = T(), + rabbit_core_metrics:queue_deleted(QueueName), + ok = rabbit_event:notify(queue_deleted, + [{name, QueueName}, + {user_who_performed_action, ActingUser}]) + end + end + end). + +forget_all_durable(Node) -> + %% Note rabbit is not running so we avoid e.g. the worker pool. Also why + %% we don't invoke the return from rabbit_binding:process_deletions/1. + {atomic, ok} = + mnesia:sync_transaction( + fun () -> + Qs = mnesia:match_object(rabbit_durable_queue, + #amqqueue{_ = '_'}, write), + [forget_node_for_queue(Node, Q) || + #amqqueue{pid = Pid} = Q <- Qs, + node(Pid) =:= Node], + ok + end), + ok. + +%% Try to promote a slave while down - it should recover as a +%% master. We try to take the oldest slave here for best chance of +%% recovery. +forget_node_for_queue(DeadNode, Q = #amqqueue{recoverable_slaves = RS}) -> + forget_node_for_queue(DeadNode, RS, Q). + +forget_node_for_queue(_DeadNode, [], #amqqueue{name = Name}) -> + %% No slaves to recover from, queue is gone. + %% Don't process_deletions since that just calls callbacks and we + %% are not really up. + internal_delete1(Name, true); + +%% Should not happen, but let's be conservative. +forget_node_for_queue(DeadNode, [DeadNode | T], Q) -> + forget_node_for_queue(DeadNode, T, Q); + +forget_node_for_queue(DeadNode, [H|T], Q) -> + case node_permits_offline_promotion(H) of + false -> forget_node_for_queue(DeadNode, T, Q); + true -> Q1 = Q#amqqueue{pid = rabbit_misc:node_to_fake_pid(H)}, + ok = mnesia:write(rabbit_durable_queue, Q1, write) + end. + +node_permits_offline_promotion(Node) -> + case node() of + Node -> not rabbit:is_running(); %% [1] + _ -> All = rabbit_mnesia:cluster_nodes(all), + Running = rabbit_mnesia:cluster_nodes(running), + lists:member(Node, All) andalso + not lists:member(Node, Running) %% [2] + end. +%% [1] In this case if we are a real running node (i.e. rabbitmqctl +%% has RPCed into us) then we cannot allow promotion. If on the other +%% hand we *are* rabbitmqctl impersonating the node for offline +%% node-forgetting then we can. +%% +%% [2] This is simpler; as long as it's down that's OK + +run_backing_queue(QPid, Mod, Fun) -> + gen_server2:cast(QPid, {run_backing_queue, Mod, Fun}). + +set_ram_duration_target(QPid, Duration) -> + gen_server2:cast(QPid, {set_ram_duration_target, Duration}). + +set_maximum_since_use(QPid, Age) -> + gen_server2:cast(QPid, {set_maximum_since_use, Age}). + +update_mirroring(QPid) -> ok = delegate:cast(QPid, update_mirroring). + +sync_mirrors(#amqqueue{pid = QPid}) -> delegate:call(QPid, sync_mirrors); +sync_mirrors(QPid) -> delegate:call(QPid, sync_mirrors). +cancel_sync_mirrors(#amqqueue{pid = QPid}) -> delegate:call(QPid, cancel_sync_mirrors); +cancel_sync_mirrors(QPid) -> delegate:call(QPid, cancel_sync_mirrors). + +is_mirrored(Q) -> + rabbit_mirror_queue_misc:is_mirrored(Q). + +on_node_up(Node) -> + ok = rabbit_misc:execute_mnesia_transaction( + fun () -> + Qs = mnesia:match_object(rabbit_queue, + #amqqueue{_ = '_'}, write), + [maybe_clear_recoverable_node(Node, Q) || Q <- Qs], + ok + end). + +maybe_clear_recoverable_node(Node, + #amqqueue{sync_slave_pids = SPids, + recoverable_slaves = RSs} = Q) -> + case lists:member(Node, RSs) of + true -> + %% There is a race with + %% rabbit_mirror_queue_slave:record_synchronised/1 called + %% by the incoming slave node and this function, called + %% by the master node. If this function is executed after + %% record_synchronised/1, the node is erroneously removed + %% from the recoverable slaves list. + %% + %% We check if the slave node's queue PID is alive. If it is + %% the case, then this function is executed after. In this + %% situation, we don't touch the queue record, it is already + %% correct. + DoClearNode = + case [SP || SP <- SPids, node(SP) =:= Node] of + [SPid] -> not rabbit_misc:is_process_alive(SPid); + _ -> true + end, + if + DoClearNode -> RSs1 = RSs -- [Node], + store_queue( + Q#amqqueue{recoverable_slaves = RSs1}); + true -> ok + end; + false -> + ok + end. + +on_node_down(Node) -> + rabbit_misc:execute_mnesia_tx_with_tail( + fun () -> QsDels = + qlc:e(qlc:q([{QName, delete_queue(QName)} || + #amqqueue{name = QName, pid = Pid} = Q + <- mnesia:table(rabbit_queue), + not rabbit_amqqueue:is_mirrored(Q) andalso + node(Pid) == Node andalso + not rabbit_mnesia:is_process_alive(Pid)])), + {Qs, Dels} = lists:unzip(QsDels), + T = rabbit_binding:process_deletions( + lists:foldl(fun rabbit_binding:combine_deletions/2, + rabbit_binding:new_deletions(), Dels), + ?INTERNAL_USER), + fun () -> + T(), + lists:foreach( + fun(QName) -> + rabbit_core_metrics:queue_deleted(QName), + ok = rabbit_event:notify(queue_deleted, + [{name, QName}, + {user, ?INTERNAL_USER}]) + end, Qs) + end + end). + +delete_queue(QueueName) -> + ok = mnesia:delete({rabbit_queue, QueueName}), + rabbit_binding:remove_transient_for_destination(QueueName). + +pseudo_queue(QueueName, Pid) -> + #amqqueue{name = QueueName, + durable = false, + auto_delete = false, + arguments = [], + pid = Pid, + slave_pids = []}. + +immutable(Q) -> Q#amqqueue{pid = none, + slave_pids = none, + sync_slave_pids = none, + recoverable_slaves = none, + gm_pids = none, + policy = none, + decorators = none, + state = none}. + +deliver([], _Delivery) -> + %% /dev/null optimisation + []; + +deliver(Qs, Delivery = #delivery{flow = Flow}) -> + {MPids, SPids} = qpids(Qs), + QPids = MPids ++ SPids, + %% We use up two credits to send to a slave since the message + %% arrives at the slave from two directions. We will ack one when + %% the slave receives the message direct from the channel, and the + %% other when it receives it via GM. + case Flow of + %% Here we are tracking messages sent by the rabbit_channel + %% process. We are accessing the rabbit_channel process + %% dictionary. + flow -> [credit_flow:send(QPid) || QPid <- QPids], + [credit_flow:send(QPid) || QPid <- SPids]; + noflow -> ok + end, + + %% We let slaves know that they were being addressed as slaves at + %% the time - if they receive such a message from the channel + %% after they have become master they should mark the message as + %% 'delivered' since they do not know what the master may have + %% done with it. + MMsg = {deliver, Delivery, false}, + SMsg = {deliver, Delivery, true}, + delegate:cast(MPids, MMsg), + delegate:cast(SPids, SMsg), + QPids. + +qpids([]) -> {[], []}; %% optimisation +qpids([#amqqueue{pid = QPid, slave_pids = SPids}]) -> {[QPid], SPids}; %% opt +qpids(Qs) -> + {MPids, SPids} = lists:foldl(fun (#amqqueue{pid = QPid, slave_pids = SPids}, + {MPidAcc, SPidAcc}) -> + {[QPid | MPidAcc], [SPids | SPidAcc]} + end, {[], []}, Qs), + {MPids, lists:append(SPids)}. diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl new file mode 100644 index 0000000000..7335a35242 --- /dev/null +++ b/src/rabbit_auth_backend_internal.erl @@ -0,0 +1,604 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_auth_backend_internal). +-include("rabbit.hrl"). + +-behaviour(rabbit_authn_backend). +-behaviour(rabbit_authz_backend). + +-export([user_login_authentication/2, user_login_authorization/1, + check_vhost_access/3, check_resource_access/3, check_topic_access/4]). + +-export([add_user/3, delete_user/2, lookup_user/1, + change_password/3, clear_password/2, + hash_password/2, change_password_hash/2, change_password_hash/3, + set_tags/3, set_permissions/6, clear_permissions/3, + set_topic_permissions/6, clear_topic_permissions/3, clear_topic_permissions/4, + add_user_sans_validation/3]). + +-export([user_info_keys/0, perms_info_keys/0, + user_perms_info_keys/0, vhost_perms_info_keys/0, + user_vhost_perms_info_keys/0, + list_users/0, list_users/2, list_permissions/0, + list_user_permissions/1, list_user_permissions/3, + list_topic_permissions/0, + list_vhost_permissions/1, list_vhost_permissions/3, + list_user_vhost_permissions/2, + list_user_topic_permissions/1, list_vhost_topic_permissions/1, list_user_vhost_topic_permissions/2]). + +%% for testing +-export([hashing_module_for_user/1, expand_topic_permission/2]). + +%%---------------------------------------------------------------------------- + +-type regexp() :: binary(). + +-spec add_user(rabbit_types:username(), rabbit_types:password(), + rabbit_types:username()) -> 'ok' | {'error', string()}. +-spec delete_user(rabbit_types:username(), rabbit_types:username()) -> 'ok'. +-spec lookup_user + (rabbit_types:username()) -> + rabbit_types:ok(rabbit_types:internal_user()) | + rabbit_types:error('not_found'). +-spec change_password + (rabbit_types:username(), rabbit_types:password(), rabbit_types:username()) -> 'ok'. +-spec clear_password(rabbit_types:username(), rabbit_types:username()) -> 'ok'. +-spec hash_password + (module(), rabbit_types:password()) -> rabbit_types:password_hash(). +-spec change_password_hash + (rabbit_types:username(), rabbit_types:password_hash()) -> 'ok'. +-spec set_tags(rabbit_types:username(), [atom()], rabbit_types:username()) -> 'ok'. +-spec set_permissions + (rabbit_types:username(), rabbit_types:vhost(), regexp(), regexp(), + regexp(), rabbit_types:username()) -> + 'ok'. +-spec clear_permissions + (rabbit_types:username(), rabbit_types:vhost(), rabbit_types:username()) -> 'ok'. +-spec user_info_keys() -> rabbit_types:info_keys(). +-spec perms_info_keys() -> rabbit_types:info_keys(). +-spec user_perms_info_keys() -> rabbit_types:info_keys(). +-spec vhost_perms_info_keys() -> rabbit_types:info_keys(). +-spec user_vhost_perms_info_keys() -> rabbit_types:info_keys(). +-spec list_users() -> [rabbit_types:infos()]. +-spec list_users(reference(), pid()) -> 'ok'. +-spec list_permissions() -> [rabbit_types:infos()]. +-spec list_user_permissions + (rabbit_types:username()) -> [rabbit_types:infos()]. +-spec list_user_permissions + (rabbit_types:username(), reference(), pid()) -> 'ok'. +-spec list_vhost_permissions + (rabbit_types:vhost()) -> [rabbit_types:infos()]. +-spec list_vhost_permissions + (rabbit_types:vhost(), reference(), pid()) -> 'ok'. +-spec list_user_vhost_permissions + (rabbit_types:username(), rabbit_types:vhost()) -> [rabbit_types:infos()]. + +%%---------------------------------------------------------------------------- +%% Implementation of rabbit_auth_backend + +%% Returns a password hashing module for the user record provided. If +%% there is no information in the record, we consider it to be legacy +%% (inserted by a version older than 3.6.0) and fall back to MD5, the +%% now obsolete hashing function. +hashing_module_for_user(#internal_user{ + hashing_algorithm = ModOrUndefined}) -> + rabbit_password:hashing_mod(ModOrUndefined). + +user_login_authentication(Username, []) -> + internal_check_user_login(Username, fun(_) -> true end); +user_login_authentication(Username, AuthProps) -> + case lists:keyfind(password, 1, AuthProps) of + {password, Cleartext} -> + internal_check_user_login( + Username, + fun (#internal_user{ + password_hash = <<Salt:4/binary, Hash/binary>> + } = U) -> + Hash =:= rabbit_password:salted_hash( + hashing_module_for_user(U), Salt, Cleartext); + (#internal_user{}) -> + false + end); + false -> exit({unknown_auth_props, Username, AuthProps}) + end. + +user_login_authorization(Username) -> + case user_login_authentication(Username, []) of + {ok, #auth_user{impl = Impl, tags = Tags}} -> {ok, Impl, Tags}; + Else -> Else + end. + +internal_check_user_login(Username, Fun) -> + Refused = {refused, "user '~s' - invalid credentials", [Username]}, + case lookup_user(Username) of + {ok, User = #internal_user{tags = Tags}} -> + case Fun(User) of + true -> {ok, #auth_user{username = Username, + tags = Tags, + impl = none}}; + _ -> Refused + end; + {error, not_found} -> + Refused + end. + +check_vhost_access(#auth_user{username = Username}, VHostPath, _Sock) -> + case mnesia:dirty_read({rabbit_user_permission, + #user_vhost{username = Username, + virtual_host = VHostPath}}) of + [] -> false; + [_R] -> true + end. + +check_resource_access(#auth_user{username = Username}, + #resource{virtual_host = VHostPath, name = Name}, + Permission) -> + case mnesia:dirty_read({rabbit_user_permission, + #user_vhost{username = Username, + virtual_host = VHostPath}}) of + [] -> + false; + [#user_permission{permission = P}] -> + PermRegexp = case element(permission_index(Permission), P) of + %% <<"^$">> breaks Emacs' erlang mode + <<"">> -> <<$^, $$>>; + RE -> RE + end, + case re:run(Name, PermRegexp, [{capture, none}]) of + match -> true; + nomatch -> false + end + end. + +check_topic_access(#auth_user{username = Username}, + #resource{virtual_host = VHostPath, name = Name, kind = topic}, + Permission, + Context) -> + case mnesia:dirty_read({rabbit_topic_permission, + #topic_permission_key{user_vhost = #user_vhost{username = Username, + virtual_host = VHostPath}, + exchange = Name + }}) of + [] -> + true; + [#topic_permission{permission = P}] -> + PermRegexp = case element(permission_index(Permission), P) of + %% <<"^$">> breaks Emacs' erlang mode + <<"">> -> <<$^, $$>>; + RE -> RE + end, + PermRegexpExpanded = expand_topic_permission( + PermRegexp, + maps:get(variable_map, Context, undefined) + ), + case re:run(maps:get(routing_key, Context), PermRegexpExpanded, [{capture, none}]) of + match -> true; + nomatch -> false + end + end. + +expand_topic_permission(Permission, ToExpand) when is_map(ToExpand) -> + Opening = <<"{">>, + Closing = <<"}">>, + ReplaceFun = fun(K, V, Acc) -> + Placeholder = <<Opening/binary, K/binary, Closing/binary>>, + binary:replace(Acc, Placeholder, V, [global]) + end, + maps:fold(ReplaceFun, Permission, ToExpand); +expand_topic_permission(Permission, _ToExpand) -> + Permission. + +permission_index(configure) -> #permission.configure; +permission_index(write) -> #permission.write; +permission_index(read) -> #permission.read. + +%%---------------------------------------------------------------------------- +%% Manipulation of the user database + +validate_credentials(Username, Password) -> + rabbit_credential_validation:validate(Username, Password). + +validate_and_alternate_credentials(Username, Password, ActingUser, Fun) -> + case validate_credentials(Username, Password) of + ok -> + Fun(Username, Password, ActingUser); + {error, Err} -> + rabbit_log:error("Credential validation for '~s' failed!~n", [Username]), + {error, Err} + end. + +add_user(Username, Password, ActingUser) -> + validate_and_alternate_credentials(Username, Password, ActingUser, + fun add_user_sans_validation/3). + +add_user_sans_validation(Username, Password, ActingUser) -> + rabbit_log:info("Creating user '~s'~n", [Username]), + %% hash_password will pick the hashing function configured for us + %% but we also need to store a hint as part of the record, so we + %% retrieve it here one more time + HashingMod = rabbit_password:hashing_mod(), + User = #internal_user{username = Username, + password_hash = hash_password(HashingMod, Password), + tags = [], + hashing_algorithm = HashingMod}, + R = rabbit_misc:execute_mnesia_transaction( + fun () -> + case mnesia:wread({rabbit_user, Username}) of + [] -> + ok = mnesia:write(rabbit_user, User, write); + _ -> + mnesia:abort({user_already_exists, Username}) + end + end), + rabbit_event:notify(user_created, [{name, Username}, + {user_who_performed_action, ActingUser}]), + R. + +delete_user(Username, ActingUser) -> + rabbit_log:info("Deleting user '~s'~n", [Username]), + R = rabbit_misc:execute_mnesia_transaction( + rabbit_misc:with_user( + Username, + fun () -> + ok = mnesia:delete({rabbit_user, Username}), + [ok = mnesia:delete_object( + rabbit_user_permission, R, write) || + R <- mnesia:match_object( + rabbit_user_permission, + #user_permission{user_vhost = #user_vhost{ + username = Username, + virtual_host = '_'}, + permission = '_'}, + write)], + UserTopicPermissionsQuery = match_user_vhost_topic_permission(Username, '_'), + UserTopicPermissions = UserTopicPermissionsQuery(), + [ok = mnesia:delete_object(rabbit_topic_permission, R, write) || R <- UserTopicPermissions], + ok + end)), + rabbit_event:notify(user_deleted, + [{name, Username}, + {user_who_performed_action, ActingUser}]), + R. + +lookup_user(Username) -> + rabbit_misc:dirty_read({rabbit_user, Username}). + +change_password(Username, Password, ActingUser) -> + validate_and_alternate_credentials(Username, Password, ActingUser, + fun change_password_sans_validation/3). + +change_password_sans_validation(Username, Password, ActingUser) -> + rabbit_log:info("Changing password for '~s'~n", [Username]), + HashingAlgorithm = rabbit_password:hashing_mod(), + R = change_password_hash(Username, + hash_password(rabbit_password:hashing_mod(), + Password), + HashingAlgorithm), + rabbit_event:notify(user_password_changed, + [{name, Username}, + {user_who_performed_action, ActingUser}]), + R. + +clear_password(Username, ActingUser) -> + rabbit_log:info("Clearing password for '~s'~n", [Username]), + R = change_password_hash(Username, <<"">>), + rabbit_event:notify(user_password_cleared, + [{name, Username}, + {user_who_performed_action, ActingUser}]), + R. + +hash_password(HashingMod, Cleartext) -> + rabbit_password:hash(HashingMod, Cleartext). + +change_password_hash(Username, PasswordHash) -> + change_password_hash(Username, PasswordHash, rabbit_password:hashing_mod()). + + +change_password_hash(Username, PasswordHash, HashingAlgorithm) -> + update_user(Username, fun(User) -> + User#internal_user{ + password_hash = PasswordHash, + hashing_algorithm = HashingAlgorithm } + end). + +set_tags(Username, Tags, ActingUser) -> + ConvertedTags = [rabbit_data_coercion:to_atom(I) || I <- Tags], + rabbit_log:info("Setting user tags for user '~s' to ~p~n", + [Username, ConvertedTags]), + R = update_user(Username, fun(User) -> + User#internal_user{tags = ConvertedTags} + end), + rabbit_event:notify(user_tags_set, [{name, Username}, {tags, ConvertedTags}, + {user_who_performed_action, ActingUser}]), + R. + +set_permissions(Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm, ActingUser) -> + rabbit_log:info("Setting permissions for " + "'~s' in '~s' to '~s', '~s', '~s'~n", + [Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm]), + lists:map( + fun (RegexpBin) -> + Regexp = binary_to_list(RegexpBin), + case re:compile(Regexp) of + {ok, _} -> ok; + {error, Reason} -> throw({error, {invalid_regexp, + Regexp, Reason}}) + end + end, [ConfigurePerm, WritePerm, ReadPerm]), + R = rabbit_misc:execute_mnesia_transaction( + rabbit_vhost:with_user_and_vhost( + Username, VHostPath, + fun () -> ok = mnesia:write( + rabbit_user_permission, + #user_permission{user_vhost = #user_vhost{ + username = Username, + virtual_host = VHostPath}, + permission = #permission{ + configure = ConfigurePerm, + write = WritePerm, + read = ReadPerm}}, + write) + end)), + rabbit_event:notify(permission_created, [{user, Username}, + {vhost, VHostPath}, + {configure, ConfigurePerm}, + {write, WritePerm}, + {read, ReadPerm}, + {user_who_performed_action, ActingUser}]), + R. + +clear_permissions(Username, VHostPath, ActingUser) -> + R = rabbit_misc:execute_mnesia_transaction( + rabbit_vhost:with_user_and_vhost( + Username, VHostPath, + fun () -> + ok = mnesia:delete({rabbit_user_permission, + #user_vhost{username = Username, + virtual_host = VHostPath}}) + end)), + rabbit_event:notify(permission_deleted, [{user, Username}, + {vhost, VHostPath}, + {user_who_performed_action, ActingUser}]), + R. + + +update_user(Username, Fun) -> + rabbit_misc:execute_mnesia_transaction( + rabbit_misc:with_user( + Username, + fun () -> + {ok, User} = lookup_user(Username), + ok = mnesia:write(rabbit_user, Fun(User), write) + end)). + +set_topic_permissions(Username, VHostPath, Exchange, WritePerm, ReadPerm, ActingUser) -> + WritePermRegex = rabbit_data_coercion:to_binary(WritePerm), + ReadPermRegex = rabbit_data_coercion:to_binary(ReadPerm), + lists:map( + fun (RegexpBin) -> + case re:compile(RegexpBin) of + {ok, _} -> ok; + {error, Reason} -> throw({error, {invalid_regexp, + RegexpBin, Reason}}) + end + end, [WritePerm, ReadPerm]), + R = rabbit_misc:execute_mnesia_transaction( + rabbit_vhost:with_user_and_vhost( + Username, VHostPath, + fun () -> ok = mnesia:write( + rabbit_topic_permission, + #topic_permission{ + topic_permission_key = #topic_permission_key{ + user_vhost = #user_vhost{ + username = Username, + virtual_host = VHostPath}, + exchange = Exchange + }, + permission = #permission{ + write = WritePermRegex, + read = ReadPermRegex + } + }, + write) + end)), + rabbit_event:notify(topic_permission_created, [ + {user, Username}, + {vhost, VHostPath}, + {exchange, Exchange}, + {write, WritePermRegex}, + {read, ReadPermRegex}, + {user_who_performed_action, ActingUser}]), + R. + +clear_topic_permissions(Username, VHostPath, ActingUser) -> + R = rabbit_misc:execute_mnesia_transaction( + rabbit_vhost:with_user_and_vhost( + Username, VHostPath, + fun () -> + ListFunction = match_user_vhost_topic_permission(Username, VHostPath), + List = ListFunction(), + lists:foreach(fun(X) -> + ok = mnesia:delete_object(rabbit_topic_permission, X, write) + end, List) + end)), + rabbit_event:notify(topic_permission_deleted, [{user, Username}, + {vhost, VHostPath}, + {user_who_performed_action, ActingUser}]), + R. + +clear_topic_permissions(Username, VHostPath, Exchange, ActingUser) -> + R = rabbit_misc:execute_mnesia_transaction( + rabbit_vhost:with_user_and_vhost( + Username, VHostPath, + fun () -> + ok = mnesia:delete(rabbit_topic_permission, + #topic_permission_key{ + user_vhost = #user_vhost{ + username = Username, + virtual_host = VHostPath}, + exchange = Exchange + }, write) + end)), + rabbit_event:notify(permission_deleted, [{user, Username}, + {vhost, VHostPath}, + {user_who_performed_action, ActingUser}]), + R. + +%%---------------------------------------------------------------------------- +%% Listing + +-define(PERMS_INFO_KEYS, [configure, write, read]). +-define(USER_INFO_KEYS, [user, tags]). + +user_info_keys() -> ?USER_INFO_KEYS. + +perms_info_keys() -> [user, vhost | ?PERMS_INFO_KEYS]. +vhost_perms_info_keys() -> [user | ?PERMS_INFO_KEYS]. +user_perms_info_keys() -> [vhost | ?PERMS_INFO_KEYS]. +user_vhost_perms_info_keys() -> ?PERMS_INFO_KEYS. + +topic_perms_info_keys() -> [user, vhost, exchange, write, read]. +user_topic_perms_info_keys() -> [vhost, exchange, write, read]. +vhost_topic_perms_info_keys() -> [user, exchange, write, read]. +user_vhost_topic_perms_info_keys() -> [exchange, write, read]. + +list_users() -> + [extract_internal_user_params(U) || + U <- mnesia:dirty_match_object(rabbit_user, #internal_user{_ = '_'})]. + +list_users(Ref, AggregatorPid) -> + rabbit_control_misc:emitting_map( + AggregatorPid, Ref, + fun(U) -> extract_internal_user_params(U) end, + mnesia:dirty_match_object(rabbit_user, #internal_user{_ = '_'})). + +list_permissions() -> + list_permissions(perms_info_keys(), match_user_vhost('_', '_')). + +list_permissions(Keys, QueryThunk) -> + [extract_user_permission_params(Keys, U) || + U <- rabbit_misc:execute_mnesia_transaction(QueryThunk)]. + +list_permissions(Keys, QueryThunk, Ref, AggregatorPid) -> + rabbit_control_misc:emitting_map( + AggregatorPid, Ref, fun(U) -> extract_user_permission_params(Keys, U) end, + rabbit_misc:execute_mnesia_transaction(QueryThunk)). + +filter_props(Keys, Props) -> [T || T = {K, _} <- Props, lists:member(K, Keys)]. + +list_user_permissions(Username) -> + list_permissions( + user_perms_info_keys(), + rabbit_misc:with_user(Username, match_user_vhost(Username, '_'))). + +list_user_permissions(Username, Ref, AggregatorPid) -> + list_permissions( + user_perms_info_keys(), + rabbit_misc:with_user(Username, match_user_vhost(Username, '_')), + Ref, AggregatorPid). + +list_vhost_permissions(VHostPath) -> + list_permissions( + vhost_perms_info_keys(), + rabbit_vhost:with(VHostPath, match_user_vhost('_', VHostPath))). + +list_vhost_permissions(VHostPath, Ref, AggregatorPid) -> + list_permissions( + vhost_perms_info_keys(), + rabbit_vhost:with(VHostPath, match_user_vhost('_', VHostPath)), + Ref, AggregatorPid). + +list_user_vhost_permissions(Username, VHostPath) -> + list_permissions( + user_vhost_perms_info_keys(), + rabbit_vhost:with_user_and_vhost( + Username, VHostPath, match_user_vhost(Username, VHostPath))). + +extract_user_permission_params(Keys, #user_permission{ + user_vhost = + #user_vhost{username = Username, + virtual_host = VHostPath}, + permission = #permission{ + configure = ConfigurePerm, + write = WritePerm, + read = ReadPerm}}) -> + filter_props(Keys, [{user, Username}, + {vhost, VHostPath}, + {configure, ConfigurePerm}, + {write, WritePerm}, + {read, ReadPerm}]). + +extract_internal_user_params(#internal_user{username = Username, tags = Tags}) -> + [{user, Username}, {tags, Tags}]. + +match_user_vhost(Username, VHostPath) -> + fun () -> mnesia:match_object( + rabbit_user_permission, + #user_permission{user_vhost = #user_vhost{ + username = Username, + virtual_host = VHostPath}, + permission = '_'}, + read) + end. + +list_topic_permissions() -> + list_topic_permissions(topic_perms_info_keys(), match_user_vhost_topic_permission('_', '_')). + +list_user_topic_permissions(Username) -> + list_topic_permissions(user_topic_perms_info_keys(), + rabbit_misc:with_user(Username, match_user_vhost_topic_permission(Username, '_'))). + +list_vhost_topic_permissions(VHost) -> + list_topic_permissions(vhost_topic_perms_info_keys(), + rabbit_vhost:with(VHost, match_user_vhost_topic_permission('_', VHost))). + +list_user_vhost_topic_permissions(Username, VHost) -> + list_topic_permissions(user_vhost_topic_perms_info_keys(), + rabbit_vhost:with_user_and_vhost(Username, VHost, match_user_vhost_topic_permission(Username, VHost))). + +list_topic_permissions(Keys, QueryThunk) -> + [extract_topic_permission_params(Keys, U) || + U <- rabbit_misc:execute_mnesia_transaction(QueryThunk)]. + +match_user_vhost_topic_permission(Username, VHostPath) -> + match_user_vhost_topic_permission(Username, VHostPath, '_'). + +match_user_vhost_topic_permission(Username, VHostPath, Exchange) -> + fun () -> mnesia:match_object( + rabbit_topic_permission, + #topic_permission{topic_permission_key = #topic_permission_key{ + user_vhost = #user_vhost{ + username = Username, + virtual_host = VHostPath}, + exchange = Exchange}, + permission = '_'}, + read) + end. + +extract_topic_permission_params(Keys, #topic_permission{ + topic_permission_key = #topic_permission_key{ + user_vhost = #user_vhost{username = Username, + virtual_host = VHostPath}, + exchange = Exchange}, + permission = #permission{ + write = WritePerm, + read = ReadPerm}}) -> + filter_props(Keys, [{user, Username}, + {vhost, VHostPath}, + {exchange, Exchange}, + {write, WritePerm}, + {read, ReadPerm}]). diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl new file mode 100644 index 0000000000..d474e9cad3 --- /dev/null +++ b/src/rabbit_basic.erl @@ -0,0 +1,302 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_basic). +-include("rabbit.hrl"). +-include("rabbit_framing.hrl"). + +-export([publish/4, publish/5, publish/1, + message/3, message/4, properties/1, prepend_table_header/3, + extract_headers/1, extract_timestamp/1, map_headers/2, delivery/4, + header_routes/1, parse_expiration/1, header/2, header/3]). +-export([build_content/2, from_content/1, msg_size/1, maybe_gc_large_msg/1]). + +%%---------------------------------------------------------------------------- + +-type properties_input() :: + rabbit_framing:amqp_property_record() | [{atom(), any()}]. +-type publish_result() :: + {ok, [pid()]} | rabbit_types:error('not_found'). +-type header() :: any(). +-type headers() :: rabbit_framing:amqp_table() | 'undefined'. + +-type exchange_input() :: rabbit_types:exchange() | rabbit_exchange:name(). +-type body_input() :: binary() | [binary()]. + +-spec publish + (exchange_input(), rabbit_router:routing_key(), properties_input(), + body_input()) -> + publish_result(). +-spec publish + (exchange_input(), rabbit_router:routing_key(), boolean(), + properties_input(), body_input()) -> + publish_result(). +-spec publish(rabbit_types:delivery()) -> publish_result(). +-spec delivery + (boolean(), boolean(), rabbit_types:message(), undefined | integer()) -> + rabbit_types:delivery(). +-spec message + (rabbit_exchange:name(), rabbit_router:routing_key(), properties_input(), + binary()) -> + rabbit_types:message(). +-spec message + (rabbit_exchange:name(), rabbit_router:routing_key(), + rabbit_types:decoded_content()) -> + rabbit_types:ok_or_error2(rabbit_types:message(), any()). +-spec properties + (properties_input()) -> rabbit_framing:amqp_property_record(). + +-spec prepend_table_header + (binary(), rabbit_framing:amqp_table(), headers()) -> headers(). + +-spec header(header(), headers()) -> 'undefined' | any(). +-spec header(header(), headers(), any()) -> 'undefined' | any(). + +-spec extract_headers(rabbit_types:content()) -> headers(). + +-spec map_headers + (fun((headers()) -> headers()), rabbit_types:content()) -> + rabbit_types:content(). + +-spec header_routes(undefined | rabbit_framing:amqp_table()) -> [string()]. +-spec build_content + (rabbit_framing:amqp_property_record(), binary() | [binary()]) -> + rabbit_types:content(). +-spec from_content + (rabbit_types:content()) -> + {rabbit_framing:amqp_property_record(), binary()}. +-spec parse_expiration + (rabbit_framing:amqp_property_record()) -> + rabbit_types:ok_or_error2('undefined' | non_neg_integer(), any()). + +%%---------------------------------------------------------------------------- + +%% Convenience function, for avoiding round-trips in calls across the +%% erlang distributed network. +publish(Exchange, RoutingKeyBin, Properties, Body) -> + publish(Exchange, RoutingKeyBin, false, Properties, Body). + +%% Convenience function, for avoiding round-trips in calls across the +%% erlang distributed network. +publish(X = #exchange{name = XName}, RKey, Mandatory, Props, Body) -> + Message = message(XName, RKey, properties(Props), Body), + publish(X, delivery(Mandatory, false, Message, undefined)); +publish(XName, RKey, Mandatory, Props, Body) -> + Message = message(XName, RKey, properties(Props), Body), + publish(delivery(Mandatory, false, Message, undefined)). + +publish(Delivery = #delivery{ + message = #basic_message{exchange_name = XName}}) -> + case rabbit_exchange:lookup(XName) of + {ok, X} -> publish(X, Delivery); + Err -> Err + end. + +publish(X, Delivery) -> + Qs = rabbit_amqqueue:lookup(rabbit_exchange:route(X, Delivery)), + DeliveredQPids = rabbit_amqqueue:deliver(Qs, Delivery), + {ok, DeliveredQPids}. + +delivery(Mandatory, Confirm, Message, MsgSeqNo) -> + #delivery{mandatory = Mandatory, confirm = Confirm, sender = self(), + message = Message, msg_seq_no = MsgSeqNo, flow = noflow}. + +build_content(Properties, BodyBin) when is_binary(BodyBin) -> + build_content(Properties, [BodyBin]); + +build_content(Properties, PFR) -> + %% basic.publish hasn't changed so we can just hard-code amqp_0_9_1 + {ClassId, _MethodId} = + rabbit_framing_amqp_0_9_1:method_id('basic.publish'), + #content{class_id = ClassId, + properties = Properties, + properties_bin = none, + protocol = none, + payload_fragments_rev = PFR}. + +from_content(Content) -> + #content{class_id = ClassId, + properties = Props, + payload_fragments_rev = FragmentsRev} = + rabbit_binary_parser:ensure_content_decoded(Content), + %% basic.publish hasn't changed so we can just hard-code amqp_0_9_1 + {ClassId, _MethodId} = + rabbit_framing_amqp_0_9_1:method_id('basic.publish'), + {Props, list_to_binary(lists:reverse(FragmentsRev))}. + +%% This breaks the spec rule forbidding message modification +strip_header(#content{properties = #'P_basic'{headers = undefined}} + = DecodedContent, _Key) -> + DecodedContent; +strip_header(#content{properties = Props = #'P_basic'{headers = Headers}} + = DecodedContent, Key) -> + case lists:keysearch(Key, 1, Headers) of + false -> DecodedContent; + {value, Found} -> Headers0 = lists:delete(Found, Headers), + rabbit_binary_generator:clear_encoded_content( + DecodedContent#content{ + properties = Props#'P_basic'{ + headers = Headers0}}) + end. + +message(XName, RoutingKey, #content{properties = Props} = DecodedContent) -> + try + {ok, #basic_message{ + exchange_name = XName, + content = strip_header(DecodedContent, ?DELETED_HEADER), + id = rabbit_guid:gen(), + is_persistent = is_message_persistent(DecodedContent), + routing_keys = [RoutingKey | + header_routes(Props#'P_basic'.headers)]}} + catch + {error, _Reason} = Error -> Error + end. + +message(XName, RoutingKey, RawProperties, Body) -> + Properties = properties(RawProperties), + Content = build_content(Properties, Body), + {ok, Msg} = message(XName, RoutingKey, Content), + Msg. + +properties(P = #'P_basic'{}) -> + P; +properties(P) when is_list(P) -> + %% Yes, this is O(length(P) * record_info(size, 'P_basic') / 2), + %% i.e. slow. Use the definition of 'P_basic' directly if + %% possible! + lists:foldl(fun ({Key, Value}, Acc) -> + case indexof(record_info(fields, 'P_basic'), Key) of + 0 -> throw({unknown_basic_property, Key}); + N -> setelement(N + 1, Acc, Value) + end + end, #'P_basic'{}, P). + +prepend_table_header(Name, Info, undefined) -> + prepend_table_header(Name, Info, []); +prepend_table_header(Name, Info, Headers) -> + case rabbit_misc:table_lookup(Headers, Name) of + {array, Existing} -> + prepend_table(Name, Info, Existing, Headers); + undefined -> + prepend_table(Name, Info, [], Headers); + Other -> + Headers2 = prepend_table(Name, Info, [], Headers), + set_invalid_header(Name, Other, Headers2) + end. + +prepend_table(Name, Info, Prior, Headers) -> + rabbit_misc:set_table_value(Headers, Name, array, [{table, Info} | Prior]). + +set_invalid_header(Name, {_, _}=Value, Headers) when is_list(Headers) -> + case rabbit_misc:table_lookup(Headers, ?INVALID_HEADERS_KEY) of + undefined -> + set_invalid([{Name, array, [Value]}], Headers); + {table, ExistingHdr} -> + update_invalid(Name, Value, ExistingHdr, Headers); + Other -> + %% somehow the x-invalid-headers header is corrupt + Invalid = [{?INVALID_HEADERS_KEY, array, [Other]}], + set_invalid_header(Name, Value, set_invalid(Invalid, Headers)) + end. + +set_invalid(NewHdr, Headers) -> + rabbit_misc:set_table_value(Headers, ?INVALID_HEADERS_KEY, table, NewHdr). + +update_invalid(Name, Value, ExistingHdr, Header) -> + Values = case rabbit_misc:table_lookup(ExistingHdr, Name) of + undefined -> [Value]; + {array, Prior} -> [Value | Prior] + end, + NewHdr = rabbit_misc:set_table_value(ExistingHdr, Name, array, Values), + set_invalid(NewHdr, Header). + +header(_Header, undefined) -> + undefined; +header(_Header, []) -> + undefined; +header(Header, Headers) -> + header(Header, Headers, undefined). + +header(Header, Headers, Default) -> + case lists:keysearch(Header, 1, Headers) of + false -> Default; + {value, Val} -> Val + end. + +extract_headers(Content) -> + #content{properties = #'P_basic'{headers = Headers}} = + rabbit_binary_parser:ensure_content_decoded(Content), + Headers. + +extract_timestamp(Content) -> + #content{properties = #'P_basic'{timestamp = Timestamp}} = + rabbit_binary_parser:ensure_content_decoded(Content), + Timestamp. + +map_headers(F, Content) -> + Content1 = rabbit_binary_parser:ensure_content_decoded(Content), + #content{properties = #'P_basic'{headers = Headers} = Props} = Content1, + Headers1 = F(Headers), + rabbit_binary_generator:clear_encoded_content( + Content1#content{properties = Props#'P_basic'{headers = Headers1}}). + +indexof(L, Element) -> indexof(L, Element, 1). + +indexof([], _Element, _N) -> 0; +indexof([Element | _Rest], Element, N) -> N; +indexof([_ | Rest], Element, N) -> indexof(Rest, Element, N + 1). + +is_message_persistent(#content{properties = #'P_basic'{ + delivery_mode = Mode}}) -> + case Mode of + 1 -> false; + 2 -> true; + undefined -> false; + Other -> throw({error, {delivery_mode_unknown, Other}}) + end. + +%% Extract CC routes from headers +header_routes(undefined) -> + []; +header_routes(HeadersTable) -> + lists:append( + [case rabbit_misc:table_lookup(HeadersTable, HeaderKey) of + {array, Routes} -> [Route || {longstr, Route} <- Routes]; + undefined -> []; + {Type, _Val} -> throw({error, {unacceptable_type_in_header, + binary_to_list(HeaderKey), Type}}) + end || HeaderKey <- ?ROUTING_HEADERS]). + +parse_expiration(#'P_basic'{expiration = undefined}) -> + {ok, undefined}; +parse_expiration(#'P_basic'{expiration = Expiration}) -> + case string:to_integer(binary_to_list(Expiration)) of + {error, no_integer} = E -> + E; + {N, ""} -> + case rabbit_misc:check_expiry(N) of + ok -> {ok, N}; + E = {error, _} -> E + end; + {_, S} -> + {error, {leftover_string, S}} + end. + +maybe_gc_large_msg(Content) -> + rabbit_writer:maybe_gc_large_msg(Content). + +msg_size(Content) -> + rabbit_writer:msg_size(Content). diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl new file mode 100644 index 0000000000..378c1e583b --- /dev/null +++ b/src/rabbit_channel.erl @@ -0,0 +1,2150 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_channel). + +%% rabbit_channel processes represent an AMQP 0-9-1 channels. +%% +%% Connections parse protocol frames coming from clients and +%% dispatch them to channel processes. +%% Channels are responsible for implementing the logic behind +%% the various protocol methods, involving other processes as +%% needed: +%% +%% * Routing messages (using functions in various exchange type +%% modules) to queue processes. +%% * Managing queues, exchanges, and bindings. +%% * Keeping track of consumers +%% * Keeping track of unacknowledged deliveries to consumers +%% * Keeping track of publisher confirms +%% * Keeping track of mandatory message routing confirmations +%% and returns +%% * Transaction management +%% * Authorisation (enforcing permissions) +%% * Publishing trace events if tracing is enabled +%% +%% Every channel has a number of dependent processes: +%% +%% * A writer which is responsible for sending frames to clients. +%% * A limiter which controls how many messages can be delivered +%% to consumers according to active QoS prefetch and internal +%% flow control logic. +%% +%% Channels are also aware of their connection's queue collector. +%% When a queue is declared as exclusive on a channel, the channel +%% will notify queue collector of that queue. + +-include("rabbit_framing.hrl"). +-include("rabbit.hrl"). + +-behaviour(gen_server2). + +-export([start_link/11, do/2, do/3, do_flow/3, flush/1, shutdown/1]). +-export([send_command/2, deliver/4, deliver_reply/2, + send_credit_reply/2, send_drained/2]). +-export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1, + emit_info_all/4, info_local/1]). +-export([refresh_config_local/0, ready_for_close/1]). +-export([refresh_interceptors/0]). +-export([force_event_refresh/1]). + +-export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2, + handle_info/2, handle_pre_hibernate/1, prioritise_call/4, + prioritise_cast/3, prioritise_info/3, format_message_queue/2]). +%% Internal +-export([list_local/0, emit_info_local/3, deliver_reply_local/3]). +-export([get_vhost/1, get_user/1]). +%% For testing +-export([build_topic_variable_map/3]). + +-record(ch, { + %% starting | running | flow | closing + state, + %% same as reader's protocol. Used when instantiating + %% (protocol) exceptions. + protocol, + %% channel number + channel, + %% reader process + reader_pid, + %% writer process + writer_pid, + %% + conn_pid, + %% same as reader's name, see #v1.name + %% in rabbit_reader + conn_name, + %% limiter pid, see rabbit_limiter + limiter, + %% none | {Msgs, Acks} | committing | failed | + tx, + %% (consumer) delivery tag sequence + next_tag, + %% messages pending consumer acknowledgement + unacked_message_q, + %% same as #v1.user in the reader, used in + %% authorisation checks + user, + %% same as #v1.user in the reader + virtual_host, + %% when queue.bind's queue field is empty, + %% this name will be used instead + most_recently_declared_queue, + %% a map of queue pid to queue name + queue_names, + %% queue processes are monitored to update + %% queue names + queue_monitors, + %% a map of consumer tags to + %% consumer details: #amqqueue record, acknowledgement mode, + %% consumer exclusivity, etc + consumer_mapping, + %% a map of queue pids to consumer tag lists + queue_consumers, + %% a set of pids of queues that have unacknowledged + %% deliveries + delivering_queues, + %% when a queue is declared as exclusive, queue + %% collector must be notified. + %% see rabbit_queue_collector for more info. + queue_collector_pid, + %% timer used to emit statistics + stats_timer, + %% are publisher confirms enabled for this channel? + confirm_enabled, + %% publisher confirm delivery tag sequence + publish_seqno, + %% a dtree used to track unconfirmed + %% (to publishers) messages + unconfirmed, + %% a list of tags for published messages that were + %% delivered but are yet to be confirmed to the client + confirmed, + %% a dtree used to track oustanding notifications + %% for messages published as mandatory + mandatory, + %% same as capabilities in the reader + capabilities, + %% tracing exchange resource if tracing is enabled, + %% 'none' otherwise + trace_state, + consumer_prefetch, + %% used by "one shot RPC" (amq. + reply_consumer, + %% flow | noflow, see rabbitmq-server#114 + delivery_flow, + interceptor_state +}). + + +-define(MAX_PERMISSION_CACHE_SIZE, 12). + +-define(REFRESH_TIMEOUT, 15000). + +-define(STATISTICS_KEYS, + [reductions, + pid, + transactional, + confirm, + consumer_count, + messages_unacknowledged, + messages_unconfirmed, + messages_uncommitted, + acks_uncommitted, + prefetch_count, + global_prefetch_count, + state, + garbage_collection]). + +-define(CREATION_EVENT_KEYS, + [pid, + name, + connection, + number, + user, + vhost, + user_who_performed_action]). + +-define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]). + +-define(INCR_STATS(Incs, Measure, State), + case rabbit_event:stats_level(State, #ch.stats_timer) of + fine -> incr_stats(Incs, Measure); + _ -> ok + end). + +%%---------------------------------------------------------------------------- + +-export_type([channel_number/0]). + +-type channel_number() :: non_neg_integer(). + +-export_type([channel/0]). + +-type channel() :: #ch{}. + +-spec start_link + (channel_number(), pid(), pid(), pid(), string(), rabbit_types:protocol(), + rabbit_types:user(), rabbit_types:vhost(), rabbit_framing:amqp_table(), + pid(), pid()) -> + rabbit_types:ok_pid_or_error(). +-spec do(pid(), rabbit_framing:amqp_method_record()) -> 'ok'. +-spec do + (pid(), rabbit_framing:amqp_method_record(), + rabbit_types:maybe(rabbit_types:content())) -> + 'ok'. +-spec do_flow + (pid(), rabbit_framing:amqp_method_record(), + rabbit_types:maybe(rabbit_types:content())) -> + 'ok'. +-spec flush(pid()) -> 'ok'. +-spec shutdown(pid()) -> 'ok'. +-spec send_command(pid(), rabbit_framing:amqp_method_record()) -> 'ok'. +-spec deliver + (pid(), rabbit_types:ctag(), boolean(), rabbit_amqqueue:qmsg()) -> 'ok'. +-spec deliver_reply(binary(), rabbit_types:delivery()) -> 'ok'. +-spec deliver_reply_local(pid(), binary(), rabbit_types:delivery()) -> 'ok'. +-spec send_credit_reply(pid(), non_neg_integer()) -> 'ok'. +-spec send_drained(pid(), [{rabbit_types:ctag(), non_neg_integer()}]) -> 'ok'. +-spec list() -> [pid()]. +-spec list_local() -> [pid()]. +-spec info_keys() -> rabbit_types:info_keys(). +-spec info(pid()) -> rabbit_types:infos(). +-spec info(pid(), rabbit_types:info_keys()) -> rabbit_types:infos(). +-spec info_all() -> [rabbit_types:infos()]. +-spec info_all(rabbit_types:info_keys()) -> [rabbit_types:infos()]. +-spec refresh_config_local() -> 'ok'. +-spec ready_for_close(pid()) -> 'ok'. +-spec force_event_refresh(reference()) -> 'ok'. + +%%---------------------------------------------------------------------------- + +start_link(Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, + VHost, Capabilities, CollectorPid, Limiter) -> + gen_server2:start_link( + ?MODULE, [Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, + User, VHost, Capabilities, CollectorPid, Limiter], []). + +do(Pid, Method) -> + rabbit_channel_common:do(Pid, Method). + +do(Pid, Method, Content) -> + rabbit_channel_common:do(Pid, Method, Content). + +do_flow(Pid, Method, Content) -> + rabbit_channel_common:do_flow(Pid, Method, Content). + +flush(Pid) -> + gen_server2:call(Pid, flush, infinity). + +shutdown(Pid) -> + gen_server2:cast(Pid, terminate). + +send_command(Pid, Msg) -> + gen_server2:cast(Pid, {command, Msg}). + +deliver(Pid, ConsumerTag, AckRequired, Msg) -> + gen_server2:cast(Pid, {deliver, ConsumerTag, AckRequired, Msg}). + +deliver_reply(<<"amq.rabbitmq.reply-to.", Rest/binary>>, Delivery) -> + case decode_fast_reply_to(Rest) of + {ok, Pid, Key} -> + delegate:invoke_no_result( + Pid, {?MODULE, deliver_reply_local, [Key, Delivery]}); + error -> + ok + end. + +%% We want to ensure people can't use this mechanism to send a message +%% to an arbitrary process and kill it! +deliver_reply_local(Pid, Key, Delivery) -> + case pg_local:in_group(rabbit_channels, Pid) of + true -> gen_server2:cast(Pid, {deliver_reply, Key, Delivery}); + false -> ok + end. + +declare_fast_reply_to(<<"amq.rabbitmq.reply-to">>) -> + exists; +declare_fast_reply_to(<<"amq.rabbitmq.reply-to.", Rest/binary>>) -> + case decode_fast_reply_to(Rest) of + {ok, Pid, Key} -> + Msg = {declare_fast_reply_to, Key}, + rabbit_misc:with_exit_handler( + rabbit_misc:const(not_found), + fun() -> gen_server2:call(Pid, Msg, infinity) end); + error -> + not_found + end; +declare_fast_reply_to(_) -> + not_found. + +decode_fast_reply_to(Rest) -> + case string:tokens(binary_to_list(Rest), ".") of + [PidEnc, Key] -> Pid = binary_to_term(base64:decode(PidEnc)), + {ok, Pid, Key}; + _ -> error + end. + +send_credit_reply(Pid, Len) -> + gen_server2:cast(Pid, {send_credit_reply, Len}). + +send_drained(Pid, CTagCredit) -> + gen_server2:cast(Pid, {send_drained, CTagCredit}). + +list() -> + rabbit_misc:append_rpc_all_nodes(rabbit_mnesia:cluster_nodes(running), + rabbit_channel, list_local, []). + +list_local() -> + pg_local:get_members(rabbit_channels). + +info_keys() -> ?INFO_KEYS. + +info(Pid) -> + gen_server2:call(Pid, info, infinity). + +info(Pid, Items) -> + case gen_server2:call(Pid, {info, Items}, infinity) of + {ok, Res} -> Res; + {error, Error} -> throw(Error) + end. + +info_all() -> + rabbit_misc:filter_exit_map(fun (C) -> info(C) end, list()). + +info_all(Items) -> + rabbit_misc:filter_exit_map(fun (C) -> info(C, Items) end, list()). + +info_local(Items) -> + rabbit_misc:filter_exit_map(fun (C) -> info(C, Items) end, list_local()). + +emit_info_all(Nodes, Items, Ref, AggregatorPid) -> + Pids = [ spawn_link(Node, rabbit_channel, emit_info_local, [Items, Ref, AggregatorPid]) || Node <- Nodes ], + rabbit_control_misc:await_emitters_termination(Pids). + +emit_info_local(Items, Ref, AggregatorPid) -> + emit_info(list_local(), Items, Ref, AggregatorPid). + +emit_info(PidList, InfoItems, Ref, AggregatorPid) -> + rabbit_control_misc:emitting_map_with_exit_handler( + AggregatorPid, Ref, fun(C) -> info(C, InfoItems) end, PidList). + +refresh_config_local() -> + rabbit_misc:upmap( + fun (C) -> gen_server2:call(C, refresh_config, infinity) end, + list_local()), + ok. + +refresh_interceptors() -> + rabbit_misc:upmap( + fun (C) -> gen_server2:call(C, refresh_interceptors, ?REFRESH_TIMEOUT) end, + list_local()), + ok. + +ready_for_close(Pid) -> + rabbit_channel_common:ready_for_close(Pid). + +force_event_refresh(Ref) -> + [gen_server2:cast(C, {force_event_refresh, Ref}) || C <- list()], + ok. + +%%--------------------------------------------------------------------------- + +init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost, + Capabilities, CollectorPid, LimiterPid]) -> + process_flag(trap_exit, true), + ?store_proc_name({ConnName, Channel}), + ok = pg_local:join(rabbit_channels, self()), + Flow = case rabbit_misc:get_env(rabbit, mirroring_flow_control, true) of + true -> flow; + false -> noflow + end, + + State = #ch{state = starting, + protocol = Protocol, + channel = Channel, + reader_pid = ReaderPid, + writer_pid = WriterPid, + conn_pid = ConnPid, + conn_name = ConnName, + limiter = rabbit_limiter:new(LimiterPid), + tx = none, + next_tag = 1, + unacked_message_q = queue:new(), + user = User, + virtual_host = VHost, + most_recently_declared_queue = <<>>, + queue_names = #{}, + queue_monitors = pmon:new(), + consumer_mapping = #{}, + queue_consumers = #{}, + delivering_queues = sets:new(), + queue_collector_pid = CollectorPid, + confirm_enabled = false, + publish_seqno = 1, + unconfirmed = dtree:empty(), + confirmed = [], + mandatory = dtree:empty(), + capabilities = Capabilities, + trace_state = rabbit_trace:init(VHost), + consumer_prefetch = 0, + reply_consumer = none, + delivery_flow = Flow, + interceptor_state = undefined}, + State1 = State#ch{ + interceptor_state = rabbit_channel_interceptor:init(State)}, + State2 = rabbit_event:init_stats_timer(State1, #ch.stats_timer), + Infos = infos(?CREATION_EVENT_KEYS, State2), + rabbit_core_metrics:channel_created(self(), Infos), + rabbit_event:notify(channel_created, Infos), + rabbit_event:if_enabled(State2, #ch.stats_timer, + fun() -> emit_stats(State2) end), + put_operation_timeout(), + {ok, State2, hibernate, + {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. + +prioritise_call(Msg, _From, _Len, _State) -> + case Msg of + info -> 9; + {info, _Items} -> 9; + _ -> 0 + end. + +prioritise_cast(Msg, _Len, _State) -> + case Msg of + {confirm, _MsgSeqNos, _QPid} -> 5; + {mandatory_received, _MsgSeqNo, _QPid} -> 5; + _ -> 0 + end. + +prioritise_info(Msg, _Len, _State) -> + case Msg of + emit_stats -> 7; + _ -> 0 + end. + +handle_call(flush, _From, State) -> + reply(ok, State); + +handle_call(info, _From, State) -> + reply(infos(?INFO_KEYS, State), State); + +handle_call({info, Items}, _From, State) -> + try + reply({ok, infos(Items, State)}, State) + catch Error -> reply({error, Error}, State) + end; + +handle_call(refresh_config, _From, State = #ch{virtual_host = VHost}) -> + reply(ok, State#ch{trace_state = rabbit_trace:init(VHost)}); + +handle_call(refresh_interceptors, _From, State) -> + IState = rabbit_channel_interceptor:init(State), + reply(ok, State#ch{interceptor_state = IState}); + +handle_call({declare_fast_reply_to, Key}, _From, + State = #ch{reply_consumer = Consumer}) -> + reply(case Consumer of + {_, _, Key} -> exists; + _ -> not_found + end, State); + +handle_call(_Request, _From, State) -> + noreply(State). + +handle_cast({method, Method, Content, Flow}, + State = #ch{reader_pid = Reader, + interceptor_state = IState}) -> + case Flow of + %% We are going to process a message from the rabbit_reader + %% process, so here we ack it. In this case we are accessing + %% the rabbit_channel process dictionary. + flow -> credit_flow:ack(Reader); + noflow -> ok + end, + + try handle_method(rabbit_channel_interceptor:intercept_in( + expand_shortcuts(Method, State), Content, IState), + State) of + {reply, Reply, NewState} -> + ok = send(Reply, NewState), + noreply(NewState); + {noreply, NewState} -> + noreply(NewState); + stop -> + {stop, normal, State} + catch + exit:Reason = #amqp_error{} -> + MethodName = rabbit_misc:method_record_type(Method), + handle_exception(Reason#amqp_error{method = MethodName}, State); + _:Reason -> + {stop, {Reason, erlang:get_stacktrace()}, State} + end; + +handle_cast(ready_for_close, State = #ch{state = closing, + writer_pid = WriterPid}) -> + ok = rabbit_writer:send_command_sync(WriterPid, #'channel.close_ok'{}), + {stop, normal, State}; + +handle_cast(terminate, State = #ch{writer_pid = WriterPid}) -> + ok = rabbit_writer:flush(WriterPid), + {stop, normal, State}; + +handle_cast({command, #'basic.consume_ok'{consumer_tag = CTag} = Msg}, State) -> + ok = send(Msg, State), + noreply(consumer_monitor(CTag, State)); + +handle_cast({command, Msg}, State) -> + ok = send(Msg, State), + noreply(State); + +handle_cast({deliver, _CTag, _AckReq, _Msg}, State = #ch{state = closing}) -> + noreply(State); +handle_cast({deliver, ConsumerTag, AckRequired, + Msg = {_QName, QPid, _MsgId, Redelivered, + #basic_message{exchange_name = ExchangeName, + routing_keys = [RoutingKey | _CcRoutes], + content = Content}}}, + State = #ch{writer_pid = WriterPid, + next_tag = DeliveryTag}) -> + ok = rabbit_writer:send_command_and_notify( + WriterPid, QPid, self(), + #'basic.deliver'{consumer_tag = ConsumerTag, + delivery_tag = DeliveryTag, + redelivered = Redelivered, + exchange = ExchangeName#resource.name, + routing_key = RoutingKey}, + Content), + rabbit_basic:maybe_gc_large_msg(Content), + noreply(record_sent(ConsumerTag, AckRequired, Msg, State)); + +handle_cast({deliver_reply, _K, _Del}, State = #ch{state = closing}) -> + noreply(State); +handle_cast({deliver_reply, _K, _Del}, State = #ch{reply_consumer = none}) -> + noreply(State); +handle_cast({deliver_reply, Key, #delivery{message = + #basic_message{exchange_name = ExchangeName, + routing_keys = [RoutingKey | _CcRoutes], + content = Content}}}, + State = #ch{writer_pid = WriterPid, + next_tag = DeliveryTag, + reply_consumer = {ConsumerTag, _Suffix, Key}}) -> + ok = rabbit_writer:send_command( + WriterPid, + #'basic.deliver'{consumer_tag = ConsumerTag, + delivery_tag = DeliveryTag, + redelivered = false, + exchange = ExchangeName#resource.name, + routing_key = RoutingKey}, + Content), + noreply(State); +handle_cast({deliver_reply, _K1, _}, State=#ch{reply_consumer = {_, _, _K2}}) -> + noreply(State); + +handle_cast({send_credit_reply, Len}, State = #ch{writer_pid = WriterPid}) -> + ok = rabbit_writer:send_command( + WriterPid, #'basic.credit_ok'{available = Len}), + noreply(State); + +handle_cast({send_drained, CTagCredit}, State = #ch{writer_pid = WriterPid}) -> + [ok = rabbit_writer:send_command( + WriterPid, #'basic.credit_drained'{consumer_tag = ConsumerTag, + credit_drained = CreditDrained}) + || {ConsumerTag, CreditDrained} <- CTagCredit], + noreply(State); + +handle_cast({force_event_refresh, Ref}, State) -> + rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State), + Ref), + noreply(rabbit_event:init_stats_timer(State, #ch.stats_timer)); + +handle_cast({mandatory_received, MsgSeqNo}, State = #ch{mandatory = Mand}) -> + %% NB: don't call noreply/1 since we don't want to send confirms. + noreply_coalesce(State#ch{mandatory = dtree:drop(MsgSeqNo, Mand)}); + +handle_cast({confirm, MsgSeqNos, QPid}, State = #ch{unconfirmed = UC}) -> + {MXs, UC1} = dtree:take(MsgSeqNos, QPid, UC), + %% NB: don't call noreply/1 since we don't want to send confirms. + noreply_coalesce(record_confirms(MXs, State#ch{unconfirmed = UC1})). + +handle_info({bump_credit, Msg}, State) -> + %% A rabbit_amqqueue_process is granting credit to our channel. If + %% our channel was being blocked by this process, and no other + %% process is blocking our channel, then this channel will be + %% unblocked. This means that any credit that was deferred will be + %% sent to rabbit_reader processs that might be blocked by this + %% particular channel. + credit_flow:handle_bump_msg(Msg), + noreply(State); + +handle_info(timeout, State) -> + noreply(State); + +handle_info(emit_stats, State) -> + emit_stats(State), + State1 = rabbit_event:reset_stats_timer(State, #ch.stats_timer), + %% NB: don't call noreply/1 since we don't want to kick off the + %% stats timer. + {noreply, send_confirms(State1), hibernate}; + +handle_info({'DOWN', _MRef, process, QPid, Reason}, State) -> + State1 = handle_publishing_queue_down(QPid, Reason, State), + State3 = handle_consuming_queue_down(QPid, State1), + State4 = handle_delivering_queue_down(QPid, State3), + %% A rabbit_amqqueue_process has died. If our channel was being + %% blocked by this process, and no other process is blocking our + %% channel, then this channel will be unblocked. This means that + %% any credit that was deferred will be sent to the rabbit_reader + %% processs that might be blocked by this particular channel. + credit_flow:peer_down(QPid), + #ch{queue_names = QNames, queue_monitors = QMons} = State4, + case maps:find(QPid, QNames) of + {ok, QName} -> erase_queue_stats(QName); + error -> ok + end, + noreply(State4#ch{queue_names = maps:remove(QPid, QNames), + queue_monitors = pmon:erase(QPid, QMons)}); + +handle_info({'EXIT', _Pid, Reason}, State) -> + {stop, Reason, State}; + +handle_info({{Ref, Node}, LateAnswer}, State = #ch{channel = Channel}) + when is_reference(Ref) -> + rabbit_log_channel:warning("Channel ~p ignoring late answer ~p from ~p", + [Channel, LateAnswer, Node]), + noreply(State). + +handle_pre_hibernate(State) -> + ok = clear_permission_cache(), + rabbit_event:if_enabled( + State, #ch.stats_timer, + fun () -> emit_stats(State, + [{idle_since, + os:system_time(milli_seconds)}]) + end), + {hibernate, rabbit_event:stop_stats_timer(State, #ch.stats_timer)}. + +terminate(_Reason, State = #ch{user = #user{username = Username}}) -> + {_Res, _State1} = notify_queues(State), + pg_local:leave(rabbit_channels, self()), + rabbit_event:if_enabled(State, #ch.stats_timer, + fun() -> emit_stats(State) end), + [delete_stats(Tag) || {Tag, _} <- get()], + rabbit_core_metrics:channel_closed(self()), + rabbit_event:notify(channel_closed, [{pid, self()}, + {user_who_performed_action, Username}]). + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +format_message_queue(Opt, MQ) -> rabbit_misc:format_message_queue(Opt, MQ). + +%%--------------------------------------------------------------------------- + +reply(Reply, NewState) -> {reply, Reply, next_state(NewState), hibernate}. + +noreply(NewState) -> {noreply, next_state(NewState), hibernate}. + +next_state(State) -> ensure_stats_timer(send_confirms(State)). + +noreply_coalesce(State = #ch{confirmed = C}) -> + Timeout = case C of [] -> hibernate; _ -> 0 end, + {noreply, ensure_stats_timer(State), Timeout}. + +ensure_stats_timer(State) -> + rabbit_event:ensure_stats_timer(State, #ch.stats_timer, emit_stats). + +return_ok(State, true, _Msg) -> {noreply, State}; +return_ok(State, false, Msg) -> {reply, Msg, State}. + +ok_msg(true, _Msg) -> undefined; +ok_msg(false, Msg) -> Msg. + +send(_Command, #ch{state = closing}) -> + ok; +send(Command, #ch{writer_pid = WriterPid}) -> + ok = rabbit_writer:send_command(WriterPid, Command). + +format_soft_error(#amqp_error{name = N, explanation = E, method = M}) -> + io_lib:format("operation ~s caused a channel exception ~s: ~ts", [M, N, E]); +format_soft_error(Reason) -> + Reason. + +handle_exception(Reason, State = #ch{protocol = Protocol, + channel = Channel, + writer_pid = WriterPid, + reader_pid = ReaderPid, + conn_pid = ConnPid, + conn_name = ConnName, + virtual_host = VHost, + user = User}) -> + %% something bad's happened: notify_queues may not be 'ok' + {_Result, State1} = notify_queues(State), + case rabbit_binary_generator:map_exception(Channel, Reason, Protocol) of + {Channel, CloseMethod} -> + rabbit_log_channel:error( + "Channel error on connection ~p (~s, vhost: '~s'," + " user: '~s'), channel ~p:~n~s~n", + [ConnPid, ConnName, VHost, User#user.username, + Channel, format_soft_error(Reason)]), + ok = rabbit_writer:send_command(WriterPid, CloseMethod), + {noreply, State1}; + {0, _} -> + ReaderPid ! {channel_exit, Channel, Reason}, + {stop, normal, State1} + end. + +-spec precondition_failed(string()) -> no_return(). + +precondition_failed(Format) -> precondition_failed(Format, []). + +-spec precondition_failed(string(), [any()]) -> no_return(). + +precondition_failed(Format, Params) -> + rabbit_misc:protocol_error(precondition_failed, Format, Params). + +return_queue_declare_ok(#resource{name = ActualName}, + NoWait, MessageCount, ConsumerCount, State) -> + return_ok(State#ch{most_recently_declared_queue = ActualName}, NoWait, + #'queue.declare_ok'{queue = ActualName, + message_count = MessageCount, + consumer_count = ConsumerCount}). + +check_resource_access(User, Resource, Perm) -> + V = {Resource, Perm}, + Cache = case get(permission_cache) of + undefined -> []; + Other -> Other + end, + case lists:member(V, Cache) of + true -> ok; + false -> ok = rabbit_access_control:check_resource_access( + User, Resource, Perm), + CacheTail = lists:sublist(Cache, ?MAX_PERMISSION_CACHE_SIZE-1), + put(permission_cache, [V | CacheTail]) + end. + +clear_permission_cache() -> erase(permission_cache), + erase(topic_permission_cache), + ok. + +check_configure_permitted(Resource, #ch{user = User}) -> + check_resource_access(User, Resource, configure). + +check_write_permitted(Resource, #ch{user = User}) -> + check_resource_access(User, Resource, write). + +check_read_permitted(Resource, #ch{user = User}) -> + check_resource_access(User, Resource, read). + +check_write_permitted_on_topic(Resource, Channel, RoutingKey) -> + check_topic_authorisation(Resource, Channel, RoutingKey, write). + +check_read_permitted_on_topic(Resource, Channel, RoutingKey) -> + check_topic_authorisation(Resource, Channel, RoutingKey, read). + +check_user_id_header(#'P_basic'{user_id = undefined}, _) -> + ok; +check_user_id_header(#'P_basic'{user_id = Username}, + #ch{user = #user{username = Username}}) -> + ok; +check_user_id_header( + #'P_basic'{}, #ch{user = #user{authz_backends = + [{rabbit_auth_backend_dummy, _}]}}) -> + ok; +check_user_id_header(#'P_basic'{user_id = Claimed}, + #ch{user = #user{username = Actual, + tags = Tags}}) -> + case lists:member(impersonator, Tags) of + true -> ok; + false -> precondition_failed( + "user_id property set to '~s' but authenticated user was " + "'~s'", [Claimed, Actual]) + end. + +check_expiration_header(Props) -> + case rabbit_basic:parse_expiration(Props) of + {ok, _} -> ok; + {error, E} -> precondition_failed("invalid expiration '~s': ~p", + [Props#'P_basic'.expiration, E]) + end. + +check_internal_exchange(#exchange{name = Name, internal = true}) -> + rabbit_misc:protocol_error(access_refused, + "cannot publish to internal ~s", + [rabbit_misc:rs(Name)]); +check_internal_exchange(_) -> + ok. + +check_topic_authorisation(#exchange{name = Name = #resource{virtual_host = VHost}, type = topic}, + #ch{user = User = #user{username = Username}, conn_pid = ConnPid}, + RoutingKey, + Permission) -> + Resource = Name#resource{kind = topic}, + Timeout = get_operation_timeout(), + AmqpParams = rabbit_amqp_connection:amqp_params(ConnPid, Timeout), + VariableMap = build_topic_variable_map(AmqpParams, VHost, Username), + Context = #{routing_key => RoutingKey, + variable_map => VariableMap + }, + Cache = case get(topic_permission_cache) of + undefined -> []; + Other -> Other + end, + case lists:member({Resource, Context, Permission}, Cache) of + true -> ok; + false -> ok = rabbit_access_control:check_topic_access( + User, Resource, Permission, Context), + CacheTail = lists:sublist(Cache, ?MAX_PERMISSION_CACHE_SIZE-1), + put(topic_permission_cache, [{Resource, Context, Permission} | CacheTail]) + end; +check_topic_authorisation(_, _, _, _) -> + ok. + +build_topic_variable_map(AmqpParams, VHost, Username) -> + VariableFromAmqpParams = extract_topic_variable_map_from_amqp_params(AmqpParams), + maps:merge(VariableFromAmqpParams, #{<<"vhost">> => VHost, <<"username">> => Username}). + +%% use tuple representation of amqp_params to avoid coupling. +%% get variable map only from amqp_params_direct, not amqp_params_network. +%% amqp_params_direct are usually used from plugins (e.g. MQTT, STOMP) +extract_topic_variable_map_from_amqp_params([{amqp_params, {amqp_params_direct, _, _, _, _, + {amqp_adapter_info, _,_,_,_,_,_,AdditionalInfo}, _}}]) -> + proplists:get_value(variable_map, AdditionalInfo, #{}); +extract_topic_variable_map_from_amqp_params(_) -> + #{}. + +check_msg_size(Content) -> + Size = rabbit_basic:maybe_gc_large_msg(Content), + case Size > ?MAX_MSG_SIZE of + true -> precondition_failed("message size ~B larger than max size ~B", + [Size, ?MAX_MSG_SIZE]); + false -> ok + end. + +check_vhost_queue_limit(#resource{name = QueueName}, VHost) -> + case rabbit_vhost_limit:is_over_queue_limit(VHost) of + false -> ok; + {true, Limit} -> precondition_failed("cannot declare queue '~s': " + "queue limit in vhost '~s' (~p) is reached", + [QueueName, VHost, Limit]) + + end. + +qbin_to_resource(QueueNameBin, State) -> + name_to_resource(queue, QueueNameBin, State). + +name_to_resource(Type, NameBin, #ch{virtual_host = VHostPath}) -> + rabbit_misc:r(VHostPath, Type, NameBin). + +expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = <<>>}) -> + rabbit_misc:protocol_error(not_found, "no previously declared queue", []); +expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = MRDQ}) -> + MRDQ; +expand_queue_name_shortcut(QueueNameBin, _) -> + QueueNameBin. + +expand_routing_key_shortcut(<<>>, <<>>, + #ch{most_recently_declared_queue = <<>>}) -> + rabbit_misc:protocol_error(not_found, "no previously declared queue", []); +expand_routing_key_shortcut(<<>>, <<>>, + #ch{most_recently_declared_queue = MRDQ}) -> + MRDQ; +expand_routing_key_shortcut(_QueueNameBin, RoutingKey, _State) -> + RoutingKey. + +expand_shortcuts(#'basic.get' {queue = Q} = M, State) -> + M#'basic.get' {queue = expand_queue_name_shortcut(Q, State)}; +expand_shortcuts(#'basic.consume'{queue = Q} = M, State) -> + M#'basic.consume'{queue = expand_queue_name_shortcut(Q, State)}; +expand_shortcuts(#'queue.delete' {queue = Q} = M, State) -> + M#'queue.delete' {queue = expand_queue_name_shortcut(Q, State)}; +expand_shortcuts(#'queue.purge' {queue = Q} = M, State) -> + M#'queue.purge' {queue = expand_queue_name_shortcut(Q, State)}; +expand_shortcuts(#'queue.bind' {queue = Q, routing_key = K} = M, State) -> + M#'queue.bind' {queue = expand_queue_name_shortcut(Q, State), + routing_key = expand_routing_key_shortcut(Q, K, State)}; +expand_shortcuts(#'queue.unbind' {queue = Q, routing_key = K} = M, State) -> + M#'queue.unbind' {queue = expand_queue_name_shortcut(Q, State), + routing_key = expand_routing_key_shortcut(Q, K, State)}; +expand_shortcuts(M, _State) -> + M. + +check_not_default_exchange(#resource{kind = exchange, name = <<"">>}) -> + rabbit_misc:protocol_error( + access_refused, "operation not permitted on the default exchange", []); +check_not_default_exchange(_) -> + ok. + +check_exchange_deletion(XName = #resource{name = <<"amq.", _/binary>>, + kind = exchange}) -> + rabbit_misc:protocol_error( + access_refused, "deletion of system ~s not allowed", + [rabbit_misc:rs(XName)]); +check_exchange_deletion(_) -> + ok. + +%% check that an exchange/queue name does not contain the reserved +%% "amq." prefix. +%% +%% As per the AMQP 0-9-1 spec, the exclusion of "amq." prefixed names +%% only applies on actual creation, and not in the cases where the +%% entity already exists or passive=true. +%% +%% NB: We deliberately do not enforce the other constraints on names +%% required by the spec. +check_name(Kind, NameBin = <<"amq.", _/binary>>) -> + rabbit_misc:protocol_error( + access_refused, + "~s name '~s' contains reserved prefix 'amq.*'",[Kind, NameBin]); +check_name(_Kind, NameBin) -> + NameBin. + +strip_cr_lf(NameBin) -> + binary:replace(NameBin, [<<"\n">>, <<"\r">>], <<"">>, [global]). + + +maybe_set_fast_reply_to( + C = #content{properties = P = #'P_basic'{reply_to = + <<"amq.rabbitmq.reply-to">>}}, + #ch{reply_consumer = ReplyConsumer}) -> + case ReplyConsumer of + none -> rabbit_misc:protocol_error( + precondition_failed, + "fast reply consumer does not exist", []); + {_, Suf, _K} -> Rep = <<"amq.rabbitmq.reply-to.", Suf/binary>>, + rabbit_binary_generator:clear_encoded_content( + C#content{properties = P#'P_basic'{reply_to = Rep}}) + end; +maybe_set_fast_reply_to(C, _State) -> + C. + +record_confirms([], State) -> + State; +record_confirms(MXs, State = #ch{confirmed = C}) -> + State#ch{confirmed = [MXs | C]}. + +handle_method({Method, Content}, State) -> + handle_method(Method, Content, State). + +handle_method(#'channel.open'{}, _, State = #ch{state = starting}) -> + %% Don't leave "starting" as the state for 5s. TODO is this TRTTD? + State1 = State#ch{state = running}, + rabbit_event:if_enabled(State1, #ch.stats_timer, + fun() -> emit_stats(State1) end), + {reply, #'channel.open_ok'{}, State1}; + +handle_method(#'channel.open'{}, _, _State) -> + rabbit_misc:protocol_error( + channel_error, "second 'channel.open' seen", []); + +handle_method(_Method, _, #ch{state = starting}) -> + rabbit_misc:protocol_error(channel_error, "expected 'channel.open'", []); + +handle_method(#'channel.close_ok'{}, _, #ch{state = closing}) -> + stop; + +handle_method(#'channel.close'{}, _, State = #ch{writer_pid = WriterPid, + state = closing}) -> + ok = rabbit_writer:send_command(WriterPid, #'channel.close_ok'{}), + {noreply, State}; + +handle_method(_Method, _, State = #ch{state = closing}) -> + {noreply, State}; + +handle_method(#'channel.close'{}, _, State = #ch{reader_pid = ReaderPid}) -> + {_Result, State1} = notify_queues(State), + %% We issue the channel.close_ok response after a handshake with + %% the reader, the other half of which is ready_for_close. That + %% way the reader forgets about the channel before we send the + %% response (and this channel process terminates). If we didn't do + %% that, a channel.open for the same channel number, which a + %% client is entitled to send as soon as it has received the + %% close_ok, might be received by the reader before it has seen + %% the termination and hence be sent to the old, now dead/dying + %% channel process, instead of a new process, and thus lost. + ReaderPid ! {channel_closing, self()}, + {noreply, State1}; + +%% Even though the spec prohibits the client from sending commands +%% while waiting for the reply to a synchronous command, we generally +%% do allow this...except in the case of a pending tx.commit, where +%% it could wreak havoc. +handle_method(_Method, _, #ch{tx = Tx}) + when Tx =:= committing orelse Tx =:= failed -> + rabbit_misc:protocol_error( + channel_error, "unexpected command while processing 'tx.commit'", []); + +handle_method(#'access.request'{},_, State) -> + {reply, #'access.request_ok'{ticket = 1}, State}; + +handle_method(#'basic.publish'{immediate = true}, _Content, _State) -> + rabbit_misc:protocol_error(not_implemented, "immediate=true", []); + +handle_method(#'basic.publish'{exchange = ExchangeNameBin, + routing_key = RoutingKey, + mandatory = Mandatory}, + Content, State = #ch{virtual_host = VHostPath, + tx = Tx, + channel = ChannelNum, + confirm_enabled = ConfirmEnabled, + trace_state = TraceState, + user = #user{username = Username}, + conn_name = ConnName, + delivery_flow = Flow}) -> + check_msg_size(Content), + ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin), + check_write_permitted(ExchangeName, State), + Exchange = rabbit_exchange:lookup_or_die(ExchangeName), + check_internal_exchange(Exchange), + check_write_permitted_on_topic(Exchange, State, RoutingKey), + %% We decode the content's properties here because we're almost + %% certain to want to look at delivery-mode and priority. + DecodedContent = #content {properties = Props} = + maybe_set_fast_reply_to( + rabbit_binary_parser:ensure_content_decoded(Content), State), + check_user_id_header(Props, State), + check_expiration_header(Props), + DoConfirm = Tx =/= none orelse ConfirmEnabled, + {MsgSeqNo, State1} = + case DoConfirm orelse Mandatory of + false -> {undefined, State}; + true -> SeqNo = State#ch.publish_seqno, + {SeqNo, State#ch{publish_seqno = SeqNo + 1}} + end, + case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of + {ok, Message} -> + Delivery = rabbit_basic:delivery( + Mandatory, DoConfirm, Message, MsgSeqNo), + QNames = rabbit_exchange:route(Exchange, Delivery), + rabbit_trace:tap_in(Message, QNames, ConnName, ChannelNum, + Username, TraceState), + DQ = {Delivery#delivery{flow = Flow}, QNames}, + {noreply, case Tx of + none -> deliver_to_queues(DQ, State1); + {Msgs, Acks} -> Msgs1 = queue:in(DQ, Msgs), + State1#ch{tx = {Msgs1, Acks}} + end}; + {error, Reason} -> + precondition_failed("invalid message: ~p", [Reason]) + end; + +handle_method(#'basic.nack'{delivery_tag = DeliveryTag, + multiple = Multiple, + requeue = Requeue}, _, State) -> + reject(DeliveryTag, Requeue, Multiple, State); + +handle_method(#'basic.ack'{delivery_tag = DeliveryTag, + multiple = Multiple}, + _, State = #ch{unacked_message_q = UAMQ, tx = Tx}) -> + {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, Multiple), + State1 = State#ch{unacked_message_q = Remaining}, + {noreply, case Tx of + none -> ack(Acked, State1), + State1; + {Msgs, Acks} -> Acks1 = ack_cons(ack, Acked, Acks), + State1#ch{tx = {Msgs, Acks1}} + end}; + +handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck}, + _, State = #ch{writer_pid = WriterPid, + conn_pid = ConnPid, + limiter = Limiter, + next_tag = DeliveryTag}) -> + QueueName = qbin_to_resource(QueueNameBin, State), + check_read_permitted(QueueName, State), + case rabbit_amqqueue:with_exclusive_access_or_die( + QueueName, ConnPid, + fun (Q) -> rabbit_amqqueue:basic_get( + Q, self(), NoAck, rabbit_limiter:pid(Limiter)) + end) of + {ok, MessageCount, + Msg = {QName, QPid, _MsgId, Redelivered, + #basic_message{exchange_name = ExchangeName, + routing_keys = [RoutingKey | _CcRoutes], + content = Content}}} -> + ok = rabbit_writer:send_command( + WriterPid, + #'basic.get_ok'{delivery_tag = DeliveryTag, + redelivered = Redelivered, + exchange = ExchangeName#resource.name, + routing_key = RoutingKey, + message_count = MessageCount}, + Content), + State1 = monitor_delivering_queue(NoAck, QPid, QName, State), + {noreply, record_sent(none, not(NoAck), Msg, State1)}; + empty -> + {reply, #'basic.get_empty'{}, State} + end; + +handle_method(#'basic.consume'{queue = <<"amq.rabbitmq.reply-to">>, + consumer_tag = CTag0, + no_ack = NoAck, + nowait = NoWait}, + _, State = #ch{reply_consumer = ReplyConsumer, + consumer_mapping = ConsumerMapping}) -> + case maps:find(CTag0, ConsumerMapping) of + error -> + case {ReplyConsumer, NoAck} of + {none, true} -> + CTag = case CTag0 of + <<>> -> rabbit_guid:binary( + rabbit_guid:gen_secure(), "amq.ctag"); + Other -> Other + end, + %% Precalculate both suffix and key; base64 encoding is + %% expensive + Key = base64:encode(rabbit_guid:gen_secure()), + PidEnc = base64:encode(term_to_binary(self())), + Suffix = <<PidEnc/binary, ".", Key/binary>>, + Consumer = {CTag, Suffix, binary_to_list(Key)}, + State1 = State#ch{reply_consumer = Consumer}, + case NoWait of + true -> {noreply, State1}; + false -> Rep = #'basic.consume_ok'{consumer_tag = CTag}, + {reply, Rep, State1} + end; + {_, false} -> + rabbit_misc:protocol_error( + precondition_failed, + "reply consumer cannot acknowledge", []); + _ -> + rabbit_misc:protocol_error( + precondition_failed, "reply consumer already set", []) + end; + {ok, _} -> + %% Attempted reuse of consumer tag. + rabbit_misc:protocol_error( + not_allowed, "attempt to reuse consumer tag '~s'", [CTag0]) + end; + +handle_method(#'basic.cancel'{consumer_tag = ConsumerTag, nowait = NoWait}, + _, State = #ch{reply_consumer = {ConsumerTag, _, _}}) -> + State1 = State#ch{reply_consumer = none}, + case NoWait of + true -> {noreply, State1}; + false -> Rep = #'basic.cancel_ok'{consumer_tag = ConsumerTag}, + {reply, Rep, State1} + end; + +handle_method(#'basic.consume'{queue = QueueNameBin, + consumer_tag = ConsumerTag, + no_local = _, % FIXME: implement + no_ack = NoAck, + exclusive = ExclusiveConsume, + nowait = NoWait, + arguments = Args}, + _, State = #ch{consumer_prefetch = ConsumerPrefetch, + consumer_mapping = ConsumerMapping}) -> + case maps:find(ConsumerTag, ConsumerMapping) of + error -> + QueueName = qbin_to_resource(QueueNameBin, State), + check_read_permitted(QueueName, State), + ActualConsumerTag = + case ConsumerTag of + <<>> -> rabbit_guid:binary(rabbit_guid:gen_secure(), + "amq.ctag"); + Other -> Other + end, + case basic_consume( + QueueName, NoAck, ConsumerPrefetch, ActualConsumerTag, + ExclusiveConsume, Args, NoWait, State) of + {ok, State1} -> + {noreply, State1}; + {error, exclusive_consume_unavailable} -> + rabbit_misc:protocol_error( + access_refused, "~s in exclusive use", + [rabbit_misc:rs(QueueName)]) + end; + {ok, _} -> + %% Attempted reuse of consumer tag. + rabbit_misc:protocol_error( + not_allowed, "attempt to reuse consumer tag '~s'", [ConsumerTag]) + end; + +handle_method(#'basic.cancel'{consumer_tag = ConsumerTag, nowait = NoWait}, + _, State = #ch{consumer_mapping = ConsumerMapping, + queue_consumers = QCons, + user = #user{username = Username}}) -> + OkMsg = #'basic.cancel_ok'{consumer_tag = ConsumerTag}, + case maps:find(ConsumerTag, ConsumerMapping) of + error -> + %% Spec requires we ignore this situation. + return_ok(State, NoWait, OkMsg); + {ok, {Q = #amqqueue{pid = QPid}, _CParams}} -> + ConsumerMapping1 = maps:remove(ConsumerTag, ConsumerMapping), + QCons1 = + case maps:find(QPid, QCons) of + error -> QCons; + {ok, CTags} -> CTags1 = gb_sets:delete(ConsumerTag, CTags), + case gb_sets:is_empty(CTags1) of + true -> maps:remove(QPid, QCons); + false -> maps:put(QPid, CTags1, QCons) + end + end, + NewState = State#ch{consumer_mapping = ConsumerMapping1, + queue_consumers = QCons1}, + %% In order to ensure that no more messages are sent to + %% the consumer after the cancel_ok has been sent, we get + %% the queue process to send the cancel_ok on our + %% behalf. If we were sending the cancel_ok ourselves it + %% might overtake a message sent previously by the queue. + case rabbit_misc:with_exit_handler( + fun () -> {error, not_found} end, + fun () -> + rabbit_amqqueue:basic_cancel( + Q, self(), ConsumerTag, ok_msg(NoWait, OkMsg), + Username) + end) of + ok -> + {noreply, NewState}; + {error, not_found} -> + %% Spec requires we ignore this situation. + return_ok(NewState, NoWait, OkMsg) + end + end; + +handle_method(#'basic.qos'{prefetch_size = Size}, _, _State) when Size /= 0 -> + rabbit_misc:protocol_error(not_implemented, + "prefetch_size!=0 (~w)", [Size]); + +handle_method(#'basic.qos'{global = false, + prefetch_count = PrefetchCount}, _, State) -> + {reply, #'basic.qos_ok'{}, State#ch{consumer_prefetch = PrefetchCount}}; + +handle_method(#'basic.qos'{global = true, + prefetch_count = 0}, + _, State = #ch{limiter = Limiter}) -> + Limiter1 = rabbit_limiter:unlimit_prefetch(Limiter), + {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}}; + +handle_method(#'basic.qos'{global = true, + prefetch_count = PrefetchCount}, + _, State = #ch{limiter = Limiter, unacked_message_q = UAMQ}) -> + %% TODO queue:len(UAMQ) is not strictly right since that counts + %% unacked messages from basic.get too. Pretty obscure though. + Limiter1 = rabbit_limiter:limit_prefetch(Limiter, + PrefetchCount, queue:len(UAMQ)), + case ((not rabbit_limiter:is_active(Limiter)) andalso + rabbit_limiter:is_active(Limiter1)) of + true -> rabbit_amqqueue:activate_limit_all( + consumer_queues(State#ch.consumer_mapping), self()); + false -> ok + end, + {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}}; + +handle_method(#'basic.recover_async'{requeue = true}, + _, State = #ch{unacked_message_q = UAMQ, limiter = Limiter}) -> + OkFun = fun () -> ok end, + UAMQL = queue:to_list(UAMQ), + foreach_per_queue( + fun (QPid, MsgIds) -> + rabbit_misc:with_exit_handler( + OkFun, + fun () -> rabbit_amqqueue:requeue(QPid, MsgIds, self()) end) + end, lists:reverse(UAMQL)), + ok = notify_limiter(Limiter, UAMQL), + %% No answer required - basic.recover is the newer, synchronous + %% variant of this method + {noreply, State#ch{unacked_message_q = queue:new()}}; + +handle_method(#'basic.recover_async'{requeue = false}, _, _State) -> + rabbit_misc:protocol_error(not_implemented, "requeue=false", []); + +handle_method(#'basic.recover'{requeue = Requeue}, Content, State) -> + {noreply, State1} = handle_method(#'basic.recover_async'{requeue = Requeue}, + Content, State), + {reply, #'basic.recover_ok'{}, State1}; + +handle_method(#'basic.reject'{delivery_tag = DeliveryTag, requeue = Requeue}, + _, State) -> + reject(DeliveryTag, Requeue, false, State); + +handle_method(#'exchange.declare'{exchange = ExchangeNameBin, + type = TypeNameBin, + passive = false, + durable = Durable, + auto_delete = AutoDelete, + internal = Internal, + nowait = NoWait, + arguments = Args}, + _, State = #ch{virtual_host = VHostPath, + user = #user{username = Username}}) -> + CheckedType = rabbit_exchange:check_type(TypeNameBin), + ExchangeName = rabbit_misc:r(VHostPath, exchange, strip_cr_lf(ExchangeNameBin)), + check_not_default_exchange(ExchangeName), + check_configure_permitted(ExchangeName, State), + X = case rabbit_exchange:lookup(ExchangeName) of + {ok, FoundX} -> FoundX; + {error, not_found} -> + check_name('exchange', strip_cr_lf(ExchangeNameBin)), + AeKey = <<"alternate-exchange">>, + case rabbit_misc:r_arg(VHostPath, exchange, Args, AeKey) of + undefined -> ok; + {error, {invalid_type, Type}} -> + precondition_failed( + "invalid type '~s' for arg '~s' in ~s", + [Type, AeKey, rabbit_misc:rs(ExchangeName)]); + AName -> check_read_permitted(ExchangeName, State), + check_write_permitted(AName, State), + ok + end, + rabbit_exchange:declare(ExchangeName, + CheckedType, + Durable, + AutoDelete, + Internal, + Args, + Username) + end, + ok = rabbit_exchange:assert_equivalence(X, CheckedType, Durable, + AutoDelete, Internal, Args), + return_ok(State, NoWait, #'exchange.declare_ok'{}); + +handle_method(#'exchange.declare'{exchange = ExchangeNameBin, + passive = true, + nowait = NoWait}, + _, State = #ch{virtual_host = VHostPath}) -> + ExchangeName = rabbit_misc:r(VHostPath, exchange, strip_cr_lf(ExchangeNameBin)), + check_not_default_exchange(ExchangeName), + _ = rabbit_exchange:lookup_or_die(ExchangeName), + return_ok(State, NoWait, #'exchange.declare_ok'{}); + +handle_method(#'exchange.delete'{exchange = ExchangeNameBin, + if_unused = IfUnused, + nowait = NoWait}, + _, State = #ch{virtual_host = VHostPath, + user = #user{username = Username}}) -> + StrippedExchangeNameBin = strip_cr_lf(ExchangeNameBin), + ExchangeName = rabbit_misc:r(VHostPath, exchange, StrippedExchangeNameBin), + check_not_default_exchange(ExchangeName), + check_exchange_deletion(ExchangeName), + check_configure_permitted(ExchangeName, State), + case rabbit_exchange:delete(ExchangeName, IfUnused, Username) of + {error, not_found} -> + return_ok(State, NoWait, #'exchange.delete_ok'{}); + {error, in_use} -> + precondition_failed("~s in use", [rabbit_misc:rs(ExchangeName)]); + ok -> + return_ok(State, NoWait, #'exchange.delete_ok'{}) + end; + +handle_method(#'exchange.bind'{destination = DestinationNameBin, + source = SourceNameBin, + routing_key = RoutingKey, + nowait = NoWait, + arguments = Arguments}, _, State) -> + binding_action(fun rabbit_binding:add/3, + strip_cr_lf(SourceNameBin), exchange, strip_cr_lf(DestinationNameBin), RoutingKey, + Arguments, #'exchange.bind_ok'{}, NoWait, State); + +handle_method(#'exchange.unbind'{destination = DestinationNameBin, + source = SourceNameBin, + routing_key = RoutingKey, + nowait = NoWait, + arguments = Arguments}, _, State) -> + binding_action(fun rabbit_binding:remove/3, + strip_cr_lf(SourceNameBin), exchange, strip_cr_lf(DestinationNameBin), RoutingKey, + Arguments, #'exchange.unbind_ok'{}, NoWait, State); + +%% Note that all declares to these are effectively passive. If it +%% exists it by definition has one consumer. +handle_method(#'queue.declare'{queue = <<"amq.rabbitmq.reply-to", + _/binary>> = QueueNameBin, + nowait = NoWait}, _, + State = #ch{virtual_host = VHost}) -> + StrippedQueueNameBin = strip_cr_lf(QueueNameBin), + QueueName = rabbit_misc:r(VHost, queue, StrippedQueueNameBin), + case declare_fast_reply_to(StrippedQueueNameBin) of + exists -> return_queue_declare_ok(QueueName, NoWait, 0, 1, State); + not_found -> rabbit_misc:not_found(QueueName) + end; + +handle_method(#'queue.declare'{queue = QueueNameBin, + passive = false, + durable = DurableDeclare, + exclusive = ExclusiveDeclare, + auto_delete = AutoDelete, + nowait = NoWait, + arguments = Args} = Declare, + _, State = #ch{virtual_host = VHostPath, + conn_pid = ConnPid, + queue_collector_pid = CollectorPid, + user = #user{username = Username}}) -> + Owner = case ExclusiveDeclare of + true -> ConnPid; + false -> none + end, + StrippedQueueNameBin = strip_cr_lf(QueueNameBin), + Durable = DurableDeclare andalso not ExclusiveDeclare, + ActualNameBin = case StrippedQueueNameBin of + <<>> -> rabbit_guid:binary(rabbit_guid:gen_secure(), + "amq.gen"); + Other -> check_name('queue', Other) + end, + QueueName = rabbit_misc:r(VHostPath, queue, ActualNameBin), + check_configure_permitted(QueueName, State), + case rabbit_amqqueue:with( + QueueName, + fun (Q) -> ok = rabbit_amqqueue:assert_equivalence( + Q, Durable, AutoDelete, Args, Owner), + maybe_stat(NoWait, Q) + end) of + {ok, MessageCount, ConsumerCount} -> + return_queue_declare_ok(QueueName, NoWait, MessageCount, + ConsumerCount, State); + {error, not_found} -> + %% enforce the limit for newly declared queues only + check_vhost_queue_limit(QueueName, VHostPath), + DlxKey = <<"x-dead-letter-exchange">>, + case rabbit_misc:r_arg(VHostPath, exchange, Args, DlxKey) of + undefined -> + ok; + {error, {invalid_type, Type}} -> + precondition_failed( + "invalid type '~s' for arg '~s' in ~s", + [Type, DlxKey, rabbit_misc:rs(QueueName)]); + DLX -> + check_read_permitted(QueueName, State), + check_write_permitted(DLX, State), + ok + end, + case rabbit_amqqueue:declare(QueueName, Durable, AutoDelete, + Args, Owner, Username) of + {new, #amqqueue{pid = QPid}} -> + %% We need to notify the reader within the channel + %% process so that we can be sure there are no + %% outstanding exclusive queues being declared as + %% the connection shuts down. + ok = case Owner of + none -> ok; + _ -> rabbit_queue_collector:register( + CollectorPid, QPid) + end, + return_queue_declare_ok(QueueName, NoWait, 0, 0, State); + {existing, _Q} -> + %% must have been created between the stat and the + %% declare. Loop around again. + handle_method(Declare, none, State); + {absent, Q, Reason} -> + rabbit_misc:absent(Q, Reason); + {owner_died, _Q} -> + %% Presumably our own days are numbered since the + %% connection has died. Pretend the queue exists though, + %% just so nothing fails. + return_queue_declare_ok(QueueName, NoWait, 0, 0, State) + end; + {error, {absent, Q, Reason}} -> + rabbit_misc:absent(Q, Reason) + end; + +handle_method(#'queue.declare'{queue = QueueNameBin, + passive = true, + nowait = NoWait}, + _, State = #ch{virtual_host = VHostPath, + conn_pid = ConnPid}) -> + StrippedQueueNameBin = strip_cr_lf(QueueNameBin), + QueueName = rabbit_misc:r(VHostPath, queue, StrippedQueueNameBin), + {{ok, MessageCount, ConsumerCount}, #amqqueue{} = Q} = + rabbit_amqqueue:with_or_die( + QueueName, fun (Q) -> {maybe_stat(NoWait, Q), Q} end), + ok = rabbit_amqqueue:check_exclusive_access(Q, ConnPid), + return_queue_declare_ok(QueueName, NoWait, MessageCount, ConsumerCount, + State); + +handle_method(#'queue.delete'{queue = QueueNameBin, + if_unused = IfUnused, + if_empty = IfEmpty, + nowait = NoWait}, + _, State = #ch{conn_pid = ConnPid, + user = #user{username = Username}}) -> + StrippedQueueNameBin = strip_cr_lf(QueueNameBin), + QueueName = qbin_to_resource(StrippedQueueNameBin, State), + check_configure_permitted(QueueName, State), + case rabbit_amqqueue:with( + QueueName, + fun (Q) -> + rabbit_amqqueue:check_exclusive_access(Q, ConnPid), + rabbit_amqqueue:delete(Q, IfUnused, IfEmpty, Username) + end, + fun (not_found) -> {ok, 0}; + ({absent, Q, crashed}) -> rabbit_amqqueue:delete_crashed(Q, Username), + {ok, 0}; + ({absent, Q, Reason}) -> rabbit_misc:absent(Q, Reason) + end) of + {error, in_use} -> + precondition_failed("~s in use", [rabbit_misc:rs(QueueName)]); + {error, not_empty} -> + precondition_failed("~s not empty", [rabbit_misc:rs(QueueName)]); + {ok, PurgedMessageCount} -> + return_ok(State, NoWait, + #'queue.delete_ok'{message_count = PurgedMessageCount}) + end; + +handle_method(#'queue.bind'{queue = QueueNameBin, + exchange = ExchangeNameBin, + routing_key = RoutingKey, + nowait = NoWait, + arguments = Arguments}, _, State) -> + binding_action(fun rabbit_binding:add/3, + strip_cr_lf(ExchangeNameBin), queue, strip_cr_lf(QueueNameBin), RoutingKey, Arguments, + #'queue.bind_ok'{}, NoWait, State); + +handle_method(#'queue.unbind'{queue = QueueNameBin, + exchange = ExchangeNameBin, + routing_key = RoutingKey, + arguments = Arguments}, _, State) -> + binding_action(fun rabbit_binding:remove/3, + strip_cr_lf(ExchangeNameBin), queue, strip_cr_lf(QueueNameBin), RoutingKey, Arguments, + #'queue.unbind_ok'{}, false, State); + +handle_method(#'queue.purge'{queue = QueueNameBin, nowait = NoWait}, + _, State = #ch{conn_pid = ConnPid}) -> + QueueName = qbin_to_resource(QueueNameBin, State), + check_read_permitted(QueueName, State), + {ok, PurgedMessageCount} = rabbit_amqqueue:with_exclusive_access_or_die( + QueueName, ConnPid, + fun (Q) -> rabbit_amqqueue:purge(Q) end), + return_ok(State, NoWait, + #'queue.purge_ok'{message_count = PurgedMessageCount}); + +handle_method(#'tx.select'{}, _, #ch{confirm_enabled = true}) -> + precondition_failed("cannot switch from confirm to tx mode"); + +handle_method(#'tx.select'{}, _, State = #ch{tx = none}) -> + {reply, #'tx.select_ok'{}, State#ch{tx = new_tx()}}; + +handle_method(#'tx.select'{}, _, State) -> + {reply, #'tx.select_ok'{}, State}; + +handle_method(#'tx.commit'{}, _, #ch{tx = none}) -> + precondition_failed("channel is not transactional"); + +handle_method(#'tx.commit'{}, _, State = #ch{tx = {Msgs, Acks}, + limiter = Limiter}) -> + State1 = rabbit_misc:queue_fold(fun deliver_to_queues/2, State, Msgs), + Rev = fun (X) -> lists:reverse(lists:sort(X)) end, + lists:foreach(fun ({ack, A}) -> ack(Rev(A), State1); + ({Requeue, A}) -> reject(Requeue, Rev(A), Limiter) + end, lists:reverse(Acks)), + {noreply, maybe_complete_tx(State1#ch{tx = committing})}; + +handle_method(#'tx.rollback'{}, _, #ch{tx = none}) -> + precondition_failed("channel is not transactional"); + +handle_method(#'tx.rollback'{}, _, State = #ch{unacked_message_q = UAMQ, + tx = {_Msgs, Acks}}) -> + AcksL = lists:append(lists:reverse([lists:reverse(L) || {_, L} <- Acks])), + UAMQ1 = queue:from_list(lists:usort(AcksL ++ queue:to_list(UAMQ))), + {reply, #'tx.rollback_ok'{}, State#ch{unacked_message_q = UAMQ1, + tx = new_tx()}}; + +handle_method(#'confirm.select'{}, _, #ch{tx = {_, _}}) -> + precondition_failed("cannot switch from tx to confirm mode"); + +handle_method(#'confirm.select'{nowait = NoWait}, _, State) -> + return_ok(State#ch{confirm_enabled = true}, + NoWait, #'confirm.select_ok'{}); + +handle_method(#'channel.flow'{active = true}, _, State) -> + {reply, #'channel.flow_ok'{active = true}, State}; + +handle_method(#'channel.flow'{active = false}, _, _State) -> + rabbit_misc:protocol_error(not_implemented, "active=false", []); + +handle_method(#'basic.credit'{consumer_tag = CTag, + credit = Credit, + drain = Drain}, + _, State = #ch{consumer_mapping = Consumers}) -> + case maps:find(CTag, Consumers) of + {ok, {Q, _CParams}} -> ok = rabbit_amqqueue:credit( + Q, self(), CTag, Credit, Drain), + {noreply, State}; + error -> precondition_failed( + "unknown consumer tag '~s'", [CTag]) + end; + +handle_method(_MethodRecord, _Content, _State) -> + rabbit_misc:protocol_error( + command_invalid, "unimplemented method", []). + +%%---------------------------------------------------------------------------- + +%% We get the queue process to send the consume_ok on our behalf. This +%% is for symmetry with basic.cancel - see the comment in that method +%% for why. +basic_consume(QueueName, NoAck, ConsumerPrefetch, ActualConsumerTag, + ExclusiveConsume, Args, NoWait, + State = #ch{conn_pid = ConnPid, + limiter = Limiter, + consumer_mapping = ConsumerMapping, + user = #user{username = Username}}) -> + case rabbit_amqqueue:with_exclusive_access_or_die( + QueueName, ConnPid, + fun (Q) -> + {rabbit_amqqueue:basic_consume( + Q, NoAck, self(), + rabbit_limiter:pid(Limiter), + rabbit_limiter:is_active(Limiter), + ConsumerPrefetch, ActualConsumerTag, + ExclusiveConsume, Args, + ok_msg(NoWait, #'basic.consume_ok'{ + consumer_tag = ActualConsumerTag}), + Username), + Q} + end) of + {ok, Q = #amqqueue{pid = QPid, name = QName}} -> + CM1 = maps:put( + ActualConsumerTag, + {Q, {NoAck, ConsumerPrefetch, ExclusiveConsume, Args}}, + ConsumerMapping), + State1 = monitor_delivering_queue( + NoAck, QPid, QName, + State#ch{consumer_mapping = CM1}), + {ok, case NoWait of + true -> consumer_monitor(ActualConsumerTag, State1); + false -> State1 + end}; + {{error, exclusive_consume_unavailable} = E, _Q} -> + E + end. + +maybe_stat(false, Q) -> rabbit_amqqueue:stat(Q); +maybe_stat(true, _Q) -> {ok, 0, 0}. + +consumer_monitor(ConsumerTag, + State = #ch{consumer_mapping = ConsumerMapping, + queue_monitors = QMons, + queue_consumers = QCons}) -> + {#amqqueue{pid = QPid}, _CParams} = + maps:get(ConsumerTag, ConsumerMapping), + CTags1 = case maps:find(QPid, QCons) of + {ok, CTags} -> gb_sets:insert(ConsumerTag, CTags); + error -> gb_sets:singleton(ConsumerTag) + end, + QCons1 = maps:put(QPid, CTags1, QCons), + State#ch{queue_monitors = pmon:monitor(QPid, QMons), + queue_consumers = QCons1}. + +monitor_delivering_queue(NoAck, QPid, QName, + State = #ch{queue_names = QNames, + queue_monitors = QMons, + delivering_queues = DQ}) -> + State#ch{queue_names = maps:put(QPid, QName, QNames), + queue_monitors = pmon:monitor(QPid, QMons), + delivering_queues = case NoAck of + true -> DQ; + false -> sets:add_element(QPid, DQ) + end}. + +handle_publishing_queue_down(QPid, Reason, State = #ch{unconfirmed = UC, + mandatory = Mand}) -> + {MMsgs, Mand1} = dtree:take(QPid, Mand), + [basic_return(Msg, State, no_route) || {_, Msg} <- MMsgs], + State1 = State#ch{mandatory = Mand1}, + case rabbit_misc:is_abnormal_exit(Reason) of + true -> {MXs, UC1} = dtree:take_all(QPid, UC), + send_nacks(MXs, State1#ch{unconfirmed = UC1}); + false -> {MXs, UC1} = dtree:take(QPid, UC), + record_confirms(MXs, State1#ch{unconfirmed = UC1}) + + end. + +handle_consuming_queue_down(QPid, State = #ch{queue_consumers = QCons, + queue_names = QNames}) -> + ConsumerTags = case maps:find(QPid, QCons) of + error -> gb_sets:new(); + {ok, CTags} -> CTags + end, + gb_sets:fold( + fun (CTag, StateN = #ch{consumer_mapping = CMap}) -> + QName = maps:get(QPid, QNames), + case queue_down_consumer_action(CTag, CMap) of + remove -> + cancel_consumer(CTag, QName, StateN); + {recover, {NoAck, ConsumerPrefetch, Exclusive, Args}} -> + case catch basic_consume( %% [0] + QName, NoAck, ConsumerPrefetch, CTag, + Exclusive, Args, true, StateN) of + {ok, StateN1} -> StateN1; + _ -> cancel_consumer(CTag, QName, StateN) + end + end + end, State#ch{queue_consumers = maps:remove(QPid, QCons)}, ConsumerTags). + +%% [0] There is a slight danger here that if a queue is deleted and +%% then recreated again the reconsume will succeed even though it was +%% not an HA failover. But the likelihood is not great and most users +%% are unlikely to care. + +cancel_consumer(CTag, QName, State = #ch{capabilities = Capabilities, + consumer_mapping = CMap}) -> + case rabbit_misc:table_lookup( + Capabilities, <<"consumer_cancel_notify">>) of + {bool, true} -> ok = send(#'basic.cancel'{consumer_tag = CTag, + nowait = true}, State); + _ -> ok + end, + rabbit_event:notify(consumer_deleted, [{consumer_tag, CTag}, + {channel, self()}, + {queue, QName}]), + State#ch{consumer_mapping = maps:remove(CTag, CMap)}. + +queue_down_consumer_action(CTag, CMap) -> + {_, {_, _, _, Args} = ConsumeSpec} = maps:get(CTag, CMap), + case rabbit_misc:table_lookup(Args, <<"x-cancel-on-ha-failover">>) of + {bool, true} -> remove; + _ -> {recover, ConsumeSpec} + end. + +handle_delivering_queue_down(QPid, State = #ch{delivering_queues = DQ}) -> + State#ch{delivering_queues = sets:del_element(QPid, DQ)}. + +binding_action(Fun, ExchangeNameBin, DestinationType, DestinationNameBin, + RoutingKey, Arguments, ReturnMethod, NoWait, + State = #ch{virtual_host = VHostPath, + conn_pid = ConnPid, + user = #user{username = Username}}) -> + DestinationName = name_to_resource(DestinationType, DestinationNameBin, State), + check_write_permitted(DestinationName, State), + ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin), + [check_not_default_exchange(N) || N <- [DestinationName, ExchangeName]], + check_read_permitted(ExchangeName, State), + ExchangeLookup = rabbit_exchange:lookup(ExchangeName), + case ExchangeLookup of + {error, not_found} -> + %% no-op + ExchangeLookup; + {ok, Exchange} -> + check_read_permitted_on_topic(Exchange, State, RoutingKey), + ExchangeLookup + end, + case Fun(#binding{source = ExchangeName, + destination = DestinationName, + key = RoutingKey, + args = Arguments}, + fun (_X, Q = #amqqueue{}) -> + try rabbit_amqqueue:check_exclusive_access(Q, ConnPid) + catch exit:Reason -> {error, Reason} + end; + (_X, #exchange{}) -> + ok + end, + Username) of + {error, {resources_missing, [{not_found, Name} | _]}} -> + rabbit_misc:not_found(Name); + {error, {resources_missing, [{absent, Q, Reason} | _]}} -> + rabbit_misc:absent(Q, Reason); + {error, binding_not_found} -> + rabbit_misc:protocol_error( + not_found, "no binding ~s between ~s and ~s", + [RoutingKey, rabbit_misc:rs(ExchangeName), + rabbit_misc:rs(DestinationName)]); + {error, {binding_invalid, Fmt, Args}} -> + rabbit_misc:protocol_error(precondition_failed, Fmt, Args); + {error, #amqp_error{} = Error} -> + rabbit_misc:protocol_error(Error); + ok -> return_ok(State, NoWait, ReturnMethod) + end. + +basic_return(#basic_message{exchange_name = ExchangeName, + routing_keys = [RoutingKey | _CcRoutes], + content = Content}, + State = #ch{protocol = Protocol, writer_pid = WriterPid}, + Reason) -> + ?INCR_STATS([{exchange_stats, ExchangeName, 1}], return_unroutable, State), + {_Close, ReplyCode, ReplyText} = Protocol:lookup_amqp_exception(Reason), + ok = rabbit_writer:send_command( + WriterPid, + #'basic.return'{reply_code = ReplyCode, + reply_text = ReplyText, + exchange = ExchangeName#resource.name, + routing_key = RoutingKey}, + Content). + +reject(DeliveryTag, Requeue, Multiple, + State = #ch{unacked_message_q = UAMQ, tx = Tx}) -> + {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, Multiple), + State1 = State#ch{unacked_message_q = Remaining}, + {noreply, case Tx of + none -> reject(Requeue, Acked, State1#ch.limiter), + State1; + {Msgs, Acks} -> Acks1 = ack_cons(Requeue, Acked, Acks), + State1#ch{tx = {Msgs, Acks1}} + end}. + +%% NB: Acked is in youngest-first order +reject(Requeue, Acked, Limiter) -> + foreach_per_queue( + fun (QPid, MsgIds) -> + rabbit_amqqueue:reject(QPid, Requeue, MsgIds, self()) + end, Acked), + ok = notify_limiter(Limiter, Acked). + +record_sent(ConsumerTag, AckRequired, + Msg = {QName, QPid, MsgId, Redelivered, _Message}, + State = #ch{unacked_message_q = UAMQ, + next_tag = DeliveryTag, + trace_state = TraceState, + user = #user{username = Username}, + conn_name = ConnName, + channel = ChannelNum}) -> + ?INCR_STATS([{queue_stats, QName, 1}], case {ConsumerTag, AckRequired} of + {none, true} -> get; + {none, false} -> get_no_ack; + {_ , true} -> deliver; + {_ , false} -> deliver_no_ack + end, State), + case Redelivered of + true -> ?INCR_STATS([{queue_stats, QName, 1}], redeliver, State); + false -> ok + end, + rabbit_trace:tap_out(Msg, ConnName, ChannelNum, Username, TraceState), + UAMQ1 = case AckRequired of + true -> queue:in({DeliveryTag, ConsumerTag, {QPid, MsgId}}, + UAMQ); + false -> UAMQ + end, + State#ch{unacked_message_q = UAMQ1, next_tag = DeliveryTag + 1}. + +%% NB: returns acks in youngest-first order +collect_acks(Q, 0, true) -> + {lists:reverse(queue:to_list(Q)), queue:new()}; +collect_acks(Q, DeliveryTag, Multiple) -> + collect_acks([], [], Q, DeliveryTag, Multiple). + +collect_acks(ToAcc, PrefixAcc, Q, DeliveryTag, Multiple) -> + case queue:out(Q) of + {{value, UnackedMsg = {CurrentDeliveryTag, _ConsumerTag, _Msg}}, + QTail} -> + if CurrentDeliveryTag == DeliveryTag -> + {[UnackedMsg | ToAcc], + case PrefixAcc of + [] -> QTail; + _ -> queue:join( + queue:from_list(lists:reverse(PrefixAcc)), + QTail) + end}; + Multiple -> + collect_acks([UnackedMsg | ToAcc], PrefixAcc, + QTail, DeliveryTag, Multiple); + true -> + collect_acks(ToAcc, [UnackedMsg | PrefixAcc], + QTail, DeliveryTag, Multiple) + end; + {empty, _} -> + precondition_failed("unknown delivery tag ~w", [DeliveryTag]) + end. + +%% NB: Acked is in youngest-first order +ack(Acked, State = #ch{queue_names = QNames}) -> + foreach_per_queue( + fun (QPid, MsgIds) -> + ok = rabbit_amqqueue:ack(QPid, MsgIds, self()), + ?INCR_STATS(case maps:find(QPid, QNames) of + {ok, QName} -> Count = length(MsgIds), + [{queue_stats, QName, Count}]; + error -> [] + end, ack, State) + end, Acked), + ok = notify_limiter(State#ch.limiter, Acked). + +%% {Msgs, Acks} +%% +%% Msgs is a queue. +%% +%% Acks looks s.t. like this: +%% [{false,[5,4]},{true,[3]},{ack,[2,1]}, ...] +%% +%% Each element is a pair consisting of a tag and a list of +%% ack'ed/reject'ed msg ids. The tag is one of 'ack' (to ack), 'true' +%% (reject w requeue), 'false' (reject w/o requeue). The msg ids, as +%% well as the list overall, are in "most-recent (generally youngest) +%% ack first" order. +new_tx() -> {queue:new(), []}. + +notify_queues(State = #ch{state = closing}) -> + {ok, State}; +notify_queues(State = #ch{consumer_mapping = Consumers, + delivering_queues = DQ }) -> + QPids = sets:to_list( + sets:union(sets:from_list(consumer_queues(Consumers)), DQ)), + Timeout = get_operation_timeout(), + {rabbit_amqqueue:notify_down_all(QPids, self(), Timeout), + State#ch{state = closing}}. + +foreach_per_queue(_F, []) -> + ok; +foreach_per_queue(F, [{_DTag, _CTag, {QPid, MsgId}}]) -> %% common case + F(QPid, [MsgId]); +%% NB: UAL should be in youngest-first order; the tree values will +%% then be in oldest-first order +foreach_per_queue(F, UAL) -> + T = lists:foldl(fun ({_DTag, _CTag, {QPid, MsgId}}, T) -> + rabbit_misc:gb_trees_cons(QPid, MsgId, T) + end, gb_trees:empty(), UAL), + rabbit_misc:gb_trees_foreach(F, T). + +consumer_queues(Consumers) -> + lists:usort([QPid || {_Key, {#amqqueue{pid = QPid}, _CParams}} + <- maps:to_list(Consumers)]). + +%% tell the limiter about the number of acks that have been received +%% for messages delivered to subscribed consumers, but not acks for +%% messages sent in a response to a basic.get (identified by their +%% 'none' consumer tag) +notify_limiter(Limiter, Acked) -> + %% optimisation: avoid the potentially expensive 'foldl' in the + %% common case. + case rabbit_limiter:is_active(Limiter) of + false -> ok; + true -> case lists:foldl(fun ({_, none, _}, Acc) -> Acc; + ({_, _, _}, Acc) -> Acc + 1 + end, 0, Acked) of + 0 -> ok; + Count -> rabbit_limiter:ack(Limiter, Count) + end + end. + +deliver_to_queues({#delivery{message = #basic_message{exchange_name = XName}, + confirm = false, + mandatory = false}, + []}, State) -> %% optimisation + ?INCR_STATS([{exchange_stats, XName, 1}], publish, State), + State; +deliver_to_queues({Delivery = #delivery{message = Message = #basic_message{ + exchange_name = XName}, + mandatory = Mandatory, + confirm = Confirm, + msg_seq_no = MsgSeqNo}, + DelQNames}, State = #ch{queue_names = QNames, + queue_monitors = QMons}) -> + Qs = rabbit_amqqueue:lookup(DelQNames), + DeliveredQPids = rabbit_amqqueue:deliver(Qs, Delivery), + %% The pmon:monitor_all/2 monitors all queues to which we + %% delivered. But we want to monitor even queues we didn't deliver + %% to, since we need their 'DOWN' messages to clean + %% queue_names. So we also need to monitor each QPid from + %% queues. But that only gets the masters (which is fine for + %% cleaning queue_names), so we need the union of both. + %% + %% ...and we need to add even non-delivered queues to queue_names + %% since alternative algorithms to update queue_names less + %% frequently would in fact be more expensive in the common case. + {QNames1, QMons1} = + lists:foldl(fun (#amqqueue{pid = QPid, name = QName}, + {QNames0, QMons0}) -> + {case maps:is_key(QPid, QNames0) of + true -> QNames0; + false -> maps:put(QPid, QName, QNames0) + end, pmon:monitor(QPid, QMons0)} + end, {QNames, pmon:monitor_all(DeliveredQPids, QMons)}, Qs), + State1 = State#ch{queue_names = QNames1, + queue_monitors = QMons1}, + %% NB: the order here is important since basic.returns must be + %% sent before confirms. + State2 = process_routing_mandatory(Mandatory, DeliveredQPids, MsgSeqNo, + Message, State1), + State3 = process_routing_confirm( Confirm, DeliveredQPids, MsgSeqNo, + XName, State2), + ?INCR_STATS([{exchange_stats, XName, 1} | + [{queue_exchange_stats, {QName, XName}, 1} || + QPid <- DeliveredQPids, + {ok, QName} <- [maps:find(QPid, QNames1)]]], + publish, State3), + State3. + +process_routing_mandatory(false, _, _MsgSeqNo, _Msg, State) -> + State; +process_routing_mandatory(true, [], _MsgSeqNo, Msg, State) -> + ok = basic_return(Msg, State, no_route), + State; +process_routing_mandatory(true, QPids, MsgSeqNo, Msg, State) -> + State#ch{mandatory = dtree:insert(MsgSeqNo, QPids, Msg, + State#ch.mandatory)}. + +process_routing_confirm(false, _, _MsgSeqNo, _XName, State) -> + State; +process_routing_confirm(true, [], MsgSeqNo, XName, State) -> + record_confirms([{MsgSeqNo, XName}], State); +process_routing_confirm(true, QPids, MsgSeqNo, XName, State) -> + State#ch{unconfirmed = dtree:insert(MsgSeqNo, QPids, XName, + State#ch.unconfirmed)}. + +send_nacks([], State) -> + State; +send_nacks(_MXs, State = #ch{state = closing, + tx = none}) -> %% optimisation + State; +send_nacks(MXs, State = #ch{tx = none}) -> + coalesce_and_send([MsgSeqNo || {MsgSeqNo, _} <- MXs], + fun(MsgSeqNo, Multiple) -> + #'basic.nack'{delivery_tag = MsgSeqNo, + multiple = Multiple} + end, State); +send_nacks(_MXs, State = #ch{state = closing}) -> %% optimisation + State#ch{tx = failed}; +send_nacks(_, State) -> + maybe_complete_tx(State#ch{tx = failed}). + +send_confirms(State = #ch{tx = none, confirmed = []}) -> + State; +send_confirms(State = #ch{tx = none, confirmed = C}) -> + case rabbit_node_monitor:pause_partition_guard() of + ok -> MsgSeqNos = + lists:foldl( + fun ({MsgSeqNo, XName}, MSNs) -> + ?INCR_STATS([{exchange_stats, XName, 1}], + confirm, State), + [MsgSeqNo | MSNs] + end, [], lists:append(C)), + send_confirms(MsgSeqNos, State#ch{confirmed = []}); + pausing -> State + end; +send_confirms(State) -> + case rabbit_node_monitor:pause_partition_guard() of + ok -> maybe_complete_tx(State); + pausing -> State + end. + +send_confirms([], State) -> + State; +send_confirms(_Cs, State = #ch{state = closing}) -> %% optimisation + State; +send_confirms([MsgSeqNo], State) -> + ok = send(#'basic.ack'{delivery_tag = MsgSeqNo}, State), + State; +send_confirms(Cs, State) -> + coalesce_and_send(Cs, fun(MsgSeqNo, Multiple) -> + #'basic.ack'{delivery_tag = MsgSeqNo, + multiple = Multiple} + end, State). + +coalesce_and_send(MsgSeqNos, MkMsgFun, State = #ch{unconfirmed = UC}) -> + SMsgSeqNos = lists:usort(MsgSeqNos), + CutOff = case dtree:is_empty(UC) of + true -> lists:last(SMsgSeqNos) + 1; + false -> {SeqNo, _XName} = dtree:smallest(UC), SeqNo + end, + {Ms, Ss} = lists:splitwith(fun(X) -> X < CutOff end, SMsgSeqNos), + case Ms of + [] -> ok; + _ -> ok = send(MkMsgFun(lists:last(Ms), true), State) + end, + [ok = send(MkMsgFun(SeqNo, false), State) || SeqNo <- Ss], + State. + +ack_cons(Tag, Acked, [{Tag, Acks} | L]) -> [{Tag, Acked ++ Acks} | L]; +ack_cons(Tag, Acked, Acks) -> [{Tag, Acked} | Acks]. + +ack_len(Acks) -> lists:sum([length(L) || {ack, L} <- Acks]). + +maybe_complete_tx(State = #ch{tx = {_, _}}) -> + State; +maybe_complete_tx(State = #ch{unconfirmed = UC}) -> + case dtree:is_empty(UC) of + false -> State; + true -> complete_tx(State#ch{confirmed = []}) + end. + +complete_tx(State = #ch{tx = committing}) -> + ok = send(#'tx.commit_ok'{}, State), + State#ch{tx = new_tx()}; +complete_tx(State = #ch{tx = failed}) -> + {noreply, State1} = handle_exception( + rabbit_misc:amqp_error( + precondition_failed, "partial tx completion", [], + 'tx.commit'), + State), + State1#ch{tx = new_tx()}. + +infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items]. + +i(pid, _) -> self(); +i(connection, #ch{conn_pid = ConnPid}) -> ConnPid; +i(number, #ch{channel = Channel}) -> Channel; +i(user, #ch{user = User}) -> User#user.username; +i(user_who_performed_action, Ch) -> i(user, Ch); +i(vhost, #ch{virtual_host = VHost}) -> VHost; +i(transactional, #ch{tx = Tx}) -> Tx =/= none; +i(confirm, #ch{confirm_enabled = CE}) -> CE; +i(name, State) -> name(State); +i(consumer_count, #ch{consumer_mapping = CM}) -> maps:size(CM); +i(messages_unconfirmed, #ch{unconfirmed = UC}) -> dtree:size(UC); +i(messages_unacknowledged, #ch{unacked_message_q = UAMQ}) -> queue:len(UAMQ); +i(messages_uncommitted, #ch{tx = {Msgs, _Acks}}) -> queue:len(Msgs); +i(messages_uncommitted, #ch{}) -> 0; +i(acks_uncommitted, #ch{tx = {_Msgs, Acks}}) -> ack_len(Acks); +i(acks_uncommitted, #ch{}) -> 0; +i(state, #ch{state = running}) -> credit_flow:state(); +i(state, #ch{state = State}) -> State; +i(prefetch_count, #ch{consumer_prefetch = C}) -> C; +i(global_prefetch_count, #ch{limiter = Limiter}) -> + rabbit_limiter:get_prefetch_limit(Limiter); +i(interceptors, #ch{interceptor_state = IState}) -> + IState; +i(garbage_collection, _State) -> + rabbit_misc:get_gc_info(self()); +i(reductions, _State) -> + {reductions, Reductions} = erlang:process_info(self(), reductions), + Reductions; +i(Item, _) -> + throw({bad_argument, Item}). + +name(#ch{conn_name = ConnName, channel = Channel}) -> + list_to_binary(rabbit_misc:format("~s (~p)", [ConnName, Channel])). + +incr_stats(Incs, Measure) -> + [begin + rabbit_core_metrics:channel_stats(Type, Measure, {self(), Key}, Inc), + %% Keys in the process dictionary are used to clean up the core metrics + put({Type, Key}, none) + end || {Type, Key, Inc} <- Incs]. + +emit_stats(State) -> emit_stats(State, []). + +emit_stats(State, Extra) -> + [{reductions, Red} | Coarse0] = infos(?STATISTICS_KEYS, State), + rabbit_core_metrics:channel_stats(self(), Extra ++ Coarse0), + rabbit_core_metrics:channel_stats(reductions, self(), Red). + +erase_queue_stats(QName) -> + rabbit_core_metrics:channel_queue_down({self(), QName}), + erase({queue_stats, QName}), + [begin + rabbit_core_metrics:channel_queue_exchange_down({self(), QX}), + erase({queue_exchange_stats, QX}) + end || {{queue_exchange_stats, QX = {QName0, _}}, _} <- get(), + QName0 =:= QName]. + +get_vhost(#ch{virtual_host = VHost}) -> VHost. + +get_user(#ch{user = User}) -> User. + +delete_stats({queue_stats, QName}) -> + rabbit_core_metrics:channel_queue_down({self(), QName}); +delete_stats({exchange_stats, XName}) -> + rabbit_core_metrics:channel_exchange_down({self(), XName}); +delete_stats({queue_exchange_stats, QX}) -> + rabbit_core_metrics:channel_queue_exchange_down({self(), QX}); +delete_stats(_) -> + ok. + +put_operation_timeout() -> + put(channel_operation_timeout, ?CHANNEL_OPERATION_TIMEOUT). + +get_operation_timeout() -> + get(channel_operation_timeout). diff --git a/src/rabbit_channel_interceptor.erl b/src/rabbit_channel_interceptor.erl new file mode 100644 index 0000000000..f01c7feb6f --- /dev/null +++ b/src/rabbit_channel_interceptor.erl @@ -0,0 +1,113 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_channel_interceptor). + +-include("rabbit_framing.hrl"). +-include("rabbit.hrl"). + +-export([init/1, intercept_in/3]). + +-behaviour(rabbit_registry_class). + +-export([added_to_rabbit_registry/2, removed_from_rabbit_registry/1]). + +-type(method_name() :: rabbit_framing:amqp_method_name()). +-type(original_method() :: rabbit_framing:amqp_method_record()). +-type(processed_method() :: rabbit_framing:amqp_method_record()). +-type(original_content() :: rabbit_types:maybe(rabbit_types:content())). +-type(processed_content() :: rabbit_types:maybe(rabbit_types:content())). +-type(interceptor_state() :: term()). + +-callback description() -> [proplists:property()]. +%% Derive some initial state from the channel. This will be passed back +%% as the third argument of intercept/3. +-callback init(rabbit_channel:channel()) -> interceptor_state(). +-callback intercept(original_method(), original_content(), + interceptor_state()) -> + {processed_method(), processed_content()} | + rabbit_misc:channel_or_connection_exit(). +-callback applies_to() -> list(method_name()). + +added_to_rabbit_registry(_Type, _ModuleName) -> + rabbit_channel:refresh_interceptors(). +removed_from_rabbit_registry(_Type) -> + rabbit_channel:refresh_interceptors(). + +init(Ch) -> + Mods = [M || {_, M} <- rabbit_registry:lookup_all(channel_interceptor)], + check_no_overlap(Mods), + [{Mod, Mod:init(Ch)} || Mod <- Mods]. + +check_no_overlap(Mods) -> + check_no_overlap1([sets:from_list(Mod:applies_to()) || Mod <- Mods]). + +%% Check no non-empty pairwise intersection in a list of sets +check_no_overlap1(Sets) -> + lists:foldl(fun(Set, Union) -> + Is = sets:intersection(Set, Union), + case sets:size(Is) of + 0 -> ok; + _ -> + internal_error("Interceptor: more than one " + "module handles ~p~n", [Is]) + end, + sets:union(Set, Union) + end, + sets:new(), + Sets), + ok. + +intercept_in(M, C, Mods) -> + lists:foldl(fun({Mod, ModState}, {M1, C1}) -> + call_module(Mod, ModState, M1, C1) + end, + {M, C}, + Mods). + +call_module(Mod, St, M, C) -> + % this little dance is because Mod might be unloaded at any point + case (catch {ok, Mod:intercept(M, C, St)}) of + {ok, R} -> validate_response(Mod, M, C, R); + {'EXIT', {undef, [{Mod, intercept, _, _} | _]}} -> {M, C} + end. + +validate_response(Mod, M1, C1, R = {M2, C2}) -> + case {validate_method(M1, M2), validate_content(C1, C2)} of + {true, true} -> R; + {false, _} -> + internal_error("Interceptor: ~p expected to return " + "method: ~p but returned: ~p", + [Mod, rabbit_misc:method_record_type(M1), + rabbit_misc:method_record_type(M2)]); + {_, false} -> + internal_error("Interceptor: ~p expected to return " + "content iff content is provided but " + "content in = ~p; content out = ~p", + [Mod, C1, C2]) + end. + +validate_method(M, M2) -> + rabbit_misc:method_record_type(M) =:= rabbit_misc:method_record_type(M2). + +validate_content(none, none) -> true; +validate_content(#content{}, #content{}) -> true; +validate_content(_, _) -> false. + +%% keep dialyzer happy +-spec internal_error(string(), [any()]) -> no_return(). +internal_error(Format, Args) -> + rabbit_misc:protocol_error(internal_error, Format, Args). diff --git a/src/rabbit_exchange_decorator.erl b/src/rabbit_exchange_decorator.erl new file mode 100644 index 0000000000..58df5b87be --- /dev/null +++ b/src/rabbit_exchange_decorator.erl @@ -0,0 +1,114 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_exchange_decorator). + +-include("rabbit.hrl"). + +-export([select/2, set/1]). + +-behaviour(rabbit_registry_class). + +-export([added_to_rabbit_registry/2, removed_from_rabbit_registry/1]). + +%% This is like an exchange type except that: +%% +%% 1) It applies to all exchanges as soon as it is installed, therefore +%% 2) It is not allowed to affect validation, so no validate/1 or +%% assert_args_equivalence/2 +%% +%% It's possible in the future we might make decorators +%% able to manipulate messages as they are published. + +-type(tx() :: 'transaction' | 'none'). +-type(serial() :: pos_integer() | tx()). + +-callback description() -> [proplists:property()]. + +%% Should Rabbit ensure that all binding events that are +%% delivered to an individual exchange can be serialised? (they +%% might still be delivered out of order, but there'll be a +%% serial number). +-callback serialise_events(rabbit_types:exchange()) -> boolean(). + +%% called after declaration and recovery +-callback create(tx(), rabbit_types:exchange()) -> 'ok'. + +%% called after exchange (auto)deletion. +-callback delete(tx(), rabbit_types:exchange(), [rabbit_types:binding()]) -> + 'ok'. + +%% called when the policy attached to this exchange changes. +-callback policy_changed(rabbit_types:exchange(), rabbit_types:exchange()) -> + 'ok'. + +%% called after a binding has been added or recovered +-callback add_binding(serial(), rabbit_types:exchange(), + rabbit_types:binding()) -> 'ok'. + +%% called after bindings have been deleted. +-callback remove_bindings(serial(), rabbit_types:exchange(), + [rabbit_types:binding()]) -> 'ok'. + +%% Allows additional destinations to be added to the routing decision. +-callback route(rabbit_types:exchange(), rabbit_types:delivery()) -> + [rabbit_amqqueue:name() | rabbit_exchange:name()]. + +%% Whether the decorator wishes to receive callbacks for the exchange +%% none:no callbacks, noroute:all callbacks except route, all:all callbacks +-callback active_for(rabbit_types:exchange()) -> 'none' | 'noroute' | 'all'. + +%%---------------------------------------------------------------------------- + +added_to_rabbit_registry(_Type, _ModuleName) -> + [maybe_recover(X) || X <- rabbit_exchange:list()], + ok. +removed_from_rabbit_registry(_Type) -> + [maybe_recover(X) || X <- rabbit_exchange:list()], + ok. + +%% select a subset of active decorators +select(all, {Route, NoRoute}) -> filter(Route ++ NoRoute); +select(route, {Route, _NoRoute}) -> filter(Route); +select(raw, {Route, NoRoute}) -> Route ++ NoRoute. + +filter(Modules) -> + [M || M <- Modules, code:which(M) =/= non_existing]. + +set(X) -> + Decs = lists:foldl(fun (D, {Route, NoRoute}) -> + ActiveFor = D:active_for(X), + {cons_if_eq(all, ActiveFor, D, Route), + cons_if_eq(noroute, ActiveFor, D, NoRoute)} + end, {[], []}, list()), + X#exchange{decorators = Decs}. + +list() -> [M || {_, M} <- rabbit_registry:lookup_all(exchange_decorator)]. + +cons_if_eq(Select, Select, Item, List) -> [Item | List]; +cons_if_eq(_Select, _Other, _Item, List) -> List. + +maybe_recover(X = #exchange{name = Name, + decorators = Decs}) -> + #exchange{decorators = Decs1} = set(X), + Old = lists:sort(select(all, Decs)), + New = lists:sort(select(all, Decs1)), + case New of + Old -> ok; + _ -> %% TODO create a tx here for non-federation decorators + [M:create(none, X) || M <- New -- Old], + rabbit_exchange:update_decorators(Name) + end. diff --git a/src/rabbit_health_check.erl b/src/rabbit_health_check.erl new file mode 100644 index 0000000000..d8d4cc240b --- /dev/null +++ b/src/rabbit_health_check.erl @@ -0,0 +1,95 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% +-module(rabbit_health_check). + +%% External API +-export([node/1, node/2]). + +%% Internal API +-export([local/0]). + +-spec node(node(), timeout()) -> ok | {badrpc, term()} | {error_string, string()}. +-spec local() -> ok | {error_string, string()}. + +%%---------------------------------------------------------------------------- +%% External functions +%%---------------------------------------------------------------------------- + +node(Node) -> + %% same default as in CLI + node(Node, 70000). +node(Node, Timeout) -> + rabbit_misc:rpc_call(Node, rabbit_health_check, local, [], Timeout). + +local() -> + run_checks([list_channels, list_queues, alarms, rabbit_node_monitor]). + +%%---------------------------------------------------------------------------- +%% Internal functions +%%---------------------------------------------------------------------------- +run_checks([]) -> + ok; +run_checks([C|Cs]) -> + case node_health_check(C) of + ok -> + run_checks(Cs); + Error -> + Error + end. + +node_health_check(list_channels) -> + case rabbit_channel:info_local([pid]) of + L when is_list(L) -> + ok; + Other -> + ErrorMsg = io_lib:format("list_channels unexpected output: ~p", + [Other]), + {error_string, ErrorMsg} + end; + +node_health_check(list_queues) -> + health_check_queues(rabbit_vhost:list()); + +node_health_check(rabbit_node_monitor) -> + case rabbit_node_monitor:partitions() of + L when is_list(L) -> + ok; + Other -> + ErrorMsg = io_lib:format("rabbit_node_monitor reports unexpected partitions value: ~p", + [Other]), + {error_string, ErrorMsg} + end; + +node_health_check(alarms) -> + case proplists:get_value(alarms, rabbit:status()) of + [] -> + ok; + Alarms -> + ErrorMsg = io_lib:format("resource alarm(s) in effect:~p", [Alarms]), + {error_string, ErrorMsg} + end. + +health_check_queues([]) -> + ok; +health_check_queues([VHost|RestVHosts]) -> + case rabbit_amqqueue:info_local(VHost) of + L when is_list(L) -> + health_check_queues(RestVHosts); + Other -> + ErrorMsg = io_lib:format("list_queues unexpected output for vhost ~s: ~p", + [VHost, Other]), + {error_string, ErrorMsg} + end. diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl deleted file mode 100644 index 01e171c13c..0000000000 --- a/src/rabbit_log.erl +++ /dev/null @@ -1,128 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - --module(rabbit_log). - --export([log/3, log/4]). --export([debug/1, debug/2, debug/3, - info/1, info/2, info/3, - notice/1, notice/2, notice/3, - warning/1, warning/2, warning/3, - error/1, error/2, error/3, - critical/1, critical/2, critical/3, - alert/1, alert/2, alert/3, - emergency/1, emergency/2, emergency/3, - none/1, none/2, none/3]). - --include("rabbit_log.hrl"). -%%---------------------------------------------------------------------------- - --type category() :: atom(). - --spec debug(string()) -> 'ok'. --spec debug(string(), [any()]) -> 'ok'. --spec debug(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec info(string()) -> 'ok'. --spec info(string(), [any()]) -> 'ok'. --spec info(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec notice(string()) -> 'ok'. --spec notice(string(), [any()]) -> 'ok'. --spec notice(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec warning(string()) -> 'ok'. --spec warning(string(), [any()]) -> 'ok'. --spec warning(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec error(string()) -> 'ok'. --spec error(string(), [any()]) -> 'ok'. --spec error(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec critical(string()) -> 'ok'. --spec critical(string(), [any()]) -> 'ok'. --spec critical(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec alert(string()) -> 'ok'. --spec alert(string(), [any()]) -> 'ok'. --spec alert(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec emergency(string()) -> 'ok'. --spec emergency(string(), [any()]) -> 'ok'. --spec emergency(pid() | [tuple()], string(), [any()]) -> 'ok'. --spec none(string()) -> 'ok'. --spec none(string(), [any()]) -> 'ok'. --spec none(pid() | [tuple()], string(), [any()]) -> 'ok'. - -%%---------------------------------------------------------------------------- - --spec log(category(), lager:log_level(), string()) -> 'ok'. -log(Category, Level, Fmt) -> log(Category, Level, Fmt, []). - --spec log(category(), lager:log_level(), string(), [any()]) -> 'ok'. -log(Category, Level, Fmt, Args) when is_list(Args) -> - Sink = case Category of - default -> ?LAGER_SINK; - _ -> make_internal_sink_name(Category) - end, - lager:log(Sink, Level, self(), Fmt, Args). - -make_internal_sink_name(rabbit_log_connection) -> rabbit_log_connection_lager_event; -make_internal_sink_name(rabbit_log_channel) -> rabbit_log_channel_lager_event; -make_internal_sink_name(rabbit_log_mirroring) -> rabbit_log_mirroring_lager_event; -make_internal_sink_name(rabbit_log_queue) -> rabbit_log_queue_lager_event; -make_internal_sink_name(rabbit_log_federation) -> rabbit_log_federation_lager_event; -make_internal_sink_name(rabbit_log_upgrade) -> rabbit_log_upgrade_lager_event; -make_internal_sink_name(Category) -> - lager_util:make_internal_sink_name(Category). - -debug(Format) -> debug(Format, []). -debug(Format, Args) -> debug(self(), Format, Args). -debug(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, debug, Metadata, Format, Args). - -info(Format) -> info(Format, []). -info(Format, Args) -> info(self(), Format, Args). -info(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, info, Metadata, Format, Args). - -notice(Format) -> notice(Format, []). -notice(Format, Args) -> notice(self(), Format, Args). -notice(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, notice, Metadata, Format, Args). - -warning(Format) -> warning(Format, []). -warning(Format, Args) -> warning(self(), Format, Args). -warning(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, warning, Metadata, Format, Args). - -error(Format) -> ?MODULE:error(Format, []). -error(Format, Args) -> ?MODULE:error(self(), Format, Args). -error(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, error, Metadata, Format, Args). - -critical(Format) -> critical(Format, []). -critical(Format, Args) -> critical(self(), Format, Args). -critical(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, critical, Metadata, Format, Args). - -alert(Format) -> alert(Format, []). -alert(Format, Args) -> alert(self(), Format, Args). -alert(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, alert, Metadata, Format, Args). - -emergency(Format) -> emergency(Format, []). -emergency(Format, Args) -> emergency(self(), Format, Args). -emergency(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, emergency, Metadata, Format, Args). - -none(Format) -> none(Format, []). -none(Format, Args) -> none(self(), Format, Args). -none(Metadata, Format, Args) -> - lager:log(?LAGER_SINK, none, Metadata, Format, Args). diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl new file mode 100644 index 0000000000..cd65841933 --- /dev/null +++ b/src/rabbit_networking.erl @@ -0,0 +1,481 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_networking). + +%% This module contains various functions that deal with networking, +%% TCP and TLS listeners, and connection information. +%% +%% It also contains a boot step — boot/0 — that starts networking machinery. +%% This module primarily covers AMQP 0-9-1 but some bits are reused in +%% plugins that provide protocol support, e.g. STOMP or MQTT. +%% +%% Functions in this module take care of normalising TCP listener options, +%% including dual IP stack cases, and starting the AMQP 0-9-1 listener(s). +%% +%% See also tcp_listener_sup and tcp_listener. + +-export([boot/0, start_tcp_listener/2, start_ssl_listener/3, + stop_tcp_listener/1, on_node_down/1, active_listeners/0, + node_listeners/1, register_connection/1, unregister_connection/1, + connections/0, connection_info_keys/0, + connection_info/1, connection_info/2, + connection_info_all/0, connection_info_all/1, + emit_connection_info_all/4, emit_connection_info_local/3, + close_connection/2, force_connection_event_refresh/1, accept_ack/2, + tcp_host/1]). + +%% Used by TCP-based transports, e.g. STOMP adapter +-export([tcp_listener_addresses/1, tcp_listener_spec/9, + ensure_ssl/0, fix_ssl_options/1, poodle_check/1]). + +-export([tcp_listener_started/4, tcp_listener_stopped/4]). + +%% Internal +-export([connections_local/0]). + +-include("rabbit.hrl"). + +%% IANA-suggested ephemeral port range is 49152 to 65535 +-define(FIRST_TEST_BIND_PORT, 49152). + +%%---------------------------------------------------------------------------- + +-export_type([ip_port/0, hostname/0]). + +-type hostname() :: rabbit_net:hostname(). +-type ip_port() :: rabbit_net:ip_port(). + +-type family() :: atom(). +-type listener_config() :: ip_port() | + {hostname(), ip_port()} | + {hostname(), ip_port(), family()}. +-type address() :: {inet:ip_address(), ip_port(), family()}. +-type name_prefix() :: atom(). +-type protocol() :: atom(). +-type label() :: string(). + +-spec start_tcp_listener(listener_config(), integer()) -> 'ok'. +-spec start_ssl_listener + (listener_config(), rabbit_types:infos(), integer()) -> 'ok'. +-spec stop_tcp_listener(listener_config()) -> 'ok'. +-spec active_listeners() -> [rabbit_types:listener()]. +-spec node_listeners(node()) -> [rabbit_types:listener()]. +-spec register_connection(pid()) -> ok. +-spec unregister_connection(pid()) -> ok. +-spec connections() -> [rabbit_types:connection()]. +-spec connections_local() -> [rabbit_types:connection()]. +-spec connection_info_keys() -> rabbit_types:info_keys(). +-spec connection_info(rabbit_types:connection()) -> rabbit_types:infos(). +-spec connection_info(rabbit_types:connection(), rabbit_types:info_keys()) -> + rabbit_types:infos(). +-spec connection_info_all() -> [rabbit_types:infos()]. +-spec connection_info_all(rabbit_types:info_keys()) -> + [rabbit_types:infos()]. +-spec close_connection(pid(), string()) -> 'ok'. +-spec force_connection_event_refresh(reference()) -> 'ok'. +-spec accept_ack(any(), rabbit_net:socket()) -> ok. + +-spec on_node_down(node()) -> 'ok'. +-spec tcp_listener_addresses(listener_config()) -> [address()]. +-spec tcp_listener_spec + (name_prefix(), address(), [gen_tcp:listen_option()], module(), module(), + protocol(), any(), non_neg_integer(), label()) -> + supervisor:child_spec(). +-spec ensure_ssl() -> rabbit_types:infos(). +-spec poodle_check(atom()) -> 'ok' | 'danger'. + +-spec boot() -> 'ok'. +-spec tcp_listener_started + (_, _, + string() | + {byte(),byte(),byte(),byte()} | + {char(),char(),char(),char(),char(),char(),char(),char()}, _) -> + 'ok'. +-spec tcp_listener_stopped + (_, _, + string() | + {byte(),byte(),byte(),byte()} | + {char(),char(),char(),char(),char(),char(),char(),char()}, + _) -> + 'ok'. + +%%---------------------------------------------------------------------------- + +boot() -> + ok = record_distribution_listener(), + _ = application:start(ranch), + ok = boot_tcp(application:get_env(rabbit, num_tcp_acceptors, 10)), + ok = boot_ssl(application:get_env(rabbit, num_ssl_acceptors, 1)), + _ = maybe_start_proxy_protocol(), + ok. + +boot_tcp(NumAcceptors) -> + {ok, TcpListeners} = application:get_env(tcp_listeners), + [ok = start_tcp_listener(Listener, NumAcceptors) || Listener <- TcpListeners], + ok. + +boot_ssl(NumAcceptors) -> + case application:get_env(ssl_listeners) of + {ok, []} -> + ok; + {ok, SslListeners} -> + SslOpts = ensure_ssl(), + case poodle_check('AMQP') of + ok -> [start_ssl_listener(L, SslOpts, NumAcceptors) || L <- SslListeners]; + danger -> ok + end, + ok + end. + +ensure_ssl() -> + {ok, SslAppsConfig} = application:get_env(rabbit, ssl_apps), + ok = app_utils:start_applications(SslAppsConfig), + {ok, SslOptsConfig} = application:get_env(rabbit, ssl_options), + rabbit_ssl_options:fix(SslOptsConfig). + +poodle_check(Context) -> + {ok, Vsn} = application:get_key(ssl, vsn), + case rabbit_misc:version_compare(Vsn, "5.3", gte) of %% R16B01 + true -> ok; + false -> case application:get_env(rabbit, ssl_allow_poodle_attack) of + {ok, true} -> ok; + _ -> log_poodle_fail(Context), + danger + end + end. + +log_poodle_fail(Context) -> + rabbit_log:error( + "The installed version of Erlang (~s) contains the bug OTP-10905,~n" + "which makes it impossible to disable SSLv3. This makes the system~n" + "vulnerable to the POODLE attack. SSL listeners for ~s have therefore~n" + "been disabled.~n~n" + "You are advised to upgrade to a recent Erlang version; R16B01 is the~n" + "first version in which this bug is fixed, but later is usually~n" + "better.~n~n" + "If you cannot upgrade now and want to re-enable SSL listeners, you can~n" + "set the config item 'ssl_allow_poodle_attack' to 'true' in the~n" + "'rabbit' section of your configuration file.~n", + [rabbit_misc:otp_release(), Context]). + +maybe_start_proxy_protocol() -> + case application:get_env(rabbit, proxy_protocol, false) of + false -> ok; + true -> application:start(ranch_proxy_protocol) + end. + +fix_ssl_options(Config) -> + rabbit_ssl_options:fix(Config). + +tcp_listener_addresses(Port) when is_integer(Port) -> + tcp_listener_addresses_auto(Port); +tcp_listener_addresses({"auto", Port}) -> + %% Variant to prevent lots of hacking around in bash and batch files + tcp_listener_addresses_auto(Port); +tcp_listener_addresses({Host, Port}) -> + %% auto: determine family IPv4 / IPv6 after converting to IP address + tcp_listener_addresses({Host, Port, auto}); +tcp_listener_addresses({Host, Port, Family0}) + when is_integer(Port) andalso (Port >= 0) andalso (Port =< 65535) -> + [{IPAddress, Port, Family} || + {IPAddress, Family} <- getaddr(Host, Family0)]; +tcp_listener_addresses({_Host, Port, _Family0}) -> + rabbit_log:error("invalid port ~p - not 0..65535~n", [Port]), + throw({error, {invalid_port, Port}}). + +tcp_listener_addresses_auto(Port) -> + lists:append([tcp_listener_addresses(Listener) || + Listener <- port_to_listeners(Port)]). + +tcp_listener_spec(NamePrefix, {IPAddress, Port, Family}, SocketOpts, + Transport, ProtoSup, ProtoOpts, Protocol, NumAcceptors, Label) -> + {rabbit_misc:tcp_name(NamePrefix, IPAddress, Port), + {tcp_listener_sup, start_link, + [IPAddress, Port, Transport, [Family | SocketOpts], ProtoSup, ProtoOpts, + {?MODULE, tcp_listener_started, [Protocol, SocketOpts]}, + {?MODULE, tcp_listener_stopped, [Protocol, SocketOpts]}, + NumAcceptors, Label]}, + transient, infinity, supervisor, [tcp_listener_sup]}. + +start_tcp_listener(Listener, NumAcceptors) -> + start_listener(Listener, NumAcceptors, amqp, "TCP Listener", tcp_opts()). + +start_ssl_listener(Listener, SslOpts, NumAcceptors) -> + start_listener(Listener, NumAcceptors, 'amqp/ssl', "SSL Listener", tcp_opts() ++ SslOpts). + +start_listener(Listener, NumAcceptors, Protocol, Label, Opts) -> + [start_listener0(Address, NumAcceptors, Protocol, Label, Opts) || + Address <- tcp_listener_addresses(Listener)], + ok. + +start_listener0(Address, NumAcceptors, Protocol, Label, Opts) -> + Transport = transport(Protocol), + Spec = tcp_listener_spec(rabbit_tcp_listener_sup, Address, Opts, + Transport, rabbit_connection_sup, [], Protocol, + NumAcceptors, Label), + case supervisor:start_child(rabbit_sup, Spec) of + {ok, _} -> ok; + {error, {shutdown, _}} -> {IPAddress, Port, _Family} = Address, + exit({could_not_start_tcp_listener, + {rabbit_misc:ntoa(IPAddress), Port}}) + end. + +transport(Protocol) -> + ProxyProtocol = application:get_env(rabbit, proxy_protocol, false), + case {Protocol, ProxyProtocol} of + {amqp, false} -> ranch_tcp; + {amqp, true} -> ranch_proxy; + {'amqp/ssl', false} -> ranch_ssl; + {'amqp/ssl', true} -> ranch_proxy_ssl + end. + + +stop_tcp_listener(Listener) -> + [stop_tcp_listener0(Address) || + Address <- tcp_listener_addresses(Listener)], + ok. + +stop_tcp_listener0({IPAddress, Port, _Family}) -> + Name = rabbit_misc:tcp_name(rabbit_tcp_listener_sup, IPAddress, Port), + ok = supervisor:terminate_child(rabbit_sup, Name), + ok = supervisor:delete_child(rabbit_sup, Name). + +tcp_listener_started(Protocol, Opts, IPAddress, Port) -> + %% We need the ip to distinguish e.g. 0.0.0.0 and 127.0.0.1 + %% We need the host so we can distinguish multiple instances of the above + %% in a cluster. + ok = mnesia:dirty_write( + rabbit_listener, + #listener{node = node(), + protocol = Protocol, + host = tcp_host(IPAddress), + ip_address = IPAddress, + port = Port, + opts = Opts}). + +tcp_listener_stopped(Protocol, Opts, IPAddress, Port) -> + ok = mnesia:dirty_delete_object( + rabbit_listener, + #listener{node = node(), + protocol = Protocol, + host = tcp_host(IPAddress), + ip_address = IPAddress, + port = Port, + opts = Opts}). + +record_distribution_listener() -> + {Name, Host} = rabbit_nodes:parts(node()), + {port, Port, _Version} = erl_epmd:port_please(Name, Host), + tcp_listener_started(clustering, [], {0,0,0,0,0,0,0,0}, Port). + +active_listeners() -> + rabbit_misc:dirty_read_all(rabbit_listener). + +node_listeners(Node) -> + mnesia:dirty_read(rabbit_listener, Node). + +on_node_down(Node) -> + case lists:member(Node, nodes()) of + false -> + rabbit_log:info( + "Node ~s is down, deleting its listeners~n", [Node]), + ok = mnesia:dirty_delete(rabbit_listener, Node); + true -> + rabbit_log:info( + "Keeping ~s listeners: the node is already back~n", [Node]) + end. + +register_connection(Pid) -> pg_local:join(rabbit_connections, Pid). + +unregister_connection(Pid) -> pg_local:leave(rabbit_connections, Pid). + +connections() -> + rabbit_misc:append_rpc_all_nodes(rabbit_mnesia:cluster_nodes(running), + rabbit_networking, connections_local, []). + +connections_local() -> pg_local:get_members(rabbit_connections). + +connection_info_keys() -> rabbit_reader:info_keys(). + +connection_info(Pid) -> rabbit_reader:info(Pid). +connection_info(Pid, Items) -> rabbit_reader:info(Pid, Items). + +connection_info_all() -> cmap(fun (Q) -> connection_info(Q) end). +connection_info_all(Items) -> cmap(fun (Q) -> connection_info(Q, Items) end). + +emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) -> + Pids = [ spawn_link(Node, rabbit_networking, emit_connection_info_local, [Items, Ref, AggregatorPid]) || Node <- Nodes ], + rabbit_control_misc:await_emitters_termination(Pids), + ok. + +emit_connection_info_local(Items, Ref, AggregatorPid) -> + rabbit_control_misc:emitting_map_with_exit_handler( + AggregatorPid, Ref, fun(Q) -> connection_info(Q, Items) end, + connections_local()). + +close_connection(Pid, Explanation) -> + rabbit_log:info("Closing connection ~p because ~p~n", [Pid, Explanation]), + case lists:member(Pid, connections()) of + true -> rabbit_reader:shutdown(Pid, Explanation); + false -> throw({error, {not_a_connection_pid, Pid}}) + end. + +force_connection_event_refresh(Ref) -> + [rabbit_reader:force_event_refresh(C, Ref) || C <- connections()], + ok. + +accept_ack(Ref, Sock) -> + ok = ranch:accept_ack(Ref), + case tune_buffer_size(Sock) of + ok -> ok; + {error, _} -> rabbit_net:fast_close(Sock), + exit(normal) + end, + ok = file_handle_cache:obtain(). + +tune_buffer_size(Sock) -> + case rabbit_net:getopts(Sock, [sndbuf, recbuf, buffer]) of + {ok, BufSizes} -> BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]), + rabbit_net:setopts(Sock, [{buffer, BufSz}]); + Error -> Error + end. + +%%-------------------------------------------------------------------- + +tcp_host(IPAddress) -> + rabbit_net:tcp_host(IPAddress). + +cmap(F) -> rabbit_misc:filter_exit_map(F, connections()). + +tcp_opts() -> + {ok, ConfigOpts} = application:get_env(rabbit, tcp_listen_options), + ConfigOpts. + +%% inet_parse:address takes care of ip string, like "0.0.0.0" +%% inet:getaddr returns immediately for ip tuple {0,0,0,0}, +%% and runs 'inet_gethost' port process for dns lookups. +%% On Windows inet:getaddr runs dns resolver for ip string, which may fail. +getaddr(Host, Family) -> + case inet_parse:address(Host) of + {ok, IPAddress} -> [{IPAddress, resolve_family(IPAddress, Family)}]; + {error, _} -> gethostaddr(Host, Family) + end. + +gethostaddr(Host, auto) -> + Lookups = [{Family, inet:getaddr(Host, Family)} || Family <- [inet, inet6]], + case [{IP, Family} || {Family, {ok, IP}} <- Lookups] of + [] -> host_lookup_error(Host, Lookups); + IPs -> IPs + end; + +gethostaddr(Host, Family) -> + case inet:getaddr(Host, Family) of + {ok, IPAddress} -> [{IPAddress, Family}]; + {error, Reason} -> host_lookup_error(Host, Reason) + end. + +host_lookup_error(Host, Reason) -> + rabbit_log:error("invalid host ~p - ~p~n", [Host, Reason]), + throw({error, {invalid_host, Host, Reason}}). + +resolve_family({_,_,_,_}, auto) -> inet; +resolve_family({_,_,_,_,_,_,_,_}, auto) -> inet6; +resolve_family(IP, auto) -> throw({error, {strange_family, IP}}); +resolve_family(_, F) -> F. + +%%-------------------------------------------------------------------- + +%% There are three kinds of machine (for our purposes). +%% +%% * Those which treat IPv4 addresses as a special kind of IPv6 address +%% ("Single stack") +%% - Linux by default, Windows Vista and later +%% - We also treat any (hypothetical?) IPv6-only machine the same way +%% * Those which consider IPv6 and IPv4 to be completely separate things +%% ("Dual stack") +%% - OpenBSD, Windows XP / 2003, Linux if so configured +%% * Those which do not support IPv6. +%% - Ancient/weird OSes, Linux if so configured +%% +%% How to reconfigure Linux to test this: +%% Single stack (default): +%% echo 0 > /proc/sys/net/ipv6/bindv6only +%% Dual stack: +%% echo 1 > /proc/sys/net/ipv6/bindv6only +%% IPv4 only: +%% add ipv6.disable=1 to GRUB_CMDLINE_LINUX_DEFAULT in /etc/default/grub then +%% sudo update-grub && sudo reboot +%% +%% This matters in (and only in) the case where the sysadmin (or the +%% app descriptor) has only supplied a port and we wish to bind to +%% "all addresses". This means different things depending on whether +%% we're single or dual stack. On single stack binding to "::" +%% implicitly includes all IPv4 addresses, and subsequently attempting +%% to bind to "0.0.0.0" will fail. On dual stack, binding to "::" will +%% only bind to IPv6 addresses, and we need another listener bound to +%% "0.0.0.0" for IPv4. Finally, on IPv4-only systems we of course only +%% want to bind to "0.0.0.0". +%% +%% Unfortunately it seems there is no way to detect single vs dual stack +%% apart from attempting to bind to the port. +port_to_listeners(Port) -> + IPv4 = {"0.0.0.0", Port, inet}, + IPv6 = {"::", Port, inet6}, + case ipv6_status(?FIRST_TEST_BIND_PORT) of + single_stack -> [IPv6]; + ipv6_only -> [IPv6]; + dual_stack -> [IPv6, IPv4]; + ipv4_only -> [IPv4] + end. + +ipv6_status(TestPort) -> + IPv4 = [inet, {ip, {0,0,0,0}}], + IPv6 = [inet6, {ip, {0,0,0,0,0,0,0,0}}], + case gen_tcp:listen(TestPort, IPv6) of + {ok, LSock6} -> + case gen_tcp:listen(TestPort, IPv4) of + {ok, LSock4} -> + %% Dual stack + gen_tcp:close(LSock6), + gen_tcp:close(LSock4), + dual_stack; + %% Checking the error here would only let us + %% distinguish single stack IPv6 / IPv4 vs IPv6 only, + %% which we figure out below anyway. + {error, _} -> + gen_tcp:close(LSock6), + case gen_tcp:listen(TestPort, IPv4) of + %% Single stack + {ok, LSock4} -> gen_tcp:close(LSock4), + single_stack; + %% IPv6-only machine. Welcome to the future. + {error, eafnosupport} -> ipv6_only; %% Linux + {error, eprotonosupport}-> ipv6_only; %% FreeBSD + %% Dual stack machine with something already + %% on IPv4. + {error, _} -> ipv6_status(TestPort + 1) + end + end; + %% IPv4-only machine. Welcome to the 90s. + {error, eafnosupport} -> %% Linux + ipv4_only; + {error, eprotonosupport} -> %% FreeBSD + ipv4_only; + %% Port in use + {error, _} -> + ipv6_status(TestPort + 1) + end. diff --git a/src/rabbit_nodes.erl b/src/rabbit_nodes.erl new file mode 100644 index 0000000000..7a4b1c9dea --- /dev/null +++ b/src/rabbit_nodes.erl @@ -0,0 +1,232 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_nodes). + +-export([names/1, diagnostics/1, make/1, parts/1, cookie_hash/0, + is_running/2, is_process_running/2, + cluster_name/0, set_cluster_name/2, ensure_epmd/0, + all_running/0]). + +-include_lib("kernel/include/inet.hrl"). + +-define(EPMD_TIMEOUT, 30000). +-define(TCP_DIAGNOSTIC_TIMEOUT, 5000). +-define(ERROR_LOGGER_HANDLER, rabbit_error_logger_handler). + +%%---------------------------------------------------------------------------- +%% Specs +%%---------------------------------------------------------------------------- + +-spec names(string()) -> + rabbit_types:ok_or_error2([{string(), integer()}], term()). +-spec diagnostics([node()]) -> string(). +-spec cookie_hash() -> string(). +-spec is_running(node(), atom()) -> boolean(). +-spec is_process_running(node(), atom()) -> boolean(). +-spec cluster_name() -> binary(). +-spec set_cluster_name(binary(), rabbit_types:username()) -> 'ok'. +-spec ensure_epmd() -> 'ok'. +-spec all_running() -> [node()]. + +%%---------------------------------------------------------------------------- + +names(Hostname) -> + Self = self(), + Ref = make_ref(), + {Pid, MRef} = spawn_monitor( + fun () -> Self ! {Ref, net_adm:names(Hostname)} end), + timer:exit_after(?EPMD_TIMEOUT, Pid, timeout), + receive + {Ref, Names} -> erlang:demonitor(MRef, [flush]), + Names; + {'DOWN', MRef, process, Pid, Reason} -> {error, Reason} + end. + +diagnostics(Nodes) -> + verbose_erlang_distribution(true), + NodeDiags = [{"~nDIAGNOSTICS~n===========~n~n" + "attempted to contact: ~p~n", [Nodes]}] ++ + [diagnostics_node(Node) || Node <- Nodes] ++ + current_node_details(), + verbose_erlang_distribution(false), + rabbit_misc:format_many(lists:flatten(NodeDiags)). + +verbose_erlang_distribution(true) -> + net_kernel:verbose(1), + error_logger:add_report_handler(?ERROR_LOGGER_HANDLER); +verbose_erlang_distribution(false) -> + net_kernel:verbose(0), + error_logger:delete_report_handler(?ERROR_LOGGER_HANDLER). + +current_node_details() -> + [{"~ncurrent node details:~n- node name: ~w", [node()]}, + case init:get_argument(home) of + {ok, [[Home]]} -> {"- home dir: ~s", [Home]}; + Other -> {"- no home dir: ~p", [Other]} + end, + {"- cookie hash: ~s", [cookie_hash()]}]. + +diagnostics_node(Node) -> + {Name, Host} = parts(Node), + [{"~s:", [Node]} | + case names(Host) of + {error, Reason} -> + [{" * unable to connect to epmd (port ~s) on ~s: ~s~n", + [epmd_port(), Host, rabbit_misc:format_inet_error(Reason)]}]; + {ok, NamePorts} -> + [{" * connected to epmd (port ~s) on ~s", + [epmd_port(), Host]}] ++ + case net_adm:ping(Node) of + pong -> dist_working_diagnostics(Node); + pang -> dist_broken_diagnostics(Name, Host, NamePorts) + end + end]. + +epmd_port() -> + case init:get_argument(epmd_port) of + {ok, [[Port | _] | _]} when is_list(Port) -> Port; + error -> "4369" + end. + +dist_working_diagnostics(Node) -> + case is_process_running(Node, rabbit) of + true -> [{" * node ~s up, 'rabbit' application running", [Node]}]; + false -> [{" * node ~s up, 'rabbit' application not running~n" + " * running applications on ~s: ~p~n" + " * suggestion: start_app on ~s", + [Node, Node, remote_apps(Node), Node]}] + end. + +remote_apps(Node) -> + %% We want a timeout here because really, we don't trust the node, + %% the last thing we want to do is hang. + case rpc:call(Node, application, which_applications, [5000]) of + {badrpc, _} = E -> E; + Apps -> [App || {App, _, _} <- Apps] + end. + +dist_broken_diagnostics(Name, Host, NamePorts) -> + case [{N, P} || {N, P} <- NamePorts, N =:= Name] of + [] -> + {SelfName, SelfHost} = parts(node()), + Others = [list_to_atom(N) || {N, _} <- NamePorts, + N =/= case SelfHost of + Host -> SelfName; + _ -> never_matches + end], + OthersDiag = case Others of + [] -> [{" no other nodes on ~s", + [Host]}]; + _ -> [{" other nodes on ~s: ~p", + [Host, Others]}] + end, + [{" * epmd reports: node '~s' not running at all", [Name]}, + OthersDiag, {" * suggestion: start the node", []}]; + [{Name, Port}] -> + [{" * epmd reports node '~s' running on port ~b", [Name, Port]} | + case diagnose_connect(Host, Port) of + ok -> + connection_succeeded_diagnostics(); + {error, Reason} -> + [{" * can't establish TCP connection, reason: ~s~n" + " * suggestion: blocked by firewall?", + [rabbit_misc:format_inet_error(Reason)]}] + end] + end. + +connection_succeeded_diagnostics() -> + case gen_event:call(error_logger, ?ERROR_LOGGER_HANDLER, get_connection_report) of + [] -> + [{" * TCP connection succeeded but Erlang distribution " + "failed~n" + " * suggestion: hostname mismatch?~n" + " * suggestion: is the cookie set correctly?~n" + " * suggestion: is the Erlang distribution using TLS?", []}]; + Report -> + [{" * TCP connection succeeded but Erlang distribution " + "failed~n", []}] + ++ Report + end. + +diagnose_connect(Host, Port) -> + case inet:gethostbyname(Host) of + {ok, #hostent{h_addrtype = Family}} -> + case gen_tcp:connect(Host, Port, [Family], + ?TCP_DIAGNOSTIC_TIMEOUT) of + {ok, Socket} -> gen_tcp:close(Socket), + ok; + {error, _} = E -> E + end; + {error, _} = E -> + E + end. + +make(NodeStr) -> + rabbit_nodes_common:make(NodeStr). + +parts(NodeStr) -> + rabbit_nodes_common:parts(NodeStr). + +cookie_hash() -> + base64:encode_to_string(erlang:md5(atom_to_list(erlang:get_cookie()))). + +is_running(Node, Application) -> + case rpc:call(Node, rabbit_misc, which_applications, []) of + {badrpc, _} -> false; + Apps -> proplists:is_defined(Application, Apps) + end. + +is_process_running(Node, Process) -> + case rpc:call(Node, erlang, whereis, [Process]) of + {badrpc, _} -> false; + undefined -> false; + P when is_pid(P) -> true + end. + +cluster_name() -> + rabbit_runtime_parameters:value_global( + cluster_name, cluster_name_default()). + +cluster_name_default() -> + {ID, _} = rabbit_nodes:parts(node()), + {ok, Host} = inet:gethostname(), + {ok, #hostent{h_name = FQDN}} = inet:gethostbyname(Host), + list_to_binary(atom_to_list(rabbit_nodes:make({ID, FQDN}))). + +set_cluster_name(Name, Username) -> + %% Cluster name should be binary + BinaryName = rabbit_data_coercion:to_binary(Name), + rabbit_runtime_parameters:set_global(cluster_name, BinaryName, Username). + +ensure_epmd() -> + {ok, Prog} = init:get_argument(progname), + ID = rabbit_misc:random(1000000000), + Port = open_port( + {spawn_executable, os:find_executable(Prog)}, + [{args, ["-sname", rabbit_misc:format("epmd-starter-~b", [ID]), + "-noshell", "-eval", "halt()."]}, + exit_status, stderr_to_stdout, use_stdio]), + port_shutdown_loop(Port). + +port_shutdown_loop(Port) -> + receive + {Port, {exit_status, _Rc}} -> ok; + {Port, _} -> port_shutdown_loop(Port) + end. + +all_running() -> rabbit_mnesia:cluster_nodes(running). diff --git a/src/rabbit_queue_collector.erl b/src/rabbit_queue_collector.erl new file mode 100644 index 0000000000..dd8423e7a0 --- /dev/null +++ b/src/rabbit_queue_collector.erl @@ -0,0 +1,89 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_queue_collector). + +%% Queue collector keeps track of exclusive queues and cleans them +%% up e.g. when their connection is closed. + +-behaviour(gen_server). + +-export([start_link/1, register/2, delete_all/1]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-record(state, {monitors, delete_from}). + +-include("rabbit.hrl"). + +%%---------------------------------------------------------------------------- + +-spec start_link(rabbit_types:proc_name()) -> rabbit_types:ok_pid_or_error(). +-spec register(pid(), pid()) -> 'ok'. + +%%---------------------------------------------------------------------------- + +start_link(ProcName) -> + gen_server:start_link(?MODULE, [ProcName], []). + +register(CollectorPid, Q) -> + gen_server:call(CollectorPid, {register, Q}, infinity). + +delete_all(CollectorPid) -> + rabbit_queue_collector_common:delete_all(CollectorPid). + +%%---------------------------------------------------------------------------- + +init([ProcName]) -> + ?store_proc_name(ProcName), + {ok, #state{monitors = pmon:new(), delete_from = undefined}}. + +%%-------------------------------------------------------------------------- + +handle_call({register, QPid}, _From, + State = #state{monitors = QMons, delete_from = Deleting}) -> + case Deleting of + undefined -> ok; + _ -> ok = rabbit_amqqueue:delete_exclusive([QPid], Deleting) + end, + {reply, ok, State#state{monitors = pmon:monitor(QPid, QMons)}}; + +handle_call(delete_all, From, State = #state{monitors = QMons, + delete_from = undefined}) -> + case pmon:monitored(QMons) of + [] -> {reply, ok, State#state{delete_from = From}}; + QPids -> ok = rabbit_amqqueue:delete_exclusive(QPids, From), + {noreply, State#state{delete_from = From}} + end. + +handle_cast(Msg, State) -> + {stop, {unhandled_cast, Msg}, State}. + +handle_info({'DOWN', _MRef, process, DownPid, _Reason}, + State = #state{monitors = QMons, delete_from = Deleting}) -> + QMons1 = pmon:erase(DownPid, QMons), + case Deleting =/= undefined andalso pmon:is_empty(QMons1) of + true -> gen_server:reply(Deleting, ok); + false -> ok + end, + {noreply, State#state{monitors = QMons1}}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. diff --git a/src/rabbit_queue_decorator.erl b/src/rabbit_queue_decorator.erl new file mode 100644 index 0000000000..512fe4c874 --- /dev/null +++ b/src/rabbit_queue_decorator.erl @@ -0,0 +1,73 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_queue_decorator). + +-include("rabbit.hrl"). + +-export([select/1, set/1, register/2, unregister/1]). + +-behaviour(rabbit_registry_class). + +-export([added_to_rabbit_registry/2, removed_from_rabbit_registry/1]). + +%%---------------------------------------------------------------------------- + +-callback startup(rabbit_types:amqqueue()) -> 'ok'. + +-callback shutdown(rabbit_types:amqqueue()) -> 'ok'. + +-callback policy_changed(rabbit_types:amqqueue(), rabbit_types:amqqueue()) -> + 'ok'. + +-callback active_for(rabbit_types:amqqueue()) -> boolean(). + +%% called with Queue, MaxActivePriority, IsEmpty +-callback consumer_state_changed( + rabbit_types:amqqueue(), integer(), boolean()) -> 'ok'. + +%%---------------------------------------------------------------------------- + +added_to_rabbit_registry(_Type, _ModuleName) -> ok. +removed_from_rabbit_registry(_Type) -> ok. + +select(Modules) -> + [M || M <- Modules, code:which(M) =/= non_existing]. + +set(Q) -> Q#amqqueue{decorators = [D || D <- list(), D:active_for(Q)]}. + +list() -> [M || {_, M} <- rabbit_registry:lookup_all(queue_decorator)]. + +register(TypeName, ModuleName) -> + rabbit_registry:register(queue_decorator, TypeName, ModuleName), + [maybe_recover(Q) || Q <- rabbit_amqqueue:list()], + ok. + +unregister(TypeName) -> + rabbit_registry:unregister(queue_decorator, TypeName), + [maybe_recover(Q) || Q <- rabbit_amqqueue:list()], + ok. + +maybe_recover(Q = #amqqueue{name = Name, + decorators = Decs}) -> + #amqqueue{decorators = Decs1} = set(Q), + Old = lists:sort(select(Decs)), + New = lists:sort(select(Decs1)), + case New of + Old -> ok; + _ -> [M:startup(Q) || M <- New -- Old], + rabbit_amqqueue:update_decorators(Name) + end. diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl new file mode 100644 index 0000000000..9f3445731b --- /dev/null +++ b/src/rabbit_reader.erl @@ -0,0 +1,1652 @@ +%% 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 Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. +%% + +-module(rabbit_reader). + +%% This is an AMQP 0-9-1 connection implementation. If AMQP 1.0 plugin is enabled, +%% this module passes control of incoming AMQP 1.0 connections to it. +%% +%% Every connection (as in, a process using this module) +%% is a controlling process for a server socket. +%% +%% Connections have a number of responsibilities: +%% +%% * Performing protocol handshake +%% * Parsing incoming data and dispatching protocol methods +%% * Authenticating clients (with the help of authentication backends) +%% * Enforcing TCP backpressure (throttling clients) +%% * Enforcing connection limits, e.g. channel_max +%% * Channel management +%% * Setting up heartbeater and alarm notifications +%% * Emitting connection and network activity metric events +%% * Gracefully handling client disconnects, channel termination, etc +%% +%% and a few more. +%% +%% Every connection has +%% +%% * a queue collector which is responsible for keeping +%% track of exclusive queues on the connection and their cleanup. +%% * a heartbeater that's responsible for sending heartbeat frames to clients, +%% keeping track of the incoming ones and notifying connection about +%% heartbeat timeouts +%% * Stats timer, a timer that is used to periodically emit metric events +%% +%% Some dependencies are started under a separate supervisor to avoid deadlocks +%% during system shutdown. See rabbit_channel_sup:start_link/0 for details. +%% +%% Reader processes are special processes (in the OTP sense). + +-include("rabbit_framing.hrl"). +-include("rabbit.hrl"). + +-export([start_link/3, info_keys/0, info/1, info/2, force_event_refresh/2, + shutdown/2]). + +-export([system_continue/3, system_terminate/4, system_code_change/4]). + +-export([init/4, mainloop/4, recvloop/4]). + +-export([conserve_resources/3, server_properties/1]). + +-define(NORMAL_TIMEOUT, 3). +-define(CLOSING_TIMEOUT, 30). +-define(CHANNEL_TERMINATION_TIMEOUT, 3). +%% we wait for this many seconds before closing TCP connection +%% with a client that failed to log in. Provides some relief +%% from connection storms and DoS. +-define(SILENT_CLOSE_DELAY, 3). +-define(CHANNEL_MIN, 1). + +%%-------------------------------------------------------------------------- + +-record(v1, { + %% parent process + parent, + %% socket + sock, + %% connection state, see connection record + connection, + callback, + recv_len, + pending_recv, + %% pre_init | securing | running | blocking | blocked | closing | closed | {become, F} + connection_state, + %% see comment in rabbit_connection_sup:start_link/0 + helper_sup, + %% takes care of cleaning up exclusive queues, + %% see rabbit_queue_collector + queue_collector, + %% sends and receives heartbeat frames, + %% see rabbit_heartbeat + heartbeater, + %% timer used to emit statistics + stats_timer, + %% channel supervisor + channel_sup_sup_pid, + %% how many channels this connection has + channel_count, + %% throttling state, for both + %% credit- and resource-driven flow control + throttle, + proxy_socket}). + +-record(throttle, { + %% never | timestamp() + last_blocked_at, + %% a set of the reasons why we are + %% blocked: {resource, memory}, {resource, disk}. + %% More reasons can be added in the future. + blocked_by, + %% true if received any publishes, false otherwise + %% note that this will also be true when connection is + %% already blocked + should_block, + %% true if we had we sent a connection.blocked, + %% false otherwise + connection_blocked_message_sent +}). + +-define(STATISTICS_KEYS, [pid, recv_oct, recv_cnt, send_oct, send_cnt, + send_pend, state, channels, reductions, + garbage_collection]). + +-define(SIMPLE_METRICS, [pid, recv_oct, send_oct, reductions]). +-define(OTHER_METRICS, [recv_cnt, send_cnt, send_pend, state, channels, + garbage_collection]). + +-define(CREATION_EVENT_KEYS, + [pid, name, port, peer_port, host, + peer_host, ssl, peer_cert_subject, peer_cert_issuer, + peer_cert_validity, auth_mechanism, ssl_protocol, + ssl_key_exchange, ssl_cipher, ssl_hash, protocol, user, vhost, + timeout, frame_max, channel_max, client_properties, connected_at, + node, user_who_performed_action]). + +-define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]). + +-define(AUTH_NOTIFICATION_INFO_KEYS, + [host, name, peer_host, peer_port, protocol, auth_mechanism, + ssl, ssl_protocol, ssl_cipher, peer_cert_issuer, peer_cert_subject, + peer_cert_validity]). + +-define(IS_RUNNING(State), + (State#v1.connection_state =:= running orelse + State#v1.connection_state =:= blocked)). + +-define(IS_STOPPING(State), + (State#v1.connection_state =:= closing orelse + State#v1.connection_state =:= closed)). + +%%-------------------------------------------------------------------------- + +-spec start_link(pid(), any(), rabbit_net:socket()) -> rabbit_types:ok(pid()). +-spec info_keys() -> rabbit_types:info_keys(). +-spec info(pid()) -> rabbit_types:infos(). +-spec info(pid(), rabbit_types:info_keys()) -> rabbit_types:infos(). +-spec force_event_refresh(pid(), reference()) -> 'ok'. +-spec shutdown(pid(), string()) -> 'ok'. +-type resource_alert() :: {WasAlarmSetForNode :: boolean(), + IsThereAnyAlarmsWithSameSourceInTheCluster :: boolean(), + NodeForWhichAlarmWasSetOrCleared :: node()}. +-spec conserve_resources(pid(), atom(), resource_alert()) -> 'ok'. +-spec server_properties(rabbit_types:protocol()) -> + rabbit_framing:amqp_table(). + +%% These specs only exists to add no_return() to keep dialyzer happy +-spec init(pid(), pid(), any(), rabbit_net:socket()) -> no_return(). +-spec start_connection(pid(), pid(), any(), rabbit_net:socket()) -> + no_return(). + +-spec mainloop(_,[binary()], non_neg_integer(), #v1{}) -> any(). +-spec system_code_change(_,_,_,_) -> {'ok',_}. +-spec system_continue(_,_,{[binary()], non_neg_integer(), #v1{}}) -> any(). +-spec system_terminate(_,_,_,_) -> none(). + +%%-------------------------------------------------------------------------- + +start_link(HelperSup, Ref, Sock) -> + Pid = proc_lib:spawn_link(?MODULE, init, [self(), HelperSup, Ref, Sock]), + + %% In the event that somebody floods us with connections, the + %% reader processes can spew log events at error_logger faster + %% than it can keep up, causing its mailbox to grow unbounded + %% until we eat all the memory available and crash. So here is a + %% meaningless synchronous call to the underlying gen_event + %% mechanism. When it returns the mailbox is drained, and we + %% return to our caller to accept more connections. + gen_event:which_handlers(error_logger), + + {ok, Pid}. + +shutdown(Pid, Explanation) -> + gen_server:call(Pid, {shutdown, Explanation}, infinity). + +init(Parent, HelperSup, Ref, Sock) -> + RealSocket = rabbit_net:unwrap_socket(Sock), + rabbit_networking:accept_ack(Ref, RealSocket), + Deb = sys:debug_options([]), + start_connection(Parent, HelperSup, Deb, Sock). + +system_continue(Parent, Deb, {Buf, BufLen, State}) -> + mainloop(Deb, Buf, BufLen, State#v1{parent = Parent}). + +system_terminate(Reason, _Parent, _Deb, _State) -> + exit(Reason). + +system_code_change(Misc, _Module, _OldVsn, _Extra) -> + {ok, Misc}. + +info_keys() -> ?INFO_KEYS. + +info(Pid) -> + gen_server:call(Pid, info, infinity). + +info(Pid, Items) -> + case gen_server:call(Pid, {info, Items}, infinity) of + {ok, Res} -> Res; + {error, Error} -> throw(Error) + end. + +force_event_refresh(Pid, Ref) -> + gen_server:cast(Pid, {force_event_refresh, Ref}). + +conserve_resources(Pid, Source, {_, Conserve, _}) -> + Pid ! {conserve_resources, Source, Conserve}, + ok. + +server_properties(Protocol) -> + {ok, Product} = application:get_key(rabbit, description), + {ok, Version} = application:get_key(rabbit, vsn), + + %% Get any configuration-specified server properties + {ok, RawConfigServerProps} = application:get_env(rabbit, + server_properties), + + %% Normalize the simplifed (2-tuple) and unsimplified (3-tuple) forms + %% from the config and merge them with the generated built-in properties + NormalizedConfigServerProps = + [{<<"capabilities">>, table, server_capabilities(Protocol)} | + [case X of + {KeyAtom, Value} -> {list_to_binary(atom_to_list(KeyAtom)), + longstr, + maybe_list_to_binary(Value)}; + {BinKey, Type, Value} -> {BinKey, Type, Value} + end || X <- RawConfigServerProps ++ + [{product, Product}, + {version, Version}, + {cluster_name, rabbit_nodes:cluster_name()}, + {platform, "Erlang/OTP"}, + {copyright, ?COPYRIGHT_MESSAGE}, + {information, ?INFORMATION_MESSAGE}]]], + + %% Filter duplicated properties in favour of config file provided values + lists:usort(fun ({K1,_,_}, {K2,_,_}) -> K1 =< K2 end, + NormalizedConfigServerProps). + +maybe_list_to_binary(V) when is_binary(V) -> V; +maybe_list_to_binary(V) when is_list(V) -> list_to_binary(V). + +server_capabilities(rabbit_framing_amqp_0_9_1) -> + [{<<"publisher_confirms">>, bool, true}, + {<<"exchange_exchange_bindings">>, bool, true}, + {<<"basic.nack">>, bool, true}, + {<<"consumer_cancel_notify">>, bool, true}, + {<<"connection.blocked">>, bool, true}, + {<<"consumer_priorities">>, bool, true}, + {<<"authentication_failure_close">>, bool, true}, + {<<"per_consumer_qos">>, bool, true}, + {<<"direct_reply_to">>, bool, true}]; +server_capabilities(_) -> + []. + +%%-------------------------------------------------------------------------- + +socket_error(Reason) when is_atom(Reason) -> + rabbit_log_connection:error("Error on AMQP connection ~p: ~s~n", + [self(), rabbit_misc:format_inet_error(Reason)]); +socket_error(Reason) -> + Level = + case Reason of + {ssl_upgrade_error, closed} -> + %% The socket was closed while upgrading to SSL. + %% This is presumably a TCP healthcheck, so don't log + %% it unless specified otherwise. + debug; + _ -> + error + end, + rabbit_log:log(rabbit_log_connection, Level, + "Error on AMQP connection ~p:~n~p~n", [self(), Reason]). + +inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F). + +socket_op(Sock, Fun) -> + RealSocket = rabbit_net:unwrap_socket(Sock), + case Fun(Sock) of + {ok, Res} -> Res; + {error, Reason} -> socket_error(Reason), + rabbit_net:fast_close(RealSocket), + exit(normal) + end. + +start_connection(Parent, HelperSup, Deb, Sock) -> + process_flag(trap_exit, true), + RealSocket = rabbit_net:unwrap_socket(Sock), + Name = case rabbit_net:connection_string(Sock, inbound) of + {ok, Str} -> list_to_binary(Str); + {error, enotconn} -> rabbit_net:fast_close(RealSocket), + exit(normal); + {error, Reason} -> socket_error(Reason), + rabbit_net:fast_close(RealSocket), + exit(normal) + end, + {ok, HandshakeTimeout} = application:get_env(rabbit, handshake_timeout), + InitialFrameMax = application:get_env(rabbit, initial_frame_max, ?FRAME_MIN_SIZE), + 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(Name), + State = #v1{parent = Parent, + sock = RealSocket, + connection = #connection{ + name = Name, + log_name = Name, + host = Host, + peer_host = PeerHost, + port = Port, + peer_port = PeerPort, + protocol = none, + user = none, + timeout_sec = (HandshakeTimeout / 1000), + frame_max = InitialFrameMax, + vhost = none, + client_properties = none, + capabilities = [], + auth_mechanism = none, + auth_state = none, + connected_at = os:system_time( + milli_seconds)}, + callback = uninitialized_callback, + recv_len = 0, + pending_recv = false, + connection_state = pre_init, + queue_collector = undefined, %% started on tune-ok + helper_sup = HelperSup, + heartbeater = none, + channel_sup_sup_pid = none, + channel_count = 0, + throttle = #throttle{ + last_blocked_at = never, + should_block = false, + blocked_by = sets:new(), + connection_blocked_message_sent = false + }, + proxy_socket = rabbit_net:maybe_get_proxy_socket(Sock)}, + try + case run({?MODULE, recvloop, + [Deb, [], 0, switch_callback(rabbit_event:init_stats_timer( + State, #v1.stats_timer), + handshake, 8)]}) of + %% connection was closed cleanly by the client + #v1{connection = #connection{user = #user{username = Username}, + vhost = VHost}} -> + rabbit_log_connection:info("closing AMQP connection ~p (~s, vhost: '~s', user: '~s')~n", + [self(), dynamic_connection_name(Name), VHost, Username]); + %% just to be more defensive + _ -> + rabbit_log_connection:info("closing AMQP connection ~p (~s)~n", + [self(), dynamic_connection_name(Name)]) + end + catch + Ex -> + log_connection_exception(dynamic_connection_name(Name), Ex) + after + %% We don't call gen_tcp:close/1 here since it waits for + %% pending output to be sent, which results in unnecessary + %% delays. We could just terminate - the reader is the + %% controlling process and hence its termination will close + %% the socket. However, to keep the file_handle_cache + %% accounting as accurate as possible we ought to close the + %% socket w/o delay before termination. + rabbit_net:fast_close(RealSocket), + rabbit_networking:unregister_connection(self()), + rabbit_core_metrics:connection_closed(self()), + rabbit_event:notify(connection_closed, [{name, Name}, + {pid, self()}, + {node, node()}]) + end, + done. + +log_connection_exception(Name, Ex) -> + Severity = case Ex of + connection_closed_with_no_data_received -> debug; + {connection_closed_abruptly, _} -> warning; + connection_closed_abruptly -> warning; + _ -> error + end, + log_connection_exception(Severity, Name, Ex). + +log_connection_exception(Severity, Name, {heartbeat_timeout, TimeoutSec}) -> + %% Long line to avoid extra spaces and line breaks in log + rabbit_log:log(rabbit_log_connection, Severity, + "closing AMQP connection ~p (~s):~n" + "missed heartbeats from client, timeout: ~ps~n", + [self(), Name, TimeoutSec]); +log_connection_exception(Severity, Name, {connection_closed_abruptly, + #v1{connection = #connection{user = #user{username = Username}, + vhost = VHost}}}) -> + rabbit_log:log(rabbit_log_connection, Severity, "closing AMQP connection ~p (~s, vhost: '~s', user: '~s'):~nclient unexpectedly closed TCP connection~n", + [self(), Name, VHost, Username]); +%% when client abruptly closes connection before connection.open/authentication/authorization +%% succeeded, don't log username and vhost as 'none' +log_connection_exception(Severity, Name, {connection_closed_abruptly, _}) -> + rabbit_log:log(rabbit_log_connection, Severity, "closing AMQP connection ~p (~s):~nclient unexpectedly closed TCP connection~n", + [self(), Name]); +%% old exception structure +log_connection_exception(Severity, Name, connection_closed_abruptly) -> + rabbit_log:log(rabbit_log_connection, Severity, + "closing AMQP connection ~p (~s):~n" + "client unexpectedly closed TCP connection~n", + [self(), Name]); +log_connection_exception(Severity, Name, Ex) -> + rabbit_log:log(rabbit_log_connection, Severity, + "closing AMQP connection ~p (~s):~n~p~n", + [self(), Name, Ex]). + +run({M, F, A}) -> + try apply(M, F, A) + catch {become, MFA} -> run(MFA) + end. + +recvloop(Deb, Buf, BufLen, State = #v1{pending_recv = true}) -> + mainloop(Deb, Buf, BufLen, State); +recvloop(Deb, Buf, BufLen, State = #v1{connection_state = blocked}) -> + mainloop(Deb, Buf, BufLen, State); +recvloop(Deb, Buf, BufLen, State = #v1{connection_state = {become, F}}) -> + throw({become, F(Deb, Buf, BufLen, State)}); +recvloop(Deb, Buf, BufLen, State = #v1{sock = Sock, recv_len = RecvLen}) + when BufLen < RecvLen -> + case rabbit_net:setopts(Sock, [{active, once}]) of + ok -> mainloop(Deb, Buf, BufLen, + State#v1{pending_recv = true}); + {error, Reason} -> stop(Reason, State) + end; +recvloop(Deb, [B], _BufLen, State) -> + {Rest, State1} = handle_input(State#v1.callback, B, State), + recvloop(Deb, [Rest], size(Rest), State1); +recvloop(Deb, Buf, BufLen, State = #v1{recv_len = RecvLen}) -> + {DataLRev, RestLRev} = binlist_split(BufLen - RecvLen, Buf, []), + Data = list_to_binary(lists:reverse(DataLRev)), + {<<>>, State1} = handle_input(State#v1.callback, Data, State), + recvloop(Deb, lists:reverse(RestLRev), BufLen - RecvLen, State1). + +binlist_split(0, L, Acc) -> + {L, Acc}; +binlist_split(Len, L, [Acc0|Acc]) when Len < 0 -> + {H, T} = split_binary(Acc0, -Len), + {[H|L], [T|Acc]}; +binlist_split(Len, [H|T], Acc) -> + binlist_split(Len - size(H), T, [H|Acc]). + +mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock, + connection_state = CS, + connection = #connection{ + name = ConnName}}) -> + Recv = rabbit_net:recv(Sock), + case CS of + pre_init when Buf =:= [] -> + %% We only log incoming connections when either the + %% first byte was received or there was an error (eg. a + %% timeout). + %% + %% The goal is to not log TCP healthchecks (a connection + %% with no data received) unless specified otherwise. + Level = case Recv of + closed -> debug; + _ -> info + end, + rabbit_log:log(rabbit_log_connection, Level, + "accepting AMQP connection ~p (~s)~n", + [self(), ConnName]); + _ -> + ok + end, + case Recv of + {data, Data} -> + recvloop(Deb, [Data | Buf], BufLen + size(Data), + State#v1{pending_recv = false}); + closed when State#v1.connection_state =:= closed -> + State; + closed when CS =:= pre_init andalso Buf =:= [] -> + stop(tcp_healthcheck, State); + closed -> + stop(closed, State); + {other, {heartbeat_send_error, Reason}} -> + %% The only portable way to detect disconnect on blocked + %% connection is to wait for heartbeat send failure. + stop(Reason, State); + {error, Reason} -> + stop(Reason, State); + {other, {system, From, Request}} -> + sys:handle_system_msg(Request, From, State#v1.parent, + ?MODULE, Deb, {Buf, BufLen, State}); + {other, Other} -> + case handle_other(Other, State) of + stop -> State; + NewState -> recvloop(Deb, Buf, BufLen, NewState) + end + end. + +stop(tcp_healthcheck, State) -> + %% The connection was closed before any packet was received. It's + %% probably a load-balancer healthcheck: don't consider this a + %% failure. + maybe_emit_stats(State), + throw(connection_closed_with_no_data_received); +stop(closed, State) -> + maybe_emit_stats(State), + throw({connection_closed_abruptly, State}); +stop(Reason, State) -> + maybe_emit_stats(State), + throw({inet_error, Reason}). + +handle_other({conserve_resources, Source, Conserve}, + State = #v1{throttle = Throttle = #throttle{blocked_by = Blockers}}) -> + Resource = {resource, Source}, + Blockers1 = case Conserve of + true -> sets:add_element(Resource, Blockers); + false -> sets:del_element(Resource, Blockers) + end, + control_throttle(State#v1{throttle = Throttle#throttle{blocked_by = Blockers1}}); +handle_other({channel_closing, ChPid}, State) -> + ok = rabbit_channel:ready_for_close(ChPid), + {_, State1} = channel_cleanup(ChPid, State), + maybe_close(control_throttle(State1)); +handle_other({'EXIT', Parent, Reason}, State = #v1{parent = Parent}) -> + terminate(io_lib:format("broker forced connection closure " + "with reason '~w'", [Reason]), State), + %% this is what we are expected to do according to + %% http://www.erlang.org/doc/man/sys.html + %% + %% If we wanted to be *really* nice we should wait for a while for + %% clients to close the socket at their end, just as we do in the + %% ordinary error case. However, since this termination is + %% initiated by our parent it is probably more important to exit + %% quickly. + maybe_emit_stats(State), + exit(Reason); +handle_other({channel_exit, _Channel, E = {writer, send_failed, _E}}, State) -> + maybe_emit_stats(State), + throw(E); +handle_other({channel_exit, Channel, Reason}, State) -> + handle_exception(State, Channel, Reason); +handle_other({'DOWN', _MRef, process, ChPid, Reason}, State) -> + handle_dependent_exit(ChPid, Reason, State); +handle_other(terminate_connection, State) -> + maybe_emit_stats(State), + stop; +handle_other(handshake_timeout, State) + when ?IS_RUNNING(State) orelse ?IS_STOPPING(State) -> + State; +handle_other(handshake_timeout, State) -> + maybe_emit_stats(State), + throw({handshake_timeout, State#v1.callback}); +handle_other(heartbeat_timeout, State = #v1{connection_state = closed}) -> + State; +handle_other(heartbeat_timeout, + State = #v1{connection = #connection{timeout_sec = T}}) -> + maybe_emit_stats(State), + throw({heartbeat_timeout, T}); +handle_other({'$gen_call', From, {shutdown, Explanation}}, State) -> + {ForceTermination, NewState} = terminate(Explanation, State), + gen_server:reply(From, ok), + case ForceTermination of + force -> stop; + normal -> NewState + end; +handle_other({'$gen_call', From, info}, State) -> + gen_server:reply(From, infos(?INFO_KEYS, State)), + State; +handle_other({'$gen_call', From, {info, Items}}, State) -> + gen_server:reply(From, try {ok, infos(Items, State)} + catch Error -> {error, Error} + end), + State; +handle_other({'$gen_cast', {force_event_refresh, Ref}}, State) + when ?IS_RUNNING(State) -> + rabbit_event:notify( + connection_created, + [{type, network} | infos(?CREATION_EVENT_KEYS, State)], Ref), + rabbit_event:init_stats_timer(State, #v1.stats_timer); +handle_other({'$gen_cast', {force_event_refresh, _Ref}}, State) -> + %% Ignore, we will emit a created event once we start running. + State; +handle_other(ensure_stats, State) -> + ensure_stats_timer(State); +handle_other(emit_stats, State) -> + emit_stats(State); +handle_other({bump_credit, Msg}, State) -> + %% Here we are receiving credit by some channel process. + credit_flow:handle_bump_msg(Msg), + control_throttle(State); +handle_other(Other, State) -> + %% internal error -> something worth dying for + maybe_emit_stats(State), + exit({unexpected_message, Other}). + +switch_callback(State, Callback, Length) -> + State#v1{callback = Callback, recv_len = Length}. + +terminate(Explanation, State) when ?IS_RUNNING(State) -> + {normal, handle_exception(State, 0, + rabbit_misc:amqp_error( + connection_forced, Explanation, [], none))}; +terminate(_Explanation, State) -> + {force, State}. + +send_blocked(#v1{connection = #connection{protocol = Protocol, + capabilities = Capabilities}, + sock = Sock}, Reason) -> + case rabbit_misc:table_lookup(Capabilities, <<"connection.blocked">>) of + {bool, true} -> + + ok = send_on_channel0(Sock, #'connection.blocked'{reason = Reason}, + Protocol); + _ -> + ok + end. + +send_unblocked(#v1{connection = #connection{protocol = Protocol, + capabilities = Capabilities}, + sock = Sock}) -> + case rabbit_misc:table_lookup(Capabilities, <<"connection.blocked">>) of + {bool, true} -> + ok = send_on_channel0(Sock, #'connection.unblocked'{}, Protocol); + _ -> + ok + end. + +%%-------------------------------------------------------------------------- +%% error handling / termination + +close_connection(State = #v1{queue_collector = Collector, + connection = #connection{ + timeout_sec = TimeoutSec}}) -> + %% The spec says "Exclusive queues may only be accessed by the + %% current connection, and are deleted when that connection + %% closes." This does not strictly imply synchrony, but in + %% practice it seems to be what people assume. + clean_up_exclusive_queues(Collector), + %% We terminate the connection after the specified interval, but + %% no later than ?CLOSING_TIMEOUT seconds. + erlang:send_after((if TimeoutSec > 0 andalso + TimeoutSec < ?CLOSING_TIMEOUT -> TimeoutSec; + true -> ?CLOSING_TIMEOUT + end) * 1000, self(), terminate_connection), + State#v1{connection_state = closed}. + +%% queue collector will be undefined when connection +%% tuning was never performed or didn't finish. In such cases +%% there's also nothing to clean up. +clean_up_exclusive_queues(undefined) -> + ok; + +clean_up_exclusive_queues(Collector) -> + rabbit_queue_collector:delete_all(Collector). + +handle_dependent_exit(ChPid, Reason, State) -> + {Channel, State1} = channel_cleanup(ChPid, State), + case {Channel, termination_kind(Reason)} of + {undefined, controlled} -> State1; + {undefined, uncontrolled} -> handle_uncontrolled_channel_close(ChPid), + exit({abnormal_dependent_exit, + ChPid, Reason}); + {_, controlled} -> maybe_close(control_throttle(State1)); + {_, uncontrolled} -> handle_uncontrolled_channel_close(ChPid), + State2 = handle_exception( + State1, Channel, Reason), + maybe_close(control_throttle(State2)) + end. + +terminate_channels(#v1{channel_count = 0} = State) -> + State; +terminate_channels(#v1{channel_count = ChannelCount} = State) -> + lists:foreach(fun rabbit_channel:shutdown/1, all_channels()), + Timeout = 1000 * ?CHANNEL_TERMINATION_TIMEOUT * ChannelCount, + TimerRef = erlang:send_after(Timeout, self(), cancel_wait), + wait_for_channel_termination(ChannelCount, TimerRef, State). + +wait_for_channel_termination(0, TimerRef, State) -> + case erlang:cancel_timer(TimerRef) of + false -> receive + cancel_wait -> State + end; + _ -> State + end; +wait_for_channel_termination(N, TimerRef, + State = #v1{connection_state = CS, + connection = #connection{ + log_name = ConnName, + user = User, + vhost = VHost}, + sock = Sock}) -> + receive + {'DOWN', _MRef, process, ChPid, Reason} -> + {Channel, State1} = channel_cleanup(ChPid, State), + case {Channel, termination_kind(Reason)} of + {undefined, _} -> + exit({abnormal_dependent_exit, ChPid, Reason}); + {_, controlled} -> + wait_for_channel_termination(N-1, TimerRef, State1); + {_, uncontrolled} -> + rabbit_log_connection:error( + "Error on AMQP connection ~p (~s, vhost: '~s'," + " user: '~s', state: ~p), channel ~p:" + "error while terminating:~n~p~n", + [self(), ConnName, VHost, User#user.username, + CS, Channel, Reason]), + handle_uncontrolled_channel_close(ChPid), + wait_for_channel_termination(N-1, TimerRef, State1) + end; + {'EXIT', Sock, _Reason} -> + [channel_cleanup(ChPid, State) || ChPid <- all_channels()], + exit(normal); + cancel_wait -> + exit(channel_termination_timeout) + end. + +maybe_close(State = #v1{connection_state = closing, + channel_count = 0, + connection = #connection{protocol = Protocol}, + sock = Sock}) -> + NewState = close_connection(State), + ok = send_on_channel0(Sock, #'connection.close_ok'{}, Protocol), + NewState; +maybe_close(State) -> + State. + +termination_kind(normal) -> controlled; +termination_kind(_) -> uncontrolled. + +format_hard_error(#amqp_error{name = N, explanation = E, method = M}) -> + io_lib:format("operation ~s caused a connection exception ~s: ~p", [M, N, E]); +format_hard_error(Reason) -> + case io_lib:deep_char_list(Reason) of + true -> Reason; + false -> rabbit_misc:format("~p", [Reason]) + end. + +log_hard_error(#v1{connection_state = CS, + connection = #connection{ + log_name = ConnName, + user = User, + vhost = VHost}}, Channel, Reason) -> + rabbit_log_connection:error( + "Error on AMQP connection ~p (~s, vhost: '~s'," + " user: '~s', state: ~p), channel ~p:~n ~s~n", + [self(), ConnName, VHost, User#user.username, CS, Channel, format_hard_error(Reason)]). + +handle_exception(State = #v1{connection_state = closed}, Channel, Reason) -> + log_hard_error(State, Channel, Reason), + State; +handle_exception(State = #v1{connection = #connection{protocol = Protocol}, + connection_state = CS}, + Channel, Reason) + when ?IS_RUNNING(State) orelse CS =:= closing -> + respond_and_close(State, Channel, Protocol, Reason, Reason); +%% authentication failure +handle_exception(State = #v1{connection = #connection{protocol = Protocol, + log_name = ConnName, + capabilities = Capabilities}, + connection_state = starting}, + Channel, Reason = #amqp_error{name = access_refused, + explanation = ErrMsg}) -> + rabbit_log_connection:error( + "Error on AMQP connection ~p (~s, state: ~p):~n~s~n", + [self(), ConnName, starting, ErrMsg]), + %% respect authentication failure notification capability + case rabbit_misc:table_lookup(Capabilities, + <<"authentication_failure_close">>) of + {bool, true} -> + send_error_on_channel0_and_close(Channel, Protocol, Reason, State); + _ -> + close_connection(terminate_channels(State)) + end; +%% when loopback-only user tries to connect from a non-local host +%% when user tries to access a vhost it has no permissions for +handle_exception(State = #v1{connection = #connection{protocol = Protocol, + log_name = ConnName, + user = User}, + connection_state = opening}, + Channel, Reason = #amqp_error{name = not_allowed, + explanation = ErrMsg}) -> + rabbit_log_connection:error( + "Error on AMQP connection ~p (~s, user: '~s', state: ~p):~n~s~n", + [self(), ConnName, User#user.username, opening, ErrMsg]), + send_error_on_channel0_and_close(Channel, Protocol, Reason, State); +handle_exception(State = #v1{connection = #connection{protocol = Protocol}, + connection_state = CS = opening}, + Channel, Reason = #amqp_error{}) -> + respond_and_close(State, Channel, Protocol, Reason, + {handshake_error, CS, Reason}); +%% when negotiation fails, e.g. due to channel_max being higher than the +%% maxiumum allowed limit +handle_exception(State = #v1{connection = #connection{protocol = Protocol, + log_name = ConnName, + user = User}, + connection_state = tuning}, + Channel, Reason = #amqp_error{name = not_allowed, + explanation = ErrMsg}) -> + rabbit_log_connection:error( + "Error on AMQP connection ~p (~s," + " user: '~s', state: ~p):~n~s~n", + [self(), ConnName, User#user.username, tuning, ErrMsg]), + send_error_on_channel0_and_close(Channel, Protocol, Reason, State); +handle_exception(State, Channel, Reason) -> + %% We don't trust the client at this point - force them to wait + %% for a bit so they can't DOS us with repeated failed logins etc. + timer:sleep(?SILENT_CLOSE_DELAY * 1000), + throw({handshake_error, State#v1.connection_state, Channel, Reason}). + +%% we've "lost sync" with the client and hence must not accept any +%% more input +fatal_frame_error(Error, Type, Channel, Payload, State) -> + frame_error(Error, Type, Channel, Payload, State), + %% grace period to allow transmission of error + timer:sleep(?SILENT_CLOSE_DELAY * 1000), + throw(fatal_frame_error). + +frame_error(Error, Type, Channel, Payload, State) -> + {Str, Bin} = payload_snippet(Payload), + handle_exception(State, Channel, + rabbit_misc:amqp_error(frame_error, + "type ~p, ~s octets = ~p: ~p", + [Type, Str, Bin, Error], none)). + +unexpected_frame(Type, Channel, Payload, State) -> + {Str, Bin} = payload_snippet(Payload), + handle_exception(State, Channel, + rabbit_misc:amqp_error(unexpected_frame, + "type ~p, ~s octets = ~p", + [Type, Str, Bin], none)). + +payload_snippet(Payload) when size(Payload) =< 16 -> + {"all", Payload}; +payload_snippet(<<Snippet:16/binary, _/binary>>) -> + {"first 16", Snippet}. + +%%-------------------------------------------------------------------------- + +create_channel(_Channel, + #v1{channel_count = ChannelCount, + connection = #connection{channel_max = ChannelMax}}) + when ChannelMax /= 0 andalso ChannelCount >= ChannelMax -> + {error, rabbit_misc:amqp_error( + not_allowed, "number of channels opened (~w) has reached the " + "negotiated channel_max (~w)", + [ChannelCount, ChannelMax], 'none')}; +create_channel(Channel, + #v1{sock = Sock, + queue_collector = Collector, + channel_sup_sup_pid = ChanSupSup, + channel_count = ChannelCount, + connection = + #connection{name = Name, + protocol = Protocol, + frame_max = FrameMax, + user = User, + vhost = VHost, + capabilities = Capabilities}} = State) -> + {ok, _ChSupPid, {ChPid, AState}} = + rabbit_channel_sup_sup:start_channel( + ChanSupSup, {tcp, Sock, Channel, FrameMax, self(), Name, + Protocol, User, VHost, Capabilities, Collector}), + MRef = erlang:monitor(process, ChPid), + put({ch_pid, ChPid}, {Channel, MRef}), + put({channel, Channel}, {ChPid, AState}), + {ok, {ChPid, AState}, State#v1{channel_count = ChannelCount + 1}}. + +channel_cleanup(ChPid, State = #v1{channel_count = ChannelCount}) -> + case get({ch_pid, ChPid}) of + undefined -> {undefined, State}; + {Channel, MRef} -> credit_flow:peer_down(ChPid), + erase({channel, Channel}), + erase({ch_pid, ChPid}), + erlang:demonitor(MRef, [flush]), + {Channel, State#v1{channel_count = ChannelCount - 1}} + end. + +all_channels() -> [ChPid || {{ch_pid, ChPid}, _ChannelMRef} <- get()]. + +%%-------------------------------------------------------------------------- + +handle_frame(Type, 0, Payload, + State = #v1{connection = #connection{protocol = Protocol}}) + when ?IS_STOPPING(State) -> + case rabbit_command_assembler:analyze_frame(Type, Payload, Protocol) of + {method, MethodName, FieldsBin} -> + handle_method0(MethodName, FieldsBin, State); + _Other -> State + end; +handle_frame(Type, 0, Payload, + State = #v1{connection = #connection{protocol = Protocol}}) -> + case rabbit_command_assembler:analyze_frame(Type, Payload, Protocol) of + error -> frame_error(unknown_frame, Type, 0, Payload, State); + heartbeat -> State; + {method, MethodName, FieldsBin} -> + handle_method0(MethodName, FieldsBin, State); + _Other -> unexpected_frame(Type, 0, Payload, State) + end; +handle_frame(Type, Channel, Payload, + State = #v1{connection = #connection{protocol = Protocol}}) + when ?IS_RUNNING(State) -> + case rabbit_command_assembler:analyze_frame(Type, Payload, Protocol) of + error -> frame_error(unknown_frame, Type, Channel, Payload, State); + heartbeat -> unexpected_frame(Type, Channel, Payload, State); + Frame -> process_frame(Frame, Channel, State) + end; +handle_frame(_Type, _Channel, _Payload, State) when ?IS_STOPPING(State) -> + State; +handle_frame(Type, Channel, Payload, State) -> + unexpected_frame(Type, Channel, Payload, State). + +process_frame(Frame, Channel, State) -> + ChKey = {channel, Channel}, + case (case get(ChKey) of + undefined -> create_channel(Channel, State); + Other -> {ok, Other, State} + end) of + {error, Error} -> + handle_exception(State, Channel, Error); + {ok, {ChPid, AState}, State1} -> + case rabbit_command_assembler:process(Frame, AState) of + {ok, NewAState} -> + put(ChKey, {ChPid, NewAState}), + post_process_frame(Frame, ChPid, State1); + {ok, Method, NewAState} -> + rabbit_channel:do(ChPid, Method), + put(ChKey, {ChPid, NewAState}), + post_process_frame(Frame, ChPid, State1); + {ok, Method, Content, NewAState} -> + rabbit_channel:do_flow(ChPid, Method, Content), + put(ChKey, {ChPid, NewAState}), + post_process_frame(Frame, ChPid, control_throttle(State1)); + {error, Reason} -> + handle_exception(State1, Channel, Reason) + end + end. + +post_process_frame({method, 'channel.close_ok', _}, ChPid, State) -> + {_, State1} = channel_cleanup(ChPid, State), + %% This is not strictly necessary, but more obviously + %% correct. Also note that we do not need to call maybe_close/1 + %% since we cannot possibly be in the 'closing' state. + control_throttle(State1); +post_process_frame({content_header, _, _, _, _}, _ChPid, State) -> + publish_received(State); +post_process_frame({content_body, _}, _ChPid, State) -> + publish_received(State); +post_process_frame(_Frame, _ChPid, State) -> + State. + +%%-------------------------------------------------------------------------- + +%% We allow clients to exceed the frame size a little bit since quite +%% a few get it wrong - off-by 1 or 8 (empty frame size) are typical. +-define(FRAME_SIZE_FUDGE, ?EMPTY_FRAME_SIZE). + +handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32, _/binary>>, + State = #v1{connection = #connection{frame_max = FrameMax}}) + when FrameMax /= 0 andalso + PayloadSize > FrameMax - ?EMPTY_FRAME_SIZE + ?FRAME_SIZE_FUDGE -> + fatal_frame_error( + {frame_too_large, PayloadSize, FrameMax - ?EMPTY_FRAME_SIZE}, + Type, Channel, <<>>, State); +handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32, + Payload:PayloadSize/binary, ?FRAME_END, + Rest/binary>>, + State) -> + {Rest, ensure_stats_timer(handle_frame(Type, Channel, Payload, State))}; +handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32, Rest/binary>>, + State) -> + {Rest, ensure_stats_timer( + switch_callback(State, + {frame_payload, Type, Channel, PayloadSize}, + PayloadSize + 1))}; +handle_input({frame_payload, Type, Channel, PayloadSize}, Data, State) -> + <<Payload:PayloadSize/binary, EndMarker, Rest/binary>> = Data, + case EndMarker of + ?FRAME_END -> State1 = handle_frame(Type, Channel, Payload, State), + {Rest, switch_callback(State1, frame_header, 7)}; + _ -> fatal_frame_error({invalid_frame_end_marker, EndMarker}, + Type, Channel, Payload, State) + end; +handle_input(handshake, <<"AMQP", A, B, C, D, Rest/binary>>, State) -> + {Rest, handshake({A, B, C, D}, State)}; +handle_input(handshake, <<Other:8/binary, _/binary>>, #v1{sock = Sock}) -> + refuse_connection(Sock, {bad_header, Other}); +handle_input(Callback, Data, _State) -> + throw({bad_input, Callback, Data}). + +%% The two rules pertaining to version negotiation: +%% +%% * If the server cannot support the protocol specified in the +%% protocol header, it MUST respond with a valid protocol header and +%% then close the socket connection. +%% +%% * The server MUST provide a protocol version that is lower than or +%% equal to that requested by the client in the protocol header. +handshake({0, 0, 9, 1}, State) -> + start_connection({0, 9, 1}, rabbit_framing_amqp_0_9_1, State); + +%% This is the protocol header for 0-9, which we can safely treat as +%% though it were 0-9-1. +handshake({1, 1, 0, 9}, State) -> + start_connection({0, 9, 0}, rabbit_framing_amqp_0_9_1, State); + +%% This is what most clients send for 0-8. The 0-8 spec, confusingly, +%% defines the version as 8-0. +handshake({1, 1, 8, 0}, State) -> + start_connection({8, 0, 0}, rabbit_framing_amqp_0_8, State); + +%% The 0-8 spec as on the AMQP web site actually has this as the +%% protocol header; some libraries e.g., py-amqplib, send it when they +%% want 0-8. +handshake({1, 1, 9, 1}, State) -> + start_connection({8, 0, 0}, rabbit_framing_amqp_0_8, State); + +%% ... and finally, the 1.0 spec is crystal clear! +handshake({Id, 1, 0, 0}, State) -> + become_1_0(Id, State); + +handshake(Vsn, #v1{sock = Sock}) -> + refuse_connection(Sock, {bad_version, Vsn}). + +%% Offer a protocol version to the client. Connection.start only +%% includes a major and minor version number, Luckily 0-9 and 0-9-1 +%% are similar enough that clients will be happy with either. +start_connection({ProtocolMajor, ProtocolMinor, _ProtocolRevision}, + Protocol, + State = #v1{sock = Sock, connection = Connection}) -> + rabbit_networking:register_connection(self()), + Start = #'connection.start'{ + version_major = ProtocolMajor, + version_minor = ProtocolMinor, + server_properties = server_properties(Protocol), + mechanisms = auth_mechanisms_binary(Sock), + locales = <<"en_US">> }, + ok = send_on_channel0(Sock, Start, Protocol), + switch_callback(State#v1{connection = Connection#connection{ + timeout_sec = ?NORMAL_TIMEOUT, + protocol = Protocol}, + connection_state = starting}, + frame_header, 7). + +refuse_connection(Sock, Exception, {A, B, C, D}) -> + ok = inet_op(fun () -> rabbit_net:send(Sock, <<"AMQP",A,B,C,D>>) end), + throw(Exception). + +-spec refuse_connection(rabbit_net:socket(), any()) -> no_return(). + +refuse_connection(Sock, Exception) -> + refuse_connection(Sock, Exception, {0, 0, 9, 1}). + +ensure_stats_timer(State = #v1{connection_state = running}) -> + rabbit_event:ensure_stats_timer(State, #v1.stats_timer, emit_stats); +ensure_stats_timer(State) -> + State. + +%%-------------------------------------------------------------------------- + +handle_method0(MethodName, FieldsBin, + State = #v1{connection = #connection{protocol = Protocol}}) -> + try + handle_method0(Protocol:decode_method_fields(MethodName, FieldsBin), + State) + catch throw:{inet_error, E} when E =:= closed; E =:= enotconn -> + maybe_emit_stats(State), + throw({connection_closed_abruptly, State}); + exit:#amqp_error{method = none} = Reason -> + handle_exception(State, 0, Reason#amqp_error{method = MethodName}); + Type:Reason -> + Stack = erlang:get_stacktrace(), + handle_exception(State, 0, {Type, Reason, MethodName, Stack}) + end. + +handle_method0(#'connection.start_ok'{mechanism = Mechanism, + response = Response, + client_properties = ClientProperties}, + State0 = #v1{connection_state = starting, + connection = Connection0, + sock = Sock}) -> + AuthMechanism = auth_mechanism_to_module(Mechanism, Sock), + Capabilities = + case rabbit_misc:table_lookup(ClientProperties, <<"capabilities">>) of + {table, Capabilities1} -> Capabilities1; + _ -> [] + end, + Connection1 = Connection0#connection{ + client_properties = ClientProperties, + capabilities = Capabilities, + auth_mechanism = {Mechanism, AuthMechanism}, + auth_state = AuthMechanism:init(Sock)}, + Connection2 = augment_connection_log_name(Connection1), + State = State0#v1{connection_state = securing, + connection = Connection2}, + auth_phase(Response, State); + +handle_method0(#'connection.secure_ok'{response = Response}, + State = #v1{connection_state = securing}) -> + auth_phase(Response, State); + +handle_method0(#'connection.tune_ok'{frame_max = FrameMax, + channel_max = ChannelMax, + heartbeat = ClientHeartbeat}, + State = #v1{connection_state = tuning, + connection = Connection, + helper_sup = SupPid, + sock = Sock}) -> + ok = validate_negotiated_integer_value( + frame_max, ?FRAME_MIN_SIZE, FrameMax), + ok = validate_negotiated_integer_value( + channel_max, ?CHANNEL_MIN, ChannelMax), + {ok, Collector} = rabbit_connection_helper_sup:start_queue_collector( + SupPid, Connection#connection.name), + Frame = rabbit_binary_generator:build_heartbeat_frame(), + Parent = self(), + SendFun = + fun() -> + case catch rabbit_net:send(Sock, Frame) of + ok -> + ok; + {error, Reason} -> + Parent ! {heartbeat_send_error, Reason}; + Unexpected -> + Parent ! {heartbeat_send_error, Unexpected} + end, + ok + end, + ReceiveFun = fun() -> Parent ! heartbeat_timeout end, + Heartbeater = rabbit_heartbeat:start( + SupPid, Sock, Connection#connection.name, + ClientHeartbeat, SendFun, ClientHeartbeat, ReceiveFun), + State#v1{connection_state = opening, + connection = Connection#connection{ + frame_max = FrameMax, + channel_max = ChannelMax, + timeout_sec = ClientHeartbeat}, + queue_collector = Collector, + heartbeater = Heartbeater}; + +handle_method0(#'connection.open'{virtual_host = VHost}, + State = #v1{connection_state = opening, + connection = Connection = #connection{ + log_name = ConnName, + user = User = #user{username = Username}, + protocol = Protocol}, + helper_sup = SupPid, + sock = Sock, + throttle = Throttle}) -> + + ok = is_over_connection_limit(VHost, User), + ok = rabbit_access_control:check_vhost_access(User, VHost, Sock), + NewConnection = Connection#connection{vhost = VHost}, + ok = send_on_channel0(Sock, #'connection.open_ok'{}, Protocol), + + Alarms = rabbit_alarm:register(self(), {?MODULE, conserve_resources, []}), + BlockedBy = sets:from_list([{resource, Alarm} || Alarm <- Alarms]), + Throttle1 = Throttle#throttle{blocked_by = BlockedBy}, + + {ok, ChannelSupSupPid} = + rabbit_connection_helper_sup:start_channel_sup_sup(SupPid), + State1 = control_throttle( + State#v1{connection_state = running, + connection = NewConnection, + channel_sup_sup_pid = ChannelSupSupPid, + throttle = Throttle1}), + Infos = [{type, network} | infos(?CREATION_EVENT_KEYS, State1)], + rabbit_core_metrics:connection_created(proplists:get_value(pid, Infos), + Infos), + rabbit_event:notify(connection_created, Infos), + maybe_emit_stats(State1), + rabbit_log_connection:info( + "connection ~p (~s): " + "user '~s' authenticated and granted access to vhost '~s'~n", + [self(), dynamic_connection_name(ConnName), Username, VHost]), + State1; +handle_method0(#'connection.close'{}, State) when ?IS_RUNNING(State) -> + lists:foreach(fun rabbit_channel:shutdown/1, all_channels()), + maybe_close(State#v1{connection_state = closing}); +handle_method0(#'connection.close'{}, + State = #v1{connection = #connection{protocol = Protocol}, + sock = Sock}) + when ?IS_STOPPING(State) -> + %% We're already closed or closing, so we don't need to cleanup + %% anything. + ok = send_on_channel0(Sock, #'connection.close_ok'{}, Protocol), + State; +handle_method0(#'connection.close_ok'{}, + State = #v1{connection_state = closed}) -> + self() ! terminate_connection, + State; +handle_method0(_Method, State) when ?IS_STOPPING(State) -> + State; +handle_method0(_Method, #v1{connection_state = S}) -> + rabbit_misc:protocol_error( + channel_error, "unexpected method in connection state ~w", [S]). + +is_over_connection_limit(VHostPath, User) -> + try rabbit_vhost_limit:is_over_connection_limit(VHostPath) of + false -> ok; + {true, Limit} -> rabbit_misc:protocol_error(not_allowed, + "access to vhost '~s' refused for user '~s': " + "connection limit (~p) is reached", + [VHostPath, User#user.username, Limit]) + catch + throw:{error, {no_such_vhost, VHostPath}} -> + rabbit_misc:protocol_error(not_allowed, "vhost ~s not found", [VHostPath]) + end. + + +validate_negotiated_integer_value(Field, Min, ClientValue) -> + ServerValue = get_env(Field), + if ClientValue /= 0 andalso ClientValue < Min -> + fail_negotiation(Field, min, Min, ClientValue); + ServerValue /= 0 andalso (ClientValue =:= 0 orelse + ClientValue > ServerValue) -> + fail_negotiation(Field, max, ServerValue, ClientValue); + true -> + ok + end. + +%% keep dialyzer happy +-spec fail_negotiation(atom(), 'min' | 'max', integer(), integer()) -> + no_return(). +fail_negotiation(Field, MinOrMax, ServerValue, ClientValue) -> + {S1, S2} = case MinOrMax of + min -> {lower, minimum}; + max -> {higher, maximum} + end, + rabbit_misc:protocol_error( + not_allowed, "negotiated ~w = ~w is ~w than the ~w allowed value (~w)", + [Field, ClientValue, S1, S2, ServerValue], 'connection.tune'). + +get_env(Key) -> + {ok, Value} = application:get_env(rabbit, Key), + Value. + +send_on_channel0(Sock, Method, Protocol) -> + ok = rabbit_writer:internal_send_command(Sock, 0, Method, Protocol). + +auth_mechanism_to_module(TypeBin, Sock) -> + case rabbit_registry:binary_to_type(TypeBin) of + {error, not_found} -> + rabbit_misc:protocol_error( + command_invalid, "unknown authentication mechanism '~s'", + [TypeBin]); + T -> + case {lists:member(T, auth_mechanisms(Sock)), + rabbit_registry:lookup_module(auth_mechanism, T)} of + {true, {ok, Module}} -> + Module; + _ -> + rabbit_misc:protocol_error( + command_invalid, + "invalid authentication mechanism '~s'", [T]) + end + end. + +auth_mechanisms(Sock) -> + {ok, Configured} = application:get_env(auth_mechanisms), + [Name || {Name, Module} <- rabbit_registry:lookup_all(auth_mechanism), + Module:should_offer(Sock), lists:member(Name, Configured)]. + +auth_mechanisms_binary(Sock) -> + list_to_binary( + string:join([atom_to_list(A) || A <- auth_mechanisms(Sock)], " ")). + +auth_phase(Response, + State = #v1{connection = Connection = + #connection{protocol = Protocol, + auth_mechanism = {Name, AuthMechanism}, + auth_state = AuthState}, + sock = Sock}) -> + case AuthMechanism:handle_response(Response, AuthState) of + {refused, Username, Msg, Args} -> + auth_fail(Username, Msg, Args, Name, State); + {protocol_error, Msg, Args} -> + notify_auth_result(none, user_authentication_failure, + [{error, rabbit_misc:format(Msg, Args)}], + State), + rabbit_misc:protocol_error(syntax_error, Msg, Args); + {challenge, Challenge, AuthState1} -> + Secure = #'connection.secure'{challenge = Challenge}, + ok = send_on_channel0(Sock, Secure, Protocol), + State#v1{connection = Connection#connection{ + auth_state = AuthState1}}; + {ok, User = #user{username = Username}} -> + case rabbit_access_control:check_user_loopback(Username, Sock) of + ok -> + notify_auth_result(Username, user_authentication_success, + [], State); + not_allowed -> + auth_fail(Username, "user '~s' can only connect via " + "localhost", [Username], Name, State) + end, + Tune = #'connection.tune'{frame_max = get_env(frame_max), + channel_max = get_env(channel_max), + heartbeat = get_env(heartbeat)}, + ok = send_on_channel0(Sock, Tune, Protocol), + State#v1{connection_state = tuning, + connection = Connection#connection{user = User, + auth_state = none}} + end. + +-spec auth_fail + (rabbit_types:username() | none, string(), [any()], binary(), #v1{}) -> + no_return(). + +auth_fail(Username, Msg, Args, AuthName, + State = #v1{connection = #connection{protocol = Protocol, + capabilities = Capabilities}}) -> + notify_auth_result(Username, user_authentication_failure, + [{error, rabbit_misc:format(Msg, Args)}], State), + AmqpError = rabbit_misc:amqp_error( + access_refused, "~s login refused: ~s", + [AuthName, io_lib:format(Msg, Args)], none), + case rabbit_misc:table_lookup(Capabilities, + <<"authentication_failure_close">>) of + {bool, true} -> + SafeMsg = io_lib:format( + "Login was refused using authentication " + "mechanism ~s. For details see the broker " + "logfile.", [AuthName]), + AmqpError1 = AmqpError#amqp_error{explanation = SafeMsg}, + {0, CloseMethod} = rabbit_binary_generator:map_exception( + 0, AmqpError1, Protocol), + ok = send_on_channel0(State#v1.sock, CloseMethod, Protocol); + _ -> ok + end, + rabbit_misc:protocol_error(AmqpError). + +notify_auth_result(Username, AuthResult, ExtraProps, State) -> + EventProps = [{connection_type, network}, + {name, case Username of none -> ''; _ -> Username end}] ++ + [case Item of + name -> {connection_name, i(name, State)}; + _ -> {Item, i(Item, State)} + end || Item <- ?AUTH_NOTIFICATION_INFO_KEYS] ++ + ExtraProps, + rabbit_event:notify(AuthResult, [P || {_, V} = P <- EventProps, V =/= '']). + +%%-------------------------------------------------------------------------- + +infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items]. + +i(pid, #v1{}) -> self(); +i(node, #v1{}) -> node(); +i(SockStat, S) when SockStat =:= recv_oct; + SockStat =:= recv_cnt; + SockStat =:= send_oct; + SockStat =:= send_cnt; + SockStat =:= send_pend -> + socket_info(fun (Sock) -> rabbit_net:getstat(Sock, [SockStat]) end, + fun ([{_, I}]) -> I end, S); +i(ssl, #v1{sock = Sock}) -> rabbit_net:is_ssl(Sock); +i(ssl_protocol, S) -> ssl_info(fun ({P, _}) -> P end, S); +i(ssl_key_exchange, S) -> ssl_info(fun ({_, {K, _, _}}) -> K end, S); +i(ssl_cipher, S) -> ssl_info(fun ({_, {_, C, _}}) -> C end, S); +i(ssl_hash, S) -> ssl_info(fun ({_, {_, _, H}}) -> H end, S); +i(peer_cert_issuer, S) -> cert_info(fun rabbit_ssl:peer_cert_issuer/1, S); +i(peer_cert_subject, S) -> cert_info(fun rabbit_ssl:peer_cert_subject/1, S); +i(peer_cert_validity, S) -> cert_info(fun rabbit_ssl:peer_cert_validity/1, S); +i(channels, #v1{channel_count = ChannelCount}) -> ChannelCount; +i(state, #v1{connection_state = ConnectionState, + throttle = #throttle{blocked_by = Reasons, + last_blocked_at = T} = Throttle}) -> + %% not throttled by resource or other longer-term reasons + %% TODO: come up with a sensible function name + case sets:size(sets:del_element(flow, Reasons)) =:= 0 andalso + (credit_flow:blocked() %% throttled by flow now + orelse %% throttled by flow recently + (is_blocked_by_flow(Throttle) andalso T =/= never andalso + erlang:convert_time_unit(erlang:monotonic_time() - T, + native, + micro_seconds) < 5000000)) of + true -> flow; + false -> + case {has_reasons_to_block(Throttle), ConnectionState} of + %% blocked + {_, blocked} -> blocked; + %% not yet blocked (there were no publishes) + {true, running} -> blocking; + %% not blocked + {false, _} -> ConnectionState; + %% catch all to be defensive + _ -> ConnectionState + end + end; +i(garbage_collection, _State) -> + rabbit_misc:get_gc_info(self()); +i(reductions, _State) -> + {reductions, Reductions} = erlang:process_info(self(), reductions), + Reductions; +i(Item, #v1{connection = Conn}) -> ic(Item, Conn). + +ic(name, #connection{name = Name}) -> Name; +ic(host, #connection{host = Host}) -> Host; +ic(peer_host, #connection{peer_host = PeerHost}) -> PeerHost; +ic(port, #connection{port = Port}) -> Port; +ic(peer_port, #connection{peer_port = PeerPort}) -> PeerPort; +ic(protocol, #connection{protocol = none}) -> none; +ic(protocol, #connection{protocol = P}) -> P:version(); +ic(user, #connection{user = none}) -> ''; +ic(user, #connection{user = U}) -> U#user.username; +ic(user_who_performed_action, C) -> ic(user, C); +ic(vhost, #connection{vhost = VHost}) -> VHost; +ic(timeout, #connection{timeout_sec = Timeout}) -> Timeout; +ic(frame_max, #connection{frame_max = FrameMax}) -> FrameMax; +ic(channel_max, #connection{channel_max = ChMax}) -> ChMax; +ic(client_properties, #connection{client_properties = CP}) -> CP; +ic(auth_mechanism, #connection{auth_mechanism = none}) -> none; +ic(auth_mechanism, #connection{auth_mechanism = {Name, _Mod}}) -> Name; +ic(connected_at, #connection{connected_at = T}) -> T; +ic(Item, #connection{}) -> throw({bad_argument, Item}). + +socket_info(Get, Select, #v1{sock = Sock}) -> + case Get(Sock) of + {ok, T} -> case Select(T) of + N when is_number(N) -> N; + _ -> 0 + end; + {error, _} -> 0 + end. + +ssl_info(F, #v1{sock = Sock}) -> + case rabbit_net:ssl_info(Sock) of + nossl -> ''; + {error, _} -> ''; + {ok, Items} -> + P = proplists:get_value(protocol, Items), + CS = proplists:get_value(cipher_suite, Items), + %% The first form is R14. + %% The second is R13 - the extra term is exportability (by + %% inspection, the docs are wrong). + case CS of + {K, C, H} -> F({P, {K, C, H}}); + {K, C, H, _} -> F({P, {K, C, H}}) + end + end. + +cert_info(F, #v1{sock = Sock}) -> + case rabbit_net:peercert(Sock) of + nossl -> ''; + {error, _} -> ''; + {ok, Cert} -> list_to_binary(F(Cert)) + end. + +maybe_emit_stats(State) -> + rabbit_event:if_enabled(State, #v1.stats_timer, + fun() -> emit_stats(State) end). + +emit_stats(State) -> + [{_, Pid}, {_, Recv_oct}, {_, Send_oct}, {_, Reductions}] = I + = infos(?SIMPLE_METRICS, State), + Infos = infos(?OTHER_METRICS, State), + rabbit_core_metrics:connection_stats(Pid, Infos), + rabbit_core_metrics:connection_stats(Pid, Recv_oct, Send_oct, Reductions), + rabbit_event:notify(connection_stats, Infos ++ I), + State1 = rabbit_event:reset_stats_timer(State, #v1.stats_timer), + ensure_stats_timer(State1). + +%% 1.0 stub +-spec become_1_0(non_neg_integer(), #v1{}) -> no_return(). + +become_1_0(Id, State = #v1{sock = Sock}) -> + case code:is_loaded(rabbit_amqp1_0_reader) of + false -> refuse_connection(Sock, amqp1_0_plugin_not_enabled); + _ -> Mode = case Id of + 0 -> amqp; + 3 -> sasl; + _ -> refuse_connection( + Sock, {unsupported_amqp1_0_protocol_id, Id}, + {3, 1, 0, 0}) + end, + F = fun (_Deb, Buf, BufLen, S) -> + {rabbit_amqp1_0_reader, init, + [Mode, pack_for_1_0(Buf, BufLen, S)]} + end, + State#v1{connection_state = {become, F}} + end. + +pack_for_1_0(Buf, BufLen, #v1{parent = Parent, + sock = Sock, + recv_len = RecvLen, + pending_recv = PendingRecv, + helper_sup = SupPid, + proxy_socket = ProxySocket}) -> + {Parent, Sock, RecvLen, PendingRecv, SupPid, Buf, BufLen, ProxySocket}. + +respond_and_close(State, Channel, Protocol, Reason, LogErr) -> + log_hard_error(State, Channel, LogErr), + send_error_on_channel0_and_close(Channel, Protocol, Reason, State). + +send_error_on_channel0_and_close(Channel, Protocol, Reason, State) -> + {0, CloseMethod} = + rabbit_binary_generator:map_exception(Channel, Reason, Protocol), + State1 = close_connection(terminate_channels(State)), + ok = send_on_channel0(State#v1.sock, CloseMethod, Protocol), + State1. + +%% +%% Publisher throttling +%% + +blocked_by_message(#throttle{blocked_by = Reasons}) -> + %% we don't want to report internal flow as a reason here since + %% it is entirely transient + Reasons1 = sets:del_element(flow, Reasons), + RStr = string:join([format_blocked_by(R) || R <- sets:to_list(Reasons1)], " & "), + list_to_binary(rabbit_misc:format("low on ~s", [RStr])). + +format_blocked_by({resource, memory}) -> "memory"; +format_blocked_by({resource, disk}) -> "disk"; +format_blocked_by({resource, disc}) -> "disk". + +update_last_blocked_at(Throttle) -> + Throttle#throttle{last_blocked_at = erlang:monotonic_time()}. + +connection_blocked_message_sent( + #throttle{connection_blocked_message_sent = BS}) -> BS. + +should_send_blocked(Throttle = #throttle{blocked_by = Reasons}) -> + should_block(Throttle) + andalso + sets:size(sets:del_element(flow, Reasons)) =/= 0 + andalso + not connection_blocked_message_sent(Throttle). + +should_send_unblocked(Throttle = #throttle{blocked_by = Reasons}) -> + connection_blocked_message_sent(Throttle) + andalso + sets:size(sets:del_element(flow, Reasons)) == 0. + +%% Returns true if we have a reason to block +%% this connection. +has_reasons_to_block(#throttle{blocked_by = Reasons}) -> + sets:size(Reasons) > 0. + +is_blocked_by_flow(#throttle{blocked_by = Reasons}) -> + sets:is_element(flow, Reasons). + +should_block(#throttle{should_block = Val}) -> Val. + +should_block_connection(Throttle) -> + should_block(Throttle) andalso has_reasons_to_block(Throttle). + +should_unblock_connection(Throttle) -> + not should_block_connection(Throttle). + +maybe_block(State = #v1{connection_state = CS, throttle = Throttle}) -> + case should_block_connection(Throttle) of + true -> + State1 = State#v1{connection_state = blocked, + throttle = update_last_blocked_at(Throttle)}, + case CS of + running -> + ok = rabbit_heartbeat:pause_monitor(State#v1.heartbeater); + _ -> ok + end, + maybe_send_blocked_or_unblocked(State1); + false -> State + end. + +maybe_unblock(State = #v1{throttle = Throttle}) -> + case should_unblock_connection(Throttle) of + true -> + ok = rabbit_heartbeat:resume_monitor(State#v1.heartbeater), + State1 = State#v1{connection_state = running, + throttle = Throttle#throttle{should_block = false}}, + maybe_send_unblocked(State1); + false -> State + end. + +maybe_send_unblocked(State = #v1{throttle = Throttle}) -> + case should_send_unblocked(Throttle) of + true -> + ok = send_unblocked(State), + State#v1{throttle = + Throttle#throttle{connection_blocked_message_sent = false}}; + false -> State + end. + +maybe_send_blocked_or_unblocked(State = #v1{throttle = Throttle}) -> + case should_send_blocked(Throttle) of + true -> + ok = send_blocked(State, blocked_by_message(Throttle)), + State#v1{throttle = + Throttle#throttle{connection_blocked_message_sent = true}}; + false -> maybe_send_unblocked(State) + end. + +publish_received(State = #v1{throttle = Throttle}) -> + case has_reasons_to_block(Throttle) of + false -> State; + true -> + Throttle1 = Throttle#throttle{should_block = true}, + maybe_block(State#v1{throttle = Throttle1}) + end. + +control_throttle(State = #v1{connection_state = CS, + throttle = #throttle{blocked_by = Reasons} = Throttle}) -> + Throttle1 = case credit_flow:blocked() of + true -> + Throttle#throttle{blocked_by = sets:add_element(flow, Reasons)}; + false -> + Throttle#throttle{blocked_by = sets:del_element(flow, Reasons)} + end, + State1 = State#v1{throttle = Throttle1}, + case CS of + running -> maybe_block(State1); + %% unblock or re-enable blocking + blocked -> maybe_block(maybe_unblock(State1)); + _ -> State1 + end. + +augment_connection_log_name(#connection{client_properties = ClientProperties, + name = Name} = Connection) -> + case rabbit_misc:table_lookup(ClientProperties, <<"connection_name">>) of + {longstr, UserSpecifiedName} -> + LogName = <<Name/binary, " - ", UserSpecifiedName/binary>>, + rabbit_log_connection:info("Connection ~p (~s) has a client-provided name: ~s~n", [self(), Name, UserSpecifiedName]), + ?store_proc_name(LogName), + Connection#connection{log_name = LogName}; + _ -> + Connection + end. + +dynamic_connection_name(Default) -> + case rabbit_misc:get_proc_name() of + {ok, Name} -> + Name; + _ -> + Default + end. + +handle_uncontrolled_channel_close(ChPid) -> + rabbit_core_metrics:channel_closed(ChPid), + rabbit_event:notify(channel_closed, [{pid, ChPid}]). diff --git a/src/rabbit_resource_monitor_misc.erl b/src/rabbit_resource_monitor_misc.erl deleted file mode 100644 index 8d743b9a24..0000000000 --- a/src/rabbit_resource_monitor_misc.erl +++ /dev/null @@ -1,47 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - - --module(rabbit_resource_monitor_misc). - --export([parse_information_unit/1]). - --spec parse_information_unit(integer() | string()) -> - {ok, integer()} | {error, parse_error}. - -parse_information_unit(Value) when is_integer(Value) -> {ok, Value}; -parse_information_unit(Value) when is_list(Value) -> - case re:run(Value, - "^(?<VAL>[0-9]+)(?<UNIT>kB|KB|MB|GB|kb|mb|gb|Kb|Mb|Gb|kiB|KiB|MiB|GiB|kib|mib|gib|KIB|MIB|GIB|k|K|m|M|g|G)?$", - [{capture, all_but_first, list}]) of - {match, [[], _]} -> - {ok, list_to_integer(Value)}; - {match, [Num]} -> - {ok, list_to_integer(Num)}; - {match, [Num, Unit]} -> - Multiplier = case Unit of - KiB when KiB =:= "k"; KiB =:= "kiB"; KiB =:= "K"; KiB =:= "KIB"; KiB =:= "kib" -> 1024; - MiB when MiB =:= "m"; MiB =:= "MiB"; MiB =:= "M"; MiB =:= "MIB"; MiB =:= "mib" -> 1024*1024; - GiB when GiB =:= "g"; GiB =:= "GiB"; GiB =:= "G"; GiB =:= "GIB"; GiB =:= "gib" -> 1024*1024*1024; - KB when KB =:= "KB"; KB =:= "kB"; KB =:= "kb"; KB =:= "Kb" -> 1000; - MB when MB =:= "MB"; MB =:= "mB"; MB =:= "mb"; MB =:= "Mb" -> 1000000; - GB when GB =:= "GB"; GB =:= "gB"; GB =:= "gb"; GB =:= "Gb" -> 1000000000 - end, - {ok, list_to_integer(Num) * Multiplier}; - nomatch -> - % log error - {error, parse_error} - end. diff --git a/src/rabbit_vhost.erl b/src/rabbit_vhost.erl index 1d1ea16cca..55c79d4ead 100644 --- a/src/rabbit_vhost.erl +++ b/src/rabbit_vhost.erl @@ -21,7 +21,7 @@ %%---------------------------------------------------------------------------- -export([recover/0, recover/1]). --export([add/2, delete/2, exists/1, list/0, with/2, assert/1, update/2, +-export([add/2, delete/2, exists/1, list/0, with/2, with_user_and_vhost/3, assert/1, update/2, set_limits/2, limits_of/1]). -export([info/1, info/2, info_all/0, info_all/1, info_all/2, info_all/3]). -export([dir/1, msg_store_dir_path/1, msg_store_dir_wildcard/0]). @@ -33,6 +33,8 @@ -spec exists(rabbit_types:vhost()) -> boolean(). -spec list() -> [rabbit_types:vhost()]. -spec with(rabbit_types:vhost(), rabbit_misc:thunk(A)) -> A. +-spec with_user_and_vhost + (rabbit_types:username(), rabbit_types:vhost(), rabbit_misc:thunk(A)) -> A. -spec assert(rabbit_types:vhost()) -> 'ok'. -spec info(rabbit_types:vhost()) -> rabbit_types:infos(). @@ -179,6 +181,9 @@ with(VHostPath, Thunk) -> end end. +with_user_and_vhost(Username, VHostPath, Thunk) -> + rabbit_misc:with_user(Username, with(VHostPath, Thunk)). + %% Like with/2 but outside an Mnesia tx assert(VHostPath) -> case exists(VHostPath) of true -> ok; diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl deleted file mode 100644 index b07ff27f17..0000000000 --- a/src/vm_memory_monitor.erl +++ /dev/null @@ -1,540 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - -%% 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. - --module(vm_memory_monitor). - --behaviour(gen_server). - --export([start_link/1, start_link/3]). - --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - --export([get_total_memory/0, get_vm_limit/0, - get_check_interval/0, set_check_interval/1, - get_vm_memory_high_watermark/0, set_vm_memory_high_watermark/1, - get_memory_limit/0, get_memory_use/1, - get_process_memory/0]). - -%% for tests --export([parse_line_linux/1]). - - --define(SERVER, ?MODULE). --define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000). --define(ONE_MiB, 1048576). - -%% 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). --define(DEFAULT_VM_MEMORY_HIGH_WATERMARK, 0.4). - --record(state, {total_memory, - memory_limit, - memory_config_limit, - timeout, - timer, - alarmed, - alarm_funs - }). - -%%---------------------------------------------------------------------------- - --type vm_memory_high_watermark() :: (float() | {'absolute', integer() | string()}). --spec start_link(float()) -> rabbit_types:ok_pid_or_error(). --spec start_link(float(), fun ((any()) -> 'ok'), - fun ((any()) -> 'ok')) -> rabbit_types:ok_pid_or_error(). --spec get_total_memory() -> (non_neg_integer() | 'unknown'). --spec get_vm_limit() -> non_neg_integer(). --spec get_check_interval() -> non_neg_integer(). --spec set_check_interval(non_neg_integer()) -> 'ok'. --spec get_vm_memory_high_watermark() -> vm_memory_high_watermark(). --spec set_vm_memory_high_watermark(vm_memory_high_watermark()) -> 'ok'. --spec get_memory_limit() -> non_neg_integer(). --spec get_memory_use(bytes) -> {non_neg_integer(), float() | infinity}; - (ratio) -> float() | infinity. - -%%---------------------------------------------------------------------------- -%% Public API -%%---------------------------------------------------------------------------- - -get_total_memory() -> - case application:get_env(rabbit, total_memory_available_override_value) of - {ok, Value} -> - case rabbit_resource_monitor_misc:parse_information_unit(Value) of - {ok, ParsedTotal} -> - ParsedTotal; - {error, parse_error} -> - rabbit_log:warning( - "The override value for the total memmory available is " - "not a valid value: ~p, getting total from the system.~n", - [Value]), - get_total_memory_from_os() - end; - undefined -> - get_total_memory_from_os() - end. - -get_vm_limit() -> get_vm_limit(os:type()). - -get_check_interval() -> - gen_server:call(?MODULE, get_check_interval, infinity). - -set_check_interval(Fraction) -> - gen_server:call(?MODULE, {set_check_interval, Fraction}, infinity). - -get_vm_memory_high_watermark() -> - gen_server:call(?MODULE, get_vm_memory_high_watermark, infinity). - -set_vm_memory_high_watermark(Fraction) -> - gen_server:call(?MODULE, {set_vm_memory_high_watermark, Fraction}, - infinity). - -get_memory_limit() -> - gen_server:call(?MODULE, get_memory_limit, infinity). - -get_memory_use(bytes) -> - MemoryLimit = get_memory_limit(), - {get_process_memory(), case MemoryLimit > 0.0 of - true -> MemoryLimit; - false -> infinity - end}; -get_memory_use(ratio) -> - MemoryLimit = get_memory_limit(), - case MemoryLimit > 0.0 of - true -> get_process_memory() / MemoryLimit; - false -> infinity - end. - -%% Memory reported by erlang:memory(total) is not supposed to -%% be equal to the total size of all pages mapped to the emulator, -%% according to http://erlang.org/doc/man/erlang.html#memory-0 -%% erlang:memory(total) under-reports memory usage by around 20% --spec get_process_memory() -> Bytes :: integer(). -get_process_memory() -> - case get_memory_calculation_strategy() of - rss -> - case get_system_process_resident_memory() of - {ok, MemInBytes} -> - MemInBytes; - {error, Reason} -> - rabbit_log:debug("Unable to get system memory used. Reason: ~p." - " Falling back to erlang memory reporting", - [Reason]), - erlang:memory(total) - end; - erlang -> - erlang:memory(total) - end. - --spec get_memory_calculation_strategy() -> rss | erlang. -get_memory_calculation_strategy() -> - case application:get_env(rabbit, vm_memory_calculation_strategy, rss) of - erlang -> - erlang; - rss -> - rss; - UnsupportedValue -> - rabbit_log:warning( - "Unsupported value '~p' for vm_memory_calculation_strategy. " - "Supported values: (rss|erlang). " - "Defaulting to 'rss'", - [UnsupportedValue] - ), - rss - end. - --spec get_system_process_resident_memory() -> {ok, Bytes :: integer()} | {error, term()}. -get_system_process_resident_memory() -> - try - get_system_process_resident_memory(os:type()) - catch _:Error -> - {error, {"Failed to get process resident memory", Error}} - end. - -get_system_process_resident_memory({unix,darwin}) -> - get_ps_memory(); - -get_system_process_resident_memory({unix, linux}) -> - get_ps_memory(); - -get_system_process_resident_memory({unix,freebsd}) -> - get_ps_memory(); - -get_system_process_resident_memory({unix,openbsd}) -> - get_ps_memory(); - -get_system_process_resident_memory({win32,_OSname}) -> - OsPid = os:getpid(), - Cmd = "wmic process where processid=" ++ OsPid ++ " get WorkingSetSize /value 2>&1", - CmdOutput = os:cmd(Cmd), - %% Memory usage is displayed in bytes - case re:run(CmdOutput, "WorkingSetSize=([0-9]+)", [{capture, all_but_first, binary}]) of - {match, [Match]} -> - {ok, binary_to_integer(Match)}; - _ -> - {error, {unexpected_output_from_command, Cmd, CmdOutput}} - end; - -get_system_process_resident_memory({unix, sunos}) -> - get_ps_memory(); - -get_system_process_resident_memory({unix, aix}) -> - get_ps_memory(); - -get_system_process_resident_memory(_OsType) -> - {error, not_implemented_for_os}. - -get_ps_memory() -> - OsPid = os:getpid(), - Cmd = "ps -p " ++ OsPid ++ " -o rss=", - CmdOutput = os:cmd(Cmd), - case re:run(CmdOutput, "[0-9]+", [{capture, first, list}]) of - {match, [Match]} -> - {ok, list_to_integer(Match) * 1024}; - _ -> - {error, {unexpected_output_from_command, Cmd, CmdOutput}} - end. - -%%---------------------------------------------------------------------------- -%% gen_server callbacks -%%---------------------------------------------------------------------------- - -start_link(MemFraction) -> - start_link(MemFraction, - fun alarm_handler:set_alarm/1, fun alarm_handler:clear_alarm/1). - -start_link(MemFraction, AlarmSet, AlarmClear) -> - gen_server:start_link({local, ?SERVER}, ?MODULE, - [MemFraction, {AlarmSet, AlarmClear}], []). - -init([MemFraction, AlarmFuns]) -> - TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL), - State = #state { timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL, - timer = TRef, - alarmed = false, - alarm_funs = AlarmFuns }, - {ok, set_mem_limits(State, MemFraction)}. - -handle_call(get_vm_memory_high_watermark, _From, - #state{memory_config_limit = MemLimit} = State) -> - {reply, MemLimit, State}; - -handle_call({set_vm_memory_high_watermark, MemLimit}, _From, State) -> - {reply, ok, set_mem_limits(State, 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(get_memory_limit, _From, State) -> - {reply, State#state.memory_limit, State}; - -handle_call(_Request, _From, State) -> - {noreply, State}. - -handle_cast(_Request, State) -> - {noreply, State}. - -handle_info(update, State) -> - {noreply, internal_update(State)}; - -handle_info(_Info, State) -> - {noreply, State}. - -terminate(_Reason, _State) -> - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -%%---------------------------------------------------------------------------- -%% Server Internals -%%---------------------------------------------------------------------------- -get_total_memory_from_os() -> - try - get_total_memory(os:type()) - catch _:Error -> - rabbit_log:warning( - "Failed to get total system memory: ~n~p~n~p~n", - [Error, erlang:get_stacktrace()]), - unknown - end. - -set_mem_limits(State, MemLimit) -> - case erlang:system_info(wordsize) of - 4 -> - error_logger:warning_msg( - "You are using a 32-bit version of Erlang: you may run into " - "memory address~n" - "space exhaustion or statistic counters overflow.~n"); - _ -> - ok - end, - TotalMemory = - case get_total_memory() of - unknown -> - case State of - #state { total_memory = undefined, - memory_limit = undefined } -> - error_logger:warning_msg( - "Unknown total memory size for your OS ~p. " - "Assuming memory size is ~p MiB (~p bytes).~n", - [os:type(), - trunc(?MEMORY_SIZE_FOR_UNKNOWN_OS/?ONE_MiB), - ?MEMORY_SIZE_FOR_UNKNOWN_OS]); - _ -> - ok - end, - ?MEMORY_SIZE_FOR_UNKNOWN_OS; - M -> M - end, - UsableMemory = - case get_vm_limit() of - Limit when Limit < TotalMemory -> - error_logger:warning_msg( - "Only ~p MiB (~p bytes) of ~p MiB (~p bytes) memory usable due to " - "limited address space.~n" - "Crashes due to memory exhaustion are possible - see~n" - "http://www.rabbitmq.com/memory.html#address-space~n", - [trunc(Limit/?ONE_MiB), Limit, trunc(TotalMemory/?ONE_MiB), - TotalMemory]), - Limit; - _ -> - TotalMemory - end, - MemLim = interpret_limit(parse_mem_limit(MemLimit), UsableMemory), - error_logger:info_msg("Memory limit set to ~p MiB (~p bytes) of ~p MiB (~p bytes) total.~n", - [trunc(MemLim/?ONE_MiB), MemLim, trunc(TotalMemory/?ONE_MiB), - TotalMemory]), - internal_update(State #state { total_memory = TotalMemory, - memory_limit = MemLim, - memory_config_limit = MemLimit}). - -interpret_limit({'absolute', MemLim}, UsableMemory) -> - erlang:min(MemLim, UsableMemory); -interpret_limit(MemFraction, UsableMemory) -> - trunc(MemFraction * UsableMemory). - - -parse_mem_limit({absolute, Limit}) -> - case rabbit_resource_monitor_misc:parse_information_unit(Limit) of - {ok, ParsedLimit} -> {absolute, ParsedLimit}; - {error, parse_error} -> - rabbit_log:error("Unable to parse vm_memory_high_watermark value ~p", [Limit]), - ?DEFAULT_VM_MEMORY_HIGH_WATERMARK - end; -parse_mem_limit(Relative) when is_float(Relative), Relative < 1 -> - Relative; -parse_mem_limit(_) -> - ?DEFAULT_VM_MEMORY_HIGH_WATERMARK. - - -internal_update(State = #state { memory_limit = MemLimit, - alarmed = Alarmed, - alarm_funs = {AlarmSet, AlarmClear} }) -> - MemUsed = get_process_memory(), - NewAlarmed = MemUsed > MemLimit, - case {Alarmed, NewAlarmed} of - {false, true} -> emit_update_info(set, MemUsed, MemLimit), - AlarmSet({{resource_limit, memory, node()}, []}); - {true, false} -> emit_update_info(clear, MemUsed, MemLimit), - AlarmClear({resource_limit, memory, node()}); - _ -> ok - end, - State #state {alarmed = NewAlarmed}. - -emit_update_info(AlarmState, MemUsed, MemLimit) -> - error_logger:info_msg( - "vm_memory_high_watermark ~p. Memory used:~p allowed:~p~n", - [AlarmState, MemUsed, MemLimit]). - -start_timer(Timeout) -> - {ok, TRef} = timer:send_interval(Timeout, update), - TRef. - -%% According to http://msdn.microsoft.com/en-us/library/aa366778(VS.85).aspx -%% Windows has 2GB and 8TB of address space for 32 and 64 bit accordingly. -get_vm_limit({win32,_OSname}) -> - case erlang:system_info(wordsize) of - 4 -> 2*1024*1024*1024; %% 2 GB for 32 bits 2^31 - 8 -> 8*1024*1024*1024*1024 %% 8 TB for 64 bits 2^42 - end; - -%% On a 32-bit machine, if you're using more than 2 gigs of RAM you're -%% in big trouble anyway. -get_vm_limit(_OsType) -> - case erlang:system_info(wordsize) of - 4 -> 2*1024*1024*1024; %% 2 GB for 32 bits 2^31 - 8 -> 256*1024*1024*1024*1024 %% 256 TB for 64 bits 2^48 - %%http://en.wikipedia.org/wiki/X86-64#Virtual_address_space_details - end. - -%%---------------------------------------------------------------------------- -%% 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 = sysctl("vm.stats.vm.v_page_size"), - PageCount = sysctl("vm.stats.vm.v_page_count"), - PageCount * PageSize; - -get_total_memory({unix,openbsd}) -> - sysctl("hw.usermem"); - -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({unix, aix}) -> - File = cmd("/usr/bin/vmstat -v"), - Lines = string:tokens(File, "\n"), - Dict = dict:from_list(lists:map(fun parse_line_aix/1, Lines)), - dict:fetch('memory pages', Dict) * 4096; - -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 "MemTotal: 502968 kB" -%% or (with broken OS/modules) "Readahead 123456 kB" -parse_line_linux(Line) -> - {Name, Value, UnitRest} = - case string:tokens(Line, ":") of - %% no colon in the line - [S] -> - [K, RHS] = re:split(S, "\s", [{parts, 2}, {return, list}]), - [V | Unit] = string:tokens(RHS, " "), - {K, V, Unit}; - [K, RHS | _Rest] -> - [V | Unit] = string:tokens(RHS, " "), - {K, V, Unit} - end, - Value1 = case UnitRest 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) * ?ONE_MiB * 1024; - ["Megabytes"] -> - list_to_integer(Value1) * ?ONE_MiB; - ["Kilobytes"] -> - list_to_integer(Value1) * 1024; - _ -> - Value1 ++ UnitsRest %% no known units - end, - {list_to_atom(Name), Value2}; - [Name] -> {list_to_atom(Name), none} - end. - -%% Lines look like " 12345 memory pages" -%% or " 80.1 maxpin percentage" -parse_line_aix(Line) -> - [Value | NameWords] = string:tokens(Line, " "), - Name = string:join(NameWords, " "), - {list_to_atom(Name), - case lists:member($., Value) of - true -> trunc(list_to_float(Value)); - false -> list_to_integer(Value) - end}. - -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. diff --git a/src/worker_pool.erl b/src/worker_pool.erl deleted file mode 100644 index 71ed2a359a..0000000000 --- a/src/worker_pool.erl +++ /dev/null @@ -1,172 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - --module(worker_pool). - -%% Generic worker pool manager. -%% -%% Submitted jobs are functions. They can be executed synchronously -%% (using worker_pool:submit/1, worker_pool:submit/2) or asynchronously -%% (using worker_pool:submit_async/1). -%% -%% We typically use the worker pool if we want to limit the maximum -%% parallelism of some job. We are not trying to dodge the cost of -%% creating Erlang processes. -%% -%% Supports nested submission of jobs and two execution modes: -%% 'single' and 'reuse'. Jobs executed in 'single' mode are invoked in -%% a one-off process. Those executed in 'reuse' mode are invoked in a -%% worker process out of the pool. Nested jobs are always executed -%% immediately in current worker process. -%% -%% 'single' mode is offered to work around a bug in Mnesia: after -%% network partitions reply messages for prior failed requests can be -%% sent to Mnesia clients - a reused worker pool process can crash on -%% receiving one. -%% -%% Caller submissions are enqueued internally. When the next worker -%% process is available, it communicates it to the pool and is -%% assigned a job to execute. If job execution fails with an error, no -%% response is returned to the caller. -%% -%% Worker processes prioritise certain command-and-control messages -%% from the pool. -%% -%% Future improvement points: job prioritisation. - --behaviour(gen_server2). - --export([start_link/1, - submit/1, submit/2, submit/3, - submit_async/1, submit_async/2, - ready/2, - idle/2, - default_pool/0]). - --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - -%%---------------------------------------------------------------------------- - --type mfargs() :: {atom(), atom(), [any()]}. - --spec start_link(atom()) -> {'ok', pid()} | {'error', any()}. --spec submit(fun (() -> A) | mfargs()) -> A. --spec submit(fun (() -> A) | mfargs(), 'reuse' | 'single') -> A. --spec submit(atom(), fun (() -> A) | mfargs(), 'reuse' | 'single') -> A. --spec submit_async(fun (() -> any()) | mfargs()) -> 'ok'. --spec ready(atom(), pid()) -> 'ok'. --spec idle(atom(), pid()) -> 'ok'. --spec default_pool() -> atom(). - -%%---------------------------------------------------------------------------- - --define(DEFAULT_POOL, ?MODULE). --define(HIBERNATE_AFTER_MIN, 1000). --define(DESIRED_HIBERNATE, 10000). - --record(state, { available, pending }). - -%%---------------------------------------------------------------------------- - -start_link(Name) -> gen_server2:start_link({local, Name}, ?MODULE, [], - [{timeout, infinity}]). - -submit(Fun) -> - submit(?DEFAULT_POOL, Fun, reuse). - -%% ProcessModel =:= single is for working around the mnesia_locker bug. -submit(Fun, ProcessModel) -> - submit(?DEFAULT_POOL, Fun, ProcessModel). - -submit(Server, Fun, ProcessModel) -> - case get(worker_pool_worker) of - true -> worker_pool_worker:run(Fun); - _ -> Pid = gen_server2:call(Server, {next_free, self()}, infinity), - worker_pool_worker:submit(Pid, Fun, ProcessModel) - end. - -submit_async(Fun) -> submit_async(?DEFAULT_POOL, Fun). - -submit_async(Server, Fun) -> gen_server2:cast(Server, {run_async, Fun}). - -ready(Server, WPid) -> gen_server2:cast(Server, {ready, WPid}). - -idle(Server, WPid) -> gen_server2:cast(Server, {idle, WPid}). - -default_pool() -> ?DEFAULT_POOL. - -%%---------------------------------------------------------------------------- - -init([]) -> - {ok, #state { pending = queue:new(), available = ordsets:new() }, hibernate, - {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. - -handle_call({next_free, CPid}, From, State = #state { available = [], - pending = Pending }) -> - {noreply, State#state{pending = queue:in({next_free, From, CPid}, Pending)}, - hibernate}; -handle_call({next_free, CPid}, _From, State = #state { available = - [WPid | Avail1] }) -> - worker_pool_worker:next_job_from(WPid, CPid), - {reply, WPid, State #state { available = Avail1 }, hibernate}; - -handle_call(Msg, _From, State) -> - {stop, {unexpected_call, Msg}, State}. - -handle_cast({ready, WPid}, State) -> - erlang:monitor(process, WPid), - handle_cast({idle, WPid}, State); - -handle_cast({idle, WPid}, State = #state { available = Avail, - pending = Pending }) -> - {noreply, - case queue:out(Pending) of - {empty, _Pending} -> - State #state { available = ordsets:add_element(WPid, Avail) }; - {{value, {next_free, From, CPid}}, Pending1} -> - worker_pool_worker:next_job_from(WPid, CPid), - gen_server2:reply(From, WPid), - State #state { pending = Pending1 }; - {{value, {run_async, Fun}}, Pending1} -> - worker_pool_worker:submit_async(WPid, Fun), - State #state { pending = Pending1 } - end, hibernate}; - -handle_cast({run_async, Fun}, State = #state { available = [], - pending = Pending }) -> - {noreply, State #state { pending = queue:in({run_async, Fun}, Pending)}, - hibernate}; -handle_cast({run_async, Fun}, State = #state { available = [WPid | Avail1] }) -> - worker_pool_worker:submit_async(WPid, Fun), - {noreply, State #state { available = Avail1 }, hibernate}; - -handle_cast(Msg, State) -> - {stop, {unexpected_cast, Msg}, State}. - -handle_info({'DOWN', _MRef, process, WPid, _Reason}, - State = #state { available = Avail }) -> - {noreply, State #state { available = ordsets:del_element(WPid, Avail) }, - hibernate}; - -handle_info(Msg, State) -> - {stop, {unexpected_info, Msg}, State}. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -terminate(_Reason, State) -> - State. diff --git a/src/worker_pool_sup.erl b/src/worker_pool_sup.erl deleted file mode 100644 index 2608f5c2e6..0000000000 --- a/src/worker_pool_sup.erl +++ /dev/null @@ -1,56 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - --module(worker_pool_sup). - --behaviour(supervisor). - --export([start_link/0, start_link/1, start_link/2]). - --export([init/1]). - -%%---------------------------------------------------------------------------- - --spec start_link() -> rabbit_types:ok_pid_or_error(). --spec start_link(non_neg_integer()) -> rabbit_types:ok_pid_or_error(). --spec start_link(non_neg_integer(), atom()) - -> rabbit_types:ok_pid_or_error(). - -%%---------------------------------------------------------------------------- - -start_link() -> - start_link(erlang:system_info(schedulers)). - -start_link(WCount) -> - start_link(WCount, worker_pool:default_pool()). - -start_link(WCount, PoolName) -> - SupName = list_to_atom(atom_to_list(PoolName) ++ "_sup"), - supervisor:start_link({local, SupName}, ?MODULE, [WCount, PoolName]). - -%%---------------------------------------------------------------------------- - -init([WCount, PoolName]) -> - %% we want to survive up to 1K of worker restarts per second, - %% e.g. when a large worker pool used for network connections - %% encounters a network failure. This is the case in the LDAP authentication - %% backend plugin. - {ok, {{one_for_one, 1000, 1}, - [{worker_pool, {worker_pool, start_link, [PoolName]}, transient, - 16#ffffffff, worker, [worker_pool]} | - [{N, {worker_pool_worker, start_link, [PoolName]}, transient, - 16#ffffffff, worker, [worker_pool_worker]} - || N <- lists:seq(1, WCount)]]}}. diff --git a/src/worker_pool_worker.erl b/src/worker_pool_worker.erl deleted file mode 100644 index 23db127def..0000000000 --- a/src/worker_pool_worker.erl +++ /dev/null @@ -1,193 +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 Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2017 Pivotal Software, Inc. All rights reserved. -%% - --module(worker_pool_worker). - -%% Executes jobs (functions) submitted to a worker pool with worker_pool:submit/1, -%% worker_pool:submit/2 or worker_pool:submit_async/1. -%% -%% See worker_pool for an overview. - --behaviour(gen_server2). - --export([start_link/1, next_job_from/2, submit/3, submit_async/2, - run/1]). - --export([set_maximum_since_use/2]). --export([set_timeout/2, set_timeout/3, clear_timeout/1]). - --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3, prioritise_cast/3]). - -%%---------------------------------------------------------------------------- - --type mfargs() :: {atom(), atom(), [any()]}. - --spec start_link(atom) -> {'ok', pid()} | {'error', any()}. --spec next_job_from(pid(), pid()) -> 'ok'. --spec submit(pid(), fun (() -> A) | mfargs(), 'reuse' | 'single') -> A. --spec submit_async(pid(), fun (() -> any()) | mfargs()) -> 'ok'. --spec run(fun (() -> A)) -> A; (mfargs()) -> any(). --spec set_maximum_since_use(pid(), non_neg_integer()) -> 'ok'. - -%%---------------------------------------------------------------------------- - --define(HIBERNATE_AFTER_MIN, 1000). --define(DESIRED_HIBERNATE, 10000). - -%%---------------------------------------------------------------------------- - -start_link(PoolName) -> - gen_server2:start_link(?MODULE, [PoolName], [{timeout, infinity}]). - -next_job_from(Pid, CPid) -> - gen_server2:cast(Pid, {next_job_from, CPid}). - -submit(Pid, Fun, ProcessModel) -> - gen_server2:call(Pid, {submit, Fun, self(), ProcessModel}, infinity). - -submit_async(Pid, Fun) -> - gen_server2:cast(Pid, {submit_async, Fun}). - -set_maximum_since_use(Pid, Age) -> - gen_server2:cast(Pid, {set_maximum_since_use, Age}). - -run({M, F, A}) -> apply(M, F, A); -run(Fun) -> Fun(). - -run(Fun, reuse) -> - run(Fun); -run(Fun, single) -> - Self = self(), - Ref = make_ref(), - spawn_link(fun () -> - put(worker_pool_worker, true), - Self ! {Ref, run(Fun)}, - unlink(Self) - end), - receive - {Ref, Res} -> Res - end. - -%%---------------------------------------------------------------------------- - -init([PoolName]) -> - ok = file_handle_cache:register_callback(?MODULE, set_maximum_since_use, - [self()]), - ok = worker_pool:ready(PoolName, self()), - put(worker_pool_worker, true), - put(worker_pool_name, PoolName), - {ok, undefined, hibernate, - {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. - -prioritise_cast({set_maximum_since_use, _Age}, _Len, _State) -> 8; -prioritise_cast({next_job_from, _CPid}, _Len, _State) -> 7; -prioritise_cast(_Msg, _Len, _State) -> 0. - -handle_call({submit, Fun, CPid, ProcessModel}, From, undefined) -> - {noreply, {job, CPid, From, Fun, ProcessModel}, hibernate}; - -handle_call({submit, Fun, CPid, ProcessModel}, From, {from, CPid, MRef}) -> - erlang:demonitor(MRef), - gen_server2:reply(From, run(Fun, ProcessModel)), - ok = worker_pool:idle(get(worker_pool_name), self()), - {noreply, undefined, hibernate}; - -handle_call(Msg, _From, State) -> - {stop, {unexpected_call, Msg}, State}. - -handle_cast({next_job_from, CPid}, undefined) -> - MRef = erlang:monitor(process, CPid), - {noreply, {from, CPid, MRef}, hibernate}; - -handle_cast({next_job_from, CPid}, {job, CPid, From, Fun, ProcessModel}) -> - gen_server2:reply(From, run(Fun, ProcessModel)), - ok = worker_pool:idle(get(worker_pool_name), self()), - {noreply, undefined, hibernate}; - -handle_cast({submit_async, Fun}, undefined) -> - run(Fun), - ok = worker_pool:idle(get(worker_pool_name), self()), - {noreply, undefined, hibernate}; - -handle_cast({set_maximum_since_use, Age}, State) -> - ok = file_handle_cache:set_maximum_since_use(Age), - {noreply, State, hibernate}; - -handle_cast(Msg, State) -> - {stop, {unexpected_cast, Msg}, State}. - -handle_info({'DOWN', MRef, process, CPid, _Reason}, {from, CPid, MRef}) -> - ok = worker_pool:idle(get(worker_pool_name), self()), - {noreply, undefined, hibernate}; - -handle_info({'DOWN', _MRef, process, _Pid, _Reason}, State) -> - {noreply, State, hibernate}; - -handle_info({timeout, Key, Fun}, State) -> - clear_timeout(Key), - Fun(), - {noreply, State, hibernate}; - -handle_info(Msg, State) -> - {stop, {unexpected_info, Msg}, State}. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -terminate(_Reason, State) -> - State. - --spec set_timeout(integer(), fun(() -> any())) -> reference(). -set_timeout(Time, Fun) -> - Key = make_ref(), - set_timeout(Key, Time, Fun). - --spec set_timeout(Key, integer(), fun(() -> any())) -> Key when Key :: any(). -set_timeout(Key, Time, Fun) -> - Timeouts = get_timeouts(), - set_timeout(Key, Time, Fun, Timeouts). - --spec clear_timeout(any()) -> ok. -clear_timeout(Key) -> - NewTimeouts = cancel_timeout(Key, get_timeouts()), - put(timeouts, NewTimeouts), - ok. - -get_timeouts() -> - case get(timeouts) of - undefined -> dict:new(); - Dict -> Dict - end. - -set_timeout(Key, Time, Fun, Timeouts) -> - cancel_timeout(Key, Timeouts), - {ok, TRef} = timer:send_after(Time, {timeout, Key, Fun}), - NewTimeouts = dict:store(Key, TRef, Timeouts), - put(timeouts, NewTimeouts), - {ok, Key}. - -cancel_timeout(Key, Timeouts) -> - case dict:find(Key, Timeouts) of - {ok, TRef} -> - timer:cancel(TRef), - receive {timeout, Key, _} -> ok - after 0 -> ok - end, - dict:erase(Key, Timeouts); - error -> - Timeouts - end. |