partisan_erpc (partisan v6.0.0)

View Source

Executes a function call on a remote node over the Partisan transport.

This is Partisan's counterpart of Erlang's erpc, and the primary RPC surface — prefer it to partisan_rpc, which is the legacy rpc-shaped API built on top of this one.

Every function erpc exports is exported here with the same arity and the same contract, so code written against erpc behaves identically when pointed at this module. The differences are all supersets:

How a call is carried

A request is an explicit correlated message to the target's partisan_rpc_backend, which runs it in a process of its own and replies directly to the caller. Upstream erpc instead uses the spawn_request/5 BIF and reads the result from a monitor's exit reason — both are distribution mechanisms that ride Erlang distribution, not Partisan, which is why the transport here is native rather than vendored.

Two properties follow, and callers can rely on them:

  • A slow call delays only itself. Each request runs in its own process on the target, so one blocking M:F(A) does not stall unrelated calls.
  • A late reply is discarded, never mis-delivered. The correlation reference is a process alias; abandoning a request deactivates it, so a reply arriving after its caller gave up is dropped by the runtime.

Failures

Failures are raised, not returned. A remote exception is re-raised locally with its class preserved; an unreachable target raises error({partisan_erpc, noconnection}); an expired timeout raises error({partisan_erpc, timeout}); and a target already running rpc_max_concurrency requests raises error({partisan_erpc, overloaded}). partisan_rpc translates all of these into {badrpc, _} for callers that prefer the rpc convention.

Peer service manager

This module currently works only with partisan_pluggable_peer_service_manager.

Summary

Types

caught_call_exception()
monitor_ref()
request_id()
request_id_collection()
stack_item()

Functions

call(Node, Fun)

Equivalent to call(Node, Fun, infinity).

call(Node, Fun, Timeout)

Evaluates Fun() on Node and returns its value.

call(Node, Module, Function, Args)

Equivalent to call(Node, Module, Function, Args, infinity).

call(Node, Module, Function, Args, TimeoutOrOpts)

Evaluates apply(Module, Function, Args) on Node and returns its value.

cast(Node, Fun)

Equivalent to cast(Node, erlang, apply, [Fun, []]).

cast(Node, Module, Function, Args)

Evaluates apply(Module, Function, Args) on Node and discards the result.

cast(Node, Module, Function, Args, Opts)

Evaluates apply(Module, Function, Args) on Node with per-call transport options, discarding the result.

check_response(Message, RequestId)

Tests whether Message is the response to RequestId.

check_response/3

Tests whether Message is a response to any request in a collection.

multicall(Nodes, Fun)

Equivalent to multicall(Nodes, Fun, infinity).

multicall(Nodes, Fun, Timeout)

Equivalent to multicall(Nodes, erlang, apply, [Fun, []], Timeout).

multicall(Nodes, Module, Function, Args)

Equivalent to multicall(Nodes, Module, Function, Args, infinity).

multicall(Nodes, Module, Function, Args, TimeoutOrOpts)

Evaluates apply(Module, Function, Args) on every node in Nodes.

multicast(Nodes, Fun)

Equivalent to multicast(Nodes, erlang, apply, [Fun, []]).

multicast(Nodes, Module, Function, Args)

Evaluates apply(Module, Function, Args) on every node in Nodes, discarding the results.

multicast(Nodes, Module, Function, Args, Opts)

Evaluates apply(Module, Function, Args) on every node in Nodes with per-call transport options, discarding the results.

receive_response(RequestId)

Equivalent to receive_response(RequestId, infinity).

receive_response(RequestId, Timeout)

Waits for the response to RequestId and returns its value.

receive_response/3

Waits for the next response in a collection.

reqids_add/3

Adds an already-issued request to a collection under Label.

reqids_new()

Returns a new, empty request-identifier collection.

reqids_size(ReqIdCollection)

Returns the number of requests still outstanding in the collection.

reqids_to_list/1

Returns the collection's outstanding requests as a list of {RequestId, Label} pairs.

send_request(Node, Fun)

Equivalent to send_request(Node, erlang, apply, [Fun, []]).

send_request(Node, Module, Function, Args)

Sends a request to evaluate apply(Module, Function, Args) on Node and returns immediately.

send_request(Node, Module, Function, Args, Opts)

Sends a request as send_request/4, with per-call transport options.

send_request/6

Sends a request as send_request/4 and adds it to a collection under Label.

send_request/7

Sends a request as send_request/6, with per-call transport options.

wait_response(RequestId)

Equivalent to wait_response(RequestId, 0) — checks for a response without waiting.

wait_response(RequestId, WaitTime)

Waits up to WaitTime for the response to RequestId.

wait_response/3

Waits up to WaitTime for the next response in a collection.

Types

caught_call_exception()

-type caught_call_exception() ::
          {throw, Throw :: term()} |
          {exit, {exception, Reason :: term()}} |
          {error, {exception, Reason :: term(), StackTrace :: [stack_item()]}} |
          {exit, {signal, Reason :: term()}} |
          {error, {partisan_erpc, Reason :: term()}}.

monitor_ref()

-type monitor_ref() :: reference() | partisan:remote_reference().

request_id()

-opaque request_id()

request_id_collection()

-type request_id_collection() ::
          #{Res :: partisan:remote_reference() => {ReqId :: monitor_ref(), Label :: term()}}.

stack_item()

-type stack_item() ::
          {Module :: atom(),
           Function :: atom(),
           Arity :: arity() | (Args :: [term()]),
           Location :: [{file, Filename :: string()} | {line, Line :: pos_integer()}]}.

Functions

call(Node, Fun)

-spec call(Node, Fun) -> Result when Node :: node(), Fun :: function(), Result :: term().

Equivalent to call(Node, Fun, infinity).

call(Node, Fun, Timeout)

-spec call(Node, Fun, Timeout) -> Result
              when
                  Node :: node(),
                  Fun :: function(),
                  Timeout :: 0..4294967295 | infinity,
                  Result :: term().

Evaluates Fun() on Node and returns its value.

Equivalent to call(Node, erlang, apply, [Fun, []], Timeout). Fun must be loadable on the target.

call(Node, Module, Function, Args)

-spec call(Node, Module, Function, Args) -> Result
              when
                  Node :: node(),
                  Module :: atom(),
                  Function :: atom(),
                  Args :: [term()],
                  Result :: term().

Equivalent to call(Node, Module, Function, Args, infinity).

A call to the local node with an infinity timeout is evaluated in the calling process, without going near the transport.

call(Node, Module, Function, Args, TimeoutOrOpts)

-spec call(Node, Module, Function, Args, TimeoutOrOpts) -> Result
              when
                  Node :: node(),
                  Module :: atom(),
                  Function :: atom(),
                  Args :: [term()],
                  TimeoutOrOpts ::
                      0..4294967295 | infinity | partisan_peer_service_manager:forward_opts(),
                  Result :: term().

Evaluates apply(Module, Function, Args) on Node and returns its value.

Blocks until the result arrives, the timeout expires, or the target becomes unreachable. The call runs in a process of its own on the target, so it does not delay unrelated calls there.

TimeoutOrOpts is either a timeout or a partisan_peer_service_manager:forward_opts/0 map from which timeout is read — use the latter to place the request on a specific channel.

Raises on failure; see the module documentation for the error terms.

cast(Node, Fun)

-spec cast(Node, Fun) -> ok when Node :: node(), Fun :: function().

Equivalent to cast(Node, erlang, apply, [Fun, []]).

cast(Node, Module, Function, Args)

-spec cast(Node, Module, Function, Args) -> ok
              when Node :: node(), Module :: atom(), Function :: atom(), Args :: [term()].

Evaluates apply(Module, Function, Args) on Node and discards the result.

Returns ok as soon as the request has been handed to the transport: there is no reply, no correlation and no confirmation that the target ran anything. The call still runs in its own process on the target.

cast(Node, Module, Function, Args, Opts)

-spec cast(Node, Module, Function, Args, Opts) -> ok
              when
                  Node :: node(),
                  Module :: atom(),
                  Function :: atom(),
                  Args :: [term()],
                  Opts :: partisan_peer_service_manager:forward_opts().

Evaluates apply(Module, Function, Args) on Node with per-call transport options, discarding the result.

Partisan-specific; erpc has no equivalent. See cast/4 for the semantics and send_request/5 for the options.

check_response(Message, RequestId)

-spec check_response(Message, RequestId) -> {response, Result} | no_response
                        when Message :: term(), RequestId :: request_id(), Result :: term().

Tests whether Message is the response to RequestId.

Returns {response, Result} if it is, and no_response for any other message — which lets a process with its own receive loop handle responses without giving up control of its mailbox.

check_response/3

-spec check_response(Message :: term(), request_id_collection(), Delete :: boolean()) ->
                        {{response, Result :: term()}, Label :: term(), request_id_collection()} |
                        no_response.

Tests whether Message is a response to any request in a collection.

Returns {{response, Result}, Label, Collection} if it is, and no_response otherwise, so a process can service its own mailbox and still collect responses.

multicall(Nodes, Fun)

-spec multicall(Nodes, Fun) -> Result when Nodes :: [atom()], Fun :: function(), Result :: term().

Equivalent to multicall(Nodes, Fun, infinity).

multicall(Nodes, Fun, Timeout)

-spec multicall(Nodes, Fun, Timeout) -> Result
                   when
                       Nodes :: [atom()],
                       Fun :: function(),
                       Timeout :: 0..4294967295 | infinity,
                       Result :: term().

Equivalent to multicall(Nodes, erlang, apply, [Fun, []], Timeout).

multicall(Nodes, Module, Function, Args)

-spec multicall(Nodes, Module, Function, Args) -> Result
                   when
                       Nodes :: [atom()],
                       Module :: atom(),
                       Function :: atom(),
                       Args :: [term()],
                       Result :: [{ok, ReturnValue :: term()} | caught_call_exception()].

Equivalent to multicall(Nodes, Module, Function, Args, infinity).

multicall(Nodes, Module, Function, Args, TimeoutOrOpts)

-spec multicall(Nodes, Module, Function, Args, TimeoutOrOpts) -> Result
                   when
                       Nodes :: [atom()],
                       Module :: atom(),
                       Function :: atom(),
                       Args :: [term()],
                       TimeoutOrOpts ::
                           0..4294967295 | infinity | partisan_peer_service_manager:forward_opts(),
                       Result :: [{ok, ReturnValue :: term()} | caught_call_exception()].

Evaluates apply(Module, Function, Args) on every node in Nodes.

Returns one result per node, in the order of Nodes — {ok, Value} for a node that answered, or the caught exception for one that did not. Unlike the single-node calls, a failure on one node is reported in its own element rather than raised, so a timeout or an unreachable peer never hides the other results.

TimeoutOrOpts is either a timeout or a partisan_peer_service_manager:forward_opts/0 map, as for call/5. The timeout applies to the fan-out as a whole.

multicast(Nodes, Fun)

-spec multicast(Nodes, Fun) -> ok when Nodes :: [node()], Fun :: function().

Equivalent to multicast(Nodes, erlang, apply, [Fun, []]).

multicast(Nodes, Module, Function, Args)

-spec multicast(Nodes, Module, Function, Args) -> ok
                   when Nodes :: [node()], Module :: atom(), Function :: atom(), Args :: [term()].

Evaluates apply(Module, Function, Args) on every node in Nodes, discarding the results.

The fan-out counterpart of cast/4: no replies, no correlation, no indication of which nodes ran anything.

multicast(Nodes, Module, Function, Args, Opts)

-spec multicast(Nodes, Module, Function, Args, Opts) -> ok
                   when
                       Nodes :: [node()],
                       Module :: atom(),
                       Function :: atom(),
                       Args :: [term()],
                       Opts :: partisan_peer_service_manager:forward_opts().

Evaluates apply(Module, Function, Args) on every node in Nodes with per-call transport options, discarding the results.

Partisan-specific; erpc has no equivalent. See multicast/4 for the semantics and send_request/5 for the options.

receive_response(RequestId)

-spec receive_response(RequestId) -> Result when RequestId :: request_id(), Result :: term().

Equivalent to receive_response(RequestId, infinity).

receive_response(RequestId, Timeout)

-spec receive_response(RequestId, Timeout) -> Result
                          when
                              RequestId :: request_id(),
                              Timeout :: 0..4294967295 | infinity,
                              Result :: term().

Waits for the response to RequestId and returns its value.

Blocks until the response arrives, the timeout expires, or the target becomes unreachable, and raises on failure exactly as call/5 does. The request is abandoned on timeout, so a reply that arrives afterwards is discarded rather than left in the mailbox.

receive_response/3

-spec receive_response(request_id_collection(),
                       Timeout :: 0..4294967295 | infinity,
                       Delete :: boolean()) ->
                          {Result :: term(), Label :: term(), request_id_collection()}.

Waits for the next response in a collection.

Returns the result, the label the request was added under, and the collection — with that request removed when Delete is true. Responses are returned in the order they arrive, not the order the requests were issued.

Raises error({partisan_erpc, timeout}) when the timeout expires, abandoning every request in the collection.

reqids_add/3

-spec reqids_add(request_id(), term(), request_id_collection()) -> request_id_collection().

Adds an already-issued request to a collection under Label.

Use it to fold a request made with send_request/4 into a collection. Raises error({partisan_erpc, badarg}) if the identifier is already present.

reqids_new()

-spec reqids_new() -> request_id_collection().

Returns a new, empty request-identifier collection.

reqids_size(ReqIdCollection)

-spec reqids_size(request_id_collection()) -> non_neg_integer().

Returns the number of requests still outstanding in the collection.

reqids_to_list/1

-spec reqids_to_list(request_id_collection()) -> [{request_id(), Label :: term()}].

Returns the collection's outstanding requests as a list of {RequestId, Label} pairs.

send_request(Node, Fun)

-spec send_request(Node, Fun) -> RequestId
                      when Node :: node(), Fun :: function(), RequestId :: request_id().

Equivalent to send_request(Node, erlang, apply, [Fun, []]).

send_request(Node, Module, Function, Args)

-spec send_request(Node, Module, Function, Args) -> RequestId
                      when
                          Node :: node(),
                          Module :: atom(),
                          Function :: atom(),
                          Args :: [term()],
                          RequestId :: request_id().

Sends a request to evaluate apply(Module, Function, Args) on Node and returns immediately.

Collect the result with receive_response/1,2, wait_response/1,2 or check_response/2. Several requests may be outstanding at once: each carries its own correlation reference, so replies cannot be confused with one another, and a reply to a request you have abandoned is discarded rather than delivered to the next call.

The returned identifier is opaque and belongs to the calling process — only that process can collect the response.

send_request(Node, Module, Function, Args, Opts)

-spec send_request(Node, Module, Function, Args, Opts) -> RequestId
                      when
                          Node :: node(),
                          Module :: atom(),
                          Function :: atom(),
                          Args :: [term()],
                          Opts :: partisan_peer_service_manager:forward_opts(),
                          RequestId :: request_id().

Sends a request as send_request/4, with per-call transport options.

Partisan-specific; erpc has no equivalent. Opts is a partisan_peer_service_manager:forward_opts/0 map or proplist — most usefully channel, to keep a burst of asynchronous requests off the channel carrying latency-sensitive traffic. Per-call keys win over the global forward_options.

send_request/6

-spec send_request(Node :: node(),
                   Module :: atom(),
                   Function :: atom(),
                   Args :: [term()],
                   Label :: term(),
                   request_id_collection()) ->
                      request_id_collection().

Sends a request as send_request/4 and adds it to a collection under Label.

Use this to hold many outstanding requests together and collect them as they arrive, with receive_response/3, wait_response/3 or check_response/3. The label is returned alongside the result, so the caller can tell which request answered without tracking identifiers itself.

send_request/7

-spec send_request(Node :: node(),
                   Module :: atom(),
                   Function :: atom(),
                   Args :: [term()],
                   Label :: term(),
                   request_id_collection(),
                   Opts :: partisan_peer_service_manager:forward_opts()) ->
                      request_id_collection().

Sends a request as send_request/6, with per-call transport options.

Partisan-specific; erpc has no equivalent. This is the collection form, so it is the one to reach for when issuing a large fan-out that should not share a channel with everything else.

wait_response(RequestId)

-spec wait_response(RequestId) -> {response, Result} | no_response
                       when RequestId :: request_id(), Result :: term().

Equivalent to wait_response(RequestId, 0) — checks for a response without waiting.

wait_response(RequestId, WaitTime)

-spec wait_response(RequestId, WaitTime) -> {response, Result} | no_response
                       when
                           RequestId :: request_id(),
                           WaitTime :: 0..4294967295 | infinity,
                           Result :: term().

Waits up to WaitTime for the response to RequestId.

Returns {response, Result} if it arrives, or no_response if the wait expires. Unlike receive_response/2, expiry is not an error and does not abandon the request: the identifier stays valid and can be waited on again.

wait_response/3

-spec wait_response(request_id_collection(), WaitTime :: 0..4294967295 | infinity, Delete :: boolean()) ->
                       {{response, Result :: term()}, Label :: term(), request_id_collection()} |
                       no_response.

Waits up to WaitTime for the next response in a collection.

Returns {{response, Result}, Label, Collection} or no_response. As with wait_response/2, expiry is not an error and leaves the collection intact.