magmalake Dista

Dista · API

dista.client

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

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__

def __init__(out self, id: UInt64)

write_to

def write_to(self, mut writer: T)

ActorHandle

struct ActorHandle[T: ActorLike]

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

Implements Copyable, ImplicitlyCopyable, Movable, Writable

Fields
id: UInt64

__init__

def __init__(out self, id: UInt64)

write_to

def write_to(self, mut writer: T)

TaskSpan

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

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

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__

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

def status(mut self) -> String

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

status_json

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

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

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

Every driver's finished tasks the head still holds.

stop

def stop(mut self)

Shuts the head down, and its workers with it.

Client

struct Client

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

Implements Movable

__init__

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

def close(mut self)

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

take_driver_spans

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

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

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.

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

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

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

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.

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.

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.

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.

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.

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.

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.

bytes_fetched

def bytes_fetched(self) -> Int

How many result bytes this driver has fetched over sockets.

location

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

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__

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

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

CtxTask

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.

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

comptime A = A_
comptime R = R_

name

def name() -> StaticString

id

def id() -> UInt64

run

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

run_ctx

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

CtxMethod

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

comptime A = A_
comptime R = R_
comptime Act = Act_

name

def name() -> StaticString

id

def id() -> UInt64

owner

def owner() -> UInt64

call

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

call_ctx

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

Traits

CtxTaskLike

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

Implements TaskLike

run_ctx

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

CtxMethodLike

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

Implements MethodLike, TaskLike

call_ctx

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

Functions

parse_timeline

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

A timeline frame's spans, unnamed.

write_trace

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

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

Splits host:port.

connect_head

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

comptime MAX_DRIVER_SPANS = 1000000

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