Merge branch 'develop' of github.com:HenrikBengtsson/future into develop

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future
-Version: 1.70.0-9018
+Version: 1.70.0-9024
 Title: Unified Parallel and Distributed Processing in R for Everyone
 Depends:
     R (>= 3.2.0)

NEWS.md

@@ -11,8 +11,24 @@
    `FutureJournalList` list of them now reports on parallelization
    metrics, including total speedup and total efficiency.
 
+ * `all.equal()` for `future` now returns `"Not both functions"` when
+   comparing against a non-function. Similarly, `"Not both lists"` is
+   returned when comparing a `FutureStrategyList` object against a not
+   list object.
+   
 ## Bug Fixes
 
+ * On MS Windows, a future startup script specified by environment
+   variable `R_FUTURE_STARTUP_SCRIPT` was silently ignored due to a
+   bug in parsing the value.
+
+ * Conditions relayed via the file system could produce an error when
+   relayed if a condition file was corrupt. Now such conditions are
+   silently ignored as intended.
+ 
+ * `all.equal()` for `FutureStrategyList` would throw an error if the
+   object checked against was not a list.
+
  * The default for R option `future.globals.maxSize` is now `+Inf`
    everywhere. It was still 500 MiB is some cases, e.g. `futureCall()`
    had a limit although `future()` did not.

R/backend_api-03.MultiprocessFutureBackend-class.R

@@ -99,7 +99,7 @@ nbrOfFreeWorkers.MultiprocessFutureBackend <- function(evaluator, background = F
 
 
 
-D#' A multiprocess future is a future whose value will be resolved asynchronously in a parallel process
+#' A multiprocess future is a future whose value will be resolved asynchronously in a parallel process
 #'
 #' @inheritParams Future-class
 #' 

R/utils-immediateCondition.R

@@ -64,7 +64,9 @@ readImmediateConditions <- function(path = immediateConditionsPath(rootPath = ro
   if (length(files) == 0L) return(list())
 
   ## Read objects from file
-  objs <- lapply(files, FUN = tryCatch(readRDS, error = identity))
+  objs <- lapply(files, FUN = function(file) {
+    tryCatch(readRDS(file), error = identity)
+  })
 
   ## Drop the ones that failed to be read
   keep <- !vapply(objs, FUN = inherits, "error", FUN.VALUE = FALSE)

R/utils-marshalling.R

@@ -18,7 +18,7 @@ reference_filters <- local({
         FALSE  ## don't drop reference
       }
     } else if (action == "set") {
-      filters <- list(...)
+      filters <<- list(...)
     } else if (action == "reset") {
       filters <<- default
     } else if (action == "append") {

R/utils_api-plan.R

@@ -6,6 +6,13 @@ all.equal.future <- function(target, current, ..., debug = FALSE) {
     mstr(list(target = target, current = current))
   }
 
+  ## Comparing a future strategy against a non-function, e.g. when 'all.equal()'
+  ## recurses into unrelated objects of two different 'future' namespaces
+  if (!is.function(target) || !is.function(current)) {
+    if (debug) mdebug("Not both functions")
+    return("Not both functions")
+  }
+
   ## Compare formals
   if (!isTRUE(all.equal(formals(target), formals(current)))) {
     if (debug) mdebug("Formals differ")
@@ -51,7 +58,12 @@ all.equal.FutureStrategyList <- function(target, current, ..., debug = FALSE) {
     on.exit(mdebug_pop())
   }
 
-  stop_if_not(is.list(target), is.list(current))
+  ## Comparing a future strategy against a non-list, e.g. when 'all.equal()'
+  ## recurses into unrelated objects of two different 'future' namespaces
+  if (!is.list(target) || !is.list(current)) {
+    if (debug) mdebug("Not both lists")
+    return("Not both lists")
+  }
 
   if (length(target) != length(current)) {
     if (debug) mdebug("Different lengths")

R/zzz.R

@@ -109,7 +109,10 @@ sourceFutureStartupScript <- function(default = c(".future.R", "~/.future.R"), d
     pathnames <- TRUE
   } else {
     if (debug) mdebug("R_FUTURE_STARTUP_SCRIPT: ", sQuote(pathnames))
-    pathnames <- strsplit(pathnames, split = "[:;]", fixed = FALSE)[[1]]
+    ## On MS Windows, ':' must not be used as a separator, because it is
+    ## part of the drive letter of an absolute pathname, e.g. 'C:/x/.future.R'
+    split <- if (.Platform$OS.type == "windows") ";" else "[:;]"
+    pathnames <- strsplit(pathnames, split = split, fixed = FALSE)[[1]]
     if (identical(toupper(pathnames), "TRUE")) {
       pathnames <- TRUE
     } else if (identical(toupper(pathnames), "FALSE")) {

inst/testme/run.R

@@ -118,6 +118,11 @@ main <- function() {
     if (!utils::file_test("-f", "DESCRIPTION")) {
       stop("Current folder does not look like a package folder")
     }
+    ## IMPORTANT: Load 'covr' already here, i.e. before the prologue scripts
+    ## are sourced, so that they and the test scripts can detect coverage
+    ## testing via ("covr" %in% loadedNamespaces()), which is how it is
+    ## detected when testing via covr::package_coverage()
+    loadNamespace("covr")
   }
 
   ## Fallback for 'testme_name'?
@@ -296,7 +301,6 @@ testme_run_test <- function(testme) {
   if (testme[["status"]] != "skipped") {
     if (testme[["debug"]]) message("Running test script: ", sQuote(testme[["script"]]))
     testme[["status"]] <- "failed"
-    str(testme[["coverage"]])
     if (testme[["coverage"]] != "none") {
       pkg_env <- pkgload::load_all()
       cov <- covr::environment_coverage(pkg_env[["env"]], test_files = testme[["script"]])

inst/testme/test-FutureBackend-factories.R

@@ -0,0 +1,161 @@
+#' @tags backend sequential multisession cluster
+#' @tags detritus-connections
+
+library(future)
+
+SequentialFutureBackend <- future:::SequentialFutureBackend
+MultisessionFutureBackend <- future:::MultisessionFutureBackend
+ClusterFutureBackend <- future:::ClusterFutureBackend
+MulticoreFutureBackend <- future:::MulticoreFutureBackend
+stopWorkers <- future:::stopWorkers
+
+## NOTE: A backend factory is reached via attr(<strategy>, "factory"), which
+## holds the function object as it was when the package namespace was built.
+## Calls made through that copy are not attributed to these functions, which
+## is why the factories are called directly below.
+
+message("*** SequentialFutureBackend() ...")
+
+backend <- SequentialFutureBackend()
+print(class(backend))
+stopifnot(inherits(backend, "SequentialFutureBackend"),
+          inherits(backend, "FutureBackend"))
+
+## A sequential backend has a single worker, which is always free
+stopifnot(nbrOfWorkers(backend) == 1L)
+stopifnot(nbrOfFreeWorkers(backend) == 1L)
+
+print(backend)
+
+message("*** SequentialFutureBackend() ... DONE")
+
+
+message("*** MultisessionFutureBackend() ...")
+
+## 'workers' must be numeric
+res <- tryCatch(MultisessionFutureBackend(workers = "two"), error = identity)
+print(res)
+stopifnot(inherits(res, "error"))
+
+## ... and a single, finite value >= 1
+res <- tryCatch(MultisessionFutureBackend(workers = c(1L, 2L)), error = identity)
+print(res)
+stopifnot(inherits(res, "error"))
+
+res <- tryCatch(MultisessionFutureBackend(workers = 0L), error = identity)
+print(res)
+stopifnot(inherits(res, "error"))
+
+## 'workers = 1' falls back to a sequential backend
+backend <- MultisessionFutureBackend(workers = 1L)
+print(class(backend))
+stopifnot(inherits(backend, "SequentialFutureBackend"))
+
+## ... which can be overridden with I(1)
+backend <- MultisessionFutureBackend(workers = I(1L))
+print(class(backend))
+stopifnot(inherits(backend, "MultisessionFutureBackend"))
+stopifnot(nbrOfWorkers(backend) == 1L)
+stopWorkers(backend)
+
+## 'workers' may be given as a function
+backend <- MultisessionFutureBackend(workers = function() 2L)
+print(class(backend))
+stopifnot(inherits(backend, "MultisessionFutureBackend"),
+          inherits(backend, "ClusterFutureBackend"))
+stopifnot(nbrOfWorkers(backend) == 2L)
+stopifnot(nbrOfFreeWorkers(backend) == 2L)
+
+## A MultisessionFuture is what this backend produces
+stopifnot("MultisessionFuture" %in% backend[["futureClasses"]])
+
+print(backend)
+stopWorkers(backend)
+
+message("*** MultisessionFutureBackend() ... DONE")
+
+
+message("*** ClusterFutureBackend() ...")
+
+backend <- ClusterFutureBackend(workers = 2L)
+print(class(backend))
+stopifnot(inherits(backend, "ClusterFutureBackend"),
+          inherits(backend, "FutureBackend"))
+stopifnot(nbrOfWorkers(backend) == 2L)
+stopifnot(nbrOfFreeWorkers(backend) == 2L)
+
+print(backend)
+
+## The backend can be shut down, and doing so is idempotent
+stopifnot(isTRUE(stopWorkers(backend)))
+
+message("*** ClusterFutureBackend() ... DONE")
+
+
+message("*** listFutures() ...")
+
+listFutures <- future:::listFutures
+
+## A backend without a 'reg' element cannot list its futures
+fake <- structure(list(), class = c("MultiprocessFutureBackend",
+                                    "FutureBackend"))
+res <- tryCatch(listFutures(fake), error = identity)
+print(res)
+stopifnot(inherits(res, "FutureError"),
+          grepl("does not implement listFutures", conditionMessage(res)))
+
+## nbrOfWorkers()/nbrOfFreeWorkers() are not implemented at this level either
+res <- tryCatch(nbrOfWorkers(fake), error = identity)
+print(res)
+stopifnot(inherits(res, "error"), grepl("not implemented", conditionMessage(res)))
+
+res <- tryCatch(nbrOfFreeWorkers(fake), error = identity)
+print(res)
+stopifnot(inherits(res, "error"), grepl("not implemented", conditionMessage(res)))
+
+## A backend without any futures lists none
+backend <- MultisessionFutureBackend(workers = 2L)
+data <- listFutures(backend)
+print(data)
+stopifnot(is.data.frame(data), nrow(data) == 0L)
+stopifnot(all(c("counter", "start", "label", "resolved") %in% names(data)))
+stopWorkers(backend)
+
+## With futures running, they are listed
+plan(multisession, workers = 2L)
+fs <- list(future({ Sys.sleep(0.5); 1L }, label = "one"),
+           future({ Sys.sleep(0.5); 2L }, label = "two"))
+data <- listFutures(plan("backend"), debug = TRUE)
+print(data)
+stopifnot(is.data.frame(data), nrow(data) == 2L)
+stopifnot(all(c("one", "two") %in% data[["label"]]))
+stopifnot(is.logical(data[["resolved"]]))
+
+## Collect the values, so that no futures are left behind
+stopifnot(identical(unlist(value(fs)), c(1L, 2L)))
+plan(sequential)
+
+message("*** listFutures() ... DONE")
+
+
+message("*** MulticoreFutureBackend() ...")
+
+## IMPORTANT: Skip when testing with 'covr', because forked workers do not
+## write back their 'covr' traces, and may truncate them
+if (!covr_testing && parallelly::supportsMulticore() &&
+    availableCores("multicore") >= 2L) {
+  backend <- MulticoreFutureBackend(workers = 2L)
+  print(class(backend))
+  stopifnot(inherits(backend, "MulticoreFutureBackend"))
+  stopifnot(nbrOfWorkers(backend) == 2L)
+  stopWorkers(backend)
+
+  ## 'workers = 1' falls back to a sequential backend
+  backend <- MulticoreFutureBackend(workers = 1L)
+  print(class(backend))
+  stopifnot(inherits(backend, "SequentialFutureBackend"))
+} else {
+  message("Skipping MulticoreFutureBackend() tests")
+}
+
+message("*** MulticoreFutureBackend() ... DONE")