PERFORMANCE: Pre-validation of cluster futures had a thinko, causing it to take 0.1-0.2 seconds for each future being launched

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future
-Version: 1.58.0-9003
+Version: 1.58.0-9004
 Title: Unified Parallel and Distributed Processing in R for Everyone
 Depends:
     R (>= 3.2.0)

NEWS.md

@@ -2,6 +2,11 @@
 
 ## Performance
 
+ * The pre-validation of the cluster worker allotted to a future when
+   launched was unnecessarily expensive due to a thinko, e.g. it would
+   take ~0.1-0.2 seconds for a multisession future, whereas after the
+   fix it is effectly 0.0 seconds.
+ 
  * Calling `resolved()` on a lazy `ClusterFuture` would collect the
    result for the first _resolved_ future in order to free up one
    worker slot. Now this is only done if all slots are occupied. The

R/backend_api-11.ClusterFutureBackend-class.R

@@ -1025,7 +1025,7 @@ receiveMessageFromWorker <- local({
 ## to backend[["workers"]], before returning.
 #' @importFrom parallelly isConnectionValid isNodeAlive cloneNode
 #' @importFrom parallel clusterCall
-requestNode <- function(await, backend, timeout, delta, alpha) {
+requestNode <- function(await, backend, timeout, delta, alpha, validateWorker = TRUE) {
   debug <- isTRUE(getOption("future.debug"))
   if (debug) {
     mdebug_push("requestNode() ...")
@@ -1113,96 +1113,99 @@ requestNode <- function(await, backend, timeout, delta, alpha) {
   stop_if_not(is.numeric(node_idx), is.finite(node_idx), node_idx >= 1, node_idx <= length(workers))
   if (debug) mdebugf("Index of first available worker: %d", node_idx)
 
-  ## Validate that the cluster node is working
-  if (debug) mdebug_push("Validate that the worker is functional ...")
-  cl <- workers[node_idx]
-  stop_if_not(length(cl) == 1L, inherits(cl, "cluster"))
-
-  maxTries <- 3L
-  for (kk in maxTries:1) {
-    okay <- TRUE
+  ## Validate that the cluster node is working - if not, relaunch it
+  if (validateWorker) {
+    if (debug) mdebug_push("Validate that the worker is functional ...")
+    cl <- workers[node_idx]
+    stop_if_not(length(cl) == 1L, inherits(cl, "cluster"))
+  
     truth <- "future:::requestNode() validation call"
-    res <- tryCatch({
-      suppressWarnings({
-        clusterCall(cl = cl, function(x) x, truth)[[1]]
-      })
-    }, error = identity)
 
-    ## If not working, investigate why, and relaunch a new worker
-    if (inherits(res, "error") || !identical(res, truth)) {
-      if (debug) {
-        mdebug("Worker is non-functional")
-        if (inherits(res, "error")) {
-          mdebug("Error received: ", conditionMessage(res))
-        } else {
-          mdebug("Result received: ", sQuote(res))
+    maxTries <- 3L
+    for (kk in maxTries:1) {
+      okay <- TRUE
+      res <- tryCatch({
+        suppressWarnings({
+          clusterCall(cl = cl, identity, truth)[[1]]
+        })
+      }, error = identity)
+      
+      ## If not working, investigate why, and relaunch a new worker
+      if (inherits(res, "error") || !identical(res, truth)) {
+        if (debug) {
+          mdebug("Worker is non-functional")
+          if (inherits(res, "error")) {
+            mdebug("Error received: ", conditionMessage(res))
+          } else {
+            mdebug("Result received: ", sQuote(res))
+          }
         }
-      }
-      okay <- FALSE
-  
-      ## Is the connection working?
-      node <- cl[[1]]
-      con <- node[["con"]]
-      connectionOkay <- NA
-      if (inherits(con, "connection")) {
-        connectionOkay <- isConnectionValid(con)
-        if (debug) mdebug("Connection is valid: ", connectionOkay)
-      }
-  
-      if (is.na(connectionOkay) || connectionOkay) {
-        ## If the node does not use a connection, or the connection is working,
-        ## we can only assume the worker is also alive. If so, we should try to
-        ## kill the worker.
-        res <- suppressWarnings(killNode(node))
-        if (debug) mdebugf("Killed %s: %s", class(node)[1], res)
-      } else {
-        ## If connection is not working, we could assume the worker is no longer
-        ## alive, but it could also be a network issues. In either case, we
-        ## should try to kill it, just in case.
-        res <- suppressWarnings(killNode(node))
-        if (debug) mdebugf("Killed %s: %s", class(node)[1], res)
-      }
-      if (kk == 1L) {
-        stop(FutureError(sprintf("Failed to find a functional cluster worker, after attempting to relaunch the parallel worker %d times", maxTries)))
-      }
-    } else {
-      if (debug) mdebug("Worker is functional")
-      break
-    }
+        okay <- FALSE
 
-    ## Relaunch worker?
-    if (!okay) {
-      if (debug) mdebugf_push("Restarting non-alive cluster node %d ...", node_idx)
-      node2 <- tryCatch({
-        cloneNode(node)
-      }, error = identity)
-      if (inherits(node2, "error")) {
-        msg <- sprintf("One of the future workers of class %s, part of a cluster of class %s, was interrupted and attempts to relaunch it failed", sQuote(class(node)[1]), sQuote(class(cl)[1]))
-        if (inherits(node, c("SOCKnode", "SOCK0node")) &&
-            !inherits(node, c("RichSOCKnode"))) {
-          msg <- sprintf("%s. If you created your cluster with parallel::makeCluster(), try with parallelly::makeClusterPSOCK() instead", msg)
+        ## Is the connection working?
+        node <- cl[[1]]
+        con <- node[["con"]]
+        connectionOkay <- NA
+        if (inherits(con, "connection")) {
+          connectionOkay <- isConnectionValid(con)
+          if (debug) mdebug("Connection is valid: ", connectionOkay)
+        }
+    
+        if (is.na(connectionOkay) || connectionOkay) {
+          ## If the node does not use a connection, or the connection is working,
+          ## we can only assume the worker is also alive. If so, we should try to
+          ## kill the worker.
+          res <- suppressWarnings(killNode(node))
+          if (debug) mdebugf("Killed %s: %s", class(node)[1], res)
+        } else {
+          ## If connection is not working, we could assume the worker is no longer
+          ## alive, but it could also be a network issues. In either case, we
+          ## should try to kill it, just in case.
+          res <- suppressWarnings(killNode(node))
+          if (debug) mdebugf("Killed %s: %s", class(node)[1], res)
+        }
+        if (kk == 1L) {
+          stop(FutureError(sprintf("Failed to find a functional cluster worker, after attempting to relaunch the parallel worker %d times", maxTries)))
         }
-        msg <- sprintf("%s. The reported reason was: %s", msg, conditionMessage(node2))
-        stop(FutureError(msg))
       } else {
-        node <- node2
+        if (debug) mdebug("Worker is functional")
+        break
       }
 
-      cl[[1]] <- node
-      
-      workers[[node_idx]] <- node
-      backend[["workers"]] <- workers
-      
-      if (debug) {
-        mdebug("Re-launched cluster node:")
-        mprint(node)
-        mdebugf_pop()
+      ## Relaunch worker?
+      if (!okay) {
+        if (debug) mdebugf_push("Restarting non-alive cluster node %d ...", node_idx)
+        node2 <- tryCatch({
+          cloneNode(node)
+        }, error = identity)
+        if (inherits(node2, "error")) {
+          msg <- sprintf("One of the future workers of class %s, part of a cluster of class %s, was interrupted and attempts to relaunch it failed", sQuote(class(node)[1]), sQuote(class(cl)[1]))
+          if (inherits(node, c("SOCKnode", "SOCK0node")) &&
+              !inherits(node, c("RichSOCKnode"))) {
+            msg <- sprintf("%s. If you created your cluster with parallel::makeCluster(), try with parallelly::makeClusterPSOCK() instead", msg)
+          }
+          msg <- sprintf("%s. The reported reason was: %s", msg, conditionMessage(node2))
+          stop(FutureError(msg))
+        } else {
+          node <- node2
+        }
+        
+        cl[[1]] <- node
+        
+        workers[[node_idx]] <- node
+        backend[["workers"]] <- workers
+        
+        if (debug) {
+          mdebug("Re-launched cluster node:")
+          mprint(node)
+          mdebugf_pop()
+        }
       }
-    }
-
-    ## Try again
-    Sys.sleep(0.1)
-  } ##  for (kk in maxTries:1)
+  
+      ## Try again
+      Sys.sleep(0.1)
+    } ##  for (kk in maxTries:1)
+  } ## if (validateWorker)
 
   ## Assert that there is no other registered future that is using
   ## the found node
@@ -1214,7 +1217,6 @@ requestNode <- function(await, backend, timeout, delta, alpha) {
     }
   }
 
-
   if (debug) mdebug_pop()
 
   node_idx