# dista.frame > A module of Dista's API. Length-prefixed messages between drivers, the head and workers. Every frame is a 4-byte little-endian length, then that many bytes: | offset | field | |---|---| | 0 | kind (below) | | 1 | object ID, u64 | | 9 | task ID, u64 (0 unless the kind uses it) | | 17 | payload | | kind | sent by | payload | |---|---|---| | 1 task | driver to head, head to worker | the encoded argument | | 2 result | worker to head, head to driver | the encoded result | | 3 error | worker to head, head to driver | a UTF-8 message | | 4 hello | driver, worker or observer to head | role byte (`ROLE_*`), the registry hash u64 (an observer's is ignored); a worker adds its object server's address | | 5 welcome | head to driver | the driver's number, u64, then the head's epoch, u64 | | 6 task-after | driver to head | flags byte (`GATHER`, `WITH_VALUE`), count u64, then that many object IDs; with `WITH_VALUE`, a length u64 and the encoded value | | 7 stored | worker to head | the result's size, u64: it stayed on the worker | | 8 located | head to driver | the address of the worker holding the result | | 9 fetch | driver or worker to a worker | none, or the largest part the fetcher takes, u64: send that object back as a result frame, or if it's larger than that part, as part frames | | 10 task-inputs | head to worker | flags byte, count u64, then per input a tag byte, its object ID, then for tag 0 a length u64 and the bytes, or for tag 1 its location; with `WITH_VALUE`, a length u64 and the value | | 11 release | driver to head | none: the driver is done with this object | | 12 free | head to worker | none: delete this object | | 13 stats | driver to head, head to driver | none; the reply is five u64 counts (see `HeadStats`) | | 14 holding | worker to head, on rejoining | count u64, then per object its ID u64 and its location | | 15 claim | driver to head, on resuming | none: this driver still holds this object, which lives on a worker | | 16 resumed | driver to head | none: this driver has restored, claimed or resubmitted everything | | 17 span | worker to head, before the result | three u64 wall-clock nanoseconds: when the task arrived, started and ended | | 18 timeline | driver to head, head to driver | none; the reply is a count u64, then per finished task its object ID, task ID, the four head stamps (submitted, queued, dispatched, settled), the three worker stamps, and the worker's address | | 19 actor-create | driver to head, head to worker | from the driver, `checkpoint_every` u64 first; then the encoded constructor argument. The object ID names the actor, the task ID its type | | 20 actor-call | driver to head, head to worker | the actor's ID u64, then the encoded argument; the task ID is the method's | | 21 actor-kill | driver to head, head to worker | none: destroy the actor the object ID names | | 22 checkpoint | driver to head, head to worker | from the driver, the actor's ID u64, and the object ID is the checkpoint's own; from the head, the calls covered by the checkpoint it would restore from, i64 (-1 for none), the object ID names the actor and the task ID is the number of calls the new one covers | | 23 checkpointed | worker to head | none if the checkpoint is written, else why not; the object and task IDs are the checkpoint frame's | | 24 actor-restore | head to worker | the number of calls the checkpoint covers, u64; the object ID names the actor, the task ID its type | | 25 covered | head to driver | the newest of this driver's calls to the actor a checkpoint now includes, u64; the object ID names the actor | | 26 status | observer to head, head to observer | none; the reply is the head's state as UTF-8 text | | 27 stop | observer to head, head to worker | none: shut down; a worker exits instead of rejoining | | 28 task-many | driver to head, head to worker | the number of results u64, then the encoded argument; the results are this object ID and the ones after it | | 29 drain | worker to head | none: the worker was told to stop; it takes no more work, and leaves once the head says stop | | 30 adopt | head to worker | a draining worker's location of the object: fetch it and keep it | | 31 adopted | worker to head | 1 and the object's new location, or 0 and why not | | 32 cancel | driver to head | none: fail this object's task with "cancelled", and drop it if it hasn't run | | 33 part | worker to a driver or worker that fetched | the object's size u64, this part's offset u64, then its bytes; parts come in order, and the object is whole when they reach its size | A frame holds at most `MAX_FRAME` bytes, so a result larger than that crosses hosts in parts. A fetcher says it takes parts by putting the largest it wants in the fetch frame. One that sends an empty fetch, as before parts existed, gets the whole object in one result frame, or an error frame if it doesn't fit. Writing a frame that doesn't fit raises; it used to wrap the length past 4 GiB and garble the stream. An observer (`dista status`, `timeline`, `stop`) says hello without a registry and owns no objects. Its timeline frame gets every driver's finished tasks, and leaves them for the drivers too. After a head restart, a driver's hello carries its old number as a u64 after the registry hash, so its object IDs stay valid, and then the epoch of the head that first numbered it, u64. A head numbers new drivers from a random start, so a driver of the head before rarely shares a number with one of this head's; if it does, the epoch tells them apart, and the head refuses the old one with an error frame rather than give it the new one's objects. A driver resuming under a number that is still connected, from the same epoch, is the same driver on a new connection: the old one is dropped. It then sends, in order of submission, each object it still needs: a small result or error back as a result or error frame, a result on a worker as claim, and an unfinished task as the task or task-after frame it first sent; then resumed. A rejoining worker sends holding. An actor lives on one worker, which runs nothing else while it hosts it. The worker answers actor-create with an empty result, or the constructor's error, and each call like a task: a result, stored or error frame for the call's object ID. Calls run in the order the driver sent them. A checkpoint runs between calls too: the worker writes the actor's state to `DISTA_CHECKPOINT_URI` and answers checkpointed. Actor-restore builds an actor from its checkpoint instead of its constructor, and is answered like actor-create. Actors survive a head restart through their checkpoints. Every caller keeps its calls to an actor until covered says a checkpoint includes them. A resuming driver sends each live actor's actor-create again, in order with everything else, then every call it still keeps: those answered too, after their answers, since the actor's state hasn't seen them. The new head restores the actor from its newest checkpoint; a call that checkpoint already includes isn't run again. A rejoining worker destroys the actor it hosted: its state may hold calls no caller would resend. A task-after frame queues a task whose argument is built from other results: the single result as is, or with gather set, a `Seq` of all of them; with `WITH_VALUE` set, after an encoded value (a `TaskWith`). With `MANY` set it's a `TaskMany`, whose results, the object ID and the ones after it, number the u64 that follows the flags byte; the task-inputs frame carries it on. A task-many frame runs a `TaskMany`: its worker answers each of its results with its own result, stored or error frame, as if each were a task. When some of those results live on workers, the head sends the task as task-inputs, and the worker fetches them before it runs the task. An address is 12 bytes: the IPv4 address, then the port as a u64. The head counts references to every object: its driver's, until a release, and one per submitted task that takes it as input, until that task settles. At zero it forgets the object, and sends free to the worker holding it. ## Structs ### `Frame` ```mojo struct Frame ``` Implements: `Movable` Fields: - `kind: UInt8` - `object_id: UInt64` - `task_id: UInt64` - `payload: List[UInt8]` #### `__init__` ```mojo def __init__(out self, kind: UInt8, object_id: UInt64, task_id: UInt64, var payload: List[UInt8]) ``` #### `take_payload` ```mojo def take_payload(deinit self) -> List[UInt8] ``` Consumes the frame and returns its payload without copying it. ### `Parts` ```mojo struct Parts ``` An object arriving in part frames: `add` each, in order. Implements: `Movable` Fields: - `data: List[UInt8]` - `size: Int` #### `__init__` ```mojo def __init__(out self) ``` #### `add` ```mojo def add(mut self, payload: Span[UInt8]) -> Bool ``` Adds one part frame's payload; whether the object is now whole. #### `take` ```mojo def take(deinit self) -> List[UInt8] ``` ## Functions ### `message` ```mojo def message(payload: Span[UInt8]) -> String ``` An error frame's text. It comes from another process, so bytes that aren't UTF-8 become replacement characters instead of a broken String. ### `check_payload` ```mojo def check_payload(n: Int) ``` Raises unless a frame can carry a payload of `n` bytes. ### `append_frame` ```mojo def append_frame(mut out: List[UInt8], kind: UInt8, object_id: UInt64, task_id: UInt64, payload: Span[UInt8]) ``` Appends one whole frame to `out`. Raises, appending nothing, if the payload is too large for a frame. ### `append_parts` ```mojo def append_parts(mut out: List[UInt8], object_id: UInt64, data: Span[UInt8], part: Int) ``` Appends `data`, object `object_id`, as part frames of at most `part` bytes each. ### `append_header` ```mojo def append_header(mut out: List[UInt8], kind: UInt8, object_id: UInt64, task_id: UInt64, size: Int) ``` Appends a frame's length and header, for a payload of `size` bytes that the caller appends after it: a payload sent from where it lies. ### `append_part_header` ```mojo def append_part_header(mut out: List[UInt8], object_id: UInt64, total: Int, offset: Int, n: Int) ``` Appends a part frame's length, header and part fields, for the `n` bytes of an object of `total` at `offset`, which the caller appends. ### `part_payload_size` ```mojo def part_payload_size(part: Int) -> Int ``` The bytes of an object one part frame carries, for a fetch asking for parts of `part`: what `append_parts` cuts it into. ### `fetch_payload` ```mojo def fetch_payload(part: Int = Int(67108864)) -> List[UInt8] ``` A fetch frame's payload: the largest part this fetcher takes. ### `fetch_part` ```mojo def fetch_part(payload: Span[UInt8]) -> Int ``` The largest part a fetch frame asks for; 0 for an empty fetch, which wants the whole object in one frame. ### `write_frame` ```mojo def write_frame(s: Stream, kind: UInt8, object_id: UInt64, task_id: UInt64, payload: Span[UInt8]) ``` ### `frame_length` ```mojo def frame_length(head: Span[UInt8]) -> Int ``` The body length a 4-byte frame prefix announces, checked. ### `parse_body` ```mojo def parse_body(body: Span[UInt8]) -> Frame ``` A frame from its body: everything after the length prefix. ### `read_frame` ```mojo def read_frame(s: Stream) -> Optional[Frame] ``` Reads the next frame, or returns `None` if the peer closed cleanly. ### `encode_addr` ```mojo def encode_addr(ip: Array[UInt8, Int(4)], port: Int) -> List[UInt8] ``` An address as frames carry it: the IPv4 address, then the port, u64. ### `decode_addr` ```mojo def decode_addr(buf: Span[UInt8], mut pos: Int) -> Tuple[Array[UInt8, Int(4)], Int] ``` Reads an address written by `encode_addr` and advances `pos`. ## Aliases ### `KIND_TASK` ```mojo comptime KIND_TASK = UInt8(1) ``` ### `KIND_RESULT` ```mojo comptime KIND_RESULT = UInt8(2) ``` ### `KIND_ERROR` ```mojo comptime KIND_ERROR = UInt8(3) ``` ### `KIND_HELLO` ```mojo comptime KIND_HELLO = UInt8(4) ``` ### `KIND_WELCOME` ```mojo comptime KIND_WELCOME = UInt8(5) ``` ### `KIND_TASK_AFTER` ```mojo comptime KIND_TASK_AFTER = UInt8(6) ``` ### `KIND_STORED` ```mojo comptime KIND_STORED = UInt8(7) ``` ### `KIND_LOCATED` ```mojo comptime KIND_LOCATED = UInt8(8) ``` ### `KIND_FETCH` ```mojo comptime KIND_FETCH = UInt8(9) ``` ### `KIND_TASK_INPUTS` ```mojo comptime KIND_TASK_INPUTS = UInt8(10) ``` ### `KIND_RELEASE` ```mojo comptime KIND_RELEASE = UInt8(11) ``` ### `KIND_FREE` ```mojo comptime KIND_FREE = UInt8(12) ``` ### `KIND_STATS` ```mojo comptime KIND_STATS = UInt8(13) ``` ### `KIND_HOLDING` ```mojo comptime KIND_HOLDING = UInt8(14) ``` ### `KIND_CLAIM` ```mojo comptime KIND_CLAIM = UInt8(15) ``` ### `KIND_RESUMED` ```mojo comptime KIND_RESUMED = UInt8(16) ``` ### `KIND_SPAN` ```mojo comptime KIND_SPAN = UInt8(17) ``` ### `KIND_TIMELINE` ```mojo comptime KIND_TIMELINE = UInt8(18) ``` ### `KIND_ACTOR_CREATE` ```mojo comptime KIND_ACTOR_CREATE = UInt8(19) ``` ### `KIND_ACTOR_CALL` ```mojo comptime KIND_ACTOR_CALL = UInt8(20) ``` ### `KIND_ACTOR_KILL` ```mojo comptime KIND_ACTOR_KILL = UInt8(21) ``` ### `KIND_CHECKPOINT` ```mojo comptime KIND_CHECKPOINT = UInt8(22) ``` ### `KIND_CHECKPOINTED` ```mojo comptime KIND_CHECKPOINTED = UInt8(23) ``` ### `KIND_ACTOR_RESTORE` ```mojo comptime KIND_ACTOR_RESTORE = UInt8(24) ``` ### `KIND_COVERED` ```mojo comptime KIND_COVERED = UInt8(25) ``` ### `KIND_STATUS` ```mojo comptime KIND_STATUS = UInt8(26) ``` ### `KIND_STOP` ```mojo comptime KIND_STOP = UInt8(27) ``` ### `KIND_TASK_MANY` ```mojo comptime KIND_TASK_MANY = UInt8(28) ``` ### `KIND_DRAIN` ```mojo comptime KIND_DRAIN = UInt8(29) ``` Worker to head: it was told to stop (SIGTERM), so send it no more work and move what it holds; the head says `KIND_STOP` once it may go. ### `KIND_ADOPT` ```mojo comptime KIND_ADOPT = UInt8(30) ``` Head to worker: fetch object `object_id` from the location in the payload, a draining worker's, and keep it as its own. ### `KIND_ADOPTED` ```mojo comptime KIND_ADOPTED = UInt8(31) ``` Worker to head: 1 and the object's new location, or 0 and why it couldn't take it. ### `KIND_CANCEL` ```mojo comptime KIND_CANCEL = UInt8(32) ``` Driver to head: this driver won't read `object_id`. A task still queued or waiting for inputs never runs; one running finishes, and its result is dropped. Either way the object fails with "cancelled". ### `KIND_PART` ```mojo comptime KIND_PART = UInt8(33) ``` Worker to a fetcher: one part of an object too large for one frame. ### `GATHER` ```mojo comptime GATHER = UInt8(1) ``` Task-after and task-inputs flag: the inputs gather into a `Seq`. ### `WITH_VALUE` ```mojo comptime WITH_VALUE = UInt8(2) ``` Task-after and task-inputs flag: an encoded value goes before them. ### `MANY` ```mojo comptime MANY = UInt8(4) ``` Task-after and task-inputs flag: the task is a `TaskMany`, and the number of its results, a u64, follows the flags byte. ### `ADDR_SIZE` ```mojo comptime ADDR_SIZE = 12 ``` ### `ROLE_DRIVER` ```mojo comptime ROLE_DRIVER = UInt8(1) ``` ### `ROLE_WORKER` ```mojo comptime ROLE_WORKER = UInt8(2) ``` ### `ROLE_CLIENT` ```mojo comptime ROLE_CLIENT = UInt8(3) ``` ### `ROLE_OBSERVER` ```mojo comptime ROLE_OBSERVER = UInt8(4) ``` A driver inside a worker, for a task that calls actors: the head treats it as a driver, except that it doesn't keep a local-mode head alive. ### `HEADER` ```mojo comptime HEADER = 17 ``` ### `MAX_FRAME` ```mojo comptime MAX_FRAME = 1073741824 ``` The largest frame body a receiver takes: the kind, the IDs and the payload. ### `MAX_PAYLOAD` ```mojo comptime MAX_PAYLOAD = 1073741807 ``` ### `PART_SIZE` ```mojo comptime PART_SIZE = 67108864 ``` The largest part a fetcher asks for: an object larger than this comes in parts of it. ### `MAX_OBJECT` ```mojo comptime MAX_OBJECT = 1099511627776 ``` A part frame announcing an object larger than this is garbage.