improved control over sleep duration; nodename in cfSlurm

Author: mllg

.gitignore

@@ -1,6 +1,8 @@
 .DS_Store
-inst/doc
-src/*.so
-src/*.o
-vignettes/*.html
-vignettes/*.pdf
+/inst/doc
+/src/*.so
+/src/*.o
+/vignettes/*.html
+/vignettes/*.pdf
+/.Rproj.user
+/batchtools.Rproj

NEWS.md

@@ -2,6 +2,7 @@
 
 * Added a missing routine to upgrade registries created with batchtools prior to v0.9.3.
 * Fixed a bug where the registry could not be synced if jobs failed during initialization (#135).
+* The sleep duration used in `waitForJobs()` and `submitJobs()` can now be set via the configuration file.
 * A new heuristic will try to detect if the registry has been altered by a simultaneously running R session.
   If this is detected, the registry will be set to a read-only state.
 * New argument `writeable` for `loadRegistry()` to allow loading registries explicitly as read-only.

R/chunkIds.R

@@ -23,7 +23,7 @@ chunkIds = function(ids = NULL, n.chunks = NULL, chunk.size = NULL, group.by = c
 
   if (length(group.by) > 0L) {
     job.id = NULL
-    if (any(group.by %nin% names(ids)))
+    if (any(group.by %chnin% names(ids)))
       stop("All columns to group by must be provided in the 'ids' table")
     ids[, "chunk" := chunk(job.id, n.chunks = n.chunks, chunk.size = chunk.size), by = group.by]
     ids[, "chunk" := .GRP, by = c(group.by, "chunk")]

R/clusterFunctions.R

@@ -31,9 +31,13 @@
 #' @param array.var [\code{character(1)}]\cr
 #'   Name of the environment variable set by the scheduler to identify IDs of job arrays.
 #'   Default is \code{NA} for no array support.
-#' @param store.job [\code{logical(1)}]\cr
+#' @param store.job.collection [\code{logical(1)}]\cr
 #'   Flag to indicate that the cluster function implementation of \code{submitJob} can not directly handle \code{\link{JobCollection}} objects.
 #'   If set to \code{FALSE}, the \code{\link{JobCollection}} is serialized to the file system before submitting the job.
+#' @param store.job.files [\code{logical(1)}]\cr
+#'   Flag to indicate that job files need to be stored in the file directory.
+#'   If set to \code{FALSE} (default), the job file is created in a temporary directory, otherwise (or if the debug mode is enabled) in
+#'   the subdirectory \code{jobs} of the \code{file.dir}.
 #' @param scheduler.latency [\code{numeric(1)}]\cr
 #'   Time to sleep after important interactions with the scheduler to ensure a sane state.
 #'   Currently only triggered after calling \code{\link{submitJobs}}.
@@ -50,7 +54,8 @@
 #' @family ClusterFunctions
 #' @family ClusterFunctionsHelper
 makeClusterFunctions = function(name, submitJob, killJob = NULL, listJobsQueued = NULL, listJobsRunning = NULL,
-  array.var = NA_character_, store.job = FALSE, scheduler.latency = 0, fs.latency = NA_real_, hooks = list()) {
+  array.var = NA_character_, store.job.collection = FALSE, store.job.files = FALSE, scheduler.latency = 0,
+  fs.latency = NA_real_, hooks = list()) {
   assertList(hooks, types = "function", names = "unique")
   assertSubset(names(hooks), unlist(batchtools$hooks, use.names = FALSE))
 
@@ -61,7 +66,8 @@ makeClusterFunctions = function(name, submitJob, killJob = NULL, listJobsQueued
       listJobsQueued = assertFunction(listJobsQueued, "reg", null.ok = TRUE),
       listJobsRunning = assertFunction(listJobsRunning, "reg", null.ok = TRUE),
       array.var = assertString(array.var, na.ok = TRUE),
-      store.job = assertFlag(store.job),
+      store.job.collection = assertFlag(store.job.collection),
+      store.job.files = assertFlag(store.job.files),
       scheduler.latency = assertNumber(scheduler.latency, lower = 0),
       fs.latency = assertNumber(fs.latency, lower = 0, na.ok = TRUE),
       hooks = hooks),
@@ -181,7 +187,11 @@ cfReadBrewTemplate = function(template, comment.string = NA_character_) {
 cfBrewTemplate = function(reg, text, jc) {
   assertString(text)
 
-  outfile = if (batchtools$debug) fp(reg$file.dir, "jobs", sprintf("%s.job", jc$job.hash)) else tempfile("job")
+  outfile = if (batchtools$debug || reg$cluster.functions$store.job.files) {
+    fp(reg$file.dir, "jobs", sprintf("%s.job", jc$job.hash))
+  } else {
+    tempfile(fileext = "job")
+  }
   parent.env(jc) = asNamespace("batchtools")
   on.exit(parent.env(jc) <- emptyenv())
   "!DEBUG [cfBrewTemplate]: Brewing template to file '`outfile`'"
@@ -235,16 +245,18 @@ cfHandleUnknownSubmitError = function(cmd, exit.code, output) {
 #' @param max.tries [\code{integer(1)}]\cr
 #'   Number of total times to try execute the OS command in cases of failures.
 #'   Default is \code{3}.
+#' @inheritParams runOSCommand
 #' @return \code{TRUE} on success. An exception is raised otherwise.
 #' @family ClusterFunctionsHelper
 #' @export
-cfKillJob = function(reg, cmd, args = character(0L), max.tries = 3L) {
+cfKillJob = function(reg, cmd, args = character(0L), max.tries = 3L, nodename = "localhost") {
   assertString(cmd, min.chars = 1L)
   assertCharacter(args, any.missing = FALSE)
+  assertString(nodename)
   max.tries = asCount(max.tries)
 
   for (i in seq_len(max.tries)) {
-    res = runOSCommand(cmd, args)
+    res = runOSCommand(cmd, args, nodename = nodename)
     if (res$exit.code == 0L)
       return(TRUE)
     Sys.sleep(1)

R/clusterFunctionsDocker.R

@@ -108,6 +108,6 @@ makeClusterFunctionsDocker = function(image, docker.args = character(0L), image.
   }
 
   makeClusterFunctions(name = "Docker", submitJob = submitJob, killJob = killJob, listJobsRunning = listJobsRunning,
-    store.job = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency,
+    store.job.collection = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency,
     hooks = list(post.submit = housekeeping, post.sync = housekeeping))
 } # nocov end

R/clusterFunctionsInteractive.R

@@ -36,5 +36,5 @@ makeClusterFunctionsInteractive = function(external = FALSE, write.logs = TRUE,
     }
   }
 
-  makeClusterFunctions(name = "Interactive", submitJob = submitJob, store.job = external, fs.latency = fs.latency)
+  makeClusterFunctions(name = "Interactive", submitJob = submitJob, store.job.collection = external, fs.latency = fs.latency)
 }

R/clusterFunctionsLSF.R

@@ -73,5 +73,5 @@ makeClusterFunctionsLSF = function(template = "lsf", scheduler.latency = 1, fs.l
   }
 
   makeClusterFunctions(name = "LSF", submitJob = submitJob, killJob = killJob, listJobsQueued = listJobsQueued,
-    listJobsRunning = listJobsRunning, store.job = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency)
+    listJobsRunning = listJobsRunning, store.job.collection = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency)
 } # nocov end

R/clusterFunctionsMulticore.R

@@ -99,5 +99,5 @@ makeClusterFunctionsMulticore = function(ncpus = NA_integer_, fs.latency = NA_re
   }
 
   makeClusterFunctions(name = "Multicore", submitJob = submitJob, listJobsRunning = listJobsRunning,
-    store.job = FALSE, fs.latency = fs.latency, hooks = list(pre.sync = function(reg, fns) p$collect(1)))
+    store.job.collection = FALSE, fs.latency = fs.latency, hooks = list(pre.sync = function(reg, fns) p$collect(1)))
 }

R/clusterFunctionsOpenLava.R

@@ -73,5 +73,5 @@ makeClusterFunctionsOpenLava = function(template = "openlava", scheduler.latency
   }
 
   makeClusterFunctions(name = "OpenLava", submitJob = submitJob, killJob = killJob, listJobsQueued = listJobsQueued,
-    listJobsRunning = listJobsRunning, store.job = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency)
+    listJobsRunning = listJobsRunning, store.job.collection = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency)
 } # nocov end

R/clusterFunctionsSGE.R

@@ -67,5 +67,5 @@ makeClusterFunctionsSGE = function(template = "sge", scheduler.latency = 1, fs.l
   }
 
   makeClusterFunctions(name = "SGE", submitJob = submitJob, killJob = killJob, listJobsQueued = listJobsQueued,
-    listJobsRunning = listJobsRunning, store.job = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency)
+    listJobsRunning = listJobsRunning, store.job.collection = TRUE, scheduler.latency = scheduler.latency, fs.latency = fs.latency)
 } # nocov end

R/clusterFunctionsSSH.R

@@ -62,5 +62,5 @@ makeClusterFunctionsSSH = function(workers, fs.latency = 65) { # nocov start
   }
 
   makeClusterFunctions(name = "SSH", submitJob = submitJob, killJob = killJob, listJobsRunning = listJobsRunning,
-    store.job = TRUE, fs.latency = fs.latency)
+    store.job.collection = TRUE, fs.latency = fs.latency)
 } # nocov end

R/clusterFunctionsSlurm.R

@@ -26,14 +26,23 @@
 #'  Note that you should not select the cluster in your template file via \code{#SBATCH --clusters}.
 #' @param array.jobs [\code{logical(1)}]\cr
 #'  If array jobs are disabled on the computing site, set to \code{FALSE}.
+#' @param nodename [\code{character(1)}]\cr
+#'  Nodename of the master. All commands are send via SSH to this host. Only works iff
+#'  \enumerate{
+#'    \item{Passwordless authentication (e.g., via SSH public key authentication) is set up.}
+#'    \item{The file directory is shared across machines, e.g. mounted via SSHFS.}
+#'    \item{The absolute path to the \code{file.dir} are identical on the machines, or paths are provided relative to the
+#'      home directory.}
+#'  }
 #' @inheritParams makeClusterFunctions
 #' @return [\code{\link{ClusterFunctions}}].
 #' @family ClusterFunctions
 #' @export
-makeClusterFunctionsSlurm = function(template = "slurm", clusters = NULL, array.jobs = TRUE, scheduler.latency = 1, fs.latency = 65) { # nocov start
+makeClusterFunctionsSlurm = function(template = "slurm", clusters = NULL, array.jobs = TRUE, nodename = "localhost", scheduler.latency = 1, fs.latency = 65) { # nocov start
   if (!is.null(clusters))
     assertString(clusters, min.chars = 1L)
   assertFlag(array.jobs)
+  assertString(nodename)
   template = findTemplateFile(template)
   template = cfReadBrewTemplate(template, "##")
 
@@ -47,7 +56,7 @@ makeClusterFunctionsSlurm = function(template = "slurm", clusters = NULL, array.
       jc$log.file = stri_join(jc$log.file, "_%a")
     }
     outfile = cfBrewTemplate(reg, template, jc)
-    res = runOSCommand("sbatch", shQuote(outfile))
+    res = runOSCommand("sbatch", shQuote(outfile), nodename = nodename)
 
     max.jobs.msg = "sbatch: error: Batch job submission failed: Job violates accounting policy (job submit limit, user's size and/or time limits)"
     temp.error = "Socket timed out on send/recv operation"
@@ -76,7 +85,7 @@ makeClusterFunctionsSlurm = function(template = "slurm", clusters = NULL, array.
     assertRegistry(reg, writeable = FALSE)
     if (array.jobs)
       args = c(args, "-r")
-    res = runOSCommand("squeue", args)
+    res = runOSCommand("squeue", args, nodename = nodename)
     if (res$exit.code > 0L)
       OSError("Listing of jobs failed", res)
     if (!is.null(clusters)) tail(res$output, -1L) else res$output
@@ -95,10 +104,10 @@ makeClusterFunctionsSlurm = function(template = "slurm", clusters = NULL, array.
   killJob = function(reg, batch.id) {
     assertRegistry(reg, writeable = TRUE)
     assertString(batch.id)
-    cfKillJob(reg, "scancel", c(sprintf("--clusters=%s", clusters), batch.id))
+    cfKillJob(reg, "scancel", c(sprintf("--clusters=%s", clusters), batch.id), nodename = nodename)
   }
 
   makeClusterFunctions(name = "Slurm", submitJob = submitJob, killJob = killJob, listJobsRunning = listJobsRunning,
-    listJobsQueued = listJobsQueued, array.var = "SLURM_ARRAY_TASK_ID", store.job = TRUE,
-    scheduler.latency = scheduler.latency, fs.latency = fs.latency)
+    listJobsQueued = listJobsQueued, array.var = "SLURM_ARRAY_TASK_ID", store.job.collection = TRUE,
+    store.job.files = !isLocalHost(nodename), scheduler.latency = scheduler.latency, fs.latency = fs.latency)
 } # nocov end

R/clusterFunctionsSocket.R

@@ -71,5 +71,5 @@ makeClusterFunctionsSocket = function(ncpus = NA_integer_, fs.latency = 65) {
   }
 
   makeClusterFunctions(name = "Socket", submitJob = submitJob, listJobsRunning = listJobsRunning,
-    store.job = FALSE, fs.latency = fs.latency, hooks = list(pre.sync = function(reg, fns) p$list()))
+    store.job.collection = FALSE, fs.latency = fs.latency, hooks = list(pre.sync = function(reg, fns) p$list()))
 }

R/clusterFunctionsTORQUE.R

@@ -73,6 +73,6 @@ makeClusterFunctionsTORQUE = function(template = "torque", scheduler.latency = 1
   }
 
   makeClusterFunctions(name = "TORQUE", submitJob = submitJob, killJob = killJob, listJobsQueued = listJobsQueued,
-    listJobsRunning = listJobsRunning, array.var = "PBS_ARRAYID", store.job = TRUE,
+    listJobsRunning = listJobsRunning, array.var = "PBS_ARRAYID", store.job.collection = TRUE,
     scheduler.latency = scheduler.latency, fs.latency = fs.latency)
 } # nocov end

R/findJobs.R

@@ -258,7 +258,7 @@ findExpired = function(ids = NULL, reg = getDefaultRegistry()) {
 
 .findExpired = function(reg, ids = NULL, batch.ids = getBatchIds(reg)) {
   submitted = done = batch.id = NULL
-  filter(reg$status, ids, c("job.id", "submitted", "done", "batch.id"))[!is.na(submitted) & is.na(done) & batch.id %nin% batch.ids$batch.id, "job.id"]
+  filter(reg$status, ids, c("job.id", "submitted", "done", "batch.id"))[!is.na(submitted) & is.na(done) & batch.id %chnin% batch.ids$batch.id, "job.id"]
 }
 
 #' @export

R/helpers.R

@@ -98,13 +98,17 @@ stopf = function (...) {
   !match(x, y, nomatch = 0L)
 }
 
+`%chnin%` = function(x, y) {
+  !chmatch(x, y, nomatch = 0L)
+}
+
 setClasses = function(x, cl) {
   setattr(x, "class", cl)
   x
 }
 
 addlevel = function(x, lvl) {
-  if (lvl %nin% levels(x))
+  if (lvl %chnin% levels(x))
     levels(x) = c(levels(x), lvl)
   x
 }

R/runOSCommand.R

@@ -28,8 +28,8 @@ runOSCommand = function(sys.cmd, sys.args = character(0L), stdin = "", nodename
   assertCharacter(sys.args, any.missing = FALSE)
   assertString(nodename, min.chars = 1L)
 
-  if (nodename != "localhost") {
-    sys.args = c(nodename, shQuote(stri_flatten(c(sys.cmd, sys.args), " ")))
+  if (!isLocalHost(nodename)) {
+    sys.args = c("-q", nodename, shQuote(stri_flatten(c(sys.cmd, sys.args), " ")))
     sys.cmd = "ssh"
   }
 
@@ -50,7 +50,11 @@ runOSCommand = function(sys.cmd, sys.args = character(0L), stdin = "", nodename
   return(list(sys.cmd = sys.cmd, sys.args = sys.args, exit.code = exit.code, output = output))
 }
 
+isLocalHost = function(nodename) {
+  is.null(nodename) || nodename %chin% c("localhost", "127.0.0.1", "::1")
+}
+
 OSError = function(msg, res) {
-  stopf("%s (exit code %i); cmd: '%s'; output: %s",
-    msg, res$exit.code, stri_flatten(c(res$sys.cmd, res$sys.args), collapse = " "), res$output)
+  stopf("%s (exit code %i);\ncmd: '%s'\noutput:\n%s",
+    msg, res$exit.code, stri_flatten(c(res$sys.cmd, res$sys.args), collapse = " "), stri_flatten(res$output, "\n"))
 }

R/sleep.R

@@ -1,11 +1,18 @@
-default.sleep = function(i) {
-   5 + 115 * pexp(i - 1, rate = 0.01)
-}
+getSleepFunction = function(reg, sleep) {
+  if (is.null(sleep)) {
+    if (is.null(reg$sleep))
+      return(function(i) { 5 + 115 * pexp(i - 1, rate = 0.01) })
+    sleep = reg$sleep
+  }
+
+  if (is.numeric(sleep)) {
+    assertNumber(sleep, lower = 0)
+    return(function(i) Sys.sleep(sleep))
+  }
+
+  if (is.function(sleep)) {
+    return(function(i) Sys.sleep(sleep(i)))
+  }
 
-getSleepFunction = function(sleep) {
-  assert(checkNumber(sleep, lower = 0), checkFunction(sleep, args = "i"))
-  if (is.numeric(sleep))
-    function(i) Sys.sleep(sleep)
-  else
-    function(i) Sys.sleep(sleep(i))
+  stop("Argument 'sleep' must be either a numeric value or function(i)")
 }

R/submitJobs.R

@@ -54,8 +54,11 @@
 #'   Defaults can be stored in the configuration file by providing the named list \code{default.resources}.
 #'   Settings in \code{resources} overwrite those in \code{default.resources}.
 #' @param sleep [\code{function(i)} | \code{numeric(1)}]\cr
-#'   Function which returns the duration to sleep in the \code{i}-th iteration between temporary errors.
-#'   Alternatively, you can pass a single positive numeric value.
+#'   Parameter to control the duration to sleep between temporary errors.
+#'   You can pass an absolute numeric value in seconds or a \code{function(i)} which returns the number of seconds to sleep in the \code{i}-th
+#'   iteration between temporary errors.
+#'   If not provided (\code{NULL}), tries to read the value (number/function) from the configuration file (stored in \code{reg$sleep}) or defaults to
+#'   a function with exponential backoff between 5 and 120 seconds.
 #' @template reg
 #' @return [\code{\link{data.table}}] with columns \dQuote{job.id} and \dQuote{chunk}.
 #' @export
@@ -114,7 +117,7 @@
 #' # There should also be a note in the log:
 #' grepLogs(pattern = "parallelMap", reg = tmp)
 #' }
-submitJobs = function(ids = NULL, resources = list(), sleep = default.sleep, reg = getDefaultRegistry()) {
+submitJobs = function(ids = NULL, resources = list(), sleep = NULL, reg = getDefaultRegistry()) {
   assertRegistry(reg, writeable = TRUE, sync = TRUE)
   assertList(resources, names = "strict")
   resources = insert(reg$default.resources, resources)
@@ -133,7 +136,7 @@ submitJobs = function(ids = NULL, resources = list(), sleep = default.sleep, reg
       resources$chunks.as.arrayjobs = NULL
     }
   }
-  sleep = getSleepFunction(sleep)
+  sleep = getSleepFunction(reg, sleep)
 
   ids = convertIds(reg, ids, default = .findNotSubmitted(reg = reg), keep.extra = "chunk")
   if (nrow(ids) == 0L)
@@ -185,7 +188,7 @@ submitJobs = function(ids = NULL, resources = list(), sleep = default.sleep, reg
   for (ch in chunks) {
     ids.chunk = ids[chunk == ch, "job.id"]
     jc = makeJobCollection(ids.chunk, resources = resources, reg = reg)
-    if (reg$cluster.functions$store.job)
+    if (reg$cluster.functions$store.job.collection)
       writeRDS(jc, file = jc$uri)
 
     if (!is.na(max.concurrent.jobs)) {

R/updateRegisty.R

@@ -17,7 +17,7 @@ updateRegistry = function(reg = getDefaultRegistry()) { # nocov start
     }
 
     ### hotfix for log.file column
-    if ("log.file" %nin% names(reg$status)) {
+    if ("log.file" %chnin% names(reg$status)) {
       info("Adding column 'log.file'")
       reg$status[, ("log.file") := rep(NA_character_, .N)]
     }
@@ -44,7 +44,7 @@ updateRegistry = function(reg = getDefaultRegistry()) { # nocov start
   }
 
   if (reg$version < "0.9.4-9001") {
-    if ("job.name" %nin% names(reg$status)) {
+    if ("job.name" %chnin% names(reg$status)) {
       info("Adding column 'job.name'")
       reg$status[, ("job.name") := rep(NA_character_, .N)]
     }

R/waitForJobs.R

@@ -6,8 +6,12 @@
 #' @templateVar ids.default findSubmitted
 #' @template ids
 #' @param sleep [\code{function(i)} | \code{numeric(1)}]\cr
-#'   Function which returns the duration to sleep in the \code{i}-th iteration.
-#'   Alternatively, you can pass a single positive numeric value.
+#'   Parameter to control the duration to sleep between queries.
+#'   You can pass an absolute numeric value in seconds or a \code{function(i)} which returns
+#'   the number of seconds to sleep in the \code{i}-th iteration.
+#'   If not provided (\code{NULL}), tries to read the value (number/function) from the configuration file
+#'   (stored in \code{reg$sleep}) or defaults to a function with exponential backoff between
+#'   5 and 120 seconds.
 #' @param timeout [\code{numeric(1)}]\cr
 #'   After waiting \code{timeout} seconds, show a message and return
 #'   \code{FALSE}. This argument may be required on some systems where, e.g.,
@@ -21,11 +25,11 @@
 #'   successfully and \code{FALSE} if either the timeout is reached or at least
 #'   one job terminated with an exception.
 #' @export
-waitForJobs = function(ids = NULL, sleep = default.sleep, timeout = 604800, stop.on.error = FALSE, reg = getDefaultRegistry()) {
+waitForJobs = function(ids = NULL, sleep = NULL, timeout = 604800, stop.on.error = FALSE, reg = getDefaultRegistry()) {
   assertRegistry(reg, writeable = FALSE, sync = TRUE)
   assertNumber(timeout, lower = 0)
   assertFlag(stop.on.error)
-  sleep = getSleepFunction(sleep)
+  sleep = getSleepFunction(reg, sleep)
   ids = convertIds(reg, ids, default = .findSubmitted(reg = reg))
 
   .findNotTerminated = function(reg, ids = NULL) {