Workers & Barriers

Async functions that live across frames and keep their state, and the barrier where several of them meet to share one computation.

A worker is an exported async function marked @worker. Worker.spawn starts it in its own instance, with copies of the arguments you pass. Unlike a job, a worker keeps its instance and its state until it returns.

A worker waits on signals, ports and Worker.sleep, and the host resumes it when one fires. Natively it runs on a thread from the worker pool between awaits, and uses no thread at all while it awaits.

A @worker function must be exported under its own name and must not be a generator. Promise needs no import, as in TypeScript.

A worker as a service

The usual shape is a service: a loop that awaits a request on a port, works on it, and posts the reply to another port. A port carries typed messages from one lane to another, each one a deep copy. The ports themselves cross to the worker by id, so both lanes use the same two ports.

import { Layout, Context } from "rive/host";
import { Worker } from "rive/workers";
import { Port } from "rive/ports";

@worker export async function doubler(requests: Port<f64>, replies: Port<f64>): Promise<void> {
    while (true) {
        let request = await requests.receive();
        if (request == null) {
            return;
        }
        replies.post(request.value * 2);
    }
}

export class Doubling extends Layout {
    requests: Port<f64> = Port.create<f64>();
    replies: Port<f64> = Port.create<f64>();
    worker: Worker? = null;
    latest: f64 = 0;

    override init(context: Context): bool {
        this.worker = Worker.spawn(doubler, this.requests, this.replies);
        this.requests.post(21);
        return true;
    }

    override advance(seconds: f64): bool {
        let self = this;
        this.replies.drain((value: f64): void => {
            self.latest = value;
        });
        return true;
    }
}

receive settles with null when the worker is being stopped or the port is gone, so the loop returns then. In advance, drain hands each queued reply to the handler and never waits.

drainFor(handler, seconds) drains the same way, but stops once the seconds pass. It checks the time after each message, so it hands over at least one when any is queued. A processor has no clock, so it cannot use drainFor.

Answering within the frame

The main lane can also wait for a worker’s answer inside the call that asked. Port.receiveWithin(seconds) waits for a message, and Signal.waitWithin(after, seconds) waits for a notify. They block as a job’s join does, never past the call’s budget. Only the main lane may block this way. A worker awaits instead.

import { Layout, Context } from "rive/host";
import { Worker } from "rive/workers";
import { Port } from "rive/ports";

@worker export async function halver(requests: Port<f64>, replies: Port<f64>): Promise<void> {
    while (true) {
        let request = await requests.receive();
        if (request == null) {
            return;
        }
        replies.post(request.value / 2);
    }
}

export class Halving extends Layout {
    requests: Port<f64> = Port.create<f64>();
    replies: Port<f64> = Port.create<f64>();
    worker: Worker? = null;
    latest: f64 = 0;

    override init(context: Context): bool {
        this.worker = Worker.spawn(halver, this.requests, this.replies);
        return true;
    }

    override advance(seconds: f64): bool {
        this.requests.post(42);
        // The answer to this frame's request, or null after 4 ms.
        let reply = this.replies.receiveWithin(0.004);
        if (reply != null) {
            this.latest = reply.value;
        }
        return true;
    }
}

Without a budget, these waits last at most two seconds. They throw inside SharedRegion.exclusive. In a browser they spin for at most a tenth of a second. In a build without threads they never wait, since a worker only steps between advances.

Stopping a worker

stop() asks a worker to end. Its awaits settle with kWaitStopping, which imports from rive/lane_codes, and Lanes.stopping(), from rive/lanes, turns true. A worker that awaits again ends at that await. Natively, one still running past a short grace is interrupted. In a browser that takes fuel (see Platform differences); without it, a stopped worker ends only at its next await.

A worker can also return on its own. Its promise then resolves with what it returned, which crosses back as a deep copy. Read it with peek, peekInto or promise(), as for a job. Releasing the handle, or letting it be collected, stops the worker. So keep the Worker in a field for as long as it should run.

Lanes cannot deadlock, but a worker can stall

A deadlock is threads each blocked waiting for another. Lanes cannot deadlock:

• A worker waits for signals, ports and sleeps only by awaiting, which blocks no thread.
• A region lock waits at most two seconds, and no lane can wait while it holds one.
• Every wait that blocks the main lane gives up within the call’s budget.

A call without a budget is the one exception. There a join waits for its job to end, so a job that loops until the main lane acts hangs both.

A worker can stall, though. One that awaits an event that never comes, such as a notify that the other workers at a barrier never send, waits until it is stopped. Nothing detects a stall. But when a headless run of rive ends, the log names every worker still waiting, what it awaits and for how long:

worker phase is still waiting for signal 2147487744 after 1.25 s

A port’s messages queue in a ring, so a worker awaiting a port’s receive waits for a write to that ring. A script reads the ring’s id as port.ring.id. Nothing checks whether anything could still wake a worker, so a service waiting for its next request is named too.

How many workers to spawn

Spawn a worker per service, not per core. Natively, a worker holds no thread while it awaits, and a step that runs past 20 ms gets a thread of its own. Workers whose steps are shorter run at once only up to the worker pool’s size: one thread fewer than the device has cores, and at most 4.

• Spread independent pieces of work over the cores with jobs. Jobs.threads() says how many jobs run at once.
• Jobs cannot meet between steps, though. The parts of one computation that keep large state from step to step, or must meet between steps, go to workers that meet at a barrier, as the next section shows.
• Workers that share one request port, to keep state from one request to the next, gain little past Jobs.threads() of them.

Where Jobs.threads() is 0 there are no threads, and workers step between frames.

Splitting one computation

One computation can be split over workers that meet between its steps. An example is a model whose layers each normalize over the data that all of the workers hold. The workers meet at a Barrier, from rive/shared, which crosses to them through Worker.spawn.

Barrier.create(maxParticipants, shareFloats) makes one for rounds of up to that many participants, each bringing a share of up to that many f32 values. In each round, every participant awaits one of these:

• arrive(round, participants) only meets.
• sum, min or max(round, participants, index, share, out) also combines the participants’ shares. The last to arrive combines them in index order, so out holds the same bits in every participant. out may be the share itself.

Each call settles true once every participant of the round has arrived. The rules:

• Number the rounds from 0, with no gaps, over every use of the barrier.
• Pass each round’s participant count. The count can change from round to round, but the participants of one round all pass the same count.
• Each participant passes its own index, from 0 to the count less one.
• All participants of a round call the same method, with shares of the same length.
• A worker that sits rounds out calls the next round it takes part in. That call waits until the rounds before it are over.

import { Layout, Context } from "rive/host";
import { Worker } from "rive/workers";
import { Port } from "rive/ports";
import { Barrier } from "rive/shared";

const kWorkers: i32 = 4;
const kSteps: i32 = 8;
const kSlice: i32 = 1024;

// Short chunks are split over two parts, long ones over all four.
function parts(chunk: i32): i32 {
    return chunk % 3 == 2 ? kWorkers : 2;
}

@worker export async function part(index: i32, chunks: i32, barrier: Barrier, done: Port<f32>): Promise<void> {
    // This part's slice of the chunk, kept from step to step.
    let values = new Float32Array(kSlice);
    let moments = new Float32Array(2);
    for (let chunk = 0; chunk < chunks; chunk++) {
        let count = parts(chunk);
        if (index >= count) {
            continue;
        }
        for (let i = 0; i < kSlice; i++) {
            values[i] = <f32>((chunk + index * kSlice + i) % 13);
        }
        for (let step = 0; step < kSteps; step++) {
            moments[0] = 0;
            moments[1] = 0;
            for (let i = 0; i < kSlice; i++) {
                moments[0] += values[i];
                moments[1] += values[i] * values[i];
            }
            // Rounds count on over every chunk, and moments then holds the
            // totals over every part.
            let met = await barrier.sum(chunk * kSteps + step, count, index, moments, moments);
            if (!met) {
                return;
            }
            let n = <f32>(count * kSlice);
            let mean = moments[0] / n;
            let deviation = Mathf.sqrt(max<f32>(moments[1] / n - mean * mean, 1e-6));
            for (let i = 0; i < kSlice; i++) {
                values[i] = (values[i] - mean) / deviation;
            }
        }
    }
    done.post(values[0]);
}

export class Normalize extends Layout {
    barrier: Barrier = Barrier.create(kWorkers, 2);
    done: Port<f32> = Port.create<f32>();
    workers: Array<Worker> = new Array<Worker>();
    first: f32 = 0;

    override init(context: Context): bool {
        for (let i = 0; i < kWorkers; i++) {
            this.workers.push(Worker.spawn(part, i, 6, this.barrier, this.done));
        }
        return true;
    }

    override advance(seconds: f64): bool {
        for (let i = 0; i < this.workers.length; i++) {
            // The other parts would wait for a failed one forever.
            if (this.workers[i].failed != null) {
                this.barrier.break();
            }
        }
        let self = this;
        this.done.drain((value: f32): void => {
            self.first = value;
        });
        return true;
    }
}

When a participant fails

A call settles false once the barrier is broken or released, and inside a worker being stopped. A participant that fails or is stopped, or whose call is cancelled, leaves the others waiting for it. A call is cancelled, for example, when it is one of the promises given to Promise.all and another of them rejects.

So stop every participant together, or call break(), as the example’s advance does. It settles every wait on the barrier false, and every call from then on.

What a round costs

A round ends only when its slowest part arrives. So give the parts equal work, and split the computation into no more parts than can run at once. Natively that is the worker pool’s size, unless the host sizes it otherwise, as rive --worker-threads=<N> does.

Parts that meet more often than every 20 ms never get a thread of their own. So a part past the pool’s size waits for a thread every round, and a round then takes at least twice as long as one part’s work. In a browser with threads, each worker keeps a thread of its own, up to about 8.

A round also costs a few hundredths of a millisecond, mostly to wake the parts that wait for it. Natively on an Apple M4 Pro, on a pool of 8 threads, with a fifth of a millisecond of work in each part between rounds:

With much less work between rounds, a round costs up to twice that. A part that reaches the next round first waits there for the others to leave the last one.

For every detail of the API, run rive docs animascript/api/workers, rive docs animascript/api/ports and rive docs animascript/api/shared. The full guide is rive docs animascript/threading.