Utilities

Air provides a handful of small, independent utilities that do not depend on the collection types. Most also do not depend on the transaction system, with one exception: a Promise may be delivered inside a transaction, in which case the delivery is staged until the transaction commits (see Promises and transactions).

Lazy values: `Delay` and `@delay`

A Delay{T} computes its value the first time it is requested, and returns that same value on every later request. Unlike a plain closure, a Delay is thread-safe: if several tasks ask for the value at once, the underlying function still runs exactly once.

julia> using Air

julia> d = Delay{Int}(() -> (println("computing..."); 7));

julia> isready(d);      # nothing has been computed yet

julia> d[]
computing...
7

julia> d[]              # the function is not run again
7

@delay builds a Delay from an expression. The expression may be a closure, whose arguments are closed over, or a plain expression, which behaves as () -> expression:

julia> d = @delay 10;

julia> d[]
10

julia> n = 3;

julia> d2 = @delay (n + 1);

julia> n = 100;         # the value was captured, not re-read

julia> d2[]
4

An optional trailing type annotation on the expression sets the element type of the resulting Delay:

julia> d = @delay (2 + 3)::Int;

julia> typeof(d)
Delay{Int64}

Delay objects are used by LazyDict to represent values that have not been computed yet.

Memoized functions: `@memoize`

@memoize declares a function whose results are cached, keyed by the arguments. The function body is evaluated at most once per distinct set of arguments, and never concurrently with itself.

julia> @memoize square(n::Int) = (println("computing..."); n * n)
square (generic function with 1 method)

julia> square(4)
computing...
16

julia> square(4)
16

A trailing type annotation names the type cached in the memoization dictionary:

julia> @memoize half(n::Int) = (n / 2)::Float64;

Arguments are compared by equality, so passing mutable arguments that are later modified leads to undefined behaviour.

One-shot values: `Promise`

A Promise{T} is a container for a single value that is produced at some later point, typically by another task. Reading a Promise blocks until the value is available. When the values form a sequence rather than a single result, use a Stream instead; see Streams.

julia> p = Promise{Int}();

julia> isready(p)
false

julia> Threads.@spawn put!(p, 7);

julia> take(p)
7

A promise settles exactly once. It may be settled successfully with put!, or as failed with fail!; after that, its result cannot change. isready is true after either settlement, isfailed tells the two apart, and geterror reads a failure's payload without throwing.

julia> p = Promise{Int}();

julia> fail!(p, "no value");

julia> isready(p), isfailed(p)
(true, true)

julia> try
           take(p)
       catch e
           sprint(showerror, e)
       end
"promise failed: \"no value\""

julia> geterror(p)
"no value"

A promise may be given an ondelivery and/or an onfailure callback when it is constructed. It is called with the settlement's payload, exactly once, after the settlement has been made; its return value is ignored, and an exception propagates while the promise stays settled. See Promises and transactions for what a settlement inside a transaction means.

julia> p = Promise{Int}(ondelivery = x -> println("got ", x),
                        onfailure = e -> println("failed: ", e));

julia> put!(p, 7);
got 7

julia> take(p)
7

Promises compare by identity: a promise is isequal only to itself, and its hash never waits for a settlement, so an undelivered promise is a perfectly good dictionary key.

Cancellation and timeouts

take can bound its wait with timeout, in seconds. If the promise is still pending when the timeout expires, take throws ReaderTimeout instead of waiting forever; a Stream's take and put! throw the same exception.

julia> p = Promise{Int}();

julia> try
           take(p; timeout = 0.1)
       catch e
           sprint(showerror, e)
       end
"reader timed out after 0.1 seconds"

Every task also has a CancelToken: it is created on first use and is available as canceltoken. Blocking Air calls observe the calling task's token by default, so a task is cancellable without any setup — another task only needs the token, typically recorded when the task is created. cancel! cancels a token and wakes every task waiting under it with a Cancelled exception, carrying the reason given to cancel!. Cancellation is one-way and idempotent; the first reason given is kept.

julia> tok = CancelToken();

julia> cancel!(tok, "shutting down");

julia> try
           take(Promise{Int}(); token = tok)
       catch e
           sprint(showerror, e)
       end
"cancelled: shutting down"

Cancellation wins over settlement: a cancelled token makes take throw even when the promise is already settled, because a task whose work was cancelled should not go on to use the value. To wait regardless of cancellation — in the cleanup of a cancelled task, for example — pass token = nothing. Inside a transaction, a take on a promise that has to wait is a deadlock, since the settlement it waits for cannot happen until the body returns (a Stream may be written by another task while the body waits; see Streams and transactions). A Cancelled thrown inside a transaction body aborts that transaction and is not retried. The same holds for ReaderTimeout.

checkcancel observes cancellation without blocking (iscancelled is its non-throwing form), and setcancaltoken! replaces or clears a task's token. Air's internal Air.sleep_interruptible(seconds; token) is the cancellable counterpart of sleep; it is not exported, because it would shadow Base.sleep for anyone who does using Air.

Cancellation is observed by blocking waits, not everywhere: a transaction that is retrying, and a Delay's [], do not check a token, so a cancelled task should call checkcancel between transactions.

Locking several objects: `lockall`

lockall takes a collection of locks, acquires all of them (in a consistent order to avoid deadlock), runs the given function, and releases them:

julia> r1, r2 = ReentrantLock(), ReentrantLock();

julia> lockall(() -> :done, r1, r2)
:done

API

Air.@delay — Macro

@delay expression

Yields a Delay object that matches the given expression. The expression may be one of the following:

A function of no arguments, such as () -> 10; in this case the delay is made from this function (i.e., the RHS is the -> expression that is delayed).
A function with a set of symbol arguments evaluates the RHS but uses a let statement to wrap all the symbols in the LHS into a closure.
An expression, which is treated as equivalent to () -> expression.

Optionally, the expression or LHS may be tagged with a type T. In this case, a Delay{T} object is yielded instead of a Delay{Any}.

See also: Delay, LazyDict

Examples

DocTestSetup = quote
    using Air
end

julia> # Create a Delay with a long run time.
       d = (@delay (println("Running."); sleep(2); 10)::Int64)
Delay{Int64}(<...>)

julia> # Start a few threads, each of which attempt to read it. The function will
       # only run once.
       for th in [(Threads.@spawn d[]) for _ in 1:5]; wait(th) end
Running.

julia> # Ensure that it produced the correct value and doesn't run again.
       d[]
10

julia> # The display now shows the realized value also.
       d
Delay{Int64}(10)

julia> # Create a Delay with a locally-bound symbol.
       d2 = (@delay (d) -> (d[] / 20.0)::Float64)
Delay{Float64}(<...>)

julia> # We can rebind d without affecting d2.
       d = 10
10

julia> d2[]
0.5

source

Air.@memoize — Macro

@memoize name(args...) = expr
@memoize name(args...) where {...} = expr

@memoize is a macro for declaring that the function declaration that follows should be memoized in a private dictionary and any pre-calculated value should be returned from that dictionary instead of being recalculated. All memoization is thread-safe: expr is only ever evaluated by one thread at a time, and is only ever evaluated once per unique set of arguments.

Note that arguments are memoized according to equality, so the use of mutable arguments can result in undefined behavior of those arguments are later changed.

Examples

DocTestSetup = quote
    using Air
end

julia> @memoize fib(n::Int) = begin
           println("Calculating fib($n)...")
           if n < 1
               return 0
           elseif n == 1
               return 1
           else
               return fib(n-1) + fib(n - 2)
           end
       end::Int
fib (generic function with 1 method)

julia> fib(5)
Calculating fib(5)...
Calculating fib(4)...
Calculating fib(3)...
Calculating fib(2)...
Calculating fib(1)...
Calculating fib(0)...
5

julia> fib(6)
Calculating fib(6)...
8

source

Air.take — Function

take(promise; timeout = nothing, token = canceltoken())

Yields the value delivered to the given Promise object after suspending the current thread to wait for the value if necessary, and throws PromiseFailed if the promise was failed. If the current transaction has staged a settlement for the promise, that settlement is used without waiting; see Promise.

timeout bounds the wait in seconds: if the promise is still pending when it expires, take throws ReaderTimeout. A settled promise is returned or thrown immediately, even with timeout = 0.

token is the CancelToken to observe; the default is the calling task's own token, see canceltoken. If the token has been cancelled, take throws Cancelled even when the promise is already settled, so pass token = nothing to wait regardless — in the cleanup of a cancelled task, for example. Inside a transaction, a take that has to wait is a deadlock: the delivery it is waiting for cannot happen until the body returns. (Waiting on a Stream inside a transaction is not a deadlock in the same way; see its docstring.)

DocTestSetup = quote
    using Air
end

julia> p = Promise()
Promise{Any}(<...>)

julia> put!(p, :done)
:done

julia> p
Promise{Any}(:done)

julia> take(p)
:done

source

take(stream; timeout = nothing, token = canceltoken())

Removes and yields the oldest value of the given Stream, suspending the current task until a value is available if necessary. A closed stream that still holds values yields them, and then throws StreamClosed.

timeout bounds the wait in seconds: if no value is available when it expires, take throws ReaderTimeout. token is the CancelToken to observe, as for take(::Promise); cancellation wins even over an available value. Inside a transaction, take is staged and the value is removed only if the transaction commits; the transaction also sees its own staged writes. A take that has to wait inside a transaction waits normally, but a stream created in that transaction raises StreamDeadlock instead.

Utilities

Lazy values: Delay and @delay

Memoized functions: @memoize

One-shot values: Promise

Cancellation and timeouts

Locking several objects: lockall

API

Lazy values: `Delay` and `@delay`

Memoized functions: `@memoize`

One-shot values: `Promise`

Locking several objects: `lockall`