🥮 mooncakes.io

async

Asynchronous programming library for MoonBit

moon add moonbitlang/async@0.20.5
Download zip
Version
0.20.5
License
Apache-2.0
Repository
Last updated
5 days ago
Downloads
422K
README

#Asynchronous programming library for MoonBit

This library provides basic asynchronous IO functionality for MoonBit, as well as useful asynchronous programming facilities. Currently, this library only supports native/LLVM backends on Linux/MacOS.

API document is available at https://mooncakes.io/docs/moonbitlang/async. You can also find small examples in examples, these examples can be run via moon run -C examples <example-name> in project root.

WARNING: this library is current experimental, API is subject to future change.

#Installation

In your MoonBit project root, run:
moon add moonbitlang/async@0.16.2
This library provides the following packages:

  • moonbitlang/async: most basic asynchronous operations
  • moonbitlang/async/socket: TCP and UDP socket
  • moonbitlang/async/tls: TLS support via OpenSSL
  • moonbitlang/async/stdio: operations on standard output channels
  • moonbitlang/async/pipe: operations on pipes
  • moonbitlang/async/fs: file system operations, such as file IO and directory reading
  • moonbitlang/async/process: spawning system process
  • moonbitlang/async/aqueue: asynchronous queue data structure for inter-task communication
  • moonbitlang/async/semaphore: semaphore for concurrency control
  • moonbitlang/async/cond_var: condition variable with broadcasting support
  • moonbitlang/async/io: generic IO abstraction & utilities, such as buffering
  • moonbitlang/async/http: HTTP client and server support. Features:
    • HTTPS client support
    • CONNECT based HTTP/HTTPS proxy support
  • moonbitlang/async/websocket: WebSocket client and server support. Features:
    • TLS encrypted WebSocket (wss://) support
    • CONNECT based HTTP/HTTPS proxy support
    • native integration with moonbitlang/async/http

To use these packages, add them to the import field of moon.pkg.json.

#Features

#Structured concurrency and error propagation

moonbitlang/async features structured concurrency. In moonbitlang/async, every asynchronous task must be spawned in a task group. Task groups can be created with the with_task_group function

async fn[X] with_task_group(async (TaskGroup[X]) -> X)  -> X

When with_task_group returns, it is guaranteed that all tasks spawned in the group already terminate, leaving no room for orphan task and resource leak.

If any child task in a task group fail with error, all other tasks in the group will be cancelled. So there will be no silently ignored error.

For more behavior detail and useful API, consult the API document.

#Task cancellation

In moonbitlang/async, all asynchronous operations are by default cancellable. So no need to worry about accidentally creating uncancellable task.

In moonbitlang/async, when a task is cancelled, it will receive an error at where it suspended. The cancelled task can then perform cleanup logic using try .. catch. Since most asynchronous operations may throw other error anyway, correct error handling automatically gives correct cancellation handling, so most of the time correct cancellation handling just come for free in moonbitlang/async.

Currently, it is not allowed to perform other asynchronous operation after a task is cancelled. Those operations will be cancelled immediately if current task is already cancelled. Spawn a task in some parent context if asynchronous cleanup is necessary.

#Caveats

Currently, moonbitlang/async features a single-threaded, cooperative multitasking model. With this single-threaded model, code without suspension point can always be considered atomic. So no need for expensive lock and less bug. However, this model also come with some caveats:

  • task scheduling can only happen when current task suspend itself, by performing some asynchronous IO operation or manually calling @async.pause. If you perform heavy computation loop without pausing from time to time, the whole program will be blocked until the loop terminates, and other task will not get executed before that. Similarly, performing blocking IO operation not provided by moonbitlang/async may block the whole program as well

  • in the same way, task cancellation can only happen when a task is in suspended state (blocked by IO operation or manually pause'ed)

  • although internally moonbitlang/async may use OS threads to perform some IO job, user code can only utilize one hardware processor

There are several other points to notice when using this library:

  • At most one task can read/write to socket etc. at anytime, to avoid race condition. If multiple reader/writer is desired, you should create a dedicated worker task for reading/writing and use @async.Queue to distribute/gather data

#
CondVar

source

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

#
Queue

source

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

#
Semaphore

source

#
AlreadyTerminated

source
#deprecated("this error is no longer emitted")
pub suberror AlreadyTerminated

#
TimeoutError

source
pub suberror TimeoutError

#
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::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.

#
RetryMethod

source
pub(all) enum RetryMethod {
  Immediate
  FixedDelay(Int)
  ExponentialDelay(Int, Double, 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. Note that if the task is not spawned with allow_failure=true, the whole task group will fail too.

#
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 current task is cancelled, wait return immediately with error.

#
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) -> Unit

Attach a defer block, represented as a cleanup function, to a task group. The clenaup function will be invoked when the group terminates. 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.

Note that if the whole task group is cancelled, async operations in group defer block will be cancelled immediately too. Users can use protect_from_cancel to prevent async tasks from being cancelled. 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
fn[X] TaskGroup::return_immediately(self : TaskGroup[X], value : X) -> Unit raise

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
fn[G, X] TaskGroup::spawn(self : TaskGroup[G], f : async () -> X, no_wait? : Bool, allow_failure? : Bool) -> 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 whole task group will only exit after this child task terminates.

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.

If the task group is already cancelled or has been terminated, spawn will fail with error and the child task will not be spawned.

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

#
TaskGroup::spawn_bg

source
fn[X] TaskGroup::spawn_bg(self : TaskGroup[X], f : async () -> Unit, no_wait? : Bool, allow_failure? : Bool) -> Unit

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

Unless no_wait (false by default) is true, the whole task group will only exit after this child task terminates.

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.

If the task group is already cancelled or has been terminated, spawn_bg will fail with error and the child task will not be spawned.

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 () -> IterResult, 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 returning IterEnd.

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.

#
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.

#
is_being_cancelled

source
fn is_being_cancelled() -> Bool

#
is_cancellation_error

source
fn is_cancellation_error(error : Error) -> Bool

is_cancellation_error(err) return true if err is the special error used to represent cancellation internally.

Note that is_cancellation_error may not be accurate: async code may fail and raise other error during cancellation handling, in this case the error may be replaced by something else. For accurate detection of cancellation, use is_being_cancelled instead.

#
lazy_init

source
fn[X] lazy_init(f : async () -> X) -> 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.

#
now

source
fn now() -> Int64

Get current time, measured in milliseconds. now uses the same clock as timers in moonbitlang/async.

now can be used to measure elapsed time, such as benchmarking, but the meaning of the absolute time value returned by now is undefined, and should not be depended on.

#
pause

source
async fn pause() -> Unit

#
protect_from_cancel

source
async fn[X] protect_from_cancel(f : async () -> X, resume_on_cancel? : Bool) -> X

#
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.

#
run_async_main

source
fn run_async_main(main : async () -> Unit) -> Unit

This function is used for integration of moonbitlang/async into the MoonBit toolchain, do not call directly.

#
sleep

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

#
with_event_loop

source
#deprecated("use `async fn main` or `async test` instead")
fn with_event_loop(f : async (TaskGroup[Unit]) -> Unit) -> Unit raise

Create a fresh event loop and run a async program inside the loop. A new task group will be created for convenience, that is, with_event_loop(f) will run with_task_group(f) using the event loop.

There can only one event loop running for every program, calling with_event_loop inside another event loop is invalid, and will result in immediate failure.

#
with_task_group

source
async fn[X] with_task_group(f : async (TaskGroup[X]) -> X) -> 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.

Source Files

Powered by MoonBit

Site sourceReport issuePackagesBuild queueSkillsStatistics

© 2026 mooncakes.io