Beyond Sequential: Scaling Existing Medical Pipelines with ‘futurize’

- Easy!

Henrik Bengtsson

University of California, San Francisco
R Foundation, R Consortium
@HenrikBengtsson

Medical research is powered by trusted R packages

boot	Bootstrap resampling for robust confidence intervals
lme4	Mixed-effects models for longitudinal patient data
survival	Time-to-event analysis for clinical endpoints
DESeq2	Differential expression in RNA-seq data
scater	Single-cell RNA-seq analysis, e.g. PCA, t-SNE, UMAP
…	…

R is very effective for medical and scientific research, mostly due to its robust ecosystem. We rely on large number of well-tested R packages like ‘survival’, ‘LME4’ and ‘boot’. These packages provide us with validated implementations of statistical methods that have been refined and peer-reviewed over several years. This means that whenever we use them, we are using tools that are recognized and trusted by the scientific community. This foundation of trust is essential for reproducible research and science in general.

Packages are distributed via the highly-trusted CRAN and Bioconductor repositories:

With supporting repositories such as R-universe, Pharmaverse, and R-multiverse:

I would like to point out that I do not think we would be here today if it wouldn’t be for package repositories like CRAN and Bioconductor. We trust these ecosystems because they have been proven for thousands of hours of real world use. However, this established trust creates a constraint for us today when we’re talking paralisation. Because when we need to scale up our computations that are currently running sequentially, we should not have to rewrite or modify our code that we already have validated just to achieve better performance because major rewrite comes with major risk. So instead, the goal should be to increase performance with a minimum amount of changes to our already existing code.

Growing datasets make sequential analysis a bottleneck

• Bootstrapping confidence intervals - 1,000 replicates × large cohort = hours
• Mixed models across patient subgroups - dozens of subgroup fits, one after another
• Simulation studies for sample-size planning - 10,000 iterations on a laptop overnight

← 32× CPU cores sitting idle

Parallelization can help →

Barrier is friction!

Now, when we work with large data sets and they grow in size, our regular sequential analysis becomes a significant bottleneck. For example if you run a thousand bootstrap replicates on a large clinical cohort that can take several hours. Now, if you monitor your computer during these tasks here is our images from H top you might find that most of the cores are idle. As most of us know, paralisation is the solution to this problem. However, there is a major barrier here. The friction to change the code can sometimes be so large so we just give up and keep running our code sequentially.

Parallelizing used to require abandoning existing code

Before: sequential, readable, auditable

res <- lapply(patients, fit_model)

Refactoring tax

• Many parallelization APIs
• Complicated to test
• Expensive to maintain

After: parallel, hard to read, hard to maintain

library(parallel)

if (parallel) {
  if (.Platform$OS.type == "windows") {
    cl <- makeCluster(8)
    res <- parLapply(cl, patients, fit_model)
    stopCluster(cl)
  } else {
    options(mc.cores = 8)
    res <- mclapply(patients, fit_model)
  }
} else {
  res <- lapply(patients, fit_model)
}

Let’s look at how paralisation in our used to be in the past. So consider a very simple sequential code where we apply a function fit model to a set of patients and we use the lapply() function in base R for this. That code is very readable, it’s easy to understand, someone can validate it and understand what’s going on. If you have a lot of patients, so the fifth model takes a long time. We are typically looking into parallelization. Now, traditionally in order to do that, we could use the parallel package. And when we use that and we do code that we want to share with collaborators, some are on Windows, some on Linux, some on Mac, or we typically end up with code like to the right, if we want to run parallel, the first thing we do is like are we running on Windows, then we set up a cluster of workers in this case eight parallel workers and then we call the par lapply() function. And when done we’ll have to remember to stop the cluster of workers. Alternatively, if we’re not on Windows, we can use the mclapply() function which use fork parallelism and that’s case we said it it we allows it to run eight workers and we parallelize. And if you don’t want to run in parallel, we can fall back to regular lapply() function.

Futureverse was introduced to lower refactoring tax

Before: sequential, readable, auditable

After: parallel, readable, maintainable

res <- lapply(patients, fit_model)

library(future.apply)
plan(multisession)
res <- future_lapply(patients, fit_model)

So that was the type of code I used to write to maintain functions that anyone could work on Windows, Mac, Linux, small and big computers. But that became unmanageable and it was really hard to test. So I scratched that in 2015. I started to work on what is today the Futureverse ecosystem. I started out by providing a unified interface for parallelization. That was the goal. And the goal was to provide researchers and other art developers to separate what to parallelize and how to parallelize. So, instead of writing platform-specific code, you could just define a future backend, the plan() function and say I want to use the ‘multisession’ for local parallelization and then you just pick the corresponding “future” function that mimics what you used to work. So for example, if you used to use lapply(), you could use the future_lapply() version that is available from the ‘future.apply’ package. And we can see to the right how the code looks like after palestation.

library(purrr)

res <- map(patients, fit_model)

library(furrr)
plan(multisession)
res <- future_map(patients, fit_model)

And, a little bit later, because there were a lot of people used the ‘purrr’ package and the map() functions there. David Vaughan, he created a ‘furrr’ package and the corresponding future_map(). And, similarly it was very easy to parallelize ‘purrr’ functions using the ‘furrr’ package.

library(foreach)

res <- foreach(p = patients) %do% {
  fit_model(p) 
}

library(doFuture)
plan(multisession)
res <- foreach(p = patients) %dofuture% {
  fit_model(p) 
}

And as many of you know, there is a this fantastic package called ‘foreach’ by Steve Weston. It was created back in 2009 and it’s been very popular for parallelization. It provides the foreach() map-reduce construct, where the sequential version is where you call foreach() and you apply the ‘%do%’ operator. It looks like a for loop, it’s not a for loop, but it’s more similar to lapply() and map(). But, the big thing and the big pickup with ‘foreach’ is that you can also apply it to parallelize. So, in the ‘future’ ecosystem, there is the ‘%dofuture%’ package that allows you to replace the ‘%do%’ operator with the ‘%dofuture%’ operator. And now that will scale out on the futureverse backend. in this case also the ‘multisession’.

The R community has embraced the Futureverse

+30% reverse dependencies yearly

Top 0.7% most downloaded

Just to share the uptake from the our community since the releases of these packages starting in 2015. We have the number of packages that depend on the core package called ‘future’ is now over 500 packages and it grows by 30% on average every year. And similarly these packages are among the top 0.7% most downloaded packages on CRAN. So, this growth demonstrates there was a need for something simpler. It was also we can also see this is there is a need for parallelization because data sets keep growing. So I expect more packages will use parallelization going forward.

… but, we can simplify it further with ‘futurize’

All you need to remember is futurize()

So, as I said before, one of my objectives with the Futureverse ecosystem is to minimize the friction of parallelization as much as possible. Ideally a researcher should be able to parallelize their code without modifying a single function name or restructuring the code. The goal is to take code that has already been validated, or maybe even peer-reviewed, and enable parallel execution with a single transparent step. So, these functions and packages are already shown, they do a great job getting towards that, but I would say we’re not there yet. They’re not perfect. We can simplify it even more. And that’s the purpose of the ‘futurize’ package. It acts as a single adapter that allows you to maintain your original code by simply piping to the futurize() function. That’s all. It removes most of the needs for learning a new API and rewriting existing code.

New package futurize preserves your original code

• Universal adapter: One unifying function futurize()
• Zero rewrites: Original logic unchanged

res <- lapply(patients, fit_model) |>
         futurize()

So, the new package ‘futurize’ helps you preserve your original code. It has a universal adapter, one unifying function futurize(), that’s all you need to know. It’s helps you do zero rewrite, the original logic is unchanged. All you have to do is pipe your lapply() call into the futurize() function.

library(purrr)
res <- map(patients, fit_model) |>
         futurize()

Or if you use ‘purrr’, you can take for instance the map() function as seen here and pipe it into futurize().

library(foreach)
res <- foreach(p = patients) %do% {
  fit_model(p)
} |> futurize()

Or if you’re a person that prefers the foreach() construct, you can just pipe that into futurize().

library(plyr)
res <- llply(patients, fit_model) |>
         futurize()

It doesn’t stop there. If you’re a person who prefers the ‘plyr’ package, you can take the llply() function and pipe it into futurize().

library(BiocParallel)
res <- bplapply(patients, fit_model) |>
         futurize()

If you’re coming from the Bioconductor world and are used to use ‘BiocParallel’, you can use its functions like bplapply() and just pipe it into futurize(), if you like that.

library(crossmap)
res <- xmap(x, ~ .y * .x) |>
         futurize()

Other map-reduce-like packages are also supported like. Here is an example with a ‘crossmap’ function that takes the xmap() call and pipe it into futurize().

Easy!

So, this allows researchers and other programmers to parallelize legacy or establish code with minimal intervention which therefore comes with minimal risks for bugs.

Same code scales from laptop to cloud to HPC

Without changing any code, you can switch from local and remote parallel processing, to large-scale high-performance compute (HPC) processing:

plan()	Environment	Use case
sequential	Single machine	Sequential (default, debugging)
multisession	Single machine	Parallel across multiple cores
mirai_multisession	Single machine	Same as above; powered by mirai
cluster	Many machines	Parallel across many machines (desktops, cloud)
batchtools_*	Slurm/SGE/LSF	Scheduler-based HPC clusters

Like everything else in Futureverse, the futurize() function can take full advantage of all the parallel backends available in the ecosystem. You just specify the execution plan at the beginning of your script or your code.

The default is ‘sequential’ processing and that can be useful for local debugging.

The next step is typically to parallelize on your local computer and take advantage of all the cores and that’s you can use the built-in ‘multisession’ backend. There is also the ‘mirai_multisession’, which provides an alternative to ‘multisession’ with less latency and it comes from the ‘mirai’ package.

You can also scale out to computers, local computers, remote computers by using the ‘cluster’ backend. All you need to have is SSH access and R running on them.

Some of us have access to high-performance-compute clusters, where you submit jobs to a queue which then where the jobs are then distributed out to thousands of cores and that’s done by different job schedulers like Slurm, SGE, LSF, Torque, PBS, and so on.

library(futurize)
plan(future.batchtools::batchtools_slurm)
res <- patients |> purrr::map(fit_model) |> futurize()

Here I added an example where we have a set of patients that we pipe to the purrr::map() function to process for each patient with it a model and if we have a lot of patients we can then pipe it to futurize() so it is parallelized. And in this setup here I use the ‘batchtools_slurm’ backend, meaning the parallelization will take each of this task, each patient will be a job on the job scheduler and the scheduler can then distribute it to thousands of cores.

… we can do even more with ‘futurize’

All you need to remember is futurize()

Now, we can do even more with the ‘futurize’ package. Remember, all you need to know is the futurize() function.

futurize() works with a growing set of domain-specific packages

CRAN Package	Use
boot	Bootstrap resampling, confidence intervals
caret	Classification and regression training
fwb	Bootstrap resampling, confidence intervals
gamlss	Generalized additive models (GAMLSS)
glmnet	Lasso and elastic-net regularization
glmmTMB	Generalized linear mixed models (GLMMs)
kernelshap	Kernel SHAP (Shapley Additive Explanations)
lme4	Linear and non-linear mixed-effects models
metafor	Meta-analysis models
mgcv	Generalized additive models (GAMs)
partykit	Recursive partitioning (trees)
riskRegression	Risk regression for survival analysis
seriation	Data ordering (seriation)
stars	Spatiotemporal data cubes
structchange	Testing for structural changes
tm	Text mining
vegan	Community ecology

Bioconductor Package	Use
BiocParallel	Map-reduce and parallel infrastructure
DESeq2	Differential gene expression analysis
GenomicAlignments	Genomic alignments (BAM/CRAM)
GSVA	Gene set variation analysis
Rsamtools	Binary alignment (BAM) and tabix utilities
scater	Single-cell transformations
scuttle	Single-cell analysis utilities
SingleCellExperiment	Single-cell data containers
sva	Surrogate variable analysis

‘futurize’ is not limited to simple map-produce patterns, as we’ve seen thus far. It is also designed to intercept and parallelize internal loops and map-reduce calls within other R packages, provided those packages expose the necessary settings for doing so. This can be very beneficial for specialized domains because many of those packages have multiprocess support, but they often use their own unique implementation logic. So futurize() serves as common interface with those variety of implementations and arguments and standardize them to simplify it for you as an end-user and developer.

Bootstrap simulations accelerated with futurize()

Sequential:

library(boot)
b <- boot(data = cohort, statistic = cox_stat, R = 100e3)

100,000 bootstrap replicates takes hours on large cohorts!

A single worker

Here is an example using the ‘boot’ package. The ‘boot’ package provides ways to bootstrap, and here is an example where we run 100,000 bootstrap samples to calculate the Cox-proportional hazard model on a cohort of data. Here we do this sequentially so this can take several hours if the cohort is very large.

Parallel:

plan(future.mirai::mirai_multisession)
library(boot)
b <- boot(data = cohort, statistic = cox_stat, R = 100e3) |>
     futurize()

Faster when distributed across parallel workers.
Identical results.

32 parallel workers

And if we run this on a 32-core machine, as we see to the right, that machine is sitting idle most of the time. But, we can just pipe this boot() call to futurize() and if we set it up to parallelize on the local machine, here again I’m using the mirai_multisession() to illustrate how to use that to parallelize on the 32 core machines. We can see we utilize the machine to 100% and this will finish much faster. And I also would like to add that futurize() and the ‘future’ framework is designed so you get identical call results even here when random number generation is included and used.

Traditional parallelization is more cumbersome and less robust

library(boot)

library(parallel)
cl <- makeCluster(32)

b <- boot(data = cohort, statistic = cox_stat, R = 100e3,
          parallel = "snow", ncpus = length(cl), cl = cl)

stopCluster(cl)

• Parallelization arguments blur the bootstrapping logic
• All three parallelization arguments must be specified
• Does not interrupt nicely
• Does not handle crashed parallel workers
• Not easy to scale to cloud or HPC job schedulers

And here is what it would take to parallelize the boot() function using traditional parallelization mechanism from the ‘parallel’ package. You need to set up a cluster, you need to make sure to stop it. You need to pass the correct arguments, and we see this blurs the whole original code logic. It doesn’t interrupt as nicely as the futurize() function do. It doesn’t handle crashed parallel workers, it’s not easy to scale to cloud or to HPC job schedulers.

Yes, we can do progress reporting too

All you need to remember is progressify()

Now, when we parallelize, we typically run things for a very long time, because that’s why we parallelize in the first time. And in those scenarios we would like to know how far things have progressed while it’s running. And when we do parallelization, that’s typically lost, because parallel updates and progress updates are really hard in parallelization. And that’s where the ‘progressify’ package comes in and the progressify() function.

Progress reporting with ‘progressify’

Vanilla call:

res <- lapply(patients, fit_model)

Let’s consider our basic lapply() example, where we take a set of patients and we fit a model on each of them. When we call this, there is zero progress updates. So, this can run for hours and be silent until the very end.

With progress reporting:

library(progressify)
handlers("cli", globals = TRUE)

res <- lapply(patients, fit_model) |> 
         progressify()
         

■■■■■■■■■■■■■■■■80% | ETA: 12m

And here is how you can get progress reporting from that long-running, sequential lapply() call. You just pipe it to progressify() function.

In order to see something, we need to set a handler, and here I’m using the “cli” handler from the ‘cli’ package on CRAN. When that’s set up, we will get a nice progress bar in the terminal.

You can also get audio reporting notification in your desktop, and so on.

In parallel with progress reporting:

library(progressify)
handlers("cli", globals = TRUE)

library(futurize)
plan(multisession)

res <- lapply(patients, fit_model) |> 
         progressify() |> futurize()

■■■■■■■■■■■■■■■■80% | ETA: 23s

Now, progressify() works very similar to futurize(). They’re both so-called transpiler. I won’t go into the details, but you can mix and match them. So here is an example how we can get the lapply() call, get progress updates from it, and then run it in parallel through the futurize() function. And here I’m parallelizing on the local computer.

Easy!

Easy!

Structured concurrency allows for automatic optimization

Because futurize() limits the life-span of the parallel tasks, it can:

• cancel remaining parallel tasks
  • if there is an error
  • if the user or the operating system requests an interrupt
• estimate efficiency of parallelization
  • is it worth it?
  • suggest a better parallel backend
• optimize distribution of objects to parallel workers
  • by chunking
  • by remote caching
  • via shared memory (e.g. new mori package by C. Gao 2026)
• be agile to resource specifications
  • memory, run-time, GPU, …

Before I wrap up this presentation, I would like to say that the ‘futurize’ package implements a principle known as structured concurrency. This is because it managed the lifecycle of each parallel tasks and, because of that, it can provide quite advanced features automatically. For example, if a worker process encounters an error or gets interrupted, futurize() can immediately cancel the remaining parallel workers and tasks running, which prevents resources from being wasted. It can handle interrupts gracefully.

But looking ahead we can also imagine that it estimates the efficiency of a parallelization. It can try to run things sequentially and in parallel on different backends so it can conclude if it’s worth parallelizing the call. It can suggest a better parallel backend than the one you’re using.

It can also optimize the distribution of objects to parallel workers. We already do this with chunking, but you can imagine remote caching, or by using shared memory, which some of you might have heard of, e.g. the ‘mori’ package by Charlie Gao that just came out. I also want to point out that Bioconductor have had the ‘SharedObject’ package, which was a little bit more complicated to use, but they have been using it for several years with this spirit.

We can also allow to implement features that are on the roadmap, where you can specify resources that you need, like the amount of memory, the amount of runtime, if you need a GPU, and so on.

So with that, I want to say even though we’ve done and achieved a lot thus far, there is a lot more on the roadmap that will improve things and you do do not have to do anything. Your code using these functions will perform better over the next generations of package releases.

Go compute and may the future be with you!

Easy to install:

install.packages(c("futurize", "progressify"))

Easy to use:

ys <- lapply(xs, fcn) |> progressify() |> futurize()

Stay with your favorite coding style:

ys <- xs |> map(fcn) |> progressify() |> futurize()

Available elsewhere too:

ys <- glmnet::cv.glmnet(x, y) |> futurize()

https://www.futureverse.org

And with that, I want to say thank you for your attention. Go compute and may the future be with you. Thank you!