serious performance tuning

Author: shikokuchuo

R/aio.R

@@ -49,7 +49,7 @@ send_aio <- function(con, data, mode = c("serial", "raw"), timeout = -2L) UseMet
 #'
 send_aio.nanoSocket <- function(con, data, mode = c("serial", "raw"), timeout = -2L) {
 
-  if (missing(mode) || match.arg1(mode) == 1L)
+  if (.Call(rnng_serial, mode))
     data <- serialize(object = data, connection = NULL)
   aio <- .Call(rnng_send_aio, con, data, timeout)
   is.integer(aio) && return(aio)
@@ -77,7 +77,7 @@ send_aio.nanoSocket <- function(con, data, mode = c("serial", "raw"), timeout =
 #'
 send_aio.nanoContext <- function(con, data, mode = c("serial", "raw"), timeout = -2L) {
 
-  if (missing(mode) || match.arg1(mode) == 1L)
+  if (.Call(rnng_serial, mode))
     data <- serialize(object = data, connection = NULL)
   aio <- .Call(rnng_ctx_send_aio, con, data, timeout)
   is.integer(aio) && return(aio)
@@ -200,8 +200,7 @@ recv_aio.nanoSocket <- function(con,
                                 keep.raw = TRUE,
                                 ...) {
 
-  mode <- match.arg2(mode, c("serial", "character", "complex", "double",
-                             "integer", "logical", "numeric", "raw"))
+  mode <- .Call(rnng_matcharg, mode)
   aio <- .Call(rnng_recv_aio, con, timeout)
   is.integer(aio) && return(aio)
 
@@ -258,8 +257,7 @@ recv_aio.nanoContext <- function(con,
                                  keep.raw = TRUE,
                                  ...) {
 
-  mode <- match.arg2(mode, c("serial", "character", "complex", "double",
-                             "integer", "logical", "numeric", "raw"))
+  mode <- .Call(rnng_matcharg, mode)
   aio <- .Call(rnng_ctx_recv_aio, con, timeout)
   is.integer(aio) && return(aio)
 
@@ -317,8 +315,7 @@ recv_aio.nanoStream <- function(con,
                                 n = 65536L,
                                 ...) {
 
-  mode <- match.arg2(mode, c("character", "complex", "double", "integer",
-                             "logical", "numeric", "raw")) + 1L
+  mode <- .Call(rnng_matchargs, mode)
   aio <- .Call(rnng_stream_recv_aio, con, n, timeout)
   is.integer(aio) && return(aio)
 

R/context.R

@@ -113,18 +113,15 @@ reply <- function(context,
                   timeout = -2L,
                   ...) {
 
-  recv_mode <- match.arg2(recv_mode, c("serial", "character", "complex", "double",
-                                       "integer", "logical", "numeric", "raw"))
   res <- .Call(rnng_ctx_recv, context, recv_mode, timeout, FALSE)
   is_error_value(res) && return(invisible(res))
   on.exit(expr = send_aio(context, as.raw(0L), mode = send_mode))
   data <- execute(res, ...)
-  if (missing(send_mode) || match.arg1(send_mode) == 1L)
+  if (.Call(rnng_serial, send_mode))
     data <- serialize(object = data, connection = NULL)
-  res <- .Call(rnng_ctx_send, context, data, timeout)
+  res <- .Call(rnng_ctx_send, context, data, timeout, FALSE)
   on.exit()
-  is.integer(res) && return(invisible(res))
-  invisible(0L)
+  if (missing(res)) invisible(0L) else invisible(res)
 
 }
 
@@ -183,9 +180,8 @@ request <- function(context,
                     timeout = -2L,
                     keep.raw = TRUE) {
 
-  recv_mode <- match.arg2(recv_mode, c("serial", "character", "complex", "double",
-                                       "integer", "logical", "numeric", "raw"))
-  if (missing(send_mode) || match.arg1(send_mode) == 1L)
+  recv_mode <- .Call(rnng_matcharg, recv_mode)
+  if (.Call(rnng_serial, send_mode))
     data <- serialize(object = data, connection = NULL)
   res <- .Call(rnng_ctx_send_aio, context, data, -2L)
   is.integer(res) && return(res)

R/messenger.R

@@ -29,7 +29,7 @@ messenger <- function(url) {
   sock <- .Call(rnng_messenger, url)
   is.integer(sock) && return(invisible(sock))
   on.exit(expr = {
-    s <- suppressWarnings(.Call(rnng_send, sock, writeBin(":d ", raw()), 0L))
+    suppressWarnings(.Call(rnng_send, sock, writeBin(":d ", raw()), 0L, FALSE))
     .Call(rnng_close, sock)
     invisible()
   })
@@ -43,7 +43,7 @@ messenger <- function(url) {
   cat(sprintf("\n| url: %s\n", url), file = stdout())
   cat("| connecting... ", file = stderr())
 
-  s <- suppressWarnings(.Call(rnng_send, sock, writeBin(":c ", raw()), 1000L))
+  s <- suppressWarnings(.Call(rnng_send, sock, writeBin(":c ", raw()), 1000L, TRUE))
   if (is.integer(s)) {
     cat(sprintf("\r| peer offline: %s\n", format.POSIXct(Sys.time())), file = stderr())
   } else {
@@ -55,7 +55,7 @@ messenger <- function(url) {
     data <- readline()
     if (identical(data, ":q")) break
     if (identical(data, "")) next
-    s <- suppressWarnings(.Call(rnng_send, sock, data, 0L))
+    s <- suppressWarnings(.Call(rnng_send, sock, data, 0L, TRUE))
     if (is.integer(s)) {
       cat(sprintf("%*s > not sent: peer offline: %s\n", nchar(data), "", format.POSIXct(Sys.time())),
           file = stderr())

R/sendrecv.R

@@ -74,11 +74,10 @@ send.nanoSocket <- function(con,
                             block = FALSE,
                             echo = TRUE) {
 
-  if (missing(mode) || match.arg1(mode) == 1L)
+  if (.Call(rnng_serial, mode))
     data <- serialize(object = data, connection = NULL)
-  res <- .Call(rnng_send, con, data, block)
-  is.integer(res) && return(res)
-  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
+  res <- .Call(rnng_send, con, data, block, echo)
+  if (missing(res)) invisible(0L) else res
 
 }
 
@@ -92,12 +91,10 @@ send.nanoContext <- function(con,
                              block = TRUE,
                              echo = TRUE) {
 
-  if (missing(mode) || match.arg1(mode) == 1L)
+  if (.Call(rnng_serial, mode))
     data <- serialize(object = data, connection = NULL)
-  if (missing(block) || isTRUE(block)) block <- -2L
-  res <- .Call(rnng_ctx_send, con, data, block)
-  is.integer(res) && return(res)
-  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
+  res <- .Call(rnng_ctx_send, con, data, block, echo)
+  if (missing(res)) invisible(0L) else res
 
 }
 
@@ -111,14 +108,11 @@ send.nanoStream <- function(con,
                             block = TRUE,
                             echo = TRUE) {
 
-  if (missing(block) || isTRUE(block)) block <- -2L
-  res <- .Call(rnng_stream_send, con, data, block)
-  is.integer(res) && return(res)
-  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
+  res <- .Call(rnng_stream_send, con, data, block, echo)
+  if (missing(res)) invisible(0L) else res
 
 }
 
-
 #' Receive
 #'
 #' Receive data over a connection (Socket, Context or Stream).
@@ -220,8 +214,6 @@ recv.nanoSocket <- function(con,
                             keep.raw = TRUE,
                             ...) {
 
-  mode <- match.arg2(mode, c("serial", "character", "complex", "double",
-                             "integer", "logical", "numeric", "raw"))
   .Call(rnng_recv, con, mode, block, keep.raw)
 
 }
@@ -237,9 +229,6 @@ recv.nanoContext <- function(con,
                              keep.raw = TRUE,
                              ...) {
 
-  mode <- match.arg2(mode, c("serial", "character", "complex", "double",
-                             "integer", "logical", "numeric", "raw"))
-  if (missing(block) || isTRUE(block)) block <- -2L
   .Call(rnng_ctx_recv, con, mode, block, keep.raw)
 
 }
@@ -256,9 +245,6 @@ recv.nanoStream <- function(con,
                             n = 65536L,
                             ...) {
 
-  mode <- match.arg2(mode, c("character", "complex", "double", "integer",
-                             "logical", "numeric", "raw")) + 1L
-  if (missing(block) || isTRUE(block)) block <- -2L
   .Call(rnng_stream_recv, con, mode, block, keep.raw, n)
 
 }

R/utils.R

@@ -264,9 +264,3 @@ match.arg2 <- function(choice, choices) {
   index
 }
 
-match.arg1 <- function(choice) {
-  index <- pmatch(choice[1L], c("serial", "raw"), nomatch = 0L, duplicates.ok = TRUE)
-  index || stop(sprintf("'%s' should be one of serial, raw", deparse(substitute(choice))))
-  index
-}
-

README.Rmd

@@ -107,7 +107,7 @@ nano2$recv()
 
 #### Functional Interface 
 
-The primary object in the functional interface is the Socket. Use `socket()` to create a socket, and optionally dial or listen at an address. The socket is then passed as the first argument of subsequent actions such as `send()` or `recv()`.
+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 Pipeline (Push/Pull) protocol with TCP/IP transport:*
 
@@ -268,7 +268,7 @@ close(s2)
 
 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.
 
-Server process: `reply()` will wait for a message and apply a function, in this case `rnorm()`, before sending back the result.
+[S] Server process: `reply()` will wait for a message and apply a function, in this case `rnorm()`, before sending back the result.
 
 ```{r rpcserver, eval=FALSE}
 
@@ -279,7 +279,7 @@ reply(ctxp, execute = rnorm, send_mode = "raw")
 
 ```
 
-Client process: `request()` performs an async send and receive request and returns immediately with a `recvAio` object.
+[C] Client process: `request()` performs an async send and receive request and returns immediately with a `recvAio` object.
 
 ```{r rpcclient}
 

README.md

@@ -126,9 +126,9 @@ nano2$recv()
 #### Functional Interface
 
 The primary object in the functional interface is the Socket. Use
-`socket()` to create a socket, and optionally dial or listen at an
-address. The socket is then passed as the first argument of subsequent
-actions such as `send()` or `recv()`.
+`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 Pipeline (Push/Pull) protocol with TCP/IP transport:*
 
@@ -339,8 +339,8 @@ 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.
 
-Server process: `reply()` will wait for a message and apply a function,
-in this case `rnorm()`, before sending back the result.
+\[S\] Server process: `reply()` will wait for a message and apply a
+function, in this case `rnorm()`, before sending back the result.
 
 ``` r
 library(nanonext)
@@ -349,8 +349,8 @@ ctxp <- context(rep)
 reply(ctxp, execute = rnorm, send_mode = "raw") 
 ```
 
-Client process: `request()` performs an async send and receive request
-and returns immediately with a `recvAio` object.
+\[C\] Client process: `request()` performs an async send and receive
+request and returns immediately with a `recvAio` object.
 
 ``` r
 library(nanonext)
@@ -379,7 +379,7 @@ aio
 #> < recvAio >
 #>  - $data for message data
 aio$data |> str()
-#>  num [1:100000000] -1.072 1.447 0.962 -0.483 -0.843 ...
+#>  num [1:100000000] 2.698 0.0645 -1.0922 2.1422 0.8848 ...
 ```
 
 As `call_aio()` is blocking and will wait for completion, an alternative
@@ -516,11 +516,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 36 35 62 32 62 31 2d 33 33 30 61 30 30 30 37 32 33 61 32 61 39 63
-#> [101] 36 36 61 30 36 66 38 36 61 22 0a 20 20 7d 0a 7d 0a
+#>  [76] 2d 36 32 36 37 31 36 30 32 2d 32 35 64 63 33 32 39 63 30 39 35 34 64 36 33
+#> [101] 39 37 37 62 34 30 35 32 39 22 0a 20 20 7d 0a 7d 0a
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-6265b2b1-330a000723a2a9c66a06f86a\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62671602-25dc329c0954d63977b40529\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.
@@ -535,7 +535,7 @@ res
 #>  - $raw for raw message
 
 call_aio(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-6265b2b1-09b62748201d88f5057952ab\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"78.145.225.121\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
+#> [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-62671603-121afeba10cc694a091360a9\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"78.145.225.121\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
 ```
 
 In this respect, it may be used as a performant and lightweight method