std/task

The generated reference of std/task has every public construct with its signature, its doc comment and its examples.

std/task is a computation that finishes later. 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 can be cancelled, and await() passes a cancellation on instead of answering it: it answers the Value, and a task that awaits a cancelled task is cancelled in turn. result() answers Result<Value, Cancelled> for the code that has to know. Task, Channel, ChannelClosed, Cancelled, TimedOut, spawn and both are in scope through the prelude; pause, offload, Workers, Permits, Pool, Lease and PoolError are imported from here.

Import#

use Task, Channel, ChannelClosed, Cancelled, TimedOut, Workers, Permits, Pool, Lease, PoolError, spawn, both, pause, offload from "std/task"

const numbers: List<Int> = [1, 2, 3, 4]

// Stages are lazy: without `toList()` nothing would be spawned until somebody iterates
const tasks = numbers.map { number => spawn { number * number } }.toList()

// Top-level `await()` is allowed in entry files and scripts. `Task.all` waits for all of them at once
const squares = Task.all(tasks).await()
print squares.sum()

Declarations#

Task#

native shared type Task<Value> {
  fn await(): Value
  fn result(): Result<Value, Cancelled>
  native fn runToEnd(): Result<Value, Cancelled>
  var fn cancel()
  var fn within(limit: Duration): Task<Result<Value, TimedOut>>
  fn map<Output>(transform: Transform<Value, Output>): Task<Output>
  fn flatMap<Output>(transform: Transform<Value, Task<Output>>): Task<Output>
  static fn all(tasks: Iterate<Self>): Task<List<Value>>
  static fn first(a: Self, b: Self): Task<Value>
}

A task starts running the moment it is created (spawn { ... }, or calling a function that returns one). await() waits for the value and is allowed in a function that returns a Task, in a closure passed to spawn, and at the top level of an entry file or a script - everywhere else it is a compile error, because a function that waits says so in its return type. It answers the Value itself, and where the task ended cancelled the waiter is cancelled too, at that very await(): it stops there, its scopes end in reverse order (every using is closed) as at any cancellation, and whoever awaits it 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. So a task whose value is a Result needs one ? and nothing more: file.addLine(text).await()?. result() waits the same way and answers a cancellation as Fail(Cancelled) instead of passing it on - for a supervisor, a test, or the task that cancelled another one and wants to confirm it.

runToEnd() is the wait of synchronous code: it runs the scheduler until the task has completed and answers as result() does, so a function that answers a plain value can still make a request or sleep on the way, in a program whose top level never calls await(). The other tasks keep running meanwhile, and the ones still alive when it returns go on at the next wait. Only the main thread outside every task may call it: inside a task, and on any other thread, it panics 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. A browser cannot wait without giving the page its turn, so there it panics as well.

cancel() asks the task to stop where it next waits or where a loop of it turns around; the tasks it started stop with it. within(limit) cancels the task once the limit has passed and answers Fail(TimedOut) then. Both are var fns, so a task somebody may stop sits in a var binding. Dropping the last handle stops nothing. Task belongs to the vocabulary of Option and Result (map, flatMap, all), and all three are TorbScript over await() and result(); a cancelled task stays cancelled through map and flatMap.

Task.first(a, b) waits for whichever of two tasks completes first and answers its value. The other one keeps running, for the caller to await later or to cancel, and where both have completed already a wins, so a tie never depends on the scheduler. Where the first to complete ended cancelled the waiter is cancelled too, as by await(), and cancelling the answer cancels neither task. Both tasks answer the same type: to race two kinds of work, map each into a type with a case for either.

spawn, both, pause#

native fn spawn<Value>(body: () => Value): Task<Value>
fn both<First, Second>(first: Task<First>, second: Task<Second>): Task<(First, Second)>
native fn pause(): Task<Void>

spawn runs the closure as a new task; it gets copies of what it captures and cannot capture a var binding, so there is nothing to race for. The free function both waits for two tasks of different types at once (const (user, posts) = both(fetchUser(1), fetchPosts(1)).await()); Task.all is the same idea for any number of tasks of the same type. Where one of them is cancelled, the others are cancelled too, and so is the whole - and with it whoever awaits it. pause() puts the running task at the back of the queue, which is what makes a long loop fair.

Workers#

native type Workers {
  native static fn count(): Int
  native static fn blocking(): Int
  native static fn copies(): Int
}

How many workers - operating-system threads - run the tasks of this process: TORB_WORKERS where it is set, and one per logical processor otherwise, at most 1024. It is fixed before the first line runs, because every worker owns a heap. A task that is started goes to the worker that started it; an idle worker takes it before its first run where every value it holds may cross to another thread - numbers, a Channel of numbers, a text or a list nobody else holds - and a task never moves once it has run. With TORB_WORKERS=1 the tasks run one at a time in the order they became ready, which is the order a program that prints from several tasks at once can rely on. Workers is imported from here; it is what parallel(workers:) defaults to. blocking() is the size of the blocking pool below: TORB_BLOCKING where it is set, and 4 otherwise, at most 1024. copies() counts the values copied at a crossing to another worker since the process started - a statistic, which a program reads to show that something crossed without one, as windows does.

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: offload({ File.readText("notes.txt") }).await(). The pool has Workers.blocking() threads, started the first time a body moves there, so it is for one blocking call and not for a long computation, which belongs on the workers. The body moves only where its closure may cross to another thread - where it captures nothing counted, or only literals and 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, with the same answer and that worker blocked. The body cannot await, and a cancellation does not interrupt a body that already runs: it runs to its end and its answer is thrown away.

Channel#

native shared type Channel<Item> {
  capacity: Int = 0
  fn source(): ChannelSource<Item>
  fn sink(): ChannelSink<Item>
}

A stream in memory of which one holder has both ends. capacity: 0 hands every item over directly, so add waits until somebody pulls; source() is a Source<Item, Never> and sink() a Sink<Item, ChannelClosed> (see std/stream), and they can be handed out separately, so a producer never sees the reading end and a consumer never sees the writing one. Reading cannot fail: a closed channel is the end of the stream, not a failure.

Permits#

shared type Permits {
  count: Int
  static fn of(count: Int): Self
  var fn acquire(): Task<Void>
  var fn release()
}

A semaphore for tasks: a count of free permits, and the tasks that wait for one, oldest first. Permits.of(count) makes count permits, all free; a count below 1 is taken as 1. It is for a resource that only so many tasks may use at once - the connections a pool opens, or a connection that runs one statement at a time (Permits.of 1).

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

acquire() takes a free permit at once, and otherwise waits; the task that has waited longest is served first. A task that is cancelled while it waits takes no permit, and passes on the one it was handed if the permit reached it before the cancellation did. release() hands the permit to the oldest task that still waits, and otherwise adds it to the free ones. It waits for nothing and starts no task, so it may run in a task that is being cancelled - for instance from the close() of a value bound with using, which is how a permit is given back however the work ends. A permit that was taken has to be given back once: a release() without an acquire() adds a permit.

Pool, Lease, PoolError#

shared type Pool<Resource, Failure> {
  maximum: Int
  static fn of(maximum: Int, open: () => Task<Result<Resource, Failure>>, reusable: (resource: Resource) => Bool = { _ => true }): Self
  var fn acquire(): Task<Result<Lease<Resource>, PoolError<Failure>>>
  var fn stop()
  var fn close()
  fn idleCount(): Int
  fn openCount(): Int
}
shared type Lease<Resource> with Close {
  resource: Resource
  var fn giveBack()
  var fn discard()
  var fn close()
}
type PoolError<Failure> {
  case Closed
  case Open(failure: Failure)
}
extend<Failure: Show> PoolError<Failure> with Show

A pool of resources, written once over Permits: at most maximum of them are open at once, the idle ones are kept for the next acquire(), and a caller beyond maximum waits, oldest first. It is for a resource that is dear to open and harmful in numbers - connections to a server, handles of a device, helper processes. Pool.of(maximum, open) makes it; a maximum below 1 is taken as 1, and reusable (true for every resource unless given) tells whether an idle resource may be lent again.

const pool = Pool.of(4, { connect(address) })
using lease = pool.acquire().await()?
lease.resource.send(request).await()?
lease.giveBack()

acquire() waits for a free slot, takes the newest idle resource that is reusable - the others it finds on the way are dropped - and otherwise calls open. An open that fails frees the slot and answers PoolError.Open(failure); a pool that was stopped answers PoolError.Closed, also where it was stopped while the caller waited. A caller cancelled while it waits takes no slot. What the pool leaves to the caller: a key per origin (keep one Pool for each), a timeout (pool.acquire().within(limit), as Permits.acquire is a task), and transactions or redirects.

A Lease settles exactly once, however often it is asked: giveBack() keeps the resource for the next caller when the pool is running and reusable agrees, discard() drops it, and both free the slot. The release of a lease discards where neither ran, so a lease bound with using gives its slot back when the work fails or is cancelled. stop() drops the idle resources, later returns are dropped and later acquire() calls answer Closed; stopping twice is nothing, and close() (the release of the pool) does the same. idleCount() and openCount() serve tests and monitoring. reusable runs inside acquire() for every idle resource it looks at, so keep it cheap, and do not let it wait. Show for PoolError needs a Show failure.

ChannelClosed, Cancelled, TimedOut#

type ChannelClosed with Show, Error {
  fn show(): String
}
extend ChannelClosed with From<ChannelClosed>
type Cancelled with Show, Error {
  fn show(): String
}
type TimedOut with Show, Error {
  limit: Duration
  fn show(): String
}

ChannelClosed is the only way a Channel's writing end fails: nobody is reading any more. Cancelled is what result() answers for a task that stopped because somebody asked it to; it carries no reason, and await() never answers it. TimedOut is the failure within answers, with the limit that was too small.

Related#

• Tasks - the rules of Task, await() and cancellation.
• std/stream - Source and Sink, the two ends a Channel hands out.
• The standard library - the other packages.