magmalake Orca

Orca · API

orca.tasks

The dista tasks a pipeline's steps call, and the runtime a run gets.

comptime Square = Task[Int, Int, square]
comptime TaxiTasks = Tasks[Square]
comptime Taxi = Pipeline["taxi", Download, Load, tasks=TaxiTasks]

def load(mut ctx: Ctx, raw: RawFile) raises -> Loaded:
    ref rt = ctx.runtime[TaxiTasks]()
    var r = rt.remote[Square](7)
    ...

A run of a pipeline with tasks starts one dista Runtime before its first step and keeps it until the run ends: in local mode it forks a head and ORCA_WORKERS workers (default 4); with DISTA_HEAD set it joins that cluster instead. The same binary is dista's worker: run with DISTA_ROLE set, orca_main hands the process to dista before reading its arguments.

The runtime starts before orca opens a database connection or its heartbeat thread, so the forked head and workers inherit neither.

A driver or worker given DISTA_HEAD first waits until the head accepts a connection (wait_for_head), trying for a second at a time. On Kubernetes the head's Service can be a moment old: a connection sent before the node routes its address waits on a connection-tracking entry that never routes, for minutes, and dista connects without a timeout. A fresh try each second gets a fresh entry.

Structs

Tasks

struct Tasks[*Ts: TaskLike]

A pipeline's dista tasks.

Implements Movable, TaskSet

comptime RT = Runtime[Ts]
comptime HAS_RUNTIME = True

start

def start(workers: Int) -> Self.RT

names

def names() -> List[String]

timeline

def timeline(mut rt: Runtime[Ts], path: String) -> Int

NoTasks

struct NoTasks

A pipeline that calls no dista tasks: no runtime is started.

Implements Movable, TaskSet

comptime RT = NoneType
comptime HAS_RUNTIME = False

start

def start(workers: Int) -> Self.RT

names

def names() -> List[String]

timeline

def timeline(mut rt: NoneType, path: String) -> Int

Traits

TaskSet

The tasks a pipeline registers with dista.

comptime RT

The runtime type: Runtime of the tasks, or NoneType.

comptime HAS_RUNTIME

start

def start(workers: Int) -> Self.RT

Starts the runtime: forks a local cluster, or joins DISTA_HEAD.

names

def names() -> List[String]

The tasks' linkage names, for the pipeline's registry hash.

timeline

def timeline(mut rt: Self.RT, path: String) -> Int

Writes the run's dista timeline (Chrome's trace format) to path and returns the tasks in it.

Functions

wait_for_head

def wait_for_head(addr: String, timeout_s: Float64)

Returns once the dista head at host:port accepts a connection; raises after timeout_s seconds.

task_list

def task_list[T: TaskSet]() -> String

T's task names, one per line.

workers

def workers() -> Int

ORCA_WORKERS, the local workers a run forks (default 4).