Parallel tasks and deadlines

Named parallel results, await deadlines and lazy delegation.

Named parallel results¶

Named tasks in parallel produce a typed result record after the structured join. Every task captures the outer scope; task labels are not variables for siblings. Failure cancels and drains siblings cooperatively. Transitively linear result payloads are rejected.

Practice in By Example · Source program

// Feature: typed named results from structured parallel tasks
// Each label names a result field; each RHS captures only the outer scope.
struct User { name: str }
async fn load_user() -> User = User { name: "Ada" }
async fn load_orders() -> [int] = [101, 102]

let user = "outer"
let data = parallel {
    user = await load_user()
    orders = await load_orders()
    captured = user
}
print(data.user.name)
print(data.orders.len())
print(data.captured)
print(user)
// expected:
//   Ada
//   2
//   outer
//   outer

Run this example locally with the installed CLI. The browser playground does not provide the host scheduler.

Await deadlines¶

await task within duration returns a typed timeout Result. On expiry the scheduler requests cancellation and drains the task's cooperative cleanup. A task returning Result produces a nested Result; choose explicitly which layer to propagate. Native work without cancellation checkpoints cannot be forcibly interrupted.

Practice in By Example · Source program

async fn load() -> int {
    sleep 20ms
    42
}

fn main() {
    let result = await load() within 1s
    print(result.unwrap())
}

Run this example locally with the installed CLI. The browser playground does not provide the host scheduler.

Lazy generator delegation¶

yield ... child delegates one yielded value at a time and retains the child cursor between resumes. It does not materialize an array. Closing the outer generator closes the active child and runs its cleanup; ordinary yield is unchanged.

Practice in By Example · Source program

fn* middle()->int{yield 2;yield 3}
fn* values()->int{yield 1;yield ... middle();yield 4}
for value in values(){print(value)}

Reading an asynchronous source¶

for await asks the source for one value at a time and awaits next() before running the loop body. Nil ends the iteration. A known next() method must return Future<T?> or Task<T?>, and the source must implement core::cleanup::Cleanup. This example consumes a complete source and stops a second source early; both are closed.

use core::cleanup::Cleanup
struct Events { remaining: int }
impl Events {
    async fn next(self) -> int? {
        if self.remaining <= 0 { return nil }
        let value = self.remaining
        self.remaining -= 1
        return value
    }
}
impl Cleanup for Events {
    fn close(self) { print("closed") }
}
let complete = Events { remaining: 3 }
var sum = 0
for await value in complete { sum += value }
assert_eq(sum, 6)
let early = Events { remaining: 5 }
:events for await value in early {
    if value == 4 { break :events }
    print(value)
}
print(sum)

The source is acquired once and closed on completion, break, return, failure or cooperative cancellation. Pattern bindings, including for await case, apply to the awaited value. Synchronous arrays do not become asynchronous sources. Run this example with the CLI; wasm-aot rejects scheduler operations explicitly.

By Example

Limiting concurrent work¶

parallel(limit: 2) [for id in ids => await load_profile(id)] processes an array with at most two active jobs and returns results in source order. The example counts active jobs so the bound is observable, then verifies an empty input and a source snapshot.

var active = 0
var peak = 0
async fn load_profile(id: int) -> int {
    active += 1
    if active > peak { peak = active }
    defer { active -= 1 }
    sleep 2ms
    return id * 10
}
let profiles = parallel(limit: 2) [for id in [1, 2, 3, 4] => await load_profile(id)]
assert_eq(peak, 2)
assert_eq(active, 0)
assert_eq(profiles.len(), 4)
assert_eq(profiles[0], 10)
assert_eq(profiles[3], 40)
print(peak)
print(profiles[0])
print(profiles[3])
let no_ids: [int] = []
let empty = parallel(limit: 4) [for id in no_ids => id]
assert_eq(empty.len(), 0)

// The pool snapshots source items before evaluating any mapper.
let mut ids = [1, 2, 3]
let original = parallel(limit: 1) [for id in ids => { ids[1] = 99; id }]
assert_eq(original[1], 2)
assert_eq(ids[1], 99)

The positive integer limit and source are evaluated once, in that order. A fixed worker pool claims the next job after completing the current one; it does not spawn a task per input. The source snapshot protects pending items from later mutations. Failure cancels siblings and drains cleanup before propagating the error. This is cooperative concurrency; CPU-only jobs do not gain OS threads. Available on VM, native and LLVM; wasm-aot reports the scheduler limitation.

By Example

Continue learning¶

Documentation · Practice in By Example