Introduce originating source to channels, to help avoid

synchronous channel requests back to the connection process

4 files changed, 143 insertions, 3 deletions

diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 036aa9a60c..dcb4befee6 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -63,6 +63,7 @@
 -export([refresh_config_local/0, ready_for_close/1]).
 -export([refresh_interceptors/0]).
 -export([force_event_refresh/1]).
+-export([source/2]).
 
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2, handle_pre_hibernate/1, prioritise_call/4,
@@ -448,6 +449,14 @@ force_event_refresh(Ref) ->
 list_queue_states(Pid) ->
     gen_server2:call(Pid, list_queue_states).
 
+-spec source(pid(), any()) -> any().
+
+source(Pid, Source) when is_pid(Pid) ->
+    case erlang:is_process_alive(Pid) of
+        true  -> Pid ! {channel_source, Source};
+        false -> {error, channel_terminated}
+    end.
+
 %%---------------------------------------------------------------------------
 
 init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
@@ -805,7 +814,11 @@ handle_info(queue_cleanup, State = #ch{queue_states = QueueStates0}) ->
                             QName =  rabbit_quorum_queue:queue_name(QS),
                             [] /= rabbit_amqqueue:lookup(QName)
                     end, QueueStates0),
-    noreply(init_queue_cleanup_timer(State#ch{queue_states = QueueStates})).
+    noreply(init_queue_cleanup_timer(State#ch{queue_states = QueueStates}));
+
+handle_info({channel_source, Source}, State) ->
+    put(channel_source, Source),
+    noreply(State).
 
 handle_pre_hibernate(State) ->
     ok = clear_permission_cache(),
@@ -984,7 +997,7 @@ check_topic_authorisation(Resource = #exchange{type = topic},
     check_topic_authorisation(Resource, User, AmqpParams, RoutingKey, Permission);
 check_topic_authorisation(Resource = #exchange{type = topic},
                           User, ConnPid, RoutingKey, Permission) when is_pid(ConnPid) ->
-    AmqpParams = get_amqp_params(ConnPid),
+    AmqpParams = get_amqp_params(ConnPid, get(channel_source)),
     check_topic_authorisation(Resource, User, AmqpParams, RoutingKey, Permission);
 check_topic_authorisation(#exchange{name = Name = #resource{virtual_host = VHost}, type = topic},
                           User = #user{username = Username},
@@ -1007,7 +1020,8 @@ check_topic_authorisation(#exchange{name = Name = #resource{virtual_host = VHost
 check_topic_authorisation(_, _, _, _, _) ->
     ok.
 
-get_amqp_params(ConnPid) when is_pid(ConnPid) ->
+get_amqp_params(_ConnPid, rabbit_reader) -> [];
+get_amqp_params(ConnPid, _Any) when is_pid(ConnPid) ->
     Timeout = get_operation_timeout(),
     get_amqp_params(ConnPid, rabbit_misc:is_process_alive(ConnPid), Timeout).
 
@@ -2246,6 +2260,7 @@ i(garbage_collection, _State) ->
 i(reductions, _State) ->
     {reductions, Reductions} = erlang:process_info(self(), reductions),
     Reductions;
+i(channel_source, _State = #ch{}) -> get(channel_source);
 i(Item, _) ->
     throw({bad_argument, Item}).
 
diff --git a/src/rabbit_direct.erl b/src/rabbit_direct.erl
index 50e8f3d2b0..e43bfba90c 100644
--- a/src/rabbit_direct.erl
+++ b/src/rabbit_direct.erl
@@ -211,6 +211,7 @@ start_channel(Number, ClientChannelPid, ConnPid, ConnName, Protocol, User,
           rabbit_direct_client_sup,
           [{direct, Number, ClientChannelPid, ConnPid, ConnName, Protocol,
             User, VHost, Capabilities, Collector}]),
+    _ = rabbit_channel:source(ChannelPid, ?MODULE),
     {ok, ChannelPid}.
 
 -spec disconnect(pid(), rabbit_event:event_props()) -> 'ok'.
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index c0cb9c57d5..6f0b0a5ea5 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -924,6 +924,7 @@ create_channel(Channel,
         rabbit_channel_sup_sup:start_channel(
           ChanSupSup, {tcp, Sock, Channel, FrameMax, self(), Name,
                        Protocol, User, VHost, Capabilities, Collector}),
+    _ = rabbit_channel:source(ChPid, ?MODULE),
     MRef = erlang:monitor(process, ChPid),
     put({ch_pid, ChPid}, {Channel, MRef}),
     put({channel, Channel}, {ChPid, AState}),
diff --git a/test/channel_source_SUITE.erl b/test/channel_source_SUITE.erl
new file mode 100644
index 0000000000..11f87c7fde
--- /dev/null
+++ b/test/channel_source_SUITE.erl
@@ -0,0 +1,123 @@
+%% 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-2019 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(channel_source_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-compile(export_all).
+
+all() ->
+    [
+      {group, non_parallel_tests}
+    ].
+
+groups() ->
+    [
+      {non_parallel_tests, [], [
+          network_channel_source_notifications,
+          direct_channel_source_notifications,
+          undefined_channel_source_notifications
+        ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(_, Config) ->
+    Config.
+
+end_per_group(_, Config) ->
+    Config.
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase),
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodename_suffix, Testcase}
+      ]),
+    rabbit_ct_helpers:run_steps(Config1,
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps()).
+
+end_per_testcase(Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:run_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()),
+    rabbit_ct_helpers:testcase_finished(Config1, Testcase).
+
+%% -------------------------------------------------------------------
+%% Testcases.
+%% -------------------------------------------------------------------
+
+network_channel_source_notifications(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, network_channel_source_notifications1, [Config]).
+
+network_channel_source_notifications1(Config) ->
+    ExistingChannels = rabbit_channel:list(),
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, _ClientCh} = amqp_connection:open_channel(Conn),
+    [ServerCh] = rabbit_channel:list() -- ExistingChannels,
+    [{channel_source, rabbit_reader}] =
+        rabbit_channel:info(ServerCh, [channel_source]),
+    rabbit_channel:source(ServerCh, ?MODULE),
+    [{channel_source, ?MODULE}] =
+        rabbit_channel:info(ServerCh, [channel_source]),
+    amqp_connection:close(Conn),
+    {error, channel_terminated} = rabbit_channel:source(ServerCh, ?MODULE),
+    passed.
+
+direct_channel_source_notifications(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, direct_channel_source_notifications1, [Config]).
+
+direct_channel_source_notifications1(Config) ->
+    ExistingChannels = rabbit_channel:list(),
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection_direct(Config),
+    {ok, _ClientCh} = amqp_connection:open_channel(Conn),
+    [ServerCh] = rabbit_channel:list() -- ExistingChannels,
+    [{channel_source, rabbit_direct}] =
+        rabbit_channel:info(ServerCh, [channel_source]),
+    rabbit_channel:source(ServerCh, ?MODULE),
+    [{channel_source, ?MODULE}] =
+        rabbit_channel:info(ServerCh, [channel_source]),
+    amqp_connection:close(Conn),
+    {error, channel_terminated} = rabbit_channel:source(ServerCh, ?MODULE),
+    passed.
+
+undefined_channel_source_notifications(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, undefined_channel_source_notifications1, [Config]).
+
+undefined_channel_source_notifications1(_Config) ->
+    ExistingChannels = rabbit_channel:list(),
+    {_Writer, _Limiter, ServerCh} = rabbit_ct_broker_helpers:test_channel(),
+    [ServerCh] = rabbit_channel:list() -- ExistingChannels,
+    [{channel_source, undefined}] =
+        rabbit_channel:info(ServerCh, [channel_source]),
+    rabbit_channel:source(ServerCh, ?MODULE),
+    [{channel_source, ?MODULE}] =
+        rabbit_channel:info(ServerCh, [channel_source]),
+    passed.