Merge branch 'release/0.4.0'

Author: HenrikBengtsson

DESCRIPTION

@@ -1,11 +1,12 @@
 Package: future.p2p
 Title: A Peer-to-Peer Compute Cluster via Futureverse
-Version: 0.3.0
+Version: 0.4.0
 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),
     processx,
-    callr
+    callr,
+    utils
 Suggests:
     commonmark,
     base64enc
@@ -18,4 +19,4 @@ URL: https://future.p2p.futureverse.org/
 BugReports: https://github.com/futureverse/future.p2p/issues
 Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
-RoxygenNote: 7.3.2
+RoxygenNote: 7.3.3

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)
@@ -31,9 +35,11 @@ importFrom(future,resolved)
 importFrom(future,result)
 importFrom(future,run)
 importFrom(future,sequential)
+importFrom(processx,poll)
 importFrom(processx,process)
 importFrom(utils,capture.output)
 importFrom(utils,download.file)
 importFrom(utils,file_test)
+importFrom(utils,head)
 importFrom(utils,packageVersion)
 importFrom(utils,str)

NEWS.md

@@ -1,3 +1,28 @@
+# Version 0.4.0 [2025-11-20]
+
+## New Features
+
+ * Add support for canceling and interrupting 'future.p2p' futures.
+
+## Bug Fixes
+
+ * The `future.p2p` backend was holding on to temporary **callr**
+   files longer than necessary. Such files were only removed when the
+   future object itself was removed. This would result in a large
+   number of temporary files accumulating where there were many
+   futures processed. Now the backend finalizes the **callr** process
+   as soon as the future results have been collected, which results in
+   removing temporary files created by callr sooner. Previously, the
+   finalizer was only run when the future object was removed and
+   garbage collected.
+
+ * Package gave errors on "Error in as.POSIXct.numeric(time1) :
+   'origin' must be supplied" when using R (< 4.3.0).
+
+ * Package failed to  install 'wormhole-williams' automatically on ARM7
+   machines like Raspberry Pi.
+
+
 # Version 0.3.0 [2025-08-26]
 
 ## Significant Changes
@@ -24,4 +49,3 @@
 # Version 0.1.0 [2025-08-10]
 
 This is the first public version of the **future.p2p** package.
-

R/001.import_future_functions.R

@@ -15,3 +15,13 @@ import_future_functions <- function() {
 
   .debug <<- import_future(".debug", mode = "environment", default = new.env(parent = emptyenv()))
 }
+
+
+#' @importFrom utils packageVersion
+future_supports_state_submitted <- local({
+  .value <- NA
+  function() {
+    if (is.na(.value)) .value <<- (packageVersion("future") > "1.67.0")
+    .value
+  }
+})

R/PicoP2PFuture-class.R

@@ -24,11 +24,41 @@ resolved.PicoP2PFuture <- function(x, .signalEarly = TRUE, ...) {
 
   ## Still running?
   rx <- future[["rx"]]
-  resolved <- !rx$is_alive()
+  if (is.null(rx)) {
+    resolved <- TRUE
+  } else {
+    resolved <- !rx$is_alive()
+    dispatcher_status <- process_dispatcher_messages(rx, debug = debug)
+  }
+
+  state <- future[["state"]]
+  
+  ## Update state?
+  if (state == "submitted" || state == "running") {
+    if (debug) mdebugf("Future state before: %s", commaq(state))
+    if (debug) mdebugf("Child process updates: [n=%d] %s", length(dispatcher_status), commaq(dispatcher_status))
+    if ("wait" %in% dispatcher_status) state <- "running"
+    future[["state"]] <- state
+    if (debug) mdebugf("Future state after: %s", commaq(state))
+  }
+
+  ## Remove communication channels?
+  if (resolved) {
+    future[["state"]] <- "finished"
+    channels <- attr(rx, "channels", exact = TRUE)
+    if (length(channels) > 0) {
+      if (debug) mdebugf("Removing communication channel files: [n=%d] %s", length(channels), commaq(channels))
+      stop_if_not(is.character(channels))
+      file.remove(channels)
+      attr(rx, "channels") <- NULL
+      future[["rx"]] <- rx
+    }
+  }
 
   resolved
 }
 
+
 #' @importFrom future result UnexpectedFutureResultError
 #' @keywords internal
 #' @export
@@ -51,25 +81,66 @@ result.PicoP2PFuture <- function(future, ...) {
   }
 
   rx <- future[["rx"]]
+
   if (debug) mdebug("Waiting for dispatch process to finish")
   rx$wait()
-  file <- rx$get_result()
-  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)
+
+  ## Get the results
+  response <- rx$get_result()
+
+  ## Relay output, handle messages
+  dispatcher_status <- process_dispatcher_messages(rx, debug = debug)
+
+  ## Finalize the 'callr' process, which includes removing any temporary
+  ## files that it created
+  rx$finalize()
+
+  ## Remove communication channels
+  channels <- attr(rx, "channels", exact = TRUE)
+  if (length(channels) > 0) {
+    if (debug) mdebugf("Removing communication channel files: [n=%d] %s", length(channels), commaq(channels))
+    stop_if_not(is.character(channels))
+    file.remove(channels)
   }
-  
-  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())
+
+  future[["rx"]] <- NULL
+
+  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))))
+    }
+  } 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 <- readRDS(file)
-    file.remove(file)
-    result
-  })
+  
+    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)
+      stop_if_not(is.character(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

@@ -100,7 +100,7 @@ PicoP2PFutureBackend <- function(cluster = p2p_cluster_name(), host = "pipe.pico
   core <- structure(core, class = c("PicoP2PFutureBackend", "MultiprocessFutureBackend", "FutureBackend", class(core)))
 
   if (!p2p_can_connect(cluster, name = name, host = host, ssh_args = ssh_args)) {
-    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username())))
+    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username(host = host, ssh_args = ssh_args))))
   }
 
   core
@@ -144,6 +144,48 @@ 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)
+  }
+
+  local({
+    con <- file(tx, open = "w")
+    on.exit(close(con))
+    cat("interrupt\n", file = con)
+    flush(con)
+  })
+  if (debug) mdebug("Sent 'interrupt' to future dispatcher processes")
+  
+  future[["state"]] <- "interrupted"
+  
+  future
+}
+
+
 #' @importFrom future nbrOfWorkers
 #' @export
 nbrOfWorkers.PicoP2PFutureBackend <- function(evaluator) {
@@ -208,19 +250,21 @@ print.PicoP2PFutureBackend <- function(x, ...) {
   cat(sprintf("P2P cluster: %s\n", sQuote(backend[["cluster"]])))
   cat(sprintf("P2P client ID: %s\n", sQuote(backend[["name"]])))
 
-  clusters <- pico_p2p_hosted_clusters(backend[["host"]], backend[["ssh_args"]])
+  host <- backend[["host"]]
+  ssh_args <- backend[["ssh_args"]]
+  clusters <- pico_p2p_hosted_clusters(host = host, ssh_args = ssh_args)
   cat(sprintf("P2P clusters you are hosting: [n=%d]\n", nrow(clusters)))
   for (kk in seq_len(nrow(clusters))) {
     cluster <- clusters[kk, ]
     users <- strsplit(cluster$users, split = ",", fixed = TRUE)[[1]]
-    users <- unique(c(users, pico_username()))
+    users <- unique(c(users, pico_username(host = host, ssh_args = ssh_args)))
     users <- paste(users, collapse = ", ")
     cat(sprintf(" %2d. %s (users: %s)\n", kk, sQuote(cluster$name), users))
   }
 
   cat("Message board:\n")
   cat(sprintf(" - Server: %s\n", backend[["host"]]))
-  username <- pico_username(backend[["host"]], backend[["ssh_args"]])
+  username <- pico_username(host = host, ssh_args = ssh_args)
   cat(sprintf(" - Username: %s\n", sQuote(username)))
 
   cat("Data transfer tools:\n")
@@ -236,7 +280,7 @@ print.PicoP2PFutureBackend <- function(x, ...) {
 
 p2p_can_connect <- function(cluster, name, host = "pipe.pico.sh", ssh_args = NULL, timeout = 10.0) {
   cluster_owner <- dirname(cluster)
-  if (cluster_owner == pico_username()) {
+  if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
     topic <- sprintf("%s/future.p2p", basename(cluster))
   } else {
     topic <- sprintf("%s/future.p2p", cluster)

R/host_cluster.R

@@ -16,25 +16,25 @@
 #'
 #' @importFrom future resolve plan sequential
 #' @export
-host_cluster <- function(cluster = p2p_cluster_name(users), users = character(0L), host = "pipe.pico.sh", ssh_args = NULL, duration = 14*24*60*60) {
+host_cluster <- function(cluster = p2p_cluster_name(users, host = host, ssh_args = ssh_args), users = character(0L), host = "pipe.pico.sh", ssh_args = NULL, duration = 14*24*60*60) {
   stopifnot(length(cluster) == 1L, is.character(cluster), !is.na(cluster), nzchar(cluster))
 
   parts <- strsplit(cluster, split = "/", fixed = TRUE)[[1]]
   okay <- FALSE
   if (length(parts) == 1L) {
     okay <- TRUE
     cluster_name <- cluster
-    cluster_owner <- pico_username()
+    cluster_owner <- pico_username(host = host, ssh_args = ssh_args)
     cluster <- sprintf("%s/%s", cluster_owner, cluster_name)
   } else if (length(parts) == 2L) {
     cluster_owner <- parts[1]
-    if (cluster_owner == pico_username()) {
+    if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
       okay <- TRUE
       cluster_name <- parts[2]
     }
   }
   if (!okay) {
-    stop(sprintf("Argument 'cluster' must be of format '{owner}/{name}' or '{name}' where '{owner}' is your Pico username (%s): %s", sQuote(cluster), sQuote(pico_username())))
+    stop(sprintf("Argument 'cluster' must be of format '{owner}/{name}' or '{name}' where '{owner}' is your Pico username (%s): %s", sQuote(cluster), sQuote(pico_username(host = host, ssh_args = ssh_args))))
   }
 
   stopifnot(
@@ -56,7 +56,7 @@ host_cluster <- function(cluster = p2p_cluster_name(users), users = character(0L
   now <- pico_p2p_time()
 
   expires <- pico_p2p_time(delta = duration)
-  duration <- difftime(duration, 0)
+  duration <- difftime2(duration, 0)
 
   info("Launch p2p cluster %s for %d users (%s) until %s (%s)", sQuote(cluster), length(users), commaq(users), format(Sys.time() + duration), format(duration))
   topic <- sprintf("%s/future.p2p", cluster_name)

R/p2p_client_id.R

@@ -3,19 +3,21 @@
 #' @param users Users to have access to the cluster. This controls whether
 #' the default cluster names should be "personal" or "friends".
 #'
+#' @param \ldots Passed as-is to [pico_username()].
+#'
 #' @return
 #' `p2p_cluster_name()` returns R option `future.p2p.cluster`, if set.
 #' If not set, it returns `{pico_name}/personal` if `length(users) == 0`,
 #' otherwise `{pico_name}/friends`.
 #' 
 #' @export
-p2p_cluster_name <- function(users = character(0)) {
+p2p_cluster_name <- function(users = character(0), ...) {
   users <- unique(users)
   name <- getOption("future.p2p.cluster")
   if (is.null(name)) {
-    users <- setdiff(users, pico_username())
+    users <- setdiff(users, pico_username(...))
     name <- if (length(users) == 0) "personal" else "friends"
-    name <- sprintf("%s/%s", pico_username(), name)
+    name <- sprintf("%s/%s", pico_username(...), name)
   }
   name
 }
@@ -30,9 +32,9 @@ p2p_cluster_name <- function(users = character(0)) {
 #' @keywords internal
 p2p_client_id <- local({
   name <- NULL
-  function() {
+  function(...) {
     if (is.null(name)) {
-      user <- pico_username()
+      user <- pico_username(...)
       hostname <- Sys.info()[["nodename"]]
       pid <- Sys.getpid()
       name <<- sprintf("%s@%s:%d", user, hostname, pid)