Running Future issue

[Tomer9w]

Hi
Trying to use 'future' to parallelize for-loops, however measuring time for sequencial versus multiprocess yielded no optimal result for the parallel process. and i wonder why.
Could be something in my code or perhaps my understanding, wither way i'm here to learn :)

In the code i used FindAllMarkers function (from Seurat package) on a subset of my data.
A snippet of my code:
Sequencial - took 33 seconds

plan(sequential)
p <- list()
start_time <- Sys.time()
for (i in seq(1:3)){
  markers <- FindAllMarkers(CD3, only.pos = TRUE, min.pct = 0.25, logfc.threshold = 0.25, return.thresh = 0.05)
  #print(p[i])
}
end_time <- Sys.time()
end_time-start_time

Concole:
Calculating cluster 1
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=01s
Calculating cluster 3
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=02s
Calculating cluster 5
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=05s
Calculating cluster 1
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=01s
Calculating cluster 3
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=01s
Calculating cluster 5
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=05s
Calculating cluster 1
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=01s
Calculating cluster 3
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=02s
Calculating cluster 5
|++++++++++++++++++++++++++++++++++++++++++++++++++| 100% elapsed=05s

end_time <- Sys.time()
end_time-start_time
Time difference of 32.96481 secs

multiprocess - took 38 seconds

plan("multiprocess", workers=6)
p <- list()
start_time <- Sys.time()
for (i in seq(1:3)){
  markers <- FindAllMarkers(CD3, only.pos = TRUE, min.pct = 0.25, logfc.threshold = 0.25, return.thresh = 0.05)
  #print(p[i])
}
end_time <- Sys.time()
end_time-start_time

console:
Calculating cluster 1
Calculating cluster 3
Calculating cluster 5
Calculating cluster 1
Calculating cluster 3
Calculating cluster 5
Calculating cluster 1
Calculating cluster 3
Calculating cluster 5

end_time <- Sys.time()
end_time-start_time
Time difference of 38.50585 secs

First thing that comes up to my eyes in the console is that the multiprocess is actually running sequencially.
If more data is needed, i'd be happy to share.

Thanks very much!!

[HenrikBengtsson]

(Moved to Discussions)

Hi.

1. Let's start with establishing your work environment. What's your sessionInfo() after you run your code?

2. The 'multiprocess' backend strategy has been deprecated; please use plan(multisession, workers = 6) instead.

FYI, it's always better to provide a small, reproducible example to illustrate your problem. Maybe you can use one of the examples that Seurat provides to illustrate the problem?

PS. Unrelated, but I suspect you don't want seq(1:3), but rather 1:3.

[Tomer9w]

Hi!
Two changes:

1. I replaced seq(1:3) to 1:3.
2. I replaced the function 'FindAllMarkers' to 'FindMarkers', which is the one used in Seurat example in https://satijalab.org/seurat/archive/v3.0/future_vignette.html

I measured the time: 41.38317 seconds for sequencinal and 4.397994 mins for multisession.
Code below:

library(Seurat)
library(tidyverse)
library(future)
memory.limit(10^13)
setwd("C:\\Users\\Tomer\\Desktop\\Master\\Projects\\Teva")

options(future.globals.maxSize = 16000 * 1024^2)

CD3 <- readRDS('Objects\\CD3.integrated.final.RDS')
CD3 <- subset(CD3, idents=c(1,2,3))

plan(sequential)
start_time <- Sys.time()
for (i in 1:3){
  markers <- FindMarkers(CD3, only.pos = TRUE, ident.1 = i, min.pct = 0.25, logfc.threshold = 0.25, return.thresh = 0.05, verbose = FALSE)
}
end_time <- Sys.time()
end_time-start_time


plan(multisession, workers=6)
start_time <- Sys.time()
for (i in 1:3){
  markers <- FindMarkers(CD3, only.pos = TRUE, ident.1 = i, min.pct = 0.25, logfc.threshold = 0.25, return.thresh = 0.05, verbose = FALSE)
}
end_time <- Sys.time()
end_time-start_time

Just to clarify things up, I'm using 'FindMarkers' as an example, not the main issue itself.
I want to parallelize another process i'm working on, which runs a certain function on a dataset. My desire was to speed things up by sending each process to different core.
I used FindMarkers to see if 'future' runs optimally, that's when i realized something's not working properly.
Is this the right approach?

sessionInfo()
R version 4.0.4 (2021-02-15)
Platform: x86_64-w64-mingw32/x64 (64-bit)
Running under: Windows 10 x64 (build 19042)

Matrix products: default

locale:
[1] LC_COLLATE=Hebrew_Israel.1255 LC_CTYPE=Hebrew_Israel.1255 LC_MONETARY=Hebrew_Israel.1255
[4] LC_NUMERIC=C LC_TIME=Hebrew_Israel.1255

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

other attached packages:
[1] RColorBrewer_1.1-2 patchwork_1.1.1 writexl_1.3.1 plyr_1.8.6 gridExtra_2.3 sleepwalk_0.3.1
[7] forcats_0.5.1 stringr_1.4.0 dplyr_1.0.6 purrr_0.3.4 readr_1.4.0 tidyr_1.1.2
[13] tibble_3.1.0 ggplot2_3.3.3 tidyverse_1.3.0 SeuratObject_4.0.0 Seurat_4.0.0 future_1.21.0

loaded via a namespace (and not attached):
[1] Rtsne_0.15 colorspace_2.0-0 deldir_0.2-10 ellipsis_0.3.1 ggridges_0.5.3
[6] fs_1.5.0 rstudioapi_0.13 spatstat.data_2.0-0 farver_2.1.0 leiden_0.3.7
[11] listenv_0.8.0 ggrepel_0.9.1 fansi_0.4.2 lubridate_1.7.10 xml2_1.3.2
[16] R.methodsS3_1.8.1 codetools_0.2-18 splines_4.0.4 knitr_1.31 polyclip_1.10-0
[21] jsonlite_1.7.2 broom_0.7.5 ica_1.0-2 cluster_2.1.1 dbplyr_2.1.0
[26] R.oo_1.24.0 png_0.1-7 uwot_0.1.10 shiny_1.6.0 sctransform_0.3.2
[31] compiler_4.0.4 httr_1.4.2 backports_1.2.1 assertthat_0.2.1 Matrix_1.3-2
[36] fastmap_1.1.0 lazyeval_0.2.2 limma_3.46.0 cli_2.3.1 later_1.1.0.1
[41] htmltools_0.5.1.1 tools_4.0.4 igraph_1.2.6 gtable_0.3.0 glue_1.4.2
[46] RANN_2.6.1 reshape2_1.4.4 tinytex_0.30 Rcpp_1.0.6 spatstat_1.64-1
[51] scattermore_0.7 cellranger_1.1.0 vctrs_0.3.6 nlme_3.1-152 lmtest_0.9-38
[56] xfun_0.21 globals_0.14.0 rvest_1.0.0 mime_0.10 miniUI_0.1.1.1
[61] lifecycle_1.0.0 irlba_2.3.3 goftest_1.2-2 jrc_0.4.0 MASS_7.3-53.1
[66] zoo_1.8-8 scales_1.1.1 hms_1.0.0 promises_1.2.0.1 spatstat.utils_2.0-0
[71] parallel_4.0.4 yaml_2.2.1 reticulate_1.18 pbapply_1.4-3 rpart_4.1-15
[76] stringi_1.5.3 rlang_0.4.10 pkgconfig_2.0.3 matrixStats_0.58.0 evaluate_0.14
[81] lattice_0.20-41 ROCR_1.0-11 tensor_1.5 labeling_0.4.2 htmlwidgets_1.5.3
[86] cowplot_1.1.1 tidyselect_1.1.0 parallelly_1.24.0 RcppAnnoy_0.0.18 magrittr_2.0.1
[91] R6_2.5.0 generics_0.1.0 DBI_1.1.1 withr_2.4.1 pillar_1.5.1
[96] haven_2.3.1 mgcv_1.8-34 fitdistrplus_1.1-3 survival_3.2-7 abind_1.4-5
[101] future.apply_1.7.0 modelr_0.1.8 crayon_1.4.1 KernSmooth_2.23-18 utf8_1.1.4
[106] plotly_4.9.3 rmarkdown_2.7 readxl_1.3.1 grid_4.0.4 data.table_1.14.0
[111] reprex_1.0.0 digest_0.6.27 xtable_1.8-4 httpuv_1.5.5 R.utils_2.10.1
[116] munsell_0.5.0 viridisLite_0.3.0

[HenrikBengtsson]

Here are a few suggestions:

1. Convince yourself that you are actually running N workers, e.g.

> library(future.apply)
> plan(multisession, workers = 4)
> future_sapply(1:nbrOfWorkers(), FUN = function(ii) c(worker = ii, pid = Sys.getpid()))
        [,1]  [,2]  [,3]  [,4]
worker     1     2     3     4
pid    24542 24541 24539 24540

You should see an equal number of unique process IDs as you have workers.

2. The fact that you bump up future.globals.maxSize to ~16 GiB suggests that you're working with a very large data set. If that is the case, note that you are exporting 16 GiB of data to each parallel worker, which takes time.

3. You can try with plan(multicore), which is forked processing and uses shared memory instead exporting. However, I'm not sure Seurat supports that (=has been validated to be stable with forked processing).

4. Ask on the Seurat forums whether your timings are reasonable.

BTW, I assume your 1:3 loop is to get the average time across three identical runs.

[izahn]

The future package is almost like magic, but it won't magically make your for-loop parallel. For that you need https://dofuture.futureverse.org/ and some small modifications (untested) to your code:

library(doFuture)
library(Seurat)

registerDoFuture()
plan(multisession, workers=6)

start_time <- Sys.time()
markers <- foreach (i = 1:3, .export = "CD3") %dopar% {
   FindMarkers(CD3, only.pos = TRUE, ident.1 = i, min.pct = 0.25, logfc.threshold = 0.25, return.thresh = 0.05, verbose = FALSE)
}
end_time <- Sys.time()
end_time-start_time