Add rexi ping message

Add a ping message to rexi to avoid any long running operations from
timing out. Long running operations at the node level can exceed the
fabric timeout and be cancelled. Sending a ping message back will
stop that from happening.

2 files changed, 11 insertions, 0 deletions

diff --git a/src/rexi/src/rexi.erl b/src/rexi/src/rexi.erl
index fea4d6453..f774dc9d4 100644
--- a/src/rexi/src/rexi.erl
+++ b/src/rexi/src/rexi.erl
@@ -19,6 +19,7 @@
 -export([stream_start/1, stream_cancel/1]).
 -export([stream/1, stream/2, stream/3, stream_ack/1, stream_ack/2]).
 -export([stream2/1, stream2/2, stream2/3, stream_last/1, stream_last/2]).
+-export([ping/0]).
 
 -include_lib("rexi/include/rexi.hrl").
 
@@ -233,6 +234,14 @@ stream_ack(Client) ->
 stream_ack(Client, N) ->
     erlang:send(Client, {rexi_ack, N}).
 
+
+%% Sends a ping message to the coordinator. This is for long running
+%% operations on a node that could exceed the rexi timeout
+ping() -> 
+    {Caller, _} = get(rexi_from),
+    erlang:send(Caller, {rexi, '$rexi_ping'}).
+
+
 %% internal functions %%
 
 cast_msg(Msg) -> {'$gen_cast', Msg}.
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
index e3eaa6fcc..11dbb252a 100644
--- a/src/rexi/src/rexi_utils.erl
+++ b/src/rexi/src/rexi_utils.erl
@@ -81,6 +81,8 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
         Worker ->
             Fun(Msg, {Worker, From}, Acc0)
         end;
+    {rexi, '$rexi_ping'} ->
+        {ok, Acc0};
     {Ref, Msg} ->
         case lists:keyfind(Ref, Keypos, RefList) of
         false ->