# Running a cluster

> Running Dista: local mode, roles from the environment, the generic dista head, Kubernetes manifests and EC2, draining a worker on SIGTERM or by name, autoscaling with dista scale, and every environment variable.

[magmalake](https://magmalake.dev/) [Dista](https://magmalake.dev/s/dista)

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](https://magmalake.dev/s/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

| Environment | The process is |
| --- | --- |
| DISTA\_ROLE=head | runs the head on DISTA\_PORT and never returns |
| DISTA\_ROLE=worker | serves the head at DISTA\_HEAD=host:port and never returns |
| DISTA\_HEAD=host:port | a driver that submits to that head |
| neither | local mode: forks a head and workers workers, then drives them |

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](https://magmalake.dev/s/dista/docs/cli). 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/](https://github.com/magmalake/dista.mojo/tree/main/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](https://github.com/magmalake/dista.mojo/tree/main/bench/cloud) runs Dista on EC2 machines with k3s, for the [performance numbers](https://magmalake.dev/s/dista/performance). 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](https://magmalake.dev/s/orca/docs/run#runners) 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](https://magmalake.dev/s/dista/docs/failures) for the timeouts and retries, [Tuning](https://magmalake.dev/s/dista/docs/tuning) for the store and the limits.

| Variable | What it sets | Default |
| --- | --- | --- |
| DISTA\_ROLE | head or worker; unset, the process is a driver | driver |
| DISTA\_HEAD | host:port of the head, for a driver or a worker in a cluster, and for dista's observer commands | local mode |
| DISTA\_PORT | the port a head built into the program listens on; 0 lets the OS pick, and the head prints it | 6390 |
| DISTA\_ADVERTISE | the address a worker gives peers and drivers to fetch its results from | the address it reaches the head from |
| DISTA\_NODE | the node a worker says it's on, for placement | its host name |
| DISTA\_CONNECT\_TIMEOUT | seconds a driver or worker keeps retrying the head, and an observer waits for it | 60 |
| DISTA\_PEER\_CONNECT\_TIMEOUT | seconds, fractions allowed, a worker or driver waits to connect to a worker holding a result | 5 |
| DISTA\_KEEPALIVE\_SECONDS | seconds of silence before a connection is probed; it fails after 3 unanswered probes. 0 turns keepalive off | 10 |
| DISTA\_TASK\_TIMEOUT | seconds a task may run after it was sent before it fails | no limit |
| DISTA\_TASK\_RETRIES | how many times a task whose worker died runs again | 3 |
| DISTA\_ORPHAN\_SECONDS | seconds a head keeps the results workers report for a driver that hasn't come back, after a head restart | 120 |
| DISTA\_DRAIN\_SECONDS | how long a draining worker may take before it leaves anyway | 25 |
| DISTA\_CHECKPOINT\_URI | where actor checkpoints go: a directory, file://, or s3://bucket/prefix | none: checkpoints fail |
| DISTA\_STORE\_DIR | where a worker writes large results for its host to map | /dev/shm on Linux, else $TMPDIR |
| DISTA\_SHM | 0 keeps large results in the worker's memory instead | on |
| DISTA\_STORE\_BYTES | bytes of results a worker keeps in the store before it spills | no limit |
| DISTA\_SPILL\_DIR | where results past the store go, on disk | $TMPDIR, else /tmp |
| DISTA\_MAX\_QUEUED | tasks queued or waiting at the head before remote blocks; 0, no limit | 100,000 |
| DISTA\_MAX\_QUEUED\_BYTES | bytes of their arguments before remote blocks; 0, no limit | 1 GiB |
| DISTA\_MAX\_DRIVER\_BUFFER | bytes waiting for a driver that isn't reading before its tasks wait; 0, no limit | 256 MiB |
| DISTA\_WARN\_HELD | objects a driver holds before the head logs it, and again at each doubling; 0, never | 100,000 |
| DISTA\_METRICS\_PORT | the port a head or worker serves Prometheus metrics on, at /metrics | off |
| DISTA\_DEBUG | 1 logs each task's inputs as a worker assembles them | off |

[A magmalake project](https://magmalake.dev/)

Apache 2.0 · © 2026 Marius Seritan