Launch local daemons from host not dispatcher (#403)

* Launch local daemons from host not dispatcher

* Simplify a bit

Author: shikokuchuo

NEWS.md

@@ -18,6 +18,7 @@
   If the serialization hook function errors or otherwise fails to return a raw vector, this will error out rather than be silently ignored (thanks @dipterix, #378).
 * `require_daemons()` arguments are swapped so that `.compute` comes before `call` for ease of use.
   Previous usage will work for the time being, although is deprecated and will be defunct in a future version.
+* Fixes `daemons(n)` failing to launch local daemons if mirai was installed in a custom user library set by an explicit `.libPaths()` call in '.Rprofile' (thanks @erydit and @dpastoor, #390).
 * Enhancements to `everywhere()`:
   + Consecutive `everywhere()` calls are permissible again when using dispatcher (behaviour update in v2.4.1) (#354).
   + No longer has any effect on the RNG stream when using a reproducible `seed` value at `daemons()` (#356).

R/daemons.R

@@ -621,23 +621,13 @@ args_daemon_disp <- function(url, dots, rs = NULL, tls = NULL) {
   shQuote(sprintf("mirai::daemon(\"%s\"%s%s)", url, dots, parse_tls(tls)))
 }
 
-args_dispatcher_local <- function(urld, n, dots) {
+args_dispatcher <- function(urld, url, n) {
   shQuote(sprintf(
-    ".libPaths(c(\"%s\",.libPaths()));mirai::dispatcher(\"%s\",n=%d%s)",
-    libp(),
-    urld,
-    n,
-    dots
-  ))
-}
-
-args_dispatcher_remote <- function(urld, url, dots) {
-  shQuote(sprintf(
-    ".libPaths(c(\"%s\",.libPaths()));mirai::dispatcher(\"%s\",url=\"%s\"%s)",
+    ".libPaths(c(\"%s\",.libPaths()));mirai::dispatcher(\"%s\",url=\"%s\",n=%d)",
     libp(),
     urld,
     url,
-    dots
+    n
   ))
 }
 
@@ -649,15 +639,17 @@ query_dispatcher <- function(sock, command, send_mode = 2L, recv_mode = 5L, bloc
   recv(sock, mode = recv_mode, block = block)
 }
 
-launch_dispatcher <- function(arg, dots, envir, serial, tls = NULL, pass = NULL) {
+launch_dispatcher <- function(url, dots, envir, serial, tls = NULL, pass = NULL) {
   cv <- cv()
   urld <- local_url()
   sock <- req_socket(urld)
   pipe_notify(sock, cv, add = TRUE)
-  write_args <- if (is.character(arg)) args_dispatcher_remote else args_dispatcher_local
+  local <- is.numeric(url)
+  n <- if (local) url else 0L
+  if (local) url <- local_url()
   system2(
     .command,
-    args = c("--default-packages=NULL", "--vanilla", "-e", write_args(urld, arg, dots)),
+    args = c("--default-packages=NULL", "--vanilla", "-e", args_dispatcher(urld, url, n)),
     wait = FALSE
   )
   if (is.null(serial)) serial <- .[["serial"]]
@@ -672,8 +664,14 @@ launch_dispatcher <- function(arg, dots, envir, serial, tls = NULL, pass = NULL)
     message(sprintf(._[["sync_dispatcher"]], sync <- sync + .limit_long_secs))
 
   pipe_notify(sock, NULL, add = TRUE)
-  req <- request(.context(sock), data, send_mode = 1L, recv_mode = 2L, cv = cv)
-
+  send(sock, data, mode = 1L, block = TRUE)
+  if (local) {
+    launch_args <- args_daemon_disp(url, dots)
+    for (i in seq_len(n)) {
+      launch_daemon(launch_args)
+    }
+  }
+  req <- recv_aio(sock, mode = 2L, cv = cv)
   while(!until(cv, .limit_long))
     message(sprintf(._[["sync_dispatcher"]], sync <- sync + .limit_long_secs))
 

R/dispatcher.R

@@ -16,25 +16,28 @@
 #' @inheritParams daemons
 #' @param host the character URL dispatcher should dial in to, typically an IPC
 #'   address.
-#' @param url (optional) the character URL dispatcher should listen at (and
-#'   daemons should dial in to), including the port to connect to e.g.
-#'   'tcp://hostname:5555' or 'tcp://10.75.32.70:5555'. Specify 'tls+tcp://' to
-#'   use secure TLS connections.
-#' @param n (optional) if specified, the integer number of daemons to launch. In
-#'   this case, a local url is automatically generated.
+#' @param url the character URL dispatcher should listen at (and daemons should
+#'   dial in to), including the port to connect to e.g. tcp://hostname:5555' or
+#'   'tcp://10.75.32.70:5555'. Specify 'tls+tcp://' to use secure TLS
+#'   connections.
+#' @param n \[default 0L\] if specified, the integer number of daemons to be
+#'   launched locally by the host process.
 #'
 #' @return Invisible NULL.
 #'
 #' @export
 #'
-dispatcher <- function(host, url = NULL, n = NULL, ...) {
-  n <- if (is.numeric(n)) as.integer(n) else length(url)
-  n > 0L || stop(._[["missing_url"]])
-
+dispatcher <- function(host, url = NULL, n = 0L, ...) {
   cv <- cv()
   sock <- socket("rep")
   on.exit(reap(sock))
   pipe_notify(sock, cv, remove = TRUE, flag = flag_value())
+
+  psock <- socket("poly")
+  on.exit(reap(psock), add = TRUE, after = TRUE)
+  m <- monitor(psock, cv)
+  n && listen(psock, url = url, fail = 2L)
+
   dial_sync_socket(sock, host)
 
   raio <- recv_aio(sock, mode = 1L, cv = cv)
@@ -47,10 +50,7 @@ dispatcher <- function(host, url = NULL, n = NULL, ...) {
   }
 
   tls <- NULL
-  auto <- is.null(url)
-  if (auto) {
-    url <- local_url()
-  } else {
+  if (!n) {
     if (is.character(res[[2L]])) {
       tls <- res[[2L]]
       pass <- res[[3L]]
@@ -61,21 +61,13 @@ dispatcher <- function(host, url = NULL, n = NULL, ...) {
   serial <- res[[4L]]
   res <- res[[5L]]
 
-  psock <- socket("poly")
-  on.exit(reap(psock), add = TRUE, after = TRUE)
-  m <- monitor(psock, cv)
-  listen(psock, url = url, tls = tls, fail = 2L)
-
   inq <- outq <- list()
   events <- integer()
   count <- 0L
   envir <- new.env(hash = FALSE, parent = emptyenv())
   `[[<-`(envir, "stream", res)
-  if (auto) {
-    dots <- parse_dots(...)
-    for (i in seq_len(n)) {
-      launch_daemon(args_daemon_disp(url, dots))
-    }
+
+  if (n) {
     for (i in seq_len(n))
       while(!until(cv, .limit_long))
         cv_signal(cv) || wait(cv) || return()
@@ -88,13 +80,13 @@ dispatcher <- function(host, url = NULL, n = NULL, ...) {
       }
     }
   } else {
+    listen(psock, url = url, tls = tls, fail = 2L)
     listener <- attr(psock, "listener")[[1L]]
     url <- opt(listener, "url")
     if (parse_url(url)[["port"]] == "0") {
       url <- sub_real_port(opt(listener, "tcp-bound-port"), url)
     }
   }
-
   send(sock, c(Sys.getpid(), url), mode = 2L, block = TRUE)
 
   ctx <- .context(sock)

R/mirai-package.R

@@ -74,7 +74,6 @@
     function_required = "`.f` must be of type function, not %s",
     localhost = "SSH tunnelling requires daemons `url` hostname to be `127.0.0.1`",
     missing_expression = "missing expression, perhaps wrap in {}?",
-    missing_url = "`n` must be 1 or greater, or else `url` must be supplied",
     named_args = "all items in `.args` must be named, unless supplying an environment",
     named_dots = "all `...` arguments must be named, unless supplying an environment",
     n_one = "`n` must be 1 or greater",

man/dispatcher.Rd

@@ -28,7 +28,6 @@ test_error(daemons(-1), "zero or greater")
 test_error(daemons(raw(0L)), "must be numeric")
 test_error(daemons(1, dispatcher = ""))
 test_error(daemons(url = local_url(), dispatcher = NA))
-test_error(dispatcher(client = "URL"), "must be 1 or greater")
 test_error(daemon("URL"), "Invalid argument")
 test_error(launch_local(1L), "daemons must be set")
 test_true(!daemons_set())