upgrades collect_mirai()

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: mirai
 Type: Package
 Title: Minimalist Async Evaluation Framework for R
-Version: 1.3.1.9026
+Version: 1.3.1.9027
 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.9026 (development)
+# mirai 1.3.1.9027 (development)
 
 #### New Architecture
 
@@ -15,6 +15,7 @@
 #### Updates
 
 * `status()` using the new dispatcher is updated to provide more concise information.
+* `collect_mirai()` is now interruptible and takes a '...' argument accepting the collection options provided to the 'mirai_map' `[]` method, such as `.flat` etc.
 * `everywhere()` now returns a list of mirai, which may be waited for and inspected (thanks @dgkf  #164).
 * `everywhere()` drops argument '.serial' as serialization configurations are now registered via an argument at `daemons()`.
 * `launch_local()` and `launch_remote()` simplified to take the argument 'n' instead of 'url' for how many daemons to launch.

R/map.R

@@ -52,7 +52,7 @@
 #'
 #' @return A \sQuote{mirai_map} (list of \sQuote{mirai} objects).
 #'
-#' @section Results:
+#' @section Collection Options:
 #'
 #' \code{x[]} collects the results of a \sQuote{mirai_map} \code{x} and returns
 #' a list. This will wait for all asynchronous operations to complete if still
@@ -215,21 +215,7 @@ mirai_map <- function(.x, .f, ..., .args = list(), .promise = NULL, .compute = "
 `[.mirai_map` <- function(x, ...) {
 
   missing(..1) && return(collect_aio_(x))
-
-  dots <- eval(`[[<-`(substitute(alist(...)), 1L, quote(list)), envir = .)
-  expr <- if (length(dots) > 1L) do.call(expression, dots) else dots[[1L]]
-  xlen <- length(x)
-  i <- 0L
-  typ <- xi <- FALSE
-  collect_map <- function(i) {
-    xi <- collect_aio_(x[[i]])
-    eval(expr)
-    xi
-  }
-  eval(expr)
-  out <- `names<-`(lapply(seq_len(xlen), collect_map), names(x))
-  xi && return(unlist(out, recursive = FALSE))
-  out
+  map(x, ...)
 
 }
 
@@ -245,10 +231,10 @@ print.mirai_map <- function(x, ...) {
 
 #' mirai Map Options
 #'
-#' Expressions to insert into the \code{[]} method for \sQuote{mirai_map}
-#' objects.
+#' Expressions to be provided to the \code{[]} method for \sQuote{mirai_map}
+#' objects, or the \code{...} argument of \code{\link{collect_mirai}}.
 #'
-#' @inheritSection mirai_map Results
+#' @inheritSection mirai_map Collection Options
 #'
 #' @keywords internal
 #' @export
@@ -288,3 +274,24 @@ print.mirai_map <- function(x, ...) {
 .stop <- compiler::compile(
   quote(if (is_error_value(xi)) { lapply(x, stop_mirai); stop(xi, call. = FALSE) })
 )
+
+# internals --------------------------------------------------------------------
+
+map <- function(x, ...) {
+
+  dots <- eval(`[[<-`(substitute(alist(...)), 1L, quote(list)), envir = .)
+  expr <- if (length(dots) > 1L) do.call(expression, dots) else dots[[1L]]
+  xlen <- length(x)
+  i <- 0L
+  typ <- xi <- FALSE
+  collect_map <- function(i) {
+    xi <- collect_aio_(x[[i]])
+    eval(expr)
+    xi
+  }
+  eval(expr)
+  out <- `names<-`(lapply(seq_len(xlen), collect_map), names(x))
+  xi && return(unlist(out, recursive = FALSE))
+  out
+
+}

R/mirai.R

@@ -341,17 +341,19 @@ call_mirai_ <- call_aio_
 
 #' mirai (Collect Value)
 #'
-#' \code{collect_mirai} waits for the \sQuote{mirai} to resolve if still in
-#' progress, and returns its value directly. It is a more efifcient version of
-#' and equivalent to \code{call_mirai(x)$data}.
+#' Waits for the \sQuote{mirai} to resolve if still in progress, and returns its
+#' value directly. It is a more efficient version of and equivalent to
+#' \code{call_mirai_(x)$data}.
 #'
 #' This function will wait for the asynchronous operation(s) to complete if
-#' still in progress (blocking), and is not interruptible.
+#' still in progress, blocking but interruptible.
 #'
-#' \code{x[]} may be used to wait for and return the value of a mirai \code{x},
-#' and is the user-interruptible counterpart to \code{collect_mirai(x)}.
+#' \code{x[]} may also be used to wait for and return the value of a mirai
+#' \code{x}, and is equivalent to \code{collect_mirai(x)}.
 #'
 #' @inheritParams call_mirai
+#' @param ... (if \sQuote{x} is a list of mirai) any of the collection options
+#'   for \code{\link{mirai_map}}, such as \code{\link{.flat}}.
 #'
 #' @return An object (the return value of the \sQuote{mirai}), or a list of such
 #'   objects (the same length as \sQuote{x}, preserving names).
@@ -372,11 +374,22 @@ call_mirai_ <- call_aio_
 #' # using x[]
 #' m[]
 #'
+#' # mirai_map with collection options
+#' daemons(1, dispatcher = FALSE)
+#' m <- mirai_map(1:3, rnorm)
+#' collect_mirai(m, .flat, .progress)
+#' daemons(0)
+#'
 #' }
 #'
 #' @export
 #'
-collect_mirai <- collect_aio
+collect_mirai <- function(x, ...) {
+
+  is.list(x) && ...length() && return(map(x, ...))
+  collect_aio_(x)
+
+}
 
 #' mirai (Stop)
 #'

mirai.Rproj

@@ -1,5 +1,5 @@
 Version: 1.0
-ProjectId: 6c04a62e-266d-423f-8aef-506f99a7fe8d
+ProjectId: 516ec7dc-782e-4bc6-a1b3-cdd75f9f0c22
 
 RestoreWorkspace: Default
 SaveWorkspace: Default