Docs

Running a cluster

The same binary runs on a laptop and on a cluster. On a cluster it takes its role from the environment: one head, any number of workers, and drivers that connect to the head. Dista starts no machines or pods itself; Kubernetes, Orca or a script of yours does.

Locally

Runtime[...](workers=n) with nothing in the environment forks a head and n workers (default 4) and drives them from the same process, which is how tests and laptops run. The forked head is the same code a cluster runs.

Roles

Processes start in any order: workers and drivers retry the head for DISTA_CONNECT_TIMEOUT seconds. A worker gives its peers the address it reaches the head from, or DISTA_ADVERTISE. Each hello carries a hash of the program's tasks, so a cluster can't mix builds.

The generic head

The head doesn't run your code, so it can be the generic one, dista head, from the dista command. It serves drivers and workers built from any program, several at once, and gives each driver's tasks and actors only to workers built from the same program. A head built into your program, DISTA_ROLE=head, serves only that program.

dista head --port 6390 &                            # the head, for any program
DISTA_ROLE=worker DISTA_HEAD=headhost:6390 ./app    # once per worker
DISTA_HEAD=headhost:6390 ./app                      # the driver

Kubernetes

The repository's deploy/ folder builds one image for every role, with your program at /opt/dista/bin/app, its entrypoint, and the command at /opt/dista/bin/dista. deploy/k8s/cluster.yaml runs the head as dista head in a StatefulSet behind the headless Service dista-head, and the workers as a Deployment, one per node, with DISTA_ADVERTISE set to the pod IP. A driver is a Job with DISTA_HEAD=dista-head:6390 (deploy/k8s/driver-job.yaml). tests/kind.sh runs it on a kind cluster of three nodes, in CI too.

Each pod has its own /dev/shm, so large results cross pods over sockets. Give the workers' /dev/shm room for the results they keep, and set DISTA_STORE_BYTES below its size limit: Kubernetes evicts a pod that passes it rather than failing the write. The AWS manifests keep 1 GiB per worker. A pod that's deleted gets SIGTERM, and its worker drains.

EC2

The repository's bench/cloud runs Dista on EC2 machines with k3s, for the performance numbers. Orca's ec2 runner starts a run's head and workers as instances of their own, and with orca register --workers MIN..MAX it scales them while the run goes; Orca's docs say how.

Draining a worker

A worker sent SIGTERM, as Kubernetes stops a pod, drains instead of dying. The head sends it no more tasks, it finishes what it was sent, the other workers adopt its results, and an actor it hosts moves, restored from a checkpoint. Then it exits 0: nothing fails or runs twice. DISTA_DRAIN_SECONDS (default 25) bounds the drain; past it the worker leaves anyway, and what it still holds is lost. A second SIGTERM ends it at once. The head itself exits at SIGTERM.

dista drain <ip:port> drains a worker by its address, as dista status shows it, for a provider that can't signal the worker's process. The head refuses to drain the last worker while tasks are queued or running.

Autoscaling

A job's workers can grow while tasks wait and shrink once they're idle. Dista says how many workers it wants and which ones may go, and whatever owns the machines starts and stops them. Three pieces, each usable alone:

• Demand. dista status --json prints the head's tasks ready, waiting for inputs and running, its actors waiting for a worker, and per worker its address, node, tasks in flight, seconds idle, bytes held, actor and whether it's draining. /metrics has dista_tasks_ready, dista_tasks_waiting, dista_tasks_running and dista_workers_idle.
• A policy. decide(demand, policy, now, pending, short_since), from dista.scale, says how many workers to start and which to drain, from Demand.parse(status_json) and a Policy. It has no clock or I/O of its own.
• Draining by name. dista drain, above.

Policy holds min_workers (default 1, and it may be 0), max_workers (4), tasks_per_worker (0, the head's pipeline depth), up_after (5 s: tasks ready that long with no free slot start workers), idle_before_down (60 s) and max_step (4, the most workers started at once). Idle workers drain fewest held bytes first, never below min_workers; a worker hosting an actor drains only with a checkpoint store, and the last worker only once no task is left and it holds nothing. Workers asked for but not yet joined are passed in as pending, so a slow boot isn't asked for twice.

dista scale --once prints the decision for the head's current demand, a line each: want N, start N, drain ADDRESS per worker to drain, and short_since T to pass back next time. A provider loop runs it, starts workers and drains the ones it names; tests/autoscale.sh does that in shell, from 0 workers to 6 and back to 0. For a long-lived cluster on Kubernetes, a HorizontalPodAutoscaler on the head's dista_queue_depth works too.

The environment

Every variable Dista reads. The other pages say more about each: Failure and recovery for the timeouts and retries, Tuning for the store and the limits.