Add rudimentary support for canceling and interrupting future.p2p futures

Author: HenrikBengtsson

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: future.p2p
 Title: A Peer-to-Peer Compute Cluster via Futureverse
-Version: 0.3.0-9004
+Version: 0.3.0-9005
 Description: Implementation of the 'Future' API <doi:10.32614/RJ-2021-048> that resolves futures on a peer-to-peer ('P2P') compute environment. By using this future backend, you and your friends can share your spare compute resources with each other.
 Imports:
     future (>= 1.67.0),

NAMESPACE

@@ -1,5 +1,6 @@
 # Generated by roxygen2: do not edit by hand
 
+S3method(interruptFuture,PicoP2PFutureBackend)
 S3method(launchFuture,PicoP2PFutureBackend)
 S3method(nbrOfFreeWorkers,PicoP2PFutureBackend)
 S3method(nbrOfWorkers,PicoP2PFutureBackend)
@@ -19,8 +20,11 @@ importFrom(callr,r_bg)
 importFrom(future,Future)
 importFrom(future,FutureBackend)
 importFrom(future,FutureError)
+importFrom(future,FutureInterruptError)
+importFrom(future,FutureResult)
 importFrom(future,UnexpectedFutureResultError)
 importFrom(future,future)
+importFrom(future,interruptFuture)
 importFrom(future,launchFuture)
 importFrom(future,nbrOfFreeWorkers)
 importFrom(future,nbrOfWorkers)

R/PicoP2PFuture-class.R

@@ -88,7 +88,7 @@ result.PicoP2PFuture <- function(future, ...) {
   rx$wait()
 
   ## Get the results
-  file <- rx$get_result()
+  response <- rx$get_result()
 
   ## Finalize the 'callr' process, which includes removing any temporary
   ## files that it created
@@ -102,21 +102,42 @@ result.PicoP2PFuture <- function(future, ...) {
   }
 
   future[["rx"]] <- NULL
-  if (debug) mdebugf("FutureResult file: %s [%g bytes]", sQuote(file), file.size(file))
-  if (!file_test("-f", file)) {
-    stop(FutureError(sprintf("FutureResult file not found: ", sQuote(file))), future = future)
-  }
-  
-  result <- local({
-    if (debug) {
-      mdebug_push("Reading FutureResult from file")
-      mdebugf("FutureResult file: %s [%g bytes]", sQuote(file), file.size(file))
-      on.exit(mdebug_pop())
+
+  stop_if_not(is.list(response))
+  type <- response[["type"]]
+  if (type == "event") {
+    event <- response[["value"]]
+    if (event == "interrupted") {
+      result <- FutureResult(
+        conditions = list(list(
+          condition = structure(list(), class = c("interrupt", "condition")),
+          signaled = 0L
+        )),
+        uuid = future[["uuid"]]
+      )
+    } else {
+      stop(FutureError(sprintf("Unknown event from future dispatcher: event = %s", sQuote(event))))
     }
-    result <- readRDS(file)
-    file.remove(file)
-    result
-  })
+  } else if (type == "file") {
+    file <- response[["value"]]
+    if (debug) mdebugf("FutureResult file: %s [%g bytes]", sQuote(file), file.size(file))
+    if (!file_test("-f", file)) {
+      stop(FutureError(sprintf("FutureResult file not found: ", sQuote(file))), future = future)
+    }
+  
+    result <- local({
+      if (debug) {
+        mdebug_push("Reading FutureResult from file")
+        mdebugf("FutureResult file: %s [%g bytes]", sQuote(file), file.size(file))
+        on.exit(mdebug_pop())
+      }
+      result <- readRDS(file)
+      file.remove(file)
+      result
+    })
+  } else {
+    stop(FutureError(sprintf("Unknown response from future dispatcher: type = %s", sQuote(type))))
+  }
 
   future[["result"]] <- result
 

R/PicoP2PFutureBackend-class.R

@@ -144,6 +144,43 @@ launchFuture.PicoP2PFutureBackend <- function(backend, future, ...) {
 } ## launchFuture()
 
 
+#' @importFrom future interruptFuture
+#' @export
+interruptFuture.PicoP2PFutureBackend <- function(backend, future, ...) {
+  debug <- isTRUE(getOption("future.debug"))
+  if (debug) {
+    mdebugf_push("interruptFuture(<%s>, future = <%s>, ...) ...", class(backend)[1], class(future)[1])
+    on.exit(mdebugf_pop())
+  }
+  
+  ## Has interrupts been disabled by user?
+  if (!backend[["interrupts"]]) {
+    if (debug) mdebug("Skipping, because interrupts are disabled for this backend")
+    return(future)
+  }
+  
+  rx <- future[["rx"]]
+  if (is.null(rx)) {
+    if (debug) mdebug("Skipping, because there is no future dispatcher processes")
+    return(future)
+  }
+
+  channels <- attr(rx, "channels", exact = TRUE)
+  tx <- channels[["tx"]]
+  if (is.null(tx)) {
+    if (debug) mdebug("Skipping, because there is no communication channel to future dispatcher processes")
+    return(future)
+  }
+
+  cat("interrupt", file = tx, append = TRUE)
+  if (debug) mdebug("Sent 'interrupt' to future dispatcher processes")
+  
+  future[["state"]] <- "interrupted"
+  
+  future
+}
+
+
 #' @importFrom future nbrOfWorkers
 #' @export
 nbrOfWorkers.PicoP2PFutureBackend <- function(evaluator) {

R/pico_p2p.R

@@ -294,6 +294,7 @@ pico_p2p_hosted_clusters <- function(host = "pipe.pico.sh", ssh_args = NULL, tim
 
 
 
+#' @importFrom future FutureInterruptError FutureResult
 #' @importFrom callr r_bg
 #' @importFrom utils file_test
 pico_p2p_dispatch_future <- function(future) {
@@ -312,21 +313,59 @@ pico_p2p_dispatch_future <- function(future) {
     pico_p2p_receive_result <- import_future.p2p("pico_p2p_receive_result")
 
     update_parent <- function(msg, ...) {
+      rx <- channels[["rx"]]
+      if (is.null(rx)) return()
       msg <- sprintf("%s\n", msg)
-      cat(msg, file = channels[["rx"]], append = TRUE)
+      cat(msg, file = rx, append = TRUE)
     }
 
+    listen_parent <- function(...) {
+      tx <- channels[["tx"]]
+      if (is.null(tx)) return(character(0L))
+      readLines(tx, n = 1e6, warn = FALSE)
+    }
+
+    ## Check for interrupts
+    if ("interrupt" %in% listen_parent()) {
+      return(list(type = "event", value = "interrupted"))
+    }
+    
     update_parent("connecting")
     pico <- pico_pipe(topic, user = name, host = host, ssh_args = ssh_args)
+    
+    ## Check for interrupts
+    if ("interrupt" %in% listen_parent()) {
+      return(list(type = "event", value = "interrupted"))
+    }
+    
     update_parent("announcement")
     m <- pico_p2p_hello(pico, type = "client")
+    ## Check for interrupts
+    if ("interrupt" %in% listen_parent()) {
+      return(list(type = "event", value = "interrupted"))
+    }
+    
     update_parent("connected")
 
     ## 2. Announce future
     repeat {
       update_parent("request")
       m1 <- pico_p2p_have_future(pico, future = file, duration = duration)
+
+      ## Check for interrupts
+      if ("interrupt" %in% listen_parent()) {
+        ## TODO: Withdraw future
+        return(list(type = "event", value = "interrupted"))
+      }
+
       m2 <- pico_p2p_wait_for(pico, type = "offer", futures = m1[["future"]], expires = m1[["expires"]])
+      
+      ## Check for interrupts
+      if ("interrupt" %in% listen_parent()) {
+        ## TODO: Withdraw future
+        return(list(type = "event", value = "interrupted"))
+      }
+
       if (m2[["type"]] != "expired") break
     }
 
@@ -339,18 +378,25 @@ pico_p2p_dispatch_future <- function(future) {
     ## 4. Remove temporary file
     file.remove(file)
 
+    ## Check for interrupts
+    if ("interrupt" %in% listen_parent()) {
+      ## TODO: Withdraw future
+      return(list(type = "event", value = "interrupted"))
+    }
+
     ## 5. Wait for and receive FutureResult file
     update_parent("wait")
     path <- file.path(dirname(dirname(file)), "results")
     tryCatch({
       file <- pico_p2p_receive_result(pico, via = via, path = path)
     }, interrupt = function(int) {
-      cat(file = "foo.log", "interrupted\n")
+      ## TODO: Withdraw future
+      return(list(type = "event", value = "interrupted"))
     })
 
     update_parent("result")
 
-    invisible(file)
+    list(type = "file", value = file)
   } ## send_future()
 
   debug <- isTRUE(getOption("future.p2p.debug"))