Distribute jobs to cores as long as they become free

[elgabbas]

Hello,

I want to use the future.lapply function to run many jobs (e.g. 1000) and I know that these jobs vary in complexity and time-consumed. I sorted them in an estimated descending order and need to force future.lapply to distribute the most complex jobs to cores, then the next complex jobs, etc. This will balance the processing time. I need to implement something like:

Cores (8 cores available for parallel processing)
 ┌───────────┬───────────┬───────────┬───────────┬───────────┬───────────┬───────────┬───────────┐
 │  Core 1   │  Core 2   │  Core 3   │  Core 4   │  Core 5   │  Core 6   │  Core 7   │  Core 8   │
 ├───────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┤
 │ Job 1     │ Job 2     │ Job 3     │ Job 4     │ Job 5     │ Job 6     │ Job 7     │ Job 8     │
 │ Job 9     │ Job 10    │ Job 11    │ Job 12    │ Job 13    │ Job 14    │ Job 15    │ Job 16    │
 │ Job 17    │ Job 18    │ Job 19    │ Job 20    │ Job 21    │ Job 22    │ Job 23    │ Job 24    │
 │   ...     │   ...     │   ...     │   ...     │   ...     │   ...     │   ...     │   ...     │
 └───────────┴───────────┴───────────┴───────────┴───────────┴───────────┴───────────┴───────────┘
 … continues until all 1000 jobs are assigned to cores.

I use something similar to this

library(dplyr)
library(future.apply)
job_list <- sort(1:(8*2), decreasing = TRUE)
plan(multisession, workers = 8)

results <- future_lapply(
  job_list,
  function(job) {
    T1 <- Sys.time()
    Sys.sleep(job)
    tibble::tibble(job = job, T1 = T1,  jobID = Sys.getpid())
  },
  future.chunk.size = 1) %>% 
  dplyr::bind_rows()

results

This forces the most complex 8 jobs to be executed first, then the next 8, etc.

results
# A tibble: 16 × 3
job T1                  jobID
<int> <dttm>              <int>
  1    16 2024-11-03 15:39:06 26080
2    15 2024-11-03 15:39:06 28060
3    14 2024-11-03 15:39:06 17420
4    13 2024-11-03 15:39:07  4608
5    12 2024-11-03 15:39:07 23916
6    11 2024-11-03 15:39:07 18744
7    10 2024-11-03 15:39:07 16224
8     9 2024-11-03 15:39:07  2436
9     8 2024-11-03 15:39:17  2436
10     7 2024-11-03 15:39:18 16224
11     6 2024-11-03 15:39:19 23916
12     5 2024-11-03 15:39:20 18744
13     4 2024-11-03 15:39:20  4608
14     3 2024-11-03 15:39:21 17420
15     2 2024-11-03 15:39:23 26080
16     1 2024-11-03 15:39:23 28060

I expect Process ID 26080 to process jobs 16 and 9, not 16 and 1. How can I ensure that the jobs are distributed in the order they are processed? Using future.scheduling = Inf instead of future.chunk.size = 1 leads to the same results.

A related question is that I need to allow jobs to start in the desired order (i.e. most complex jobs first) but distribute jobs to cores as soon as they are finished with previous job. When I monitor the current implementation, I find that some cores may become idle by the end of the processing, while many left jobs are waiting to be done on only a couple of cores, which influence the total time consumed by the job. Is this achievable without affecting the overall performance of the task?

Thanks