Progress & Cancellation

Progress reporting with events

A thread reports while it works the way a component does: it declares an event and its functions fire it.

thread downloader {
    event on_progress(bytes: i64, total: i64);

    fn download(url: string) -> []u8 or NetError {
        let total = get_content_length(url) or return;
        var received: i64 = 0;

        while (received < total) {
            let chunk = fetch_chunk(url, received) or return;
            received = received + chunk.len;
            fire on_progress(received, total);
        }

        return buffer;
    }
}

An event takes any parameters, none included. fire never waits and never fails: the arguments are copied and handed to the caller’s thread, and fires from one task arrive in order, before the result.

Updating the UI

At the UI boundary, bind a handler to the event by its name, next to on_done and on_error. The handler runs on the UI thread, so it can update a component directly:

component Downloader {
    var pct: i64 = 0;

    fn start(url: string) {
        downloader.download(url) {
            on_done: on_complete,
            on_error: on_failed,
            on_progress: show_progress,
        };
    }

    fn show_progress(bytes: i64, total: i64) {
        pct = bytes * 100 / total;       // the progress bar repaints
    }

    fn on_complete(data: []u8) { save_file(data); }
    fn on_failed(e: NetError) { show_error(e.desc()); }
}

The handler’s signature matches the event’s parameters (on_progress(bytes: i64, total: i64) takes a fn(i64, i64)). One handler per event, as for a component’s events; an event nobody bound goes nowhere.

Passing an event down

A thread hands an event to a nested thread as a parameter of type event(T...), the event’s parameter types; the nested thread fires it and it reaches the same handler:

thread scanner {
    event on_found(path: string);

    thread walker {
        fn scan(dir: string, found: event(string)) {
            for (file in list_files(dir)) {
                fire found(file);
            }
        }
    }

    fn run(dir: string) {
        walker.scan(dir, on_found);
    }
}

Only a top-level thread declares events (only top-level threads are called async, so only their events can be bound), and an event is not named on_done or on_error. An event(T...) is a parameter type only: no variable, field or result holds one, and no cast makes one.

Cooperative cancellation

cancel(handle) cancels a task and, automatically, every thread call it is making. Cancellation is checked at each thread → thread call boundary. A compute-heavy loop without such calls checks explicitly with @check_cancel:

thread search {
    fn find(readonly items: []Item, query: string) -> Vec_Item or SearchError {
        var results: Vec_Item = vec_item_new();
        for (item in items) {
            @check_cancel or return SearchError.Cancelled;
            if (matches(item, query)) {
                results.push(item);
            }
        }
        return results;
    }
}

@check_cancel has type void or cancel: it succeeds, or produces the builtin cancel error. Handle it like any fallible expression:

• @check_cancel or return; — propagate cancel (the function must return T or cancel)
• @check_cancel or return MyError.Cancelled; — map it to your own error
• @check_cancel or { ... } — handle it inline

After cancellation, fired events not yet delivered are discarded, fire does nothing, and on_done / on_error never fire.

Fan-out parallel processing

Inside a thread function, @fan splits a pipe value and runs each piece on a nested thread’s pool in parallel:

thread work {
    var ranges: [32]Range;              // the splitter's pieces, in the thread's state

    @pool(1, 32)
    thread sums {
        fn part(r: Range) -> f64 { ... }    // one piece: the sum of 1/k over r
    }

    fn split(n: i64, parts: i64) -> []Range {
        for (i in 0..parts) { ranges[i] = ...; }
        return ranges[0..parts];
    }
    fn add(xs: []f64) -> f64 { ... }

    fn total(n: i64) -> f64 {
        return n @fan(split(32)) |> sums.part() |> add();
    }
}

• @fan(splitter(args)) calls splitter(pipe_value, args...), which returns []T. The splitter is any function; its pieces must outlive the call, so keep them in the thread’s state (as ranges above) or on the heap, not in a local array.
• Each element goes to the thread function on the right of |>: a function of a thread nested in this one (sums.part), run on that thread’s pool. Its @pool(min, max) says how many run at once; without one, a nested thread has one worker and the pieces run one at a time. Arguments after the piece are passed as written: |> sums.part(scale) calls part(piece, scale).
• The results are collected as []U in input order, whatever order the pieces finish in, and go on down the pipe (|> add()), so a sum of floating-point parts is the same every run. The slice lives as long as the task that made the fan.
• A piece function may fail (-> U or E): the fan is then []U or E, handled before the next stage: let parts: []f64 = (n @fan(split(8)) |> sums.risky()) or return;. The first failing piece’s error (in input order) is the fan’s; once a piece has failed, pieces not started yet are skipped and a running one sees its next @check_cancel fail.
• Cancelling the task that waits in a fan reaches every piece the same way: at its next @check_cancel.
• @fan is valid only inside thread functions (it blocks until every piece is done), and the function after it must be a nested thread’s.

Last modified: 2026-10-31 09:22:42 UTC