availableWorkers() now supports LSF/OpenLava [#118]

NEWS

@@ -37,8 +37,9 @@ NEW FEATURES:
    --slurm-cpus-per-task=n' is specified and SLURM_JOB_NUM_NODES=1, then it
    falls back to using 'SLURM_CPUS_ON_NODE', e.g. when using '--ntasks=n'.
 
- * Now availableCores() supports LSF/OpenLava.  Specifically, environment
-   variable 'LSB_DJOB_NUMPROC' is acknowledged.
+ * Now availableCores() and availableWorkers() supports LSF/OpenLava.
+   Specifically, they acknowledge environment variable 'LSB_DJOB_NUMPROC'
+   and 'LSB_HOSTS', respectively.
 
 PERFORMANCE:
 

R/availableWorkers.R

@@ -38,6 +38,8 @@
 #'    An example of a job submission that results in this is
 #'    `qsub -pe mpi 8` (or `qsub -pe ompi 8`), which
 #'    requests eight cores on a any number of machines.
+#'  \item `"LSF"` -
+#'    Query LSF/OpenLava environment variable \env{LSB_HOSTS}.
 #'  \item `"custom"` -
 #'    If option \option{future.availableWorkers.custom} is set and a function,
 #'    then this function will be called (without arguments) and it's value
@@ -52,7 +54,7 @@
 #' @importFrom utils file_test
 #' @export
 #' @keywords internal
-availableWorkers <- function(methods = getOption("future.availableWorkers.methods", c("mc.cores", "_R_CHECK_LIMIT_CORES_", "PBS", "SGE", "Slurm", "custom", "system", "fallback")), na.rm = TRUE, default = "localhost", which = c("auto", "min", "max", "all")) {
+availableWorkers <- function(methods = getOption("future.availableWorkers.methods", c("mc.cores", "_R_CHECK_LIMIT_CORES_", "PBS", "SGE", "Slurm", "LSF", "custom", "system", "fallback")), na.rm = TRUE, default = "localhost", which = c("auto", "min", "max", "all")) {
   ## Local functions
   getenv <- function(name) {
     as.character(trim(Sys.getenv(name, NA_character_)))
@@ -135,6 +137,12 @@ availableWorkers <- function(methods = getOption("future.availableWorkers.method
       ## TODO: Parse 'data' into a hostnames /HB 2020-09-18
       ## ...
       next
+    } else if (method == "LSF") {
+      data <- getenv("LSB_HOSTS")
+      if (is.na(data)) next
+      w <- strsplit(data, split = " ", fixed = TRUE)[[1]]
+      w <- w[nzchar(w)]
+      if (length(w) == 0L) next
     } else if (method == "custom") {
       fcn <- getOption("future.availableWorkers.custom", NULL)
       if (!is.function(fcn)) next

R/options.R

@@ -54,7 +54,7 @@
 #' 
 #'  \item{\option{future.availableCores.system}:}{(integer) Number of "system" cores used instead of what is reported by \code{\link{availableCores}(which = "system")}. If not specified, this option is set according to system environment variable \env{R_FUTURE_AVAILABLECORES_SYSTEM} when the \pkg{future} package is _loaded_. This option allows you to effectively override what `parallel::detectCores()` reports the system has.}
 #'
-#'  \item{\option{future.availableWorkers.methods}:}{(character vector) Default lookup methods for [availableWorkers()]. (Default: `c("mc.cores", "_R_CHECK_LIMIT_CORES_", "PBS", "SGE", "Slurm", "custom", "system", "fallback")`)}
+#'  \item{\option{future.availableWorkers.methods}:}{(character vector) Default lookup methods for [availableWorkers()]. (Default: `c("mc.cores", "_R_CHECK_LIMIT_CORES_", "PBS", "SGE", "Slurm", "LSF", "custom", "system", "fallback")`)}
 #'
 #'  \item{\option{future.availableWorkers.custom}:}{(function) If set and a function, then this function will be called (without arguments) by [availableWorkers()] where its value, coerced to a character vector, is interpreted as hostnames of available workers.}
 #'

tests/availableCores.R

@@ -30,6 +30,7 @@ stopifnot(length(n) == 1, is.numeric(n), is.finite(n), n >= 1)
 print(availableCores(methods = "PBS"))
 print(availableCores(methods = "SGE"))
 print(availableCores(methods = "Slurm"))
+print(availableCores(methods = "LSF"))
 
 ## Any R options and system environment variable
 print(availableCores(methods = c("width", "FOO_BAR_ENV"),
@@ -41,6 +42,16 @@ res <- try(availableCores(methods = "FOO_BAR_ENV"), silent = TRUE)
 stopifnot(inherits(res, "try-error"))
 
 
+ncores0 <- 42L
+
+message("*** LSF ...")
+Sys.setenv(LSB_DJOB_NUMPROC = as.character(ncores0))
+ncores <- availableCores(methods = "LSF")
+print(ncores)
+stopifnot(ncores == ncores0)
+message("*** LSF ... done")
+
+
 message("*** Internal detectCores() ...")
 
 ## Option 'future.availableCores.system'

tests/availableWorkers.R

@@ -25,6 +25,7 @@ stopifnot(is.character(w), length(w) >= 1)
 print(availableWorkers(methods = "PBS"))
 print(availableWorkers(methods = "SGE"))
 print(availableWorkers(methods = "Slurm"))
+print(availableWorkers(methods = "LSF"))
 
 
 
@@ -39,6 +40,18 @@ data0 <- as.data.frame(table(workers0), stringsAsFactors = FALSE)
 colnames(data0) <- c("node", "count")
 data0 <- data0[order(data0$node, data0$count), ]
 
+
+message("*** LSF ...")
+
+Sys.setenv(LSB_HOSTS = paste(workers0, collapse = " "))
+workers <- availableWorkers(methods = "LSF")
+print(workers)
+stopifnot(length(workers) == length(workers0))
+
+message("*** LSF ... done")
+
+
+
 message("*** read_pbs_nodefile() ...")
 
 workers <- workers0
@@ -89,7 +102,6 @@ res <- tryCatch({
 }, warning = identity)
 stopifnot(inherits(res, "warning"))
 
-
 message("*** read_pbs_nodefile() ... DONE")