You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2024/07/26 19:07:06 UTC
(couchdb) branch improve-rexi-stream-clean-on-coordinator-exit created (now a11236fde)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch improve-rexi-stream-clean-on-coordinator-exit
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at a11236fde Improve worker cleanup on early coordinator exit
This branch includes the following new commits:
new a11236fde Improve worker cleanup on early coordinator exit
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(couchdb) 01/01: Improve worker cleanup on early coordinator exit
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch improve-rexi-stream-clean-on-coordinator-exit
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit a11236fde098e9a7573f3ddad01697911858aa1c
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Fri Jul 26 13:47:06 2024 -0400
Improve worker cleanup on early coordinator exit
Previously, if the coordinator process is killed too quickly, before the stream
worker cleanup process is spawned, remote workers may be left around
waiting until the default 5 minute timeout expires.
In order to reliably clean up processes in that state, need to start the
cleaner process, with all the job references, before we start submitting them
for execution.
At first, it may seem impossible to monitor a process until after it's already
spawned. That's true for regular processes, however rexi operates on plain
references. For each process we spawn remotely we create a reference on the
coordinator side, which we can then use to track that job. Those are just plain
manually created references. Nothing stops us from creating them first, adding
them to a cleaner process, and only then submitting them.
That's exactly what this commit accomplishes:
* Create a streams specific `fabric_streams:submit_jobs/4` function, which
spawns the cleanup process early, generates worker references, and then
submits the jobs. This way, all the existing streaming submit_jobs can be
replaced easily in one line: fabric_util -> fabric_streams.
* The cleanup process operates as previously: monitors the coordinator for
exits, and fires off `kill_all` message to each node.
* Create `rexi:cast_ref(...)` variants of `rexi:cast(...)` calls, where the
caller specifies the references a new argument. This is what allows us to
start the cleanup process before the even get submitted. Older calls can just
be easily call into the `cast_ref` versions with their own created
references.
Since we added the new `rexi:cast_ref(...)` variants, ensure to add more test
coverage, including the streaming logic as well. It's not 100% yet, but getting
there.
Also, the comments in `rexi.erl` were full of erldoc stanzas and we don't
actually build erldocs anywhere, so replace them with something more helpful.
The streaming protocol itself was never quite described anywhere, and it can
take sometime to figure it out (at least it took me), so took the chance to
also add a very basic, high level description of the message flow.
Related: https://github.com/apache/couchdb/issues/5127#issuecomment-2253261222
---
.../src/couch_replicator_fabric.erl | 2 +-
src/fabric/src/fabric_streams.erl | 113 ++++++++++++++--
src/fabric/src/fabric_view_all_docs.erl | 2 +-
src/fabric/src/fabric_view_map.erl | 4 +-
src/fabric/src/fabric_view_reduce.erl | 4 +-
src/rexi/src/rexi.erl | 150 +++++++++++++++------
src/rexi/test/rexi_tests.erl | 119 ++++++++++++++--
7 files changed, 326 insertions(+), 68 deletions(-)
diff --git a/src/couch_replicator/src/couch_replicator_fabric.erl b/src/couch_replicator/src/couch_replicator_fabric.erl
index d570d6f77..43321f26f 100644
--- a/src/couch_replicator/src/couch_replicator_fabric.erl
+++ b/src/couch_replicator/src/couch_replicator_fabric.erl
@@ -22,7 +22,7 @@
docs(DbName, Options, QueryArgs, Callback, Acc) ->
Shards = mem3:shards(DbName),
- Workers0 = fabric_util:submit_jobs(
+ Workers0 = fabric_streams:submit_jobs(
Shards, couch_replicator_fabric_rpc, docs, [Options, QueryArgs]
),
RexiMon = fabric_util:create_monitors(Workers0),
diff --git a/src/fabric/src/fabric_streams.erl b/src/fabric/src/fabric_streams.erl
index b7c6b3afc..3f14cde45 100644
--- a/src/fabric/src/fabric_streams.erl
+++ b/src/fabric/src/fabric_streams.erl
@@ -13,6 +13,7 @@
-module(fabric_streams).
-export([
+ submit_jobs/4,
start/2,
start/3,
start/4,
@@ -27,6 +28,23 @@
-define(WORKER_CLEANER, fabric_worker_cleaner).
+% This is the streams equivalent of fabric_util:submit_jobs/4. Besides
+% submitting the jobs it also starts the worker cleaner and adds each started
+% job to the cleaner first before the job is submitted.
+%
+submit_jobs(Shards, Module, EndPoint, ExtraArgs) ->
+ % Create refs first and add them to the cleaner to ensure if our process
+ % gets killed, the remote workers will be cleaned up as well.
+ RefFun = fun(#shard{} = Shard) -> Shard#shard{ref = make_ref()} end,
+ Workers = lists:map(RefFun, Shards),
+ ClientReq = chttpd_util:mochiweb_client_req_get(),
+ spawn_worker_cleaner(self(), Workers, ClientReq),
+ SubmitFun = fun(#shard{node = Node, name = ShardName, ref = Ref}) ->
+ rexi:cast_ref(Ref, Node, {Module, EndPoint, [ShardName | ExtraArgs]})
+ end,
+ ok = lists:foreach(SubmitFun, Workers),
+ Workers.
+
start(Workers, Keypos) ->
start(Workers, Keypos, undefined, undefined).
@@ -158,39 +176,49 @@ handle_stream_start(Else, _, _) ->
% Spawn an auxiliary rexi worker cleaner. This will be used in cases
% when the coordinator (request) process is forceably killed and doesn't
% get a chance to process its `after` fabric:clean/1 clause.
-spawn_worker_cleaner(Coordinator, Workers, ClientReq) ->
+spawn_worker_cleaner(Coordinator, Workers, ClientReq) when
+ is_pid(Coordinator), is_list(Workers)
+->
case get(?WORKER_CLEANER) of
undefined ->
Pid = spawn(fun() ->
erlang:monitor(process, Coordinator),
- cleaner_loop(Coordinator, Workers, ClientReq)
+ NodeRefSet = set_from_list(shards_to_node_refs(Workers)),
+ cleaner_loop(Coordinator, NodeRefSet, ClientReq)
end),
put(?WORKER_CLEANER, Pid),
Pid;
- ExistingCleaner ->
+ ExistingCleaner when is_pid(ExistingCleaner) ->
ExistingCleaner
end.
-cleaner_loop(Pid, Workers, ClientReq) ->
+cleaner_loop(Pid, NodeRefSet, ClientReq) ->
CheckMSec = chttpd_util:mochiweb_client_req_check_msec(),
receive
- {add_worker, Pid, Worker} ->
- cleaner_loop(Pid, [Worker | Workers], ClientReq);
+ {add_node_ref, Pid, {_, _} = NodeRef} ->
+ cleaner_loop(Pid, sets:add_element(NodeRef, NodeRefSet), ClientReq);
{'DOWN', _, _, Pid, _} ->
- fabric_util:cleanup(Workers)
+ rexi:kill_all(sets:to_list(NodeRefSet))
after CheckMSec ->
chttpd_util:stop_client_process_if_disconnected(Pid, ClientReq),
- cleaner_loop(Pid, Workers, ClientReq)
+ cleaner_loop(Pid, NodeRefSet, ClientReq)
end.
-add_worker_to_cleaner(CoordinatorPid, Worker) ->
+add_worker_to_cleaner(CoordinatorPid, #shard{node = Node, ref = Ref}) ->
case get(?WORKER_CLEANER) of
CleanerPid when is_pid(CleanerPid) ->
- CleanerPid ! {add_worker, CoordinatorPid, Worker};
+ CleanerPid ! {add_node_ref, CoordinatorPid, {Node, Ref}};
_ ->
ok
end.
+set_from_list(List) when is_list(List) ->
+ sets:from_list(List, [{version, 2}]).
+
+shards_to_node_refs(Workers) when is_list(Workers) ->
+ Fun = fun(#shard{node = Node, ref = Ref}) -> {Node, Ref} end,
+ lists:map(Fun, Workers).
+
-ifdef(TEST).
-include_lib("couch/include/couch_eunit.hrl").
@@ -207,7 +235,8 @@ worker_cleaner_test_() ->
?TDEF_FE(does_not_fire_if_cleanup_called),
?TDEF_FE(should_clean_additional_worker_too),
?TDEF_FE(coordinator_is_killed_if_client_disconnects),
- ?TDEF_FE(coordinator_is_not_killed_if_client_is_connected)
+ ?TDEF_FE(coordinator_is_not_killed_if_client_is_connected),
+ ?TDEF_FE(submit_jobs_sets_up_cleaner)
]
}
}.
@@ -351,6 +380,68 @@ coordinator_is_not_killed_if_client_is_connected(_) ->
{'DOWN', CleanerRef, _, _, _} -> ok
end.
+submit_jobs_sets_up_cleaner(_) ->
+ meck:reset(rexi),
+ erase(?WORKER_CLEANER),
+ Shards = [
+ #shard{node = 'n1'},
+ #shard{node = 'n2'}
+ ],
+ meck:expect(rexi, cast_ref, fun(Ref, _, _) -> Ref end),
+ {Coord, CoordRef} = spawn_monitor(fun() ->
+ Workers = submit_jobs(Shards, fabric_rpc, potatoes, []),
+ receive
+ {get_workers_and_cleaner, From} ->
+ From ! {Workers, get(?WORKER_CLEANER)},
+ timer:sleep(999999)
+ end
+ end),
+ Coord ! {get_workers_and_cleaner, self()},
+ {Workers, Cleaner} =
+ receive
+ Msg -> Msg
+ end,
+ ?assert(is_pid(Cleaner)),
+ ?assert(is_process_alive(Cleaner)),
+ ?assert(is_process_alive(Coord)),
+ CheckWorkerFun = fun(#shard{node = Node, ref = Ref}) ->
+ ?assert(is_reference(Ref)),
+ {Node, Ref}
+ end,
+ NodeRefs = lists:map(CheckWorkerFun, Workers),
+ ?assertEqual(length(Shards), length(Workers)),
+ ?assertEqual(length(lists:usort(NodeRefs)), length(NodeRefs)),
+ % Were the jobs actually submitted?
+ meck:wait(2, rexi, cast_ref, '_', 1000),
+ % If we kill the coordinator, the cleaner should kill the workers
+ meck:reset(rexi),
+ CleanupMon = erlang:monitor(process, Cleaner),
+ exit(Coord, kill),
+ receive
+ {'DOWN', CoordRef, _, _, WorkerReason} ->
+ ?assertEqual(killed, WorkerReason)
+ after 1000 ->
+ ?assert(is_process_alive(Coord))
+ end,
+ % Cleaner should do the cleanup
+ meck:wait(1, rexi, kill_all, '_', 1000),
+ History = meck:history(rexi),
+ ?assertMatch([{_, {rexi, kill_all, _}, ok}], History),
+ [{Pid, {rexi, kill_all, Args}, ok}] = History,
+ % It was the cleaner who called it
+ ?assertEqual(Cleaner, Pid),
+ ?assertMatch([[{_, _}, {_, _}]], Args),
+ [NodeRefsKilled] = Args,
+ % The node refs killed are the ones we expect
+ ?assertEqual(lists:sort(NodeRefs), lists:sort(NodeRefsKilled)),
+ % Cleanup process should exit when done
+ receive
+ {'DOWN', CleanupMon, _, _, CleanerReason} ->
+ ?assertEqual(normal, CleanerReason)
+ after 1000 ->
+ ?assert(is_process_alive(Cleaner))
+ end.
+
setup() ->
ok = meck:expect(rexi, kill_all, fun(_) -> ok end),
% Speed up disconnect socket timeout for the test to 200 msec
diff --git a/src/fabric/src/fabric_view_all_docs.erl b/src/fabric/src/fabric_view_all_docs.erl
index decc49cf4..3a03357c2 100644
--- a/src/fabric/src/fabric_view_all_docs.erl
+++ b/src/fabric/src/fabric_view_all_docs.erl
@@ -25,7 +25,7 @@ go(Db, Options, #mrargs{keys = undefined} = QueryArgs, Callback, Acc) ->
{CoordArgs, WorkerArgs} = fabric_view:fix_skip_and_limit(QueryArgs),
DbName = fabric:dbname(Db),
{Shards, RingOpts} = shards(Db, QueryArgs),
- Workers0 = fabric_util:submit_jobs(
+ Workers0 = fabric_streams:submit_jobs(
Shards, fabric_rpc, all_docs, [Options, WorkerArgs]
),
RexiMon = fabric_util:create_monitors(Workers0),
diff --git a/src/fabric/src/fabric_view_map.erl b/src/fabric/src/fabric_view_map.erl
index 62077ac9f..6f13270a9 100644
--- a/src/fabric/src/fabric_view_map.erl
+++ b/src/fabric/src/fabric_view_map.erl
@@ -40,9 +40,9 @@ go(Db, Options, DDoc, View, Args0, Callback, Acc, VInfo) ->
Repls = fabric_ring:get_shard_replacements(DbName, Shards),
RPCArgs = [DocIdAndRev, View, WorkerArgs, Options],
StartFun = fun(Shard) ->
- hd(fabric_util:submit_jobs([Shard], fabric_rpc, map_view, RPCArgs))
+ hd(fabric_streams:submit_jobs([Shard], fabric_rpc, map_view, RPCArgs))
end,
- Workers0 = fabric_util:submit_jobs(Shards, fabric_rpc, map_view, RPCArgs),
+ Workers0 = fabric_streams:submit_jobs(Shards, fabric_rpc, map_view, RPCArgs),
RexiMon = fabric_util:create_monitors(Workers0),
try
case
diff --git a/src/fabric/src/fabric_view_reduce.erl b/src/fabric/src/fabric_view_reduce.erl
index f32cd0ce5..04d73bd94 100644
--- a/src/fabric/src/fabric_view_reduce.erl
+++ b/src/fabric/src/fabric_view_reduce.erl
@@ -31,9 +31,9 @@ go(Db, DDoc, VName, Args, Callback, Acc, VInfo) ->
fabric_view:maybe_update_others(DbName, DocIdAndRev, Shards, VName, Args),
Repls = fabric_ring:get_shard_replacements(DbName, Shards),
StartFun = fun(Shard) ->
- hd(fabric_util:submit_jobs([Shard], fabric_rpc, reduce_view, RPCArgs))
+ hd(fabric_streams:submit_jobs([Shard], fabric_rpc, reduce_view, RPCArgs))
end,
- Workers0 = fabric_util:submit_jobs(Shards, fabric_rpc, reduce_view, RPCArgs),
+ Workers0 = fabric_streams:submit_jobs(Shards, fabric_rpc, reduce_view, RPCArgs),
RexiMon = fabric_util:create_monitors(Workers0),
try
case
diff --git a/src/rexi/src/rexi.erl b/src/rexi/src/rexi.erl
index 4d5736530..02d3a9e55 100644
--- a/src/rexi/src/rexi.erl
+++ b/src/rexi/src/rexi.erl
@@ -12,7 +12,9 @@
-module(rexi).
--export([cast/2, cast/3, cast/4, kill_all/1]).
+-export([cast/2, cast/3, cast/4]).
+-export([cast_ref/3, cast_ref/4, cast_ref/5]).
+-export([kill_all/1]).
-export([reply/1]).
-export([stream_start/1, stream_cancel/1]).
-export([stream_ack/1]).
@@ -20,44 +22,69 @@
-export([ping/0]).
-export([aggregate_server_queue_len/0, aggregate_buffer_queue_len/0]).
-%% @equiv cast(Node, self(), MFA)
+%% Like cast(Node, self(), MFA)
-spec cast(node(), {atom(), atom(), list()}) -> reference().
cast(Node, MFA) ->
- cast(Node, self(), MFA).
-
-%% @doc Executes apply(M, F, A) on Node.
-%% You might want to use this instead of rpc:cast/4 for two reasons. First,
-%% the Caller pid and the returned reference are inserted into the remote
-%% process' dictionary as `rexi_from', so it has a way to communicate with you.
-%% Second, the remote process is monitored. If it exits with a Reason other
-%% than normal, Caller will receive a message of the form
-%% `{Ref, {rexi_EXIT, Reason}}' where Ref is the returned reference.
+ cast_ref(make_ref(), Node, self(), MFA).
+
+%% Executes apply(M, F, A) on Node.
+%% Works like cast_ref/4 but creates its own ref.
-spec cast(node(), pid(), {atom(), atom(), list()}) -> reference().
cast(Node, Caller, MFA) ->
- Ref = make_ref(),
+ cast_ref(make_ref(), Node, Caller, MFA).
+
+%% Executes apply(M, F, A) on Node.
+%% Works like cast_ref/5 but creates its own ref.
+-spec cast(node(), pid(), {atom(), atom(), list()}, [atom()]) -> reference().
+cast(Node, Caller, MFA, Options) ->
+ cast_ref(make_ref(), Node, Caller, MFA, Options).
+
+%% Like cast_ref(Ref, Node, self(), MFA)
+-spec cast_ref(reference(), node(), {atom(), atom(), list()}) -> reference().
+cast_ref(Ref, Node, MFA) when is_reference(Ref) ->
+ cast_ref(Ref, Node, self(), MFA).
+
+%% Execute apply(M, F, A) on Node. Remote rexi_server spawns and monitors a
+%% worker process to execute this function. Its process dictionary will contain
+%% the `rexi_from = {CallerPid, Ref}` tuple so it can communicate with the
+%% caller process. If the worker crashes, the remote rexi_server sends the
+%% caller a `{Ref, {rexi_EXIT, Reason}}` message.
+%%
+%% The first argument is a reference. `cast/2,3,4` versions will automatically
+%% create it. The reason to create it manually is to hand it over to cleanup
+%% process beforehand, to ensure if the caller dies the cleaner process will be
+%% able to kill the remote workers. It may be useful when unused workers may
+%% live longer consume resources (keep db handles, acces disk, etc).
+%%
+%%
+-spec cast_ref(reference(), node(), pid(), {atom(), atom(), list()}) -> reference().
+cast_ref(Ref, Node, Caller, MFA) when is_reference(Ref) ->
Msg = cast_msg({doit, {Caller, Ref}, get(nonce), MFA}),
rexi_utils:send(rexi_utils:server_pid(Node), Msg),
Ref.
-%% @doc Executes apply(M, F, A) on Node.
+%% Executes apply(M, F, A) on Node.
%% This version accepts a sync option which uses the erlang:send/2 call
%% directly in process instead of deferring to a spawned process if
%% erlang:send/2 were to block. If the sync option is omitted this call
-%% is identical to cast/3.
--spec cast(node(), pid(), {atom(), atom(), list()}, [atom()]) -> reference().
-cast(Node, Caller, MFA, Options) ->
+%% is identical to cast_ref/4.
+%%
+-spec cast_ref(reference(), node(), pid(), {atom(), atom(), list()}, [atom()]) -> reference().
+cast_ref(Ref, Node, Caller, MFA, Options) when is_reference(Ref) ->
case lists:member(sync, Options) of
true ->
- Ref = make_ref(),
Msg = cast_msg({doit, {Caller, Ref}, get(nonce), MFA}),
erlang:send(rexi_utils:server_pid(Node), Msg),
Ref;
false ->
- cast(Node, Caller, MFA)
+ cast_ref(Ref, Node, Caller, MFA)
end.
-%% @doc Sends an async kill signal to the remote processes associated with Refs.
-%% No rexi_EXIT message will be sent.
+%% Send async kill signals to the remote processes associated with Refs. It's
+%% more efficient to send a larger group of {Node, Ref} tuples as this function
+%% will batch refs per node for effiency. Workers killed this way will not send
+%% rexi_EXIT messages.
+%%
-spec kill_all([{node(), reference()}]) -> ok.
kill_all(NodeRefs) when is_list(NodeRefs) ->
% #{N1 => [Ref1, Ref2], N2 => [Ref3], ...}
@@ -72,15 +99,19 @@ kill_all(NodeRefs) when is_list(NodeRefs) ->
PerNodeMap
).
-%% @doc convenience function to reply to the original rexi Caller.
+%% Convenience function to reply to the original rexi Caller.
+%%
-spec reply(any()) -> any().
reply(Reply) ->
{Caller, Ref} = get(rexi_from),
erlang:send(Caller, {Ref, Reply}).
-%% @doc convenience function to reply to caller and wait for response. Message
-%% is of the form {OriginalRef, {self(),reference()}, Reply}, which enables the
-%% original caller to respond back.
+%% Private function used by stream2 to initialize the stream. Message is of the
+%% form {OriginalRef, {self(),reference()}, Reply}, which enables the
+%% coordinator to respond back. The function uses a gen_server From tuple for
+%% rexi_from so that the caller may respond using gen_server:reply(From, Msg).
+%% (see stream_start/1 and stream_cancel/1 functions).
+%%
-spec sync_reply(any(), pos_integer() | infinity) -> any().
sync_reply(Reply, Timeout) ->
{Caller, Ref} = get(rexi_from),
@@ -93,36 +124,63 @@ sync_reply(Reply, Timeout) ->
timeout
end.
-%% @doc Start a worker stream
+%% Start a worker stream. Coordinator sends this message to the worker to tell
+%% it to start streaming, after the worker sent a rexi_STREAM_INIT message.
+%%
+%% The `From` should be the value provided by the worker in the
+%% rexi_STREAM_INIT message.
%%
-%% If a coordinator wants to continue using a streaming worker it
-%% should use this function to inform the worker to continue
-%% sending messages. The `From` should be the value provided by
-%% the worker in the rexi_STREAM_INIT message.
-spec stream_start({pid(), any()}) -> ok.
stream_start({Pid, _Tag} = From) when is_pid(Pid) ->
gen_server:reply(From, rexi_STREAM_START).
-%% @doc Cancel a worker stream
+%% Cancel a worker stream
+%%
+%% If a coordinator decides that a worker is not going to be part of the
+%% response, it should use this function to cancel the worker. The `From`
+%% should be the value provided by the worker in the rexi_STREAM_INIT message.
%%
-%% If a coordinator decideds that a worker is not going to be part
-%% of the response it should use this function to cancel the worker.
-%% The `From` should be the value provided by the worker in the
-%% rexi_STREAM_INIT message.
-spec stream_cancel({pid(), any()}) -> ok.
stream_cancel({Pid, _Tag} = From) when is_pid(Pid) ->
gen_server:reply(From, rexi_STREAM_CANCEL).
-%% @equiv stream2(Msg, 5, 300000)
+%% Like stream2(Msg, 5, 300000)
stream2(Msg) ->
Limit = config:get_integer("rexi", "stream_limit", 5),
stream2(Msg, Limit, 300000).
-%% @doc Stream a message back to the coordinator. It limits the
-%% number of unacked messsages to Limit and throws a timeout error
-%% if it doesn't receive an ack in Timeout milliseconds. This
-%% is a combination of the old stream_start and stream functions
-%% which automatically does the stream initialization logic.
+%% Stream messages back to the coordinator. Initializes on first use. Limit
+%% the number of unacked messsages to Limit, and throw a timeout error if it
+%% doesn't receive an ack in Timeout milliseconds.
+%%
+%% The general protocol looks like this:
+%%
+%% Coordinator Worker (one of Q*N usually)
+%% ---------- --------------------------
+%% cast/2,3,4 -> {doit, ...} -> rexi_server:
+%% spawn_monitor worker process.
+%% First time stream2/1 is called it
+%% runs init_stream/1.
+%%
+%% init_stream/1:
+%% <- rexi_STREAM_INIT <- sync send, wait for reply
+%%
+%% Some workers are told to
+%% continue with rexi_STREAM_START.
+%% Others are told to stop with
+%% rexi_STREAM_CANCEL
+%%
+%% -> rexi_STREAM_START ->
+%% Keep calling rexi:stream2/3
+%% to stream data to coordinator...
+%%
+%% <- Caller ! {Ref, self(), Msg} <-
+%% ...
+%% Coordinator must acknowledge.
+%% -> {rexi_ack, 1} ->
+%% Send last message. No need for ack.
+%% <- Caller ! Msg <-
+%%
-spec stream2(any(), pos_integer(), pos_integer() | inifinity) -> any().
stream2(Msg, Limit, Timeout) ->
maybe_init_stream(Timeout),
@@ -138,24 +196,30 @@ stream2(Msg, Limit, Timeout) ->
exit(timeout)
end.
-%% @equiv stream_last(Msg, 300000)
+%% Like stream_last(Msg, 300000)
+%%
stream_last(Msg) ->
stream_last(Msg, 300000).
-%% @doc Send the last message in a stream. This difference between
+%% Send the last message in a stream. This difference between
%% this and stream is that it uses rexi:reply/1 which doesn't include
%% the worker pid and doesn't wait for a response from the controller.
+%%
stream_last(Msg, Timeout) ->
maybe_init_stream(Timeout),
rexi:reply(Msg),
ok.
-%% @doc Ack streamed messages
+%% Ack streamed messages. Coordinator must ack stream messages, except
+%% the last one sent with stream_last/1,2. Up to Limit message can be in
+%% flight un-acked before the workers will stop and wait for an ack.
+%%
stream_ack(Client) ->
erlang:send(Client, {rexi_ack, 1}).
%% 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'}).
diff --git a/src/rexi/test/rexi_tests.erl b/src/rexi/test/rexi_tests.erl
index fd544c5ad..18b05b545 100644
--- a/src/rexi/test/rexi_tests.erl
+++ b/src/rexi/test/rexi_tests.erl
@@ -24,8 +24,14 @@ rexi_buffer_test_() ->
fun setup/0,
fun teardown/1,
[
+ ?TDEF_FE(t_server_pid),
?TDEF_FE(t_cast),
+ ?TDEF_FE(t_cast_explicit_caller),
+ ?TDEF_FE(t_cast_ref),
?TDEF_FE(t_sync_cast),
+ ?TDEF_FE(t_stream2),
+ ?TDEF_FE(t_stream2_acks),
+ ?TDEF_FE(t_stream2_cancel),
?TDEF_FE(t_kill),
?TDEF_FE(t_cast_error),
?TDEF_FE(t_metrics),
@@ -46,11 +52,25 @@ rpc_test_fun({error, Error}) ->
error(Error);
rpc_test_fun(ping) ->
rexi:ping();
+rpc_test_fun(stream2_init) ->
+ rexi:stream2(stream1),
+ rexi:stream_last(stream2);
+rpc_test_fun(stream2_acks) ->
+ rexi:stream2(a),
+ rexi:stream2(b),
+ rexi:stream2(c),
+ rexi:stream2(d),
+ rexi:stream2(e),
+ rexi:stream2(f),
+ rexi:stream2(g),
+ rexi:stream_last(h);
rpc_test_fun(Arg) ->
rexi:reply({Arg, get()}).
+t_server_pid(_) ->
+ ?assertMatch({RexiServer, node42} when is_atom(RexiServer), rexi_utils:server_pid(node42)).
+
t_cast(_) ->
- ?assertMatch({RexiServer, node42} when is_atom(RexiServer), rexi_utils:server_pid(node42)),
put(nonce, yup),
Ref = rexi:cast(node(), {?MODULE, rpc_test_fun, [potato]}),
{Res, Dict} =
@@ -67,14 +87,40 @@ t_cast(_) ->
Dict
).
+t_cast_explicit_caller(_) ->
+ put(nonce, yep),
+ {CallerPid, CallerRef} = spawn_monitor(fun() ->
+ receive
+ Msg -> exit(Msg)
+ end
+ end),
+ Ref = rexi:cast(node(), CallerPid, {?MODULE, rpc_test_fun, [potato]}),
+ Result =
+ receive
+ {'DOWN', CallerRef, _, _, Exit} -> Exit
+ end,
+ ?assertMatch({Ref, {potato, [_ | _]}}, Result).
+
+t_cast_ref(_) ->
+ put(nonce, yesh),
+ Ref = make_ref(),
+ Ref2 = rexi:cast_ref(Ref, node(), {?MODULE, rpc_test_fun, [potato]}),
+ ?assertEqual(Ref, Ref2),
+ {Res, Dict} = recv(Ref),
+ ?assertEqual(potato, Res),
+ ?assertMatch(
+ #{
+ nonce := yesh,
+ '$initial_call' := {?MODULE, rpc_test_fun, 1},
+ rexi_from := {_Pid, Ref}
+ },
+ maps:from_list(Dict)
+ ).
+
t_sync_cast(_) ->
- ?assertMatch({RexiServer, node42} when is_atom(RexiServer), rexi_utils:server_pid(node42)),
put(nonce, yup),
Ref = rexi:cast(node(), self(), {?MODULE, rpc_test_fun, [potato]}, [sync]),
- {Res, Dict} =
- receive
- {Ref, {R, D}} -> {R, maps:from_list(D)}
- end,
+ {Res, Dict} = recv(Ref),
?assertEqual(potato, Res),
?assertMatch(
#{
@@ -82,11 +128,55 @@ t_sync_cast(_) ->
'$initial_call' := {?MODULE, rpc_test_fun, 1},
rexi_from := {_Pid, _Ref}
},
- Dict
+ maps:from_list(Dict)
).
+t_stream2(_) ->
+ % We act as the coordinator
+ Ref = rexi:cast(node(), {?MODULE, rpc_test_fun, [stream2_init]}),
+ rexi:stream_start(stream_init(Ref)),
+ ?assertEqual(stream1, recv(Ref)),
+ ?assertEqual(stream2, recv(Ref)),
+ % No more messages
+ ?assertEqual(timeout, recv(Ref)).
+
+t_stream2_acks(_) ->
+ Ref = rexi:cast(node(), {?MODULE, rpc_test_fun, [stream2_acks]}),
+ {WPid, _Tag} = From = stream_init(Ref),
+ Mon = erlang:monitor(process, WPid),
+ rexi:stream_start(From),
+ ?assertEqual(a, recv(Ref)),
+ ?assertEqual(b, recv(Ref)),
+ ?assertEqual(c, recv(Ref)),
+ ?assertEqual(d, recv(Ref)),
+ ?assertEqual(e, recv(Ref)),
+ ?assertEqual(timeout, recv(Ref)),
+ rexi:stream_ack(WPid),
+ ?assertEqual(f, recv(Ref)),
+ ?assertEqual(timeout, recv(Ref)),
+ rexi:stream_ack(WPid),
+ ?assertEqual(g, recv(Ref)),
+ ?assertEqual(h, recv(Ref)),
+ % Done streaming. Ensure worker is dead.
+ ?assertEqual(timeout, recv(Ref)),
+ Res =
+ receive
+ {'DOWN', Mon, _, _, Exit} -> Exit
+ end,
+ ?assertEqual(normal, Res).
+
+t_stream2_cancel(_) ->
+ Ref = rexi:cast(node(), {?MODULE, rpc_test_fun, [stream2_init]}),
+ {WPid, _Tag} = From = stream_init(Ref),
+ Mon = erlang:monitor(process, WPid),
+ rexi:stream_cancel(From),
+ Res =
+ receive
+ {'DOWN', Mon, _, _, Exit} -> Exit
+ end,
+ ?assertEqual(normal, Res).
+
t_cast_error(_) ->
- ?assertMatch({RexiServer, node42} when is_atom(RexiServer), rexi_utils:server_pid(node42)),
Ref = rexi:cast(node(), self(), {?MODULE, rpc_test_fun, [{error, tomato}]}, []),
Res =
receive
@@ -120,3 +210,16 @@ t_ping(_) ->
{rexi, Ping} -> Ping
end,
?assertEqual('$rexi_ping', Res).
+
+stream_init(Ref) ->
+ receive
+ {Ref, From, rexi_STREAM_INIT} ->
+ From
+ end.
+
+recv(Ref) when is_reference(Ref) ->
+ receive
+ {Ref, _, Msg} -> Msg;
+ {Ref, Msg} -> Msg
+ after 500 -> timeout
+ end.