Receivers and Deferred Calls | UForm for Mojo

The receiver chooses where a verb runs and how it completes, and the verb stays the same. The model runs it on the calling thread, model.on(executor) blocks on an executor, and model.deferred(executor) submits it there and answers a Deferred:

model.text_encode(texts, Outputs.EMBEDDINGS, arena)                          # the calling thread
model.on(executor).text_encode(texts, Outputs.EMBEDDINGS, arena)             # blocking; `CONTENDED` while another call holds it
var call = model.deferred(executor).text_encode(texts, Outputs.EMBEDDINGS, arena)  # Deferred[...]
print(call.wait().embeddings().value().row[DType.float32](0)[0])

ModelOn and ModelDeferredOn offer the same verbs with the same arguments, and the blocking verbs are written once, in the ModelVerbs trait the model and ModelOn conform to. A trainer has only the executor receivers, trainer.on(executor) and trainer.deferred(executor), since its verbs run on the devices it was made over. Loads and Mesh.bootstrap address no handle yet, so they are constructors taking the executor they run on.

A Deferred owns its outcome and outputs until the call is final, and borrows the model, the executor, the arena and every input it was handed, which Mojo keeps alive for as long as the Deferred is. Mojo has no runtime of its own to await on, so wait is the whole interface: it blocks in uform_outcome_wait until the call is final, then answers the result or raises the failure, as many times as it is asked. status reads the outcome without blocking, and destroying an unfinished Deferred waits for it.

§ Progress

Every verb with a progress slot takes a Progress, made from a handler that is a compile-time parameter, since a capturing closure cannot cross into C; its state crosses beside it as a pointer that must outlive the call:

def stop_halfway(completed: Int, total: Int, seen: Pointer[Int, MutAnyOrigin]) -> Continuation:
    seen[] = completed
    return Continuation.CANCEL if 2 * completed >= total else Continuation.CONTINUE

var seen = 0
var config = EncodeConfig(progress=Progress.of[stop_halfway](Pointer(to=seen), steps=10))
model.text_encode(texts, Outputs.EMBEDDINGS, arena, config)  # raises Status.CANCELLED halfway

Handlers returning a value are confirmed on Linux x86-64 with Mojo 1.2; Linux AArch64 and macOS rest on the CI matrix.

API

StructModelOnA model bound to the executor its verbs block on, as model.on(executor) answers it. Binding-only.1

struct ModelOn[origin: ImmOrigin]

Parameters

`originImmOrigin`	comptime	The model and the executor, which this receiver borrows.

Fields

`model`	`Pointer[Model, origin]`	The model every verb addresses.
`executor`	`Address`	The `uform_executor_t` every verb runs on.

Conforms to

Copyable, Deinitable, ImplicitlyCopyable, ModelVerbs, Movable

__init__Binds a model to an executor.

def __init__(out self, model: Pointer[Model, origin], executor: Int)

StructModelDeferredOnA model bound to the executor its verbs are submitted to, as model.deferred(executor) answers it: every verb answers a Deferred at once. Binding-only.9

struct ModelDeferredOn[origin: ImmOrigin]

The verbs and their arguments are those of ModelVerbs; the arena stays borrowed until final.

Parameters

`originImmOrigin`	comptime	The model and the executor, which this receiver borrows.

Fields

`model`	`Pointer[Model, origin]`	The model every verb addresses.
`executor`	`Address`	The `uform_executor_t` every verb is submitted to.

Conforms to

Copyable, Deinitable, ImplicitlyCopyable, Movable

__init__Binds a model to an executor.

def __init__(out self, model: Pointer[Model, origin], executor: Int)

describeSubmits ModelVerbs.describe. Calls uform_model_describe.

def describe(self) -> Deferred[_ModelDescribeCall, origin]

text_encodeSubmits ModelVerbs.text_encode. Calls uform_text_encode.

def text_encode[texts_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    texts: TapeView[texts_origin],
    outputs: Outputs,
    mut arena: Arena,
    config: EncodeConfig[stats_origin] = EncodeConfig(outputs_scalar=ScalarKind.F32, progress=Progress(), stats=Optional())
) -> Deferred[_EncodeCall[origin_of(arena), texts_origin, stats_origin, StringSpan("uform_text_encode")], origin]

text_chooseSubmits ModelVerbs.text_choose. Calls uform_text_choose.

def text_choose[texts_origin: ImmOrigin, choices_origin: ImmOrigin, groups_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    texts: TapeView[texts_origin],
    choices: TapeView[choices_origin],
    mut arena: Arena,
    config: TextChooseConfig[groups_origin, stats_origin] = TextChooseConfig(temperature_groups=Span(), progress=Progress(), stats=Optional())
) -> Deferred[_TextChooseCall[origin_of(arena), texts_origin, choices_origin, groups_origin, stats_origin], origin]

image_encodeSubmits ModelVerbs.image_encode. Calls uform_image_encode.

def image_encode[images_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    images: TapeView[images_origin],
    outputs: Outputs,
    mut arena: Arena,
    config: EncodeConfig[stats_origin] = EncodeConfig(outputs_scalar=ScalarKind.F32, progress=Progress(), stats=Optional())
) -> Deferred[_EncodeCall[origin_of(arena), images_origin, stats_origin, StringSpan("uform_image_encode")], origin]

image_encode_pixelsSubmits ModelVerbs.image_encode_pixels. Calls uform_image_encode_pixels.

def image_encode_pixels[pixels_origin: ImmOrigin, heights_origin: ImmOrigin, widths_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    pixels: TapeView[pixels_origin],
    heights: Span[UInt32, heights_origin],
    widths: Span[UInt32, widths_origin],
    outputs: Outputs,
    mut arena: Arena,
    config: EncodeConfig[stats_origin] = EncodeConfig(outputs_scalar=ScalarKind.F32, progress=Progress(), stats=Optional())
) -> Deferred[_ImageEncodePixelsCall[origin_of(arena), pixels_origin, heights_origin, widths_origin, stats_origin], origin]

video_encodeSubmits ModelVerbs.video_encode. Calls uform_video_encode.

def video_encode[videos_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    videos: TapeView[videos_origin],
    outputs: Outputs,
    mut arena: Arena,
    config: EncodeConfig[stats_origin] = EncodeConfig(outputs_scalar=ScalarKind.F32, progress=Progress(), stats=Optional())
) -> Deferred[_EncodeCall[origin_of(arena), videos_origin, stats_origin, StringSpan("uform_video_encode")], origin]

audio_encodeSubmits ModelVerbs.audio_encode. Calls uform_audio_encode.

def audio_encode[recordings_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    recordings: TapeView[recordings_origin],
    outputs: Outputs,
    mut arena: Arena,
    config: EncodeConfig[stats_origin] = EncodeConfig(outputs_scalar=ScalarKind.F32, progress=Progress(), stats=Optional())
) -> Deferred[_EncodeCall[origin_of(arena), recordings_origin, stats_origin, StringSpan("uform_audio_encode")], origin]

audio_encode_samplesSubmits ModelVerbs.audio_encode_samples. Calls uform_audio_encode_samples.

def audio_encode_samples[samples_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    samples: TapeView[samples_origin],
    sample_rate: Int,
    outputs: Outputs,
    mut arena: Arena,
    config: EncodeConfig[stats_origin] = EncodeConfig(outputs_scalar=ScalarKind.F32, progress=Progress(), stats=Optional())
) -> Deferred[_AudioEncodeSamplesCall[origin_of(arena), samples_origin, stats_origin], origin]

StructTrainerDeferredOnA trainer bound to the executor its verbs are submitted to: every verb answers a Deferred at once. Binding-only.3

struct TrainerDeferredOn[origin: ImmOrigin]

Parameters

`originImmOrigin`	comptime	The trainer and the executor, which this receiver borrows.

Fields

`library`	`Library`	The loaded engine every verb crosses into.
`trainer`	`Address`	The `uform_trainer_t` every verb addresses.
`executor`	`Address`	The `uform_executor_t` every verb is submitted to.

Conforms to

Copyable, Deinitable, ImplicitlyCopyable, Movable

__init__Binds a trainer to an executor.

def __init__(out self, library: Library, trainer: Int, executor: Int)

stepSubmits TrainVerbs.step with its embeddings; every input stays borrowed until final. Calls uform_trainer_step.

def step[modalities_origin: ImmOrigin, contents_origin: ImmOrigin, bags_origin: ImmOrigin, groups_origin: ImmOrigin, sides_origin: ImmOrigin, choices_origin: ImmOrigin, targets_origin: ImmOrigin, relations_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    modalities: Span[Modality, modalities_origin],
    contents: TapeView[contents_origin],
    mut arena: Arena,
    config: TrainerStepConfig[bags_origin, groups_origin, sides_origin, choices_origin, targets_origin, relations_origin, stats_origin] = TrainerStepConfig(bags=Span(), temperature_groups=Span(), sides=Span(), choices=Optional(), choice_targets=Optional(), relations=Relations(), outputs_scalar=ScalarKind.F32, stats=Optional())
) -> Deferred[_TrainerStepCall[origin_of(arena)], origin_of(bags_origin, choices_origin, contents_origin, groups_origin, modalities_origin, origin, relations_origin, sides_origin, stats_origin, targets_origin)]

def step[modalities_origin: ImmOrigin, contents_origin: ImmOrigin, bags_origin: ImmOrigin, groups_origin: ImmOrigin, sides_origin: ImmOrigin, choices_origin: ImmOrigin, targets_origin: ImmOrigin, relations_origin: ImmOrigin, stats_origin: MutOrigin, //](
    self,
    modalities: Span[Modality, modalities_origin],
    contents: TapeView[contents_origin],
    config: TrainerStepConfig[bags_origin, groups_origin, sides_origin, choices_origin, targets_origin, relations_origin, stats_origin] = TrainerStepConfig(bags=Span(), temperature_groups=Span(), sides=Span(), choices=Optional(), choice_targets=Optional(), relations=Relations(), outputs_scalar=ScalarKind.F32, stats=Optional())
) -> Deferred[_TrainerStepCall[MutUntrackedOrigin], origin_of(bags_origin, choices_origin, contents_origin, groups_origin, modalities_origin, origin, relations_origin, sides_origin, stats_origin, targets_origin)]

Submits TrainVerbs.step; every input stays borrowed until final. Calls uform_trainer_step.

saveSubmits TrainVerbs.save; the path stays borrowed until final. Calls uform_trainer_save.

def save[path_origin: ImmOrigin, //](
    self,
    path: StringSpan[path_origin],
    progress: Progress = Progress()
) -> Deferred[_TrainerSaveCall, origin_of(origin, path_origin)]

StructCompletionWhether a verb returns with its final status or once submitted, uform_completion_t.

struct Completion

The receiver picks it: the model and model.on(executor) pass BLOCKING, and model.deferred(executor) passes DEFERRED, so no verb takes it as an argument.

Fields

`code`	`c_int`	The raw `uform_completion_t` value the ABI exchanges.

Values

`BLOCKING`	`Completion(Int32(0))`	Return with the final status stored and every output readable.
`DEFERRED`	`Completion(Int32(1))`	Return once submitted; the status reads `DEFERRED` until final.

Conforms to

Copyable, Deinitable, Equatable, ImplicitlyCopyable, Movable, RegisterPassable, TrivialRegisterPassable

StructMoodWhat an executor does with its threads or streams once it goes idle, uform_mood_t.

struct Mood

Fields

`code`	`c_int`	The raw `uform_mood_t` value the ABI exchanges.

Values

`PARKED`	`Mood(Int32(0))`	Sleep on futexes or events, freeing the core; the next call pays a wake-up.
`SPINNING`	`Mood(Int32(1))`	Poll them, trading a core for the lowest wake-up latency.

Conforms to

Copyable, Deinitable, Equatable, ImplicitlyCopyable, Movable, RegisterPassable, TrivialRegisterPassable

StructContinuationWhether the call a progress handler is watching carries on or unwinds, uform_continuation_t.

struct Continuation

Fields

`code`	`c_int`	The raw `uform_continuation_t` value the ABI exchanges.

Values

`CONTINUE`	`Continuation(Int32(0))`	Carry on to the next unit of work.
`CANCEL`	`Continuation(Int32(1))`	Stop, and finish the call with `CANCELLED`.

Conforms to

Copyable, Deinitable, Equatable, ImplicitlyCopyable, Movable, RegisterPassable, TrivialRegisterPassable

StructProgressHow a long call reports its advance, uform_progress_t: a handler, its state, and how many reports it hears. The handler is never invoked concurrently, nor after its call is final.2

struct Progress

Fields

`callback`	`Address`	The handler's C entry point; `NULL_ADDRESS` skips every report.
`state`	`Address`	Forwarded into every report; it must outlive the call.
`steps`	`c_size_t`	Reports a call makes at most, the last at completion; zero derives to ten.

Conforms to

Copyable, Defaultable, Deinitable, ImplicitlyCopyable, Movable

__init__No handler: the call runs as one slice and reports nothing.

def __init__(out self)

ofReports to handler with state, which returns Continuation.CANCEL to stop the call.

def of[T: AnyType, origin: MutOrigin, //, handler: def(Int, Int, Pointer[T, MutAnyOrigin]) thin -> Continuation](
    state: Pointer[T, origin],
    steps: Int = Int(0)
) -> Self

Parameters

`TAnyType`	comptime	What `state` points at, inferred from it.
`originMutOrigin`	comptime	The state's origin, inferred with it.
`handlerdef(Int, Int, Pointer[T, MutAnyOrigin]) thin -> Continuation`	comptime	Called with the units completed, the units in total, and `state`.
`statePointer[T, origin]`		Forwarded into every report; it must outlive the call.
`stepsInt`		Reports the call makes at most; zero derives to ten.

StructDeferredA call submitted to an executor, which owns its outcome and outputs until it is final. Binding-only.4

struct Deferred[C: Call, origin: ImmOrigin]

status reads DEFERRED until then, without blocking, and wait blocks until the call is final and answers its result or raises its failure, as many times as it is asked. The result is the view a blocking call answers, borrowing the same arena. Dropping an unfinished one waits the same way, so nothing the engine still writes is ever freed under it. The origins keep the model, the executor, the arena and every borrowed input alive for as long as this is.

Parameters

`CCall`	comptime	The verb's crossing, which names the result.
`originImmOrigin`	comptime	The union of everything the call borrows.

Fields

`library`	`Library`	The loaded engine, held so the wait can still reach `uform_outcome_wait`.
`slots`	`OwnedPointer[CallSlots[C.Outputs]]`	The outcome and outputs the engine writes, on the heap since they outlive the verb.
`call`	`Optional[C]`	What the call lends the engine, until a wait concludes it.
`answer`	`Optional[C.Result]`	The result, once a wait found the call succeeded.
`failure`	`Optional[Error]`	The failure, once a wait found the call failed.
`generation`	`Int`	The stamp `served` answered at submission, so a result waited on late still reads as stale.

Conforms to

Deinitable, Movable

__init__Submits call deferred, on executor, against handle.

def __init__(out self, library: Library, var call: C, handle: Int, executor: Int)

__deinit__Waits until the call is final, then frees what it lent. Calls uform_outcome_wait.

def __deinit__(deinit self)

statusDEFERRED until the call is final, then its final status; never blocks. Calls uform_outcome_status.

def status(self) -> Status

waitBlocks until the call is final, then answers its result or raises its failure, every time it is asked. Calls uform_outcome_wait.

def wait(mut self) -> ref[self] C.Result

Raises

Error carrying the call's failure, whether the engine or the binding refused it.