Implements a caller/client for the req node of the req/rep protocol. Sends data to the rep node (executor/server) and returns an Aio, which can be called for the value when required.
A signalling version of the function takes a 'conditionVariable' as an additional argument and signals it when the async receive is complete.
request(
context,
data,
send_mode = c("serial", "raw", "next"),
recv_mode = c("serial", "character", "complex", "double", "integer", "logical",
"numeric", "raw", "string"),
timeout = NULL,
ack = FALSE
)
request_signal(
context,
data,
cv,
send_mode = c("serial", "raw", "next"),
recv_mode = c("serial", "character", "complex", "double", "integer", "logical",
"numeric", "raw", "string"),
timeout = NULL
)a Context.
an object (if send_mode = 'raw', a vector).
[default 'serial'] one of 'serial' to send serialised R objects, 'raw' to send atomic vectors of any type as a raw byte vector, or 'next' to send in a new R-compatible serialisation format. Use 'serial' to ensure perfect reproducibility within R, although 'raw' must be used when interfacing with external applications which do not understand R serialisation. Alternatively, for performance, specify an integer position in the vector of choices e.g. 1L for 'serial' or 2L for 'raw' etc.
[default 'serial'] mode of vector to be received - one of 'serial', 'character', 'complex', 'double', 'integer', 'logical', 'numeric', 'raw', or 'string'. The default 'serial' means a serialised R object, for the other modes, the raw vector received will be converted into the respective mode. Note that 'string' is defined here as a character scalar and is a faster alternative to 'character' for receiving a single string. Alternatively, for performance, specify an integer position in the vector of choices e.g. 1L for 'serial', 2L for 'character' etc.
[default NULL] integer value in milliseconds or NULL, which applies a socket-specific default, usually the same as no timeout.
[default FALSE] logical value whether to send an ack(nowledgement) back to the rep node (consisting of an empty message) when the async receive is complete.
For the signalling version: a 'conditionVariable' to signal when the async receive is complete.
A 'recvAio' (object of class 'recvAio') (invisibly).
Sending the request and receiving the result are both performed async,
hence the function will return immediately with a 'recvAio' object. Access
the return value at $data.
This is designed so that the process on the server can run concurrently without blocking the client.
Optionally use call_aio on the 'recvAio' to call (and wait
for) the result.
If an error occured in the server process, a nul byte 00 will be
received. This allows an error to be easily distinguished from a NULL
return value. is_nul_byte can be used to test for a nul byte.
It is recommended to use a new context for each request to ensure consistent state tracking. For safety, the context used for the request is closed when all references to the returned 'recvAio' are removed and the object is garbage collected.
For the signalling version: when the receive is complete, the supplied 'conditionVariable' is signalled by incrementing its value by 1. This happens asynchronously and independently of the R execution thread.
req <- socket("req", listen = "tcp://127.0.0.1:6546")
rep <- socket("rep", dial = "tcp://127.0.0.1:6546")
# works if req and rep are running in parallel in different processes
reply(.context(rep), execute = function(x) x + 1, timeout = 50)
#> 'errorValue' int 5 | Timed out
aio <- request(.context(req), data = 2022)
aio$data
#> 'unresolved' logi NA
close(req)
close(rep)
# Signalling a condition variable
req <- socket("req", listen = "tcp://127.0.0.1:6546")
ctxq <- context(req)
cv <- cv()
aio <- request_signal(ctxq, data = 2022, cv = cv)
until(cv, 10L)
close(req)
# The following should be run in another process
# rep <- socket("rep", dial = "tcp://127.0.0.1:6546")
# ctxp <- context(rep)
# reply(ctxp, execute = function(x) x + 1)
# close(rep)