PaperlayerView exact revision on arXiv

Source-linked AI summary

A Unifying Framework for Parallel and Distributed Processing in R using Futures

Henrik Bengtsson

arXiv:2008.00553v4cs.DCstat.CO

TL;DR

R parallel programming lacks a unified way to separate computational intent from backend selection while handling common parallel-processing concerns. This paper presents the Future API and ecosystem as a minimal abstraction for concurrent R evaluation and backend-independent map-reduce programming. The framework supports backend flexibility, lowers implementation and maintenance burden, and addresses several limitations of existing approaches, while remaining constrained by non-transferable process-bound objects.

  • Problem

    Existing R parallel frameworks can expose backend-specific behavior, require substantial testing, and force developers to manage framework choices and object transfer constraints.

  • Method

    The paper develops a minimal Future API that decouples expression evaluation from value assignment and separates parallel frontends from interchangeable backends.

  • Results

    The future framework enables existing coding styles and local parallel code to use different computational resources through backend changes while reducing the burden of developing and maintaining parallel code.

  • Takeaways & Limitations

    End-users can choose parallel backends while developers focus on what to parallelize, with third-party backends supported through the Future API.

  • Takeaways & Limitations

    Objects tied to the current R process, including connections and many external-pointer objects, cannot generally be transferred unchanged to parallel workers.

Abstract

from arXiv · show

A future is a programming construct designed for concurrent and asynchronous evaluation of code, making it particularly useful for parallel processing. The future package implements the Future API for programming with futures in R. This minimal API provides sufficient constructs for implementing parallel versions of well-established, high-level map-reduce APIs. The future ecosystem supports exception handling, output and condition relaying, parallel random number generation, and automatic identification of globals lowering the threshold to parallelize code. The Future API bridges parallel frontends with parallel backends following the philosophy that end-users are the ones who choose the parallel backend while the developer focuses on what to parallelize. A variety of backends exist and third-party contributions meeting the specifications, which ensure that the same code works on all backends, are automatically supported. The future framework solves several problems not addressed by other parallel frameworks in R.

Introduction

R offers multiple parallel-processing frameworks, but developers must often choose and accommodate backend-specific behavior early. The future framework proposes a minimal API that separates what to parallelize from how and where it runs.

  • Introduction: The future package aims to provide a unifying, generic, minimal API for common R parallel-processing patterns, especially manager-worker computation.The framework targets R-level parallelization rather than native-code parallelization.
  • Existing parallel frameworks: R supports parallel processing through frameworks including parallel, legacy multicore and snow functions, and foreach-based backends.The parallel package provides mclapply() and parLapply(), while foreach separates its map-reduce construct from backend registration.
  • Existing parallel frameworks: The parallel package's mclapply() and parLapply() functions parallelize lapply()-like workloads using forked processes or background worker clusters.Forking automatically inherits the main process workspace, whereas SOCK clusters require explicit export of packages and global objects.
  • Mixed responsibilities of developers or end-users: Developers choosing a parallel framework early can limit end-user control and may face expensive rewrites when supporting different processing strategies.Hard-coded cluster types can leave end-users with little control, and packages may use conditional statements with limited testing.
  • Map-reduce parallelization with more control for the end-user: Foreach separates backend and core-count specification from its map-reduce construct, allowing end-users to choose how and where computation runs.Third-party adaptors can add new backend types without updating the foreach package itself.
  • Map-reduce parallelization with more control for the end-user: Foreach adaptors lack an exact behavioral specification, so the same code can require backend-specific options, testing, and handling of globals and packages.Different adaptors may produce runtime errors or different results when their behaviors diverge.

The future framework

The Future API separates future creation from value collection, providing minimal constructs for asynchronous and parallel R evaluation. Its backend-independent design lets developers specify what to parallelize while end-users choose how and where futures resolve.

  • The Future API: The Future API provides atomic constructs for creating, collecting, and checking futures, which support higher-level parallel map-reduce functions.The core constructs are future(expr), value(f), and resolved(f).
  • The Future API: Decoupling expression evaluation from value assignment lets other work occur between creating a future and collecting its result.A future records the expression and required objects at creation, so later reassignment of x does not change the future's value.
  • The Future API: Backend choice determines how futures resolve, from sequential execution to parallel processing through SOCK clusters or forked R processes.value() waits for resolution, while future() can block when all configured workers are busy.
  • The Future API: The framework preserves results across backend choices by assigning developers responsibility for code and end-users responsibility for the execution plan.The stated goal is identical results whether futures resolve sequentially locally or in parallel on a remote cluster.
  • The Future API: The same Future API constructs can express parallel loops and lapply()-style computations, with plan() controlling the degree of parallelization.The examples create one future per element and collect their values afterward; higher-level packages provide less verbose alternatives.
  • The Future API: resolved() enables non-blocking collection of completed futures, potentially lowering latency when results are large or transferred over limited-bandwidth networks.Collecting only resolved values can allow additional futures to launch sooner.

Exception handling

Futures relay evaluation errors to the main R process when value() is called, preserving standard error-handling behavior. Failures caused by worker termination or communication problems are distinguished as FutureError conditions.

  • Exception handling: Errors raised while evaluating a future are captured and relayed as-is when value() retrieves its result.This mirrors the behavior of equivalent code executed without futures.
  • Exception handling: Creating one future per element can be suboptimal when future-creation overhead is large relative to evaluation time.The paper identifies chunking elements as a way to mitigate this overhead, although it requires more complex code.
  • Exception handling: Standard R condition-handling mechanisms can assign a missing value when a future evaluation produces an error.The same handling approach applies to errors relayed by futures.
  • Exception handling: Worker termination and failed communication produce FutureError conditions distinct from ordinary evaluation errors.These errors can be handled specifically, for example by restarting workers or relaunching a failed future elsewhere.

Relaying of standard output and conditions (e.g., messages and warnings)

Futures capture standard output and conditions on workers and relay them in the main R process, including messages, warnings, and errors. This preserves debugging and logging behavior across local and remote backends, while standard error remains unreliable.

  • Relaying of standard output and conditions (e.g., messages and warnings): Futures relay standard output, messages, warnings, and errors when value() is called, preserving their captured order except where immediate conditions are handled specially.The mechanism supports existing debugging and logging practices during parallel execution.
  • Relaying of standard output and conditions (e.g., messages and warnings): Relayed output and conditions can be captured with standard R tools such as capture.output(), with messages and warnings handled through condition handlers.The same behavior applies regardless of whether the backend is local or remote.
  • Relaying of standard output and conditions (e.g., messages and warnings): Immediate conditions may be relayed before value() is called, making them suitable for progress updates.This early relay does not preserve the ordering of other captured condition types.
  • Relaying of standard output and conditions (e.g., messages and warnings): Standard error cannot be captured reliably and is silently ignored, whereas message() output is relayed because it uses captured message conditions.This limitation follows from restrictions in R's standard-error handling.

Globals and packages

The future framework automatically identifies globals required by future expressions, reducing the setup needed for parallel execution. When static inspection misses an indirect dependency, developers can explicitly declare the global.

  • Globals and packages: future() uses static code inspection to identify, locate, and record global variables, functions, and package namespaces needed by a future expression.This usually removes the need for developers to manage globals manually.
  • Globals and packages: Indirect access such as get("k") can evade automatic global detection and cause an object-not-found error when the future resolves.The example shows k defined before the future but unavailable inside it.
  • Globals and packages: Developers can guide global detection by explicitly mentioning a missing global in the future expression or passing it through the globals argument.The package documents additional options for controlling included globals and ignoring false positives.
  • Globals and packages: The framework walks the future expression's abstract syntax tree using an optimistic search that tolerates false positives to reduce false negatives.False negatives are emphasized because they can produce resolution errors.

Proper parallel random number generation

The future ecosystem supports reproducible parallel random-number generation across backends, while future assignments provide asynchronous, promise-based evaluation with convenient syntax and explicit limitations.

  • Proper parallel random number generation: The future ecosystem uses L’Ecuyer-CMRG to produce statistically sound and reproducible random numbers across future backends.This supports analyses such as bootstrap, permutation tests, and simulation studies.
  • Proper parallel random number generation: With seed = TRUE, parallel RNG streams remain fully reproducible regardless of the selected backend or available worker count.The default is seed = FALSE because enabling reproducible streams can introduce significant overhead.
  • Future assignments: The %<-% operator offers a convenient future-assignment syntax, with infix operators for arguments such as seed = TRUE.It is designed to mimic ordinary <- assignment.
  • Future assignments: A future assignment evaluates its right-hand expression asynchronously and assigns the result only when the promise is accessed.Access can block while value() resolves the underlying future, after which the variable becomes a regular value.
  • Future assignments: Using plan(multisession), multiple future assignments can be processed in parallel.The example launches three future assignments that are resolved concurrently.
  • Future assignments: Future assignments cannot assign promises to lists because promises can only be assigned to environments.The listenv package provides list-like environments as a workaround.

Nested parallelism and protection against it

Nested parallelism can unintentionally multiply worker usage across package layers, so the future framework defaults nested workers to sequential execution unless explicitly configured.

  • Nested parallelism and protection against it: Nested parallelism can overload CPU cores when separately parallelized package layers each use all available cores.A package update that adds internal parallelism can create N^2 workers when called from another parallel package.
  • Nested parallelism and protection against it: The future package protects against nested parallelism by configuring each worker to run sequentially unless nesting is explicitly requested.It uses options and environment variables such as options(mc.cores = 1).
  • Nested parallelism and protection against it: Nested parallelism can be configured by end-users through plan().Different worker counts can be specified for successive parallelization layers.
  • Nested parallelism and protection against it: At most 2 × 3 = 6 tasks run in parallel when two workers are configured for the first layer and three for the second.Further nesting beyond these two layers is processed sequentially.
  • Nested parallelism and protection against it: On clusters, availableCores() lets multisession workers respect the number of cores assigned to each scheduled job.This supports nested configurations combining a scheduler backend with local workers.

Future backends

The future package supplies backends ranging from sequential and local parallel execution to forked and cluster-based processing, allowing the same Future API to target different execution environments.

  • Future backends: The future package implements backends based on the parallel package in addition to defining the Future API.These backends determine how futures are resolved while preserving the common interface.
  • Future backends: The default backend resolves all futures sequentially in the current R session.Users can change the plan to select parallel execution.
  • Future backends: SOCK-based backends resolve futures in parallel on a local machine.The package provides a plan for local socket-based processing.
  • Future backends: Forked-processing backends provide another way to resolve futures in parallel on the local machine.This is distinct from socket-based multisession processing.
  • Future backends: The cluster backend resolves futures through traditional SOCK or MPI “snow” clusters.Users can also provide an explicitly created cluster, such as one with four workers.
  • Future backends: Remote SOCK workers can be created on named machines when password-less SSH access and installed R are available.Reverse tunneling avoids inward-facing firewall port-forwarding that requires administrative rights.

Third-party future backends

Third-party packages extend the Future API with local process, HPC scheduler, and cloud backends, provided they conform to its specifications.

  • Third-party future backends: Future.callr resolves futures in parallel on local machines using R processes orchestrated by callr.It provides a local alternative to the built-in process-based backend.
  • Third-party future backends: Third-party backends can be used as alternatives to built-in backends when they conform to the Future API specifications.Conformance allows them to participate in the same future-based programming model.
  • Third-party future backends: Future.batchtools submits futures as jobs to HPC schedulers including Slurm, SGE, and Torque/PBS.This supports distributed future execution in high-performance computing environments.
  • Third-party future backends: googleComputeEngineR provides a “snow” cluster type for resolving futures in the Google Compute Engine cloud.The cited package also supports parLapply() functions.
  • Third-party future backends: The callr backend avoids socket connections and the PSOCK limit of 125 parallel workers.This distinction can also avoid certain Windows firewall requirements associated with PSOCK connections.

Implementation

The future ecosystem emphasizes cross-platform validation, backend conformance, and practical handling of portability and performance constraints. Its implementation supports reproducible parallel execution while documenting object-transfer limitations and backend-dependent overhead.

  • Validation: Validation spans package tests across operating systems and R versions, continuous integration services, and R-hub.The core packages are tested across Linux, Solaris, macOS, and MS Windows.
  • Validation: Reverse-dependency checks and adapted foreach examples provide additional validation across CRAN and Bioconductor packages and multiple future backends.As of November 2021, future was tested against 210 direct reverse-package dependencies.
  • Validation: Future API conformance tests require every backend to produce correct and reproducible results across compliant implementations.New backends must pass the future.tests suite before conforming to the API.
  • Portability limitations: Objects such as connections, database connections, XML documents, and Stan models may not transfer as-is to external R processes.Connections can cause runtime errors, invalid results, or writes to the wrong file; external-pointer scanning can warn about some risky globals.
  • Performance considerations: Parallelization introduces process, communication, global-identification, error-handling, and output-relaying overhead that can outweigh its benefits.Backend choice depends on workload requirements: forked processing suits low latency, whereas distributed processing suits large throughput.

Results

The Future API separates parallel code from backend choice while standardizing essential orchestration tasks across sequential and parallel backends. This enables portable higher-level APIs, broader resource support, and lower development and maintenance costs.

  • The Future API standardizes global identification, package handling, parallel RNG, output, and condition relaying for higher-level parallel frontends.These common tasks need not be reimplemented by each frontend or backend.
  • New backends can provide consistent behavior without reimplementing orchestration tasks, allowing developers to focus on what to parallelize and end-users to choose where.The separation supports alternatives ranging from local execution to cloud and HPC resources.
  • With zero code modifications, local parallel code could run across thousands of HPC cores by changing the future plan.This illustrates how backend selection can be changed independently of application code.
  • Existing sequential map-reduce frameworks can be ported to parallel execution, supporting futurized variants of apply, purrr, and foreach interfaces.The Future API reduces the need to expose low-level parallelization code in higher-level APIs.
  • Futures increase the chance that packages support multi-host execution, including cloud and HPC environments, instead of locking users to local multicore execution.The framework addresses portability across different computational resources.
  • Future-based parallel code lowers development, testing, and maintenance costs while protecting multi-tenant systems through nested-parallelism safeguards and adaptable core settings.The framework and backend packages centralize much of the orchestration and testing burden.

Future work

Future work focuses on reducing duplicated infrastructure around future-based map-reduce APIs and extending futures with suspension, restarting, serialization, resource requirements, and broader object support.

  • Map-reduce infrastructure: future.mapreduce would centralize load balancing, global handling, and parallel random-number generation across future-based map-reduce packages.This would reduce duplicated implementations and make it easier to build additional map-reduce APIs on futures.
  • Map-reduce infrastructure: A longer-term redesign could integrate map-reduce support into the core future framework while preserving backward compatibility.The proposed refactoring aims to reduce reliance on thin wrapper packages.
  • Map-reduce infrastructure: Future-based map-reduce APIs can partition work into one future per worker, reducing overhead compared with one future per element.The described merge approach groups ten element-level futures into two worker-level futures before collecting results.
  • Map-reduce infrastructure: Helper syntax could reduce future setup verbosity and lower the maintenance burden of adding future support to existing map-reduce APIs.This would allow APIs to reuse futures without requiring separate parallel implementations.
  • Future capabilities: Planned extensions include suspending and restarting futures, fuller serialization, marshaling selected objects, and specifying resource requirements for individual futures.Examples include preserving read-only file connections through marshaling and selecting environments by R version, mounts, or forbidden backends.
  • Future capabilities: The framework currently only partially supports future serialization, and not all object types can be exported unchanged during parallel processing.The roadmap also notes that backend support for suspension and restarting may vary.

Summary

The future package offers a lightweight alternative for parallel processing in R through the Future API. Its basic functions support richer higher-level APIs that closely mimic familiar map-reduce styles.

  • Summary: The future package implements a lightweight Future API with three basic functions from which richer parallel-processing APIs can be constructed.Several higher-level APIs closely mimic counterpart map-reduce APIs, allowing developers to retain familiar coding styles.
Loading 2008.00553v4…