strand
Concurrency primitives.
Every strand has a current input, which it reads values from, and a current
output, which it writes values to. A new strand starts with its creator's.
Functions here that read or write without being given a source or
destination, such as next and put, use
them. pipeline connects each stage's output to the next
stage's input, and redirect replaces them for the
duration of a call. In the shell, a program's input and output start as its
standard input and output.
Types
| Type | Description |
|---|---|
Key[T] |
Stores a strand-local value. |
Receiver[T] |
Receives values from a channel. As an Iter, it can be used
with for or next(). |
Resource |
Limits concurrent entry into application-defined scopes. |
Sender[T] |
Sends values into a channel. As a Sink, it accepts put. |
Strand[R] |
Handles a background strand created by spawn. |
Stream[R, I, O] |
A Strand with channel-backed input and output, created by
stream. |
PipeReceiver[I] |
The receiving end of a pipe between strands, as a
stream strand's input. In the shell, this is
proc.PipeReceiver; otherwise it is a
Receiver. |
PipeSender[I] |
The sending end of a pipe between strands, as a
stream strand's output. In the shell, this is
proc.PipeSender; otherwise it is a
Sender. |
PipeReceiver[I]
The receiving end of a pipe between strands, as a
stream strand's input. In the shell, this is
proc.PipeReceiver; otherwise it is a
Receiver.
PipeSender[I]
The sending end of a pipe between strands, as a
stream strand's output. In the shell, this is
proc.PipeSender; otherwise it is a
Sender.
Functions
backtrace() -> inspect.Backtrace
Captures the current strand's calls, as an
inspect.Backtrace.
channel[T] … -> Tuple[Sender[T], Receiver[T]]
Creates a channel for communication between strands.
Parameters
| Name | Type | Description |
|---|---|---|
buffer? |
Int |
Number of values the channel holds before a send waits; 1 when omitted. |
collect[I]() -> Array[I] overload 1 of 2
Collects the strand-local input into a new array.
Parameters
| Name | Type | Description |
|---|---|---|
< |
Iter[I] |
collect[I, S @ Sinkable[I]] target -> S overload 2 of 2
Collects the strand-local input into a sink, returning the sink.
Parameters
| Name | Type | Description |
|---|---|---|
target |
S |
|
< |
Iter[I] |
each[I, O] func
Transforms each value of the strand-local input and writes the result to the strand-local output.
Parameters
| Name | Type | Description |
|---|---|---|
func |
((I, <Value, >Value) -> O) |
Function that transforms one value. |
< |
Iter[I] |
|
> |
Sink[O] |
error_backtrace() -> inspect.Backtrace
Captures the calls of the exception being handled, followed by the current strand's calls.
Errors
std.StateError if no exception is being handled, in a catch or finally
block.
fork[*Rs] *blocks -> Tuple[...Rs]
Runs blocks concurrently and returns their results in block order.
Parameters
| Name | Type | Description |
|---|---|---|
*blocks |
...(() -> Rs) |
Functions to run concurrently. |
from[O] value
Writes the values of an iterable to the strand-local output.
Parameters
| Name | Type | Description |
|---|---|---|
value |
Iterable[O] |
Values to write. |
> |
Sink[O] |
input[I]() -> I
Returns the strand-local input.
Parameters
| Name | Type | Description |
|---|---|---|
< |
I |
map[I, O] count func -> nil overload 1 of 4
Applies a function concurrently to values from the strand-local input,
writing results to the strand-local output.
The function sees std.null as input and output.
Parameters
| Name | Type | Description |
|---|---|---|
count |
Int |
Number of worker strands. |
func |
((I, <Value, >Value) -> O) |
Function applied to each input value. |
< |
Iter[I] |
|
> |
Sink[O] |
map[I, O] count func :input -> nil overload 2 of 4
Applies a function concurrently to values pulled from an iterable, writing results to the strand-local output.
Parameters
| Name | Type | Description |
|---|---|---|
count |
Int |
Number of worker strands. |
func |
((I, >Value) -> O) |
Function applied to each input value. |
:input |
Iterable[I] |
Values to process. |
> |
Sink[O] |
map[I, O] count func :output -> nil overload 3 of 4
Applies a function concurrently to values from the strand-local input, writing results to a sink.
Parameters
| Name | Type | Description |
|---|---|---|
count |
Int |
Number of worker strands. |
func |
((I, <Value) -> O) |
Function applied to each input value. |
:output |
Sinkable[O] |
Sink to write results to. |
< |
Iter[I] |
map[I, O] count func :input … -> nil overload 4 of 4
Applies a function concurrently to values pulled from an iterable, writing results to a sink.
Parameters
| Name | Type | Description |
|---|---|---|
count |
Int |
Number of worker strands. |
func |
((I) -> O) |
Function applied to each input value. |
:input |
Iterable[I] |
Values to process. |
:output |
Sinkable[O] |
Sink to write results to. |
next[I]() -> I overload 1 of 3
Reads the next value of the strand-local input.
Parameters
| Name | Type | Description |
|---|---|---|
< |
Iter[I] |
Errors
std.IterStop if the input is exhausted.
next[I, D] :default -> (I | D) overload 2 of 3
Reads the next value of the strand-local input, or returns default if the
input is exhausted.
Parameters
| Name | Type | Description |
|---|---|---|
:default |
D |
|
< |
Iter[I] |
next[I, E] :else -> (I | E) overload 3 of 3
Reads the next value of the strand-local input, or returns the result of
calling else if the input is exhausted.
Parameters
| Name | Type | Description |
|---|---|---|
:else |
(() -> E) |
|
< |
Iter[I] |
output[O]() -> O
Returns the strand-local output.
Parameters
| Name | Type | Description |
|---|---|---|
> |
O |
pipeline[R] ...stages final … -> R overload 1 of 2
Connects pipeline stages and runs them concurrently, returning the last stage's result.
Parameters
| Name | Type | Description |
|---|---|---|
:input? |
Iterable[Value] |
Input for the first stage, in place of the strand-local input. |
...stages |
((<Iter[std.Unknown], >Sink[std.Unknown]) -> Value) |
Pipeline stages, each reading the previous stage's output. |
final |
((<Iter[std.Unknown]) -> R) |
Final stage. |
pipeline[R, O] :output ...stages final … -> R overload 2 of 2
Connects pipeline stages and runs them concurrently, writing the last stage's output to a sink and returning its result.
Parameters
| Name | Type | Description |
|---|---|---|
:input? |
Iterable[Value] |
Input for the first stage, in place of the strand-local input. |
:output |
Sinkable[O] |
Output for the last stage, in place of the strand-local output. |
...stages |
((<Iter[std.Unknown], >Sink[std.Unknown]) -> Value) |
Pipeline stages, each reading the previous stage's output. |
final |
((<Iter[std.Unknown], >Sink[O]) -> R) |
Final stage. |
pool[T] count input func -> nil overload 1 of 2
Runs a function over the items of an iterable with a fixed number of scoped workers.
Parameters
| Name | Type | Description |
|---|---|---|
count |
Int |
Number of worker strands. |
input |
Iterable[T] |
Items to process. |
func |
((T) -> nil) |
Function applied to each input value. |
pool[T] count func -> nil overload 2 of 2
Runs a function over the strand-local input with a fixed number of scoped workers.
Parameters
| Name | Type | Description |
|---|---|---|
count |
Int |
Number of worker strands. |
func |
((T, <Value) -> nil) |
Function applied to each input value. |
< |
Iter[T] |
put[T] value
Writes a value to the strand-local output.
Parameters
| Name | Type | Description |
|---|---|---|
value |
T |
|
> |
Sink[T] |
redirect[R, I, O] :input :output block -> R overload 1 of 4
Calls a function with the strand-local input and output replaced, returning its result.
The previous input and output are restored when block returns or raises.
Parameters
| Name | Type | Description |
|---|---|---|
:input |
Iterable[I] |
Replacement for the strand-local input. |
:output |
Sinkable[O] |
Replacement for the strand-local output. |
block |
((<Iter[I], >Sink[O]) -> R) |
Function to call. |
Example
let doubled = []
strand.redirect input: [1, 2, 3] output: $doubled do
strand.each do |x| (x * 2)
assert_eq $doubled [2, 4, 6]
redirect[R, I] :input block -> R overload 2 of 4
Calls a function with the strand-local input replaced, returning its result.
Parameters
| Name | Type | Description |
|---|---|---|
:input |
Iterable[I] |
Replacement for the strand-local input. |
block |
((<Iter[I]) -> R) |
Function to call. |
redirect[R, O] :output block -> R overload 3 of 4
Calls a function with the strand-local output replaced, returning its result.
Parameters
| Name | Type | Description |
|---|---|---|
:output |
Sinkable[O] |
Replacement for the strand-local output. |
block |
((>Sink[O]) -> R) |
Function to call. |
redirect[R] block -> R overload 4 of 4
Calls a function, returning its result.
Parameters
| Name | Type | Description |
|---|---|---|
block |
(() -> R) |
Function to call. |
spawn[R] func -> Strand[R]
Runs a function in a background strand.
Parameters
| Name | Type | Description |
|---|---|---|
func |
(() -> R) |
Function to run. |
stream[R, I = Value, O = Value] func -> Stream[R, O, I]
Runs a function in a background strand with channel-backed endpoints.
Parameters
| Name | Type | Description |
|---|---|---|
func |
((<PipeReceiver[I], >PipeSender[O]) -> R) |
Function to run. |
throw value … -> Never
Raises a value as an exception.
Parameters
| Name | Type | Description |
|---|---|---|
value |
Value |
Value to raise. |
:backtrace? |
inspect.Backtrace |
Backtrace to report instead of the current strand's calls. |
where[T] predicate
Writes the values of the strand-local input that satisfy a predicate to the strand-local output.
Parameters
| Name | Type | Description |
|---|---|---|
predicate |
((T, <Value, >Value) -> Bool) |
|
< |
Iter[T] |
|
> |
Sink[T] |
with_io[R, I, O] :input :output block -> R overload 1 of 4
Calls a function with the strand-local input and output set to the given values as they are.
Unlike redirect, this does not convert the values with
(iter) or (sink), so it can restore an input or output saved with
input or output exactly.
The previous input and output are restored when block returns or raises.
Omitted channels are unchanged.
Parameters
| Name | Type | Description |
|---|---|---|
:input |
I |
Replacement for the strand-local input. |
:output |
O |
Replacement for the strand-local output. |
block |
((<I, >O) -> R) |
Function to call. |
Example
def with_saved_input block
let saved_input = input()
redirect input: ["ready"] do
next()
with_io input: $saved_input do block()
with_io[R, I] :input block -> R overload 2 of 4
Calls a function with the strand-local input set to the given value as it is.
Parameters
| Name | Type | Description |
|---|---|---|
:input |
I |
|
block |
((<I) -> R) |
with_io[R, O] :output block -> R overload 3 of 4
Calls a function with the strand-local output set to the given value as it is.
Parameters
| Name | Type | Description |
|---|---|---|
:output |
O |
|
block |
((>O) -> R) |
with_io[R] block -> R overload 4 of 4
Calls a function, returning its result.
Parameters
| Name | Type | Description |
|---|---|---|
block |
(() -> R) |