slightly better state for nano objects, more safety for nng_timer

Author: shikokuchuo

R/aio.R

@@ -62,6 +62,8 @@ call_aio <- function(aio) {
       }
       on.exit(expr = {
         aio[["raw"]] <- res
+        rm("aio", envir = aio)
+        rm("callparams", envir = aio)
         return(invisible(aio))
       })
       data <- switch(mode,

R/context.R

@@ -156,7 +156,7 @@ recv_ctx <- function(context,
 
 }
 
-#' Reply over Context (Server for Req/Rep Protocol)
+#' Reply over Context (RPC Server for Req/Rep Protocol)
 #'
 #' Implements an executor/server for the rep node of the req/rep protocol. Awaits
 #'     data, applies an arbitrary specified function, and returns the result
@@ -176,9 +176,10 @@ recv_ctx <- function(context,
 #'     The default 'serial' means a serialised R object, for the other modes,
 #'     the raw vector received will be converted into the respective mode.
 #' @param timeout in ms. If unspecified, a socket-specific default timeout will
-#'     be used. Note this applies to each of the receive and send legs, hence the
-#'     total elapsed time could be up to twice this parameter plus the time to
-#'     perform 'execute' on the received data.
+#'     be used. Note that this applies to receiving the request. The total elapsed
+#'     time would also include the time for performing 'execute' on the received
+#'     data. The timeout then also applies to sending the result (in the event
+#'     that the requestor has become unavailable since sending the request).
 #' @param ... additional arguments passed to the function specified by 'execute'.
 #'
 #' @return Invisible NULL.
@@ -187,11 +188,10 @@ recv_ctx <- function(context,
 #'     the desired behaviour. Set a timeout to allow the function to return
 #'     if no data is forthcoming.
 #'
-#'     In the event of an error in unserialisation or conversion of the
-#'     received message, in the evaluation of the function with respect to the
-#'     data, or in the serialization or conversion of the message to be sent,
-#'     a NULL byte (or serialized NULL byte) will be sent in reply to signal an
-#'     error to the client.
+#'     In the event of an error in either processing the messages or in evaluation
+#'     of the function with respect to the data, a nul byte \code{00} (or serialized
+#'     nul byte) will be sent in reply to the client to signal an error. This makes
+#'     it easy to distigush an error from a NULL return value.
 #'
 #' @examples
 #' req <- socket("req", listen = "tcp://127.0.0.1:6546")
@@ -229,7 +229,7 @@ reply <- function(context,
     message(res, " : ", nng_error(res))
     return(invisible(res))
   }
-  on.exit(expr = send_aio(context, writeBin(object = "", con = raw()), mode = send_mode))
+  on.exit(expr = send_aio(context, as.raw(0L), mode = send_mode))
   data <- switch(recv_mode,
                  serial = unserialize(connection = res),
                  character = (r <- readBin(con = res, what = recv_mode, n = length(res)))[r != ""],
@@ -249,7 +249,7 @@ reply <- function(context,
 
 }
 
-#' Request over Context (Client for Req/Rep Protocol)
+#' Request over Context (RPC Client for Req/Rep Protocol)
 #'
 #' 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
@@ -259,8 +259,7 @@ reply <- function(context,
 #' @inheritParams recv
 #' @param data an R object (if send_mode = 'raw', an R vector).
 #' @param timeout in ms. If unspecified, a socket-specific default timeout will
-#'     be used. Note this applies to each of the send and receive legs, hence the
-#'     total elapsed time could be up to twice this parameter.
+#'     be used. Note that this applies to receiving the result.
 #'
 #' @return A recv Aio (object of class 'recvAio').
 #'
@@ -271,8 +270,10 @@ reply <- function(context,
 #'     without blocking the client. Use \code{\link{call_aio}} on the 'recvAio'
 #'     to call the result when required.
 #'
-#'     If an error occured in the server process, a NULL byte will be received
-#'     (as \code{$data} if 'recv_mode' = 'serial', as \code{$raw} otherwise).
+#'     If an error occured in the server process, a nul byte \code{00} will be
+#'     received (as \code{$data} if 'recv_mode' = 'serial', as \code{$raw}
+#'     otherwise). This allows an error to be easily distinguished from a NULL
+#'     return value.
 #'
 #' @examples
 #' req <- socket("req", listen = "tcp://127.0.0.1:6546")

R/nano.R

@@ -68,8 +68,10 @@ nano <- function(protocol = c("pair", "bus", "push", "pull", "req", "rep",
       dial(nano, url = dial, autostart = TRUE)
     } else {
       dial(nano, url = dial, autostart = FALSE)
-      nano[["dialer_start"]] <- function(async = TRUE) start(.subset2(nano, "dialer")[[1L]],
-                                                             async = async)
+      nano[["dialer_start"]] <- function(async = TRUE) {
+        rm("dialer_start", envir = nano)
+        start(.subset2(nano, "dialer")[[1L]], async = async)
+      }
     }
   }
 
@@ -78,7 +80,10 @@ nano <- function(protocol = c("pair", "bus", "push", "pull", "req", "rep",
       listen(nano, url = listen, autostart = TRUE)
     } else {
       listen(nano, url = listen, autostart = FALSE)
-      nano[["listener_start"]] <- function() start(.subset2(nano, "listener")[[1L]])
+      nano[["listener_start"]] <- function() {
+        rm("listener_start", envir = nano)
+        start(.subset2(nano, "listener")[[1L]])
+      }
     }
   }
 

R/utils.R

@@ -69,7 +69,12 @@ nng_error <- function(error) {
 #'
 nng_timer <- function(time) {
 
-  invisible(.Call(rnng_threaded_timer, as.integer(time)))
+  if (is.numeric(time) && time >= 0) {
+    time <- as.integer(time)
+  } else {
+    stop("a numeric value >= 0 is required")
+  }
+  invisible(.Call(rnng_threaded_timer, time))
 
 }
 

README.Rmd

@@ -23,7 +23,7 @@ knitr::opts_chunk$set(
 
 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 that can be used for building distributed systems.
 
-Designed for performance and reliability, the NNG library is written in C and {nanonext} is a lightweight wrapper with no external package dependencies. Supported transports include inproc (intra-process), IPC (inter-process), TCP/IP (IPv4 or IPv6), and WebSocket.
+Designed for performance and reliability, the NNG library is written in C and {nanonext} is a lightweight wrapper depending on no other packages. Supported transports include inproc (intra-process), IPC (inter-process), TCP/IP (IPv4 or IPv6), and WebSocket. The inproc transport uses zero-copy where possible for a much faster solution than alternatives.
 
 Can be used for sending data across networks, but equally as an 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.
 
@@ -117,7 +117,7 @@ recv(socket2)
 
 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 that is light on resources and provides fewer points of failure. This is especially relevant when processing real-time data, as an example. 
+Using a messaging interface provides a clean and robust approach which is light on resources and offers limited and identifiable points of failure. This is especially relevant when processing real-time data, as an example. 
 
 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.
 
@@ -169,7 +169,7 @@ s2 <- socket("pair", dial = "inproc://nano")
 
 ```
 
-For a 'sendAio' object, calling the result causes it to be stored in the AIO as `$result`. 0 denotes a successful send.
+For a 'sendAio' object, calling the result causes it to be stored in the AIO as `$result`. An exit code of 0 denotes a successful send.
 
 ```{r async2}
 
@@ -226,18 +226,20 @@ reply(ctxp, execute = rnorm, send_mode = "raw")
 Client process: `request()` performs an async send and receive request and returns immediately with a `recvAio` object.
 
 ```{r rpcclient}
+
 library(nanonext)
 req <- socket("req", dial = "tcp://127.0.0.1:6546")
 ctxq <- context(req)
-aio <- request(ctxq, data = 1e6, recv_mode = "double", keep.raw = FALSE)
+aio <- request(ctxq, data = 1e8, recv_mode = "double", keep.raw = FALSE)
+
 ```
 
 At this point, the client can run additional code concurrent with the server processing the request. 
 ```{r rpcclient2}
 # do more...
 ```
 
-When the result of the server calculation is required (or, as the case may be, an exit code confirming that the server operation has completed), the `recvAio` may be called using `call_aio()`. 
+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`.
 
@@ -250,6 +252,10 @@ str(aio$data)
 
 ```
 
+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, using `call_aio()` confirms that the operation has completed (or it will wait for completion) and calls the return value of the function, which may typically be NULL or an exit code.
+
 [&laquo; Back to ToC](#table-of-contents)
 
 ### Publisher Subscriber Model

README.md

@@ -19,9 +19,11 @@ Serves as a concurrency framework that can be used for building
 distributed systems.
 
 Designed for performance and reliability, the NNG library is written in
-C and {nanonext} is a lightweight wrapper with no external package
-dependencies. Supported transports include inproc (intra-process), IPC
-(inter-process), TCP/IP (IPv4 or IPv6), and WebSocket.
+C and {nanonext} is a lightweight wrapper depending on no other
+packages. Supported transports include inproc (intra-process), IPC
+(inter-process), TCP/IP (IPv4 or IPv6), and WebSocket. The inproc
+transport uses zero-copy where possible for a much faster solution than
+alternatives.
 
 Can be used for sending data across networks, but equally as an
 interface for code and processes to communicate with each other. Receive
@@ -150,9 +152,10 @@ 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 that is
-light on resources and provides fewer points of failure. This is
-especially relevant when processing real-time data, as an example.
+Using a messaging interface provides a clean and robust approach which
+is light on resources and offers limited and identifiable points of
+failure. This is especially relevant when processing real-time data, as
+an example.
 
 This approach can also serve as an interface / pipe between different
 processes written in the same or different languages, running on the
@@ -219,7 +222,7 @@ s2 <- socket("pair", dial = "inproc://nano")
 ```
 
 For a ‘sendAio’ object, calling the result causes it to be stored in the
-AIO as `$result`. 0 denotes a successful send.
+AIO as `$result`. An exit code of 0 denotes a successful send.
 
 ``` r
 res <- send_aio(s1, data.frame(a = 1, b = 2))
@@ -292,7 +295,7 @@ 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 = 1e6, recv_mode = "double", keep.raw = FALSE)
+aio <- request(ctxq, data = 1e8, recv_mode = "double", keep.raw = FALSE)
 ```
 
 At this point, the client can run additional code concurrent with the
@@ -302,9 +305,8 @@ server processing the request.
 # do more...
 ```
 
-When the result of the server calculation is required (or, as the case
-may be, an exit code confirming that the server operation has
-completed), the `recvAio` may be called using `call_aio()`.
+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`.
@@ -316,9 +318,16 @@ aio
 #> < recvAio >
 #>  - $data for message data
 str(aio$data)
-#>  num [1:1000000] -1.014 -0.8599 -1.7137 1.9008 0.0125 ...
+#>  num [1:100000000] -1.104 1.34 0.442 -0.738 0.66 ...
 ```
 
+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, using `call_aio()` confirms that the operation has
+completed (or it will wait for completion) and calls the return value of
+the function, which may typically be NULL or an exit code.
+
 [« Back to ToC](#table-of-contents)
 
 ### Publisher Subscriber Model
@@ -378,11 +387,11 @@ ncurl("http://httpbin.org/headers")
 #>   [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 32 30 32 35 35 39 30 2d 34 65 37 31 61 37 65 37 37 30 63 65 31 64 36
-#> [101] 64 30 61 36 61 62 37 39 39 22 0a 20 20 7d 0a 7d 0a
+#>  [76] 2d 36 32 30 34 30 34 66 65 2d 30 62 39 31 61 34 61 63 32 61 36 62 31 36 34
+#> [101] 31 36 61 30 32 38 30 30 63 22 0a 20 20 7d 0a 7d 0a
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62025590-4e71a7e770ce1d6d0a6ab799\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-620404fe-0b91a4ac2a6b16416a02800c\"\n  }\n}\n"
 ```
 
 [« Back to ToC](#table-of-contents)