Make it possible to also interrupt a future currently being processed

Author: HenrikBengtsson

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: future.p2p
 Title: A Peer-to-Peer Compute Cluster via Futureverse
-Version: 0.3.0-9025
+Version: 0.3.0-9026
 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),

R/pico_p2p.R

@@ -279,18 +279,32 @@ pico_p2p_receive_future <- function(p, via, duration = 60) {
 
 
 #' @importFrom future result
-pico_p2p_send_result <- function(p, future, via, duration = 60) {
+pico_p2p_send_result <- function(p, future, via, to, duration = 60, from = p$user) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
     mdebug_push("pico_p2p_send_result() ...")
     mdebugf("Future: %s", future_id(future))
+    mdebugf("From: %s", from)
+    mdebugf("To: %s", to)
     mdebugf("Via: %s", via)
     mdebugf("Duration: %g seconds", duration)
     on.exit({
       mdebug_pop()
     })
   }
 
+  ## Update client about incoming result on the message board
+  m <- data.frame(
+    when = now_str(),
+    expires = pico_p2p_time(delta = duration),
+    type = "result",
+    from = from,
+    to = to,
+    future = future_id(future),
+    via = via
+  )
+  m_res <- pico_send_message_dataframe(p, m)
+
   uri <- parse_transfer_uri(via)
   if (uri$protocol == "wormhole") {
     file <- file.path(tempdir(), sprintf("%s-FutureResult.rds", future_id(future)))
@@ -417,7 +431,10 @@ pico_p2p_dispatch_future <- function(future) {
       m <- pico_p2p_next_message(pico) ## This is non-block; may return NULL
 
       ## Skip if message is related to another future than ours
-      if (is.null(m[["future"]]) || m[["future"]] != future_id) next
+      if (is.null(m[["future"]]) || m[["future"]] != future_id) {
+        Sys.sleep(0.1)
+        next
+      }
 
       if (debug) mstr(list(m = m))
 
@@ -470,6 +487,44 @@ pico_p2p_dispatch_future <- function(future) {
 
     ## 5. Wait for and receive FutureResult file
     update_parent("wait")
+    repeat {
+      ## Check for interrupts
+      if ("interrupt" %in% listen_parent()) {
+        if (debug) mdebugf("interrupt (state = %s)", sQuote(state))
+        m <- pico_p2p_withdraw_future(pico, future_id = future_id, to = worker)
+        if (debug) mstr(list(m = m))
+      }
+
+      ## New message from message board?
+      m <- pico_p2p_next_message(pico) ## This is non-block; may return NULL
+
+      ## Skip if message is related to another future than ours
+      if (is.null(m[["future"]]) || m[["future"]] != future_id) {
+        Sys.sleep(0.1)
+        next
+      }
+      
+      if (debug) mstr(list(m = m))
+      
+      if (m[["type"]] == "result") {
+        ## Ignore, if result already expired
+        if (Sys.time() > as.POSIXct(as.numeric(m[["expires"]]))) {
+          if (debug) mdebug("Receiving expired results")
+          ## FIXME: We cannot just ignore the file transfer, because then the worker will
+          ## stall forever. To do that, we need the worker coordinator to interrupt the
+          ## transfer when expired.
+          ## return(list(type = "event", value = "expired"))
+        }
+        if (debug) mdebug("Receiving results")
+        state <- "receive"
+        break
+      }
+
+      Sys.sleep(0.1)
+    } ## repeat()
+
+    state <- "results"
+    update_parent("results")
     path <- file.path(dirname(dirname(file)), "results")
     tryCatch({
       file <- pico_p2p_receive_result(pico, via = via, path = path)

R/worker.R

@@ -235,7 +235,7 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
       if ("resolved" %in% worker_status) {
         state <- "resolved"
         info("Future %s has been resolved and results will be sent to client %s", sQuote(future), sQuote(client))
-        ## FIXME: Inform client that future has been resolved
+        m0 <- pico_p2p_take_on_future(p, to = client, future = future, duration = duration)
       }
       next
     } else if (state == "resolved") {
@@ -276,7 +276,7 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
           state <- "working"
 
           ## Tell worker to receive future from client
-          tx_worker(sprintf("download=%s,via=%s", future, via))
+          tx_worker(sprintf("download=%s,via=%s,from=%s", future, via, client))
 
           ## Wait for worker to *start* download future
           repeat {
@@ -306,6 +306,7 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
         ## because the client did not respect what we support)
         state <<- "waiting"
       } else if (state == "working") {
+        info("Interrupting worker")
         state <<- "interrupt"
         rx$interrupt()
       } else {
@@ -419,11 +420,12 @@ run_worker <- function(cluster, worker_id, host, ssh_args, duration, channels) {
     }
 
     ## Download and process future?
-    pattern <- "^download=([^,]+),via=(.*)$"
+    pattern <- "^download=([^,]+),via=([^,]+),from=([^,]+)$"
     if (grepl(pattern, action)) {
       future <- sub(pattern, "\\1", action)
       via <- sub(pattern, "\\2", action)
-      info("download future %s via %s", sQuote(future), sQuote(via))
+      client <- sub(pattern, "\\3", action)
+      info("download future %s via %s from %s", sQuote(future), sQuote(via), sQuote(client))
       stop_if_not(
         nzchar(future), !grepl("[,=]", future),
         nzchar(via), !grepl("[,=]", via)
@@ -440,15 +442,15 @@ run_worker <- function(cluster, worker_id, host, ssh_args, duration, channels) {
 
       info("process future %s", sQuoteLabel(f))
       dt <- system.time({
-        r <- tryCatch(result(f), error = identity)
+        r <- tryCatch({ result(f) }, error = identity)  ## Note, result() handles 'interrupt':s
       })
       dt <- difftime(dt[3], 0)
       info("Future %s resolved after %s", sQuote(future), format(dt))
       tx_parent("resolved")
 
       info("sending future result %s via %s", sQuote(future), sQuote(via))
       dt <- system.time({
-        res <- pico_p2p_send_result(p, future = f, via = via)
+        res <- pico_p2p_send_result(p, future = f, to = client, via = via)
       })
       dt <- difftime(dt[3], 0)
       info("future result %s sent in %s", sQuote(future), format(dt))