Emulating mclapply(mc.preschedule = TRUE) with future.apply::future_lapply()

[dcnorris]

The relationship between the scheduling described in parallel::mclapply and that of future.apply::future_lapply isn't entirely clear to me. Do the concepts map 1-to-1, or does the futureverse introduce a higher level of abstraction? How might I replicate the mc.preschedule=TRUE behavior in the futureverse?

mc.preschedule: if set to ‘TRUE’ then the computation is first divided
          to (at most) as many jobs are there are cores and then the
          jobs are started, each job possibly covering more than one
          value.  If set to ‘FALSE’ then one job is forked for each
          value of ‘X’.  The former is better for short computations or
          large number of values in ‘X’, the latter is better for jobs
          that have high variance of completion time and not too many
          values of ‘X’ compared to ‘mc.cores’.

The future.scheduling and future.chunk.size parameters seem not to give me the same level of control over the number of worker threads forked under plan(multicore), nor over the way they are assigned to the jobs.

[HenrikBengtsson]

mc.preschedule=TRUE and future.scheduling=TRUE are both the defaults, but here is an illustration showing that both distribute the 100 "tasks" evenly to nworkers workers:

> nworkers <- 4
> table(unlist(parallel::mclapply(1:100, FUN=function(i) Sys.getpid(), mc.preschedule=TRUE, mc.cores=nworkers)))

5089 5090 5091 5092 
  25   25   25   25 

> future::plan("multicore", workers=nworkers)
> table(unlist(future.apply::future_lapply(1:100, FUN=function(i) Sys.getpid(), future.scheduling=TRUE)))

5093 5094 5095 5096 
  25   25   25   25 

So, "one-to-one".

[dcnorris]

(Confirming your examples work for me, but behavior in my application differs as described below.)

[dcnorris]

In my own application, I'm getting different behavior. Here's a relevant section of code, inside a function that takes a ... param:

                   prog <- progressor(along = ppe)
                   parallelize <- ifelse(any(grepl("mc.", names(list(...))))
                                       , mclapply
                                       , future_lapply)
                   cpe_parts <- parallelize(ppe, function(ppe_) {
                     prog()
                     path_m <- matrix(ppe_, nrow=2, dimnames=list(c("D","T")))
                     ## Let's be sure to skip the stopped paths, tho!
                     if (any(path_m["D",] <= 0, na.rm=TRUE))
                       return(list(ppe_))
                     level <- factor(path_m["D",1:unroll], levels=1:self$max_dose())
                     tox <- path_m["T",1:unroll]
                     enr <- cohort_sizes[1:unroll]
                     n <- as.vector(xtabs(enr ~ level))
                     x <- as.vector(xtabs(tox ~ level))
                     paths.(n, x, unroll+1, path_m, cohort_sizes)
                   }
                   , ...)
                   cpe <- do.call(c, cpe_parts)

Here's what happens when I run under each parallelizer:

> parallax(C=13, mc.cores=4)
   C     J elapsed
1 13 11571   3.036
      pid cached evals skips calc.ms us/calc
 1: 58143   2478  2478  6969    1008     407
 2: 58144   1021  1021  3089     430     422
 3: 58145    437   437  1099     182     418
 4: 58146    365   365   496     153     421
 5: 58143   2694  2694  7278    1077     400
 6: 58144   1096  1096  3170     459     419
 7: 58145    542   542  1159     240     444
 8: 58146    407   407   511     169     417
 9: 58143   2702  2702  7282    1079     399
10: 58144   1227  1227  3411     524     427
11: 58145    557   557  1165     246     443
12: 58146    426   426   513     177     417
13: 58143   2717  2717  7288    1081     398
14: 58144   1246  1246  3413     533     428
15: 58145    565   565  1169     248     440
16: 58146    429   429   513     178     417
17: 58143   2720  2720  7288    1082     398
18: 58144   1249  1249  3413     534     428
19: 58146    532   532   620     233     439
20: 58143   2739  2739  7290    1092     399
21: 58144   1252  1252  3413     535     428
      pid cached evals skips calc.ms us/calc
> parallax(C=13) # with empty ..., will default to future.apply::future_lapply
   C     J elapsed
1 13 11571   6.385
      pid cached evals skips calc.ms us/calc
 1: 58147   2478  2478  6969    1052     424
 2: 58147   3427  3427 10055    1392     406
 3: 58147   3761  3761 11182    1509     401
 4: 58147   3996  3996 11733    1603     401
 5: 58147   4061  4061 12193    1625     400
 6: 58147   4124  4124 12286    1644     398
 7: 58147   4187  4187 12388    1664     397
 8: 58147   4196  4196 12436    1669     397
 9: 58147   4204  4204 12440    1673     397
10: 58147   4299  4299 12717    1708     397
11: 58147   4314  4314 12723    1714     397
12: 58147   4330  4330 12728    1719     396
13: 58147   4330  4330 12749    1719     396
14: 58148     73    73    23      16     219
15: 58148     81    81    27      19     234
16: 58148     84    84    27      20     238
17: 58148     85    85    29      20     235
18: 58148     86    86    31      21     244
19: 58149    157   157   128      68     433
20: 58149    161   161   145      69     428
21: 58149    164   164   145      70     426
      pid cached evals skips calc.ms us/calc
> 

Of course, the progress bars don't work as desired under mclapply; this is what motivated me to investigate the futureverse.
BTW, the evals are costly numerical integrations which are memoized so that any given pid will not have to redo them. Each forked process here tallies up its own work effort, and returns tally attached to the chunked answer.

[HenrikBengtsson]

And what do you get if you run with (i) plan(sequential) and (ii) plan(multisession, workers=4)?

[dcnorris]

> future::plan("multicore", workers=4)
> parallax(C=13)
   C     J elapsed
1 13 11571   6.098
      pid cached evals skips calc.ms us/calc
 1: 59121   2478  2478  6969    1056     426
 2: 59121   3427  3427 10055    1390     405
 3: 59121   3761  3761 11182    1502     399
 4: 59121   3996  3996 11733    1588     397
 5: 59121   4061  4061 12193    1611     396
 6: 59121   4124  4124 12286    1638     397
 7: 59121   4187  4187 12388    1662     396
 8: 59121   4196  4196 12436    1665     396
 9: 59121   4204  4204 12440    1668     396
10: 59121   4299  4299 12717    1687     392
11: 59121   4314  4314 12723    1699     393
12: 59121   4330  4330 12728    1702     393
13: 59121   4330  4330 12749    1702     393
14: 59122     73    73    23      21     287
15: 59122     81    81    27      23     283
16: 59122     84    84    27      25     297
17: 59122     85    85    29      26     305
18: 59122     86    86    31      26     302
19: 59123    157   157   128      72     458
20: 59123    161   161   145      73     453
21: 59123    164   164   145      74     451
      pid cached evals skips calc.ms us/calc
> future::plan("sequential")
> parallax(C=13)
   C     J elapsed
1 13 11571   3.461
      pid cached evals skips calc.ms us/calc
 1: 59103   2478  2478  6969     886     357
 2: 59103   3427  3427 10055    1218     355
 3: 59103   3761  3761 11182    1337     355
 4: 59103   3996  3996 11733    1423     356
 5: 59103   4061  4061 12193    1440     354
 6: 59103   4124  4124 12286    1460     354
 7: 59103   4187  4187 12388    1483     354
 8: 59103   4196  4196 12436    1487     354
 9: 59103   4204  4204 12440    1491     354
10: 59103   4299  4299 12717    1537     357
11: 59103   4314  4314 12723    1540     356
12: 59103   4330  4330 12728    1544     356
13: 59103   4330  4330 12749    1544     356
14: 59103   4330  4330 12770    1544     356
15: 59103   4338  4338 12774    1546     356
16: 59103   4341  4341 12774    1547     356
17: 59103   4342  4342 12776    1548     356
18: 59103   4343  4343 12778    1549     356
19: 59103   4446  4446 12885    1580     355
20: 59103   4447  4447 12905    1581     355
21: 59103   4450  4450 12905    1582     355
      pid cached evals skips calc.ms us/calc
> future::plan("multisession", workers=4)
> parallax(C=13)
   C     J elapsed
1 13 11571  37.899
      pid cached evals skips calc.ms us/calc
 1: 59151   2478  2478  6969     904     364
 2: 59151   3427  3427 10055    1227     358
 3: 59151   3761  3761 11182    1338     355
 4: 59151   3996  3996 11733    1420     355
 5: 59151   4061  4061 12193    1437     353
 6: 59151   4124  4124 12286    1453     352
 7: 59151   4187  4187 12388    1472     351
 8: 59151   4196  4196 12436    1475     351
 9: 59151   4204  4204 12440    1476     351
10: 59151   4299  4299 12717    1509     351
11: 59151   4314  4314 12723    1514     350
12: 59151   4330  4330 12728    1517     350
13: 59151   4330  4330 12749    1517     350
14: 59152     73    73    23      18     260
15: 59152     81    81    27      21     271
16: 59152     84    84    27      21     261
17: 59152     85    85    29      21     258
18: 59152     86    86    31      21     255
19: 59150    157   157   128      46     299
20: 59150    161   161   145      48     304
21: 59150    164   164   145      49     304
      pid cached evals skips calc.ms us/calc
Warning message:
In .Internal(gc(verbose, reset, full)) :
  closing unused connection 3 (localhost)
> 

(As you can see, the memoization is so important in these computations that running this sequentially—with a unified cache—is nearly as fast as running in parallel across multiple threads that don't share their caches.)

[dcnorris]

I should say that there is likely a strong gradient in problem size across X, such that the earlier elements are more costly to run CPE on. When I use the ordering="random" option, I get work assignments a lot more like those of mclapply:

> future::plan("multicore", workers=4)
> parallax(C=13, future.scheduling=structure(TRUE, ordering="random"))
   C     J elapsed
1 13 11571   3.939
      pid cached evals skips calc.ms us/calc
 1: 59319   2536  2536  6992    1020     402
 2: 59318   1163  1163  3157     521     447
 3: 59321    853   853  2219     373     437
 4: 59321    365   365   496     147     402
 5: 59321    467   467   967     209     447
 6: 59321    530   530  1060     236     445
 7: 59318    159   159    81      56     352
 8: 59319     96    96    36      31     322
 9: 59319    104   104    40      35     336
10: 59320    188   188   259      87     462
11: 59318    196   196    89      69     352
12: 59321    384   384   498     154     401
13: 59321    399   399   507     160     401
14: 59321    530   530  1081     236     445
15: 59319    112   112    44      40     357
16: 59321    387   387   498     155     400
17: 59318    181   181    83      64     353
18: 59320    191   191   259      87     455
19: 59319   2639  2639  7099    1051     398
20: 59318    178   178    83      63     353
21: 59321    402   402   507     161     400
      pid cached evals skips calc.ms us/calc
> 

[dcnorris]

I'm starting to think that the issue is how the work is chunked. mcapply(mc.preschedule = TRUE) does this in an interleaved fashion, which tends to spread the work out more evenly when X is ordered by difficulty. (Indeed, I realized today that the tail end of my X may consist of jobs so trivial that they don't contribute to the tabulations above.)

[dcnorris]

I believe I've solved this issue with this little function used like this. This 'transposed' work assignment might make sense as a default, for people transitioning from mclapply.