# dista.worker

> A module of Dista's API.

A worker: runs tasks for the head and serves its results to peers.

A small result (up to `INLINE_LIMIT` bytes) goes back to the head, which
forwards it. A larger one stays here: in a shared mapping in the node's store
(see `store.mojo`), or in memory if the store is off. The worker tells the
head where it is; a reader on this host maps the file, and one elsewhere
fetches it from this worker's object server.

So a worker is an event loop over the head, its listener, the drivers and
workers fetching from it, and the workers it fetches from. A task whose
inputs live on other workers waits, parked, until they've all arrived; the
worker keeps serving fetches meanwhile. No worker ever waits on another one
synchronously, so two workers that need each other's results can't deadlock.
Running a task does block the loop, so fetches from this worker wait until
the current task finishes.

A result here lives until the head says free, once nothing references it.
Inputs fetched from peers are copies, kept only until the tasks waiting on
them have their arguments.

A worker the head gives an actor builds its state and then runs only that
actor's calls, in the order they come. A kill destroys the state, and the
worker goes back to running tasks. Between calls the head may ask for a
checkpoint: the worker writes the state through the codec to
`DISTA_CHECKPOINT_URI` (see `checkpoint.mojo`), and deletes the checkpoint
before it. An actor restarted elsewhere is restored from the last one.

A worker told to stop (SIGTERM, as Kubernetes sends) drains: it tells the
head, which sends it no more work and has other workers adopt the results it
holds, each fetching its copy from here. It finishes what it was given, and
once the head says stop, it serves fetches a moment longer, for tasks given
its old locations, and exits. `DISTA_DRAIN_SECONDS` (default 25, inside
Kubernetes' default grace period of 30) bounds the whole drain.

If the head goes away, `run` returns and the results here stay. `rejoin`
introduces the worker to a new head and reports what it holds; the tasks it
had queued are dropped, since their drivers resubmit them. An actor it
hosted is destroyed: actors don't survive a head restart yet, though their
checkpoints stay.

## Structs

### `InputSpec`

```mojo
struct InputSpec
```

One input in a task-inputs frame: its bytes, or where to get them.

Implements: `Copyable`, `Movable`

Fields:

- `object_id: UInt64`
- `inline: Bool`
- `bytes: List[UInt8]`
- `location: Optional[Location]`

### `TaskInputs`

```mojo
struct TaskInputs
```

Implements: `Movable`

Fields:

- `many: Int`: A `TaskMany`'s number of results; 0 for any other task.
- `gather: Bool`
- `inputs: List[InputSpec]`
- `value: List[UInt8]`: A `TaskWith`'s value, which goes before the inputs; or empty.

### `Worker`

```mojo
struct Worker[dispatch: def(UInt64, List[UInt8], mut List[UInt8], mut TaskContext) raises thin -> None, create: def(UInt64, List[UInt8]) raises thin -> StatePtr, call: def(UInt64, UInt64, Pointer[NoneType, MutUntrackedOrigin], List[UInt8], mut List[UInt8], mut TaskContext) raises thin -> None, destroy: def(UInt64, Pointer[NoneType, MutUntrackedOrigin]) thin -> None, save: def(UInt64, Pointer[NoneType, MutUntrackedOrigin], mut List[UInt8]) raises thin -> None, restore: def(UInt64, List[UInt8]) raises thin -> StatePtr]
```

Implements: `Movable`

Fields:

- `stopped: Bool`: The head said stop, or the worker drained: `run` returned for good.

#### `__init__`

```mojo
def __init__(out self, var head: Stream, registry_hash: UInt64, head_addr: String = "", resume_at: String = "")
```

Starts the object server and says hello to the head on `head`.

`head_addr` is where a task's client connects, and `resume_at` where
it reconnects after a head restart ("" in local mode).

#### `__deinit__`

```mojo
def __deinit__(deinit self)
```

#### `rejoin`

```mojo
def rejoin(mut self, var head: Stream, registry_hash: UInt64)
```

Says hello to a new head and reports the results held here.

#### `run`

```mojo
def run(mut self)
```

Serves until the head hangs up, or until a drain ends.

## Functions

### `parse_task_inputs`

```mojo
def parse_task_inputs(payload: Span[UInt8]) -> TaskInputs
```

A task-inputs payload, checked: a malformed one raises.

## Aliases

### `INLINE_LIMIT`

```mojo
comptime INLINE_LIMIT = 65536
```

Results up to this size travel through the head; larger ones stay put.

### `LINGER_NS`

```mojo
comptime LINGER_NS = UInt64(1000000000)
```

How long a drained worker keeps serving fetches after the head says stop: a task sent before its inputs moved may still fetch them from here.

### `DispatchFn`

```mojo
comptime DispatchFn = def(UInt64, List[UInt8], mut List[UInt8], mut TaskContext) raises thin -> None
```

Runs task `task_id` on encoded arguments, appending the encoded result; a `CtxTask` gets the worker's `TaskContext`.

### `CreateFn`

```mojo
comptime CreateFn = def(UInt64, List[UInt8]) raises thin -> StatePtr
```

Builds actor type `type_id`'s state from its encoded argument.

### `CallFn`

```mojo
comptime CallFn = def(UInt64, UInt64, Pointer[NoneType, MutUntrackedOrigin], List[UInt8], mut List[UInt8], mut TaskContext) raises thin -> None
```

Runs method `method_id` on the state of an actor of type `type_id`.

### `DestroyFn`

```mojo
comptime DestroyFn = def(UInt64, Pointer[NoneType, MutUntrackedOrigin]) thin -> None
```

Destroys the state of an actor of type `type_id`.

### `SaveFn`

```mojo
comptime SaveFn = def(UInt64, Pointer[NoneType, MutUntrackedOrigin], mut List[UInt8]) raises thin -> None
```

Appends the encoded state of an actor of type `type_id`.

### `RestoreFn`

```mojo
comptime RestoreFn = def(UInt64, List[UInt8]) raises thin -> StatePtr
```

Builds actor type `type_id`'s state from a checkpoint.