# dista.scale

> A module of Dista's API.

How many workers a running job wants: a policy over the head's demand.

dista starts no machines. Whatever owns them (orca's `ec2` runner, a shell
loop, a Kubernetes operator) reads the head's demand, `dista status --json`,
asks `decide` what to do, starts that many workers and drains the ones it
names with `dista drain`:

    var d = Demand.parse(observer.status_json())
    var got = decide(d, Policy(min_workers=0, max_workers=12), now, pending,
                     short_since)
    # start got.start workers; drain each of got.drain; keep got.short_since

`decide` has no clock and no I/O: `now` is the provider's, in seconds, and
`short_since`, when the job first ran short of workers, comes back in the
decision for the next call. So it's tested on tables of demand over time.

Up: tasks ready with no free slot for `up_after` seconds start workers, at
most `max_step` at once, up to `max_workers`; with no worker at all, they
start at once. Workers asked for but not yet joined (`pending`) count as
there, so a slow boot isn't asked for twice.

Down: workers idle for `idle_before_down` seconds drain, fewest held bytes
first, down to `min_workers`, which may be 0. A worker hosting an actor
drains only with a checkpoint store to move it through; and the last worker
drains only once no task is left and it holds nothing, since its results
would have nowhere to go.

## Structs

### `WorkerDemand`

```mojo
struct WorkerDemand
```

One worker, as the head sees it.

Implements: `Copyable`, `Movable`

Fields:

- `address: String`: Its object server, `ip:port`: what `dista drain` takes.
- `running: Int`: Tasks in flight on it.
- `idle_seconds: Float64`: How long it's had nothing to do; 0 while it has.
- `held_bytes: Int`: Bytes of results stored on it.
- `actor: UInt64`: The actor it hosts, or 0.
- `draining: Bool`

### `Demand`

```mojo
struct Demand
```

The head's demand, from `dista status --json`.

Implements: `Copyable`, `Movable`

Fields:

- `tasks_ready: Int`: Tasks whose inputs are ready, waiting for a worker.
- `tasks_waiting: Int`: Tasks waiting for their inputs.
- `tasks_running: Int`
- `actors_waiting: Int`: Actors waiting for a worker of their own.
- `pipeline: Int`: Tasks the head keeps in flight per worker.
- `checkpoints: Bool`: Whether actors can move, through a checkpoint store.
- `workers: List[WorkerDemand]`

#### `__init__`

```mojo
def __init__(out self, tasks_ready: Int = Int(0), tasks_waiting: Int = Int(0), tasks_running: Int = Int(0), actors_waiting: Int = Int(0), pipeline: Int = Int(2), checkpoints: Bool = False, var workers: List[WorkerDemand] = List())
```

#### `parse`

```mojo
def parse(text: String) -> Self
```

Reads `dista status --json`'s object. Unknown fields are skipped, so a newer head's output still reads.

#### `live`

```mojo
def live(self) -> Int
```

Workers that aren't draining.

### `Policy`

```mojo
struct Policy
```

Implements: `Copyable`, `Movable`

Fields:

- `min_workers: Int`: May be 0: a job with nothing to do then runs no worker.
- `max_workers: Int`
- `tasks_per_worker: Int`: Tasks one worker takes at once; 0 for the head's pipeline depth.
- `up_after: Float64`: Seconds tasks stay ready with no free slot before workers start.
- `idle_before_down: Float64`: Seconds a worker stays idle before it drains.
- `max_step: Int`: Most workers started by one decision.

#### `__init__`

```mojo
def __init__(out self, min_workers: Int = Int(1), max_workers: Int = Int(4), tasks_per_worker: Int = Int(0), up_after: Float64 = Float64("5"), idle_before_down: Float64 = Float64("60"), max_step: Int = Int(4))
```

### `Decision`

```mojo
struct Decision
```

Implements: `Copyable`, `Movable`, `Writable`

Fields:

- `start: Int`: Workers to start.
- `drain: List[String]`: Workers to drain, by address.
- `want: Int`: The workers the demand calls for, within the policy's range.
- `short_since: Float64`: When the job ran short of workers, for the next `decide`; 0 if it isn't.

#### `__init__`

```mojo
def __init__(out self)
```

#### `write_to`

```mojo
def write_to(self, mut writer: T)
```

`dista scale --once`'s output: one line each.

## Functions

### `decide`

```mojo
def decide(demand: Demand, policy: Policy, now: Float64, pending: Int = Int(0), short_since: Float64 = Float64("0")) -> Decision
```

What to do now: workers to start and workers to drain. `pending` is workers started that haven't joined the head yet.