Add support for relaying immediateCondition:s in near real-time via the local file system [#29]

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future.callr
-Version: 0.8.2-9206
+Version: 0.8.2-9208
 Depends:
     R (>= 3.4.0),
     future (>= 1.40.0)

NAMESPACE

@@ -1,5 +1,6 @@
 # Generated by roxygen2: do not edit by hand
 
+S3method(getFutureBackendConfigs,CallrFutureBackend)
 S3method(interruptFuture,CallrFutureBackend)
 S3method(launchFuture,CallrFutureBackend)
 S3method(nbrOfFreeWorkers,CallrFutureBackend)

NEWS.md

@@ -9,6 +9,10 @@
    terminated is now considerred interrupted, which for instance
    means that it can be `reset()`.
 
+ * Now 'callr' futures can relay `immediateCondition`:s in near
+   real-time, e.g. `progression` contdions signals by the
+   **progressr** package.
+ 
 
 # Version 0.8.2 [2023-08-08]
 

R/CallrFutureBackend-class.R

@@ -187,6 +187,34 @@ nbrOfFreeWorkers.CallrFutureBackend <- function(evaluator = NULL, background = F
 }
 
 
+
+#' @exportS3Method getFutureBackendConfigs CallrFutureBackend
+getFutureBackendConfigs.CallrFutureBackend <- local({
+  immediateConditionsPath <- import_future("immediateConditionsPath")
+  fileImmediateConditionHandler <- import_future("fileImmediateConditionHandler")
+  
+  function(future, ..., debug = isTRUE(getOption("future.debug"))) {
+    conditionClasses <- future[["conditions"]]
+    if (is.null(conditionClasses)) {
+      capture <- list()
+    } else {
+      path <- immediateConditionsPath(rootPath = tempdir())
+      capture <- list(
+        immediateConditionHandlers = list(
+          immediateCondition = function(cond) {
+            fileImmediateConditionHandler(cond, path = path)
+          }
+        )
+      )
+    }
+  
+    list(
+      capture = capture
+    )
+  }
+})
+
+
 #' Prints a callr future
 #'
 #' @param x An CallrFuture object
@@ -233,6 +261,15 @@ resolved.CallrFuture <- function(x, .signalEarly = TRUE, ...) {
   if (!inherits(process, "r_process")) return(FALSE)
   resolved <- !process$is_alive()
 
+  ## Collect and relay immediateCondition if they exists
+  conditions <- readImmediateConditions(signal = TRUE)
+  ## Record conditions as signaled
+  signaled <- c(x[[".signaledConditions"]], conditions)
+  x[[".signaledConditions"]] <- signaled
+
+  ## Signal conditions early? (happens only iff requested)
+  if (.signalEarly) signalEarly(x, ...)
+
   ## Signal errors early?
   if (.signalEarly && resolved) {
     ## Trigger a FutureError already here, if exit code != 0
@@ -257,6 +294,12 @@ result.CallrFuture <- function(future, ...) {
 
   result <- await(future, cleanup = FALSE)
 
+  ## Collect and relay immediateCondition if they exists
+  conditions <- readImmediateConditions()
+  ## Record conditions as signaled
+  signaled <- c(future[[".signaledConditions"]], conditions)
+  future[[".signaledConditions"]] <- signaled
+
   if (!inherits(result, "FutureResult")) {
     if (inherits(result, "FutureLaunchError")) {
     } else {

R/zzz.R

@@ -1,10 +1,16 @@
 ## To be cached by .onLoad()
 FutureRegistry <- NULL
 assertOwner <- NULL
+readImmediateConditions <- NULL
+signalEarly <- NULL
+getFutureBackendConfigs <- NULL
 
 .onLoad <- function(libname, pkgname) {
   ## Import private functions from 'future'
   FutureRegistry <<- import_future("FutureRegistry")
   assertOwner <<- import_future("assertOwner")
+  readImmediateConditions <<- import_future("readImmediateConditions")
+  signalEarly <<- import_future("signalEarly")
+  getFutureBackendConfigs <<- import_future("getFutureBackendConfigs")
 }