1#' Get Set of Available Workers
2#'
3#' @param methods A character vector specifying how to infer the number
4#' of available cores.
5#'
6#' @param na.rm If TRUE, only non-missing settings are considered/returned.
7#'
8#' @param logical Passed as-is to [availableCores()].
9#'
10#' @param default The default set of workers.
11#'
12#' @param which A character specifying which set / sets to return.
13#' If `"auto"` (default), the first non-empty set found.
14#' If `"min"`, the minimum value is returned.
15#' If `"max"`, the maximum value is returned (be careful!)
16#' If `"all"`, all values are returned.
17#'
18#' @return Return a character vector of workers, which typically consists
19#' of names of machines / compute nodes, but may also be IP numbers.
20#'
21#' @details
22#' The default set of workers for each method is
23#' `rep("localhost", times = availableCores(methods = method, logical = logical))`,
24#' which means that each will at least use as many parallel workers on the
25#' current machine that [availableCores()] allows for that method.
26#'
27#' In addition, the following settings ("methods") are also acknowledged:
28#' \itemize{
29#'  \item `"PBS"` -
30#'    Query TORQUE/PBS environment variable \env{PBS_NODEFILE}.
31#'    If this is set and specifies an existing file, then the set
32#'    of workers is read from that file, where one worker (node)
33#'    is given per line.
34#'    An example of a job submission that results in this is
35#'    `qsub -l nodes = 4:ppn = 2`, which requests four nodes each
36#'    with two cores.
37#'
38#'  \item `"SGE"` -
39#'    Query Sun/Oracle Grid Engine (SGE) environment variable
40#'    \env{PE_HOSTFILE}.
41#'    An example of a job submission that results in this is
42#'    `qsub -pe mpi 8` (or `qsub -pe ompi 8`), which
43#'    requests eight cores on a any number of machines.
44#'
45#'  \item `"LSF"` -
46#'    Query LSF/OpenLava environment variable \env{LSB_HOSTS}.
47#'
48#'  \item `"Slurm"` -
49#'    Query Slurm environment variable \env{SLURM_JOB_NODELIST} (fallback
50#'    to legacy \env{SLURM_NODELIST}) and parse set of nodes.
51#'    Then query Slurm environment variable \env{SLURM_JOB_CPUS_PER_NODE}
52#'    (fallback \env{SLURM_TASKS_PER_NODE}) to infer how many CPU cores
53#'    Slurm have alloted to each of the nodes.  If \env{SLURM_CPUS_PER_TASK}
54#'    is set, which is always a scalar, then that is respected too, i.e.
55#'    if it is smaller, then that is used for all nodes.
56#'    For example, if `SLURM_NODELIST="n1,n[03-05]"` (expands to
57#'    `c("n1", "n03", "n04", "n05")`) and `SLURM_JOB_CPUS_PER_NODE="2(x2),3,2"`
58#'    (expands to `c(2, 2, 3, 2, 2)`), then
59#'    `c("n1", "n1", "n03", "n03", "n04", "n04", "n04", "n05", "n05")` is
60#'    returned.  If in addition, `SLURM_CPUS_PER_TASK=1`, which can happen
61#'    depending on hyperthreading configurations on the Slurm cluster, then
62#'    `c("n1", "n03", "n04", "n05")` is returned.
63#'
64#'  \item `"custom"` -
65#'    If option \option{parallelly.availableWorkers.custom} is set and a function,
66#'    then this function will be called (without arguments) and it's value
67#'    will be coerced to a character vector, which will be interpreted as
68#'    hostnames of available workers.
69#' }
70#'
71#' @section Known limitations:
72#' `availableWorkers(methods = "Slurm")` will expand \env{SLURM_JOB_NODELIST}
73#' using \command{scontrol show hostnames "$SLURM_JOB_NODELIST"}, if available.
74#' If not available, then it attempts to parse the compressed nodelist based
75#' on a best-guess understanding on what the possible syntax may be.
76#' One known limitation is that "multi-dimensional" ranges are not supported,
77#' e.g. `"a[1-2]b[3-4]"` is expanded by \command{scontrol} to
78#' `c("a1b3", "a1b4", "a2b3", "a2b4")`.  If \command{scontrol} is not
79#' available, then any components that failed to be parsed are dropped with
80#' an informative warning message.  If no compents could be parsed, then
81#' the result of `methods = "Slurm"` will be empty.
82#'
83#' @examples
84#' message(paste("Available workers:",
85#'         paste(sQuote(availableWorkers()), collapse = ", ")))
86#'
87#' \dontrun{
88#' options(mc.cores = 2L)
89#' message(paste("Available workers:",
90#'         paste(sQuote(availableWorkers()), collapse = ", ")))
91#' }
92#'
93#' \dontrun{
94#' ## Always use two workers on host 'n1' and one on host 'n2'
95#' options(parallelly.availableWorkers.custom = function() {
96#'   c("n1", "n1", "n2")
97#' })
98#' message(paste("Available workers:",
99#'         paste(sQuote(availableWorkers()), collapse = ", ")))
100#' }
101#'
102#' @seealso
103#' To get the number of available workers on the current machine,
104#' see [availableCores()].
105#'
106#' @importFrom utils file_test
107#' @export
108availableWorkers <- function(methods = getOption2("parallelly.availableWorkers.methods", c("mc.cores", "BiocParallel", "_R_CHECK_LIMIT_CORES_", "PBS", "SGE", "Slurm", "LSF", "custom", "system", "fallback")), na.rm = TRUE, logical = getOption2("parallelly.availableCores.logical", TRUE), default = getOption2("parallelly.localhost.hostname", "localhost"), which = c("auto", "min", "max", "all")) {
109  ## Local functions
110  getenv <- function(name) {
111    as.character(trim(getEnvVar2(name, default = NA_character_)))
112  }
113
114  getopt <- function(name) {
115    as.character(getOption2(name, default = NA_character_))
116  }
117
118  split <- function(s) {
119    x <- unlist(strsplit(s, split = "[, ]", fixed = FALSE), use.names = FALSE)
120    x <- trim(x)
121    x <- x[nzchar(x)]
122    x
123  }
124
125  which <- match.arg(which, choices = c("auto", "min", "max", "all"))
126  stop_if_not(is.character(default), length(default) >= 1, !anyNA(default))
127
128
129  ## Default is to use the current machine
130  ncores <- availableCores(methods = methods, na.rm = FALSE, logical = logical, which = "all")
131
132  localhost_hostname <- getOption2("parallelly.localhost.hostname", "localhost")
133  workers <- lapply(ncores, FUN = function(n) {
134    if (length(n) == 0 || is.na(n)) n <- 0L
135    rep(localhost_hostname, times = n)
136  })
137
138  ## Acknowledge known HPC settings (skip others)
139  methods_localhost <- c("BiocParallel", "_R_CHECK_LIMIT_CORES_", "mc.cores", "mc.cores+1", "system")
140  methodsT <- setdiff(methods, methods_localhost)
141  for (method in methodsT) {
142    if (method == "PBS") {
143      pathname <- getenv("PBS_NODEFILE")
144      if (is.na(pathname)) next
145      if (!file_test("-f", pathname)) {
146        warnf("Environment variable %s was set but no such file %s exists", sQuote("PBS_NODEFILE"), sQuote(pathname))
147        next
148      }
149      data <- read_pbs_nodefile(pathname)
150      w <- data$node
151
152      ## Sanity checks
153      pbs_np <- as.integer(getenv("PBS_NP"))
154      if (!identical(pbs_np, length(w))) {
155        warnf("Identified %d workers from the %s file (%s), which does not match environment variable %s = %d", length(w), sQuote("PBS_NODEFILE"), sQuote(pathname), sQuote("PBS_NP"), pbs_np)
156      }
157
158      pbs_nodes <- as.integer(getenv("PBS_NUM_NODES"))
159      pbs_ppn <- as.integer(getenv("PBS_NUM_PPN"))
160      pbs_np <- pbs_nodes * pbs_ppn
161      if (!identical(pbs_np, length(w))) {
162        warnf("Identified %d workers from the %s file (%s), which does not match environment variables %s * %s = %d * %d = %d", length(w), sQuote("PBS_NODEFILE"), sQuote(pathname), sQuote("PBS_NUM_NODES"), sQuote("PBS_NUM_PPN"), pbs_nodes, pbs_ppn, pbs_np)
163      }
164
165      ## TO DO: Add validation of 'w' (from PBS_HOSTFILE) toward
166      ## counts in PBS_NP and / or PBS_NUM_NODES * PBS_NUM_PPN.
167    } else if (method == "SGE") {
168      pathname <- getenv("PE_HOSTFILE")
169      if (is.na(pathname)) next
170      if (!file_test("-f", pathname)) {
171        warnf("Environment variable %s was set but no such file %s exists", sQuote("PE_HOSTFILE"), sQuote(pathname))
172        next
173      }
174      w <- read_pe_hostfile(pathname, expand = TRUE)
175
176      ## Sanity checks
177      nslots <- as.integer(getenv("NSLOTS"))
178      if (!identical(nslots, length(w))) {
179        warnf("Identified %d workers from the %s file (%s), which does not match environment variable %s = %d", length(w), sQuote("PE_HOSTFILE"), sQuote(pathname), sQuote("NSLOTS"), nslots)
180      }
181    } else if (method == "Slurm") {
182      ## From 'man sbatch':
183      ## SLURM_JOB_NODELIST (and SLURM_NODELIST for backwards compatibility)
184      ## List of nodes allocated to the job.
185      ## Example:
186      ## SLURM_JOB_NODELIST=n1,n[3-8],n[23-25]
187      nodelist <- getenv("SLURM_JOB_NODELIST")
188      if (is.na(nodelist)) data <- getenv("SLURM_NODELIST")
189      if (is.na(nodelist)) next
190
191      ## Parse and expand nodelist
192      w <- slurm_expand_nodelist(nodelist)
193
194      ## Failed to parse?
195      if (length(w) == 0) next
196
197      ## SLURM_JOB_CPUS_PER_NODE=64,12,...
198      nodecounts <- getenv("SLURM_JOB_CPUS_PER_NODE")
199      if (is.na(nodecounts)) nodecounts <- getenv("SLURM_TASKS_PER_NODE")
200      if (is.na(nodecounts)) {
201        warning("Expected either environment variable 'SLURM_JOB_CPUS_PER_NODE' or 'SLURM_TASKS_PER_NODE' to be set. Will assume one core per node.")
202      } else {
203        ## Parse counts
204	c <- slurm_expand_nodecounts(nodecounts)
205        if (any(is.na(c))) {
206          warnf("Failed to parse 'SLURM_JOB_CPUS_PER_NODE' or 'SLURM_TASKS_PER_NODE': %s", sQuote(nodecounts))
207          next
208        }
209
210        if (length(c) != length(w)) {
211          warnf("Skipping Slurm settings because the number of elements in 'SLURM_JOB_CPUS_PER_NODE'/'SLURM_TASKS_PER_NODE' (%s) does not match parsed 'SLURM_JOB_NODELIST'/'SLURM_NODELIST' (%s): %d != %d", nodelist, nodecounts, length(c), length(w))
212          next
213        }
214
215        ## Always respect 'SLURM_CPUS_PER_TASK' (always a scalar), if that exists
216        n <- getenv("SLURM_CPUS_PER_TASK")
217        if (!is.na(n)) {
218          c0 <- c
219          c <- rep(n, times = length(w))
220          ## Is our assumption that SLURM_CPUS_PER_TASK <= SLURM_JOB_NODELIST, correct?
221          if (any(c < n)) {
222            c <- pmin(c, n)
223            warnf("Unexpected values of Slurm environment variable. 'SLURM_CPUS_PER_TASK' specifies CPU counts on one or more nodes that is strictly less than what 'SLURM_CPUS_PER_TASK' specifies. Will use the minimum of the two for each node: %s < %s", sQuote(nodecounts), n)
224          }
225        }
226
227        ## Expand workers list
228        w <- as.list(w)
229        for (kk in seq_along(w)) {
230          w[[kk]] <- rep(w[[kk]], times = c[kk])
231        }
232        w <- unlist(w, use.names = FALSE)
233      }
234    } else if (method == "LSF") {
235      data <- getenv("LSB_HOSTS")
236      if (is.na(data)) next
237      w <- split(data)
238    } else if (method == "custom") {
239      fcn <- getOption2("parallelly.availableWorkers.custom", NULL)
240      if (!is.function(fcn)) next
241      w <- local({
242        ## Avoid calling the custom function recursively
243        oopts <- options(parallelly.availableWorkers.custom = NULL)
244        on.exit(options(oopts))
245        fcn()
246      })
247      w <- as.character(w)
248    } else {
249      ## Fall back to querying option and system environment variable
250      ## with the given name
251      w <- getopt(method)
252      if (is.na(w)) w <- getenv(method)
253      if (is.na(w)) next
254      w <- split(w)
255    }
256
257    ## Drop missing values?
258    if (na.rm) w <- w[!is.na(w)]
259
260    workers[[method]] <- w
261  }
262
263
264  nnodes <- unlist(lapply(workers, FUN = length), use.names = TRUE)
265
266  if (which == "auto") {
267    ## For default localhost sets, use the minimum allowed number of
268    ## workers **according to availableCores()**.
269    methodsT <- intersect(names(workers), methods_localhost)
270    methodsT <- methodsT[is.finite(ncores[methodsT]) & ncores[methodsT] > 0]
271    if (length(methodsT) > 1L) {
272      min <- min(ncores[methodsT], na.rm = TRUE)
273      if (is.finite(min)) {
274        nnodes[methodsT] <- min
275        workers[methodsT] <- list(rep(localhost_hostname, times = min))
276      }
277    }
278
279    workers <- apply_fallback(workers)
280    nnodes <- unlist(lapply(workers, FUN = length), use.names = TRUE)
281
282    ## Now, pick the first positive and finite value
283    idx <- which(nnodes > 0L, useNames = FALSE)[1]
284    workers <- if (is.na(idx)) character(0L) else workers[[idx]]
285  } else if (which == "min") {
286    workers <- apply_fallback(workers)
287    nnodes <- unlist(lapply(workers, FUN = length), use.names = TRUE)
288    idx <- which.min(nnodes)
289    workers <- workers[[idx]]
290  } else if (which == "max") {
291    workers <- apply_fallback(workers)
292    nnodes <- unlist(lapply(workers, FUN = length), use.names = TRUE)
293    idx <- which.max(nnodes)
294    workers <- workers[[idx]]
295  }
296
297  ## Fall back to default?
298  if (is.character(workers) && length(workers) == 0) workers <- default
299
300  ## Sanity checks
301  min_count <- as.integer(na.rm)
302  if (is.list(workers)) {
303    lapply(workers, FUN = function(w) {
304      stop_if_not(is.character(w), length(w) >= 0L, all(nchar(w) > 0))
305    })
306  } else {
307    stop_if_not(is.character(workers), length(workers) >= min_count, all(nchar(workers) > 0))
308  }
309
310  workers
311} # availableWorkers()
312
313
314
315#' @importFrom utils read.table
316read_pbs_nodefile <- function(pathname, sort = TRUE) {
317  ## One (node) per line
318  lines <- readLines(pathname, warn = FALSE)
319  lines <- trim(lines)
320
321  ## Sanity checks
322  stop_if_not(
323    all(nzchar(lines)),
324    !anyNA(lines),
325    !any(grepl("[[:space:]]", lines))
326  )
327
328  data <- data.frame(node = lines, stringsAsFactors = FALSE)
329
330  if (sort) {
331    data <- data[order(data$node), , drop = FALSE]
332  }
333
334  data
335}
336
337
338#' @importFrom utils read.table
339read_pe_hostfile <- function(pathname, sort = TRUE, expand = FALSE) {
340  ## One (node, ncores, queue, comment) per line, e.g.
341  ## opt88 3 short.q@opt88 UNDEFINED
342  ## iq242 2 short.q@iq242 UNDEFINED
343  ## opt116 1 short.q@opt116 UNDEFINED
344  data <- read.table(pathname, header = FALSE, sep = " ", stringsAsFactors = FALSE)
345
346  ## Sanity checks
347  stop_if_not(ncol(data) >= 2)
348
349  colnames(data)[1:2] <- c("node", "count")
350  if (ncol(data) >= 3) colnames(data)[3] <- "via"
351  if (ncol(data) >= 4) colnames(data)[4] <- "notes"
352
353  stop_if_not(
354    is.character(data$node),
355    !anyNA(data$node),
356    !any(grepl("[[:space:]]", data$node)),
357    is.integer(data$count),
358    !anyNA(data$count),
359    all(is.finite(data$count)),
360    all(data$count > 0)
361  )
362
363  if (sort) {
364    data <- data[order(data$node, data$count), , drop = FALSE]
365  }
366
367  if (expand) {
368    data <- sge_expand_node_count_pairs(data)
369  }
370
371  data
372}
373
374## Used after read_pe_hostfile()
375sge_expand_node_count_pairs <- function(data) {
376  nodes <- mapply(data$node, data$count, FUN = function(node, count) {
377    rep(node, times = count)
378  }, SIMPLIFY = FALSE, USE.NAMES = FALSE)
379  unlist(nodes, recursive = FALSE, use.names = FALSE)
380}
381
382
383#' @importFrom utils file_test
384call_slurm_show_hostname <- function(nodelist, bin = Sys.which("scontrol")) {
385  stop_if_not(file_test("-x", bin))
386
387  args <- c("show", "hostname", shQuote(nodelist))
388  res <- system2(bin, args = args, stdout = TRUE)
389  status <- attr(res, "status")
390  if (!is.null(status)) {
391    call <- sprintf("%s %s", shQuote(bin), paste(args, collapse = " "))
392    msg <- sprintf("%s failed with exit code %s", call, status)
393    stop(msg)
394  }
395
396  res
397}
398
399supports_scontrol_show_hostname <- local({
400  res <- NA
401  function() {
402    if (!is.na(res)) return(res)
403
404    ## Look for 'scontrol'
405    bin <- Sys.which("scontrol")
406    if (!nzchar(bin)) {
407      res <<- FALSE
408      return(res)
409    }
410
411    ## Try a conversion
412    truth <- c("a1", "b02", "b03", "b04", "b6", "b7")
413    nodelist <- "a1,b[02-04,6-7]"
414
415    hosts <- tryCatch({
416      call_slurm_show_hostname(nodelist, bin = bin)
417    }, error = identity)
418
419    if (inherits(hosts, "error")) {
420      res <<- FALSE
421      return(res)
422    }
423
424    ## Sanity check
425    if (!isTRUE(all.equal(sort(hosts), sort(truth)))) {
426      warnf("Internal availableWorkers() validation failed: 'scontrol show hostnames %s' did not return the expected results.  Expected c(%s) but got c(%s).  Will still use it this methods but please report this to the maintainer of the 'parallelly' package", shQuote(nodelist), commaq(truth), commaq(hosts), immediate. = TRUE)
427    }
428
429    value <- TRUE
430    attr(value, "scontrol") <- bin
431    res <<- value
432
433    res
434  }
435})
436
437
438## SLURM_JOB_NODELIST="a1,b[02-04,6-7]"
439slurm_expand_nodelist <- function(nodelist, manual = getOption2("parallelly.slurm_expand_nodelist.manual", FALSE)) {
440  ## Alt 1. Is 'scontrol show hostnames' supported?
441  if (!manual && supports_scontrol_show_hostname()) {
442    hosts <- call_slurm_show_hostname(nodelist)
443    return(hosts)
444  }
445
446  ## Alt 2. Manually parse the nodelist specification
447  data <- nodelist
448
449  ## Replace whitespace *within* square brackets with zeros
450  ## Source: scontrol show hostnames treats "n[1,  3-4]" == "n[1,003-4]"
451  pattern <- "\\[([[:digit:],-]*)[[:space:]]([[:digit:][:space:],-]*)"
452  while (grepl(pattern, data)) {
453    data <- gsub(pattern, "[\\10\\2", data)
454  }
455
456  ## Replace any commas *within* square brackets with semicolons
457  pattern <- "\\[([[:digit:][:space:];-]*),([[:digit:][:space:];-]*)"
458  while (grepl(pattern, data)) {
459    data <- gsub(pattern, "[\\1;\\2", data)
460  }
461
462  data <- strsplit(data, split = "[,[:space:]]", fixed = FALSE)
463  data <- as.list(unlist(data, use.names = FALSE))
464
465  ## Keep only non-empty entries, which may happen due to whitespace or
466  ## extra commas.  This should not happen but 'scontrol show hostnames'
467  ## handles those cases too.
468  data <- data[nzchar(data)]
469
470  for (ii in seq_along(data)) {
471    spec <- data[[ii]]
472
473    ## Already expanded?
474    if (length(spec) > 1L) next
475
476    ## 1. Expand square-bracket specifications
477    ##    e.g. "a1,b[02-04,6-7]" => c("a1", "b02", "b03", "b04", "b6", "b7")
478    pattern <- "^(.*)\\[([[:digit:];-]+)\\]$"
479    if (grepl(pattern, spec)) {
480      prefix <- gsub(pattern, "\\1", spec)
481      set <- gsub(pattern, "\\2", spec)
482
483      sets <- strsplit(set, split = ";", fixed = TRUE)
484      sets <- unlist(sets, use.names = FALSE)
485      sets <- as.list(sets)
486
487      for (jj in seq_along(sets)) {
488        set <- sets[[jj]]
489        ## Expand by evaluating them as R expressions
490        idxs <- tryCatch({
491          expr <- parse(text = gsub("-", ":", set, fixed = TRUE))
492          eval(expr, envir = baseenv())
493        }, error = function(e) NA_integer_)
494        idxs <- as.character(idxs)
495
496        ## Pad with zeros?
497        pattern <- "^([0]*)[[:digit:]]+.*"
498        if (grepl(pattern, set)) {
499          pad <- gsub(pattern, "\\1", set)
500          idxs <- paste(pad, idxs, sep = "")
501        }
502
503        set <- paste(prefix, idxs, sep = "")
504        sets[[jj]] <- set
505      } ## for (jj ...)
506
507      sets <- unlist(sets, use.names = FALSE)
508      data[[ii]] <- sets
509    }
510  } ## for (ii in ...)
511
512  hosts <- unlist(data, recursive = FALSE, use.names = FALSE)
513
514  ## Sanity check
515  if (any(!nzchar(hosts))) {
516    warnf("Unexpected result from parallelly:::slurm_expand_nodelist(..., manual = TRUE), which resulted in %d empty hostname based on nodelist specification %s", sum(!nzchar(hosts)), sQuote(nodelist))
517    hosts <- hosts[nzchar(hosts)]
518  }
519
520  ## Failed to expand all compressed ranges?  This may happen because
521  ## "multi-dimensional" ranges are given, e.g. "a[1-2]b[3-4]". This is
522  ## currently not supported by the above manual parser. /HB 2021-03-05
523  invalid <- grep("(\\[|\\]|,|;|[[:space:]])", hosts, value = TRUE)
524  if (length(invalid) > 0) {
525    warnf("Failed to parse the compressed Slurm nodelist %s. Detected invalid node names, which are dropped: %s", sQuote(nodelist), commaq(invalid))
526    hosts <- setdiff(hosts, invalid)
527  }
528
529  hosts
530}
531
532
533SLURM_TASKS_PER_NODE="2(x2),1(x3)"  # Source: 'man sbatch'
534slurm_expand_nodecounts <- function(nodecounts) {
535  counts <- strsplit(nodecounts, split = ",", fixed = TRUE)
536  counts <- unlist(counts, use.names = TRUE)
537  counts <- counts[nzchar(counts)]
538  counts <- as.list(counts)
539  counts <- lapply(counts, FUN = function(count) {
540    ## Drop whitespace
541    count <- gsub("[[:space:]]", "", count)
542    pattern <- "^([[:digit:]]+)[(]x([[:digit:]]+)[)]$"
543    if (grepl(pattern, count)) {
544      times <- gsub(pattern, "\\2", count)
545      times <- as.integer(times)
546      if (is.na(times)) return(NA_integer_)
547
548      count <- gsub(pattern, "\\1", count)
549      count <- as.integer(count)
550      if (is.na(count)) return(NA_integer_)
551
552      count <- rep(count, times = times)
553    } else {
554      count <- as.integer(count)
555    }
556  })
557  counts <- unlist(counts, use.names = TRUE)
558
559  if (any(is.na(counts))) {
560    warnf("Failed to parse Slurm node counts specification: %s", nodecounts)
561  }
562
563  counts
564}
565
566
567
568## Used by availableWorkers()
569apply_fallback <- function(workers) {
570  ## No 'fallback'?
571  idx_fallback <- which(names(workers) == "fallback")
572  if (length(idx_fallback) == 0) return(workers)
573
574  ## Number of workers per set
575  nnodes <- unlist(lapply(workers, FUN = length), use.names = TRUE)
576
577  ## No 'fallback' workers?
578  if (nnodes[idx_fallback] == 0) return(workers)
579
580  ## Only consider non-empty sets
581  nonempty <- which(nnodes > 0)
582  workers_nonempty <- workers[nonempty]
583
584  ## Nothing to do?
585  n_nonempty <- length(workers_nonempty)
586  if (n_nonempty <= 1) return(workers)
587
588  ## Drop 'fallback'?
589  if (n_nonempty > 2) {
590    workers <- workers[-idx_fallback]
591    return(workers)
592  }
593
594  ## No 'system' to override?
595  idx_system <- which(names(workers) == "system")
596  if (length(idx_system) == 0) return(workers)
597
598  ## Drop 'system' in favor or 'fallback'
599  workers <- workers[-idx_system]
600
601  workers
602} ## apply_fallback()
603