Restructure: move makeClusterFunctionsSlurm2() to its own file [ci skip]

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future.batchtools
-Version: 0.20.0-9008
+Version: 0.20.0-9010
 Depends:
     R (>= 3.2.0),
     parallelly,

NEWS.md

@@ -1,6 +1,30 @@
 # Version (development version)
 
- * ...
+## Significant Changes
+
+ * `batchtools_slurm()` now uses `makeClusterFunctionsSlurm2()`.
+
+## New Features
+
+ * Add `makeClusterFunctionsSlurm2()`, which enhances
+   `batchtools::makeClusterFunctionsSlurm()` by patching the
+   `listJobsQueued()` cluster function such that it falls back to
+   querying Slurm's account database (`sacct`), if the future was
+   _not_ found in the Slurm job queue (`squeue`), which might be the
+   case when Slurm provisions a job that was just submitted to the
+   scheduler.
+
+## Bug Fixes
+
+  `batchtools_slurm()` would produce "Future of class
+  BatchtoolsSlurmFuture expired, which indicates that it crashed or
+  was killed" errors on some Slurm clusters. We believe this happened
+  because a recently submitted future job would not immediately show
+  up on the job queue, which caused **future.batchtools** to
+  incorrectly conclude that the job had already finished, but without
+  producing any results. `batchtools_slurm()` now uses the new
+  `makeClusterFunctionsSlurm2()`, which does a better job inferring
+  whether a job is queued or not.
 
 
 # Version 0.20.0 [2025-08-25]

R/batchtools_slurm.R

@@ -48,10 +48,10 @@ print.BatchtoolsSlurmFutureBackend <- function(x, ...) {
 #'
 #'  1. Slurm 21.08.4, Rocky Linux 8, NFS global filesystem (August 2025)
 #'  2. Slurm 22.05.11, Rocky Linux 8, NFS global filesystem (August 2025)*
-#'  3. Slurm 23.02.6, Ubuntu 24.04 LTS, NFS global filesystem (August 2025)*
+#'  3. Slurm 23.02.6, Ubuntu 24.04 LTS, NFS global filesystem (August 2025)
 #'  4. Slurm 24.11.3, AlmaLinux 9, Lustre global filesystem (September 2025)*
 #'
-#' (*) Verified with **future.batchtools** 1.20.0, which used
+#' (*) Verified with **future.batchtools** 0.20.0, which used
 #'     [batchtools::makeClusterFunctionsSlurm()], which the new
 #'     [makeClusterFunctionsSlurm2()] enhances.
 #'
@@ -154,139 +154,3 @@ slurm_version <- local({
     version
   }
 })
-
-
-# Patch Slurm cluster functions listJobsQueued() and listJobsRunning()
-# to use `sacct` instead of `squeue`
-#' @importFrom batchtools assertRegistry runOSCommand
-#' @importFrom utils tail
-patchClusterFunctionsSlurm <- function(cf) {
-  OSError <- import_from("OSError", package = "batchtools")
-  stopifnot(inherits(cf, "ClusterFunctions"))
-
-  env <- environment(cf[["listJobsQueued"]])
-  array.jobs <- env[["array.jobs"]]
-  getClusters <- env[["getClusters"]]
-  nodename <- env[["nodename"]]
-  
-  listJobs <- function(reg, args) {      
-    assertRegistry(reg, writeable = FALSE)
-    args <- c(args, "--user=$USER", "--noheader", "--parsable2", "--allocations", "--format=JobID")
-    clusters <- getClusters(reg)
-    if (length(clusters) > 0) {
-       args <- c(args, sprintf("--clusters=%s", clusters))
-    }
-    res <- runOSCommand("sacct", args, nodename = nodename)
-    if (res$exit.code > 0L) {
-      OSError("Listing of jobs failed", res)
-    }
-    if (length(clusters) > 0) {
-      res <- tail(res$output, -1L)
-    } else {
-      res <- res$output
-    }
-    res
-  } ## listJobs()
-  
-  cf$listJobsQueued <- function(reg) {
-    ## List PENDING (PD) and REQUEUED (RQ) jobs
-    listJobs(reg, "--state=PD,RQ")
-  }                                                          
-  
-  cf$listJobsRunning <- function(reg) {
-    ## List RUNNING (R), SUSPENDED (S), RESIZING (RS) jobs
-    listJobs(reg, "--state=R,S,RS")
-  }
-
-  cf
-} ## patchClusterFunctionsSlurm()
-
-
-
-# Patch Slurm cluster functions listJobsQueued() and listJobsRunning()
-# to use `sacct` instead of `squeue`
-#' @importFrom batchtools assertRegistry runOSCommand
-#' @importFrom utils tail
-patchClusterFunctionsSlurm2 <- function(cf) {
-  OSError <- import_from("OSError", package = "batchtools")
-  stopifnot(inherits(cf, "ClusterFunctions"))
-
-  env <- environment(cf[["listJobsQueued"]])
-  array.jobs <- env[["array.jobs"]]
-  getClusters <- env[["getClusters"]]
-  nodename <- env[["nodename"]]
-  org_listJobsQueued <- env[["listJobsQueued"]]
-
-  ## Allow for a 15-minute offset in time between host and Slurm's sacct server
-  isJobQueued <- function(reg, batch_id, since = NULL, offset = 15*60) {
-    stopifnot(length(batch_id) == 1L, !is.na(batch_id), nzchar(batch_id))
-    stopifnot(is.null(since) || inherits(since, "POSIXct"))
-    
-    args <- c("--user=$USER", "--noheader", "--parsable2", "--allocations", "--format=State", sprintf("--jobs=%s", batch_id), sprintf("--starttime=%s", format(since - offset, format = "%FT%T")))
-    clusters <- getClusters(reg)
-    if (length(clusters) > 0) {
-       args <- c(args, sprintf("--clusters=%s", clusters))
-    }
-    res <- runOSCommand("sacct", args, nodename = nodename)
-    if (res$exit.code > 0L) {
-      OSError("Failed to check if job is pending", res)
-    }
-    if (length(clusters) > 0) {
-      res <- tail(res$output, -1L)
-    } else {
-      res <- res$output
-    }
-
-    if (length(res) == 0) return(FALSE)
-    
-    res %in% c("PENDING", "REQUEUED")
-  } ## isJobQueued()
-  
-  cf$listJobsQueued <- function(reg) {
-    batch_id <- getOption("future.batchtools.batch_id", NULL)
-    
-    ## Queued jobs according to 'squeue'
-    jobs <- org_listJobsQueued(reg)
-    if (is.null(batch_id)) return(jobs)
-
-    ## Is the job queued?
-    if (length(jobs) > 0) {
-      jobs <- intersect(jobs, as.character(batch_id))
-      if (length(jobs) > 0) return(jobs)
-    }
-
-    ## Ask 'sacct' it if is PENDING or REQUEUED
-    submitted_on <- getOption("future.batchtools.submitted_on", NULL)
-    if (isJobQueued(reg, batch_id, since = submitted_on)) {
-      jobs <- as.character(batch_id)
-    }
-
-    jobs
-  }
-
-  cf
-} ## patchClusterFunctionsSlurm2()
-
-
-
-
-#' ClusterFunctions for Slurm Systems (robustified)
-#'
-#' This functions enhances [batchtools::makeClusterFunctionsSlurm()] by
-#' patching the `listJobsQueued()` cluster function such that it falls
-#' back to querying Slurm's account database (`sacct`), if the future was
-#' _not_ found in the Slurm job queue (`squeue`), which might be the case
-#' when Slurm provisions a job that was just submitted to the scheduler.
-#' 
-#' @inheritParams batchtools::makeClusterFunctionsSlurm
-#'
-#' @return
-#' A [batchtools::ClusterFunctions] object.
-#'
-#' @importFrom batchtools makeClusterFunctionsSlurm
-#' @export
-makeClusterFunctionsSlurm2 <- function(template = "slurm", array.jobs = TRUE, nodename = "localhost", scheduler.latency = 1, fs.latency = 65) {
-  cf <- makeClusterFunctionsSlurm(template = template, array.jobs = array.jobs, nodename = nodename, scheduler.latency = scheduler.latency, fs.latency =fs.latency)
-  cf <- patchClusterFunctionsSlurm2(cf)
-  cf
-}

R/makeClusterFutureSlurm2.R

@@ -0,0 +1,87 @@
+# Patch Slurm cluster functions listJobsQueued() and listJobsRunning()
+# to use `sacct` instead of `squeue`
+#' @importFrom batchtools assertRegistry runOSCommand
+#' @importFrom utils tail
+patchClusterFunctionsSlurm2 <- function(cf) {
+  OSError <- import_from("OSError", package = "batchtools")
+  stopifnot(inherits(cf, "ClusterFunctions"))
+
+  env <- environment(cf[["listJobsQueued"]])
+  array.jobs <- env[["array.jobs"]]
+  getClusters <- env[["getClusters"]]
+  nodename <- env[["nodename"]]
+  org_listJobsQueued <- env[["listJobsQueued"]]
+
+  ## Allow for a 15-minute offset in time between host and Slurm's sacct server
+  isJobQueued <- function(reg, batch_id, since = NULL, offset = 15*60) {
+    stopifnot(length(batch_id) == 1L, !is.na(batch_id), nzchar(batch_id))
+    stopifnot(is.null(since) || inherits(since, "POSIXct"))
+    
+    args <- c("--user=$USER", "--noheader", "--parsable2", "--allocations", "--format=State", sprintf("--jobs=%s", batch_id), sprintf("--starttime=%s", format(since - offset, format = "%FT%T")))
+    clusters <- getClusters(reg)
+    if (length(clusters) > 0) {
+       args <- c(args, sprintf("--clusters=%s", clusters))
+    }
+    res <- runOSCommand("sacct", args, nodename = nodename)
+    if (res$exit.code > 0L) {
+      OSError("Failed to check if job is pending", res)
+    }
+    if (length(clusters) > 0) {
+      res <- tail(res$output, -1L)
+    } else {
+      res <- res$output
+    }
+
+    if (length(res) == 0) return(FALSE)
+    
+    res %in% c("PENDING", "REQUEUED")
+  } ## isJobQueued()
+  
+  cf$listJobsQueued <- function(reg) {
+    batch_id <- getOption("future.batchtools.batch_id", NULL)
+    
+    ## Queued jobs according to 'squeue'
+    jobs <- org_listJobsQueued(reg)
+    if (is.null(batch_id)) return(jobs)
+
+    ## Is the job queued?
+    if (length(jobs) > 0) {
+      jobs <- intersect(jobs, as.character(batch_id))
+      if (length(jobs) > 0) return(jobs)
+    }
+
+    ## Ask 'sacct' it if is PENDING or REQUEUED
+    submitted_on <- getOption("future.batchtools.submitted_on", NULL)
+    if (isJobQueued(reg, batch_id, since = submitted_on)) {
+      jobs <- as.character(batch_id)
+    }
+
+    jobs
+  }
+
+  cf
+} ## patchClusterFunctionsSlurm2()
+
+
+
+
+#' ClusterFunctions for Slurm Systems (patched)
+#'
+#' This functions enhances [batchtools::makeClusterFunctionsSlurm()] by
+#' patching the `listJobsQueued()` cluster function such that it falls
+#' back to querying Slurm's account database (`sacct`), if the future was
+#' _not_ found in the Slurm job queue (`squeue`), which might be the case
+#' when Slurm provisions a job that was just submitted to the scheduler.
+#' 
+#' @inheritParams batchtools::makeClusterFunctionsSlurm
+#'
+#' @return
+#' A [batchtools::ClusterFunctions] object.
+#'
+#' @importFrom batchtools makeClusterFunctionsSlurm
+#' @export
+makeClusterFunctionsSlurm2 <- function(template = "slurm", array.jobs = TRUE, nodename = "localhost", scheduler.latency = 1, fs.latency = 65) {
+  cf <- makeClusterFunctionsSlurm(template = template, array.jobs = array.jobs, nodename = nodename, scheduler.latency = scheduler.latency, fs.latency =fs.latency)
+  cf <- patchClusterFunctionsSlurm2(cf)
+  cf
+}