Drivers and workers connect to the head and say hello with their role and
registry hash. A head built into a program refuses a peer whose hash isn't
its own, so every process runs the same tasks. A generic head (dista head,
registry hash 0) takes any, and hands a driver's tasks and actors only to
workers built from the same program, so one head serves several programs.
Observers (dista status, timeline and stop) say hello without a hash. A driver gets a number, which becomes the top bits of
every object ID it creates, so the head knows where each result goes.
The loop polls the listener and every connection. Drivers send tasks, ready
to run or waiting on other results; the head queues them, hands them to
workers round-robin up to pipeline in flight each, and forwards each result
or error to the driver that owns it. A task whose inputs live mostly on one
node goes to a worker there when one is free (see _pick). Every socket is non-blocking with its
own output queue (Conn), so a slow peer can't stall anyone else.
Small results come through the head, which keeps them, because a task
submitted later may take one as input. A large result stays on the worker
that computed it: the head records where, tells the owning driver, and sends
a task that needs it as task-inputs, so the worker running that task fetches
it from its peer. A worker that goes away loses the results it held: dista
doesn't rebuild them (that's Ray's lineage reconstruction), so a task that
needs one fails, naming it, and an orchestrator runs the job again. The
tasks it was running are sent to another worker, up to DISTA_TASK_RETRIES
times (default 3), unless an input they read from that worker is gone too.
Submissions are admitted, not just accepted. While the tasks queued or
waiting on inputs reach DISTA_MAX_QUEUED (default 100,000), or their
arguments DISTA_MAX_QUEUED_BYTES (default 1 GiB), the head stops reading
from drivers. Their sockets fill, and remote blocks until workers drain the
queue: backpressure through TCP, with no frame of its own. Clients inside a
worker's task are read regardless, since pausing one would hold that worker,
and with every worker held, nothing would drain. 0 turns a limit off.
A driver that doesn't read what the head sends it, busy elsewhere, would
grow the head's output buffer by every result. Past DISTA_MAX_DRIVER_BUFFER
bytes waiting for it (default 256 MiB), its queued tasks aren't sent to
workers until it reads; its queue then fills, and admission control stops its
submissions. A driver waiting to submit reads meanwhile (see Client), so
this doesn't deadlock it.
Every object is reference counted: its driver holds it until it releases it
(or disconnects), and each submitted task taking it as input holds it until
that task settles. At zero the head forgets the object and tells the worker
holding it, if one does, to delete it. An object released before its task
finishes is freed when the task settles.
The head keeps nothing on disk. When it restarts, its state comes back from
its peers: workers rejoin and report the results they hold, and each driver
reconnects under its old number and restores, claims or resubmits everything
it still needs (see frame.mojo). A reported result no resumed driver
claimed is freed. A task that was running when the head died may run again.
A worker told to stop drains (see worker.mojo). The head sends it no more
tasks and has the other workers adopt every result it holds: each fetches
its copy, the head records where the result is now, and tells the driver
that owns it. A task sent the old location still finds the result, since the
draining worker keeps serving until it's gone. Once the worker runs no task
and holds nothing, the head lets it go, and it leaves without anything
failing or running again. An actor on a draining worker moves: the head
holds its new calls, asks for a checkpoint after the ones it sent, and once
that's written, restores the actor on another worker and sends the held
calls there, so none runs twice. Without DISTA_CHECKPOINT_URI the
checkpoint fails, and the actor restarts when its worker leaves, replaying
its calls, as for one lost.
An actor is placed on a worker that hosts no other, the one with the fewest
tasks in flight, as soon as it's created; the worker builds it once those
tasks finish. From then on the head sends that worker only the actor's calls,
each as it arrives, so they run in order. An actor created when every worker
hosts one waits for a worker to join or an actor to be killed. Killing an
actor, or its driver leaving, frees its worker for tasks.
The head keeps every call made to an actor since its last checkpoint. When
the actor's worker goes away, the actor is placed on another worker, built
again from that checkpoint, or from its constructor if it has none, and
those calls are run again in order, so its state is rebuilt: calls run at
least once. Each call is answered once: a call answered before the restart
keeps its first answer, and the replay's is dropped.
A checkpoint is asked for every checkpoint_every calls, if the actor was
created with one, and whenever a driver asks. The head sends it to the
actor's worker between calls, numbered by the calls it covers, and once the
worker says it's written, forgets the calls it covers, and tells each
caller the newest of its calls the checkpoint includes.
After a head restart, a resuming driver creates its actors again. The head
restores each from its newest checkpoint in DISTA_CHECKPOINT_URI, which
the head lists, or builds it if there's none, and the calls the callers
resend run on that: every call no checkpoint includes, answered or not.