Ray

TL;DR — Ray is a distributed execution layer, not a training or serving library: it schedules tasks (stateless functions) and actors (stateful processes) over a cluster-wide resource graph, and moves data between them through a per-node object store rather than through the driver’s heap. Three primitives carry all the weight — ObjectRef futures, a name-to-float resource model, and placement groups for gang scheduling. Everything below that (veRL, OpenRLHF, SkyRL, Ray Serve LLM) is built from those three. The honest accounting: ray’s own measurement is 0.47 ms of overhead per empty task, actor restarts default to 0 and do not restore state, and infeasible placement groups produce no warning at all.

1. Why this exists

Your 7B GRPO run works on one node. Now you have to scale it, and three things break at once, none of which is arithmetic.

The resource shapes stop matching. One post-training step has stages with different bottlenecks: gradient compute (GPU FLOPs, synchronous, fixed world size), rollout generation (GPU memory bandwidth, ragged batch), and reward scoring. If the reward is a verifier that executes generated code in a sandbox, that stage wants CPU and its duration varies by 100x across samples. Run all of it as one torchrun job and the GPU ranks assigned to the sandbox stage idle while a compiler runs, because the world size was fixed at launch and the layout is homogeneous by construction.

The coordination overhead stops being negligible. A reward stage that verifies 300,000 samples at even one millisecond of round-trip cost per sample is bottlenecked by overhead rather than by work. The version you measure is worse: calls that take 6 s (a flaky test) are interleaved with calls that are instant, and a synchronous loop waits for the slow ones.

The models cannot be recreated per call. An inference engine holds weights, a KV cache, and a CUDA context. Loading Qwen2.5-7B takes tens of seconds and tens of gigabytes; a multiprocessing.Pool gives you no place to put state that survives across calls, and no scheduler that knows a GPU is already occupied.

Ray’s answer is to stop pretending the pipeline is a rectangle. Three primitives, and no opinion about the math:

flowchart LR
    subgraph P["The three primitives"]
      T["task: stateless function<br/>@ray.remote def f()"]
      A["actor: stateful process<br/>@ray.remote class C"]
      O["object: immutable bytes<br/>in the per-node object store"]
    end
    D["driver process<br/>(your top-level Python)"] -->|"f.remote() -> ObjectRef"| T
    D -->|"C.remote() -> ActorHandle"| A
    T -->|"return value inline or via Plasma"| O
    A -->|"actor method -> ObjectRef"| O
    O -->|"ray.get(ref) -> value"| D

The design intent is old and stable: the OSDI 2018 paper (arXiv:1712.05889) describes Ray as a framework for emerging AI applications, centred on a dynamic task graph and a distributed fault-tolerant store for control state. Read that paper for intent, not for API — the current object store and GCS internals postdate it [as of 2026-10].

Where this matters for you. If an interviewer asks “what does Ray buy you?”, the weak answer is “it makes training faster.” It does not. Ray does not perform your all-reduce and does not batch your requests. It guarantees that the processes that will perform those operations exist at the same time, on nodes you chose, with a data plane that is not Python pickles through the driver. On a single-node fixed-topology job, torchrun gives you the same thing with one fewer distributed system.

Ray also does not replace torch.distributed or vLLM. The layering that actually gets used is:

Kubernetes / cloud      -> machines, pods, preemption signals
  Ray                   -> resource graph, scheduling, actor lifecycle, object store
    torch.distributed   -> collectives inside a gang (FSDP, TP, PP)
      FSDP / vLLM       -> the numerical work

Most placement-group bugs come from one line of that diagram being read as “Ray does the communication.” It does not; it makes the communication possible.

2. How it works

2.1 Tasks, actors, and what each one is for

A task is a Python function that Ray runs in some worker process somewhere. It is stateless in the sense that matters: Ray may run it in a fresh process, on a different node each call, and it retains nothing between calls.

An actor is a Python class whose instance lives in a dedicated worker process for as long as its handle exists. The constructor runs once, on whichever node the scheduler picked, and every method call is dispatched to that same process.

That distinction is the whole reason RL frameworks use actors. The engine’s weights, its PagedAttention KV blocks, and its CUDA context must survive across calls. A task cannot hold them.

import ray

@ray.remote(num_cpus=1)
def score(text):            # TASK: no state, any worker, any node
    return len(text)

@ray.remote(num_gpus=1, num_cpus=1)
class Engine:               # ACTOR: one long-lived process, pinned resources
    def __init__(self, model_path):
        self.model = load(model_path)   # runs ONCE, not per call
    def generate(self, prompt):
        return self.model(prompt)

ref = score.remote("hello")              # returns immediately
handle = Engine.remote("/models/qwen")   # actor scheduled; constructor once
out = handle.generate.remote("hi")       # returns immediately
print(ray.get([ref, out]))               # blocks; both fetched in parallel

Every parameter, named:

• @ray.remote(...) marks the object for remote execution and declares its logical resource requirement. A task defaults to num_cpus=1; an actor defaults to num_cpus=1 at schedule time and 0 while running. num_gpus is what causes Ray to set CUDA_VISIBLE_DEVICES so the process sees only its GPU. Source: Resources [as of 2026-10].
• Engine.remote(path) returns an ActorHandle; the constructor is scheduled asynchronously, so the handle exists before the constructor finishes and the first method call blocks on it. .remote() itself never blocks — it returns an ObjectRef, and calling ray.get after each submit is the single most common way to silently serialise a Ray program (Section 5).

Resources are logical, not physical: num_cpus=1 reserves one unit of a cluster-wide accounting quantity and does not pin the process to one core, does not enforce that it uses only one core, and provides no CPU isolation. Accelerators are the exception — GPU allocation is enforced through CUDA_VISIBLE_DEVICES. This mismatch is the root of the actor-CPU hang in Sections 2.4 and 5.

2.2 The lifecycle of one .remote() call

flowchart TD
    D["driver: f.remote(args)"] --> P["pickle function + args<br/>into the GCS KV store"]
    P --> W["NormalTaskSubmitter: wait for args to be local"]
    W --> L["RequestWorkerLease RPC to a raylet"]
    L --> PT["PushTask RPC to the leased worker"]
    PT --> R["run the function"]
    R --> S{"return value size"}
    S -->|"<= ~100 KiB"| I["inlined into the reply<br/>lives in the owner process"]
    S -->|"> ~100 KiB"| PL["written to the node's Plasma store<br/>ObjectRef returned"]
    I --> G["ray.get(ref) deserialises"]
    PL --> G

The last branch matters more than it looks. Small return values are returned inline: they bypass the object store entirely and are held in the memory of the process that owns the ref — usually the driver, usually on the head node. Many small objects therefore accumulate in one process’s heap and can OOM it, with no object-store accounting to explain why (Section 5). Sources: Task Lifecycle, Objects [as of 2026-10].

An ObjectRef is a future: an ID plus metadata, not the value. ray.get(ref) blocks and deserialises, accepts a list (fetched in parallel), and takes timeout= (raises GetTimeoutError). ray.wait(refs) returns (ready, not_ready) as soon as at least one is ready, which is the primitive that makes pipelining possible. Passing an ObjectRef as a top-level argument de-references it, but passing it nested in a list or dict does not — the callee must call ray.get itself, and a pipeline that quietly stops being zero-copy usually has a nested ref.

2.3 The object store: what lives where, and when it moves

This is the part to get exactly right, because the folk wisdom is half wrong.

Each node runs one object store (Plasma, forked out of Apache Arrow), backed by shared memory. Objects are immutable. Every object has two things: its data, which lives on some set of nodes, and its metadata, which lives at the owner — the CoreWorker that created the original ref via .remote() or ray.put().

There are two kinds of copies, and they obey different rules:

• A primary copy is created by the producing task or by ray.put. The owner registers it with its raylet through a PinObjectIDs RPC, and it becomes pinned. A primary copy cannot be evicted. Under pressure it can only be spilled to disk (or, in the fallback path, allocated directly on the local filesystem).
• A secondary copy is a cached replica pulled to another node so a task there can read it. It is pinned in Plasma only while actively referenced; once its reference count drops, the Plasma LRU evicts it.

Eviction happens before spilling, and the reason is the whole design: Ray “prioritizes evicting secondary copies to free space, because they can be re-fetched from the primary copy.” Evicting a secondary copy is free — it can be refetched. Spilling a primary copy costs disk I/O and pins disk space. So the order is: evict secondary copies, then spill primary copies. Spilling is triggered three ways, all converging on the same call: reactively when a Create RPC to Plasma fails with OutOfMemory; periodically every free_objects_period_milliseconds (default 1000 ms); and on seal, when the threshold check fires immediately for a newly sealed object.

The threshold condition, and every term in it:

$$\frac{\text{primary\_object\_bytes}}{\text{object\_store\_mem}}\ge \text{object\_spilling\_threshold}$$

• $\text{primary\_object\_bytes}$ — total bytes of objects for which this node holds the primary copy. Not the total bytes in the cluster; the quantity that makes this node’s copy irreplaceable.
• $\text{object\_store\_mem}$ — the node’s Plasma capacity, which defaults to 30% of available memory and cannot be changed after the node starts. This is why Ray’s default memory split is 70% for the heap and 30% for the object store, and why a node that looks “half empty” in free -g can be spilling.
• $\text{object\_spilling\_threshold}$ — default 0.8. Raise it and you spill later but risk OutOfMemory in the reactive path; lower it and you trade disk I/O for headroom. Setting object_spilling_config to an empty string disables spilling entirely, which is how people produce a hard OOM instead of a slow job.

Eligibility is the trap. An object is eligible to spill only when its reference count is 1 — the owner holds it and no running task is using it. So holding a Python reference prevents eviction but does not prevent spilling; object-store memory held by a Python reference is spent memory. If you need an object to stay resident, give it a home inside a long-lived actor (the actor process holds the bytes), not a variable in the driver.

Two more pieces of the model. Fallback allocation: if spilling cannot free space within oom_grace_period_s, Ray mmaps the object directly on the local filesystem instead of shared memory — slower, but it avoids deadlock. Restore: ray.get, ray.wait, task-argument fetch, and actor-task argument fetch can all trigger one; for filesystem spilling a different node pulls from the spilling node, which reads from disk and streams over the network without re-populating its own Plasma store.

Zero-copy is a same-node property: NumPy arrays returned by ray.get on the node holding the primary copy are read-only views backed by shared object store memory, so arr[0] = 1 raises “assignment destination is read-only” and arr = arr.copy() forfeits the benefit. An object on another node is pulled into the local store first, and that transfer is a network cost no reference can avoid.

Sources: Object spilling, Serialization [as of 2026-10].

2.4 The resource model and the placement-group scheduling condition

A resource is simply a name-to-float pair. CPU, GPU, and memory are predefined; anything else ({"special_hardware": 1}) is a custom resource scheduled by exactly the same machinery. accelerator_type="H100" is implemented as a custom resource requirement of 0.001 units on a name like accelerator_type:H100 — the fractional quantity exists only so that any number of tasks can “hold” it while still signalling the autoscaler what hardware to provision.

Scheduling proceeds in two phases. First the feasibility filter: a task or actor may run on node $i$ only if, for every resource key $k$ in its requirement $r$, the free capacity covers it:

$$\forall k:r(k)\le c_i(k)-u_i(k)$$

• $r(k)$ — the requirement for resource $k$, e.g. 2 for num_cpus=2. It is a float, and it is hard: one missing key removes the node entirely rather than deprioritising it, because the condition is a conjunction over all $k$.
• $c_i(k)$ — node $i$’s declared capacity for $k$. Node defaults are num_cpus = detected CPUs, num_gpus = detected GPUs, memory = 70% of available memory, and these cannot be changed after the node starts.
• $u_i(k)$ — what is already committed on node $i$ by other tasks, actors, and placement-group bundles. It is a reservation ledger, not a measurement: Ray does not check that a num_cpus=1 task actually used one core.

Then the strategy picks among feasible nodes. The default "DEFAULT" scores by utilization with a spread threshold, prefers nodes that already hold the task’s large arguments (locality-aware, tasks only, not actors), then randomly picks among the top-$k$ lowest-scoring nodes, where $k$ is governed by RAY_scheduler_top_k_fraction (0.2) and RAY_scheduler_top_k_absolute (1).

Placement groups are the gang-scheduling primitive, and they change the object being scheduled from “one task” to “a bundle list that either all fits or none does.” A bundle is a resource dict; a placement group is a list of bundles plus a strategy. The condition for placing bundle $b$ on node $i$ is the same conjunction above. It must be a single node, for a structural reason rather than a Ray limitation: the bundle is the unit of atomic co-reservation, and single-node co-residency is the only thing Ray can guarantee at reservation time — cross-node reservation would have to be transactional across independent raylet schedulers. Creation is asynchronous (block with ray.get(pg.ready())) and atomic: if a bundle cannot fit on any current node, Ray reserves nothing and the group stays PENDING. That is what converts a silent performance problem into an explicit scheduling wait — and, as Section 5 shows, into an explicit hang.

PACK being the default and being soft is the single most consequential fact in a multi-GPU launcher: “8 GPUs” under PACK is a request that will be satisfied across two 4-GPU nodes if that is what fits first.

2.5 Why a bundle must fit one node for NCCL, with the arithmetic

An all-reduce over $N$ ranks must move $2\frac{N-1}{N}\cdot S$ bytes, where $S$ is the per-rank payload (the $\frac{N-1}{N}$ factor is the ring all-reduce scaling; for large $N$ it converges to $2S$). At an effective bandwidth $BW$ the time is approximately

$$T_{\text{ar}}\approx \frac{2(N-1)}{N}\cdot \frac{S}{BW}+(N-1)\cdot L$$

• $S$ — bytes per rank of whatever is being reduced. For bf16 gradients of a 7B model that is approximately $7\times 10^9\times 2\text{B}\approx 14\text{GB}$ per rank.
• $N$ — the number of participating ranks. Every rank must participate on every collective; a missing rank does not degrade the operation, it deadlocks it.
• $BW$ — effective bandwidth of the slowest link in the path. The hierarchy is NVLink > intra-node PCIe > cross-node RDMA > Ethernet, and the slowest link determines the transfer, not the fastest.
• $L$ — per-collective fixed latency (kernel launch, synchronization, handshake). This term is why you should reduce fewer, larger buffers: it is independent of $S$.

Take $S=14$ GB as one gradient all-reduce and $N=8$:

These bandwidths illustrate the classes, not benchmarked figures (example). The ratio is the point: crossing a 100 GbE link costs roughly 36x the NVLink all-reduce, and throughput is inversely proportional to it. That is the argument for STRICT_PACK on a tensor-parallel group — not that Ray refuses to co-locate, but that co-location is what the collective’s cost model requires, and STRICT_PACK makes the requirement explicit and therefore observable. The alternative, one {"GPU": 1} bundle per rank under PACK, is a schedule that usually works and occasionally produces a job 36x slower for reasons nobody can see in the logs.

3. Implementation

All code below is against Ray 2.59.0 [as of 2026-10], and every parameter is explained. The LLM API used at the end is ray.serve.llm, which exists in the current version — it is not renamed or deprecated, and it installs from the llm extra (pip install "ray[llm]").

3.1 A real task, and the batching that makes it worth it

import ray, time
ray.init()

@ray.remote(num_cpus=1)
def score_batch(texts):
    """Tokenise + length-filter a batch. num_cpus=1 reserves one logical CPU
    slot so Ray's scheduler accounts for this work; it does not pin cores."""
    return [len(t.split()) for t in texts]

texts = ["prompt " * 50] * 100_000

# ANTI-PATTERN (illustrated, do not run this line): one task per string.
# ~0.47 ms overhead each and microseconds of work, so >90% of wall clock is
# scheduling. Delete it and keep the batched path below.
refs = [score_batch.remote([t]) for t in texts]

# PATTERN: batch coarse, submit few. Readiness is consumed with ray.wait, so
# completed batches are processed while later batches still run.
BATCH = 1000
batches = [texts[i:i + BATCH] for i in range(0, len(texts), BATCH)]
pending = [score_batch.remote(b) for b in batches]
out = []
while pending:
    ready, pending = ray.wait(pending, num_returns=1)
    out.extend(ray.get(ready[0]))

Parameters that matter: num_cpus=1 (scheduler accounting only), BATCH (the overhead amortisation knob — see Section 4.1), and num_returns=1 in ray.wait (return as soon as one batch is done, which is what enables overlap).

3.2 A real actor: the thing that must not be recreated

@ray.remote(
    num_gpus=1,
    num_cpus=2,          # MUST be explicit; see the placement-group trap below
    max_restarts=2,      # default is 0: an actor that crashes stays dead
    max_task_retries=1,  # default is 0: at-most-once method semantics
)
class Engine:
    def __init__(self, model_path):
        # Runs once per actor lifetime, and again from scratch on every restart.
        self.llm = load_engine(model_path)

    def generate(self, prompt, max_tokens=256):
        return self.llm.generate(prompt, max_tokens)

engines = [Engine.remote("/models/qwen2.5-7b") for _ in range(4)]
results = ray.get([e.generate.remote(p) for e, p in zip(engines, prompts)])

Why each one is there: num_gpus=1 fixes the device set. num_cpus=2 is load-bearing — an actor reserves 1 CPU at schedule time, so an actor with num_cpus=0 inside a placement group still demands a CPU the bundles may not contain, and the actor hangs forever with “cannot schedule actor… requires 1 CPU”. max_restarts=2 opts into restart; Ray re-runs the constructor and restores nothing else. max_task_retries=1 upgrades method calls from at-most-once to at-least-once, which is why a retried generate may produce a second sample: your code must be idempotent or deduplicated.

3.3 A real placement group for a multi-GPU gang

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

WORLD = 8

# One bundle per rank. STRICT_PACK => all 8 land on ONE node or the group is
# never created. Under the default PACK they may split 4+4 across two nodes,
# and the NCCL cost jumps by an order of magnitude (Section 2.5).
pg = placement_group(
    [{"CPU": 2, "GPU": 1}] * WORLD,
    strategy="STRICT_PACK",
)
ray.get(pg.ready())      # blocks until all bundles are placed

@ray.remote(num_gpus=1, num_cpus=2)
class TrainRank:
    def __init__(self, rank, world):
        self.rank, self.world = rank, world
    def setup(self, master_addr, master_port):
        import torch.distributed as dist
        os.environ["MASTER_ADDR"] = master_addr
        os.environ["MASTER_PORT"] = str(master_port)
        dist.init_process_group("nccl", rank=self.rank, world_size=self.world)
        return dist.get_rank()

ranks = [
    TrainRank.options(
        scheduling_strategy=PlacementGroupSchedulingStrategy(
            placement_group=pg,
            placement_group_bundle_index=r,        # pin rank r to bundle r
            placement_group_capture_child_tasks=True,
        )
    ).remote(r, WORLD)
    for r in range(WORLD)
]
ray.get([r.setup.remote("127.0.0.1", 29500) for r in ranks])

Every parameter:

• strategy="STRICT_PACK" — the hard guarantee. PACK would still usually co-locate, which is exactly what makes the failure intermittent.
• placement_group_bundle_index=r — pins this actor to reserved slot r. Omit it and Ray schedules onto a random bundle with free reserved resources — documented behaviour, and a nondeterministic rank-to-GPU mapping.
• placement_group_capture_child_tasks=True — by default child tasks do not inherit the parent’s placement group. A rank that spawns a helper without this gets the helper scheduled outside the gang and then blocks on it forever.
• master_addr / master_port — the rendezvous. Ray is the process launcher; torch.distributed does the collectives. Nothing about NCCL is Ray’s job.
• Bundle index is not GPU rank. Ray does not promise that adjacent bundles map to adjacent physical GPUs; if NVLink topology matters, read back the assignment and verify it.

3.4 A real Ray Serve deployment, then the LLM layer

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

@serve.deployment(
    num_replicas=2,
    max_ongoing_requests=32,     # default is 5; raise it for anything real
    ray_actor_options={"num_gpus": 1, "num_cpus": 2},
)
class Model:
    def __init__(self, preprocess: DeploymentHandle):
        self.pre = preprocess
    async def __call__(self, request):
        x = await self.pre.remote(request)     # await; NEVER .result() here
        return self.predict(x)

@serve.deployment
class Preprocess:
    async def __call__(self, request):
        return tokenize(request)

serve.run(Model.bind(Preprocess.bind()))

max_ongoing_requests is the highest-value knob and the most common bug: it caps concurrent in-flight requests per replica, and its default changed from 100 to 5 in Ray 2.32.0. A GPU replica with a default of 5 is idle most of the time. await handle.remote(...) yields a DeploymentResponse; calling .result() inside a deployment blocks the replica’s event loop and defeats the whole concurrency model.

Now the LLM layer. ray.serve.llm wraps a vLLM (or SGLang) engine in a Serve deployment and adds an OpenAI-compatible ingress:

from ray import serve
from ray.serve.llm import LLMConfig, build_openai_app

llm_config = LLMConfig(
    model_loading_config=dict(
        model_id="qwen-7b",                                # what clients pass as "model"
        model_source="Qwen/Qwen2.5-7B-Instruct",           # HF id or local path
    ),
    accelerator_type="H100",
    engine_kwargs=dict(
        tensor_parallel_size=2,      # world_size = TP x PP; needs 2 GPUs in one bundle set
        max_model_len=8192,
        enable_prefix_caching=True,  # REQUIRED for the prefix-aware router to do anything
    ),
    deployment_config=dict(
        autoscaling_config=dict(min_replicas=1, max_replicas=4),
    ),
)

app = build_openai_app({"llm_configs": [llm_config]})
serve.run(app, blocking=True)     # OpenAI client at http://localhost:8000/v1

• engine_kwargs is passed straight through to vLLM, so any vLLM engine argument is valid here. That is the cleanest mental model of the boundary: Ray schedules and routes; vLLM does continuous batching and PagedAttention.
• world_size = tensor_parallel_size * pipeline_parallel_size; LLMServer reserves {CPU: 1} for the replica actor plus world_size x {GPU: 1} bundles, using a PACK strategy that prioritises TP ranks on one node.
• deployment_config.num_replicas is only used with autoscaling_config absent; the two cannot both be numeric.
• Prerequisite worth memorising: prefix-aware routing is a no-op unless you set enable_prefix_caching=True in engine_kwargs. The router routes to a cache that is not being built.

The two cache-aware routers, and their status [as of 2026-10]:

PrefixCacheAffinityRouter first checks load balance (imbalanced_threshold, default infinity — effectively always balanced unless you set it), then routes to the highest prefix match above match_rate_threshold (default 0.1), else to the replica with the lowest prefix-cache utilisation. KVAwareRouter needs vLLM-native KV block events, direct streaming, and NVIDIA Dynamo’s selection service; it supports one model per application on that path. Treat both as alpha and pin the version.

4. Numbers and trade-offs

4.1 The overhead model, with Ray’s own measurement

Ray’s documented measurement of per-task overhead, on a 2018 13-inch MacBook Pro with num_cpus=4, using an empty task — a two-line definition whose measured overhead was 0.4739549160003662 ms — is our $t_o$:

@ray.remote
def no_work(x):
    return x

From Tips for first-time users [as of 2026-10]. Write $t_o=0.47$ ms. For $N$ independent tasks each taking $s$ seconds of work, submitted at once across $W$ workers:

$$T\approx \frac{N(t_o+s)}{W},\eta =\frac{s}{t_o+s}$$

• $T$ — wall clock. $N$ and $W$ appear in it; the useful fraction $\eta$ does not. Parallelism cannot amortise per-task overhead. Adding workers makes the job finish sooner; it does not stop 90% of the CPUs’ time being spent scheduling. That is the non-obvious consequence, and why batching is the only real fix.
• At $s=0.1$ ms, $\eta =0.1/(0.47+0.1)=17.5\%$; at $s=10$ ms, $\eta =95.5\%$. Overhead is 5% of the work only at $s\approx 9$ ms, which is why “a task needs to take at least a few milliseconds” understates the low end.

Ray’s end-to-end demonstration is the same effect at scale: 100,000 tasks of 0.1 ms each took 13.4 s sequentially, 27.5 s as individual Ray tasks, and 3.25 s batched 1,000-per-task. Note what that says: 27.5 s of wall clock for 100,000 tasks is roughly 0.275 ms per task of achieved throughput cost, not 0.47 ms, because tasks overrun each other across workers. The 0.47 ms is per-task latency overhead; using it as the amortisation unit above is the right use and slightly conservative.

Labelled [UNVERIFIED]: no published equivalent exists for a no-op actor method call. Do not quote 0.47 ms for actor calls — the path differs (the worker process already exists, so no lease and no process start).

4.2 The cost of moving large objects

Implicit ray.put per call is O(size) per call. Ray’s documented example passes a $5000\times 5000$ np.zeros — $25\times 10^6$ float64 values, i.e. 200 MB — to 10 no-op tasks:

Deriving the transfer rate from the first row: $200\text{MiB}=0.2097$ GB over 107.5 ms is about 1.95 GB/s, a plausible localhost memcpy figure. The doc’s own summary of the fix is “about 8 times faster”, and its second, arguably more important point is that the direct pattern fills object-store memory with nine extra copies and drives eviction — so the performance number understates the operational damage.

Zero-copy read-only tensors. With RAY_ENABLE_ZERO_COPY_TORCH_TENSORS=1 (off by default) Ray’s own example summing a 1 GiB tensor goes 23.5 s -> 7.93 s, a 66.3% latency reduction. Two caveats that belong with the number: it is a single-machine microbenchmark from the docs, not a general claim, and the tensors are effectively shared, so a write by one process can be observed by another. That is a correctness hazard, not just a performance feature.

A caveat I could not close. No current Ray doc states a network throughput figure for cross-node object transfer, so do not invent one; measure your own interconnect.

4.3 Where the memory goes, with a worked number

Suppose a node reports 64 GiB of available memory. Ray splits it by default: memory (the logical heap budget, schedulable) gets 70% and object_store_memory gets 30%, and neither can be changed after the node starts.

$$\text{object\_store}=0.30\times 64\text{GiB}=19.2\text{GiB}$$

and the spill check fires when primary objects exceed

$$0.8\times 19.2\text{GiB}=15.36\text{GiB}.$$

Relate that to what you are actually passing around. A 7B model in bf16 is $7\times 10^9\times 2\text{B}=14\text{GB}$ of weights, and the KV cache for the same model (8 KV heads, $d_{head}=128$, 32 layers, fp16) is

$$2\times 32\times 8\times 128\times 2\text{B}=131\text{KiB per token}$$

so 8 GiB of cache holds roughly $8\times 1024^2/131\approx 64,000$ tokens (example). The comparison that matters is the first one: a single copy of the 7B model’s weights is roughly 85% of the node’s spill threshold ($14\text{GB}/15.36\text{GiB}\approx 0.85$). Moving a 14 GB tensor through the object store is not a transfer, it is an eviction event waiting to happen.

4.4 Which axis a decision moves

5. Pitfalls

Holding an ObjectRef pins object-store memory until the job ends. A task that closes over an ObjectRef “pins the object through reference counting, so Ray doesn’t evict the object until the job completes” — the docs’ own words. The subtler half: pinning prevents eviction, not spilling, because a primary copy with reference count 1 and no active consumer is spillable. Symptom: object-store usage climbs monotonically over a long job and never falls; ray.status() shows high object_store_memory with no obvious owner; eventually workers die with OutOfMemory in tasks that never allocate much. Fix: do not capture refs in long-lived closures; pass the value to a long-lived actor and let the actor hold the bytes.

ray.get inside a loop destroys pipelining. Submitting four tasks that take up to 4 s each, with 1 s of post-processing each: 7.83 s when you bulk ray.get, 4.85 s when you drive the same work with ray.wait. Symptom: nvidia-smi shows GPUs alternating between busy and idle in a sawtooth; wall clock is close to the sequential estimate while CPU utilisation is low.

The actor-holding-CPU trap. Actors reserve 1 logical CPU at schedule time and 0 while running. A placement group whose bundles are pure {"GPU": 1} with an actor that does not set num_cpus will hang forever. Symptom: the actor never starts; logs say “cannot schedule actor… requires 1 CPU”; ray status shows no pending demand because the group itself was created. Fix: set num_cpus explicitly on every actor, or add CPU to the bundle.

Infeasible placement groups produce NO warning. This is a real trap, not a figure of speech: Ray does not print anything when a bundle can never be satisfied — for example {"CPU": 4} on 2-CPU nodes, or STRICT_PACK across more GPUs than any single node has. Symptom: your program blocks in ray.get(pg.ready()), or in the first actor creation, and the only output is silence. Diagnosis, in order:

ray status
# Demands:
# {'CPU': 1.0} * 1, {'GPU': 2.0} * 1 (PACK): 1+ pending placement groups
ray list placement-groups --detail   # bundle states (PENDING/CREATED) and node_ids
ray list actors                      # required_resources should include CPU_group_<pgid>

The demand line is the signal; the CPU_group_<pgid> pseudo-resource on the actor is the proof that it is really inside the group rather than merely configured to be.

Placement-group deadlock, two rungs. Rung one: the driver creates a group that reserves all of some resource and then schedules a task needing more of it outside the group — it blocks holding the reservation. Rung two: because child tasks do not inherit the parent’s group by default, a rank inside a STRICT_PACK group spawns a helper that lands elsewhere and then waits on it forever. Symptom for rung two: one rank of eight never prints its init line; the other seven are blocked inside init_process_group. Fix: placement_group_capture_child_tasks=True, or PlacementGroupSchedulingStrategy(placement_group=None) to deliberately opt a child out.

Head-node inline-object OOM. Objects under RAY_max_direct_call_object_size (default 100 KiB) bypass the object store and live in the submitting process’s memory. When a shuffle emits thousands of tiny shards, those shards accumulate in one process — usually the driver on the head node — and OOM it. Symptom: the driver dies with a Python MemoryError while ray status shows plenty of free object-store memory on every node; the number of tasks is large and each return value is small. Fix: lower the threshold so the data travels through the spillable object store.

ray.put inside a short-lived task creates an object that outlives its owner. If that task’s worker later dies, every subsequent ray.get fails with OwnerDiedError. Symptom: a pipeline works all day and then, at hour nine, ray.get starts raising OwnerDiedError on refs that were fine before. Fix: return values from tasks, not refs to objects you ray.put inside them, and keep long-lived objects owned by long-lived actors.

Using a CUDA context as a free resource. Ray does not reuse worker processes across GPU tasks by default (max_calls=1 for GPU tasks) because GPUs leak memory on process exit. Symptom: GPU task throughput is lower than expected and the gap is per-task scheduling, not compute. This is a cost of correctness, not a bug — set max_calls=0 only if you have verified your code does not leak.

6. Using it in a project

Reach for Ray when all three of these are true: the workload is heterogeneous (more than one resource type), dynamic (replica counts or batch shapes that change), and stateful (something must stay resident across calls). If any one is missing, there is usually a smaller tool.

When NOT to use Ray.

• Small single-node jobs. A job that fits on one machine and runs in under a minute with sub-millisecond units of work is faster and simpler with concurrent.futures or multiprocessing. Ray adds a scheduler, an object store, a GCS, and roughly 0.47 ms per task to save you from a library you already have. Batch your work first; if batching makes the job trivial, you did not need Ray.
• Fixed homogeneous training. One model, fixed world size, a standard FSDP/DDP recipe: torchrun is the right answer. It is one process launch, no scheduler, no object store, and it fails and restarts as a single unit — which is exactly the failure behaviour you want when partial failure is not survivable anyway. Ray earns its place when a CPU reward stage and a GPU trainer share a job, or when the rollout engine must survive independently.
• Single-model single-node serving. For one model on one machine with steady traffic, run vLLM directly. Ray Serve adds a controller, a proxy, an autoscaler, a router, and a Python actor per replica; none of it improves the token rate of a single replica, and every hop is latency. Add Serve when you need at least one of: queue-depth autoscaling, multi-node TP/PP with gang scheduling, cache-aware routing across replicas, multi-model composition, or LoRA multiplexing.
• A cluster you cannot operate. The head node is a single point of failure by default: the GCS is in-memory and not fault tolerant — “If it fails, the entire Ray cluster fails.” Making it durable means operating an HA Redis (RAY_REDIS_ADDRESS) or attaching a durable volume to the alpha RocksDB backend, and the officially supported Ray Serve path additionally requires KubeRay. On top of that: dashboard metadata growth, ray status and state-API fluency as a debugging skill, placement-group deadlocks, and a failure model that is not uniform — task max_retries=3 but retry_exceptions=False; actor max_restarts=0; Ray Train max_failures=0. Each is a deliberate opt-in that a naive deployment forgets.

Version churn is a real cost. Over the window these notes were compiled: verl moved from volcengine/verl to verl-project/verl; max_ongoing_requests’s default changed at 2.32.0; RayServeHandle was removed at 2.10; Ray Train’s restore APIs were deprecated at 2.43; and KVAwareRouter, PrefixCacheAffinityRouter, topology strategy, and RocksDB GCS are all alpha. Pin versions and re-verify any API claim before it goes in a design doc.

A note on RLlib. Ray’s own RL library has no first-class LLM RL support [as of 2026-10] — there is no LLM or RLHF path in its docs or package. The LLM post-training story in the Ray ecosystem lives entirely in veRL, OpenRLHF, and SkyRL. Do not design around “Ray’s RLHF support”; see RL Frameworks.

7. How you would prove it worked

Ray has no metric of its own. What you prove is that a distribution decision helped, against a named baseline. Three experiments that hold up:

1. Overhead amortisation, measured against multiprocessing. Take your real unit of work and run it at three granularities: per-item Ray tasks, batched Ray tasks, and a multiprocessing.Pool with the same batching. Metric: wall clock and CPU-seconds per unit of work. Prediction from Section 4.1: the per-item run is dominated by the $t_o$ term and loses badly; the batched run beats the pool only once the work is heterogeneous (GPU + CPU stages) or stateful. (example) A tokenisation stage at 0.1 ms/item shows ~17% efficiency per-item and ~99% batched; if the pool matches the batched number, Ray bought nothing and you should say so.

2. Gang scheduling, ablated by strategy. Same 8-rank job, run once with PACK and once with STRICT_PACK, on a cluster with 4-GPU nodes. Metrics: ray list placement-groups --detail (did it split?), and per-step time. The instructive outcome is the split run: it works and is several times slower, with no error anywhere. Reporting that number is the point of the experiment.

3. Cache-aware routing, ablated. Serve two replicas of the same model with a shared prefix-heavy workload. Run three configurations: default router, prefix affinity router with enable_prefix_caching=True, and prefix affinity with the flag off (the negative control). Metric: median time-to-first-token, plus the engine’s prefix-cache hit rate. (example) The third arm is what makes the result believable — it is exactly the configuration people ship by accident, and it should reproduce the default router’s number.

Reporting template (example). “Moving the verifier from per-sample Ray tasks to 1,000-sample batches raised the reward stage’s useful CPU fraction from 17% to 94% and cut end-to-end step time by 31%; the STRICT_PACK ablation showed the 8-rank group split 4+4 under the default PACK and step time rose 3.4x, so we kept the strict strategy and sized nodes to 8 GPUs.”

8. Interview drill

“What is the difference between a task and an actor?” Strong answer: a task is a stateless function whose execution process need not persist; an actor is a stateful process, created once, kept alive with its resources held. Only an actor can hold a CUDA context, model weights, and a KV cache across calls — which is why every RL framework puts an actor in front of each GPU and uses tasks for scoring and tokenisation. Follow-up: “failure semantics?” → tasks retry on system failure with max_retries=3 by default; actors have max_restarts=0 by default and do not restore state when restarted. Common wrong answer: “actors are threads.” They are separate processes, often on different nodes, with independent memory.

“What is an ObjectRef and why is it not just a future?” Strong answer: it is a future-like handle — an ID plus metadata — but the value lives in a distributed object store, has an owner whose lifetime determines whether it is recoverable, and can be read zero-copy on the node holding the primary copy. If the owner dies, Ray deletes remaining copies and later ray.get raises OwnerDiedError, with no opt-in to recover. Common wrong answer: “it’s a lazy pointer to the driver’s memory.” Data on another node must be pulled across the network; there is no remote dereference of another machine’s heap.

“Where do Ray objects actually live, and when does a copy happen?” Strong answer: the primary copy is on the node that produced it, pinned by the owner through PinObjectIDs, not evictable and only spillable. A secondary copy is a cached replica on another node, held in Plasma only while referenced and evicted by LRU afterwards. A copy is made when a task on another node needs the value as an argument, or when ray.get is called there. Follow-up: “which does Ray evict first, and why?” → secondary copies, because they can be refetched from the primary; only if pressure persists are primary copies spilled. Common wrong answer: “holding a Python reference keeps the object in memory.” It prevents eviction; it does not prevent spilling.

“What is a placement group, and why must a bundle fit on one node?” Strong answer: a bundle is a resource dict; a group is a list of bundles plus a strategy, and creation is atomic — all bundles are reserved or none are. A bundle must fit in one node because it is the unit of atomic co-reservation, and single-node co-residency is the only thing Ray can guarantee at reservation time. For a tensor-parallel group this is hard, because an all-reduce needs every rank on every collective and crossing a node boundary can cost more than an order of magnitude. Common wrong answer: “I need 8 GPUs so I ask for 8 GPUs.” Eight {"GPU": 1} bundles under the default soft PACK is routinely satisfied 4+4 across two nodes, and the job is several times slower with no error.

“I made a placement group with pure {"GPU": 1} bundles and my actor hangs forever. Why?” Strong answer: actors reserve 1 logical CPU at schedule time and 0 while running. The actor’s scheduling requirement includes CPU: 1, the bundles do not contain a CPU, so it can never be placed — and infeasible placement groups produce no warning. Fix: set num_cpus explicitly, or add CPU to the bundle. Diagnosis to describe: ray status for the demand line, ray list placement-groups --detail for bundle state, ray list actors for the CPU_group_<pgid> pseudo-resource.

“My task raised an exception and Ray didn’t retry it. Why not?” Strong answer: retries cover system-level failures — worker death, node death — for which max_retries defaults to 3. A Python exception from user code is not retried unless you set retry_exceptions=True or pass the exception classes. That gap is why “Ray retries tasks” is only half true. Follow-up: “and the actor equivalent?” → max_task_retries=0 is at-most-once by default, so you can get a RayActorError for a method that actually executed.

“I lost a worker node. Which of my objects are gone?” Strong answer: objects whose owner was a worker on that node are gone permanently — Ray deletes remaining copies to avoid a leak, and later ray.get raises OwnerDiedError. For the rest, Ray reconstructs by lineage: find a copy elsewhere, else re-execute the producing task and its arguments recursively. Follow-up: “what are the limits?” → four: the object must have been produced by a task, not ray.put; tasks must be deterministic and idempotent, and actor-task outputs are not reconstructable by default; re-execution is bounded by the retry budget; and the owner must still be alive. Common wrong answer: “Ray replicates objects so nothing is lost.” There is no replication of the primary copy; there is spilling, and there is lineage.

“Why do veRL, OpenRLHF, and SkyRL all use Ray instead of torchrun?” Strong answer: their jobs mix resource shapes and need a rendezvous. Rollout engines, trainers, critics, and reference models are separate stateful roles with different parallelism; some rewards are CPU sandboxes; and the rollout engine must be schedulable independently of the trainer. Ray supplies actors over GPUs, placement groups for gang scheduling, and the MASTER_ADDR/MASTER_PORT rendezvous for torch.distributed in one layer. Where that is not true, Ray is overhead. See RL Frameworks. Common wrong answer: “Ray makes the training faster.” Ray does not perform the all-reduce; the speedups come from placement (colocation), overlap, and cheaper weight synchronisation — decisions Ray makes expressible, not operations Ray performs.

“Does Ray Serve LLM give me anything over plain vLLM?” Strong answer: not throughput on a single replica. It adds gang-scheduled multi-node TP/PP, elastic replica counts with queue-depth autoscaling, cache-aware routing across replicas, multi-model multiplexing, and prefill/decode disaggregation as deployment classes. If you have one model that fits one GPU and constant traffic, plain vLLM is simpler and has one fewer distributed system. Follow-up: “what do the cache-aware routers require?” → they are alpha, and prefix-aware routing is a no-op unless enable_prefix_caching=True is set in engine_kwargs; KVAwareRouter additionally needs direct streaming and NVIDIA Dynamo’s selection service.

“How does a KV-aware router decide which replica is least loaded?” Strong answer: it maintains a global index of each replica’s KV blocks from vLLM’s own block creation and eviction events, and scores each replica by token load: remaining prefill work after KV-block overlap, plus active prefill, plus decode work weighted by remaining output. GPU-resident blocks score full credit and CPU-offloaded blocks score less. The view is eventually consistent, which is the assumption to state out loud.

Next: What an Agent Actually Is — Ray is the substrate under the training and serving stacks; an agent is the control loop that sits on top of a served model.