Wait for jobs rework (#140)

Author: mllg

R/clusterFunctions.R

@@ -153,7 +153,7 @@ cfReadBrewTemplate = function(template, comment.string = NA_character_) {
     "!DEBUG [cfReadBrewTemplate]: Parsing template from string"
     lines = stri_trim_both(stri_split_lines(template)[[1L]])
   } else {
-    "!DEBUG [cfReadBrewTemplate]: Parsing template form file '`template`'"
+    "!DEBUG [cfReadBrewTemplate]: Parsing template file '`template`'"
     lines = stri_trim_both(readLines(template))
   }
 

R/filenames.R

@@ -30,7 +30,7 @@ getLogFiles = function(reg, ids) {
 }
 
 getJobFiles = function(reg, hash) {
-  fp(dir(reg, "jobs"), sprintf("%s.rds", hash))
+  fp(reg$file.dir, "jobs", sprintf("%s.rds", hash))
 }
 
 getExternalDirs = function(reg, ids) {

R/findJobs.R

@@ -220,6 +220,13 @@ findErrors = function(ids = NULL, reg = getDefaultRegistry()) {
 }
 
 
+# used in waitForJobs: find jobs which are done or error
+.findTerminated = function(reg, ids = NULL) {
+  done = NULL
+  filter(reg$status, ids, c("job.id", "done"))[!is.na(done), "job.id"]
+}
+
+
 #' @export
 #' @rdname findJobs
 findOnSystem = function(ids = NULL, reg = getDefaultRegistry()) {

R/runOSCommand.R

@@ -29,7 +29,7 @@ runOSCommand = function(sys.cmd, sys.args = character(0L), stdin = "", nodename
   assertString(nodename, min.chars = 1L)
 
   if (!isLocalHost(nodename)) {
-    sys.args = c("-q", nodename, shQuote(stri_flatten(c(sys.cmd, sys.args), " ")))
+    sys.args = c("-q", nodename, sys.cmd, sys.args)
     sys.cmd = "ssh"
   }
 

R/waitForJobs.R

@@ -17,6 +17,14 @@
 #'   \code{FALSE}. This argument may be required on some systems where, e.g.,
 #'   expired jobs or jobs on hold are problematic to detect. If you don't want
 #'   a timeout, set this to \code{Inf}. Default is \code{604800} (one week).
+#' @param expire.after [\code{integer(1)}]\cr
+#'   Jobs count as \dQuote{expired} if they are not found on the system but have not communicated back
+#'   their results (or error message). This frequently happens on managed system if the scheduler kills
+#'   a job because the job has hit the walltime or request more memory than reserved.
+#'   On the other hand, network file systems often require several seconds for new files to be found,
+#'   which can lead to false positives in the detection heuristic.
+#'   \code{waitForJobs} treats such jobs as expired after they have not been detected on the system
+#'   for \code{expire.after} iterations (default 3 iterations).
 #' @param stop.on.error [\code{logical(1)}]\cr
 #'   Immediately cancel if a job terminates with an error? Default is
 #'   \code{FALSE}.
@@ -25,84 +33,75 @@
 #'   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 = NULL, timeout = 604800, stop.on.error = FALSE, reg = getDefaultRegistry()) {
+waitForJobs = function(ids = NULL, sleep = NULL, timeout = 604800, expire.after = 3L, stop.on.error = FALSE, reg = getDefaultRegistry()) {
   assertRegistry(reg, writeable = FALSE, sync = TRUE)
   assertNumber(timeout, lower = 0)
+  assertCount(expire.after, positive = TRUE)
   assertFlag(stop.on.error)
   sleep = getSleepFunction(reg, sleep)
   ids = convertIds(reg, ids, default = .findSubmitted(reg = reg))
 
-  .findNotTerminated = function(reg, ids = NULL) {
-    done = NULL
-    filter(reg$status, ids, c("job.id", "done"))[is.na(done), "job.id"]
-  }
-
   if (nrow(.findNotSubmitted(ids = ids, reg = reg)) > 0L) {
     warning("Cannot wait for unsubmitted jobs. Removing from ids.")
     ids = ids[.findSubmitted(ids = ids, reg = reg), nomatch = 0L]
   }
 
-  n.jobs = nrow(ids)
-  if (n.jobs == 0L)
+  if (nrow(ids) == 0L) {
     return(TRUE)
+  }
 
-  batch.ids = getBatchIds(reg)
-  "!DEBUG [waitForJobs]: Using `nrow(ids)` ids and `nrow(batch.ids)` initial batch ids"
+  terminated = on.sys = expire.counter = NULL
+  ids$terminated = FALSE
+  ids$on.sys = FALSE
+  ids$expire.counter = 0L
 
   timeout = Sys.time() + timeout
-  ids.disappeared = noIds()
-  pb = makeProgressBar(total = n.jobs, format = "Waiting (S::system R::running D::done E::error) [:bar] :percent eta: :eta")
-  i = 1L
+  pb = makeProgressBar(total = nrow(ids), format = "Waiting (S::system R::running D::done E::error) [:bar] :percent eta: :eta")
+  i = 0L
 
   repeat {
-    # case 1: all jobs terminated -> nothing on system
-    ids.nt = .findNotTerminated(reg, ids)
-    if (nrow(ids.nt) == 0L) {
+    ### case 1: all jobs terminated -> nothing on system
+    ids[.findTerminated(reg, ids), "terminated" := TRUE]
+    if (ids[!(terminated), .N] == 0L) {
       "!DEBUG [waitForJobs]: All jobs terminated"
       pb$update(1)
       waitForResults(reg, ids)
       return(nrow(.findErrors(reg, ids)) == 0L)
     }
 
-    stats = getStatusTable(ids = ids, batch.ids = batch.ids, reg = reg)
-    pb$update((n.jobs - nrow(ids.nt)) / n.jobs, tokens = as.list(stats))
-
-    # case 2: there are errors and stop.on.error is TRUE
+    ### case 2: there are errors and stop.on.error is TRUE
     if (stop.on.error && nrow(.findErrors(reg, ids)) > 0L) {
       "!DEBUG [waitForJobs]: Errors found and stop.on.error is TRUE"
       pb$update(1)
       return(FALSE)
     }
 
-    # case 3: we have reached a timeout
-    if (Sys.time() > timeout) {
+    batch.ids = getBatchIds(reg)
+    ids[, "on.sys" := FALSE][.findOnSystem(reg, ids, batch.ids = batch.ids), "on.sys" := TRUE]
+    ids[!(on.sys) & !(terminated), "expire.counter" := expire.counter + 1L]
+    stats = getStatusTable(ids = ids, batch.ids = batch.ids, reg = reg)
+    pb$update(mean(ids$terminated), tokens = as.list(stats))
+    "!DEBUG [waitForJobs]: batch.ids: `stri_flatten(batch.ids$batch.id, ',')`"
+
+    ### case 3: jobs disappeared, we cannot find them on the system in [expire.after] iterations
+    if (ids[!(terminated) & expire.counter > expire.after, .N] > 0L) {
+      warning("Some jobs disappeared from the system")
       pb$update(1)
-      warning("Timeout reached")
+      waitForResults(reg, ids)
       return(FALSE)
     }
 
-    # case 4: jobs disappeared, we cannot find them on the system
-    # heuristic:
-    #   job is not terminated, not on system and has not been on the system
-    #   in the previous iteration
-    ids.on.sys = .findOnSystem(reg, ids, batch.ids = batch.ids)
-    if (nrow(ids.disappeared) > 0L) {
-      if (nrow(ids.nt[!ids.on.sys, on = "job.id"][ids.disappeared, on = "job.id", nomatch = 0L]) > 0L) {
-        warning("Some jobs disappeared from the system")
-        pb$update(1)
-        waitForResults(reg, ids)
-        return(FALSE)
-      }
-    }
-
-    ids.disappeared = ids[!ids.on.sys, on = "job.id"]
-    "!DEBUG [waitForJobs]: `nrow(ids.disappeared)` jobs disappeared"
-
+    # case 4: we reach a timeout
     sleep(i)
     i = i + 1L
-    suppressMessages(sync(reg = reg))
-    saveRegistry(reg)
-    batch.ids = getBatchIds(reg)
-    "!DEBUG [waitForJobs]: New batch.ids: `stri_flatten(batch.ids$batch.id, ',')`"
+    if (Sys.time() > timeout) {
+      pb$update(1)
+      warning("Timeout reached")
+      return(FALSE)
+    }
+
+    if (suppressMessages(sync(reg = reg)))
+      saveRegistry(reg)
   }
 }
+

man/waitForJobs.Rd

@@ -9,7 +9,7 @@ test_that("cf multicore", {
   silent({
     submitJobs(1:2, reg = reg)
     expect_equal(findOnSystem(reg = reg), findJobs(reg = reg))
-    expect_true(waitForJobs(sleep = 0.2, reg = reg))
+    expect_true(waitForJobs(sleep = 0.2, expire.after = 1, reg = reg))
   })
   expect_data_table(findOnSystem(reg = reg), nrow = 0)
   expect_equal(findDone(reg = reg), findJobs(reg = reg))

tests/testthat/test_ClusterFunctionsMulticore.R

@@ -23,6 +23,20 @@ test_that("makeJobCollection", {
   expect_output(print(jc), "Collection")
 })
 
+
+test_that("makeJobCollection does not expand relative paths", {
+  skip_on_os("windows")
+  reg = makeRegistry(file.dir = NA, make.default = FALSE)
+  batchMap(identity, 1, reg = reg)
+  reg$file.dir = npath("~/foo", must.work = FALSE)
+  reg$work.dir = npath("~/bar", must.work = FALSE)
+  jc = makeJobCollection(1, reg = reg)
+  expect_true(stri_startswith_fixed(jc$file.dir, "~/foo"))
+  expect_true(stri_startswith_fixed(jc$uri, "~/foo/jobs/"))
+  expect_true(stri_startswith_fixed(jc$log.file, "~/foo/logs"))
+  expect_true(stri_startswith_fixed(jc$work.dir, "~/bar"))
+})
+
 test_that("makeJobCollection.ExperimentCollection", {
   reg = makeExperimentRegistry(file.dir = NA, make.default = FALSE)
   addProblem(reg = reg, "p1", fun = function(job, data, ...) list(data = data, ...))

tests/testthat/test_JobCollection.R

@@ -7,7 +7,7 @@ test_that("waitForJobs", {
   silent({
     submitJobs(ids, reg = reg)
     expect_true(waitForJobs(ids = ids[1], reg = reg, sleep = 1))
-    expect_false(waitForJobs(ids = ids, stop.on.error = TRUE, sleep = 1, reg = reg))
+    expect_false(waitForJobs(ids = ids, stop.on.error = TRUE, sleep = 1, expire.after = 1, reg = reg))
   })
 })