Work is not distributed 'on the fly'

[frederikziebell]

Consider the following example where all except the first work package take the same amount of time.

library("tidyverse")
library("future")
library("future.apply")

future::plan("multisession", workers=4)
num_workers <- future::nbrOfWorkers()

f <- function(i){
  if(i==1){
    Sys.sleep(10)
  } else {
    Sys.sleep(1)
  }
  
  data.frame(
    i=i,
    pid=Sys.getpid(),
    timepoint=Sys.time()
  )
}

future_lapply(1:(5*num_workers), f)  %>% 
  bind_rows() %>% 
  mutate(worker_nr=map_dbl(pid, ~which(.x==unique(pid)))) %>% 
  arrange(timepoint)

The result is that all workers except the one evaluating f(1) finish almost simultaneously, while worker nr. 1 lags behind and it's remaining work is not distributed while all other workers are idle:

Is this an issue or is there a way to specify that unfinished work should be distributed to all available workers? I observed the phenomenon with the future.apply and furrr package, so I think it's directly related to future.

[hans-ekbrand]

EDIT: perhaps this is an issue in doFuture rather than in future?

I have a similar use case and came here to report a new issue, when I found it had already been reported.

The problem you describe is well known, the technical term is "load balancing", see for example the manual for doFuture, section "Load balancing".

When running under OSes that support forking, you will achieve optimal load balancing like this:

foreach(...,
             .options.future = list(scheduling = FALSE)
             )

Under MS Windows, which does not support forking, scheduling = FALSE is still allowed, but it performs bad whenever the computation time for each job is low compared the computation time required to start R, since for each job, a new R process must be started from scratch. If your jobs are few and takes long time, this is still a possibility for you, if you are on an OS that does not support forking.

In my use case, I have many jobs that are relatively fast (relative to the time of starting R) and one job that is really slow. On linux, this is not a problem, I just sort the jobs by estimated computation time, and use scheduling = FALSE. But for portable code, I am a package developer, this does not work. In addition to having different plans, I also have to use different scheduling, and as far as I know, there is no way to get efficient load balancing with plan(multisession) when the computation time of the jobs is smaller than the computation time of starting a new R session.

I don't know the inner workings of future, but I think it should be possible to have persistent workers and efficient load balancing. As I understand the situation (I hope I am wrong though), scheduling = FALSE implies that the workers are discarded when they have finished a single job. Looking forward to suggestions to workaround this.

Kind regards,
Hans Ekbrand

[HenrikBengtsson]

... since for each job, a new R process must be started from scratch.

Actually, this is not correct. If you use plan(multisession, workers = 2), there will be two background workers running in the background processing futures until you switch to another plan. You can see this by the following example:

library(doFuture)
registerDoFuture()
plan(multisession, workers = 2)
y <- foreach(x = 1:6, .options.future = list(scheduling = FALSE)) %dopar% { c(x=x, pid=Sys.getpid()) }
do.call(rbind, y)
#      x   pid
# [1,] 1 28332
# [2,] 2 28363
# [3,] 3 28332
# [4,] 4 28363
# [5,] 5 28332
# [6,] 6 28363

Note how there are only two process IDs: 28332 and 28363. If there would have a new worker started each time, then there would have been six different.

[hans-ekbrand]

... since for each job, a new R process must be started from scratch.

Actually, this is not correct. If you use plan(multisession, workers = 2), there will be two background workers running in the background processing futures until you switch to another plan. You can see this by the following example:

library(doFuture)
registerDoFuture()
plan(multisession, workers = 2)
y <- foreach(x = 1:6, .options.future = list(scheduling = FALSE)) %dopar% { c(x=x, pid=Sys.getpid()) }
do.call(rbind, y)
#      x   pid
# [1,] 1 28332
# [2,] 2 28363
# [3,] 3 28332
# [4,] 4 28363
# [5,] 5 28332
# [6,] 6 28363

Note how there are only two process IDs: 28332 and 28363. If there would have a new worker started each time, then there would have been six different.

Oh, that's great! Then scheduling=FALSE should do exactly what I want. Thanks for your rapid response!