Workers are now keeping track of outstanding requests - not just the most recent one

HenrikBengtsson · HenrikBengtsson · commit 8b8a310f3299 · 2025-11-14T20:31:01.000-08:00
diff --git a/DESCRIPTION b/DESCRIPTION
@@ -1,6 +1,6 @@
 Package: future.p2p
 Title: A Peer-to-Peer Compute Cluster via Futureverse
-Version: 0.3.0-9020
+Version: 0.3.0-9021
 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),
diff --git a/R/worker.R b/R/worker.R
@@ -105,7 +105,15 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
   future <- NULL
   client <- NULL
   offer_expires <- Inf
-
+  
+  ## List of known requests
+  requests <- data.frame(
+    future = character(0L),
+    expires = as.POSIXct(NA_real_)[FALSE],
+    client = character(0L)
+  )
+  stop_if_not(is.data.frame(requests))
+  
   info("waiting for request")
   
   repeat tryCatch({
@@ -130,6 +138,7 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
       
       if ("ready" %in% worker_status) {
        info("worker process is ready")
+       if (state == "interrupt") state <- "waiting"
       }
       if ("interrupted" %in% worker_status) {
         signalCondition(worker_interrupt())
@@ -148,88 +157,87 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
 
     ## Any messages from the P2P message board?
 #    res <- poll(list(p), ms = 100)[[1]]
-#    if (!"ready" %in% res[["output"]]) next
+#    if ("ready" %in% res[["output"]]) {
 
-    ## Read next message?
-    m <- pico_p2p_next_message(p)
+    ## New message message?
+    m <- pico_p2p_next_message(p) ## This is non-block; may return NULL
     
     ## Expired?
     now <- Sys.time()
     if (now > expires) {
       info("expired")
-      signalCondition(future_withdraw("worker expired; terminating"))
+      signalCondition(future_withdraw("worker expired; terminating", future = future))
       next
     } else if (state == "offer" && now > offer_expires) {
       info("work offer expired")
-      signalCondition(future_withdraw("worker offer expired"))
+      signalCondition(future_withdraw("worker offer expired", future = future))
       next
     }
     
-    ## Process request?
+    ## Process messages from the message board?
     if (length(m) > 0) {
-      ## Are we read to offer to do work?
-      if (state == "waiting" && m[["type"]] == "request") {
-        stop_if_not(is.null(future), is.null(client))
-        future <- m[["future"]]
-        client <- m[["from"]]
-        
-        ## Make a work offer for 15 seconds
-        duration <- 15.0
-        info("offer to process future %s for client %s (valid for %g seconds)", sQuote(future), sQuote(client), duration)
-        
-        state <- "offer"
-        m0 <- pico_p2p_take_on_future(p, to = client, future = future, duration = duration)
-        offer_expires <- m0[["expires"]]
-      } else if (state == "offer" && future %in% m[["future"]]) {
-        info("waiting for acceptance of our work offer")
-        if (m[["type"]] == "accept") {
-          if (m[["to"]] == worker_id) {
-            info("client %s accepted our offer to process future %s", sQuote(client), sQuote(future))
-  
-            ## Do we support the file transfer protocol?
-            via <- m[["via"]]
-            uri <- parse_transfer_uri(via)
-            if (!uri[["protocol"]] %in% supported_transfer_protocols()) {
-              info("non-supported protocol")
-              signalCondition(future_withdraw(sprintf("non-supported file-transfer protocol: %s", uri[["protocol"]])))
-            }
-            
-            state <- "working"
-            
-            ## Tell worker to receive future from client
-            tx_worker(sprintf("download=%s,via=%s", future, via))
-            
-            ## Wait for worker to *start* download future
-            repeat {
-              worker_status <- process_worker_messages(rx, debug = debug)
-              if ("downloading" %in% worker_status) {
-                ## FIXME: Acknowledge to work on future
-                break
-              }
-              Sys.sleep(0.1)
-            }
-          } else {
-            info("withdraw offer for future %s, because client %s accepted another worker's offer", sQuote(future), sQuote(client))
-            signalCondition(future_withdraw("another worker took on the future"))
-          }
-        } else if (m[["type"]] == "withdraw") {
-          signalCondition(future_withdraw())
-        }
-      } else if (state == "working") {
-        ## Withdrawal of future?
-        if (m[["type"]] == "withdraw" && future %in% m[["future"]]) {
-          signalCondition(future_withdraw())
+      type <- m[["type"]]
+      
+      ## A request?
+      if (type == "request") {
+        ## A new request?
+        if (! m[["future"]] %in% requests[["future"]]) {
+          request <- data.frame(
+            future = m[["future"]],
+            expires = as.POSIXct(as.numeric(m[["expires"]])),
+            client = m[["from"]]
+          )
+          requests <- rbind(requests, request)
+          stop_if_not(is.data.frame(requests))
         }
       }
-    } ## if (length(m) > 0)
+
+      ## Request accepted by another worker
+      if (m[["type"]] == "accept" && m[["to"]] != worker_id) {
+        info("withdraw offer for future %s, because client %s accepted another worker's offer", sQuote(m[["future"]]), sQuote(m[["from"]]))
+        signalCondition(future_withdraw("another worker took on the future", future = m[["future"]]))
+      }
+
+      ## Withdrawal of future?
+      if (type == "withdraw") {
+        signalCondition(future_withdraw(future = m[["future"]]))
+      }
+    }
+
+    ## Drop expired requests
+    requests <- subset(requests, expires >= now)
+    stop_if_not(is.data.frame(requests))
     
-    if (state == "working") {
+    if (nrow(requests) > 0) {
+      if (debug) {
+        mdebugf("Known requests: [n=%d]", nrow(requests))
+        mprint(requests)
+      }
+    }
+
+    ## Are we read to offer to do work?
+    if (state == "waiting" && nrow(requests) > 0) {
+      stop_if_not(is.null(future), is.null(client))
+      request <- head(requests, 1L)
+      future <- request[["future"]]
+      client <- request[["client"]]
+      
+      ## Make a work offer for 15 seconds
+      duration <- 15.0
+      info("offer to process future %s for client %s (valid for %g seconds)", sQuote(future), sQuote(client), duration)
+        
+      state <- "offer"
+      m0 <- pico_p2p_take_on_future(p, to = client, future = future, duration = duration)
+      offer_expires <- m0[["expires"]]
+      next
+    } else if (state == "working") {
       ## Check if worker is done
       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
       }
+      next
     } else if (state == "resolved") {
       ## Check if future results have been transferred
       if ("ready" %in% worker_status) {
@@ -240,7 +248,48 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
         info("Future %s has been resolved and results have been sent to client %s", sQuote(future), sQuote(client))
         info("waiting for request")
       }
+      next
     }
+
+    if (length(m) > 0) {
+      if (state == "offer" && future %in% m[["future"]]) {
+        info("waiting for acceptance of our work offer")
+        if (m[["type"]] == "accept" && m[["to"]] == worker_id) {
+          info("client %s accepted our offer to process future %s", sQuote(client), sQuote(future))
+  
+          ## Do we support the file transfer protocol?
+          via <- m[["via"]]
+          uri <- parse_transfer_uri(via)
+          if (!uri[["protocol"]] %in% supported_transfer_protocols()) {
+            info("non-supported protocol")
+            signalCondition(future_withdraw(sprintf("non-supported file-transfer protocol: %s", uri[["protocol"]]), future = m[["future"]]))
+            state <- "waiting"
+            next
+          }
+
+          ## Drop future from list of requests
+          if (!is.null(m[["future"]])) {
+            requests <- subset(requests, future != m[["future"]])
+            stop_if_not(is.data.frame(requests))
+          }
+          
+          state <- "working"
+            
+          ## Tell worker to receive future from client
+          tx_worker(sprintf("download=%s,via=%s", future, via))
+         
+          ## Wait for worker to *start* download future
+          repeat {
+            worker_status <- process_worker_messages(rx, debug = debug)
+            if ("downloading" %in% worker_status) {
+              ## FIXME: Acknowledge to work on future
+              break
+            }
+            Sys.sleep(0.1)
+          }
+        }
+      }
+    } ## if (length(m) > 0)
   }, future_withdraw = function(c) {
     msg <- conditionMessage(c)
     info <- sprintf("state %s", sQuote(state))
@@ -249,22 +298,30 @@ worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args),
     info <- paste(info, collapse = ", ")
     msg <- sprintf("%s [%s]", msg, info)
     info(msg)
-    
-    ## Client withdrew future
-    if (state == "waiting") {
-      state <<- "waiting"
-    } else if (state == "offer") {
-      ## FIXME: Decline work offer (although we can just ignore it
-      ## because the client did not respect what we support)
-      state <<- "waiting"
-    } else if (state == "working") {
-      state <<- "interrupt"
-      rx$interrupt()
+
+    ## Client withdrew a future we're either work on or offered to work on
+    if (!is.null(future) && (future %in% c[["future"]])) {
+      if (state == "offer") {
+        ## FIXME: Decline work offer (although we can just ignore it
+        ## because the client did not respect what we support)
+        state <<- "waiting"
+      } else if (state == "working") {
+        state <<- "interrupt"
+        rx$interrupt()
+      } else {
+        stop(FutureError(sprintf("Internal error: state %s", sQuote(state))))
+      }
+      offer_expires <<- Inf
+      future <<- NULL
+      client <<- NULL
     }
-    
-    offer_expires <<- Inf
-    future <<- NULL
-    client <<- NULL
+
+    ## Drop future from list of requests, in case it's there
+    if (!is.null(c[["future"]])) {
+      requests <<- subset(requests, future != c[["future"]])
+      stop_if_not(is.data.frame(requests))
+    }
+
     info("waiting for request")
     ## FIXME: Acknowledge withdrawal of future
   }, worker_interrupt = function(c) {
@@ -412,8 +469,9 @@ run_worker <- function(cluster, worker_id, host, ssh_args, duration, channels) {
 cli_fcn(worker) <- c("--(cluster)=(.*)", "--(host)=(.*)", "--(ssh_args)=(.*)", "--(duration)=([[:digit:]]+)")
 
 
-future_withdraw <- function(message = "future withdrawn by client", call = NULL) {
+future_withdraw <- function(message = "future withdrawn by client", call = NULL, future = NULL) {
   cond <- simpleCondition(message = message, call = call)
+  cond[["future"]] <- future
   class(cond) <- c("future_withdraw", class(cond))
   cond
 }
@@ -427,7 +485,7 @@ worker_interrupt <- function(message = "worker process interrupted", call = NULL
 
 
 process_worker_messages <- function(rx, debug = FALSE) {
-  if (debug) {
+  if (debug && isTRUE(getOption("future.debug"))) {
     mdebug_push("process_worker_messages() ...")
     on.exit({
       mdebugf("worker_status: [n=%d] %s", length(worker_status), commaq((worker_status)))
