Ray: A Distributed Framework for Emerging AI Applications

TL;DR

The next wave of AI doesn’t just train a model once and serve predictions — it runs huge numbers of short, irregular computations (simulations, rollouts, model updates) that depend on each other in ways you can’t predict ahead of time. Existing big-data systems (Spark, MapReduce) and ML systems (TensorFlow, MPI) either can’t hit the throughput/latency, or they assume the computation graph is fixed in advance. Ray’s key idea is to support dynamic task graphs with two complementary abstractions — stateless tasks (remote functions returning futures) and stateful actors (long-lived objects whose methods run remotely) — backed by a logically centralized but physically sharded control store that makes every other component stateless and restartable, plus a bottom-up scheduler that decides locally first and only escalates to a global scheduler when needed. The headline result: linear scaling past 1.8 million tasks/second with sub-millisecond task latency, transparent fault recovery via lineage re-execution, and RL algorithms (Evolution Strategies, PPO) that match or beat hand-built systems while changing ~7 lines of serial code. Ray became the de-facto framework behind RLlib, modern RLHF pipelines, and a lot of today’s agent infra.

Problem & Motivation

The pain in one sentence: reinforcement-learning-style and agentic workloads generate millions of short, wildly heterogeneous, interdependent tasks whose shape isn’t known until runtime — and no existing cluster framework can schedule that at the required throughput and latency.

Walk through a concrete RL training loop to feel it. An agent interacts with an environment: it rollouts (plays a trajectory by repeatedly calling policy.compute(state) then env.step(action)), collects rewards, and periodically calls policy.update(trajectories). To do this at scale you want hundreds of rollouts running in parallel, each finishing at a different time, and you want to update the policy as soon as enough rollouts come back — not after all of them. That creates three demands that break older systems:

• Heterogeneity. Tasks differ in function (sensor processing vs. policy network vs. physics sim), duration (a game lost in 3 moves vs. won in 300), and resource type (policy inference wants a GPU; the simulator wants a CPU). A realistic run may do hundreds of millions of simulations.
• Dynamic, evolving task graphs. You can’t pre-declare the DAG. Which rollouts feed which policy update is decided on the fly, based on which ones finish first.
• Hard performance targets. Millions of tasks/second, millisecond latencies — because a robot has to act in real time, and to act well it may run more simulations right now.

Why prior tools fall short:

• Spark / MapReduce / Dryad / Dask / CIEL — built for coarse-grained, batch, Bulk-Synchronous-Parallel (BSP) workloads where all tasks in a stage do the same thing and take about the same time. They don’t deliver the throughput or the sub-ms latency, and BSP’s “wait for the whole stage” model is exactly wrong for ragged rollouts.
• TensorFlow / Naiad / MPI / Canary — assume a static computation graph fixed before execution. Great for training one network; useless when the graph mutates at runtime.

So in 2017 every serious RL group hand-rolled an ad-hoc distributed system per algorithm. Ray’s bet: provide one general framework so that scaling an algorithm is a decorator, not a six-month systems project.

What’s New (Core Contribution)

1. A dual programming model on one execution engine. Before: you picked either a task-parallel dataflow system (CIEL — stateless tasks only) or an actor system (Orleans — stateful actors only). Now: Ray gives you both on top of a single dynamic task graph — stateless tasks for cheap, idempotent, easily-recoverable parallelism, and stateful actors to wrap things that have state (a simulator, a GPU-resident model, a third-party black box). The unification trick is the stateful edge in the task graph (explained below).
2. Logically centralized control state via a sharded Global Control Store (GCS). Before: the scheduler/metadata lived inside stateful master components that became scaling bottlenecks and single points of failure. Now: all control state (task specs, function code, the computation graph, object locations, scheduling events) lives in a sharded, replicated key-value store. That makes every other component stateless — so schedulers, object stores, and workers can be replicated, restarted, and scaled horizontally, and fault tolerance becomes “restart and re-read the GCS.”
3. A bottom-up, two-level scheduler. Before: centralized schedulers (Spark, CIEL) are simple but cap throughput; hierarchical/parallel schedulers assume static graphs or independent jobs. Now: tasks go to the local scheduler first; it handles them itself unless overloaded / lacking resources / needing remote inputs, in which case it forwards to a replicated global scheduler. A single threshold knob slides the whole system from fully decentralized to fully centralized.
4. Unified lineage-based fault tolerance for stateless and stateful work. Before: dataflow systems recovered stateless tasks by re-execution but had no clean story for stateful operators. Now: by baking stateful edges into the lineage graph, the same “walk backward and replay” reconstruction recovers lost actor state too — with optional checkpointing to bound replay length.

How It Works (Technically)

Ray has two layers: an application layer (Driver, Workers, Actors — what your code touches) and a system layer (Global Control Store, distributed scheduler, distributed object store — the plumbing). The genius is in how the system layer makes the application layer’s dynamic, heterogeneous graph fast.

The computation model: tasks, actors, and three kinds of edges

Everything Ray executes is a node in a dynamic task graph that the system builds as your program runs. There are two node types and three edge types:

• Data objects (immutable values — a scalar, a numpy array, a policy) and tasks (a remote function invocation).
• Data edges: if task T outputs object D, draw T → D; if D is an input to T, draw D → T. These capture dependencies and let the system fire a task automatically the moment all its inputs exist.
• Control edges: if task T1 invokes task T2 (nested remote functions), draw T1 → T2. This is what lets the driver not be a bottleneck — any worker can spawn more tasks.
• Stateful edges: the clever bit. An actor is a stateful process whose methods run serially. Method invocations are nodes too, but to encode that method M_j ran after M_i on the same actor (and therefore depends on the mutated internal state), Ray adds a M_i → M_j stateful edge. All of an actor’s calls form a chain.

Why stateful edges matter: they let Ray embed stateful actors inside an otherwise stateless dataflow graph. The internal state of an actor is an implicit dependency between successive calls; making it an explicit edge means the same lineage-reconstruction machinery works for actors — to rebuild a lost actor output, walk backward along data and stateful edges to a node whose inputs all still exist, then replay that subgraph (re-instantiate the actor, replay its method chain in order).

The API (the whole surface is five calls)

• futures = f.remote(args) — run f asynchronously; returns futures immediately (non-blocking). Futures can be passed straight into other .remote() calls, which is how you express the dependency graph without ever blocking.
• vals = ray.get(futures) — block until results are ready and fetch them.
• done = ray.wait(futures, k, timeout) — return as soon as k of the futures finish (or timeout). This is the call that makes RL natural: update the policy when enough rollouts return, instead of waiting for the slowest.
• actor = Class.remote(args) — instantiate a stateful actor.
• futures = actor.method.remote(args) — call a method on it (non-blocking).

Two design rules make fault tolerance cheap: remote functions are stateless, side-effect-free, and operate on immutable objects → they’re idempotent → re-executing them on failure is always safe. Resource requirements are declared per-function (@ray.remote(num_gpus=2)) so the scheduler can place GPU work on GPU boxes and CPU work on cheap CPU boxes.

Architecture & data flow

flowchart TB
  subgraph APP["Application Layer"]
    D[Driver: your program]
    W[Workers: stateless, run tasks]
    A[Actors: stateful, run methods serially]
  end
  subgraph SYS["System Layer"]
    GCS[("Global Control Store<br/>sharded + replicated<br/>task specs · fn code · graph · object locations")]
    LS[Local Scheduler<br/>one per node]
    GS[Global Scheduler<br/>replicated]
    OS[Distributed Object Store<br/>shared memory · Apache Arrow · zero-copy]
  end
  D -->|f.remote / actor.method.remote| LS
  W -->|nested tasks| LS
  LS -->|"overloaded? remote inputs? lacks GPU?"| GS
  GS -->|"place using load + input locations"| LS
  LS --> W
  LS --> A
  W <-->|read/write objects| OS
  A <-->|read/write objects| OS
  LS -. heartbeat 100ms .-> GCS
  GS <-. load + object metadata .-> GCS
  OS <-. object locations .-> GCS
  D <-->|ray.get| OS

Schematic of the bottom-up scheduler: tasks born on a node try to run locally; only when the local queue exceeds a threshold do they spill to the global scheduler, which rebalances across the cluster. Drag the threshold to see the system slide from fully-local to fully-centralized. (Illustrative, not the paper's measured numbers.)

Tracing one task end-to-end

Take add.remote(a, b) where a lives on node N1, b on node N2:

1. add was auto-registered with the GCS and pushed to every worker at startup.
2. Driver submits add(a,b) to N1’s local scheduler (step 1).
3. Local scheduler can’t run it well (input b is remote), so it forwards to the global scheduler (step 2).
4. Global scheduler asks the GCS where the args are (step 3), sees b is on N2, and schedules the task on N2 (move compute to data) (step 4).
5. N2’s local scheduler checks its object store — has b, missing a (step 5) — looks up a in the GCS (step 6), and N2’s object store replicates a locally (step 7).
6. All inputs now local → invoke add on an N2 worker (step 8), which reads args via shared memory, zero-copy (step 9).
7. For ray.get(id_c) on N1: N1 checks its store (miss), registers a callback on the GCS Object Table; when N2 finishes and writes c, the GCS fires the callback, N1 pulls c, and ray.get returns.

Most real tasks skip the global scheduler entirely (run locally) and the GCS replies are cached — so the RPC count in the common case is small.

The system-layer performance tricks

• GCS = Redis, one per shard, sharded by object/task ID, hot replica per shard. Pseudo-random IDs make load balancing across shards trivial; replication gives fault tolerance with <10% overhead even when GCS is the artificial bottleneck.
• Object store = single-threaded event loop over shared memory + Apache Arrow. Pre-allocated memory-mapped file pool, SIMD-style copies, parallel content hashing (to detect nondeterminism), zero-copy reads for same-node tasks. Objects are immutable — no consistency protocol needed. Each object fits on one node (distributed objects are an app-level concern, built as collections of futures). Objects live in memory; LRU eviction to disk.
• Schedulers = single-threaded event loops that cache local object metadata and fire tasks the instant their inputs land. Heartbeats every 100ms carry queue length + resource availability to the global scheduler via GCS pub/sub.

The algorithm, simplified

The “one central idea” is the dynamic dependency graph driving non-blocking execution plus the ray.wait rollout loop. Here is the RL training loop the paper centers on, in Ray:

import ray
ray.init()

@ray.remote
def create_policy():
    return random_policy()                  # immutable object; returns a future

@ray.remote(num_gpus=2)                      # this task needs GPUs; scheduler places it accordingly
def update_policy(policy, *rollouts):
    return improved(policy, rollouts)        # stateless, idempotent -> safe to re-execute on failure

@ray.remote                                  # an ACTOR: stateful, wraps a third-party simulator
class Simulator:
    def __init__(self):
        self.env = Environment()             # internal state lives across method calls
    def rollout(self, policy, num_steps=100):
        obs, state = [], self.env.current_state()
        for _ in range(num_steps):
            action = compute(policy, state)  # policy inference
            state = self.env.step(action)    # advance the (stateful) env
            obs.append(state)
        return obs                           # a trajectory

def train_policy():
    policy = create_policy.remote()                      # future, no blocking
    sims = [Simulator.remote() for _ in range(10)]       # 10 stateful actors
    for _ in range(100):
        # launch one rollout per actor; each returns a future immediately
        pending = [s.rollout.remote(policy) for s in sims]
        # KEY: update as soon as ENOUGH rollouts return, not all of them
        ready, pending = ray.wait(pending, num_returns=4)
        policy = update_policy.remote(policy, *ready)     # graph grows dynamically here
    return ray.get(policy)

Note what never appears: no manual serialization, no socket code, no “which node runs this” logic. The decorators + futures are the distribution. ray.wait(..., num_returns=4) is the line that turns a synchronous BSP loop into a ragged, latency-tolerant pipeline.

Fault tolerance, concretely

On node failure, the monitor marks lost tasks/objects in the GCS. To rebuild a lost object, Ray walks backward along data + stateful edges until it hits inputs that still exist, then replays that subgraph. For stateless tasks this is just re-execution (safe because idempotent). For actors, replay re-instantiates the actor and re-runs its method chain — bounded in practice because simulators are short-lived, and checkpointing caps replay (paper shows: with checkpointing, 500 re-executions and 60s stall vs. 10K re-executions and 120s without). The GCS itself survives via shard replication.

Built on Prior Work

Results & Evidence

What they measured and the headline numbers:

• Throughput: near-perfect linear scaling — >1M tasks/s at 60 nodes, >1.8M tasks/s at 100 nodes, 100M tasks processed in 54 seconds. Latency for short remote tasks is sub-millisecond.
• Object store: single client hits >15 GB/s write throughput for large objects, 18K IOPS (~56µs/op) for small ones, on a 16-core box.
• Fault tolerance: transparent recovery from killed worker nodes mid-run (tasks stall during reconstruction, then throughput fully recovers); actor recovery via lineage replay, with checkpointing cutting stall ~2x. GCS replication overhead <10% in worst case, undetectable in normal workloads.
• RL workloads: Evolution Strategies on Ray scaled to 8192 cores (the reference special-purpose system died at 1024), ran in 3.7 min median (>2x faster than best published 10 min), and required changing 7 lines of serial code. PPO on Ray beat an optimized MPI implementation using fewer GPUs, partly because Ray’s per-task resource specs let CPU-only work run on cheap CPU boxes — a 4.5x cost reduction.

What the evidence does and doesn’t establish:

• Does: Ray genuinely delivers fine-grained throughput + low latency at scale, and the programming model is dramatically simpler for RL than bespoke systems. The fault-tolerance story is real and demonstrated.
• Doesn’t: This is a 2017 systems paper; many benchmarks are embarrassingly parallel (the throughput graph) or microbenchmarks. The RL comparisons are favorable but against a small set of baselines the authors chose. Actor recovery cost can be high for long-lived actors with big state (acknowledged). Single-node object size limit pushes complexity up to the app layer. And “millions of tasks/sec” assumes tasks are mostly local — adversarial graphs that force every task through the global scheduler would not scale the same way (the threshold knob is doing a lot of work).

How You’d Use It

For an AI services shop, Ray is the horizontal-scaling substrate you reach for when a Python workload outgrows one machine but isn’t a clean “train one big model” job — which is exactly the shape of agentic and RL work.

• Multi-agent orchestration at scale. Your MAS (ARC-style) maps cleanly: each agent is a Ray actor holding its own state/memory/conversation; tool calls and sub-tasks are stateless remote functions; ray.wait lets an orchestrator proceed as soon as the fastest k agents respond instead of blocking on the slowest. You get fault tolerance and cross-node scaling almost for free.
• Parallel LLM inference / batch jobs. Fan out thousands of prompt evaluations, document extractions, or evals as tasks; place embedding/inference on GPU actors and pre/post-processing on CPU tasks — the per-task resource spec is the cost lever (the 4.5x savings in the paper is the same trick).
• RL / RLHF / agent fine-tuning. Ray (via RLlib and the broader ecosystem that grew out of this paper) is the standard backbone for rollout collection + policy updates. If a client wants to train an agent with RL, you build on Ray rather than hand-rolling a cluster.
• Stateful pipelines. Anything that wraps a stateful black box (a simulator, a browser session, a per-tenant model server) is an actor; everything stateless around it is tasks. The stateful-edge design means even these recover from node death.

Build-vs-buy read: Ray is open source and mature — you don’t build this, you use it. The value you sell is the architecture on top: knowing what becomes an actor vs. a task, where the resource specs go, and how to keep tasks local for throughput.

Build Your Own (Minimal Recipe)

You won’t reimplement Ray, but to internalize it (and to build a credible “scaled agent platform” offering), build the 80% toy:

Components, in build order:

1. A futures-based remote-call layer. A decorator that, instead of running f, ships (fn_id, args) to a worker pool and returns a future. Back it with Python concurrent.futures or a multiprocessing pool to start; the contract (non-blocking call returning a future you can pass to other calls) is the lesson.
2. A control store. A single Redis instance holding: registered function code/ids, a task table, and an object-location table. This is your GCS. Sharding/replication is the hard-mode upgrade — skip it for the toy.
3. An object store. A shared dict keyed by object-id → value, plus location entries in the control store. Real Ray uses shared memory + Arrow; your toy can use Redis or a local dict per node.
4. A two-level scheduler. Local queue per worker; if len(queue) > threshold, forward to a “global” scheduler that picks the least-loaded worker (read load from the control store). This single threshold is the whole centralized↔decentralized spectrum.
5. Lineage + reconstruction. Record task → output_object and input_object → task in the control store. On a simulated worker death, walk backward to live inputs and re-run. Add stateful edges to extend this to actors.

The 1–2 genuinely hard parts: (a) the scheduler/object-store coordination — moving compute to data, replicating remote inputs before execution, and caching metadata so the common case avoids the global path; (b) zero-copy data sharing — real performance comes from shared memory + a serialization format like Arrow, which is fiddly. For the toy, fake these; in production, use real Ray.

Reach for: real ray (pip install ray) for anything beyond a teaching exercise; redis for the control store; pyarrow if you want to feel the zero-copy story.

How to Improve It

1. Smarter, data-locality-aware global scheduling. The paper’s threshold + least-loaded placement is simple. A learned or cost-model-based scheduler that predicts task duration and input-transfer cost could cut tail latency — testable against the paper’s load-balancing benchmark.
2. Cheaper actor recovery. Long-lived actors with big state replay slowly. Ideas: incremental/differential checkpointing, user annotations for read-only methods (the authors flag this), or copy-on-write actor snapshots. Measure stall time vs. checkpoint frequency.
3. First-class distributed objects. “Each object fits on one node” pushes large-tensor sharding to the app layer. Native support (object = collection of futures with a partition map) would simplify large-model and large-matrix workloads and is directly testable on big-array benchmarks.
4. Adaptive ray.wait policies for agents. For MAS orchestration, a fixed k is crude. A policy that picks k based on marginal value of waiting (e.g., stop once the returned rollouts/agents are “good enough”) would raise throughput on heterogeneous-latency agent fleets.
5. Scheduling that understands GPU memory and model residency. Per-task resource specs are coarse (num_gpus=1). Modern LLM serving needs awareness of which model is already loaded where (KV-cache/weight residency) to avoid reloads — a fertile place to push the resource model.

Glossary

• Task (remote function) — a stateless, side-effect-free function invoked with .remote(); returns a future; safe to re-execute on failure.
• Actor — a stateful process instantiated with .remote() whose methods run serially and can mutate internal state across calls.
• Future — a placeholder for a result that isn’t ready yet; returned instantly by .remote(), resolved by ray.get, and passable into other tasks to express dependencies.
• Dynamic task graph — the dependency DAG Ray builds at runtime as tasks spawn tasks; not declared up front.
• Data / control / stateful edge — graph edges for input-output dependencies, nested-call relationships, and actor method ordering respectively.
• Lineage — the recorded history of how each object was produced; replaying it reconstructs lost data.
• Global Control Store (GCS) — sharded, replicated key-value store holding all system control state, making every other component stateless.
• Bottom-up / two-level scheduler — tasks try the per-node local scheduler first and escalate to a replicated global scheduler only when needed.
• Object store — in-memory, shared-memory, zero-copy store for immutable task inputs/outputs (built on Apache Arrow).
• BSP (Bulk Synchronous Parallel) — the “all tasks in a stage do the same work and finish together” model behind Spark/MapReduce; a poor fit for ragged RL rollouts.
• Apache Arrow — a columnar in-memory data format enabling efficient zero-copy serialization.
• Evolution Strategies (ES) / PPO — RL algorithms used as benchmarks; ES is a gradient-free population method, PPO is a policy-gradient method.
• Idempotent — re-running it produces the same result; the property that makes stateless-task re-execution safe.
• Zero-copy — reading data directly from shared memory without duplicating it, key to Ray’s low latency.