additional robustness

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: mirai
 Type: Package
 Title: Minimalist Async Evaluation Framework for R
-Version: 0.10.0.9010
+Version: 0.10.0.9011
 Description: Lightweight parallel code execution and distributed computing.
     Designed for simplicity, a 'mirai' evaluates an R expression asynchronously,
     on local or network resources, resolving automatically upon completion.

NEWS.md

@@ -1,8 +1,9 @@
-# mirai 0.10.0.9010 (development)
+# mirai 0.10.0.9011 (development)
 
 * `daemon()` argument 'exitlinger' retired as daemons now synchronise with the host/dispatcher and exit as soon as possible.
 * Optimises scheduling at dispatcher: tasks are no longer assigned to a daemon if it is exiting due to specified time/task-outs.
-* An appropriate 'errorValue' is now returned if an ephemeral daemon dies (rather than remaining unresolved).
+* An 'errorValue' 19 'Connection reset' is now returned for a 'mirai' if the connection to either dispatcher or an ephemeral daemon drops, for example if they have crashed, rather than remaining unresolved.
+* Eliminates a potential crash in the host process after querying `status()` if there is no longer a connection to dispatcher.
 * Deprecated use of alias `server()` for `daemon()` is retired.
 * Requires nanonext >= [0.10.0.9009].
 

R/mirai.R

@@ -816,23 +816,24 @@ daemons <- function(n, url = NULL, dispatcher = TRUE, seed = NULL, tls = NULL, p
       length(envir[["n"]]) || return(0L)
 
       reap(envir[["sock"]])
-      length(envir[["sockc"]]) && reap(envir[["sockc"]])
+      length(envir[["sockc"]]) && attr(envir[["sockc"]], "state") == "opened" && reap(envir[["sockc"]])
       envir <- NULL
       `[[<-`(.., .compute, new.env(hash = FALSE))
 
     } else if (is.null(envir[["sock"]])) {
 
       n > 0L || stop(.messages[["n_zero"]])
       urld <- auto_tokenized_url()
-      sock <- req_socket(urld)
       create_stream(n = n, seed = seed, envir = envir)
       if (dispatcher) {
+        sock <- req_socket(urld, resend = 0L)
         urlc <- strcat(urld, "c")
         sockc <- req_socket(urlc, resend = 0L)
         launch_and_sync_daemon(sock = sock, urld, parse_dots(...), n, urlc, rs = envir[["stream"]])
         for (i in seq_len(n)) next_stream(envir)
         init_monitor(sockc = sockc, envir = envir)
       } else {
+        sock <- req_socket(urld)
         if (is.null(seed) || n == 1L) {
           for (i in seq_len(n))
             launch_daemon(urld, parse_dots(...), next_stream(envir))
@@ -1483,7 +1484,10 @@ query_dispatcher <- function(sock, command, mode) {
 
 query_status <- function(envir) {
   res <- query_dispatcher(sock = envir[["sockc"]], command = 0L, mode = 5L)
-  is.object(res) && return(res)
+  is.object(res) && {
+    res != 7L && reap(envir[["sockc"]])
+    return(res)
+  }
   `attributes<-`(res, list(dim = c(envir[["n"]], 5L),
                            dimnames = list(envir[["urls"]], c("i", "online", "instance", "assigned", "complete"))))
 }

tests/tests.R

@@ -149,9 +149,13 @@ if (Sys.getenv("NOT_CRAN") == "true" && .Platform[["OS.type"]] != "windows") {
   nanotesto(daemons(url = "wss://127.0.0.1:0", token = TRUE, pass = "test"))
   nanotestn(launch_local(1L))
   Sys.sleep(1L)
+  mq <- mirai(Sys.sleep(1L))
   nanotest(grepl("CERTIFICATE", launch_remote(1L), fixed = TRUE))
   nanotesterr(launch_local(0:1), "out of bounds")
   nanotesterr(launch_remote(1:2), "out of bounds")
+  nanotest(is.character(saisei(i = 1L, force = TRUE)))
+  Sys.sleep(0.5)
+  nanotest(is_error_value(mq[["data"]]))
   nanotestz(daemons(0L))
   Sys.sleep(1L)
   option <- 15L