Merge branch 'release/0.21.0'

Author: HenrikBengtsson

DESCRIPTION

@@ -1,12 +1,14 @@
 Package: future.batchtools
-Version: 0.20.0
+Version: 0.21.0
 Depends:
     R (>= 3.2.0),
     parallelly,
     future (>= 1.58.0)
 Imports:
     batchtools (>= 0.9.17),
-    utils
+    utils,
+    checkmate,
+    stringi
 Suggests:
     globals,
     future.apply,
@@ -32,5 +34,5 @@ URL: https://future.batchtools.futureverse.org, https://github.com/futureverse/f
 BugReports: https://github.com/futureverse/future.batchtools/issues
 Language: en-US
 Encoding: UTF-8
-RoxygenNote: 7.3.2
+RoxygenNote: 7.3.3
 Roxygen: list(markdown = TRUE)

NAMESPACE

@@ -59,7 +59,9 @@ export(batchtools_torque)
 export(loggedError)
 export(loggedOutput)
 export(makeClusterFunctionsBash)
+export(makeClusterFunctionsSlurm2)
 importFrom(batchtools,Worker)
+importFrom(batchtools,assertRegistry)
 importFrom(batchtools,batchExport)
 importFrom(batchtools,batchMap)
 importFrom(batchtools,cfBrewTemplate)
@@ -88,6 +90,9 @@ importFrom(batchtools,saveRegistry)
 importFrom(batchtools,setJobNames)
 importFrom(batchtools,submitJobs)
 importFrom(batchtools,waitForJobs)
+importFrom(checkmate,"%??%")
+importFrom(checkmate,assertCharacter)
+importFrom(checkmate,assertString)
 importFrom(future,FutureBackend)
 importFrom(future,FutureError)
 importFrom(future,FutureInterruptError)
@@ -107,6 +112,8 @@ importFrom(future,tweak)
 importFrom(parallelly,availableCores)
 importFrom(parallelly,availableWorkers)
 importFrom(parallelly,supportsMulticore)
+importFrom(stringi,stri_flatten)
+importFrom(stringi,stri_replace_all_fixed)
 importFrom(tools,pskill)
 importFrom(utils,capture.output)
 importFrom(utils,file_test)

NEWS.md

@@ -1,3 +1,53 @@
+# Version 0.21.0 [2025-09-20]
+
+## Significant Changes
+
+ * `batchtools_slurm()` now uses `makeClusterFunctionsSlurm2()`.
+
+## New Features
+
+ * Add support for controlling the 'Rscript' call in the built-in job
+   script templates. This can be done via fields `rscript` and
+   `rscript_args` of the `resources` argument.
+ 
+ * Add support for setting environment variables in the built-in job
+   script templates. This can be done via field `envs` of the
+   `resources` argument.
+ 
+ * The built-in job script templates assert that the `Rscript`
+   launcher is found, and if not, they give an informative error
+   message suggesting to declare environment modules, via the
+   `resources` argument, that should be loaded by the job script.
+
+ * Add `makeClusterFunctionsSlurm2()`, which patches
+   `batchtools::makeClusterFunctionsSlurm()`.  Firstly, it patches 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.  Secondly, it patched the `submitJob()` cluster function
+   such that the system call to `sbatch` captures stderr separately
+   from stdout, which prevents auxiliary INFO messages from `sbatch`
+   to corrupt the output to be parsed.
+
+## Documentation
+
+ * Add example on how to configure `batchtools_slurm()` to run R
+   within a Linux container.
+
+## 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]
 
 ## Significant Changes

R/BatchtoolsFutureBackend-class.R

@@ -30,14 +30,27 @@
 #' job-script template as variable `resources`. This is based on how
 #' [batchtools::submitJobs()] works, with the exception for specially
 #' reserved names defined by the \pkg{future.batchtools} package;
-#' * `resources[["asis"]]` is a character vector that are passed as-is to
-#'   the job script and are injected as job resource declarations.
-#' * `resources[["modules"]]` is character vector of Linux environment
-#'   modules to be loaded.
-#' * `resources[["startup"]]` and `resources[["shutdown"]]` are character
-#'   vectors of shell code to be injected to the job script as-is.
+#'
 #' * `resources[["details"]]`, if TRUE, results in the job script outputting
 #'   job details and job summaries at the beginning and at the end.
+#'
+#' * `resources[["startup"]]` and `resources[["shutdown"]]` are character
+#'   vectors of shell code to be injected to the job script as-is.
+#'
+#' * `resources[["modules"]]` is character vector of Linux environment
+#'   modules to be loaded.
+#'
+#' * `resources[["envs"]]`, is an optional names character vector specifying
+#'   environment variables to be set.
+#'
+#' * `resources[["rscript"]]` is an optional character vector specifying
+#'   how the 'Rscript' is launched. The `resources[["rscript_args"]]` field
+#'   is an optional character vector specifying the 'Rscript' command-line
+#'   arguments.
+#'
+#' * `resources[["asis"]]` is a character vector that are passed as-is to
+#'   the job script and are injected as job resource declarations.
+#'
 #' * All remaining `resources` named elements are injected as named resource
 #'   specification for the scheduler.
 #'
@@ -325,7 +338,8 @@ launchFuture.BatchtoolsFutureBackend <- local({
     resources <- backend[["resources"]]
     config[["resources"]] <- resources
     future[["config"]] <- config
-
+    submitted_on <- Sys.time()
+    
     ## WORKAROUND: batchtools::submitJobs() updates the RNG state,
     ## which we must make sure to undo.
     tryCatch({
@@ -343,9 +357,10 @@ launchFuture.BatchtoolsFutureBackend <- local({
       msg <- sprintf("%s\nDETAILS:\nThe batchtools registry path: %s", msg, sQuote(path))
       stop(FutureLaunchError(msg, future = future))
     })
-    
+
     if (debug) mdebugf("Launched future #%d", jobid$job.id)
 
+    future[["submitted_on"]] <- submitted_on
     future[["state"]] <- "running"
 
     ## 6. Reserve worker for future
@@ -535,6 +550,26 @@ status <- function(future, ...) {
 
   jobid <- config$jobid
   if (is.na(jobid)) return("not submitted")
+
+  ## Optionally filter by the scheduler's job ID, if it exists
+  batch_id <- reg[["status"]][["batch.id"]]
+  if (!is.null(batch_id) && inherits(future, "BatchtoolsTemplateFutureBackend")) {
+    if (!is.character(batch_id) || length(batch_id) != 1L || is.na(batch_id) || !nzchar(batch_id) || !grepl("^[[:digit:].]+$", batch_id)) {
+      stop(sprintf("Unknown value of 'batch.id': [class=%s] %s", class(batch_id)[1], paste(sQuote(batch_id), collapse = ", ")))
+    }
+  
+    ## Pass this to cluster functions listJobsQueued() and listJobsRunning()
+    ## via an R option, because we cannot pass as an argument.
+    options(
+      future.batchtools.batch_id = batch_id,
+      future.batchtools.submitted_on = future[["submitted_on"]]
+    )
+    on.exit(options(
+      future.batchtools.batch_id = NULL,
+      future.batchtools.submitted_on = NULL
+    ), add = TRUE)
+  }
+  
   status <- get_status(reg = reg, ids = jobid)
   status <- (unlist(status) == 1L)
   status <- status[status]
@@ -667,7 +702,7 @@ resolved.BatchtoolsFuture <- function(x, ...) {
   ## Assert that the process that created the future is
   ## also the one that evaluates/resolves/queries it.
   assertOwner(x)
-
+  
   ## If not, checks the batchtools registry status
   resolved <- finished(x)
   if (is.na(resolved)) return(FALSE)
@@ -845,25 +880,69 @@ await <- function(future, cleanup = TRUE, ...) {
       ## how we can distinguish the two right now, but I'll assume that
       ## started jobs have a 'submitted' or 'started' status flag too,
       ## whereas jobs that failed to launch won't. /HB 2025-07-15
+      hints <- NULL
+      
+      state <- future[["state"]]
+      info <- sprintf("Future state: %s", sQuote(state))
+      hints <- c(hints, info)
+      info <- sprintf("Batchtools status: %s", commaq(stat))
+      hints <- c(hints, info)
+
+      ## SPECIAL CASE: Some Slurm users report on 'expired' jobs, although they never started.
+      ## Output more breadcrumbs to be able to narrow in on what causes this. /HB 2025-09-07
+      if (inherits(future, "BatchtoolsSlurmFuture")) {
+        batch_id <- reg[["status"]][["batch.id"]]
+        if (length(batch_id) > 0) {
+          info <- sprintf("Slurm job ID: [n=%d] %s", length(batch_id), commaq(batch_id))
+	  
+          args <- c("--noheader", "--format='job_id=%i,state=%T,submitted_on=%V,time_used=%M'", sprintf("--jobs=%s", paste(batch_id, collapse = ",")))
+          res <- system2("squeue", args = args, stdout = TRUE, stderr = TRUE)
+	  if (length(res) == 0) {
+	    res <- "<empty>"
+	  } else {
+            res <- paste(res, collapse = "; ") ## should only be one, but just in case ...
+	  }
+          info <- c(info, sprintf("Slurm 'squeue' job status: %s", res))
+
+          args <- c("--noheader", "--parsable2", "--allocations", "--format='JobID,State,ExitCode'", sprintf("--jobs=%s", paste(batch_id, collapse = ",")))
+          res <- system2("sacct", args = args, stdout = TRUE, stderr = TRUE)
+	  if (length(res) == 0) {
+	    res <- "<empty>"
+	  } else {
+            res <- paste(res, collapse = "; ") ## should only be one, but just in case ...
+	  }
+          info <- c(info, sprintf("Slurm 'sacct' job status: %s", res))
+	} else {
+          info <- "Slurm job ID: <not found>"
+          info <- c(info, sprintf("Slurm job status: <unknown>"))
+        }
+	hints <- c(hints, info)
+      }
 
-      hint <- tryCatch({
+      ## TROUBLESHOOTING: Logged output
+      info <- tryCatch({
         output <- loggedOutput(future, timeout = 0.0)
-        hint <- unlist(strsplit(output, split = "\n", fixed = TRUE))
-        hint <- hint[nzchar(hint)]
-        hint <- tail(hint, n = getOption("future.batchtools.expiration.tail", 48L))
+        info <- unlist(strsplit(output, split = "\n", fixed = TRUE))
+        info <- info[nzchar(info)]
+        info <- tail(info, n = getOption("future.batchtools.expiration.tail", 48L))
       }, error = function(e) NULL)
-      if (length(hint) > 0) {
-        hint <- c("The last few lines of the logged output:", hint)
-        hint <- paste(hint, collapse = "\n")
+
+      if (length(info) > 0) {
+        info <- c("The last few lines of the logged output:", info)
       } else {
-        hint <- "No logged output file exist (at the moment)"
+        info <- "No logged output file exist (at the moment)"
       }
+      hints <- c(hints, info)
 
+      if (length(hints) > 0) {
+        hints <- c("\nPost-mortem details:", hints)
+        hints <- paste(hints, collapse = "\n")
+      }	
       if (any(c("submitted", "started") %in% stat)) {
-        msg <- sprintf("Future (%s) of class %s expired, which indicates that it crashed or was killed. %s", label, class(future)[1], hint)
+        msg <- sprintf("Future (%s) of class %s expired, which indicates that it crashed or was killed.%s", label, class(future)[1], hints)
         result <- FutureInterruptError(msg, future = future)
       } else {
-        msg <- sprintf("Future (%s) of class %s failed to launch. %s", label, class(future)[1], hint)
+        msg <- sprintf("Future (%s) of class %s failed to launch.%s", label, class(future)[1], hints)
         result <- FutureLaunchError(msg, future = future)
       }
     } else if (future[["state"]] %in% c("canceled", "interrupted")) {

R/BatchtoolsTemplateFutureBackend-class.R

@@ -71,7 +71,7 @@ BatchtoolsTemplateFutureBackend <- function(type, scheduler.latency = 1.0, fs.la
     lsf      = makeClusterFunctionsLSF,
     openlava = makeClusterFunctionsOpenLava,
     sge      = makeClusterFunctionsSGE,
-    slurm    = makeClusterFunctionsSlurm,
+    slurm    = makeClusterFunctionsSlurm2,
     torque   = makeClusterFunctionsTORQUE,
                makeClusterFunctions 
   )

R/batchtools_bash.R

@@ -112,16 +112,28 @@ makeClusterFunctionsBash <- function(template = "bash", fs.latency = 0.0, ...) {
     stop_if_not(inherits(reg, "Registry"))
     stop_if_not(inherits(jc, "JobCollection"))
 
-    script <- cfBrewTemplate(reg, text = template_text, jc = jc)
-    output <- system2(bin, args = c(script), stdout = TRUE, stderr = TRUE, wait = TRUE)
     debug <- isTRUE(getOption("future.debug"))
     if (debug) {
-      mdebug_push("makeClusterFunctionsBash() ...")
-      mdebug(paste(c(output, ""), collapse = "\n"))
+      mdebug_push("makeClusterFunctionsBash() -> submitJob() ...")
       on.exit(mdebug_pop())
     }
+
+    script <- cfBrewTemplate(reg, text = template_text, jc = jc)
+    if (debug) {
+      mdebugf("job script: %s\n", script)
+      bfr <- readLines(script, warn = FALSE)
+      mdebugf("[job script]: %s\n", bfr)
+    }
+    
+    output <- system2(bin, args = c(script), stdout = TRUE, stderr = TRUE, wait = TRUE)
+    if (debug) {
+      mdebug(paste(c(sprintf("[job output]: %s", output), ""), collapse = "\n"))
+    }
 
     status <- attr(output, "status")
+    if (debug) {
+      mstr(list(status = status))
+    }
     if (is.null(status)) {
       status <- 0L
       batch.id <- sprintf("bash#%d", Sys.getpid())

R/batchtools_lsf.R

@@ -50,10 +50,11 @@ BatchtoolsLsfFutureBackend <- function(...) {
 #'
 #' f <- future({
 #'   data.frame(
-#'     hostname = Sys.info()[["nodename"]],
-#'           os = Sys.info()[["sysname"]],
-#'        cores = unname(parallelly::availableCores()),
-#'      modules = Sys.getenv("LOADEDMODULES")
+#'      hostname = Sys.info()[["nodename"]],
+#'            os = Sys.info()[["sysname"]],
+#'     osVersion = utils::osVersion,
+#'         cores = unname(parallelly::availableCores()),
+#'       modules = Sys.getenv("LOADEDMODULES")
 #'   )
 #' })
 #' info <- value(f)

R/batchtools_openlava.R

@@ -50,10 +50,11 @@ BatchtoolsOpenLavaFutureBackend <- function(...) {
 #'
 #' f <- future({
 #'   data.frame(
-#'     hostname = Sys.info()[["nodename"]],
-#'           os = Sys.info()[["sysname"]],
-#'        cores = unname(parallelly::availableCores()),
-#'      modules = Sys.getenv("LOADEDMODULES")
+#'      hostname = Sys.info()[["nodename"]],
+#'            os = Sys.info()[["sysname"]],
+#'     osVersion = utils::osVersion,
+#'         cores = unname(parallelly::availableCores()),
+#'       modules = Sys.getenv("LOADEDMODULES")
 #'   )
 #' })
 #' info <- value(f)

R/batchtools_sge.R

@@ -65,10 +65,11 @@ print.BatchtoolsSGEFutureBackend <- function(x, ...) {
 #'
 #' f <- future({
 #'   data.frame(
-#'     hostname = Sys.info()[["nodename"]],
-#'           os = Sys.info()[["sysname"]],
-#'        cores = unname(parallelly::availableCores()),
-#'      modules = Sys.getenv("LOADEDMODULES")
+#'      hostname = Sys.info()[["nodename"]],
+#'            os = Sys.info()[["sysname"]],
+#'     osVersion = utils::osVersion,
+#'         cores = unname(parallelly::availableCores()),
+#'       modules = Sys.getenv("LOADEDMODULES")
 #'   )
 #' })
 #' info <- value(f)