Part I - Core mental model

Tasks, Objects, and Futures

Lesson 01 argued why Ray exists: ordinary Python hits a wall, and the runtime's job is to spread arbitrary functions and objects across a cluster while the dependency graph is still being discovered at runtime. This lesson writes the first real program against that runtime and meets the one object everything else is built on — the ObjectRef. The thesis is small and load-bearing: f.remote(x) does not run a function and wait; it submits a node into a graph and returns a future immediately. Once you see that .remote() returns instantly and ray.get is the only place that blocks, the difference between a program that parallelizes and one that quietly runs serially comes down to where you put the ray.get.

1 · .remote() returns a future, not a value

Start from the line of code that changes everything. An ordinary function call is blocking: control does not return until the function has computed its answer.

def score(record):
    return expensive_model(record)      # ~50 ms of CPU work

result = score(records[0])              # BLOCKS for ~50 ms, then you hold the value

Mark the same function as a task — a stateless unit of work Ray can ship to any worker — by decorating it with @ray.remote, and call it with .remote(...). Now the call returns immediately, before any computation has happened, handing you an ObjectRef: a future, a typed promise that a value will exist at that address once some worker finishes producing it.

import ray
ray.init()                              # start or connect to a local cluster

@ray.remote                             # this function is now a task
def score(record):
    return expensive_model(record)      # still ~50 ms of CPU work

ref = score.remote(records[0])          # RETURNS NOW — no work has run yet
# type(ref) is ray.ObjectRef — a handle, not the answer
value = ray.get(ref)                    # THIS line blocks until the value exists

The difference is the entire mental model. The blocking call interleaves "submit work" and "wait for work" into one statement, so you can only ever have one piece of work in flight. The .remote() call separates them: you can fire thousands of tasks — accumulating thousands of ObjectRefs — before you block on any of them. Those tasks run concurrently across the cluster's workers while your driver (the process running your main()) keeps moving. Asynchrony is not a feature bolted onto Ray; it is Ray's parallelism, and it is bought entirely by .remote() refusing to wait.

# the canonical shape: a wave of submissions, then ONE block
refs = [score.remote(r) for r in records]   # 100,000 refs, returns in milliseconds
results = ray.get(refs)                      # block once, here, for the whole wave

2 · ray.get blocks — and is where bytes come home

ray.get(ref) does two things, and both matter. First it blocks the calling process until the value is ready. Second, it is the point at which the value is deserialized back into the driver's heap — the bytes that lived in the object store are pulled to the caller's node (if not already local) and reconstructed into a real Python object. That second cost is invisible until your results are large: pulling a million 1 KB results back is moving a gigabyte through one process.

Because ray.get is the only blocking primitive in this lesson, where you place it is the single most consequential decision in a Ray program. The classic mistake is to place it inside the submission loop:

# ANTI-PATTERN: ray.get inside the loop — accidental serialization
results = []
for r in records:                       # 100,000 records
    ref = score.remote(r)               # submit one task...
    results.append(ray.get(ref))        # ...then immediately wait for it
# every iteration submits ONE task and blocks for it: zero overlap.
# total time ≈ 100,000 × 50 ms = 5,000 s ≈ 83 min — same as a serial loop,
# now PLUS per-task scheduling overhead. You paid for Ray and got nothing.

# FIX: submit the whole wave, collect refs, get ONCE
refs = [score.remote(r) for r in records]   # all 100,000 in flight
results = ray.get(refs)                      # one block; tasks overlap across workers
# with 50 worker cores: ≈ (100,000 / 50) × 50 ms = 100 s instead of 5,000 s — ~50× faster.

The worked contrast is the lesson in one stroke. The anti-pattern's ray.get is a synchronization barrier that re-imposes the very serialization Ray was meant to remove — it is the most common reason "I added @ray.remote and nothing got faster." The fix moves the single barrier to the end, after all work is submitted, so the 50 ms tasks run in parallel and the only serial part left is the final gather. Synchronization points define the shape of parallelism; minimize and delay them.

3 · ray.put — store a big value once, share its ref

When a task takes an argument, Ray must get that argument's bytes to whatever worker runs the task. For a small argument this is trivial. For a large argument reused by many tasks, the default behavior is wasteful: each .remote(big) call serializes big again and puts a fresh copy in the object store.

weights = load_model_weights()          # a 400 MB numpy array, used by every task

# WASTEFUL: each call re-serializes and re-stores the same 400 MB
refs = [predict.remote(weights, x) for x in batch]   # 1,000 tasks
# Ray implicitly puts `weights` up to 1,000 times → up to 400 GB of
# serialization + object-store churn for data that never changes.

ray.put(value) stores the value in the object store once and returns an ObjectRef you can hand to as many tasks as you like. Every task then shares the single stored copy — and tasks scheduled onto the same node read it zero-copy from shared memory, so passing the ref to 1,000 same-node tasks costs one materialization, not 1,000.

# RIGHT: put the big value once, pass the shared ref
weights_ref = ray.put(weights)          # 400 MB serialized ONE time → one ref

@ray.remote
def predict(w, x):                      # `w` arrives already dereferenced (see §4)
    return w @ x

refs = [predict.remote(weights_ref, x) for x in batch]   # 1,000 tasks share one copy
preds = ray.get(refs)

4 · The dependency graph — passing refs builds edges

You do not build Ray's task graph by calling a scheduler API. You build it by passing one task's ref as another task's argument. When Ray sees a task whose argument is an ObjectRef, it records an edge: the downstream task depends on the upstream one and cannot be scheduled until that ref's value exists. The driver never has to materialize the intermediate value at all.

@ray.remote
def parse(path):
    return load_rows(path)              # returns a big table

@ray.remote
def summarize(rows):                    # takes the table directly
    return compute_stats(rows)

parsed   = [parse.remote(p) for p in paths]          # stage 1: N refs
summaries = [summarize.remote(ref) for ref in parsed] # stage 2: each depends on its parse
stats = ray.get(summaries)              # block once at the very end

Notice that summarize.remote(ref) is passed an ObjectRef, not a table. Two things follow. First, the table never travels through the driver — Ray can hand the parsed bytes directly to the worker that runs summarize (and prefers to run it on the node that already holds them; locality scheduling is lesson 04). Second, inside summarize the parameter rows is the actual table, not a ref: Ray auto-dereferences a top-level ObjectRef argument before the task body runs.

@ray.remote
def good(rows):                 # rows is the TABLE — Ray dereferenced it
    return compute_stats(rows)

@ray.remote
def trap(batch):                # batch = [ref, ref, ref] — refs are NOT dereferenced
    rows = ray.get(batch)       # YOU must get them inside the task
    return [compute_stats(r) for r in rows]

good.remote(parse.remote(path))          # ✓ top-level ref → auto-dereferenced
trap.remote([parse.remote(p) for p in paths])  # ✗ list of refs → still refs inside

5 · Granularity and ray.wait — the two knobs that decide your speedup

Submitting a task is not free. The runtime spends on the order of a few hundred microseconds to ~1 ms per task routing it, reserving resources, registering the result ref, and tracking the dependency. That overhead is fixed per task and runs against the work the task does. So the first knob is granularity: how much real work each task performs.

# batch many items per task so overhead is amortized
@ray.remote
def score_batch(records):               # 10,000 records per call
    return [expensive_model(r) for r in records]

chunks = [records[i:i+10_000] for i in range(0, len(records), 10_000)]
refs = [score_batch.remote(c) for c in chunks]   # 100 tasks, not 1,000,000
results = [y for part in ray.get(refs) for y in part]

The second knob is which results you wait for, and when. ray.get(refs) blocks until every ref is ready — so a single straggler (a task that runs 10× slower than the rest) holds the whole gather hostage, and your driver sits idle while the fast 99% of results pile up unused. ray.wait fixes this: it returns as soon as some tasks are done, splitting your list into (ready, not_ready) so you can process completed results immediately and loop on the rest. This pipelines work past the long tail.

# stream results as they finish — do not block on the slowest task
pending = [score_batch.remote(c) for c in chunks]   # 100 tasks in flight
while pending:
    # return as soon as 1 is done; 1 s safety timeout so we never hang
    ready, pending = ray.wait(pending, num_returns=1, timeout=1.0)
    for ref in ready:
        handle(ray.get(ref))            # consume each result the instant it lands

Two practical notes. Always pass a timeout to ray.wait in a loop so a dead or hung task can never block you forever. And raise num_returns when you want to batch consumption — num_returns=10 waits for ten results before returning, trading a little latency for fewer loop iterations. This first-N-done shape is exactly what RLlib uses to collect rollouts and what Ray Tune uses to harvest finished trials; you are learning the primitive those libraries are built on.

6 · Failure modes & checklist

Checkpoint exercise

Where this points next

Tasks are stateless: each one starts cold, and nothing survives between calls. That is exactly wrong for two common needs — a model you want to load once and reuse across thousands of calls, or a long-lived service that owns mutable state. Re-ray.putting or re-loading per task pays the initialization cost every time. Lesson 03 introduces the actor: a stateful worker process that owns a Python object, loads its model in __init__ once, and answers method calls against that retained state — at the cost of becoming a single-threaded bottleneck and a single point of failure. The object store and the ref machinery you learned here carry straight over; what changes is who holds the state.

Interview prompts

• What does f.remote(x) return, and why is that the source of parallelism? (§1 — an ObjectRef (future) immediately, before any work runs; separating submission from waiting lets the driver fire thousands of tasks that run concurrently instead of blocking call-by-call.)
• You wrapped a function in @ray.remote and it got no faster. What is the first thing you check? (§2 — a ray.get inside the submission loop, which blocks on each task and re-serializes the program; fix by collecting all refs and calling ray.get once.)
• When and why would you use ray.put? (§3 — for a large value reused by many tasks; it serializes and stores once, returns a shared ref, and same-node tasks read it zero-copy, versus implicitly re-serializing the value per call.)
• How is Ray's dependency graph built, and what gets auto-dereferenced? (§4 — by passing one task's ObjectRef as another task's argument, which creates an edge; a top-level ref argument is auto-dereferenced before the task runs, but a ref nested in a list/dict is not — the task must ray.get it.)
• Your "parallel" job is slower than a serial loop. What is likely wrong and how do you fix it? (§5 — tasks are too fine, so ~0.5 ms scheduling overhead dominates 50 µs of work; batch many items per task so each runs tens of ms and overhead is a few percent.)
• Why prefer ray.wait over ray.get when task durations vary? (§5 — ray.get(all) idles the driver until the slowest task; ray.wait returns the ready subset so you process the fast majority and overlap handling with the straggler, keeping the driver busy.)
• What does an ObjectRef actually point to, and what does holding one guarantee? (§1 — a cheap address into the immutable object store, not a copy; the value is computed at most once, and a live ref keeps the object reference-counted and not evicted.)