R binding for NNG (Nanomsg Next Gen), a successor to ZeroMQ. NNG is a socket library providing high-performance scalability protocols, implementing a cross-platform standard for messaging and communications. Serves as a concurrency framework for building distributed applications, utilising ‘aio’ objects which resolve automatically upon completion of asynchronous operations.
Designed for performance and reliability, the NNG library is written in C and {nanonext} is a lightweight zero-dependency wrapper. Provides the interface for code and processes to communicate with each other - receive data generated in Python, perform analysis in R, and send results to a C++ program – all on the same computer or on networks spanning the globe.
Implemented scalability protocols:
Supported transports:
Web utilities:
Install the latest release from CRAN:
install.packages("nanonext")or the development version from rOpenSci R-universe:
install.packages("nanonext", repos = "https://shikokuchuo.r-universe.dev"){nanonext} offers 2 equivalent interfaces: a functional interface, and an object-oriented interface.
The primary object in the functional interface is the Socket. Use
socket() to create a socket and dial or listen at an
address. The socket is then passed as the first argument of subsequent
actions such as send() or recv().
Example using Request/Reply (REQ/REP) protocol with inproc
transport:
(The inproc transport uses zero-copy where
possible for a much faster solution than alternatives)
Create sockets:
library(nanonext)
socket1 <- socket("req", listen = "inproc://nanonext")
socket2 <- socket("rep", dial = "inproc://nanonext")Send message from ‘socket1’:
send(socket1, "hello world!")
#> [1] 0Receive message using ‘socket2’:
recv(socket2)
#> [1] "hello world!"The primary object in the object-oriented interface is the nano
object. Use nano() to create a nano object which
encapsulates a Socket and Dialer/Listener. Methods such as
$send() or $recv() can then be accessed
directly from the object.
Example using Pipeline (Push/Pull) protocol with TCP/IP transport:
Create nano objects:
library(nanonext)
nano1 <- nano("push", listen = "tcp://127.0.0.1:5555")
nano2 <- nano("pull", dial = "tcp://127.0.0.1:5555")Send message from ‘nano1’:
nano1$send("hello world!")
#> [1] 0Receive message using ‘nano2’:
nano2$recv()
#> [1] "hello world!"{nanonext} provides a fast and reliable data interface between different programming languages where NNG has an implementation, including C, C++, Java, Python, Go, Rust etc.
The following example demonstrates the exchange of numerical data between R and Python (NumPy), two of the most commonly-used languages for data science and machine learning.
Using a messaging interface provides a clean and robust approach, light on resources with limited and identifiable points of failure.
This approach can also serve as an interface / pipe between different processes written in the same or different languages, running on the same computer or distributed across networks, and is an enabler of modular software design as espoused by the Unix philosophy.
One solution it provides is that of processing real-time data where computation times exceed the data frequency - by dividing the computation into stages, this may be set up as a pipeline or ‘cascade’ of processes, each connected using NNG sockets.
Create socket in Python using the NNG binding ‘pynng’:
import numpy as np
import pynng
socket = pynng.Pair0(listen="ipc:///tmp/nanonext.socket")Create nano object in R using {nanonext}, then send a vector of ‘doubles’, specifying mode as ‘raw’:
library(nanonext)
n <- nano("pair", dial = "ipc:///tmp/nanonext.socket")
n$send(c(1.1, 2.2, 3.3, 4.4, 5.5), mode = "raw")
#> [1] 0Receive in Python as a NumPy array of ‘floats’, and send back to R:
raw = socket.recv()
array = np.frombuffer(raw)
print(array)
#> [1.1 2.2 3.3 4.4 5.5]
msg = array.tobytes()
socket.send(msg)Receive in R, specifying the receive mode as ‘double’:
n$recv(mode = "double")
#> [1] 1.1 2.2 3.3 4.4 5.5{nanonext} implements true async send and receive, leveraging NNG as a massively-scaleable concurrency framework.
s1 <- socket("pair", listen = "inproc://nano")
s2 <- socket("pair", dial = "inproc://nano")send_aio() and recv_aio() functions return
immediately with an ‘Aio’ object, but perform their operations
async.
An ‘Aio’ object returns an unresolved value whilst its asynchronous operation is ongoing, automatically resolving to a final value once complete.
# an async receive is requested, but no messages are waiting (yet to be sent)
msg <- recv_aio(s2, keep.raw = TRUE)
msg
#> < recvAio >
#> - $raw for raw message
#> - $data for message data
msg$data
#> 'unresolved' logi NAFor a ‘sendAio’ object, the result is stored at
$result.
res <- send_aio(s1, data.frame(a = 1, b = 2))
res
#> < sendAio >
#> - $result for send result
res$result
#> [1] 0Note: a return value of 0 denotes a successful send, meaning that the message has been accepted by the socket for sending; the message itself may still be buffered within the system.
For a ‘recvAio’ object, the message is stored at $data,
and the raw message at $raw (if kept).
# now that a message has been sent, the 'recvAio' resolves automatically
msg$data
#> a b
#> 1 1 2
msg$raw
#> [1] 58 0a 00 00 00 03 00 04 02 02 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
#> [26] 03 13 00 00 00 02 00 00 00 0e 00 00 00 01 3f f0 00 00 00 00 00 00 00 00 00
#> [51] 0e 00 00 00 01 40 00 00 00 00 00 00 00 00 00 04 02 00 00 00 01 00 04 00 09
#> [76] 00 00 00 05 6e 61 6d 65 73 00 00 00 10 00 00 00 02 00 04 00 09 00 00 00 01
#> [101] 61 00 04 00 09 00 00 00 01 62 00 00 04 02 00 00 00 01 00 04 00 09 00 00 00
#> [126] 05 63 6c 61 73 73 00 00 00 10 00 00 00 01 00 04 00 09 00 00 00 0a 64 61 74
#> [151] 61 2e 66 72 61 6d 65 00 00 04 02 00 00 00 01 00 04 00 09 00 00 00 09 72 6f
#> [176] 77 2e 6e 61 6d 65 73 00 00 00 0d 00 00 00 02 80 00 00 00 ff ff ff ff 00 00
#> [201] 00 feAuxiliary function unresolved() may be used in control
flow statements to perform actions which depend on resolution of the
Aio, both before and after. This means there is no need to actually wait
(block) for an Aio to resolve, as the example below demonstrates.
msg <- recv_aio(s2)
# unresolved() queries for resolution itself so no need to use it again within the while loop
while (unresolved(msg)) {
# do stuff before checking resolution again
send_aio(s1, "resolved")
cat("unresolved")
}
#> unresolved
# perform actions which depend on the Aio value outside the while loop
msg$data
#> [1] "resolved"The values may also be called explicitly using
call_aio(). This will wait for completion of the Aio
(blocking).
# will wait for completion then return the resolved Aio
call_aio(msg)
# to access the resolved value directly (waiting if required)
call_aio(msg)$data
#> [1] "resolved"
close(s1)
close(s2){nanonext} implements remote procedure calls (RPC) using NNG’s req/rep protocol to provide a basis for distributed computing.
Can be used to perform computationally-expensive calculations or I/O-bound operations such as writing large amounts of data to disk in a separate ‘server’ process running concurrently.
[S] Server process: reply() will wait for a message and
apply a function, in this case rnorm(), before sending back
the result.
library(nanonext)
rep <- socket("rep", listen = "tcp://127.0.0.1:6546")
ctxp <- context(rep)
r <- reply(ctxp, execute = rnorm, send_mode = "raw")[C] Client process: request() performs an async send and
receive request and returns immediately with a recvAio
object.
library(nanonext)
req <- socket("req", dial = "tcp://127.0.0.1:6546")
ctxq <- context(req)
aio <- request(ctxq, data = 1e8, recv_mode = "double")At this point, the client can run additional code concurrent with the server processing the request.
# do more...When the result of the server calculation is required, the
recvAio may be called using call_aio().
The return value from the server request is then retrieved and stored
in the Aio as $data.
call_aio(aio)
aio
#> < recvAio >
#> - $data for message data
aio$data |> str()
#> num [1:100000000] 2.593 -0.328 0.571 0.646 1.092 ...As call_aio() is blocking and will wait for completion,
an alternative is to query aio$data directly. This will
return an ‘unresolved’ logical NA value if the calculation is yet to
complete.
In this example the calculation is returned, but other operations may reside entirely on the server side, for example writing data to disk.
In such a case, calling or querying the value confirms that the operation has completed, and provides the return value of the function, which may typically be NULL or an exit code.
The {mirai} package https://shikokuchuo.net/mirai/ (available on CRAN) uses {nanonext} as the back-end to provide asynchronous execution of arbitrary R code using the RPC model.
{nanonext} fully implements NNG’s pub/sub protocol as per the below example. A subscriber can subscribe to one or multiple topics broadcast by a publisher.
pub <- socket("pub", listen = "inproc://nanobroadcast")
sub <- socket("sub", dial = "inproc://nanobroadcast")
sub |> subscribe(topic = "examples")
pub |> send(c("examples", "this is an example"), mode = "raw")
#> [1] 0
sub |> recv(mode = "character")
#> [1] "examples" "this is an example"
pub |> send("examples at the start of a single text message", mode = "raw")
#> [1] 0
sub |> recv(mode = "character")
#> [1] "examples at the start of a single text message"
pub |> send(c("other", "this other topic will not be received"), mode = "raw")
#> [1] 0
sub |> recv(mode = "character")
#> 'errorValue' int 8 | Try again
# specify NULL to subscribe to ALL topics
sub |> subscribe(topic = NULL)
pub |> send(c("newTopic", "this is a new topic"), mode = "raw")
#> [1] 0
sub |> recv("character")
#> [1] "newTopic" "this is a new topic"
sub |> unsubscribe(topic = NULL)
pub |> send(c("newTopic", "this topic will now not be received"), mode = "raw")
#> [1] 0
sub |> recv("character")
#> 'errorValue' int 8 | Try again
# however the topics explicitly subscribed to are still received
pub |> send(c("examples will still be received"), mode = "raw")
#> [1] 0
sub |> recv(mode = "character")
#> [1] "examples will still be received"The subscribed topic can be of any atomic type (not just character), allowing integer, double, logical, complex and raw vectors to be sent and received.
sub |> subscribe(topic = 1)
pub |> send(c(1, 10, 10, 20), mode = "raw")
#> [1] 0
sub |> recv(mode = "double")
#> [1] 1 10 10 20
close(pub)
close(sub)This type of pattern is useful for applications such as service discovery.
A surveyor sends a survey, which is broadcast to all peer respondents. Respondents are then able to reply, but are not obliged to. The survey itself is a timed event, and responses received after the timeout are discarded.
sur <- socket("surveyor", listen = "inproc://nanoservice")
res1 <- socket("respondent", dial = "inproc://nanoservice")
res2 <- socket("respondent", dial = "inproc://nanoservice")
# sur sets a survey timeout, applying to this and subsequent surveys
sur |> survey_time(500)
# sur sends a message and then requests 2 async receives
sur |> send("service check")
#> [1] 0
aio1 <- sur |> recv_aio()
aio2 <- sur |> recv_aio()
# res1 receives the message and replies using an aio send function
res1 |> recv()
#> [1] "service check"
res1 |> send_aio("res1")
# res2 receives the message but fails to reply
res2 |> recv()
#> [1] "service check"
# checking the aio - only the first will have resolved
aio1$data
#> [1] "res1"
aio2$data
#> 'unresolved' logi NA
# after the survey expires, the second resolves into a timeout error
Sys.sleep(0.5)
aio2$data
#> 'errorValue' int 5 | Timed out
close(sur)
close(res1)
close(res2)Above it can be seen that the final value resolves into a timeout, which is an integer 5 classed as ‘errorValue’. All integer error codes are classed as ‘errorValue’ to be easily distinguishable from integer message values.
ncurl() is a minimalist http(s) client.
By setting async = TRUE, it performs requests
asynchronously, returning immediately with an ‘ncurlAio’.
For normal use, it takes just the URL. It can follow redirects.
ncurl("https://httpbin.org/headers")
#> $status
#> [1] 200
#>
#> $headers
#> NULL
#>
#> $raw
#> [1] 7b 0a 20 20 22 68 65 61 64 65 72 73 22 3a 20 7b 0a 20 20 20 20 22 48 6f 73
#> [26] 74 22 3a 20 22 68 74 74 70 62 69 6e 2e 6f 72 67 22 2c 20 0a 20 20 20 20 22
#> [51] 58 2d 41 6d 7a 6e 2d 54 72 61 63 65 2d 49 64 22 3a 20 22 52 6f 6f 74 3d 31
#> [76] 2d 36 33 63 35 31 62 33 37 2d 36 32 32 38 63 38 61 37 33 36 61 37 63 63 34
#> [101] 61 34 39 34 33 39 36 66 66 22 0a 20 20 7d 0a 7d 0a
#>
#> $data
#> [1] "{\n \"headers\": {\n \"Host\": \"httpbin.org\", \n \"X-Amzn-Trace-Id\": \"Root=1-63c51b37-6228c8a736a7cc4a494396ff\"\n }\n}\n"For advanced use, supports additional HTTP methods such as POST or PUT.
res <- ncurl("http://httpbin.org/post",
async = TRUE,
method = "POST",
headers = c(`Content-Type` = "application/json", Authorization = "Bearer APIKEY"),
data = '{"key": "value"}',
response = c("Date", "Server"))
res
#> < ncurlAio >
#> - $status for response status code
#> - $headers for response headers
#> - $raw for raw message
#> - $data for message data
call_aio(res)$headers
#> $Date
#> [1] "Mon, 16 Jan 2023 09:39:03 GMT"
#>
#> $Server
#> [1] "gunicorn/19.9.0"
res$data
#> [1] "{\n \"args\": {}, \n \"data\": \"{\\\"key\\\": \\\"value\\\"}\", \n \"files\": {}, \n \"form\": {}, \n \"headers\": {\n \"Authorization\": \"Bearer APIKEY\", \n \"Content-Length\": \"16\", \n \"Content-Type\": \"application/json\", \n \"Host\": \"httpbin.org\", \n \"X-Amzn-Trace-Id\": \"Root=1-63c51b37-0a9c25b56da3a6841d860396\"\n }, \n \"json\": {\n \"key\": \"value\"\n }, \n \"origin\": \"212.36.172.203\", \n \"url\": \"http://httpbin.org/post\"\n}\n"In this respect, it may be used as a performant and lightweight method for making REST API requests.
stream() exposes NNG’s low-level byte stream interface
for communicating with raw sockets. This may be used for connecting to
arbitrary non-NNG endpoints.
The stream interface can be used to communicate with (secure)
websocket servers. The argument textframes = TRUE can be
specified where the websocket server uses text rather than binary
frames.
# official demo API key used below
s <- stream(dial = "wss://ws.eodhistoricaldata.com/ws/forex?api_token=OeAFFmMliFG5orCUuwAKQ8l4WWFQ67YX",
textframes = TRUE)
s
#> < nanoStream >
#> - type: dialer
#> - url: wss://ws.eodhistoricaldata.com/ws/forex?api_token=OeAFFmMliFG5orCUuwAKQ8l4WWFQ67YX
#> - textframes: TRUEsend() and recv(), as well as their
asynchronous counterparts send_aio() and
recv_aio() can be used on Streams in the same way as
Sockets. This affords a great deal of flexibility in ingesting and
processing streaming data.
s |> recv()
#> [1] "{\"status_code\":200,\"message\":\"Authorized\"}"
s |> send('{"action": "subscribe", "symbols": "EURUSD"}')
#> [1] 0
s |> recv()
#> [1] "{\"s\":\"EURUSD\",\"a\":1.08243,\"b\":1.08241,\"dc\":\"0.0185\",\"dd\":\"0.0002\",\"ppms\":false,\"t\":1673861944000}"
s |> recv()
#> [1] "{\"s\":\"EURUSD\",\"a\":1.08242,\"b\":1.0824,\"dc\":\"0.0176\",\"dd\":\"0.0002\",\"ppms\":false,\"t\":1673861944000}"
close(s)Functions performing hashing using the SHA-2 series of algorithms is
included: sha224(), sha256(),
sha384() and sha512().
These call the secure, optimized implementations from the ‘Mbed TLS’ library and return a hash either directly as a raw vector or converted to a character string. For use in authentication, raw vectors can be compared directly for the highest performance.
To generate an HMAC (hash-based message authentication code), simply supply the value ‘key’ to use as the secret key.
sha256("hello world!")
#> [1] "7509e5bda0c762d2bac7f90d758b5b2263fa01ccbc542ab5e3df163be08e6ca9"
sha256("hello world!", convert = FALSE)
#> [1] 75 09 e5 bd a0 c7 62 d2 ba c7 f9 0d 75 8b 5b 22 63 fa 01 cc bc 54 2a b5 e3
#> [26] df 16 3b e0 8e 6c a9
sha256("hello world!", key = "MY_SECRET")
#> [1] "d8f0e2d368ff632682d55e2c1ccd49c15f8a6a3862d8eb68f1906b6ee658890a"Optimised functions for base64 encoding and decoding from the ‘Mbed TLS’ library are also provided:
base64enc("hello world!")
#> [1] "aGVsbG8gd29ybGQh"
base64dec(base64enc("hello world!"))
#> [1] "hello world!"Installation from source requires ‘libnng’ >= v1.6.0 and ‘libmbedtls’ >= 2 - suitable installations are automatically detected - or else ‘cmake’ to compile ‘libnng’ v1.6.0 pre-release (539e559) and ‘libmbedtls’ v3.2.1 included within the package sources.
Note: ‘libnng’ v1.6.0 is not yet available in system repositories; ‘libmbedtls’ is available as libmbedtls-dev (deb) or libmbedtls-devel (rpm).
The ‘INCLUDE_DIR’ and ‘LIB_DIR’ environment variables may be set prior to package installation to specify a custom location for ‘libmbedtls’ or ‘libnng’ other than the standard filesystem locations.
Package installation will automatically build the libraries if required.
Additional requirements for Solaris: (i) the ‘xz’ package - available on OpenCSW, and (ii) a more recent version of ‘cmake’ than that available on OpenCSW - see the ‘cmake’ website for the latest source file which can be built with just a C compiler.
For R >= 4.2 using the ‘rtools42’ toolchain, ‘libnng’ v1.6.0 (539e559) and ‘libmbedtls’ v3.2.1 will be automatically compiled from the package sources during installation.
For previous R versions, pre-compiled ‘libnng’ v1.6.0 (539e559) and ‘libmbedtls’ v3.2.1 libraries are downloaded and used for installation instead.
nanonext on CRAN: https://cran.r-project.org/package=nanonext
Package website: https://shikokuchuo.net/nanonext/
nanonext is listed in CRAN Task Views:
- High Performance
Computing: https://cran.r-project.org/view=HighPerformanceComputing
- Web Technologies: https://cran.r-project.org/view=WebTechnologies
NNG website: https://nng.nanomsg.org/
Mbed TLS website: https://www.trustedfirmware.org/projects/mbed-tls/
–
Please note that this project is released with a Contributor Code of Conduct. By participating in this project you agree to abide by its terms.