align APIs

shikokuchuo · Sep 6, 2024 · 2fe3c8a · 2fe3c8a
1 parent 7d409b5
commit 2fe3c8a
Show file tree

Hide file tree

Showing 5 changed files with 8 additions and 7 deletions.
diff --git a/DESCRIPTION b/DESCRIPTION
@@ -35,4 +35,4 @@ Suggests:
     litedown
 VignetteBuilder: litedown
 RoxygenNote: 7.3.2
-Remotes: shikokuchuo/nanonext
+Remotes: shikokuchuo/nanonext@dispatcher
diff --git a/NAMESPACE b/NAMESPACE
@@ -59,6 +59,8 @@ export(unresolved)
 importFrom(nanonext,"opt<-")
 importFrom(nanonext,.advance)
 importFrom(nanonext,.context)
+importFrom(nanonext,.cv_flag)
+importFrom(nanonext,.dispatcher)
 importFrom(nanonext,.keep)
 importFrom(nanonext,.mark)
 importFrom(nanonext,.unresolved)
@@ -67,10 +69,8 @@ importFrom(nanonext,call_aio_)
 importFrom(nanonext,collect_aio)
 importFrom(nanonext,collect_aio_)
 importFrom(nanonext,cv)
-importFrom(nanonext,cv_flag)
 importFrom(nanonext,cv_value)
 importFrom(nanonext,dial)
-importFrom(nanonext,dispatcher_socket)
 importFrom(nanonext,is_error_value)
 importFrom(nanonext,listen)
 importFrom(nanonext,lock)

diff --git a/NEWS.md b/NEWS.md
@@ -1,5 +1,6 @@
 # mirai 1.2.0.9009 (development)
 
+* `daemons(dispatcher = NA)` now provides access to threaded dispatcher (experimental). This implements dispatcher using a thread rather than an external process and is faster and more efficient.
 * `everywhere()` now errors if the specified compute profile is not yet set up, rather than fail silently.
 * Fixes flatmap with `mirai_map()[.flat]` assigning a variable 'typ' to the calling environment.
 * Adds `mirai_map()[.progress_cli]` as an alternative progress indicator, using the 'cli' package to show % complete and ETA.

diff --git a/R/daemons.R b/R/daemons.R
@@ -296,7 +296,7 @@ daemons <- function(n, url = NULL, remote = NULL, dispatcher = TRUE, ...,
       if (is.na(dispatcher)) {
         n <- if (missing(n)) length(url) else if (is.numeric(n) && n >= 1L) as.integer(n) else stop(._[["n_one"]])
         cv <- cv()
-        sock <- dispatcher_socket(cv = cv, n = n, host = inproc_url(), url = url, tls = if (length(tls)) tls_config(server = tls, pass = pass))
+        sock <- .dispatcher(cv = cv, n = n, host = inproc_url(), url = url, tls = if (length(tls)) tls_config(server = tls, pass = pass))
         urls <- as.character(lapply(seq_len(n), function(x) sprintf("%s/%d", url, x)))
         `[[<-`(`[[<-`(`[[<-`(envir, "cv", cv), "urls", urls), "dispatcher", url)
       } else if (dispatcher) {
@@ -351,7 +351,7 @@ daemons <- function(n, url = NULL, remote = NULL, dispatcher = TRUE, ...,
       output <- attr(dots, "output")
       if (is.na(dispatcher)) {
         cv <- cv()
-        sock <- dispatcher_socket(cv = cv, n = n, host = inproc_url(), url = urld)
+        sock <- .dispatcher(cv = cv, n = n, host = inproc_url(), url = urld)
         urls <- as.character(lapply(seq_len(n), function(x) sprintf("%s/%d", urld, x)))
         for (i in seq_len(n))
           launch_daemon(wa3(urls[i], dots, next_stream(envir)), output)
@@ -492,7 +492,7 @@ status <- function(.compute = "default") {
   if (is.null(envir[["dispatcher"]]))
     list(connections = as.integer(stat(envir[["sock"]], "pipes")),
          daemons = if (is.null(envir[["sockc"]])) envir[["urls"]] else query_status(envir)) else
-           list(connections = cv_flag(envir[["cv"]]), daemons = envir[["dispatcher"]])
+           list(connections = .cv_flag(envir[["cv"]]), daemons = envir[["dispatcher"]])
 
 }
 

diff --git a/R/mirai-package.R b/R/mirai-package.R
@@ -46,7 +46,7 @@
 #'     (\href{https://orcid.org/0000-0002-0750-061X}{ORCID})
 #'
 #' @importFrom nanonext .advance collect_aio collect_aio_ call_aio call_aio_
-#'     .context cv cv_flag cv_value dial dispatcher_socket is_error_value .keep
+#'     .context cv .cv_flag cv_value dial .dispatcher is_error_value .keep
 #'     listen lock .mark mclock msleep nng_error opt opt<- parse_url pipe_notify
 #'     random reap recv recv_aio request send serial_config socket stat stop_aio
 #'     tls_config unresolved .unresolved until wait write_cert