Part I - Core mental model

Actors, State, and Services

Lesson 02 gave you the stateless task: a function shipped to a worker, returning an ObjectRef immediately so the driver can fan out thousands of calls. But a task is born empty and dies with its return value — it cannot remember anything between calls, cannot hold a model that took 30 seconds to load, and cannot be a long-lived endpoint other code talks to. Some of the most common ML shapes — a parameter server, a model replica, a simulator, a connection pool — are defined by the state they carry. This lesson adds Ray's second primitive, the actor: a stateful worker that owns a Python object for its whole lifetime, and whose single-threaded execution model is both a correctness guarantee and the throughput ceiling you will spend the rest of the lesson designing around.

1 · What a task cannot do

A task is a function: it receives arguments, runs, returns a value, and the worker that ran it is free to run anyone else's task next. That statelessness is exactly what makes tasks easy to parallelize and retry — Ray can re-run a failed task anywhere because it depends only on its inputs (lineage, lesson 04). But three real needs fall outside what a stateless function can express:

An actor is a class marked @ray.remote. Instantiating it (Counter.remote()) creates one long-lived worker process that constructs the object and holds it. That instance lives on one worker, owns a Python object, and survives between calls. Method calls — counter.increment.remote() — are shipped to that specific process and return ObjectRefs exactly like task calls, so actors compose into the same dependency graph tasks do. The reference you get back from .remote() on the class is an actor handle: a first-class value you can pass to other tasks and actors so they can call the same instance.

The mental model: a task is a function call shipped to whatever worker is free; an actor is a remote object with a mailbox. Methods queue in that mailbox and run on the actor's process, one at a time, against state that lives there. Everything good and everything painful about actors follows from "one process owns the state."

2 · Two actors: a parameter server and a model replica

The smallest useful actor is a shared mutable counter — the seed of a parameter server. State lives in self; methods read and mutate it; the single-threaded execution model (next section) means the increments cannot race, so you get an atomic counter across the whole cluster for free:

import ray
ray.init()

@ray.remote
class ParameterServer:
    def __init__(self):
        self.params = 0.0            # mutable state, owned by THIS process

    def apply_gradient(self, grad):  # called concurrently by many workers
        self.params -= 0.01 * grad   # cannot race: methods run one at a time
        return self.params

    def get(self):
        return self.params

ps = ParameterServer.remote()        # 'ps' is the ACTOR HANDLE
refs = [ps.apply_gradient.remote(g) for g in gradients]  # queue many updates
ray.get(refs)                        # each ran serially against shared state
final = ray.get(ps.get.remote())     # read the converged value back

The second canonical actor solves the expensive init problem. The model is loaded once in __init__ — which runs exactly once, when the actor is created — and every predict call reuses the warm, already-on-GPU object. @ray.remote(num_gpus=1) reserves one GPU for this actor's entire lifetime, so the weights stay resident:

@ray.remote(num_gpus=1)              # reserve a GPU for this actor's lifetime
class ModelReplica:
    def __init__(self, weights_path):
        self.model = load_model(weights_path)   # ~30 s, paid ONCE at creation
        self.model.eval()

    def predict(self, batch):
        return self.model(batch)     # warm: no reload, just the forward pass

replica = ModelReplica.remote("s3://bucket/model.pt")
# the 30 s load happens in the background while we prepare inputs
preds = ray.get(replica.predict.remote(batch))

3 · The concurrency model is a single thread

A plain actor processes one method call at a time, on a single thread. Calls arrive in the mailbox, run to completion in submission order per caller, and the next one starts only when the current one returns. This is the source of the consistency guarantee in §2 — no two apply_gradient calls interleave, so you never need a lock — but it is equally the source of a hard throughput ceiling.

Put a number on it. If a method takes 20 ms, a single actor can serve at most 1 / 0.020 s = 50 calls/s — and that is the ceiling on a 1-node laptop and on a 500-node cluster alike. Suppose 40 worker tasks each want to push 100 gradients/s to one parameter server: that is 40 × 100 = 4,000 calls/s of demand against a 50 calls/s actor. The mailbox grows without bound, latency climbs, and the "fast" cluster is gated by one serialized process. You have rediscovered that a shared mutable object is a global lock.

Two escape hatches exist within a single actor for the case where the method is I/O-bound (waiting on network/disk, not burning CPU):

4 · Break the ceiling: pools and sharded state

When one actor cannot keep up with CPU-bound demand, run a fixed pool of N actors and spread calls across them. Each actor is its own process on its own core, so N actors give you roughly N× the throughput — turning the 50 calls/s ceiling into ≈ N × 50 calls/s. The simplest pool is round-robin over a list of handles:

# N independent model-replica actors, each on its own GPU/core
N = 8
pool = [ModelReplica.remote("s3://bucket/model.pt") for _ in range(N)]

# round-robin: call i goes to actor i % N — work spreads across N processes
refs = [pool[i % N].predict.remote(b) for i, b in enumerate(batches)]
preds = ray.get(refs)

Ray also ships ray.util.ActorPool, which load-balances by handing each new task to whichever actor is free rather than blind round-robin — better when call durations vary, since a slow actor does not stall its share of the queue:

from ray.util import ActorPool
pool = ActorPool([ModelReplica.remote(path) for _ in range(N)])
# map_unordered assigns each input to the next AVAILABLE actor
results = list(pool.map_unordered(lambda a, b: a.predict.remote(b), batches))

That last point is the design pivot. When the bottleneck is state, you do not pool identical actors; you shard the state by a natural key so that each actor owns a disjoint slice and calls for different keys run in parallel. A single global counter becomes K counter shards keyed by, say, user_id % K:

K = 16
shards = [ParameterServer.remote() for _ in range(K)]

def update(user_id, grad):
    shard = shards[hash(user_id) % K]     # this user's state lives on ONE shard
    return shard.apply_gradient.remote(grad)
# distinct users hit distinct shards → up to K× the single-actor throughput,
# while each user's own updates stay serialized (still race-free per key)

Sharding buys throughput but costs you global atomicity: a sum across all shards is now a fan-in (read all K, add them), and you can no longer take a single lock across the whole state. That is the recurring actor trade-off — centralize for correctness, shard for scale, and you cannot have both for the same piece of state.

5 · The two costs: a single point of failure and a serialization point

You bought capabilities in §1; here is the bill. An actor concentrates state in one process, which means two things break in ways tasks do not.

State dies with the process. A task's output can be reconstructed by re-running it (lineage). An actor's accumulated state — the replay buffer, the converged parameters, the warmed cache — cannot, because re-running __init__ gives you a fresh empty object, not the state you had. If the node holding the actor dies, that state is gone. max_restarts tells Ray how many times to restart a crashed actor (e.g. @ray.remote(max_restarts=3)), but a restart re-runs __init__ and starts empty unless you persist and reload state. Durability is your job:

@ray.remote(max_restarts=3)          # Ray will recreate the actor up to 3 times
class ReplayBuffer:
    def __init__(self, ckpt="s3://b/buf.pkl"):
        self.ckpt = ckpt
        self.data = load_if_exists(ckpt) or []   # reload on restart, else empty

    def add(self, item):
        self.data.append(item)

    def checkpoint(self):                # call periodically from the driver
        save(self.ckpt, self.data)       # durability is YOUR responsibility

It is still a serialization point. Even fault-tolerant, one actor is one thread (§3). The two costs compound: the actor you made durable with checkpointing is the same actor whose 50 calls/s ceiling you must respect. Production designs therefore reach for a pool of restartable, periodically-checkpointed actors sharded by key — fault tolerance and throughput at once — which is precisely the shape Ray Serve (lesson 11) and RLlib (lesson 07) package so you do not assemble it by hand each time.

Failure modes & checklist

Checkpoint exercise

Where this points next

You now have both primitives — stateless task, stateful actor — and the handle that lets them call each other. But we have been hand-waving the machinery underneath: when you fire predict.remote(batch), which node runs it, how does the 4 MB batch get there without copying it through the driver, where does the model replica's GPU reservation actually come from, and what reconstructs a lost object? Lesson 04 opens the runtime: the driver/worker/raylet/GCS layout, the resource-and-locality-aware scheduler that placed your actors, and the plasma object store whose ownership and lineage model is what made ObjectRef work in the first place.

Interview prompts

• What can an actor do that a task cannot, and what does it cost you? (§1, §5 — hold mutable state, reuse expensive init, be a service endpoint; in exchange it's a single point of failure whose state isn't durable and a single-threaded serialization point.)
• An actor method takes 20 ms — what's the max throughput of one actor on a 500-node cluster? (§3 — 1/0.020 = 50 calls/s; cluster size is irrelevant because one actor runs methods serially on one process.)
• When does an actor pool help and when does it not? (§4 — N actors give ≈ N× throughput when calls are independent; it does NOT help when the bottleneck is one shared mutable state, which stays serialized — then shard the state by key instead.)
• Why doesn't a plain actor need a lock around shared state? (§2, §3 — single-threaded execution runs one method at a time, so mutations can't interleave; this guarantee is lost the moment you set max_concurrency>1.)
• What does max_restarts actually recover, and what does it not? (§5 — it recreates a crashed actor up to N times, but re-runs __init__ from scratch; accumulated state is lost unless you checkpoint and reload it yourself.)
• Async actor vs threaded actor vs more actors — when each? (§3, §4 — async/threaded (max_concurrency) overlap I/O-bound waiting on one process but the GIL blocks CPU work; CPU-bound demand needs more actor processes, i.e. a pool.)
• How do you choose a checkpoint interval for a stateful learner? (§5 — trade steady-state overhead (checkpoint cost / interval) against worst-case loss (work done since last checkpoint); shorter = safer but slower, longer = cheaper but loses more on a crash.)