Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
A document-processing job finishes three tasks, and its counter reports 3. Restart the counter's process and it reports 0. The task results may still exist. Restarting a worker and recovering its application state are different operations, even when both sit behind the same remote-call API.
The last chapter treated DeepSpeed as an engine inside a distributed training job. Ray handles a wider execution problem: where to run preprocessing functions, stateful workers, and serving replicas, and how to connect their results. Its three central abstractions are tasks, actors, and object references.[1]
Use a tiny document job to understand those abstractions before considering GPUs. The local lab is pinned to Ray 2.58.0 and Python 3.12. Architecture details below use that release's documentation, checked September 2, 2026; newer releases can change internal classes and defaults.[2]
Three pieces of a running job
A task is a remote function invocation. Each call has inputs, a return value, and its own execution. An actor is a remote class instance whose methods can read and update state kept in its process. An ObjectRef is a future-like handle to a result, not the result's bytes and not an instruction to place work on a particular machine.
| Primitive | In the document job | What survives another call? |
|---|---|---|
| Task | features.remote("cache") | Its returned object while reachable; no promised worker-local state |
| Actor | BatchLedger.remote() | The actor's counter while that process survives |
| Object reference | The handle returned by a task or actor method | A dependency on that result, subject to ownership and lifetime rules |
Ordinary synchronous actor methods execute one at a time. That's not a global ordering guarantee across arbitrary callers, and async or threaded actors have different concurrency behavior. Choose an actor because you need an identifiable state owner, not because every call should become a long-lived process.[3]
The job's “vectors” are just deterministic string features: length and character-code sum modulo 101. They aren't learned embeddings. Computing them locally first keeps a distributed failure from hiding an arithmetic mistake.
1def features(text: str) -> tuple[int, int]:
2 return len(text), sum(map(ord, text)) % 101
3
4vectors = [features(text) for text in ["cache", "router", "trace"]]
5assert vectors == [(5, 96), (6, 67), (5, 22)]
6print(vectors)1[(5, 96), (6, 67), (5, 22)]The calculation is too small to benefit from remote execution, but its readable output makes Ray's behavior easier to check. Real applications should batch work so serialization and scheduling overhead don't dominate useful computation.
Keep the dependency remote
Decorating features with @ray.remote(num_cpus=1) gives it a .remote(...) entry point. Submission returns an ObjectRef before the result is ready. A downstream task or actor can accept that reference without the driver first fetching its value.
This excerpt from the downloadable lab submits all three tasks and sends their references as separate, top-level arguments to the ledger:
1refs = [features.remote(text) for text in ["cache", "router", "trace"]]
2ledger = BatchLedger.remote()
3count = ray.get(ledger.record.remote(*refs), timeout=20)
4assert count == 3Ray resolves each top-level reference before record runs, so the actor receives three tuples. The driver resolves only the small count. By contrast, ledger.inspect_nested.remote(refs) passes a list containing references: nesting them inside a container doesn't automatically replace them with values. A consumer can decide when to resolve those nested references.[1]
Moving a ray.get inside the submission loop changes more than memory use. ray.get(features.remote(text)) waits before submitting the next task, serializing the pipeline. Submitting a list of references first preserves concurrency, but gathering every result at once can create a large memory peak. These are separate mistakes: serial submission and unbounded collection.
References aren't free. They consume metadata and keep objects alive through distributed reference counting. The owner also retains lineage for reconstructable work. Passing handles avoids unnecessary driver materialization; it doesn't make retained data or bookkeeping disappear.[4]
Run the local process-failure lab
Download ray_core_lab.py. With uv installed, run it from the download directory:
1uv run --python 3.12 ray_core_lab.pyThe file pins ray==2.58.0 in its script metadata. It starts a new local runtime, not an existing cluster, with two logical CPUs, no GPUs, no dashboard, and an 80 MiB object-store budget. Those settings aren't an 80 MiB total-memory limit: Ray's control processes and Python workers use additional memory. Run it on a development machine with spare capacity, not on a shared production node.
The lab performs four checks:
- Reserve two one-CPU bundles with
STRICT_PACK, execute two tasks in those reserved bundles, then remove the group. - Pass three top-level references into the ledger and verify
completed == 3; also verify that references nested in a list remain references. - Kill the ledger process with restart enabled, wait until its process ID changes, and verify the fresh constructor's count is
0. - Deliberately exit a task worker once, then verify that its configured retry succeeds.
The last check uses a temporary local marker to make the crash happen once. That's a single-node fault-injection fixture, not a shared checkpoint or a distributed idempotency design. The lab doesn't kill the raylet or GCS, and it doesn't test lost-object reconstruction across machines.
A successful run emits this result, alongside Ray startup or deliberate-worker-failure diagnostics:
1{
2 "actor_restarted": true,
3 "count_after_restart": 0,
4 "logical_cpus": 2,
5 "nested_refs_preserved": true,
6 "nodes": 1,
7 "placement_group_tasks": 2,
8 "ray": "2.58.0",
9 "recorded": 3,
10 "worker_retry": "worker crash retried"
11}The script shuts down its runtime in finally and uses a temporary directory for runtime files. Individual waits have timeouts. It doesn't run ray stop, which could disrupt unrelated local Ray sessions. An abrupt process kill or machine failure can bypass Python cleanup; inspect remaining processes before removing runtime files.
Notice the restart test's waiting condition. ray.kill initiates termination; an immediate read can race with that operation and still return the old process's state. A count alone isn't enough evidence of a restart. The test waits for a new PID before checking the constructor value.
What schedules the work?
Each Ray node runs a raylet, which manages local worker execution and resources and participates in scheduling. The Global Control Service (GCS) maintains cluster-level metadata and operations, including actors, nodes, and placement groups. GCS isn't the process that executes your Python functions.[5]
Ordinary task scheduling can consider data locality and available resources. Actor creation involves cluster-level coordination and placement, but calling a method on an existing actor doesn't move that actor to a new node. The process already owns its state. If an actor needs a large remote argument, the argument must reach the actor rather than the actor following every input.
num_cpus=1 is a logical scheduling request. It doesn't pin a worker to one CPU core or prevent it from launching many native threads. GPU resource requests similarly guide assignment; they aren't GPU-memory reservations. Declare resources accurately and separately configure worker concurrency, native thread counts, memory budgets, and OS/container isolation.
An actor explicitly requesting one CPU holds that scheduling resource during its lifetime. On a two-CPU local runtime, that leaves one CPU slot for tasks. A waiting actor method doesn't automatically free the actor's reserved resource. This matters when composing worker pools: code can wait forever for children whose resource requests can't fit.
Reserve a group, then use its bundles
Suppose a training trial needs four one-GPU workers. Launching four unrelated actors can leave two occupying GPUs while the other two wait. A placement group requests the resources together. Its initial creation is atomic: either all requested bundles can be reserved or the group remains pending without a partial initial reservation.[6]
Each bundle must fit on one node. Four {"GPU": 1} bundles can span nodes; one {"GPU": 4} bundle can't be split across four one-GPU nodes. The strategy controls the relationship among bundles:
| Strategy | Placement requirement |
|---|---|
PACK | Prefer packing bundles together; allow multiple nodes if necessary |
STRICT_PACK | All bundles must fit on one node |
SPREAD | Prefer spreading bundles across nodes; reuse nodes if necessary |
STRICT_SPREAD | Each bundle must be on a different node |
For example, with two nodes containing two free GPUs each, four one-GPU bundles can satisfy PACK but not STRICT_PACK. Having four GPUs in aggregate doesn't satisfy a same-node requirement. Neither policy alone promises NVLink connectivity or a particular physical GPU ordering.
A reservation isn't automatic enrollment. Tasks and actors must use PlacementGroupSchedulingStrategy to consume its resources. The lab passes both the group and a bundle index explicitly. It also removes the group before creating unrelated work, so reserved CPUs aren't left unavailable to ordinary tasks.
Initial atomic placement isn't an all-or-nothing failure guarantee forever. If a node later dies, some bundles may remain alive while Ray tries to recover the lost ones. The application still needs a policy for partial worker-group failure.[6]
Two nodes each have two free GPUs. A model needs four GPUs on one node. Would four one-GPU bundles using PACK enforce that requirement?
Answer
No. PACK can spread the bundles across both nodes. STRICT_PACK expresses the same-node requirement and remains pending on this cluster. You need a node with enough available resources, a different model topology, or a different requirement.
Object bytes, owners, and memory
Large remote values usually live in node-local shared-memory object stores; small values can travel inline. An object may have copies on several nodes. Its ObjectRef doesn't identify a single permanent storage location.
ray.get makes a value available to the caller. Ordinary Python structures can require deserialization into the worker heap. Eligible NumPy arrays can instead use zero-copy views backed by local shared object-store memory. Such reads still retain memory and can require inter-node transfer first. “Every get copies the payload into the driver heap” and “references cost zero bytes” are both wrong simplifications.[1]
The owner is the process that originally creates the reference, usually by submitting .remote() or calling ray.put. For a task submitted by the driver, the driver usually owns the reference while a different worker computes its value. This distinction is central to the NSDI 2021 Ownership design, which decentralizes object metadata and reference counting.[7]
Three questions explain most recovery cases:
| What disappeared? | What Ray can try | What can prevent recovery? |
|---|---|---|
| One stored copy | Fetch another copy | No remaining accessible copy |
| All copies of a task result | Re-execute its generating task and dependencies | Dead owner, unavailable lineage/dependencies, exhausted retries |
| Actor process | Restart when configured | Constructor doesn't restore prior mutable state |
| Object owner | Surface owner failure | Ray doesn't recover that object's metadata from its surviving bytes |
Objects created with ray.put have no generating task to replay. That matters when all copies of their data are lost, even if the owner is alive. Owner loss is a separate limitation applying to task outputs too. Actor-method outputs aren't reconstructed by default; enabling method retries doesn't turn arbitrary actor mutations into deterministic computations.[4]
Ray's object store can spill data to disk under pressure. Worker heap usage is separate, and disk spilling adds I/O and disk-capacity requirements. Watch both. A zero-copy array held by a Python variable can keep an object alive even after another reference is deleted. Unlimited pending tasks can also consume memory before results exist.
For a large stream, cap in-flight work and use ray.wait to consume a bounded number of ready results. Refilling a fixed-size window bounds pending work; merely calling ray.wait after submitting a million tasks doesn't. For block-oriented processing, Ray Data provides a higher-level streaming executor rather than requiring you to manage every task reference.[8]
A retry can repeat an effect
The lab's fail_worker_once is configured with max_retries=1. Its first process exits; Ray retries that task and the second execution returns. This tests worker-failure retry, not lineage reconstruction of an already completed result. User-code exceptions have a separate retry policy and aren't automatically equivalent to process failure.
For ordinary tasks, repeatability and side effects need separate checks. A deterministic calculation can regenerate the same bytes. A deterministic function that appends a billing event can still append it twice. External effects need idempotency or a transactional protocol, and their return values must remain consistent enough for downstream computation.
Actors add two controls: max_restarts governs process recreation, while max_task_retries governs method retry. Both default to zero. An actor error doesn't prove the method had no effect: the process may fail after mutation but before the caller receives its result. Retrying a non-idempotent increment can double-count.[3]
A robust ledger would associate a stable batch ID with its recorded result and commit both the count change and deduplication record atomically to durable storage. Its constructor would restore state from that storage. A detached actor changes ownership/lifetime behavior, not durability: it isn't a replacement for checkpointing.
Libraries compose the same runtime
The 2018 Ray paper introduced a unified task-and-actor interface for dynamic AI workloads.[9] The modern libraries build higher-level workflows on that runtime, but their responsibilities differ:
| Library | Work it organizes | Work still owned elsewhere |
|---|---|---|
| Ray Data | Dataset blocks, transformations, batch inference and training ingest | Model semantics and correctness of transformations |
| Ray Train | Distributed training workers and coordination | Framework-specific forward/backward computation and collectives |
| Ray Tune | Trials, resource allocation, search and stopping decisions | Validity of the objective and evaluation split |
| RLlib | Reinforcement-learning sampling and learning components | Whether the reward and environment match the intended task |
| Ray Serve | Deployment replicas, request routing and scaling | Model-engine token scheduling and kernels |
Ray Data streams data through execution stages and applies backpressure to avoid unbounded upstream production. That doesn't mean every operator is purely streaming or every workload stays within a tiny memory footprint. A global shuffle, slow sink, or oversized batch can still dominate the job.[8]
Ray Train coordinates training workers and integrates with frameworks, reporting and checkpoints. It doesn't replace PyTorch's gradient computation, DeepSpeed's optimizer partitioning, or the communication backend's collectives. Persisting a checkpoint and deciding which state it must contain remain part of the training application.[10]
A library name isn't a reason to distribute a small script. Ray earns its operational cost when dynamic parallel work, heterogeneous resources, or composition across distributed stages solves a real bottleneck. Measure task granularity and end-to-end throughput before adding more workers.
Serve routes requests; the engine schedules tokens
A Serve deployment defines a component; each replica is a Ray actor running an instance of it. A deployment handle routes calls to replicas. For LLM serving, an ingress accepts HTTP requests, routing chooses an engine replica, and the engine batches and executes token work.
The 2.58.0 architecture docs describe OpenAiIngress, LLMServer, and an LLMEngine abstraction. For the standard vLLM layout, the server replica requests a CPU and creates a placement group for GPU workers; tensor-parallel size times pipeline-parallel size determines that layout's worker count. That's a particular engine topology, not a universal formula for every expert-parallel or disaggregated deployment.[11]
The integration boundary can move. Ray 2.58.0's release notes describe tokenization and KV-aware routing in LLMRouter, with engine-cache events informing replica choice. Therefore, “Ray never sees tokens or cache information” is too strong. The useful distinction is routing among engine instances versus scheduling and allocating state inside an engine.[2]
vLLM remains responsible for its continuous batching, KV-cache allocator and kernels. SGLang is another documented integration, with its own support and execution details. Don't assume every engine implements every Serve LLM feature identically. Validate the chosen model, engine, parallelism and package versions as a combination.
If inter-token latency rises, inspect both sides of the boundary: engine queueing and cache pressure, plus routing concentration, ingress CPU saturation and network behavior. A healthy HTTP endpoint doesn't establish healthy token generation, and a busy GPU doesn't prove the ingress is keeping up.
Control-plane recovery has its own state
GCS recovery isn't actor checkpoint recovery. In the default in-memory configuration, losing GCS loses control metadata. Ray 2.58.0 documents a supported external-Redis fault-tolerance setup for KubeRay with Ray Serve, and a Linux-only alpha embedded-RocksDB backend on persistent storage. Neither is enabled by declaring task retries.[5]
During a recoverable GCS outage, existing tasks, actors and objects can remain available while actor creation, placement-group operations and other control functions pause. Worker reconnection deadlines and the survival of backing storage still matter. Losing the entire head node or actor node is different from restarting only the GCS process.
Don't infer uninterrupted requests from a recovered control plane. Test the actual failure boundary: kill one actor, lose an object-store node, interrupt GCS, or lose its storage. Record what continued, what was retried, what became unavailable, and whether any external effect repeated. The local lab covers only worker and actor processes on one node.
Read the source with one question in mind
Use the ray-2.58.0 tag for this reading exercise. Start at python/ray/remote_function.py for task submission and python/ray/actor.py for actor handles. Follow the Python-to-CoreWorker boundary before opening src/ray/raylet/ and src/ray/gcs/. Keep two traces separate: “where does this execution run?” and “who owns this result's metadata?”[12]
For Serve, follow the public builder into its deployment, engine client and routing implementation. Compare code at the pinned tag with the version's architecture pages; a conceptual diagram can lag a refactor. Explain one request's path before trying to summarize the entire repository.
Ray originated at UC Berkeley; the 2018 paper records the initial research team. Anyscale's history dates Ray's development to 2016-2017 and its company founding to 2019.[9][13] Ray's governance describes contributors, committers, a technical steering committee and lead maintainers under LF Projects. Anyscale's commercial platform and the open-source project's technical governance aren't the same thing.[14]
Ray source is distributed under Apache-2.0. That source license doesn't grant rights to arbitrary model weights, datasets or other artifacts your job processes.[15] Runtime ownership of an ObjectRef is yet another meaning of “ownership”: a fault-tolerance role, not a copyright claim.
Evaluation rubric
- Explain the difference between a returned value, an object reference, and the process owning its metadata.
- Trace top-level and nested references through the local lab without claiming that handles use no memory.
- Distinguish logical resource declarations, bundle reservations and hard placement constraints.
- Use process identity and state together to verify actor restart, then explain why a restart isn't restoration.
- Separate the tested local failure cases from untested node loss, GCS recovery and GPU serving behavior.
Follow-up questions
How would you change BatchLedger so a retried batch can't increment the count twice after a restart?
Answer
Give each batch a stable ID. Atomically persist the count change and the ID-to-result record, then restore that durable state in the constructor. A retry returns the stored result. Keeping the ID only in actor memory loses deduplication on restart.
Why can a completed task result still become unavailable even when another node has its bytes?
Answer
The process owning the original ObjectRef also owns required metadata. Ray doesn't recover objects after owner failure merely from surviving copies. Keeping a reference in another process doesn't transfer ownership.
What would a two-node extension of the lab need to establish that its local run cannot?
Answer
Verify that producers and consumers actually ran on different nodes, observe object transfer, and inject a bounded node failure while preserving the owner. Record task replay and output equality. A single-node worker crash doesn't test network transfer, node loss, or object reconstruction.