# dista.runtime

> A module of Dista's API.

The runtime: one binary that runs as driver, head or worker.

```mojo
comptime Square = Task[Int, Int, square]

def main() raises:
    var rt = Runtime[Square](workers=4)
    var r = rt.remote[Square](7)
    print(rt.get(r))  # 49
```

`Runtime(...)` reads the environment to decide what this process is:

| Environment | Role |
|---|---|
| `DISTA_ROLE=head` | runs the head on `DISTA_PORT` (default 6390) 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 |

So a cluster runs one image everywhere, and the same program runs on a laptop
unchanged. The registry is the runtime's parameter list: workers dispatch
with a `comptime for` over it, and every process's hello carries a hash of it
so the head refuses a mismatched build.

A driver sends tasks to the head and reads results back. Small results come
with the reply; a large one stays on the worker that computed it. On the same
host `get` maps it from the node store and decodes it in place; from another
host it fetches it from that worker directly. A task can take
another task's result: pass an `ObjectRef[A]` instead of an `A`, or a
`List[ObjectRef[X]]` to a task that takes a `Seq[X]`. The head runs it once
every input exists, and fails it if an input failed.

`create` starts an actor on a worker of its own and returns its handle;
`call` runs one of its methods there, in the order of the calls, and returns
an `ObjectRef` like a task does; `checkpoint` saves its state to
`DISTA_CHECKPOINT_URI`; `kill` destroys it.

## Structs

### `Runtime`

```mojo
struct Runtime[*Ts: TaskLike]
```

A driver for tasks `Ts`, or a head or worker if the environment says so.

Implements: `Movable`

#### `__init__`

```mojo
def __init__(out self, workers: Int = Int(4), pipeline: Int = Int(2))
```

Starts this process in the role its environment names.

Args:

- `workers`: Workers to fork in local mode; ignored otherwise.
- `pipeline`: Tasks in flight per worker, if this process runs a head.

#### `__deinit__`

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

#### `remote`

```mojo
def remote[T: TaskLike](mut self, arg: T.A) -> ObjectRef[T.R]
```

Submits task `T` on `arg` and returns a handle to its result.

```mojo
def remote[T: Task2Like](mut self, a: T.A, b: T.B) -> ObjectRef[T.R]
```

Submits task `T`, a `Task2`, on `a` and `b`.

```mojo
def remote[T: Task3Like](mut self, a: T.A, b: T.B, c: T.C) -> ObjectRef[T.R]
```

Submits task `T`, a `Task3`, on `a`, `b` and `c`.

```mojo
def remote[T: TaskLike](mut self, arg: ObjectRef[T.A]) -> ObjectRef[T.R]
```

Submits task `T` to run on another task's result, once it exists.

```mojo
def remote[X: AnyType, //, T: TaskLike](mut self, args: List[ObjectRef[X]]) -> ObjectRef[T.R]
```

Submits task `T`, which takes a `Seq[X]`, to run on these results.

The task runs once every result exists, and receives them in order.

```mojo
def remote[X: AnyType, //, T: TaskLike](mut self, arg: T.A, args: List[ObjectRef[X]]) -> ObjectRef[T.R]
```

Submits task `T`, a `TaskWith`, on `arg` and these results: it runs once every result exists, and receives them in order as a `Seq`.

A shuffle's reducer takes this shape: its partition number, and the
map outputs holding that partition.

#### `remote_many`

```mojo
def remote_many[T: ManyLike](mut self, arg: T.A, n: Int) -> List[ObjectRef[T.Part]]
```

Submits task `T`, a `TaskMany`, on `arg`, for `n` results: each is an object of its own, stored, fetched and freed on its own.

A shuffle's map task takes this shape: one partition per reducer.

```mojo
def remote_many[T: ManyLike](mut self, arg: ObjectRef[T.A], n: Int) -> List[ObjectRef[T.Part]]
```

Submits task `T`, a `TaskMany`, to run on another task's result once it exists, for `n` results.

```mojo
def remote_many[X: AnyType, //, T: ManyLike](mut self, args: List[ObjectRef[X]], n: Int) -> List[ObjectRef[T.Part]]
```

Submits task `T`, a `TaskMany` taking a `Seq[X]`, to run on these results once they exist, for `n` results.

A shuffle's later stage takes this shape: it partitions again what
earlier tasks made, without the driver fetching it.

#### `create`

```mojo
def create[T: ActorLike](mut self, arg: T.A, checkpoint_every: Int = Int(0)) -> ActorHandle[T]
```

Starts actor `T` on a worker of its own, built from `arg`.

It returns at once; calls made before the actor is up wait for it.
If the constructor raises, every call fails with its error. The
handle is a value: a task given it can call the actor too.

With `checkpoint_every` set, the actor's state is written to
`DISTA_CHECKPOINT_URI` after every that many calls, and a restart
restores it from there instead of running every call again.

#### `checkpoint`

```mojo
def checkpoint[T: ActorLike](mut self, actor: ActorHandle[T]) -> ObjectRef[Int]
```

Checkpoints `actor` after the calls made to it so far. The result is the number of calls the checkpoint covers, once it's written.

#### `call`

```mojo
def call[M: MethodLike](mut self, actor: ActorHandle[M.Act], arg: M.A) -> ObjectRef[M.R]
```

Calls method `M` on `actor` with `arg`, after every earlier call.

#### `kill`

```mojo
def kill[T: ActorLike](mut self, actor: ActorHandle[T])
```

Destroys `actor` once its earlier calls have run, and frees its worker. Their results stay until released.

#### `ready`

```mojo
def ready[R: AnyType](mut self, r: ObjectRef[R]) -> Bool
```

Returns whether `r`'s task has finished, without waiting for it.

#### `get`

```mojo
def get[R: Movable](mut self, r: ObjectRef[R]) -> R
```

Waits for `r`'s task and returns its result, or raises its error.

#### `release`

```mojo
def release[R: AnyType](mut self, r: ObjectRef[R])
```

Says this driver is done with `r`: see `Client.release`.

```mojo
def release[R: AnyType](mut self, refs: List[ObjectRef[R]])
```

Releases every ref in `refs`.

#### `stats`

```mojo
def stats(mut self) -> HeadStats
```

Asks the head what it holds. Frames that arrive first are kept.

#### `location`

```mojo
def location[R: AnyType](mut self, r: ObjectRef[R]) -> String
```

Where `r`'s result lives: see `Client.location`.

#### `bytes_fetched`

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

How many result bytes this driver has fetched over sockets.

#### `clock`

```mojo
def clock() -> UInt64
```

Wall-clock nanoseconds, the clock the timeline uses: the start to give `span`.

#### `span`

```mojo
def span(mut self, name: String, start: UInt64, end: UInt64 = UInt64(0))
```

Records a stretch of this driver's own work for the timeline, from `start` (from `clock`) to `end`, or to now. `timeline` draws these as the driver's track, above the workers.

#### `spans`

```mojo
def spans(mut self) -> List[TaskSpan]
```

Every task this driver submitted that has finished, with its timeline, since the last call. A head restart loses the spans it held and those of tasks running then.

#### `timeline`

```mojo
def timeline(mut self, path: String) -> Int
```

Writes the finished tasks' timeline to `path` in Chrome's trace format, for Perfetto (ui.perfetto.dev) or chrome://tracing, and returns how many tasks it holds.

Each worker is a process with two tracks: its tasks running, and its
tasks waiting for their inputs. The driver's own spans, from `span`,
are a process above them. Times are microseconds from the first
submission or driver span; each run's arguments carry the head's
waits too. Driver spans are written once, like task spans.

## Aliases

### `DEFAULT_PORT`

```mojo
comptime DEFAULT_PORT = 6390
```