From: Ayanda Dube <ayanda.dube@erlang-solutions.com>
Date: Tue, 2 Feb 2016 18:52:16 +0000
Subject: [PATCH] Adds call/3 for handling custom delegate call timeouts.

References: #248

Adds channel_operation_timeout envionment variable.

References: #248

Adds caching of channel_operation_timeout in process dictionary. Acquisition of channel_operation_timeout from dictionary on rabbit_amqqueue:notify_down_all/3 call.

References: rabbitmq/rabbitmq-server-248

Adds CHANNEL_OPERATION_TIMEOUT macro.

References: rabbitmq/rabbitmq-server-248

Adds notify_down_all/3 clause to handle timeouts.

Ref: rabbitmq/rabbitmq-server#248

Adds get_channel_operation_timeout/0.

Ref: rabbitmq/rabbitmq-server#248

Renames CHANNEL_OPERATION_TIMEOUT dict key.

Ignore result from notify_queues/1.

Ref: rabbitmq/rabbitmq-server#248

diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 6ce3465..e8facf4 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -73,5 +73,7 @@
            mnesia_lib, rpc, mnesia_tm, qlc, sofs, proplists, credit_flow,
            pmon, ssl_connection, tls_connection, ssl_record, tls_record,
            gen_fsm, ssl]},
-         {ssl_apps, [asn1, crypto, public_key, ssl]}
+         {ssl_apps, [asn1, crypto, public_key, ssl]},
+         %% see rabbitmq-server#248
+         {channel_operation_timeout, 70000}
         ]}]}.
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 5ac3197..a05514b 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -112,6 +112,8 @@
 -define(ROUTING_HEADERS, [<<"CC">>, <<"BCC">>]).
 -define(DELETED_HEADER, <<"BCC">>).
 
+-define(CHANNEL_OPERATION_TIMEOUT, rabbit_misc:get_channel_operation_timeout()).
+
 %% Trying to send a term across a cluster larger than 2^31 bytes will
 %% cause the VM to exit with "Absurdly large distribution output data
 %% buffer". So we limit the max message size to 2^31 - 10^6 bytes (1MB
diff --git a/src/delegate.erl b/src/delegate.erl
index 378759a..e358953 100644
--- a/src/delegate.erl
+++ b/src/delegate.erl
@@ -19,7 +19,7 @@
 -behaviour(gen_server2).
 
 -export([start_link/1, invoke_no_result/2, invoke/2,
-         monitor/2, demonitor/1, call/2, cast/2]).
+         monitor/2, demonitor/1, call/2, call/3, cast/2]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
@@ -47,6 +47,9 @@
 -spec(call/2 ::
         ( pid(),  any()) -> any();
         ([pid()], any()) -> {[{pid(), any()}], [{pid(), term()}]}).
+-spec(call/3 ::
+        ( pid(),  any(), non_neg_integer()) -> any();
+        ([pid()], any(), non_neg_integer()) -> {[{pid(), any()}], [{pid(), term()}]}).
 -spec(cast/2 :: (pid() | [pid()], any()) -> 'ok').
 
 -endif.
@@ -138,6 +141,9 @@ demonitor({Name, Pid}) ->
 call(PidOrPids, Msg) ->
     invoke(PidOrPids, {gen_server2, call, [Msg, infinity]}).
 
+call(PidOrPids, Msg, Timeout) ->
+    invoke(PidOrPids, {gen_server2, call, [Msg, Timeout]}).
+
 cast(PidOrPids, Msg) ->
     invoke_no_result(PidOrPids, {gen_server2, cast, [Msg]}).
 
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 1aba7ec..0ec4a96 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -28,7 +28,7 @@
 -export([consumers/1, consumers_all/1, consumer_info_keys/0]).
 -export([basic_get/4, basic_consume/10, basic_cancel/4, notify_decorators/1]).
 -export([notify_sent/2, notify_sent_queue_down/1, resume/2]).
--export([notify_down_all/2, activate_limit_all/2, credit/5]).
+-export([notify_down_all/2, notify_down_all/3, activate_limit_all/2, credit/5]).
 -export([on_node_down/1]).
 -export([update/2, store_queue/1, policy_changed/2]).
 -export([start_mirroring/1, stop_mirroring/1, sync_mirrors/1,
@@ -148,6 +148,7 @@
 -spec(ack/3 :: (pid(), [msg_id()], pid()) -> 'ok').
 -spec(reject/4 :: (pid(), [msg_id()], boolean(), pid()) -> 'ok').
 -spec(notify_down_all/2 :: (qpids(), pid()) -> ok_or_errors()).
+-spec(notify_down_all/3 :: (qpids(), pid(), non_neg_integer()) -> ok_or_errors()).
 -spec(activate_limit_all/2 :: (qpids(), pid()) -> ok_or_errors()).
 -spec(basic_get/4 :: (rabbit_types:amqqueue(), pid(), boolean(), pid()) ->
                           {'ok', non_neg_integer(), qmsg()} | 'empty').
@@ -561,13 +562,22 @@ reject(QPid, Requeue, MsgIds, ChPid) ->
     delegate:cast(QPid, {reject, Requeue, MsgIds, ChPid}).
 
 notify_down_all(QPids, ChPid) ->
-    {_, Bads} = delegate:call(QPids, {notify_down, ChPid}),
-    case lists:filter(
-           fun ({_Pid, {exit, {R, _}, _}}) -> rabbit_misc:is_abnormal_exit(R);
-               ({_Pid, _})                 -> false
-           end, Bads) of
-        []    -> ok;
-        Bads1 -> {error, Bads1}
+    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) ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 043ec7e..59f467b 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -224,6 +224,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
     rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State1)),
     rabbit_event:if_enabled(State1, #ch.stats_timer,
                             fun() -> emit_stats(State1) end),
+    put(channel_operation_timeout, ?CHANNEL_OPERATION_TIMEOUT),
     {ok, State1, hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
@@ -635,7 +636,7 @@ handle_method(_Method, _, State = #ch{state = closing}) ->
     {noreply, State};
 
 handle_method(#'channel.close'{}, _, State = #ch{reader_pid = ReaderPid}) ->
-    {ok, State1} = notify_queues(State),
+    {_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
@@ -1446,7 +1447,9 @@ notify_queues(State = #ch{consumer_mapping  = Consumers,
                           delivering_queues = DQ }) ->
     QPids = sets:to_list(
               sets:union(sets:from_list(consumer_queues(Consumers)), DQ)),
-    {rabbit_amqqueue:notify_down_all(QPids, self()), State#ch{state = closing}}.
+    {rabbit_amqqueue:notify_down_all(QPids, self(),
+                                     get(channel_operation_timeout)),
+     State#ch{state = closing}}.
 
 foreach_per_queue(_F, []) ->
     ok;
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 63083b8..feed680 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -71,6 +71,7 @@
 -export([get_parent/0]).
 -export([store_proc_name/1, store_proc_name/2]).
 -export([moving_average/4]).
+-export([get_channel_operation_timeout/0]).
 
 %% Horrible macro to use in guards
 -define(IS_BENIGN_EXIT(R),
@@ -250,6 +251,7 @@
 -spec(store_proc_name/1 :: (rabbit_types:proc_type_and_name()) -> ok).
 -spec(moving_average/4 :: (float(), float(), float(), float() | 'undefined')
                           -> float()).
+-spec(get_channel_operation_timeout/0 :: () -> non_neg_integer()).
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -1080,6 +1082,10 @@ moving_average(Time,  HalfLife,  Next, Current) ->
     Weight = math:exp(Time * math:log(0.5) / HalfLife),
     Next * (1 - Weight) + Current * Weight.
 
+get_channel_operation_timeout() ->
+    Default = (net_kernel:get_net_ticktime() + 10) * 1000,
+    application:get_env(rabbit, channel_operation_timeout, Default).
+
 %% -------------------------------------------------------------------------
 %% Begin copypasta from gen_server2.erl
 
