mooncakes.io

CondVar

using @moonbitlang/async/cond_var { type Cond as CondVar }

A condition variable that can be used for synchronization between tasks.

Queue

source

using @moonbitlang/async/aqueue { type Queue }

An asynchronous queue, where reader can wait for data to arrive in a non-blocking manner.

Semaphore

source

using @moonbitlang/async/semaphore { type Semaphore }

BreakFromSpawnLoop

source

pub(all) suberror BreakFromSpawnLoop

A special error used to break away from TaskGroup::spawn_loop.

TaskCancelled

source

pub suberror TaskCancelled derive(@moonbitlang/core/debug.Debug
Debug)

Special error indicating a waited task is cancelled. Note that this error has nothing to do with cancellation of current task.

TimeoutError

source

pub suberror TimeoutError derive(ToJson, @moonbitlang/core/debug.Debug
Debug)

impl Show for TimeoutError

TimerCancelled

source

pub suberror TimerCancelled derive(@moonbitlang/core/debug.Debug
Debug)

impl Show for TimerCancelled

Lazy

source

type Lazy[X]

A lazily initialized value that requires async operation to compute. The async initialization process will start as soon as the result is requested. the result will be cached, so the initialization process will run only once. If no one is waiting for the result anymore before initialization completes, the initialization process will be cancelled automatically.

Lazy::Lazy

source

#callsite(autofill(loc))
#alias(new, deprecated="`new` is deprecated, use `Lazy` instead")
#as_free_fn(lazy_init, deprecated="use `@async.Lazy(..)` instead")
fn[X] Lazy::Lazy(f : async () -> X, loc~ : SourceLoc) -> Lazy[X]

Create a new lazily initialized value by passing an async initialization function. The function f will be started in the background automatically when someone request for the result of the lazy value.

If all waiters are cancelled before f completes, f will be cancelled automatically. After f is cancelled, calling .wait() on the lazy value will start f again.

If f completes normally or due to an error, the result will be cached, and subsequent .wait() on the lazy value always complete immediately.

Lazy::wait

source

async fn[X] Lazy::wait(self : Lazy[X]) -> X

Wait for the result of a lazily initialized value.

If the initialization of the value has not yet started, it will be started automatically.

If the initialization of the value has already completed, the cached result will be returned automatically.

Mutex

source

type Mutex

A mutex for synchronization between multiple tasks. Essentially a mutex is just a semaphore of size one.

Mutex::Mutex

source

fn Mutex::Mutex() -> Mutex

Create a new mutex. The mutex is initially in released state.

Mutex::acquire

source

async fn Mutex::acquire(self : Mutex) -> Unit noraise

Acquire the mutex.

If the mutex is already acquired by someone else, acquire will wait until the mutex is released.

If several tasks are both waiting for the mutex, they will acquire the mutex in a first-come-first-serve manner

acquire is cancellation safe. When acquire is cancelled, it will never trigger any side effect. In particular, a cancelled acquire request will never actually acquire the mutex.

Mutex::release

source

fn Mutex::release(self : Mutex) -> Unit

Release the mutex. The mutex must have been acquired, otherwise the program will abort.

Mutex::try_acquire

source

fn Mutex::try_acquire(self : Mutex) -> Bool

Try to acquire the mutex, but do not wait for it. If the mutex is currently available, it will be acquired and true will be returned. If the mutex is currently not available, false will be returned.

try_acquire follow the same FIFO rule of acquire: if there are acquire call waiting for the mutex, try_acquire always return false.

RetryMethod

source

pub(all) enum RetryMethod {
  Immediate
  FixedDelay(Int)
  ExponentialDelay(initial~ : Int, factor~ : Double, maximum~ : Int)
}

RetryMethod describes different retry strategies used by various APIs:

Immediate: failed task will immediately be restarted.
FixedDelay(t), failed task will be restarted after sleeping for t milliseconds
ExponentialDelay(initial~, factor~, maximum~): failed task will be restarted with an exponentially growing delay. The initial delay is initial, and after every failure, the delay will be multiplied by factor, but will never exceed maximum.

Task

source

type Task[X]

Task[X] represents a running task with result type X, it can be used to wait and retrieve the result value of the task.

Task::cancel

source

fn[X] Task::cancel(self : Task[X]) -> Unit

Cancel a task. Subsequent attempt to wait for the task will receive error.

Task::try_wait

source

fn[X] Task::try_wait(self : Task[X]) -> X? raise

Try to obtain the result of the task. If the task already terminated, its result value will be returned. If the task already failed, try_wait will fail immediately. If the task is still running, try_wait returns None. try_wait is a synchoronous function: it never blocks.

Task::wait

source

async fn[X] Task::wait(self : Task[X]) -> X

Wait for a task and retrieve its result value. If the task fails, wait will also fail with the same error. If the target task terminated due to cancellation, wait will raise TaskCancelled.

TaskGroup

source

type TaskGroup[X]

A TaskGroup can be used to spawn children tasks that run in parallel. Task groups implements structured concurrency: a task group will only return after all its children task terminates.

Task groups also handles error propagation: by default, if any child task raises error, the whole task group will also raise that error, and all other remaining child tasks will be cancelled.

The type parameter X in TaskGroup[X] is the result type of the group, see with_task_group for more detail.

TaskGroup::add_defer

source

fn[X] TaskGroup::add_defer(self : TaskGroup[X], block : async () -> Unit nocancel) -> Unit

Attach a defer block, represented as a cleanup function, to a task group. The clenaup function will be invoked when the group terminates, so all child task of the group (created via spawn or spawn_bg) must have terminated when any group defer code is invoked. Calling spawn or spawn_bg inside group defer is not allowed, and will abort the program immediately.

Group scoped defer blocks are executed in FILO order, just like normal defer. with_task_group will only exit after all group defer blocks terminate.

Group scoped defer must be nocancel. If cancellable code have to be run anyway, use protect_from_cancel. It is highly recommended to add a hard timeout to async defer block in this case, to avoid infinite hanging due to blocked operation.

TaskGroup::return_immediately

source

async fn[X] TaskGroup::return_immediately(self : TaskGroup[X], value : X) -> Unit noraise

Force a task group to terminate immediately with the given result value. All child tasks in the group, including potentially the current one, will be cancelled.

TaskGroup::spawn

source

#callsite(autofill(loc))
fn[G, X] TaskGroup::spawn(self : TaskGroup[G], f : async () -> X, no_wait? : Bool, allow_failure? : Bool, loc~ : SourceLoc) -> Task[X]

Spawn a child task in a task group, compute a result asynchronously. A task handle will be returned, the result value of the task can be waited and retrieved using .wait(), or cancelled using .cancel().

Unless no_wait (false by default) is true, the task group will wait for this task to complete before normal exit. No matter what the value of no_wait is, with_task_group will only return after all child tasks terminate. The task will be cancelled automatically if it is still running when the task group wishes to terminate,

Unless allow_failure (false by default) is true, Ithe whole task group will also fail if the spawned task fails, other tasks in the group will be cancelled in this case.

Cancelled task (either cancelled by the group or manually cancelled via Task::cancel) are not consider failing, although they raise a cancellation error. However, if the task raise some other error during cancellation handling, it will be considered failing.

If the task group is already cancelled, but with_task_group is still running (i.e. there are still other running child task in the group), spawn will still succeed, but the child task will be cancelled immediately. In this case, the child task still get a chance to start execution (which means you can rely on defer at the start of child task being triggered), but will be in cancelled state from the very beginning (e.g. all async IO operation will be cancelled immediately before starting).

Calling spawn after with_task_group completed or inside group.add_defer is not allowed, and will abort the program immediately.

It is undefined whether the child task will start running immediately before spawn returns.

TaskGroup::spawn_bg

source

#callsite(autofill(loc))
fn[X] TaskGroup::spawn_bg(self : TaskGroup[X], f : async () -> Unit, no_wait? : Bool, allow_failure? : Bool, loc~ : SourceLoc) -> Unit

Spawn a child task in a task group, and run it asynchronously in the background.

Unless allow_failure (false by default) is true, the whole task group will also fail if the spawned task fails, other tasks in the group will be cancelled in this case.

If the task group is already cancelled, but with_task_group is still running (i.e. there are still other running child task in the group), spawn_bg will still succeed, but the child task will be cancelled immediately. In this case, the child task still get a chance to start execution (which means you can rely on defer at the start of child task being triggered), but will be in cancelled state from the very beginning (e.g. all async IO operation will be cancelled immediately before starting).

Calling spawn_bg after with_task_group completed or inside group.add_defer is not allowed, and will abort the program immediately.

It is undefined whether the child task will start running immediately before spawn_bg returns.

TaskGroup::spawn_loop

source

fn[X] TaskGroup::spawn_loop(self : TaskGroup[X], f : async () -> Unit, no_wait? : Bool, allow_failure? : Bool, retry? : RetryMethod, max_retry? : Int, fatal_error? : (Error) -> Bool) -> Unit

Similar to spawn_bg, but the spawn a loop, i.e. the spawned task will be restarted after it terminates. The spawned task can terminate the loop by raising the special error BreakFromSpawnLoop, the loop will terminate normally instead of failing in this case.

If retry_method is set, within each loop, the spawned task will be automatically restarted on failure. The restart strategy is determined by the value of retry_method. max_retry and fatal_error will be passed as-is to retry. See retry and RetryMethod for more details.

The meaning of no_wait and allow_failure is the same as spawn_bg.

Timer

source

pub struct Timer {
  duration : Int
  // private fields
}

A timer that expires after a fixed duration with support for multiple waiters and refreshing. The timer will expire only once.

The timer object itself has no side effect. The only way to observe the existence of a timer is by waiting for it. So if no one is waiting for a timer, the timer will not block program termination even if it is still active.

Timer::Timer

source

#alias(new, deprecated="`new` is deprecated, use `Timer` instead")
fn Timer::Timer(duration : Int) -> Timer

Create a new timer that expires after duration milliseconds. The timer will start immediately, and expire after the duration elapsed.

Timer::cancel

source

fn Timer::cancel(timer : Timer, err? : Error) -> Unit

Manually cancel a timer. All waiters on the timer will fail immediately with err (TimerCancelled by default). All subsequent waiters will also fail with err immediately.

Timer::refresh

source

fn Timer::refresh(timer : Timer) -> Unit

Refresh a timer. If the timer has already terminated, it is restarted immediately, and will be triggered again after the timer expires. If the timer is currently active, its expiration time will be delayed to current time + duration of the timer. If the timer has been cancelled, refresh has no effect.

Timer::wait

source

async fn Timer::wait(timer : Timer) -> Unit

Wait for the expiration of a timer. If the timer has already terminated, wait will return immediately. If the timer has already been cancelled, or if the timer is cancelled during the wait, wait will fail immediately.

all

source

async fn[X] all(tasks : ArrayView[async () -> X], max_concurrent? : Int) -> Array[X]

all(tasks, max_concurrent?) waits for all tasks to complete and returns their results.

All tasks are spawned and executed (optionally with max_concurrent limit).
If all tasks succeed, all returns an array of results in the same order as tasks.
If any task fails, all fails with that error and other running tasks are cancelled.

The max_concurrent parameter limits the number of tasks that can run concurrently. If not specified, all tasks run at once.

any

source

#callsite(autofill(loc))
async fn[X] any(tasks : ArrayView[async () -> X], max_concurrent? : Int, allow_failure? : Bool, loc~ : SourceLoc) -> X

any(tasks, max_concurrent?, allow_failure?) waits for the first task to complete successfully.

All tasks are spawned and executed (optionally with max_concurrent limit).
When the first task completes successfully, any returns its result immediately and all other running tasks are cancelled.
If allow_failure is false (default), any fails as soon as any task fails.
If allow_failure is true, failing tasks are ignored and any waits for the first success.
If all tasks fail, any fails with the error from a random task.

The max_concurrent parameter limits the number of tasks that can run concurrently. If not specified, all tasks run at once.

any requires at least one task and will fail if the array is empty.

handle_cancellation

source

async fn[X] handle_cancellation(f : async () -> X) -> X?

is_being_cancelled

source

fn is_being_cancelled() -> Bool

is_cancellation_error

source

#deprecated("`catch` can no longer capture cancellation, and cancellation is no longer represented as an error, so `is_cancellation_error` is now meaningless. Use `with_cancellation_handler` or `is_being_cancelled` instead.")
fn is_cancellation_error(_ : Error) -> Bool

now

source

fn now() -> Int64

pause

source

async fn pause() -> Unit noraise

protect_from_cancel

source

async fn[X] protect_from_cancel(f : async () -> X) -> X nocancel

retry

source

async fn[X] retry(strategy : RetryMethod, max_retry? : Int, fatal_error? : (Error) -> Bool, f : async () -> X) -> X

retry(strategy, f) will keep retrying some async operation until success. If f returns value normally, retry will immediately return value. If f fails with error, retry will restart f again. The restart strategy is determined by strategy, see the RetryMethod type for more details.

If current task is cancelled, retry will be cancelled too.

By default, the number of retry attempt is unbounded. However, if max_retry is set, the number of retry attempts will not exceed the value of max_retry. If retry attempts reaches limit, retry will raise the error from f's last attempt.

If fatal_error is present, and f raises an error err such that fatal_error(err) is true, retry will fail immediately with err.

set_deadlock_handler

source

fn set_deadlock_handler(f : (Array[SourceLoc]) -> Unit) -> Unit

set_external_event_loop

source

fn set_external_event_loop(extloop : &@moonbitlang/async/external_loop_integration.ExternalEventLoop
ExternalEventLoop) -> Unit

sleep

source

async fn sleep(duration : Int) -> Unit noraise

with_cancellation_handler

source

#deprecated("use `handle_cancellation` instead")
async fn[X] with_cancellation_handler(f : async () -> X, handler~ : async () -> Unit) -> X

with_cancellation_handler(f, handler~) runs the asynchronous function f. If f is cancelled and does not produce a result, handler will be invoked. If f succeed or fail with a fatal error, the result of f will be returned as-is.

with_task_group

source

#callsite(autofill(loc))
async fn[X] with_task_group(f : async (TaskGroup[X]) -> X, loc~ : SourceLoc) -> X

with_task_group(f) creates a new task group and run f with the new group. f itself will be run in a child task of the new group. with_task_group exits after all the whole group terminates, which means all child tasks in the group have terminated, including f.

If all children task terminate successfully, with_task_group will return the result of f.

with_timeout

source

async fn[X] with_timeout(time : Int, f : async () -> X, error? : Error) -> X

with_timeout(timeout, f) run the async function f.

If f return value before timeout, with_timeout will return value immediately.
If f fail before timeout, with_timeout will fail immediately.
If f is still running after timeout milliseconds, with_timeout will fail with TimeoutError, or the value of error, if explicitly specified.

with_timeout_opt

source

async fn[X] with_timeout_opt(time : Int, f : async () -> X) -> X?

with_timeout_opt(timeout, f) run the async function f.

If f return value before timeout, with_timeout_opt will return Some(value) immediately.
If f fail before timeout, with_timeout_opt will fail immediately.
If f is still running after timeout milliseconds, with_timeout_opt will return None immediately, and f will be cancelled.

async

#Asynchronous programming library for MoonBit

#Installation

#Features

#Structured concurrency and error propagation

#Task cancellation

#Caveats

#CondVar

#Queue

#Semaphore

#BreakFromSpawnLoop

#TaskCancelled

#TimeoutError

#TimerCancelled

#Lazy

#Lazy::Lazy

#Lazy::wait

#Mutex

#Mutex::Mutex

#Mutex::acquire

#Mutex::release

#Mutex::try_acquire

#RetryMethod

#Task

#Task::cancel

#Task::try_wait

#Task::wait

#TaskGroup

#TaskGroup::add_defer

#TaskGroup::return_immediately

#TaskGroup::spawn

#TaskGroup::spawn_bg

#TaskGroup::spawn_loop

#Timer

#Timer::Timer

#Timer::cancel

#Timer::refresh

#Timer::wait

#all

#any

#handle_cancellation

#is_being_cancelled

#is_cancellation_error

#now

#pause

#protect_from_cancel

#retry

#set_deadlock_handler

#set_external_event_loop

#sleep

#with_cancellation_handler

#with_task_group

#with_timeout

#with_timeout_opt

Source Files

CondVar

Queue

Semaphore

BreakFromSpawnLoop

TaskCancelled

TimeoutError

TimerCancelled

Lazy

Lazy::Lazy

Lazy::wait

Mutex

Mutex::Mutex

Mutex::acquire

Mutex::release

Mutex::try_acquire

RetryMethod

Task

Task::cancel

Task::try_wait

Task::wait

TaskGroup

TaskGroup::add_defer

TaskGroup::return_immediately

TaskGroup::spawn

TaskGroup::spawn_bg

TaskGroup::spawn_loop

Timer

Timer::Timer

Timer::cancel

Timer::refresh

Timer::wait

all

any

handle_cancellation

is_being_cancelled

is_cancellation_error

now

pause

protect_from_cancel

retry

set_deadlock_handler

set_external_event_loop

sleep

with_cancellation_handler

with_task_group

with_timeout

with_timeout_opt