Error with future::plan(future::cluster, workers = c('rey', 'r2d2'))

[ramiromagno]

future::plan(future::cluster, workers = c('rey', 'r2d2'))

fish: Unknown command: /usr/lib64/R/bin/Rscript
fish: 
'/usr/lib64/R/bin/Rscript' --default-packages=datasets,utils,grDevices,graphics,stats,methods -e 'options(socketOptions = "no-delay")' -e 'workRSOCK <- tryCatch(parallel:::.workRSOCK, error=function(e) parallel:::.slaveRSOCK); workRSOCK()' MASTER=localhost PORT=11329 OUT=/dev/null TIMEOUT=2592000 XDR=FALSE SETUPTIMEOUT=120 SETUPSTRATEGY=sequential
^

and it hangs and I have to stop the process.

Session information

> sessionInfo()
R version 4.1.2 (2021-11-01)
Platform: x86_64-pc-linux-gnu (64-bit)
Running under: Arch Linux

Matrix products: default
BLAS:   /usr/lib/libopenblasp-r0.3.18.so
LAPACK: /usr/lib/liblapack.so.3.10.0

locale:
 [1] LC_CTYPE=en_US.UTF-8       LC_NUMERIC=C               LC_TIME=en_US.UTF-8        LC_COLLATE=en_US.UTF-8    
 [5] LC_MONETARY=en_US.UTF-8    LC_MESSAGES=en_US.UTF-8    LC_PAPER=en_US.UTF-8       LC_NAME=C                 
 [9] LC_ADDRESS=C               LC_TELEPHONE=C             LC_MEASUREMENT=en_US.UTF-8 LC_IDENTIFICATION=C       

attached base packages:
[1] stats     graphics  grDevices utils     datasets  methods   base     

other attached packages:
[1] daeqtlr_0.1.0

loaded via a namespace (and not attached):
 [1] compiler_4.1.2     parallelly_1.31.1  magrittr_2.0.1     cli_3.3.0          parallel_4.1.2     tools_4.1.2       
 [7] future.apply_1.9.0 listenv_0.8.0      codetools_0.2-18   data.table_1.14.2  digest_0.6.28      globals_0.14.0    
[13] rlang_1.0.2        future_1.25.0     

> future::futureSessionInfo()
*** Package versions
future 1.25.0, parallelly 1.31.1, parallel 4.1.2, globals 0.14.0, listenv 0.8.0

*** Allocations
availableCores():
system  nproc 
     6      6 
availableWorkers():
$system
[1] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"


*** Settings
- future.plan=<not set>
- future.fork.multithreading.enable=<not set>
- future.globals.maxSize=<not set>
- future.globals.onReference=<not set>
- future.resolve.recursive=<not set>
- future.rng.onMisuse=<not set>
- future.wait.timeout=<not set>
- future.wait.interval=<not set>
- future.wait.alpha=<not set>
- future.startup.script=<not set>

*** Backends
Number of workers: 2
List of future strategies:
1. cluster:
   - args: function (..., workers = c("rey", "r2d2"), envir = parent.frame())
   - tweaked: TRUE
   - call: future::plan(future::cluster, workers = c("rey", "r2d2"))

*** Basic tests
fish: Unknown command: /usr/lib64/R/bin/Rscript
fish: 
'/usr/lib64/R/bin/Rscript' --default-packages=datasets,utils,grDevices,graphics,stats,methods -e 'options(socketOptions = "no-delay")' -e 'workRSOCK <- tryCatch(parallel:::.workRSOCK, error=function(e) parallel:::.slaveRSOCK); workRSOCK()' MASTER=localhost PORT=11091 OUT=/dev/null TIMEOUT=2592000 XDR=FALSE SETUPTIMEOUT=120 SETUPSTRATEGY=sequential
^

also hangs here.

[HenrikBengtsson]

Using

future::plan(future::cluster, workers = c('rey', 'r2d2'), homogeneous = FALSE)

should work.

Explanation

When using

future::plan(future::cluster, workers = c('rey', 'r2d2'))

the parallelly packages is used under the hood. It basically does the following:

cl <- parallelly::makeClusterPSOCK(c('rey', 'r2d2'))
future::plan(future::cluster, workers = cl)

Comment: If you call that, you'll get the same error in the first step.

Now, the default is homogeneous = NULL, which will make makeClusterPSOCK() look at each workers element to infer whether if that needs homogeneous = TRUE or homogeneous = FALSE. In your case, they end becoming homogeneous = TRUE, which you don't want, because your worker machines do not have R installed in the same location as the machine you're calling this from. You can read about the details in Section 'Default value of argument homogeneous' on help("makeClusterPSOCK", package = "parallelly"). If your workers would be specified with fully qualified domain name (FQDN), e.g. workers = c("rey.example.org", "r2d2.example.org"), the default would have become homogeneous = FALSE.

[ramiromagno]

Curiously dryrun = TRUE is failing me now (with dryrun = FALSE seems to run without errors):

> future::plan(future::cluster, workers = c('rey', 'r2d2'), homogeneous = FALSE, dryrun = TRUE)
----------------------------------------------------------------------
Manually, (i) login into external machine ‘r2d2’:

  '/usr/bin/ssh' -R 11179:localhost:11179 r2d2

and (ii) start worker #1 from there:

  'Rscript' --default-packages=datasets,utils,grDevices,graphics,stats,methods -e 'options(socketOptions = "no-delay")' -e 'workRSOCK <- tryCatch(parallel:::.workRSOCK, error=function(e) parallel:::.slaveRSOCK); workRSOCK()' MASTER=localhost PORT=11179 OUT=/dev/null TIMEOUT=2592000 XDR=FALSE SETUPTIMEOUT=120 SETUPSTRATEGY=sequential

Alternatively, start worker #1 from the local machine by combining both step in a single call:

  '/usr/bin/ssh' -R 11179:localhost:11179 r2d2 "'Rscript' --default-packages=datasets,utils,grDevices,graphics,stats,methods -e 'options(socketOptions = \"no-delay\")' -e 'workRSOCK <- tryCatch(parallel:::.workRSOCK, error=function(e) parallel:::.slaveRSOCK); workRSOCK()' MASTER=localhost PORT=11179 OUT=/dev/null TIMEOUT=2592000 XDR=FALSE SETUPTIMEOUT=120 SETUPSTRATEGY=sequential"

----------------------------------------------------------------------
Manually, (i) login into external machine ‘rey’:

  '/usr/bin/ssh' -R 11180:localhost:11179 rey

and (ii) start worker #2 from there:

  'Rscript' --default-packages=datasets,utils,grDevices,graphics,stats,methods -e 'options(socketOptions = "no-delay")' -e 'workRSOCK <- tryCatch(parallel:::.workRSOCK, error=function(e) parallel:::.slaveRSOCK); workRSOCK()' MASTER=localhost PORT=11180 OUT=/dev/null TIMEOUT=2592000 XDR=FALSE SETUPTIMEOUT=120 SETUPSTRATEGY=sequential

Alternatively, start worker #2 from the local machine by combining both step in a single call:

  '/usr/bin/ssh' -R 11180:localhost:11179 rey "'Rscript' --default-packages=datasets,utils,grDevices,graphics,stats,methods -e 'options(socketOptions = \"no-delay\")' -e 'workRSOCK <- tryCatch(parallel:::.workRSOCK, error=function(e) parallel:::.slaveRSOCK); workRSOCK()' MASTER=localhost PORT=11180 OUT=/dev/null TIMEOUT=2592000 XDR=FALSE SETUPTIMEOUT=120 SETUPSTRATEGY=sequential"

Error: ClusterFuture (future-plan-test) failed to call grmall() on cluster NULL #1 (). The reason reported was ‘no applicable method for 'sendData' applied to an object of class "NULL"’

[HenrikBengtsson]

Yes, sorry I wasn't clear. That's expected because, per design, dryrun = TRUE doesn't actually launch those commands. It's less confusing when you use it directly with makeClusterPSOCK();

cl <- parallelly::makeClusterPSOCK(c('rey', 'r2d2'), homogeneous = TRUE, dryrun = TRUE)
cl <- parallelly::makeClusterPSOCK(c('rey', 'r2d2'), homogeneous = FALSE, dryrun = TRUE)

Then you'll see what should have happened (if you didn't do a dry-run) without getting confusing errors.

[ramiromagno]

Oh nice! That works.

Now I was trying to set the number of cores per worker but it is running on one core only in each node.

Here's my setup:

library(future.apply)
library(daeqtlr)

custom_workers <- function() {
  switch(Sys.info()[["nodename"]],
         "rey" = 64L,
         "r2d2" = 40L,
         ## default:
         parallelly::availableCores()
  )
}

future::plan(list(
  future::tweak(
    future::cluster,
    workers = c('rey', 'r2d2'),
    homogeneous = FALSE
  ),
  future::tweak(future::multisession, workers = custom_workers)
))


snp_pairs <- read_snp_pairs(file = daeqtlr_example("snp_pairs.csv"))
zygosity <- read_snp_zygosity(file = daeqtlr_example("zygosity.csv"))
ae <- read_ae_ratios(file = daeqtlr_example("ae.csv"))

snp_pairs2 <- snp_pairs[sample(1:200, 10000, replace = TRUE), ]

daeqtl_mapping2(snp_pairs = snp_pairs2, zygosity = zygosity, ae = ae, .n_cores = 64 + 40)

The function daeqtl_mapping2():

#' @export
daeqtl_mapping2 <-
  function(snp_pairs,
           zygosity,
           ae,
           fn = daeqtl_test,
           ...,
           .extra_cols = 2L,
           .n_cores = 1L) {

    snp_pairs_lst <-
      split(snp_pairs, split_index(nrow(snp_pairs), .n_cores))

    for (i in seq_along(snp_pairs_lst)) {
      data.table::setkeyv(snp_pairs_lst[[i]], c('dae_snp', 'candidate_snp'))
    }

    res <- future.apply::future_lapply(
      snp_pairs_lst,
      FUN = daeqtl_mapping_,
      zygosity = zygosity,
      ae = ae,
      fn = fn,
      ...,
      .extra_cols = .extra_cols
    )

    # Put together the data table from the list of data tables
    mapping_dt <- data.table::rbindlist(res)

    data.table::setkeyv(mapping_dt, cols = c('dae_snp', 'candidate_snp'))
    return(mapping_dt)
  }

[HenrikBengtsson]

Please start a new Discussion for this.

[ramiromagno]

Sure, thanks!