# dista.client

> A module of Dista's API.

The driver's side of the protocol: submitting tasks, calling actors and getting results back from the head.

`Runtime` is the program's entry point: it starts the process in its role
and checks that what it's given is registered. `Client` is the part that
talks to the head, without the registry, so it can also run inside a
worker: a `CtxTask` gets one, and can call actors or submit tasks from
there. A worker's client connects with `ROLE_CLIENT`, which the head treats
as a driver, except that it doesn't keep a local-mode head alive.

A client checks only what it can know: that it didn't kill an actor it
calls. Whether the task, actor or method exists is the binary's: every
process runs the same registry, and the head refuses a peer that doesn't.

## Structs

### `ObjectRef`

```mojo
struct ObjectRef[T: AnyType]
```

A handle to a task's result, which may not exist yet.

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

Fields:

- `id: UInt64`

#### `__init__`

```mojo
def __init__(out self, id: UInt64)
```

#### `write_to`

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

### `ActorHandle`

```mojo
struct ActorHandle[T: ActorLike]
```

A handle to an actor of type `T`, from `Runtime.create`.

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

Fields:

- `id: UInt64`

#### `__init__`

```mojo
def __init__(out self, id: UInt64)
```

#### `write_to`

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

### `TaskSpan`

```mojo
struct TaskSpan
```

One finished task, from `Runtime.spans`: wall-clock nanoseconds.

The first four are the head's, the last three the worker's. On a
cluster the two clocks agree only as well as NTP keeps them.

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

Fields:

- `object_id: UInt64`
- `task: String`: The task's registered name.
- `worker: String`: The worker's `ip:port`, or "" if it never ran (an input failed).
- `submitted: UInt64`
- `queued: UInt64`: When its inputs existed and it joined the queue.
- `dispatched: UInt64`
- `settled: UInt64`
- `arrived: UInt64`: On the worker: when the task reached it, before any input fetches.
- `started: UInt64`
- `ended: UInt64`: When it finished, its result stored.
- `task_id: UInt64`: The task's wire ID.

### `HeadStats`

```mojo
struct HeadStats
```

What the head holds, from `Runtime.stats`.

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

Fields:

- `objects: Int`: Objects with a live reference: unreleased, or an input still pending.
- `results: Int`: Small results kept at the head.
- `errors: Int`
- `located: Int`: Results the head knows live on a worker.
- `pending: Int`: Tasks queued or waiting on inputs.

### `Observer`

```mojo
struct Observer
```

A connection to the head that only looks, and can stop it: what `dista status`, `timeline` and `stop` use. It owns no objects and needs no registry.

Implements: `Movable`

#### `__init__`

```mojo
def __init__(out self, addr: String)
```

Connects to the head at `addr`, `host:port`, once: an observer reports a head that isn't there rather than wait for one.

#### `status`

```mojo
def status(mut self) -> String
```

The head's workers, drivers and actors, as text.

#### `status_json`

```mojo
def status_json(mut self) -> String
```

The head's demand and its workers, as one JSON object: tasks ready, waiting for inputs and running, actors waiting for a worker, and each worker's address, node, tasks in flight, seconds idle, bytes held, actor and whether it's draining.

#### `drain`

```mojo
def drain(mut self, worker: String)
```

Drains the worker whose object server is at `worker`, `ip:port`, as SIGTERM would: it takes no new work, what it holds moves, and it exits once it's done. Raises if the head won't.

#### `spans`

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

Every driver's finished tasks the head still holds.

#### `stop`

```mojo
def stop(mut self)
```

Shuts the head down, and its workers with it.

### `Client`

```mojo
struct Client
```

A connection to the head, as a driver: see the module docstring.

Implements: `Movable`

#### `__init__`

```mojo
def __init__(out self, var head: Stream, head_addr: String, registry_hash: UInt64, role: UInt8 = UInt8(1))
```

Says hello to the head on `head` and becomes a driver.

Args:

- `head`: A connection to the head.
- `head_addr`: Where to reconnect if the head restarts, or "".
- `registry_hash`: The registry every peer must share.
- `role`: `ROLE_DRIVER`, or `ROLE_CLIENT` for one inside a worker.

#### `close`

```mojo
def close(mut self)
```

Hangs up on the head: everything this client held is released.

#### `take_driver_spans`

```mojo
def take_driver_spans(mut self) -> List[_DriverSpan]
```

The driver's own spans recorded with `span`, handed over once.

#### `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`.

#### `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, the `n` values of the `Seq` it returns.

```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 all exist, for `n` results: a shuffle's second stage, partitioning what the first made.

#### `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. With
`checkpoint_every` set, its state is checkpointed after every that
many calls.

#### `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.

#### `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.

#### `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`.

The result is freed, here, at the head and on its worker, once no
submitted task still needs it; a task already given `r` as input
still gets it. Releasing before the task has finished is fine: the
result is dropped when it arrives. `get(r)` raises from now on, and a
task submitted on `r` afterwards fails.

```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.

#### `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.

#### `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.

#### `bytes_fetched`

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

How many result bytes this driver has fetched over sockets.

#### `location`

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

Where `r`'s result lives: a worker's `ip:port`, then its node-store file if it has one; "" if it came back with the reply. Waits for the task to finish.

### `TaskContext`

```mojo
struct TaskContext
```

What a worker hands a `CtxTask`: a client of its own to the head, opened the first time a task asks for it.

Implements: `Movable`

#### `__init__`

```mojo
def __init__(out self, connect_to: String, resume_at: String, registry_hash: UInt64)
```

`connect_to` is the head's `host:port`; `resume_at` the same in a cluster, or "" in local mode, where the head doesn't come back.

#### `client`

```mojo
def client(mut self) -> ref[self._client._value] Client
```

### `CtxTask`

```mojo
struct CtxTask[A_: Deinitable & Movable, R_: Deinitable & Movable, f: def(mut Client, A_) raises thin -> R_]
```

A remote function that also gets a `Client`: from inside the task it can call actors, whose handles it was given, and submit tasks.

```mojo
def bump(mut c: Client, h: ActorHandle[CounterActor]) raises -> Int:
    var r = c.call[Add](h, 1)
    var n = c.get(r)
    c.release(r)
    return n

comptime Bump = CtxTask[ActorHandle[CounterActor], Int, bump]
```

A call from a task blocks that task until the answer comes; an actor
calling back into the worker that's waiting on it would deadlock, and
tasks never run on an actor's worker, so a task can't.

Implements: `CtxTaskLike`, `Movable`, `TaskLike`

#### `A`

```mojo
comptime A = A_
```

#### `R`

```mojo
comptime R = R_
```

#### `name`

```mojo
def name() -> StaticString
```

#### `id`

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

#### `run`

```mojo
def run(args: Span[UInt8], mut out: List[UInt8])
```

#### `run_ctx`

```mojo
def run_ctx(args: Span[UInt8], mut out: List[UInt8], mut ctx: TaskContext)
```

### `CtxMethod`

```mojo
struct CtxMethod[Act_: ActorLike, A_: Deinitable & Movable, R_: Deinitable & Movable, f: def(mut Act_.S, mut Client, A_) raises thin -> R_]
```

A method of `Act_` that also gets a `Client`, so an actor can call another actor, or submit tasks.

The call blocks this actor until the answer comes. An actor that calls
one which, before answering, calls it back waits on itself: a cycle of
actor calls deadlocks, as it would with one thread per actor anywhere.

Implements: `CtxMethodLike`, `MethodLike`, `Movable`, `TaskLike`

#### `A`

```mojo
comptime A = A_
```

#### `R`

```mojo
comptime R = R_
```

#### `Act`

```mojo
comptime Act = Act_
```

#### `name`

```mojo
def name() -> StaticString
```

#### `id`

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

#### `owner`

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

#### `call`

```mojo
def call(state: Pointer[NoneType, MutUntrackedOrigin], args: Span[UInt8], mut out: List[UInt8])
```

#### `call_ctx`

```mojo
def call_ctx(state: Pointer[NoneType, MutUntrackedOrigin], args: Span[UInt8], mut out: List[UInt8], mut ctx: TaskContext)
```

## Traits

### `CtxTaskLike`

```mojo
undefined
```

A task that runs with a `Client`, to call actors or submit tasks.

Implements: `TaskLike`

#### `run_ctx`

```mojo
def run_ctx(args: Span[UInt8], mut out: List[UInt8], mut ctx: TaskContext)
```

### `CtxMethodLike`

```mojo
undefined
```

A method that runs with a `Client`, to call other actors.

Implements: `MethodLike`, `TaskLike`

#### `call_ctx`

```mojo
def call_ctx(state: Pointer[NoneType, MutUntrackedOrigin], args: Span[UInt8], mut out: List[UInt8], mut ctx: TaskContext)
```

## Functions

### `parse_timeline`

```mojo
def parse_timeline(payload: Span[UInt8]) -> List[TaskSpan]
```

A timeline frame's spans, unnamed.

### `write_trace`

```mojo
def write_trace(path: String, spans: List[TaskSpan], mine: List[_DriverSpan] = List()) -> Int
```

Writes finished tasks, and a driver's own spans, 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 spans 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. A task without a name
is named by its wire ID.

### `parse_host_port`

```mojo
def parse_host_port(addr: String) -> Tuple[String, Int]
```

Splits `host:port`.

### `connect_head`

```mojo
def connect_head(addr: String) -> Stream
```

Connects to the head at `host:port`, retrying while it isn't up yet.

In a cluster, processes start in any order: a worker's pod can come up
before the head's, or before its name resolves. This keeps trying for
`DISTA_CONNECT_TIMEOUT` seconds (default 60), then raises the last error.

The wait between tries starts at 10 ms and doubles to 250 ms: a head that
restarts is usually back within a fraction of a second, and a flat 250 ms
added that much to every recovery.

## Aliases

### `MAX_DRIVER_SPANS`

```mojo
comptime MAX_DRIVER_SPANS = 1000000
```

Driver spans kept until `timeline` writes them; later ones are dropped.