Ray

Part I: The mental model

your driver          ray.init()   one Python process orchestrates the cluster
      |
      |  f.remote(x)  ->  ObjectRef      a distributed future, returns at once
      v
core worker          C++ in every process, reached from Python via _raylet.pyx
      |              owns the returned object, does distributed refcounting
      |  request a worker lease that fits the task's resources
      v
raylet (per node)    node manager (scheduler) + object manager (shared-mem store)
      |              grants a local lease or spills the request to another raylet
      v
worker process       runs f, stores the result in its local object store
      |
      v
ray.get(ref)         blocks, then reads the value (zero-copy for numpy buffers)

GCS (control plane)  nodes, actors, placement groups, runtime envs, a kv store

The one-sentence identity. Ray is a distributed futures system with two primitives, stateless tasks and stateful actors, whose results live as immutable objects in a shared-memory store, and whose lifetimes are tracked by ownership-based reference counting rather than by a central master. You write nearly ordinary Python. You add @ray.remote to a function or a class, you call it with .remote(...) instead of calling it directly, and you get back an ObjectRef, a handle to a value that may not exist yet and may be computed on another machine. Nothing blocks until you ask for the value with ray.get. That single shift, from calling to submitting, is the whole programming model.

Two load-bearing ideas make it work at scale. The first is the object store. Every result and every large argument lives once per node in a shared-memory segment, so many workers on the same machine read the same bytes with no copy, and moving data between nodes is a deliberate pull rather than a hidden serialization on every call. The second is the single-controller shape. Unlike the SPMD world of torchrun and torchtitan, where every rank runs the same script in lockstep, a Ray program has one driver that dispatches heterogeneous work to many workers and actors as plain Python control flow. That is why Ray is comfortable orchestrating a reinforcement-learning loop that juggles several distinct models at once, and it is the reason the RLHF ecosystem landed on it. This chapter is verified against a recent Ray 2.x main branch in July 2026. Ray is a large, fast-moving repository, so where an exact API or path is likely to have shifted I say so and stay at the level of role and concept.

Part II: Using it

Ray runs on Linux, macOS, and Windows, on CPUs and GPUs, from a laptop to a thousand-node cluster with the same API. The install is a wheel, and the extras decide which libraries come along.

# core plus the dashboard, cluster launcher, and Ray Jobs
pip install "ray[default]"

# pull in the ML libraries you need
pip install "ray[data,train,serve,tune,rllib]"

A first session should be local. ray.init() with no arguments starts a single-node cluster inside your process, and the two primitives are a decorator away. Tasks are stateless remote functions. Actors are remote classes that keep state between calls.

import ray
ray.init()  # start a local cluster; use address="auto" to attach to a running one

@ray.remote
def square(x):
    return x * x

# submit eight tasks; they run in parallel across your cores
refs = [square.remote(i) for i in range(8)]
print(ray.get(refs))          # [0, 1, 4, 9, 16, 25, 36, 49]

@ray.remote
class Counter:
    def __init__(self):
        self.n = 0
    def inc(self):
        self.n += 1
        return self.n

c = Counter.remote()                       # spawn a long-lived actor process
print(ray.get([c.inc.remote() for _ in range(3)]))   # [1, 2, 3]

Resources are requested declaratively, and Ray schedules around them. @ray.remote(num_gpus=1) asks for a whole GPU and sets CUDA_VISIBLE_DEVICES so the worker sees only its assigned device. num_gpus=0.5 packs two actors onto one GPU, which matters a great deal for RLHF colocation later. num_cpus, memory, and arbitrary custom resources (resources={"accelerator_type:H100": 1}) work the same way. One thing to internalize early is that these numbers are logical accounting for the scheduler, not enforced isolation. Ray will not stop a task that claimed one CPU from using four. The reservation is a promise you make to the bin-packer.

Passing large data around is where ray.put earns its keep. Putting an object into the store once and passing its ObjectRef to many tasks means the bytes are shared, not re-sent per call.

import numpy as np
big = np.zeros((10_000, 10_000))     # ~800 MB
ref = ray.put(big)                   # into the object store once, zero-copy reads

@ray.remote
def col_sum(a, j):
    return a[:, j].sum()

# each task reads the same shared-memory buffer; big is not re-serialized
totals = ray.get([col_sum.remote(ref, j) for j in range(100)])

To go multi-node, start a head node and attach workers, then point a driver at the cluster.

# on the head node (GCS listens on 6379 by default, a nod to its Redis past)
ray start --head --port=6379

# on each worker node
ray start --address='HEAD_IP:6379'

# inspect the cluster
ray status
ray list actors
ray summary tasks
# the dashboard is at http://HEAD_IP:8265

You rarely hand-roll clusters in production. The autoscaler brings nodes up and down from a YAML file with ray up cluster.yaml, and on Kubernetes the KubeRay operator manages RayCluster, RayJob, and RayService custom resources. Dependencies travel with the job through a runtime_env that can specify pip packages, a working directory, or environment variables per task or per job, so a driver on your laptop can ship code and requirements to a remote cluster with ray job submit.

Now the mistakes everyone makes once. First, calling ray.get immediately after each .remote call inside a loop, which serializes the whole program because each get blocks before the next task is even submitted. Submit everything first, then get.

# wrong: this runs one task at a time, no parallelism at all
results = []
for i in range(8):
    results.append(ray.get(square.remote(i)))   # blocks every iteration

# right: submit all, then gather
refs = [square.remote(i) for i in range(8)]
results = ray.get(refs)

Second, an actor is single-threaded by default, so a long method call blocks every other call queued on that actor. Reach for an async actor or max_concurrency only when you actually need concurrency inside one actor. Third, objects in the store are immutable. You cannot mutate a value another task is reading. You produce a new object instead. Fourth, tiny tasks are a trap. Every remote call pays for serialization with cloudpickle, a scheduling round trip, and gRPC. Wrapping a function that runs in microseconds in @ray.remote makes it slower, not faster. Ray pays off when a task does real work, a handful of milliseconds or more, and when the data is large enough that the shared-memory store saves a copy.

Part III: When it is the right tool

Ray Core is the right tool when your problem is genuinely distributed and irregular, a pipeline of Python tasks and stateful services that a rigid framework would fight. It shines at heterogeneous ML workloads, hyperparameter sweeps, distributed simulation and RL, large-scale batch inference, and any place you want to compose CPU preprocessing and GPU compute in one program. It is deliberately general, two primitives and a scheduler, so the same substrate carries very different libraries.

The honest comparisons at the core level. Dask covers similar ground for distributed Python and is stronger if your world is dataframes and arrays, while Ray's actor model and serving story make it the better fit for stateful services and RL. Spark remains the right answer for massive structured ETL and SQL on the JVM. Ray Data overlaps it for ML ingestion and inference but is not a general data warehouse. Celery is a task queue for web backends with no shared object store and no first-class actors. Slurm and Kubernetes are cluster resource managers, and Ray runs on top of them rather than replacing them. MPI and torchrun own the tightly-coupled SPMD training world, where every rank runs identical code and communicates through collectives, and that is exactly the world of torchtitan and Megatron. Ray is looser coupling by design, and when it does run tight training it wraps a PyTorch job rather than replacing the collectives.

At the library level the comparisons matter just as much. Ray Serve competes with Triton Inference Server, KServe, BentoML, and TorchServe, and it wins on Python-native multi-model composition and autoscaling while a dedicated engine like TensorRT-LLM or a vLLM server wins on raw single-model throughput. Ray Train sits alongside raw torchrun, DeepSpeed, Lightning, and Accelerate, and it is not another parallelism strategy. It is the orchestration and fault-tolerance layer that launches your DDP or FSDP job as actors and keeps it alive. Ray Data competes with Spark, the PyTorch DataLoader, tf.data, and Petastorm, and it wins where a streaming pipeline must overlap CPU and GPU stages and feed distributed training. RLlib competes with Stable-Baselines3, CleanRL, TorchRL, and Tianshou, and it trades their single-machine readability for multi-node scale and a catalog of production algorithms. There is more on the algorithms themselves in the RL section and in deep reinforcement learning.

The architecture-shaped warning is about task granularity and data placement, which is the distributed-systems cousin of the parallel computing lessons in parallel computing. Ray's overheads are small but not zero, and they are fixed per task.

dangerous:  a million microsecond tasks
   driver --submit--> scheduler --lease--> worker --gRPC--> result
            <--- per-task overhead dwarfs the work; the cluster idles on plumbing --->

safe:       coarse tasks over shared data
   ray.put(big) once, then tasks that each do real work on a slice
            <--- overhead amortized, shared-memory reads are free --->

The fix is almost always to make tasks coarser, to batch the unit of work, and to put large inputs into the store once instead of passing them as arguments repeatedly. The symptom of getting this wrong is a program that is technically parallel and yet slower than a serial loop, with the dashboard showing workers spending their time in scheduling and deserialization rather than in your code.

Part IV: The full life of one remote call

The specimen is the smallest interesting thing Ray does, one task call and one get, on a multi-node cluster. Follow y = square.remote(x) then ray.get(y). Every library in Part V is ultimately this path repeated, so it is worth tracing carefully.

Stage 1: the decorator and the call

@ray.remote wraps square into a RemoteFunction (in python/ray/remote_function.py). The plain function is preserved, but the wrapper adds .remote(). Calling square.remote(x) does not run anything. It builds a task specification, a description of the function to run, its arguments, and its resource requirements, and hands it to the process's core worker. Actors are the parallel case, wrapping a class into an ActorClass (in python/ray/actor.py) whose .remote() creates a long-lived worker instead of a one-shot task.

Stage 2: ownership and the returned ObjectRef

Before anything schedules, the driver's core worker allocates an ObjectRef for the future return value and records that the driver is its owner. Ownership is the quiet center of Ray's design. The owner is the worker that created the ref, and it alone is responsible for that object's reference count and its locations in the cluster. There is no central table of every object. This decentralization is what lets Ray schedule millions of fine-grained tasks without a metadata bottleneck, and it is described in the ownership paper (NSDI 2021) that followed the original Ray paper (OSDI 2018).

Stage 3: arguments, inline or by reference

The core worker resolves the task's arguments. Small values are inlined directly into the task spec, so a task that takes an integer carries the integer with it. Large values are put into the local object store and passed by reference, so they are transferred at most once and shared. If an argument is itself an ObjectRef, the task gains a dependency and will not be scheduled until that object exists, which is how Ray builds implicit dataflow graphs out of ordinary function composition.

Stage 4: the lease request and distributed scheduling

The core worker asks its local raylet for a worker lease that satisfies the task's resources. The raylet is the per-node C++ daemon that owns local scheduling. If the local node has the resources and an idle worker, the raylet grants a lease. If it does not, it replies with a spillback, the address of another raylet that does, using a cluster-wide resource view that the GCS keeps roughly synchronized. This is two-level scheduling. The owner drives task placement, and the raylets arbitrate the physical resources. Leases are the efficiency trick. Once a worker is leased, the owner may reuse it for many tasks, amortizing the scheduling round trip over a batch of work rather than paying it per call.

Stage 5: direct dispatch and execution

With a lease in hand, the owner pushes the task spec directly to the leased worker over gRPC. The raylet is out of the hot path now, which keeps the scheduler from becoming a funnel. The worker resolves the arguments, fetching any remote objects it needs by asking its local object manager to pull them from the node that holds them, then runs square. The return value is stored in the worker's local object store, or inlined back to the owner if it is small.

Stage 6: reference counting and eviction

The worker reports the value's location back to the owner, which updates the ObjectRef. From here the object lives as long as some reference to it exists. The owner counts references, including ones it handed to other tasks, which then become borrowers that report back when they are done. When the count reaches zero the object becomes evictable, and when the store fills up, cold objects are spilled to local disk or cloud storage rather than lost, with their refs still valid. If a node dies and takes a task-produced object with it, Ray can often reconstruct it by re-executing the lineage of tasks that produced it, which is fault tolerance without checkpointing every intermediate value.

Stage 7: ray.get and zero-copy reads

ray.get(y) blocks until the value is ready. If the object is on another node, the object manager pulls it into the local store first. Then the value is read. For numpy arrays and Arrow buffers the read is zero-copy. The consumer maps the same shared-memory pages the producer wrote, with no deserialization at all. Other Python objects are deserialized with cloudpickle. That closes the loop of one call. A future created on submit, a lease and a direct dispatch, an immutable result in shared memory, and a reference count that decides when the bytes can go. ray.wait is the same story for many refs at once, returning whichever are ready so you can process results as they complete.

Part V: Internals deep dives

Deep dive: the object store and ownership

The object store is a shared-memory segment on every node, historically built on Plasma from the Arrow project and now integrated into the raylet's object manager. Objects are immutable and stored once per node. Every local worker that reads an object maps the same pages, so N readers on a machine cost one copy, not N. For numpy and Arrow data the read is truly zero-copy, which is why Ray Data and Ray Train can move batches between CPU preprocessing and GPU consumers cheaply. When the store fills, objects spill to disk or to a configured cloud bucket, and inter-node access is a pull initiated by the object manager on the node that needs the bytes.

The subtle part is who keeps the books. A naive design has a central master track every object's location and reference count, which becomes a bottleneck the instant tasks get small and numerous. Ray's answer is ownership.

object X produced by task T, submitted by worker W

owner of X          = W (the submitter of T), and only W
W tracks            reference count of X, borrowers of X, node locations of X
no global table     of all objects; metadata is sharded across owners

if W dies           X is lost, but Ray may re-run T's lineage to rebuild X
if a holder dies    W notices via the borrower protocol and adjusts the count

By making the creator of a reference its owner, Ray scales object metadata with the number of workers instead of funneling it through one master, which is what lets it schedule millions of tiny tasks. The cost is a sharp failure rule worth memorizing. An object's fate is tied to its owner. If the owner process dies, the object cannot be fetched even if a copy physically exists somewhere, because the bookkeeping that finds and validates it is gone. Lineage-based reconstruction covers task outputs, whose producing task can be replayed, but not values placed with ray.put by a since-dead owner, which have no lineage to replay.

Deep dive: distributed scheduling, the raylet, and placement groups

The raylet is the workhorse daemon, one per node, written in C++. It has two halves. The node manager is the local scheduler and worker pool. It grants and reclaims worker leases, bin-packs the node's logical resources, and spills requests it cannot satisfy. The object manager is the store and its transfers. Scheduling in Ray is thus local first. An owner asks the nearest raylet, which either serves the request or forwards it, and the global resource picture that informs spillback is kept in the GCS and refreshed periodically rather than consulted on every decision.

Resources deserve their own note because they are logical, not enforced. A declared num_cpus is a reservation the scheduler honors when packing, not a cgroup. The exception is GPUs. When Ray assigns num_gpus to a worker it sets CUDA_VISIBLE_DEVICES so the process sees only its share, and fractional GPU requests let several workers share one device, with memory isolation left to you. Custom resources are arbitrary labels you attach to nodes and request on tasks, which is how teams route work to specific accelerator types or data-local nodes.

Placement groups are the primitive that makes training and RLHF possible, and they are easy to overlook. A placement group reserves a set of resource bundles atomically across the cluster, with a strategy that controls how they are laid out.

The point of a placement group is gang scheduling. A distributed training job needs all N of its GPUs reserved together or none of them, because a job holding half its GPUs while waiting for the rest is a deadlock in the making. Placement groups turn all-or-nothing resource acquisition into a first-class operation, which is exactly what an RLHF pipeline needs to colocate a policy trainer and a generation engine on the same GPUs. verl and OpenRLHF both lean on this heavily.

The GCS, the Global Control Store, is the cluster's control plane. It holds node membership, the actor registry, placement group tables, runtime environments, job metadata, and a general key-value store that libraries use for coordination. It was originally backed by Redis, which is why the default GCS port is 6379, and it is now a dedicated in-memory service that can optionally persist to Redis for high availability so the head node can restart without tearing down the cluster. The GCS is on the control path, not the data path. It decides where actors live and which nodes exist, while the bulk data flows worker to worker through the object managers.

Deep dive: Ray Data, streaming blocks that overlap CPU and GPU

A Ray Data Dataset is a sequence of blocks, each an Arrow table living in the object store. Operations are lazy. read_parquet, map_batches, filter, and the rest build a logical plan that is optimized into a physical plan of operators and then run by a streaming executor. Streaming is the load-bearing word. Rather than materializing the whole dataset at each stage, the executor pipelines blocks through the operators, keeping only a bounded window resident and applying backpressure so a slow GPU stage does not let a fast reader flood memory.

import ray, numpy as np

ds = ray.data.read_parquet("s3://bucket/images/")

class Classifier:
    def __init__(self):
        self.model = load_model()          # loaded once per actor, not per batch
    def __call__(self, batch):
        batch["pred"] = self.model(batch["image"])
        return batch

# an actor pool of GPU workers, each holding the model, streamed over the data
preds = ds.map_batches(
    Classifier,
    concurrency=4,          # four model-holding actors
    num_gpus=1,             # one GPU each
    batch_size=256,
)
preds.write_parquet("s3://bucket/preds/")

The pattern above is the canonical large-scale batch inference job. A stateless map_batches spins up short tasks. Passing a class with concurrency spins up an actor pool so an expensive model loads once per actor and is reused across batches, while the reader, any CPU preprocessing, and the GPU model all run as concurrent stages of one pipeline. The same machinery is how Ray Train gets fed. Each training worker asks for its shard of a Dataset and pulls batches that were prepared on CPU workers in parallel, overlapping data prep with the backward pass.

Deep dive: Ray Train, orchestration around your PyTorch loop

Ray Train does not implement a new parallelism strategy. It launches your existing distributed training as a set of worker actors and takes care of everything around the edges. A TorchTrainer takes a training function and a ScalingConfig, starts that many workers, and sets up the torch process group so your function can use DDP or FSDP normally.

import ray.train
from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer

def train_fn(config):
    model = ray.train.torch.prepare_model(build_model())   # wraps in DDP
    shard = ray.train.get_dataset_shard("train")           # this worker's data
    for epoch in range(config["epochs"]):
        for batch in shard.iter_torch_batches(batch_size=32):
            loss = step(model, batch)
        ray.train.report({"loss": float(loss)})            # metrics + checkpoint

trainer = TorchTrainer(
    train_fn,
    train_loop_config={"epochs": 10},
    scaling_config=ScalingConfig(num_workers=8, use_gpu=True),
    datasets={"train": ray.data.read_parquet("s3://bucket/train/")},
)
result = trainer.fit()

What Ray Train adds is the operational layer. It gang-schedules the workers with a placement group, wires each one to a Ray Data shard, collects ray.train.report calls into metrics and checkpoints persisted to shared or cloud storage, and restarts the worker group from the last checkpoint when a node fails. The actual sharding of parameters and gradients is still PyTorch's, which means a torchtitan-style FSDP job or a DeepSpeed job can run underneath Ray Train, gaining fault tolerance and streaming ingestion without giving up its collectives. That division of labor, Ray owns placement and recovery while the framework owns the math, is the whole point.

Deep dive: Ray Serve, deployments as composable actors

Ray Serve turns model serving into a graph of Python classes. A @serve.deployment becomes a group of replica actors. You bind deployments together to compose them, and serve.run brings the graph up.

from ray import serve
from ray.serve.handle import DeploymentHandle

@serve.deployment(num_replicas=2)
class Embedder:
    def embed(self, text: str):
        return model_embed(text)

@serve.deployment
class Router:
    def __init__(self, embedder: DeploymentHandle):
        self.embedder = embedder
    async def __call__(self, request):
        text = (await request.json())["text"]
        vec = await self.embedder.embed.remote(text)   # call another deployment
        return {"vector": vec.tolist()}

app = Router.bind(Embedder.bind())
serve.run(app)   # now serving on http://localhost:8000

Behind the API, a controller actor holds the desired state of every deployment, a proxy actor on each node terminates HTTP or gRPC and routes requests to replicas, and the replicas are ordinary Ray actors that Serve can autoscale by watching queue depth, adding replicas under load and removing them when idle. The composition story is what sets it apart. A DeploymentHandle lets one deployment call another as if it were a local method, so a preprocessor, a model, and a postprocessor, or an ensemble, or a router in front of several models, is just Python objects calling each other, each scaled independently. Ray Serve LLM builds on this to host vLLM engines with an OpenAI-compatible API, which is a natural fit because the engine is itself a Ray actor.

Deep dive: RLlib, the sampling-then-learning loop as actors

RLlib is Ray Core applied to reinforcement learning, and its shape mirrors the RL loop directly. An Algorithm such as PPO, configured through a config object and then built, orchestrates two kinds of actor. EnvRunner actors (the older stack called them RolloutWorkers) each hold copies of the environment and the current policy and step them to collect experience in parallel. A Learner, or a LearnerGroup of learner actors for multi-GPU, takes the collected batch and computes gradient updates, with the neural network defined by an RLModule. Each iteration samples from the env runners, learns on the batch, and broadcasts the updated weights back to the runners.

from ray.rllib.algorithms.ppo import PPOConfig

config = (
    PPOConfig()
    .environment("CartPole-v1")
    .env_runners(num_env_runners=4)     # four parallel experience collectors
    .learners(num_learners=1)
)
algo = config.build()                   # exact builder method has evolved across versions
for i in range(10):
    result = algo.train()               # sample, learn, sync weights
    print(i, result["env_runners"]["episode_return_mean"])

The new API stack (RLModule, Learner, EnvRunner, ConnectorV2) is a cleaner separation of the network, the update, the sampling, and the data transforms than the old stack (Policy, RolloutWorker, Trainer), and the exact spellings have moved more than once, so learn the four roles rather than the current method names. What matters for this chapter is the structural insight. RLlib is a distributed dance of many samplers and a few learners exchanging weights and batches through actors, and that same structure, scaled up and specialized to language models, is precisely what RLHF needs.

Deep dive: why Ray became the control plane for LLM RLHF

RLHF for large language models is not one model training. It is a small society of models that must run together. A policy (the actor being trained), often a separate critic, a frozen reference model for the KL penalty, one or more reward models, and a fast generation engine that produces rollouts. These pieces use different frameworks, FSDP or Megatron for the training models and vLLM or SGLang for generation, they want different parallelism layouts, and each PPO step has to route data between them and synchronize the policy's new weights into the generation engine.

one PPO step of LLM RLHF

  prompts -> [generation engine: vLLM / SGLang]  -> responses
                       |
                       v
             [reward model]      score responses
             [reference model]   log-probs for the KL term
                       |
                       v
             [critic]            value estimates, advantages
                       |
                       v
             [policy trainer: FSDP / Megatron]  update weights
                       |
                       v
             weight sync  ----------------->  reshard into the generation engine

Pure SPMD cannot express this cleanly. A torchrun world where every rank runs the same script has no natural place to say "generate with these engines, then score with that model, then update this other model, then push the new weights over there." That is branching, heterogeneous control flow, and it wants a single controller. Ray provides exactly that, one driver that treats each model group as a set of actors, placement groups that gang-reserve and colocate their GPUs, an object store to pass rollout batches, and named actors plus collectives to reshard weights between the training layout and the inference layout.

This is why the ecosystem converged on it. verl (the HybridFlow framework from ByteDance) uses Ray for the single-controller RLHF dataflow while keeping efficient multi-controller SPMD inside each model, colocating the actor, rollout, reference, and critic roles with placement groups, supporting FSDP and Megatron training backends with vLLM or SGLang rollout, and resharding weights between the train and generate layouts each step. OpenRLHF uses Ray to place vLLM generation engines and DeepSpeed ZeRO-3 training actors across GPU groups, coordinating the PPO pipeline and the weight sync from trainer to engine. Others, including NeMo-Aligner and various TRL-based stacks, adopt Ray for the same reason. The generation engine, the trainer, and the reward model are all just actors, and the loop that connects them is plain Python on a single controller. The RLHF background itself lives in the RL section.

Part VI: Reading the repository

The Ray repository is large and bilingual, a Python surface over a C++ core, built with Bazel. Paths below are the stable landmarks and are described by role where fine detail is likely to have shifted.

Stage 0, orientation. Run the Part II quickstart and open the dashboard at port 8265, then read the Ray Core docs, then python/ray/remote_function.py and python/ray/actor.py. Questions. What does .remote() actually build, how does an ObjectRef differ from the value it points to, and what is the difference between how a task and an actor acquire a worker?

Stage 1, the Python to C++ bridge. Skim python/ray/_raylet.pyx, the Cython layer where Python calls cross into the C++ core worker to submit tasks and get objects. Do not try to master it. The goal is to see the seam, that the driver you write is a thin Python skin over a core worker library embedded in every process.

Stage 2, the core worker. Read src/ray/core_worker/, which is the intellectual center of the whole system. This is where task submission, the ownership and reference-counting tables, and the object-store interface live. Read it with the Part IV trace open beside you. Questions. Where does the owner record a new object, how does a borrower report back, and how does the worker-lease reuse actually amortize scheduling?

Stage 3, the node and the cluster. Read src/ray/raylet/ for the node manager's scheduling loop, worker pool, and spillback, then src/ray/object_manager/ for the shared-memory store, spilling, and inter-node pulls, then src/ray/gcs/ for the control plane. Questions. When does a local raylet decide to spill a request, what does the GCS know that a raylet does not, and how does a placement group reserve bundles atomically?

Stage 4, one library end to end. Pick a single library and read only its Python top layer. python/ray/data/ for the Dataset, the plan optimizer, and the streaming executor. python/ray/serve/ for the controller, proxy, and replica. python/ray/train/ for the trainers and the backend that sets up the process group. python/ray/rllib/ for the Algorithm, the EnvRunner, and the Learner. Each is a Ray program, so you will recognize tasks, actors, and object refs underneath the domain API.

Where not to start. RLlib's full catalog of algorithms is breadth you do not need on a first pass. The autoscaler node-provider plugins in python/ray/autoscaler/ are cloud-specific glue. The dashboard frontend is a separate world. Meet these only after tasks, actors, ownership, and the raylet are solid.

Part VII: Hands-on labs

Labs 1 through 4 need only a laptop. Labs 5 and 6 are happier with a GPU but degrade gracefully to CPU. Log and dashboard details drift with Ray's release cadence.

Lab 1: parallel speedup and the get-in-a-loop trap. Concept: submit then gather, Part II.

import ray, time
ray.init()

@ray.remote
def slow(i):
    time.sleep(1)
    return i

t = time.time(); [ray.get(slow.remote(i)) for i in range(8)]   # get inside loop
print("serial-ish:", time.time() - t)                          # ~8 s

t = time.time(); ray.get([slow.remote(i) for i in range(8)])   # submit then gather
print("parallel:", time.time() - t)                            # ~1 s

Watch the two timings and explain the gap. Then open the dashboard and see eight tasks running at once in the parallel case versus one at a time in the first.

Lab 2: actor state and single-threaded ordering. Concept: actors keep state and serialize their calls.

@ray.remote
class Bank:
    def __init__(self):
        self.balance = 0
    def deposit(self, x):
        self.balance += x
        return self.balance

acct = Bank.remote()
print(ray.get([acct.deposit.remote(1) for _ in range(5)]))   # [1,2,3,4,5], in order

Observe that the results are exactly ordered because one actor runs one method at a time. Now add @ray.remote(max_concurrency=4) and a time.sleep inside the method, and watch the behavior change. The lesson is that an actor is a single-threaded service unless you ask otherwise.

Lab 3: ray.put and zero-copy sharing. Concept: the object store, Part V.

import numpy as np
big = np.random.rand(5000, 5000)      # ~200 MB
ref = ray.put(big)

@ray.remote
def rowmean(a, i):
    return float(a[i].mean())

means = ray.get([rowmean.remote(ref, i) for i in range(50)])

Run ray memory or watch the dashboard's cluster memory while this runs and confirm the array is stored once, not fifty times. Then pass big directly instead of ref and observe the difference in object-store traffic.

Lab 4: a placement group and gang scheduling. Concept: bundles reserved atomically, Part V.

from ray.util.placement_group import placement_group
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

pg = placement_group([{"CPU": 1}, {"CPU": 1}], strategy="STRICT_PACK")
ray.get(pg.ready())     # blocks until all bundles are reserved together

@ray.remote(num_cpus=1)
class Worker:
    def whoami(self):
        return ray.get_runtime_context().get_node_id()

ws = [Worker.options(
        scheduling_strategy=PlacementGroupSchedulingStrategy(pg, bundle_index=i)
      ).remote() for i in range(2)]
print(ray.get([w.whoami.remote() for w in ws]))   # same node under STRICT_PACK

Confirm both actors land on one node under STRICT_PACK. Switch to STRICT_SPREAD on a single-node cluster and watch the group fail to become ready, which is the gang guarantee working as intended.

Lab 5: batch inference with Ray Data. Concept: streaming actor pool, Part V.

import ray
ds = ray.data.range(10_000)

class Squarer:
    def __call__(self, batch):
        batch["out"] = batch["id"] ** 2
        return batch

out = ds.map_batches(Squarer, concurrency=4, batch_size=512)
print(out.take_batch(4))

Note that passing a class rather than a function makes each of the four workers an actor. Now give Squarer.__init__ an expensive setup and confirm from the logs that it runs four times, once per actor, not once per batch. That reuse is the whole reason batch inference uses an actor pool.

Lab 6: compose two Ray Serve deployments. Concept: deployment handles and independent scaling, Part V.

from ray import serve
from ray.serve.handle import DeploymentHandle

@serve.deployment(num_replicas=2)
class Upper:
    def do(self, s: str):
        return s.upper()

@serve.deployment
class Front:
    def __init__(self, up: DeploymentHandle):
        self.up = up
    async def __call__(self, req):
        s = (await req.json())["s"]
        return {"result": await self.up.do.remote(s)}

serve.run(Front.bind(Upper.bind()))
# curl -X POST http://localhost:8000/ -H 'content-type: application/json' -d '{"s":"hi"}'

Call it over HTTP, then bump Upper to num_replicas=4 and redeploy, confirming the two deployments scale independently. Open the Serve dashboard and watch requests fan out across the Upper replicas.

Part VIII: Questions and model answers

Understanding checks. Answer aloud before reading.

1. What is Ray, in one sentence?

A distributed futures system with two primitives, stateless tasks and stateful actors, whose results live as immutable objects in a shared-memory store and whose lifetimes are managed by ownership-based reference counting, with a stack of ML libraries (Data, Train, Serve, RLlib) built on that core.

2. What is an ObjectRef and how does it differ from a Future in plain Python?

An ObjectRef is a reference to a value that may live on another node in the shared object store. Like a future it is returned immediately from an async call and resolved with ray.get, but unlike a local future it is a distributed, first-class value. You can pass it to other tasks to build dataflow dependencies, and it is reference-counted across the cluster by its owner.

3. What is the difference between a task and an actor?

A task is a stateless remote function that runs once on any available worker. An actor is a remote class instance bound to a dedicated long-lived worker that keeps state between calls and, by default, executes its methods one at a time in submission order.

4. Explain ownership and why Ray uses it.

Every object has exactly one owner, the worker that created its reference. The owner alone tracks that object's reference count, borrowers, and locations. This shards object metadata across all workers instead of funneling it through a central master, which is what lets Ray schedule millions of fine-grained tasks without a bottleneck. The cost is that if the owner dies, the object is lost.

5. Trace one remote call from .remote to ray.get.

.remote builds a task spec, the owner allocates the return ObjectRef and records itself as owner, arguments are inlined or put by reference, the core worker asks its raylet for a worker lease, the raylet grants locally or spills to another node, the owner dispatches the task directly to the leased worker over gRPC, the worker resolves args, runs the function, and stores the result, then reports its location back so reference counting can manage it, and ray.get blocks and reads the value, zero-copy for numpy.

6. Why can wrapping a tiny function in @ray.remote make a program slower?

Every remote call has fixed overhead, cloudpickle serialization, a scheduling round trip for a lease, and gRPC. If the function runs in microseconds, that overhead dominates and the cluster spends its time on plumbing. Tasks should be coarse enough that real work dwarfs the per-task cost.

7. What is a placement group and what problem does it solve?

A placement group reserves a set of resource bundles atomically across the cluster with a layout strategy. It solves gang scheduling, the need to acquire all of a job's resources together or none, which avoids deadlock in distributed training and lets an RLHF pipeline colocate its trainer and generation engine on the same GPUs.

8. What does the GCS store, and what is it not on the path for?

The GCS holds cluster control state, node membership, the actor registry, placement groups, runtime environments, jobs, and a key-value store. It is on the control path only. Bulk data flows worker to worker through the object managers, not through the GCS.

9. How does Ray Data overlap CPU and GPU work?

A dataset is a stream of blocks, and the streaming executor pipelines them through operators, keeping only a bounded window resident with backpressure. Reading, CPU preprocessing, and GPU inference run as concurrent stages, so the GPU is fed by CPU workers preparing the next batches while it consumes the current one.

10. What does Ray Train add over just running torchrun?

Ray Train does not replace the parallelism, which is still PyTorch's DDP or FSDP. It launches the workers as a gang-scheduled actor group, wires each to a Ray Data shard, collects metrics and checkpoints to durable storage, and restarts the group from a checkpoint on failure. It is the orchestration and fault-tolerance layer around your loop.

11. How does Ray Serve compose multiple models?

Each @serve.deployment is a group of replica actors. You bind deployments together and pass DeploymentHandles, so one deployment calls another like a local method. A preprocessor, a model, a postprocessor, an ensemble, or a router are just Python objects calling each other, each autoscaled independently on queue depth.

12. Why is SPMD a poor fit for LLM RLHF, and why is Ray a good one?

RLHF coordinates several distinct models, a policy, a critic, a reference model, a reward model, and a generation engine, with branching control flow and weight syncing between training and inference layouts. SPMD assumes every rank runs the same program, which cannot express that heterogeneity. Ray's single controller treats each model as actors, uses placement groups to colocate their GPUs, and passes data through the object store, so the loop is plain Python.

13. Name two frameworks that build RLHF on Ray and what Ray gives them.

verl (HybridFlow) uses Ray for single-controller RLHF dataflow with SPMD inside each model, placement-group colocation, and FSDP or Megatron plus vLLM or SGLang. OpenRLHF uses Ray to place vLLM generation engines and DeepSpeed training actors and to coordinate the weight sync from trainer to engine. Ray gives both the actor orchestration, gang scheduling, and data passing.

14. A Ray job dies when one node fails and cannot recover an object. What happened?

The lost object's owner most likely lived on the failed node, so its reference-counting and location metadata went with it and the object became unfetchable even if a copy existed. Lineage reconstruction can replay task-produced objects, but a ray.put value from a dead owner has no lineage to replay.

Part IX: Design lessons

Two primitives, then build everything on them. Ray commits to just tasks and actors over an object store, and every library, Data, Train, Serve, RLlib, is an ordinary Ray program on that substrate. A small, sharp core that many domains reuse beats a pile of special-purpose systems, the same instinct as a good kernel interface or a minimal instruction set.

Make data movement implicit and dedupe with shared memory. Results are immutable objects read zero-copy by every local worker, and inter-node transfer is an explicit pull. Turning data placement into first-class references, rather than hidden serialization on every call, is the same win as content-addressed storage or copy-on-write pages.

Shard the metadata, not just the data. Ownership pushes reference counting and object location to the creator of each reference, so the control state scales with the number of workers instead of through one master. Whenever a central table becomes a bottleneck, the move is to partition responsibility by who created what, which is how scalable filesystems and databases avoid a single hot coordinator.

A single controller between SPMD and a workflow engine. Ray sits deliberately between rigid lockstep parallelism and a heavy DAG scheduler. One driver expressing distributed work as plain Python control flow is expressive enough for RLHF yet light enough for tight loops, and that middle ground is exactly the gap the RLHF ecosystem needed filled.

Gang scheduling deserves to be a primitive. Placement groups make all-or-nothing resource acquisition a first-class operation rather than something each application reinvents badly. Wherever partial acquisition means deadlock, training GPUs, database locks, distributed transactions, the atomic reservation belongs in the substrate.

Recover by replay, not only by checkpoint. Because tasks are deterministic functions with recorded lineage, Ray can rebuild a lost object by re-executing the tasks that produced it, paying for fault tolerance only when something actually fails. Lineage and idempotence turn recovery into recomputation, the same idea behind resilient distributed datasets and event-sourced systems.

Part X: Memorization framework

The one-sentence summary. Ray turns a Python program into a cluster program with stateless tasks and stateful actors, stores their immutable results in a shared-memory object store tracked by ownership-based reference counting, schedules them through per-node raylets with a GCS control plane and placement groups for gang scheduling, and layers Data, Train, Serve, and RLlib on top, which is why it became the single-controller substrate under LLM RLHF.

.remote() -> ObjectRef (owner allocates it, starts refcount)
  -> core worker asks raylet for a worker lease (resources)
  -> raylet grants locally or spills to another node (GCS resource view)
  -> owner dispatches directly to the leased worker over gRPC
  -> worker runs, stores result in shared-memory object store
  -> ray.get reads it (zero-copy for numpy); refcount to 0 -> evict/spill

The system mapped to source:

python api        python/ray/remote_function.py, actor.py, _private/worker.py
py->c++ bridge     python/ray/_raylet.pyx
core worker       src/ray/core_worker/   (submission, ownership, refcount)
node scheduler    src/ray/raylet/        (leases, spillback, worker pool)
object store      src/ray/object_manager/  (shared memory, transfer, spill)
control plane     src/ray/gcs/           (nodes, actors, placement groups, kv)
libraries         python/ray/{data,train,serve,rllib,tune}/

Memorize these blocks:

• Primitives: tasks are stateless functions, actors are stateful classes, both requested with @ray.remote and called with .remote() returning an ObjectRef.
• Object store: immutable objects, one copy per node in shared memory, zero-copy numpy reads, spill to disk or cloud when full.
• Ownership: the creator of a reference owns the object and does its reference counting and location tracking, no central table, and if the owner dies the object is lost.
• Scheduling: owner requests a worker lease from a raylet, which grants locally or spills, then the owner dispatches directly and reuses the lease; placement groups gang-reserve bundles (PACK, SPREAD, STRICT_PACK, STRICT_SPREAD).
• Libraries: Ray Data streams blocks over CPU and GPU stages, Ray Train orchestrates a PyTorch DDP/FSDP job as actors, Ray Serve composes deployment actors behind a proxy, RLlib runs EnvRunner samplers and Learner updaters.
• RLHF: a single controller with actors for policy, critic, reference, reward, and a vLLM or SGLang generation engine, colocated by placement groups, is why verl and OpenRLHF build on Ray.

Part XI: Papers and further reading

The ideas in this walkthrough come from a small set of papers, and each one rewards a direct read. Where this site covers the same system or algorithm in depth, the companion link points there.

1. Moritz et al., Ray, A Distributed Framework for Emerging AI Applications, OSDI 2018. The paper behind this whole chapter, the unified task and actor interface over a shared object store with a bottom-up distributed scheduler.
2. Wang et al., Ownership, A Distributed Futures System for Fine-Grained Tasks, NSDI 2021. The redesign that pushed object metadata to each reference's creator, the source of the failure rule in Part V that an object dies with its owner.
3. Hewitt et al., A Universal Modular ACTOR Formalism for Artificial Intelligence, IJCAI 1973. The origin of the actor model that Ray's stateful workers descend from.
4. Zaharia et al., Resilient Distributed Datasets, A Fault-Tolerant Abstraction for In-Memory Cluster Computing, NSDI 2012. The lineage-based recovery idea from Spark that Ray reuses when it rebuilds a lost object by replaying the tasks that produced it.
5. Liang et al., RLlib, Abstractions for Distributed Reinforcement Learning, ICML 2018. The case for composing distributed RL from top-down hierarchical control, which became the RLlib library of Part V. The algorithms it ships are treated in the deep reinforcement learning class on this site.
6. Liaw et al., Tune, A Research Platform for Distributed Model Selection and Training, 2018. The hyperparameter search layer that runs each trial as ordinary Ray work, the narrow waist between training scripts and search algorithms.
7. Schulman et al., Proximal Policy Optimization Algorithms, 2017. The clipped objective behind both the RLlib example and the RLHF loop Ray orchestrates. Derived step by step in the PPO note on this site.
8. Ouyang et al., Training language models to follow instructions with human feedback, 2022. The RLHF pipeline whose society of models is exactly the heterogeneous workload Part V argues Ray was built for.
9. Kwon et al., Efficient Memory Management for Large Language Model Serving with PagedAttention, SOSP 2023. The generation engine Ray colocates and serves, covered in the vLLM walkthrough.
10. Sheng et al., HybridFlow, A Flexible and Efficient RLHF Framework, EuroSys 2025. The single-controller RLHF design that runs on Ray as verl, covered in the verl walkthrough.
11. Hu et al., OpenRLHF, An Easy-to-use, Scalable and High-performance RLHF Framework, 2024. The Ray plus vLLM plus DeepSpeed stack from Part V, covered in the OpenRLHF walkthrough.

Part XII: Final takeaway

If the tightly-coupled training these libraries wrap is the part you want to understand from the inside, the torchtitan chapter follows one FSDP training step down to the collectives, and the generation engines Ray colocates for RLHF are dissected in the vLLM and SGLang chapters. The scheduling and data-placement instincts underneath Ray are the distributed cousins of the ideas in parallel computing. Then come back and read src/ray/core_worker/ once more with the one-call trace in hand, and the object store, the leases, and the reference counts will read like the plain mechanics they are.