Merge branch 'release/1.1.0'

Author: HenrikBengtsson

.github/workflows/R-CMD-check.yaml

@@ -27,10 +27,13 @@ jobs:
           - {os: ubuntu-latest,  r: 'oldrel'   }
           - {os: ubuntu-latest,  r: 'oldrel-1' }
           - {os: ubuntu-latest,  r: 'oldrel-2' }
-          - {os: ubuntu-latest,  r: '3.6'      }
+#          - {os: ubuntu-latest,  r: '3.6'      }
           - {os: ubuntu-latest,  r: 'release'  , plan: multicore,    label: 'w/ multicore'    }
           - {os: ubuntu-latest,  r: 'release'  , plan: multisession, label: 'w/ multisession' }
+          - {os: windows-latest,  r: 'release', future_version: develop, label: 'w/ future-develop' }
+          - {os: windows-latest,  r: 'release', globals_version: develop, future_version: develop, label: 'w/ future-develop' }
           - {os: ubuntu-latest,  r: 'release', future_version: develop, label: 'w/ future-develop' }
+          - {os: ubuntu-latest,  r: 'release', globals_version: develop, future_version: develop, label: 'w/ future-develop' }
 
     env:
       GITHUB_PAT: ${{ secrets.GITHUB_TOKEN }}
@@ -48,6 +51,8 @@ jobs:
       R_FUTURE_PLAN: ${{ matrix.config.plan }}
       ## FIXME: Eventually update to 'R_FUTURE_GLOBALS_ONREFERENCE=error'
       R_FUTURE_GLOBALS_ONREFERENCE: "warning"
+      R_PARALLELLY_VERSION: ${{ matrix.config.parallelly_version }}
+      R_GLOBALS_VERSION: ${{ matrix.config.globals_version }}
       R_FUTURE_VERSION: ${{ matrix.config.future_version }}
 
     steps:
@@ -63,20 +68,24 @@ jobs:
 
       - uses: r-lib/actions/setup-r-dependencies@v2
         with:
-          extra-packages: any::rcmdcheck
+          extra-packages: any::rcmdcheck any::remotes
           needs: check
 
       - name: Install package itself (special case)
         run: |
           install.packages(".", repos = NULL, type = "source")  ## needed by parallel workers
           lapply(c("future.batchtools", "future.callr"), FUN = function(pkg) if (grepl(pkg, "${{ matrix.config.future_plan }}")) install.packages(pkg))
+          remotes::install_github("futureverse/parallelly", ref="develop")
         shell: Rscript {0}
 
       - name: Test with specific future version?
         run: |
+          globals_version <- Sys.getenv("R_GLOBALS_VERSION")
+          if (nzchar(globals_version)) {
+            remotes::install_github("futureverse/globals", ref=globals_version)
+          }
           future_version <- Sys.getenv("R_FUTURE_VERSION")
           if (nzchar(future_version)) {
-            install.packages("remotes")
             remotes::install_github("futureverse/future", ref=future_version)
           }
         shell: Rscript {0}

.github/workflows/doFuture.tests.extra.yaml

@@ -20,9 +20,9 @@ jobs:
           - { r: "release", strategies: multicore                             }
           - { r: "release", strategies: multisession                          }
           - { r: "release", strategies: "future.callr::callr"                 }
+          - { r: "release", strategies: "future.mirai::mirai_multisession"    }
           - { r: "release", strategies: "future.batchtools::batchtools_local" }
           - { r: "release", strategies: sequential, future_version: develop, label: 'w/ future-develop' }
-          - { r: "release", strategies: sequential, future_version: feature/evalFuture-3, label: 'w/ future-feature/evalFuture-3' }
 
     env:
       GITHUB_PAT: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/test-coverage.yaml

@@ -1,71 +1,62 @@
-on: [push]
+on:
+  workflow_dispatch:  # Enables manual triggering
 
-name: covr
+name: test-coverage.yaml
 
-jobs:
-  covr:
-    if: "! contains(github.event.head_commit.message, '[ci skip]')"    
-
-    timeout-minutes: 45
+permissions: read-all
 
+jobs:
+  test-coverage:
     runs-on: ubuntu-latest
-
-    name: covr
-
-    strategy:
-      fail-fast: false
-
     env:
       GITHUB_PAT: ${{ secrets.GITHUB_TOKEN }}
-      R_KEEP_PKG_SOURCE: yes
-      R_REMOTES_NO_ERRORS_FROM_WARNINGS: true
-      ## Test in other locale (optional)
-      LANGUAGE: ${{ matrix.config.language }}
-      ## R CMD check
-      _R_CHECK_CRAN_INCOMING_: false
-      _R_CHECK_MATRIX_DATA_: true
-      _R_CHECK_SUGGESTS_ONLY_: true
-      _R_CHECK_THINGS_IN_TEMP_DIR_: true
-      RCMDCHECK_ERROR_ON: note
-      ## Specific to futures
-      R_FUTURE_RNG_ONMISUSE: error
 
     steps:
       - uses: actions/checkout@v4
 
+      - name: Assert CODECOV_TOKEN is set
+        run: |
+          if [[ -z "${{secrets.CODECOV_TOKEN}}" ]]; then
+            >&2 echo "::error::ERROR: 'secrets.CODECOV_TOKEN' not set"
+            exit 1
+          fi
+
       - uses: r-lib/actions/setup-r@v2
         with:
           use-public-rspm: true
 
       - uses: r-lib/actions/setup-r-dependencies@v2
         with:
-          extra-packages: |
-            any::rcmdcheck
-            any::remotes
-            any::sessioninfo
-            any::covr
-          needs: check
+          extra-packages: any::covr, any::xml2
+          needs: coverage
 
-      - name: Install dependencies
+      - name: Install itself
         run: |
-          remotes::install_deps(dependencies = TRUE)
-          install.packages(".", repos=NULL, type="source")
+          install.packages(".", repos = NULL, type = "source")
         shell: Rscript {0}
 
-      - name: Session info
-        run: |
-          options(width = 100)
-          pkgs <- installed.packages()[, "Package"]
-          sessioninfo::session_info(pkgs, include_base = TRUE)
-        shell: Rscript {0}
-    
       - name: Test coverage
         run: |
-          ## 1. Get 'Repository Upload Token' from Codecov:
-          ##    https://app.codecov.io/gh/<org>/<repo>/settings
-          ## 2. Set 'CODECOV_TOKEN' in GitHub Secrets:
-          ##    https://github.com/<org>/<repo>/settings/environments/
-          coverage <- covr::package_coverage(quiet = FALSE)
-          print(coverage)
-          covr::codecov(coverage = coverage, token="${{secrets.CODECOV_TOKEN}}")
+          cov <- covr::package_coverage(
+            quiet = FALSE,
+            clean = FALSE,
+            install_path = file.path(normalizePath(Sys.getenv("RUNNER_TEMP"), winslash = "/"), "package")
+          )
+          print(cov)
+          covr::to_cobertura(cov)
         shell: Rscript {0}
+
+      - uses: codecov/codecov-action@v4
+        with:
+          fail_ci_if_error: ${{ github.event_name != 'pull_request' && true || false }}
+          file: ./cobertura.xml
+          plugin: noop
+          disable_search: true
+          token: ${{ secrets.CODECOV_TOKEN }}
+
+      - name: Upload test results
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-test-failures
+          path: ${{ runner.temp }}/package

DESCRIPTION

@@ -1,9 +1,9 @@
 Package: doFuture
-Version: 1.0.2
+Version: 1.1.0
 Title: Use Foreach to Parallelize via the Future Framework
 Depends:
     foreach (>= 1.5.0),
-    future (>= 1.32.0)
+    future (>= 1.49.0)
 Imports:
     future.apply,
     globals,
@@ -25,6 +25,7 @@ License: LGPL (>= 2.1)
 LazyLoad: TRUE
 URL: https://doFuture.futureverse.org, https://github.com/futureverse/doFuture
 BugReports: https://github.com/futureverse/doFuture/issues
+Language: en-US
 Encoding: UTF-8
 RoxygenNote: 7.3.2
 Roxygen: list(markdown = TRUE)

NAMESPACE

@@ -1,5 +1,6 @@
 # Generated by roxygen2: do not edit by hand
 
+S3method(with,DoPar)
 export("%dofuture%")
 export(registerDoFuture)
 export(withDoRNG)
@@ -8,9 +9,12 @@ importFrom(foreach,getErrorIndex)
 importFrom(foreach,getErrorValue)
 importFrom(foreach,getResult)
 importFrom(foreach,makeAccum)
+importFrom(foreach,registerDoSEQ)
 importFrom(foreach,setDoPar)
 importFrom(future,Future)
 importFrom(future,FutureError)
+importFrom(future,FutureInterruptError)
+importFrom(future,cancel)
 importFrom(future,future)
 importFrom(future,getGlobalsAndPackages)
 importFrom(future,nbrOfWorkers)

NEWS.md

@@ -1,3 +1,38 @@
+# Version 1.1.0 (2025-05-19)
+
+## New Features
+
+ * `foreach()` with `%dofuture%` or `%dopar%` will now exit early as
+   soon as it detects an error in one of the iterations. It will also
+   exit early if it detects a user interrupt (e.g. Ctrl-C). Remaining
+   iterations are canceled and interrupted if the future backend
+   supports it, releasing compute resources sooner and avoiding having
+   to wait for remaining futures to be resolved.
+
+ * Add support for `foreach(..., .verbose = TRUE)`, which is part of
+   the official `foreach()` API. When set, detailed debugging
+   information useful for troubleshooting is outputted to standard
+   error through R's message condition mechanism.
+
+ * Add support for `with(registerDoFuture(), { ... })` to temporarily
+   use the doFuture adapter.
+   
+ * Add `registerDoFuture(flavor = "%dofuture%")`, which makes the
+   `%dopar%` infix operator behave as if `%dofuture%` would have been
+   used. This makes it possible for you to use `%dofuture%`, even if
+   you do not have the option to update the code that uses
+   `%dopar%`. For instance, if you use one of the many packages that
+   uses `foreach(...) %dopar% { ... }` internally, this flavor allows
+   you to effectively make that the same as `foreach(...)  %dofuture%
+   { ... }`.
+
+## Bug Fixes
+
+ * `registerDoFuture()` would return an invalid `DoPar` object if
+   there was no `%dopar%` registered. Now it returns an `DoPar` object
+   as if `foreach::registerDoSEQ()` had been called before.
+
+
 # Version 1.0.2 (2025-03-15)
 
 ## Miscellaneous

R/000.import.R

@@ -1,18 +1,14 @@
-import_from <- function(name, default = NULL, package) {
+import_from <- function(name, mode = "function", default = NULL, package) {
   ns <- getNamespace(package)
-  if (exists(name, mode = "function", envir = ns, inherits = FALSE)) {
-    get(name, mode = "function", envir = ns, inherits = FALSE)
+  if (exists(name, mode = mode, envir = ns, inherits = FALSE)) {
+    get(name, mode = mode, envir = ns, inherits = FALSE)
   } else if (!is.null(default)) {
     default
   } else {
-    stop(sprintf("No such '%s' function: %s()", package, name))
+    stop(sprintf("No such '%s' %s: %s()", package, mode, name))
   }
 }
 
-import_future <- function(name, default = NULL) {
-  import_from(name, default = default, package = "future")
-}
-
-import_future.apply <- function(name, default = NULL) {
-  import_from(name, default = default, package = "future.apply")
+import_future <- function(name, mode = "function", default = NULL) {
+  import_from(name, mode = mode, default = default, package = "future")
 }

R/001.import_future_functions.R

@@ -0,0 +1,7 @@
+## To be imported from 'future', if available
+.debug <- NULL
+
+## Import private functions from 'future'
+import_future_functions <- function() {
+  .debug <<- import_future(".debug", mode = "environment", default = new.env(parent = emptyenv()))
+}

R/condition-handlers.R

@@ -0,0 +1,43 @@
+#' @importFrom future FutureInterruptError
+onDoFutureInterrupt <- function(int, op_name = "%dofuture%", debug = FALSE) {
+  if (debug) {
+    mdebug_push("onDoFutureInterrupt() ...")
+    mdebug(sprintf("Received <%s>", class(int)[1]))
+    on.exit(mdebug_pop())
+  }
+  
+  when <- Sys.time()
+  host <- Sys.info()[["nodename"]]
+  pid <- Sys.getpid()
+  msg <- sprintf("'%s' interrupted at %s, while running on %s (pid %s)", op_name, format(when, format = "%FT%T"), sQuote(host), pid)
+
+  ## By signaling the interrupt as an error, the next handler, which should
+  ## be onDoFutureError(), will take care of canceling outstanding futures
+  stop(FutureInterruptError(msg))
+} ## onDoFutureInterrupt()
+
+
+
+#' @importFrom future cancel resolve value
+onDoFutureError <- function(ex, futures, debug = FALSE) {
+  if (debug) {
+    mdebug_push("onDoFutureError() ...")
+    mdebug(sprintf("Received <%s>", class(ex)[1]))
+    on.exit(mdebug_pop())
+  }
+  
+  ## Canceling all futures
+  msg <- sprintf("Caught %s. Canceling all iterations ...", class(ex)[1])
+  warning(msg, immediate. = TRUE, call. = FALSE)
+  futures <- cancel(futures)
+
+  ## Make sure all workers finish before continuing
+  futures <- resolve(futures)
+
+  ## Collect all results
+  for (f in futures) tryCatch(value(f), error = identity)
+
+  if (debug) mdebug(sprintf("Signaling: <%s>", class(ex)[1]))
+
+  stop(ex)
+} ## onDoFutureError()