Implement Daemon IDs (#182)

* daemon id concept

* add tests

* code formatting

* more efficient implementation

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: mirai
 Type: Package
 Title: Minimalist Async Evaluation Framework for R
-Version: 1.3.1.9028
+Version: 1.3.1.9029
 Description: Designed for simplicity, a 'mirai' evaluates an R expression
     asynchronously in a parallel process, locally or distributed over the
     network, with the result automatically available upon completion. Modern

NEWS.md

@@ -1,4 +1,4 @@
-# mirai 1.3.1.9028 (development)
+# mirai 1.3.1.9029 (development)
 
 #### New Architecture
 
@@ -24,6 +24,7 @@
 * `ssh_config()` simplified to take the argument 'port' instead of 'host'. For SSH tunnelling, this is the port that will be used, and the hostname is now required to be '127.0.0.1' (no longer accepting 'localhost'). 
 * `daemon()` gains the new argument 'dispatcher', which should be set to `TRUE` when connecting to dispatcher and `FALSE` when connecting directly to host.
 * `daemon()` '...' argument has been moved up to prevent partial matching on any of the optional arguments.
+* `daemon()` gains argument 'id' which accept an integer value that allows `status()` to track connection and disconnection events.
 * `host_url()` argument 'ws' is removed as a TCP URL is now always recommended (although websocket URLs are still supported).
 * `saisei()` is defunct as no longer required, but still available for use with the old v1 dispatcher.
 * `daemons(dispatcher = "thread")` (experimental threaded dispatcher) has been retired - as this was based on the old dispatcher architecture and future development will focus on the current design. Specifying 'dispatcher = thread' is defunct, but will point to 'dispatcher = process' for the time being.

R/daemon.R

@@ -60,6 +60,9 @@
 #'   task (idle time) before exiting.
 #' @param walltime [default Inf] integer milliseconds soft walltime (time limit)
 #'   i.e. the minimum amount of real time elapsed before exiting.
+#' @param id [default NULL] (optional) integer daemon ID provided to dispatcher
+#'   to track connection status. Causes \code{\link{status}} to report this ID
+#'   under \code{$events} when the daemon connects and disconnects.
 #' @param tls [default NULL] required for secure TLS connections over
 #'   'tls+tcp://'. \strong{Either} the character path to a file containing X.509
 #'   certificate(s) in PEM format, comprising the certificate authority
@@ -94,7 +97,7 @@
 #'
 daemon <- function(url, dispatcher = FALSE, ..., asyncdial = FALSE, autoexit = TRUE,
                    cleanup = TRUE, output = FALSE, maxtasks = Inf, idletime = Inf,
-                   walltime = Inf, tls = NULL, rs = NULL) {
+                   walltime = Inf, id = NULL, tls = NULL, rs = NULL) {
 
   missing(dispatcher) && return(
     v1_daemon(url = url, asyncdial = asyncdial, autoexit = autoexit,
@@ -128,6 +131,8 @@ daemon <- function(url, dispatcher = FALSE, ..., asyncdial = FALSE, autoexit = T
 
   if (dispatcher) {
     aio <- recv_aio(sock, mode = 1L, cv = cv)
+    if (is.numeric(id))
+      send(sock, c(.intmax, as.integer(id)), mode = 2L, block = TRUE)
     wait(cv) || return()
     serial <- collect_aio(aio)
     if (is.list(serial))

R/daemons.R

@@ -425,6 +425,13 @@ with.miraiDaemons <- function(data, expr, ...) {
 #'     the result has been received (either completed or cancelled).
 #'   }
 #'
+#' @section Events:
+#'
+#'   If dispatcher is used combined with daemon IDs, an additional element
+#'   \strong{events} will report the positive integer ID when the daemon
+#'   connects and the negative value when it disconnects. Only the events since
+#'   the previous status query are returned.
+#'
 #' @examples
 #' if (interactive()) {
 #' # Only run examples in interactive R sessions
@@ -606,11 +613,14 @@ query_status <- function(envir) {
 
 dispatcher_status <- function(envir) {
   status <- query_dispatcher(envir[["sock"]], c(0L, 0L))
-  list(connections = status[1L],
-       daemons = envir[["urls"]],
-       mirai = c(awaiting = status[2L],
-                 executing = status[3L],
-                 completed = envir[["msgid"]] - status[2L] - status[3L]))
+  out <- list(connections = status[1L],
+              daemons = envir[["urls"]],
+              mirai = c(awaiting = status[2L],
+                        executing = status[3L],
+                        completed = envir[["msgid"]] - status[2L] - status[3L]))
+  if (length(status) > 3L)
+    out <- c(out, list(events = status[4:length(status)]))
+  out
 }
 
 ._scm_. <- as.raw(c(0x42, 0x0a, 0x03, 0x00, 0x00, 0x00, 0x02, 0x03, 0x04, 0x00, 0x00, 0x05, 0x03, 0x00, 0x05, 0x00, 0x00, 0x00, 0x55, 0x54, 0x46, 0x2d, 0x38, 0xfc, 0x00, 0x00, 0x00))

R/dispatcher.R

@@ -102,6 +102,7 @@ dispatcher <- function(host, url = NULL, n = NULL, ..., tls = NULL, pass = NULL,
 
   msgid <- 0L
   inq <- outq <- list()
+  events <- integer()
   envir <- new.env(hash = FALSE)
   if (is.numeric(rs)) `[[<-`(envir, "stream", as.integer(rs))
   if (auto) {
@@ -146,6 +147,8 @@ dispatcher <- function(host, url = NULL, n = NULL, ..., tls = NULL, pass = NULL,
             if (length(outq[[id]])) {
               if (outq[[id]][["msgid"]])
                 send(outq[[id]][["ctx"]], .connectionReset, mode = 1L, block = TRUE)
+              if (length(outq[[id]][["dmnid"]]))
+                events <- c(events, outq[[id]][["dmnid"]])
               outq[[id]] <- NULL
             }
           }
@@ -162,8 +165,10 @@ dispatcher <- function(host, url = NULL, n = NULL, ..., tls = NULL, pass = NULL,
             found <- c(
               length(outq),
               length(inq),
-              sum(as.logical(unlist(lapply(outq, .subset2, "msgid"), use.names = FALSE)))
+              sum(as.logical(unlist(lapply(outq, .subset2, "msgid"), use.names = FALSE))),
+              events
             )
+            events <- integer()
           } else {
             found <- FALSE
             for (i in seq_along(outq))
@@ -199,11 +204,21 @@ dispatcher <- function(host, url = NULL, n = NULL, ..., tls = NULL, pass = NULL,
           cv_signal(cv)
           next
         }
-        send(outq[[id]][["ctx"]], value, mode = 2L, block = TRUE)
-        outq[[id]][["msgid"]] <- 0L
         if (value[4L]) {
+          if (value[4L] > 1L) {
+            dmnid <- readBin(value, "integer", n = 2L)[2L]
+            events <- c(events, dmnid)
+            outq[[id]][["dmnid"]] <- -dmnid
+            next
+          }
+          send(outq[[id]][["ctx"]], value, mode = 2L, block = TRUE)
           send(psock, ._scm_., mode = 2L, pipe = outq[[id]][["pipe"]], block = TRUE)
+          if (length(outq[[id]][["dmnid"]]))
+            events <- c(events, outq[[id]][["dmnid"]])
           outq[[id]] <- NULL
+        } else {
+          send(outq[[id]][["ctx"]], value, mode = 2L, block = TRUE)
+          outq[[id]][["msgid"]] <- 0L
         }
       }
 

man/daemon.Rd

@@ -279,16 +279,20 @@ connection && requireNamespace("promises", quietly = TRUE) && Sys.getenv("NOT_CR
 # mirai daemon limits tests
 connection && Sys.getenv("NOT_CRAN") == "true" && {
   Sys.sleep(1L)
-  test_equal(daemons(1, cleanup = FALSE, maxtasks = 2L), 1L)
+  test_equal(daemons(1, cleanup = FALSE, maxtasks = 2L, id = 125L), 1L)
   test_equal(mirai(1)[], mirai(1)[])
   m <- mirai(0L)
   Sys.sleep(1L)
-  test_zero(status()$connections)
+  res <- status()
+  test_zero(res$connections)
+  test_identical(res$events, c(125L, -125L))
   test_equal(status()$mirai[["awaiting"]], 1L)
-  test_equal(launch_local(1, idletime = 5000L, walltime = 500L), 1L)
+  test_equal(launch_local(1, idletime = 5000L, walltime = 500L, id = 129L), 1L)
   test_zero(m[])
   Sys.sleep(1L)
-  test_zero(status()$connections)
+  res <- status()
+  test_zero(res$connections)
+  test_identical(res$events, c(129L, -129L))
   test_zero(daemons(0))
 }
 # mirai cancellation tests