Added support for backend maintainers to specify "cleanup" hook funct…

…ions [#719]

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future
-Version: 1.33.2-9001
+Version: 1.33.2-9002
 Title: Unified Parallel and Distributed Processing in R for Everyone
 Imports:
     digest,

NEWS.md

@@ -1,5 +1,12 @@
 # Version (development version)
 
+## New Features
+
+ * Added support for backend maintainers to specify "cleanup" hook
+   functions on future strategies, which are called when switching
+   future plan. These hook functions are specified via the optional
+   `cleanup` attribute, cf. `attr(cluster, "cleanup")`.
+
 ## Bug Fixes
 
  * `resolved()` for `ClusterFuture`:s would produce `Error:

R/cluster.R

@@ -46,4 +46,7 @@ cluster <- function(..., persistent = FALSE, workers = availableWorkers(), envir
 }
 class(cluster) <- c("cluster", "multiprocess", "future", "function")
 attr(cluster, "init") <- TRUE
+attr(cluster, "cleanup") <- function() {
+  ClusterRegistry(action = "stop")
+}
 attr(cluster, "tweakable") <- quote(c(makeClusterPSOCK_args(), "persistent"))

R/multisession.R

@@ -81,4 +81,7 @@ multisession <- function(..., workers = availableCores(), lazy = FALSE, rscript_
 }
 class(multisession) <- c("multisession", "cluster", "multiprocess", "future", "function")
 attr(multisession, "init") <- TRUE
+attr(multisession, "cleanup") <- function() {
+  ClusterRegistry(action = "stop")
+}
 attr(multisession, "untweakable") <- c("persistent")

R/zzz.plan.R

@@ -177,7 +177,16 @@ plan <- local({
   })
 
   plan_cleanup <- function() {
-    ClusterRegistry(action = "stop")
+    evaluator <- stack[[1L]]
+
+    cleanup <- attr(evaluator, "cleanup", exact = TRUE)
+    if (!is.null(cleanup)) {
+      if (is.function(cleanup)) {
+        cleanup()
+      } else {
+        stop(FutureError(sprintf("Unknown type of 'cleanup' attribute on current future strategy: %s", sQuote(paste(class(cleanup), collapse = ", ")))))
+      }
+    }
   }
 
   plan_init <- function() {
@@ -266,11 +275,11 @@ plan <- local({
     ## Warn about 'multicore' on certain systems
     warn_about_multicore(newStack)
 
-    stack <<- newStack
-
-    ## Stop any (implicitly started) clusters?
+    ## Stop/cleanup any previously registered backends?
     if (cleanup) plan_cleanup()
 
+    stack <<- newStack
+
     ## Initiate future workers?
     if (init) plan_init()
 
@@ -575,4 +584,3 @@ resetWorkers.multicore <- function(x, ...) {
   FutureRegistry(reg, action = "collect-all", earlySignal = FALSE)
   stop_if_not(usedCores() == 0L)
 }
-