`dista.scale`

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`

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`

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`

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`

def parse(text: String) -> Self

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

`live`

def live(self) -> Int

Workers that aren't draining.

`Policy`

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`

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`

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`

def __init__(out self)

`write_to`

def write_to(self, mut writer: T)

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

Functions

`decide`

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.