magmalake Dista

Dista · API

dista.runtime

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

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

struct Runtime[*Ts: TaskLike]

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

Implements Movable

__init__

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__

def __deinit__(deinit self)

remote

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

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

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

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

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.

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.

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.

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

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.

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.

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

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

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

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

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

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

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

get

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

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

release

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

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

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

Releases every ref in refs.

stats

def stats(mut self) -> HeadStats

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

location

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

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

bytes_fetched

def bytes_fetched(self) -> Int

How many result bytes this driver has fetched over sockets.

clock

def clock() -> UInt64

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

span

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

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

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

comptime DEFAULT_PORT = 6390