Record submission time, which is used when querying Slurm's 'sacct'

Author: HenrikBengtsson

DESCRIPTION

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

R/BatchtoolsFutureBackend-class.R

@@ -325,7 +325,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 +344,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
@@ -540,8 +542,14 @@ status <- function(future, ...) {
   batch_id <- reg[["status"]][["batch.id"]]
   ## 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)
-  on.exit(options(future.batchtools.batch_id = NULL), add = TRUE)
+  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)

R/batchtools_slurm.R

@@ -213,12 +213,13 @@ patchClusterFunctionsSlurm2 <- function(cf) {
   getClusters <- env[["getClusters"]]
   nodename <- env[["nodename"]]
   org_listJobsQueued <- env[["listJobsQueued"]]
-  
-  isJobQueued <- function(reg, batch_id) {
+
+  isJobQueued <- function(reg, batch_id, since = NULL) {
     stopifnot(length(batch_id) == 1L, !is.na(batch_id), nzchar(batch_id))
+    stopifnot(is.null(since) || inherits(since, "POSIXct"))
 
     ## FIXME: Add also --starttime=<start time>, because 'sacct' only returns jobs ran today
-    args <- c("--user=$USER", "--noheader", "--parsable2", "--allocations", "--format=State", sprintf("--jobs=%s", batch_id))
+    args <- c("--user=$USER", "--noheader", "--parsable2", "--allocations", "--format=State", sprintf("--jobs=%s", batch_id), sprintf("--start-time=%s", format(since - 15*60, format = "%FT%T"))) ## Allow for 15-min time offsets
     clusters <- getClusters(reg)
     if (length(clusters) > 0) {
        args <- c(args, sprintf("--clusters=%s", clusters))
@@ -252,7 +253,10 @@ patchClusterFunctionsSlurm2 <- function(cf) {
     }
 
     ## Ask 'sacct' it if is PENDING or REQUEUED
-    if (isJobQueued(reg, batch_id)) jobs <- as.character(batch_id)
+    submitted_on <- getOption("future.batchtools.submitted_on", NULL)
+    if (isJobQueued(reg, batch_id, since = submitted_on)) {
+      jobs <- as.character(batch_id)
+    }
 
     jobs
   }