nupp.tasks

Structured concurrency: a scope owns the tasks spawned into it.

Open a scope, spawn children into it, and the block that holds it does not end until every one of them has run, been cancelled, or unwound.

local greeting = ""

with scope = nupp.tasks.open() do
    const hello = scope:spawn(function(): string
        return "hello"
    end)
    const world = scope:spawn(function(): string
        return "world"
    end)

    greeting = hello:await() .. " " .. world:await()
end

assert(greeting == "hello world")

open takes a limit, which parks spawn and fork while that many children are live, and a timeoutMs, how long the scope may take. A running child cooperates by calling checkpoint, which is where it learns the scope was cancelled or its deadline passed.

local total = 0

with scope = nupp.tasks.open(timeoutMs = 1000) do
    const sum = scope:spawn("sum values", function(): integer
        local answer: integer = 0
        for value = 1, 1000 do
            nupp.tasks.checkpoint()
            answer = answer + value
        end

        return answer
    end)

    total = sum:await()
end

assert(total == 500500)

Growing task families#

nupp.suspension's combinators own a family that is complete when the call is written: all cannot be handed a fourth body once it is running. create makes a coroutine that inherits a handler and says nothing about its result, failure, or lifetime. A server, a loading pipeline, or a scene needs somewhere to put children found over time and one answer to what happens when the body returns, a child fails, or the whole thing is cancelled.

CPU work on other threads is nupp.workers, and a task scope is the only way to it: fork hands a child to a lane, and the scope owns the family of lanes lazily and closes it through the suspension-aware path, so leaving the block inside a frame parks rather than blocking the thread.

The scheduling contract is with the host rather than with each child. The host sees one aggregate and decides when it runs; this decides which children run then, in FIFO order, up to a bounded number of activations per host turn. Nesting divides that bound rather than multiplying it, because the token belongs to the turn.

Module contents

Types

TypeKindDescription
CancellationrecordRepresents what a cancelled task raises.
ScoperecordRepresents a scope opened by open.
StatustypeWhere a task is, as its status returns: queued, running, done, failed, or cancelled.
TasktypeRepresents the handle one spawn or fork returns.

Functions

FunctionKindDescription
checkpointfunctionRaises where the current task has been cancelled or has run out of time.
deadlinefunctionReturns the effective deadline of the current task, or nil where there is none.
isCancelledfunctionReports whether a caught value is a task cancellation.
openfunction

Types#

Cancellationrecord#

record tasks.Cancellation ...

Represents what a cancelled task raises.

Nominal rather than a string, so isCancelled recognizes it without matching text, and so a program that catches everything still sees something it can ask about. tostring renders it because an uncaught one reaches a human.

Members

NameKindDescription
operationfieldWhat was cancelled, as it was named when it was started.
reasonfieldWhy, where a caller supplied a reason.

operationfield#

operation: string

What was cancelled, as it was named when it was started.

reasonfield#

reason: string?

Why, where a caller supplied a reason.

Scoperecord#

record tasks.Scope is nupp.Closeable ...

Represents a scope opened by open.

A nupp.Closeable: open returns it as an owner a with block holds, and leaving the block closes it. The block can add children to the scope and reach the worker scope it owns.

Members

NameKindDescription
spawnmethodStarts a child under this scope: scope:spawn(arguments..., f), or scope:spawn(name, f) to name one that takes no...
forkmethodStarts a child on a worker lane: scope:fork(arguments..., f).
cancelmethodRequests cancellation of every child.
closemethodSettles the scope, which is what leaving its with does.

spawnmethod#

spawn: function<F>(borrows self: tasks.Scope, ...: unpackof Parameters(F), takes body: F): tasks.Task<F> borrows (self) & function<F is Nullary>(borrows self: tasks.Scope, name: string, takes body: F): tasks.Task<F> borrows (self)

Starts a child under this scope: scope:spawn(arguments..., f), or scope:spawn(name, f) to name one that takes no arguments.

Arguments
NameTypeDescription
borrows selftasks.Scope
...unpackof Parameters(F)
takes bodyF
Returns
TypeDescription
tasks.Task<F> borrows (self)

forkmethod#

fork: function<F is Submittable>(borrows self: tasks.Scope, ...: unpackof nupp.workers.spi.Submitted(F), F): tasks.Task<F> borrows (self)

Starts a child on a worker lane: scope:fork(arguments..., f).

Arguments
NameTypeDescription
borrows selftasks.Scope
...unpackof nupp.workers.spi.Submitted(F)
?F
Returns
TypeDescription
tasks.Task<F> borrows (self)

cancelmethod#

cancel: function(borrows self: tasks.Scope, reason: string?): nil

Requests cancellation of every child.

Arguments
NameTypeDescription
borrows selftasks.Scope
reasonstring?
Returns
TypeDescription
nil

closemethod#

close: function(takes self: tasks.Scope): nil

Settles the scope, which is what leaving its with does.

Arguments
NameTypeDescription
takes selftasks.Scope
Returns
TypeDescription
nil

Statustype#

type Status = "queued" | "running" | "done" | "failed" | "cancelled"

Where a task is, as its status returns: queued, running, done, failed, or cancelled. A closed union, so a switch over it needs no else.

Tasktype#

type Task<F> = TaskType(F)

Represents the handle one spawn or fork returns.

Its await result pack is the body's, which is why this is derived from the function type rather than declared once over any. The worker contract declares it, so a coroutine child and a worker child return the same type.

Type parameters

NameDescription
F

Functions#

tasks.checkpointfunction#

function tasks.checkpoint(): nil

Raises where the current task has been cancelled or has run out of time.

The one authored cancellation point, and the only thing that reaches a body computing without parking. It never suspends, so a @nosuspend region and a worker lane can both call it, and outside any task it does nothing.

for index = 1, #items do
    tasks.checkpoint()
    consume(items[index])
end

Returns

TypeDescription
nil

Raises

TypeCondition
tasks.Cancellation

where one has been requested

tasks.deadlinefunction#

function tasks.deadline(): number?

Returns the effective deadline of the current task, or nil where there is none.

So a body can size its work rather than discover the bound by being cancelled part way through it.

Returns

TypeDescription
number?

an absolute monotonic reading, as nupp.time.now returns

tasks.isCancelledfunction#

function tasks.isCancelled(value: any): boolean

Reports whether a caught value is a task cancellation.

The one question a pcall around task work has to be able to ask, because cancellation is not a failure and should usually be re-raised rather than reported.

const ok, problem = pcall(work)
if not ok and not tasks.isCancelled(problem) then
    report(problem)
end

Arguments

NameTypeDescription
valueany

whatever was caught

Returns

TypeDescription
boolean

whether this is a cancellation

tasks.openfunction#

function tasks.open(limit: integer?, timeoutMs: number?): affine(tasks.Scope)

Arguments

NameTypeDescription
limitinteger?
timeoutMsnumber?

Returns

TypeDescription
affine(tasks.Scope)