Reference

std/task/lib

std/task/src/lib.trb

Concurrent work: Task, a computation that finishes later, and Channel, a queue between two tasks.

Asynchrony lives in the type: a function that returns Task<Value> may call await(), and its body produces the Value - the same way a function that returns Result may use ?. Task and Channel are the two shared types that connect concurrent work; everything else is copied when it crosses into one.

Every task is cancellable (Task.cancel), and a cancellation is passed on rather than answered: await() answers the Value, and a task that awaits a task that ended cancelled is cancelled itself, right there - its scopes end and whoever awaits it is cancelled in turn. So an IO line is file.addLine(text).await()?, one ? for the failure of the work and nothing for a cancellation. Task.result is the one way to observe a cancellation as a value (Cancelled), for a supervisor or for code that cancelled a task and wants to confirm it.

type Task

native shared type Task<Value>

A computation that finishes later, running from the moment it is created (spawn { ... }, or calling a function that returns one).

Reach for it wherever work should happen while the caller keeps going. map and flatMap mean the same as everywhere else and work on the finished value.

Pitfalls

  • Dropping the last handle does not stop the task: a task runs until it finishes or somebody calls Task.cancel.

Related

  • spawn - starts a task from a closure.
  • both - waits for two tasks of different types at once; Task.all does the same for any number of the same type.
  • Task.first - waits for the first of two tasks of the same type, and leaves the other one running.

fn await

fn await(): Value

Waits for the value. Allowed in functions that return a Task, in closures passed to spawn and at the top level of entry files and scripts.

A cancellation is passed on, not answered. Where the task ends cancelled, the waiter is cancelled too, at this very point: it stops here, its scopes end in reverse order (every using is closed) exactly as when it is cancelled at a loop turn, and whoever awaits the waiter is cancelled in turn. At the top level of an entry file that ends the program with cancelled: the program waited for a task that was cancelled and exit code 130. A waiter that is cancelled itself never sees an answer either: it stops where it waits.

Panics

Never where the task ended cancelled: the waiter stops before the answer, as above. The panic in the body guards a state the wait rules out, with a task read the value of a task that was cancelled.

Related

  • Task.result - waits the same way and answers a cancellation as Fail(Cancelled) instead of passing it on.

fn result

fn result(): Result<Value, Cancelled>

Waits like Task.await, and answers a cancellation instead of passing it on: Ok(value) where the task finished, Fail(Cancelled) where it ended cancelled. The waiter goes on either way. Allowed where await() is.

It is for the code that has to know: a supervisor that restarts what was cancelled, a test, the task that called Task.cancel and wants to confirm it. Everything else writes await(), and a cancellation stops it.

fn runToEnd

native fn runToEnd(): Result<Value, Cancelled>

Runs the scheduler until the task has completed, from synchronous code, and answers as Task.result does: Ok(value) where it finished, Fail(Cancelled) where it ended cancelled. Every other task keeps running while this waits, and the ones still alive when it returns go on at the next wait, or are cancelled at the end of the program as always.

It is for code that is not a task and cannot become one: a function that answers a plain value and still needs a request or a timer on the way, in a program whose top level never calls await(). Everywhere else await() and result() wait, and a top level that calls either of them anywhere is a task itself.

Panics

  • Inside a task - a function that returns a Task, a closure passed to spawn or offload, a top level that calls await() - and on a thread other than the main one, with `runToEnd()` was called inside a task: only synchronous code outside every task runs a task to its end, and a task waits with `await()` or `result()`. One wait at a time runs the scheduler, and a task that blocked its thread for another one would stop the tasks that share it.
  • In a browser, which cannot wait without giving the page its turn: there the top level waits with await().
  • On a deadlock, as every wait of the main thread does: the task is pending and nothing is left that could wake it.

A task that panics ends the program as it always does, whoever waits for it.

Related

fn cancel

var fn cancel()

Asks the task to stop at its next suspension point or loop turn. A request and not a kill; asking twice, or asking a task that has finished, changes nothing. The tasks this one started are cancelled with it.

fn within

var fn within(limit: Duration): Task<Result<Value, TimedOut>>

Cancels the task once the limit has passed: Ok(value) where it finished in time, Fail(TimedOut) where the limit passed first. A task that finished first is unaffected. Cancelling the answer does not cancel this task, and a task somebody else cancelled cancels the answer, as every await() of it would.

fn map

fn map<Output>(transform: Transform<Value, Output>): Task<Output>

Changes the value once the task finishes, without waiting for it here. A cancelled task stays cancelled.

fn flatMap

fn flatMap<Output>(transform: Transform<Value, Task<Output>>): Task<Output>

Continues with another task once this one finishes, and keeps one level: the async form of Result.flatMap.

fn all

static fn all(tasks: Iterate<Self>): Task<List<Value>>

A task for every item, running at the same time. The values arrive in the order of the tasks. Where one of them is cancelled the answer cannot be built, so the others are cancelled too and this task ends as cancelled - and whoever awaits it with it.

fn first

static fn first(a: Self, b: Self): Task<Value>

Waits for whichever of two tasks completes first and answers its value. The other task keeps running, for the caller to await later or to cancel. Where both have completed by the time the answer is decided, a wins, so a tie never depends on the scheduler.

It is for a loop that waits on two sources at once - a message of the client or an event of the server, whichever comes first - and keeps waiting on the one that did not answer in the next turn. Both tasks answer the same type: to race two kinds of work, map each into a type with a case for either, which then also says which one it was.

A cancellation is passed on, not answered, as by Task.await: where the first task to complete ended cancelled, the waiter is cancelled too. Cancelling the answer cancels neither task.

Examples

type Event {
  case Message(text: String)
  case Tick
}

const message = spawn { Event.Message "hello" }
const tick = spawn { Event.Tick }
print Task.first(message, tick).await()

Pitfalls

  • The task that lost is not cancelled. A task nobody waits for any more still runs to its end: cancel it where its work is no longer wanted.

Related

  • Task.within - races a task against a limit instead, and cancels it when the limit comes first.
  • both - waits for both tasks rather than for the first.

fn spawn

native fn spawn<Value>(body: () => Value): Task<Value>

Runs the closure as a new task, in parallel. It gets copies of what it captures and cannot capture a var.

fn both

fn both<First, Second>(first: Task<First>, second: Task<Second>): Task<(First, Second)>

Waits for two tasks of different types at once: const (user, posts) = both(fetchUser(1), fetchPosts(1)).await(). Where one of them is cancelled, the other is cancelled too and this task ends as cancelled - and whoever awaits it with it.

It is both and not all: it takes exactly two tasks, each of its own type, while Task.all waits for a list of tasks of one type and Iterate.all asks a predicate of every item.

type Workers

native type Workers

The worker pool of this process: how many operating-system threads run its tasks. A reading namespace - the count is decided before the first line of the program runs and never changes, because every worker owns a heap.

Related

  • spawn - a task an idle worker may take before it first runs.
  • offload - a body that blocks, run on the blocking pool.

fn count

native static fn count(): Int

How many workers this process runs: TORB_WORKERS where it is set, and otherwise one per logical processor, at most 1024. What parallel(workers:) defaults to.

fn blocking

native static fn blocking(): Int

How many threads the blocking pool has, which runs the bodies of offload: TORB_BLOCKING where it is set, and otherwise 4, at most 1024. The threads start the first time a body moves there.

fn copies

native static fn copies(): Int

How many values were copied at a task crossing since the process started: the frame of a task that may move, a message of a Channel, the environment of a closure (docs/design/CONCURRENCY.md section 16). A statistic for measuring - two readings around a piece of work say how much of it was copied - and nothing the program should decide by: the number depends on the workers, and is 0 where there is one.

Examples

const before = Workers.copies()
print(Workers.copies() >= before)

fn pause

native fn pause(): Task<Void>

Puts this task at the back of the queue and lets every other ready task run first. It is what makes a long loop fair; a loop of a task is cancellable at every turn without it.

fn offload

fn offload<Value>(body: () => Value): Task<Value>

Runs body on a thread of the blocking pool instead of on a worker, so a call that blocks - reading a file, running a child process, a C library that waits - does not stall the other tasks of the worker that asked.

const text = offload({ File.readText("notes.txt") }).await()

Reach for it around one blocking call, not around a computation: a computation that is merely long belongs on the workers, with spawn or parallel(), because the pool has only Workers.blocking threads.

Pitfalls

  • The body moves to the pool only where the closure may cross to another thread: where it captures nothing counted, or only literals and other values nothing else holds a count of. A closure that captures a String built at run time, a list or a shared type object runs on the worker that asked - the same answer, but it blocks that worker as a direct call would (docs/design/CONCURRENCY.md section 16, "The blocking pool, as built").
  • The body cannot await. Cancelling the task does not interrupt a body that already runs: it runs to its end on its thread, and what it answers is thrown away.

type Channel

native shared type Channel<Item>

A stream in memory, of which one holder has both ends: a queue between two tasks.

const channel = Channel<Report>(capacity: 8)
const writer = spawn { channel.sink().fill(reports).await() }
const total = channel.source().count().await()

channel.source() and channel.sink() are a Source and a Sink and can be handed out separately, so a producer never sees the reading end and a consumer never sees the writing one. They are methods and not fields because they are computed, and because a field would have to be passed to the constructor. They answer the concrete ChannelSource and ChannelSink, so a call on one is a direct call; where a Source or a Sink is asked for, either of them is one.

Reading cannot fail: a closed channel is the end of the stream, not a failure, and a cancellation is passed on rather than answered. The writing end fails with ChannelClosed once the reader is gone, which is how a producer learns that nobody wants its items any more - the cancellation signal of Source.produce.

field capacity

capacity: Int = 0

0 hands every item over directly: add waits until somebody pulls.

fn source

fn source(): ChannelSource<Item>

Pulls items in the order they were added, and ends when the writing end is finished or closed.

fn sink

fn sink(): ChannelSink<Item>

add waits while the channel is full, end() tells the reading end that the stream ended.

type ChannelSource

shared type ChannelSource<Item> with Source<Item, Never>

The reading end of a Channel, a Source<Item, Never>: every next() waits for the next item, and close() tells the writers to stop.

fn of

static fn of(channel: Channel<Item>): Self

The reading end of channel: what Channel.source answers.

fn next

var fn next(): Task<Result<Item?, Never>>

The task that waits for the item ends as cancelled where the stream ended, which is None here.

fn close

var fn close()

Closes the reading end: what the channel still buffers is dropped, and every writer fails with ChannelClosed.

type ChannelSink

shared type ChannelSink<Item> with Sink<Item, ChannelClosed>

The writing end of a Channel, a Sink<Item, ChannelClosed>: add finishes once a reader took the item, end() and close() end the stream.

fn of

static fn of(channel: Channel<Item>): Self

The writing end of channel: what Channel.sink answers.

fn add

var fn add(item: Item): Task<Result<Void, ChannelClosed>>

The task that offers the item ends as cancelled where nobody reads any more, which is ChannelClosed here.

fn end

var fn end(): Task<Result<Void, ChannelClosed>>

Finishes the writing end: what was offered is still delivered, then the stream ends. It always answers Ok.

fn close

var fn close()

Finishes the writing end like end(), for a caller that has no result to handle.

type Permits

shared type Permits

A semaphore for tasks: a count of free permits, and the tasks that wait for one, oldest first.

const permits = Permits.of 2
var turn = permits
turn.acquire().await()
// ... at most two tasks are here at once
turn.release()

Reach for it wherever only so many tasks may use a resource at once: the connections a pool opens, or a connection that runs one statement at a time (Permits.of 1). A task that waits holds the reading end of a channel of its own; a permit is handed to it by dropping the writing end, whose release ends the stream - so Permits.release waits for nothing and starts no task. That matters: a permit is given back by a release of a scope, which may run in a task that is being cancelled, and a task started there would be cancelled with it.

Pitfalls

  • A permit that was taken has to be given back, once: Permits.release without a matching Permits.acquire adds a permit. Give it back from the close() of a type that is bound with using, so that a failure or a cancellation gives it back too.
  • A task cancelled while it waits takes no permit, and passes on one it was handed; a task that holds a permit and is cancelled has to give it back itself.

field count

count: Int

How many permits it was made with: what are free when no task holds one.

fn of

static fn of(count: Int): Self

A semaphore of count permits, all free; a count below 1 is taken as 1.

fn acquire

var fn acquire(): Task<Void>

Waits for a free permit and takes it: at once where one is free, otherwise as the oldest waiter is served first. A task cancelled while it waits takes none, or passes on the one it was handed.

fn release

var fn release()

Gives a permit back: to the oldest task that still waits, or to the free ones. It waits for nothing and starts no task, so it may run in a task that is being cancelled.

type ChannelClosed

type ChannelClosed with Show, Error

Nobody is reading any more: the reading end of the channel was closed. The only way a Channel's writing end fails, and a value rather than a panic, because a producer with nobody left to feed should stop, not crash.

fn show

fn show(): String

"the channel is closed".

extend ChannelClosed with From<ChannelClosed>

extend ChannelClosed with From<ChannelClosed>

A producer whose own failure is ChannelClosed needs no conversion to learn that nobody reads any more.

fn from

static fn from(value: Self): Self

type Cancelled

type Cancelled with Show, Error

A task that stopped because somebody asked it to, as Task.result answers it. It carries no reason: who asked and why is the canceller's knowledge, and a deadline has a failure of its own, TimedOut. await() never answers it - a cancellation it meets is passed on to the waiter instead.

fn show

fn show(): String

"the task was cancelled".

type TimedOut

type TimedOut with Show, Error

A deadline that passed: what Task.within answers when the limit came first.

field limit

limit: Duration

How long the waiter was willing to wait.

fn show

fn show(): String

"the task did not finish within {limit}".