Part I - Core mental model

Runtime, Scheduling, and the Object Store

Lessons 02 and 03 gave you the two primitives — stateless tasks returning ObjectRef futures, and stateful actors — and told you the object store "carries values between them." That was a promise made on faith. This lesson opens the box: the processes that actually run on each node, the scheduler that decides where a task lands, and the shared-memory object store that lets a 4 GB array sit in one place while ten workers read it. Once you can name where work is placed, where its bytes live, and what is actually scarce, almost every Ray performance question becomes a calculation rather than a mystery.

1 · Four processes, four responsibilities

Stop picturing a cluster as one big laptop. It is a set of nodes — separate machines (or pods) with their own CPU, GPU, RAM, disk, and a network between them. On every node Ray runs a small set of processes, and knowing which one owns what is the difference between guessing and debugging.

The split matters because it tells you who to suspect. A task that won't schedule is a raylet/GCS resource question — no node advertises the resources you asked for. A program that mysteriously stalls is usually the driver blocking on ray.get. An "actor not found" is the GCS actor registry. And bytes that won't move are the object manager half of a raylet. Most of this lesson lives inside the raylet.

2 · How the scheduler decides where work lands

When you call f.remote(x), the task does not run immediately — it is queued, and the scheduler must choose a worker on some node. Three ideas drive that choice.

Resources are a contract, not enforcement. A task declares what it needs: @ray.remote(num_cpus=2, num_gpus=1). Ray treats num_cpus, num_gpus, and any resources={"...": n} as logical resources — a bookkeeping budget the scheduler will not overcommit. It is crucial to internalize that these are a promise about scarcity, not a sandbox. Declaring num_cpus=2 does not pin your task to two cores or stop it spawning thirty threads; it only means the scheduler will subtract 2 from that node's CPU budget and refuse to co-place more tasks than the node advertises. Under-declare and you oversubscribe the hardware (your "2-CPU" tasks each secretly use 8 and the node thrashes); over-declare and the node sits half-idle because the budget says "full" while the cores are not. Resources are how you tell Ray what is scarce.

Locality-aware placement. Among nodes that satisfy the resource contract, the scheduler prefers a node that already holds the bytes of the task's arguments in its local object store — because then the task reads them with no network transfer. If preprocess.remote(block_ref) can run on the node where block_ref's value already lives, that is strictly cheaper than copying the block across the network first.

Spillback. When the preferred (local, or resource-rich) node is full — no free worker matching the resource request — the local scheduler does not queue indefinitely. It spills back: hands the task to the GCS / another raylet to place on a different node that has capacity. Spillback is what keeps the cluster busy instead of letting one hot node become a queue, at the cost of giving up locality for that task.

3 · The Plasma object store: shared memory, immutability, ownership

Every task result and every ray.put(value) lands in the node's object store — Ray's implementation is called Plasma, an in-memory store backed by shared memory. Three properties make it the load-bearing component of the whole runtime.

Zero-copy reads on the same node. "Zero-copy" means exactly what it says: a large object is written once into a shared-memory segment, and every worker process on that node maps the same physical pages (via mmap) into its own address space rather than receiving a private copy. Ten workers reading a 4 GB array consume 4 GB of RAM total, not 40 GB, and the "read" is a pointer remap that takes microseconds — no deserialization, no memcpy of the payload. This is precisely why you ray.put a big shared value once and pass its ref instead of passing the value into each task (the serialization trap from lesson 00).

Objects are immutable. Once written, an object's bytes never change. Immutability is what makes zero-copy safe — many readers can share the same pages because no one can mutate them out from under another — and it is what makes a value cacheable and reconstructible. To "change" a value you produce a new object with a new ref.

Distributed reference counting and the ownership model. Ray tracks how many references exist to each object across the whole cluster; when the count hits zero, the object is evicted. But counting needs an authority, and that authority is the owner: the worker that created the ref — by submitting the task that returns it, or by calling ray.put — owns that object's metadata and its lineage (the task and arguments that produced it). The owner, not the GCS, holds the reference count and the recipe.

Lineage-based reconstruction. Here is what ownership buys you. Because the owner remembers the task and inputs that produced an object, a lost object (its node crashed, evicting the bytes) can be rebuilt by simply re-executing the task that produced it — recursively, if its inputs were lost too. You did not have to checkpoint the intermediate value; the task graph is the recovery plan. This is automatic for task results whose inputs are still reproducible, and it is the foundation of Ray's task fault tolerance — the full failure model, including where reconstruction cannot help (actor state, side effects, ray.put objects whose owner died), is lesson 14.

import ray
ray.init()

# A large, read-only lookup table the whole job shares.
# ray.put writes it into Plasma ONCE; we pass the ref, not the bytes.
lookup_ref = ray.put(load_lookup_table())   # e.g. a 4 GB embedding table

@ray.remote(num_cpus=2, num_gpus=1)          # the resource CONTRACT
def score(block, table):
    # Ray auto-dereferences a top-level ObjectRef arg before the task runs,
    # so `table` is the actual object — read zero-copy on this node.
    return model_score(block, table)

# Pass the SAME ref to every task. On a node holding `lookup_ref`'s bytes,
# all workers mmap the one copy (4 GB total), not one copy per task.
refs = [score.remote(b, lookup_ref) for b in blocks]
results = ray.get(refs)                       # block once, here

4 · Spilling, and why object-store memory is its own budget

The object store has a fixed size — by default Ray reserves about 30% of a node's RAM for Plasma, separate from the Python heap your task code allocates. This separation is the first thing to internalize: object-store memory ≠ process heap. A node with 64 GB RAM gives Plasma ≈ 19 GB; the rest is for your workers' own allocations. You can blow the heap (a worker OOMs) while the object store is empty, or fill Plasma while the heap is fine — they are tracked and exhausted independently.

Spilling is what happens when the object store fills: rather than failing, Ray writes least-recently-used objects out to disk (or remote storage) to make room — and reads them back when a task needs them. This keeps the job correct, but it introduces a latency cliff, because disk is orders of magnitude slower than the shared-memory read it replaced.

5 · Placement groups: gang-scheduling a bundle of resources

Sometimes independent placement is wrong. If you need four GPU workers that must be co-located — say tensor-parallel ranks of one model that exchange activations over the fast intra-node link — you cannot let the scheduler scatter them across nodes and hope. A placement group reserves a set of resource bundles atomically (gang scheduling: either all bundles are placed or none are) and lets you control their spread.

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

# Four bundles, each 1 GPU + 2 CPUs, all PACKed onto as few nodes as
# possible (STRICT_PACK would force a single node) — co-locate TP workers
# so their activations cross the intra-node link, not the slow network.
pg = placement_group([{"GPU": 1, "CPU": 2}] * 4, strategy="PACK")
ray.get(pg.ready())                              # gang: all 4 or none

@ray.remote(num_gpus=1, num_cpus=2)
class TPWorker:
    def step(self, shard): ...

workers = [
    TPWorker.options(
        scheduling_strategy=PlacementGroupSchedulingStrategy(
            placement_group=pg, placement_group_bundle_index=i)
    ).remote()
    for i in range(4)
]

The strategies trace the locality-vs-utilization tension again: PACK (and STRICT_PACK) maximize co-location for fast communication but can fail to schedule if no node is empty enough; SPREAD distributes bundles for fault isolation and balanced load at the cost of inter-node communication. Placement groups are how Ray Train and RLlib reserve co-located GPU workers under the hood (lessons 07, 10).

6 · Failure modes & checklist

Checkpoint exercise

Where this points next

You now know how one task is placed, how its bytes are shared, and what happens when the store fills. But real workloads are not one task — they are thousands, arranged in a shape: fan out a map, exchange data in a shuffle, and fold results back in a reduce. That shape stresses every mechanism in this lesson at once — locality during the map, the object store during the shuffle, and the spill cliff if the reduce can't keep up. Lesson 05 builds map-shuffle-reduce on top of this runtime, shows why naively ray.get-ing everything onto the driver is an anti-pattern, and gives the tree-reduce that shrinks the serial fraction Amdahl punished you for in lesson 00.

Interview prompts

• Name Ray's four runtime processes and what each owns. (§1 — driver: control flow; worker: runs a task or hosts an actor; raylet: per-node local scheduler + object manager; GCS: cluster metadata on the head node.)
• Does num_cpus=2 stop a task using more than two cores? (§2 — no; logical resources are a scheduling contract preventing overcommit, not a sandbox. Under-declaring oversubscribes the hardware; over-declaring idles capacity.)
• 4 GB input over 10 GbE: when do you move the task to the data vs the data to the task? (§2 — transfer ≈ 3.2 s; with 5 s compute that's a 64% penalty so favor locality; with 600 s compute it's ~0.5% noise so favor utilization and let spillback place it.)
• What does "zero-copy" mean in the object store, and why is immutability required for it? (§3 — workers mmap the same shared-memory pages instead of each getting a private copy; immutability makes shared readers safe and the object reconstructible.)
• What is the ownership model and what breaks when an owner dies? (§3 — the worker that created a ref owns its reference count and lineage; if it dies the object is unreconstructible even if the bytes still exist, and dependents fail with owner-died.)
• How does Ray recover a lost (evicted) task result without checkpointing it? (§3 — lineage-based reconstruction: re-execute the task that produced it, recursively, since the owner remembers the recipe; sets up the lesson 14 failure model.)
• Your job suddenly slows down mid-run with no code change. What runtime cause do you check first? (§4 — object-store spill: live objects exceeded the ~30%-of-RAM Plasma budget and are now read from disk, a 10,000×+ latency cliff. Reduce live objects / block size, overlap producer-consumer.)
• Why would you use a placement group, and what's the PACK vs SPREAD trade-off? (§5 — to gang-schedule co-located bundles, e.g. TP GPU workers on the fast intra-node link; PACK maximizes co-location (may fail to schedule), SPREAD gives fault isolation at inter-node communication cost.)

Connective tissue

The object store is a shared-memory cache with eviction and a disk-spill tier — the same storage-engine ideas (write-once immutable values, LRU eviction, spilling to a slower tier) that databases formalize. And the locality-vs-transfer math is the cluster interconnect from the systems-ML view: 10 GbE vs NVLink changes every break-even in this lesson.