weft

Owned, bounded structured concurrency.

A run is a list of tasks, a bound on how many of them may be in flight at once, and a complete account of what happened to every one of them. The account is the point: start hands back one Outcome per task, including the tasks that crashed, the tasks that were cancelled mid flight, and the tasks that never got a slot. There is no path through this module where a late failure discards work that already succeeded.

Why there is an extra process

The guarantee this module exists to make is no task outlives its scope, and no scope outlives its caller, and it is enforced by link propagation rather than by a collector loop that has to stay alive to do its job. A library that spawns unlinked workers and reaps them from a loop degrades to “no work outlives the VM” the moment something kills that loop.

So every run pays for one extra process, the scope, and the link topology does the work:

caller
  |  link          (process.spawn, which is proc_lib:spawn_link)
  v
scope              (traps exits)
  |  link   link   link
  v         v      v
 w0        w1     w2         (workers, which do NOT trap)

The caller spawns the scope linked to itself.
The scope calls process.trap_exits(True) and spawns every worker linked to itself, from inside the scope process.
A worker crashing therefore arrives at the scope as a trapped EXIT and becomes a Crashed outcome. It does not take the scope down with it.
The scope being killed — including by a brutal supervisor kill — propagates down those links to workers that do not trap, so they die with it. Nothing has to still be looping for that to hold.
The caller dying sends an EXIT to the scope, which traps it, kills its workers, waits for them, and exits.

One asymmetry in that picture is worth stating because it shaped the code: a normal exit signal is ignored by a process that does not trap exits. So the links alone do not reap workers when the scope finishes cleanly; on every path where the scope stops early it kills its workers explicitly and waits for their EXITs before returning. settled is the predicate that encodes it, and it counts workers that have already answered but not yet exited (finishing) precisely so that the scope cannot return while one of its children is still alive.

Kill then join, in that order

Cancellation sends every kill first and only then waits for the EXITs. Interleaving the two lets one slow exit postpone every later worker’s cancellation, which turns a bounded teardown into a serial one. begin_cancel does the sending; the ordinary receive loop does the joining.

Delivery is pull, not push

fold consumes outcomes in completion order, and the streaming claim would be a lie if the scope pushed each outcome at the caller as it landed: a slow reducer would make the caller’s mailbox the buffer, and mailboxes are unbounded. Instead the scope holds outcomes and sends one only when the caller has asked for it. The caller replies Next after each outcome, so at most one outcome is ever in flight, and every Delivered message carries the scope’s own inbox so no handshake is needed to bootstrap the conversation.

Back pressure reaches the workers because a slot is occupied from the moment a task is spawned until its outcome has been delivered, not until the task returns. So limit bounds work in flight plus completed-but-unconsumed results together, and a reducer that stops reading stops the run from starting new work. The one place this bound is deliberately relaxed is cancellation, which materialises one NeverStarted per unstarted task in a single burst: those are three-word records and the task list was already a materialised list, so the cost is one that the caller had already paid.

Cancellation, and how `Abandoned` stays honest

A run can be stopped from three directions: on_failure(CancelSiblings) when a task fails, deadline when the wall clock runs out, and cancel_with when some other process decides to end a run the caller is blocked inside. All three converge on begin_cancel.

The scope cancels a worker with process.kill, so a cancelled worker’s exit reason is Killed — and so is the exit reason of a worker some unrelated process killed. Those two are different facts and the module keeps them apart with one boolean: a Killed exit after the scope initiated cancellation is Abandoned; a Killed exit before it is Crashed(Killed), because nobody in this run asked for it.

What the caller may rely on

When start returns, every task has an outcome and no worker from that run is alive. When fold returns — including when the reducer halted it early — the scope has killed and joined everything it spawned before replying, so the same holds. A caller that traps exits does not see the scope’s normal exit as mailbox noise, because the scope drops the link itself just before it returns; the link is live for the whole run, which is the part that matters.

Quick start

import weft

pub fn fetch_all(urls: List(String)) {
  let outcomes =
    urls
    |> list.map(fn(url) { fn() { fetch(url) } })
    |> weft.new
    |> weft.limit(8)
    |> weft.on_failure(weft.KeepGoing)
    |> weft.deadline(30_000)
    |> weft.start

  // Every url is accounted for: bodies for the fetches that succeeded,
  // and one Outcome per fetch that failed, crashed, or ran out of time.
  let #(bodies, rest) = weft.partition(outcomes)
  #(bodies, rest)
}

Types

Cancel

opaque </>

An external stop signal. Any process holding one can end a run that another process is blocked inside, which is the case a handle-based cancel(task) cannot reach: the caller is the thing that is blocked.

A signal is a tiny process that does nothing but stay alive. cancel kills it, and every scope watching it sees the death. That is why one signal can serve several runs at once, and why a signal is one-shot: a killed process cannot come back, so a Cancel that has fired stays fired. Handing a spent signal to cancel_with cancels that run immediately, before any task starts.

pub opaque type Cancel

OnFailure

</>

What a run does to the remaining tasks when one of them fails.

pub type OnFailure {
  KeepGoing
  CancelSiblings
}

Constructors

```
KeepGoing
```
Run everything to completion and report every outcome. The default.
```
CancelSiblings
```
The first Failed or Crashed outcome cancels the rest: running tasks become Abandoned and unstarted ones become NeverStarted.

Outcome

</>

What happened to exactly one task.

Every task in a run produces exactly one of these, so a run is always fully accounted for and never collapses into a single Result that throws away the tasks which did succeed. The index is the task’s position in the list given to new, which is what lets start restore input order and what lets a caller match an outcome back to the work that produced it.

pub type Outcome(a, e) {
  Completed(index: Int, value: a)
  Failed(index: Int, error: e)
  Crashed(index: Int, reason: process.ExitReason)
  Abandoned(index: Int)
  NeverStarted(index: Int)
}

Constructors

```
Completed(index: Int, value: a)
```
The task returned Ok.

Arguments

index

The task’s position in the list given to new, counting from zero.

value

The value the task returned.
```
Failed(index: Int, error: e)
```
The task returned Error. The task itself decided this; nothing went wrong with the run.

Arguments

index

The task’s position in the list given to new, counting from zero.

error

The error the task returned.
```
Crashed(index: Int, reason: process.ExitReason)
```
The task’s process died instead of returning. The Erlang exit reason is carried rather than flattened to a string, so a caller can match on it.

Arguments

index

The task’s position in the list given to new, counting from zero.

reason

Why the worker process exited.
```
Abandoned(index: Int)
```
The task was started, then cancelled before it finished. Work was done and thrown away.

Arguments

index

The task’s position in the list given to new, counting from zero.
```
NeverStarted(index: Int)
```
The run ended before this task ever got a slot. No work was done.

Arguments

index

The task’s position in the list given to new, counting from zero.

Run

opaque </>

A configured run, not yet started.

Built with new and refined with limit, on_failure, cancel_with and deadline; start and fold are the terminal verbs. A Run is an ordinary immutable value, so the same one can be started more than once.

pub opaque type Run(a, e)

Step

</>

A reducer’s verdict on whether to keep consuming outcomes.

This is fold’s answer to the problem a plain fn(acc, outcome) -> acc cannot express: a reducer that has seen enough has no way to say so, and the caller is left reaching for an external cancel signal to make a decision the reducer already made.

pub type Step(acc) {
  Continue(accumulator: acc)
  Halt(accumulator: acc)
}

Constructors

```
Continue(accumulator: acc)
```
Keep going: ask the scope for the next outcome.

Arguments

accumulator

The accumulator to carry into the next outcome.
```
Halt(accumulator: acc)
```
Stop here: the remaining tasks are cancelled and the run is torn down before fold returns.

Arguments

accumulator

The accumulator fold will return.

Values

cancel

</>

pub fn cancel(signal: Cancel) -> Nil

Fire a cancel signal, ending every run watching it.

Firing a signal that has already fired does nothing. Runs watching the signal report Abandoned for the tasks that were running and NeverStarted for the tasks that had not begun.

Examples

let stop = weft.cancel_signal()
process.spawn(fn() { weft.new(tasks) |> weft.cancel_with(stop) |> weft.start })

// From any process, at any time, including while the run's caller is
// blocked inside `start`:
weft.cancel(stop)

cancel_signal

</>

pub fn cancel_signal() -> Cancel

Create a fresh cancel signal.

The returned value can be passed to cancel_with on any number of runs and shared with any number of processes. It costs one idle process, which lives until cancel is called on it.

Examples

let stop = weft.cancel_signal()

let outcomes =
  weft.new(tasks)
  |> weft.cancel_with(stop)
  |> weft.start

cancel_with

</>

pub fn cancel_with(run: Run(a, e), signal: Cancel) -> Run(a, e)

Watch a cancel signal for the duration of the run.

This is the only way to stop a run from outside the process that started it, because that process is blocked inside start or fold and cannot act on its own behalf.

Examples

let stop = weft.cancel_signal()

let outcomes =
  weft.new(tasks)
  |> weft.cancel_with(stop)
  |> weft.start

deadline

</>

pub fn deadline(run: Run(a, e), within: Int) -> Run(a, e)

Give the whole run a wall-clock budget, in milliseconds.

Hitting the deadline is not an error, it is a reason some entries in the account are Abandoned (started, then cut short) or NeverStarted (never got a slot). The timer is cancelled and flushed if the run finishes first, so it can never fire into a later receive.

Examples

let outcomes =
  weft.new(tasks)
  |> weft.deadline(30_000)
  |> weft.start

failures

</>

pub fn failures(outcomes: List(Outcome(a, e))) -> List(e)

The errors of the tasks that returned Error, in the order given.

Tasks that crashed, were abandoned, or never started are not errors the tasks chose to return, so they appear here in no form at all; reach for partition when you need to see them.

Examples

let outcomes = [weft.Failed(0, "a"), weft.Crashed(1, process.Killed)]
assert weft.failures(outcomes) == ["a"]

first_ok

</>

pub fn first_ok(
  tasks: List(fn() -> Result(a, e)),
) -> Result(a, List(Outcome(a, e)))

Run tasks concurrently and return the first one to succeed, falling back to the full account if none did. The losers are killed and joined before a success returns.

A Failed or Crashed task does not end the run here; that is the whole point of the function, and it is why first_ok leaves on_failure at KeepGoing. An empty list is answerable, unlike in race, because the error channel already exists: it yields Error([]).

If something outside the run destroys the scope, the Error carries only the outcomes heard before the death rather than a total account; reach for start, which fills the gap in, when that distinction matters.

Examples

let body = weft.first_ok([fn() { fetch(mirror_a) }, fn() { fetch(mirror_b) }])
// -> Ok(<-body->)

// Nothing succeeded, so the caller gets the whole account in input order.
let answer = weft.first_ok([fn() { Error("a") }, fn() { Error("b") }])
// -> Error([weft.Failed(0, "a"), weft.Failed(1, "b")])

fold

</>

pub fn fold(
  run: Run(a, e),
  from initial: acc,
  with reducer: fn(acc, Outcome(a, e)) -> Step(acc),
) -> acc

Consume outcomes in completion order as they land.

This is how a run whose results do not all fit in memory is processed, and how a caller acts on early results without waiting for the slow tail. The scope holds each outcome until the reducer asks for it, so at most one is ever in flight and a slow reducer stops new work from starting rather than filling a mailbox.

A reducer returning Halt ends the run: the remaining tasks are killed and joined before fold returns, so no worker survives the call. The outcomes of those tasks are not delivered — fold returns an accumulator and has nowhere to put them — so a caller that wants the full account of a run it stopped early should use start with a cancel signal instead.

Examples

// Count successes without holding every value.
let successes =
  weft.new(tasks)
  |> weft.fold(from: 0, with: fn(count, outcome) {
    case outcome {
      weft.Completed(..) -> weft.Continue(count + 1)
      weft.Failed(..) | weft.Crashed(..) -> weft.Continue(count)
      weft.Abandoned(..) | weft.NeverStarted(..) -> weft.Continue(count)
    }
  })

// Stop as soon as three values are in hand; the rest are cancelled.
let first_three =
  weft.new(tasks)
  |> weft.fold(from: [], with: fn(found, outcome) {
    let found = case outcome {
      weft.Completed(value:, ..) -> [value, ..found]
      weft.Failed(..) | weft.Crashed(..) -> found
      weft.Abandoned(..) | weft.NeverStarted(..) -> found
    }
    case list.length(found) >= 3 {
      True -> weft.Halt(found)
      False -> weft.Continue(found)
    }
  })

limit

</>

pub fn limit(run: Run(a, e), max: Int) -> Run(a, e)

Set how many tasks may occupy a slot at once.

A slot is held from the moment a task is spawned until its outcome has been handed to the consumer, so this bounds running work and completed-but- unconsumed results together. A max below one is raised to one; there is no unbounded setting, by design.

Examples

let outcomes =
  weft.new(tasks)
  |> weft.limit(8)
  |> weft.start

map

</>

pub fn map(
  items: List(a),
  limit max: Int,
  with fun: fn(a) -> Result(b, e),
) -> List(Outcome(b, e))

Apply a fallible function to every item, bounded, and return the account.

Exactly weft.new over the items with limit and start, which is the shape most callers want and the one it is easiest to get wrong by mapping an unbounded async over a list.

Examples

let outcomes = weft.map(urls, limit: 8, with: fetch)
let #(bodies, rest) = weft.partition(outcomes)

new

</>

pub fn new(tasks: List(fn() -> Result(a, e))) -> Run(a, e)

Begin a run over tasks, bounded by default.

The default limit is the number of schedulers online, not unbounded: spawning one process per item is fine for pure computation and actively harmful for anything holding a socket, a file handle, a database connection or a rate-limited quota. Someone who wants a wider fan-out says so with limit.

The default failure policy is KeepGoing.

Examples

let outcomes =
  urls
  |> list.map(fn(url) { fn() { fetch(url) } })
  |> weft.new
  |> weft.start

on_failure

</>

pub fn on_failure(run: Run(a, e), policy: OnFailure) -> Run(a, e)

Set what a failing task does to the tasks beside it.

Examples

// Stop at the first `Failed` or `Crashed`; the rest are reported as
// `Abandoned` or `NeverStarted` rather than being lost.
let outcomes =
  weft.new(tasks)
  |> weft.on_failure(weft.CancelSiblings)
  |> weft.start

partition

</>

pub fn partition(
  outcomes: List(Outcome(a, e)),
) -> #(List(a), List(Outcome(a, e)))

Split an account into the values that succeeded and everything else.

Both halves keep the order they were given in. The second half stays a list of Outcome rather than a list of errors, because “did not succeed” covers four different facts and flattening them is how the accounting gets thrown away one call site at a time.

Examples

let outcomes = [weft.Completed(0, "a"), weft.Failed(1, "no")]
assert weft.partition(outcomes) == #(["a"], [weft.Failed(1, "no")])

race

</>

pub fn race(
  first: fn() -> Result(a, e),
  rest: List(fn() -> Result(a, e)),
) -> Outcome(a, e)

Run tasks concurrently and return the first one to complete, however it completed. The losers are killed and joined before this returns.

The first task is a separate argument from the rest because a race over nothing has no answer: Outcome has no variant meaning “there was nothing to race”, and wrapping the result in a Result would push an error case onto every call site that already knows its list is not empty. Making the empty case unrepresentable is cheaper than making every caller handle it. A caller holding a runtime list matches on it once and decides for itself what nothing should mean.

Note the difference from first_ok: this returns the first task to finish, even if it finished by failing or crashing. Conflating the two is a common bug, which is why they are separate functions.

Examples

// Whichever mirror answers first, even if it answers with an error.
let outcome = weft.race(fn() { fetch(mirror_a) }, [fn() { fetch(mirror_b) }])
// -> weft.Completed(index: 1, value: <-body->)

start

</>

pub fn start(run: Run(a, e)) -> List(Outcome(a, e))

Run everything to completion and return every outcome in input order.

Blocks the calling process until the run is over. When it returns, every task has exactly one outcome and no worker from the run is alive.

Ordered versus unordered falls out of which terminal verb you call rather than out of a flag: start sorts the account back into input order, fold hands outcomes over in completion order as they land.

Examples

let outcomes =
  weft.new([fn() { Ok(1) }, fn() { Error("no") }])
  |> weft.start
// -> [weft.Completed(0, 1), weft.Failed(1, "no")]

let #(values, rest) =
  weft.new(tasks)
  |> weft.limit(4)
  |> weft.on_failure(weft.CancelSiblings)
  |> weft.start
  |> weft.partition

values

</>

pub fn values(outcomes: List(Outcome(a, e))) -> List(a)

The values of the tasks that returned Ok, in the order given.

Examples

let outcomes = [weft.Completed(0, 1), weft.Abandoned(1), weft.Completed(2, 3)]
assert weft.values(outcomes) == [1, 3]

Why there is an extra process

Kill then join, in that order

Delivery is pull, not push

Cancellation, and how Abandoned stays honest

What the caller may rely on

Quick start

Constructors

Constructors

Arguments

Arguments

Arguments

Arguments

Arguments

Constructors

Arguments

Arguments

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Cancellation, and how `Abandoned` stays honest