moonbit: async runtime redesign and coverage expansion

crates/moonbit/Cargo.toml

@@ -11,6 +11,7 @@ description = """
 MoonBit bindings generator for WIT and the component model, typically used
 through the `wit-bindgen-cli` crate.
 """
+readme = "README.md"
 
 [lints]
 workspace = true

crates/moonbit/README.md

@@ -0,0 +1,68 @@
+# `wit-bindgen-moonbit`
+
+MoonBit language bindings generator for WIT and the Component Model.
+
+## Usage
+
+Generate bindings via the `moonbit` subcommand:
+
+```bash
+wit-bindgen moonbit [OPTIONS] <WIT>
+```
+
+See `wit-bindgen help moonbit` for available options.
+
+## Local async usage
+
+For pure MoonBit code (no FFI), you can create local future/stream pairs.
+
+Future + Promise:
+
+```mbt
+let (f, p) = @async.Future::new[Int]()
+@async.spawn(async fn() { p.write(42) })
+let value = f.get()
+```
+
+Stream + Sink (batched reads/writes):
+
+```mbt
+let (s, sink) = @async.Stream::new[Byte]()
+@async.spawn(async fn() {
+  let chunk : Array[Byte] = [1, 2, 3, 4]
+  let _ = sink.write(chunk[:])
+  sink.close()
+})
+let chunk = s.read(4096)
+match chunk {
+  None => ()
+  Some(bytes) => {
+    let _ = bytes.length()
+  }
+}
+```
+
+`Stream::read(count)` returns up to `count` elements; `Sink::write` accepts
+`ArrayView[T]` so byte streams can batch data efficiently. `Stream::new`
+accepts an optional `capacity` (<= 0 means unbounded).
+
+## Testing
+
+From the repo root, run the MoonBit codegen tests:
+
+```bash
+cargo run test \
+  --languages rust,moonbit \
+  --artifacts target/artifacts \
+  --rust-wit-bindgen-path ./crates/guest-rust \
+  tests/codegen
+```
+
+And the async runtime tests (requires an async component-model runner):
+
+```bash
+cargo run test --languages rust,moonbit tests/runtime-async \
+  --artifacts target/artifacts \
+  --rust-wit-bindgen-path ./crates/guest-rust \
+  --runner "wasmtime -W component-model-async"
+```

crates/moonbit/src/async/async_abi.mbt

@@ -0,0 +1,266 @@
+// #region subtask
+
+///|
+priv struct SubTask {
+  handle : Int
+  state : SubTaskState
+}
+
+///|
+priv enum SubTaskState {
+  Starting = 0
+  Started = 1
+  Returned = 2
+  Cancelled_before_started = 3
+  Cancelled_before_returned = 4
+}
+
+///|
+fn SubTaskState::from(int : Int) -> SubTaskState {
+  match int {
+    0 => Starting
+    1 => Started
+    2 => Returned
+    3 => Cancelled_before_started
+    4 => Cancelled_before_returned
+    _ => panic()
+  }
+}
+
+///|
+fn SubTask::from(code : Int) -> SubTask {
+  { handle: code >> 4, state: SubTaskState::from(code & 0xf) }
+}
+
+///|
+/// None : the subtask is blocked
+fn SubTask::cancel(self : SubTask) -> SubTaskState? {
+  let result = subtask_cancel(self.handle)
+  if result == -1 {
+    None
+  } else {
+    Some(SubTaskState::from(subtask_cancel(self.handle)))
+  }
+}
+
+// #endregion
+
+// #region events
+
+///|
+priv enum EventCode {
+  None = 0
+  SubTask = 1
+  StreamRead = 2
+  StreamWrite = 3
+  FutureRead = 4
+  FutureWrite = 5
+  TaskCancelled = 6
+}
+
+///|
+fn EventCode::from(int : Int) -> EventCode {
+  match int {
+    0 => EventCode::None
+    1 => EventCode::SubTask
+    2 => EventCode::StreamRead
+    3 => EventCode::StreamWrite
+    4 => EventCode::FutureRead
+    5 => EventCode::FutureWrite
+    6 => EventCode::TaskCancelled
+    _ => panic()
+  }
+}
+
+///|
+priv enum Events {
+  None
+  Subtask(Int, SubTaskState)
+  StreamRead(Int, StreamResult)
+  StreamWrite(Int, StreamResult)
+  FutureRead(Int, FutureReadResult)
+  FutureWrite(Int, FutureWriteResult)
+  TaskCancelled
+}
+
+///|
+fn Events::new(code : EventCode, i : Int, j : Int) -> Events {
+  match code {
+    None => None
+    SubTask => Subtask(i, SubTaskState::from(j))
+    StreamRead => StreamRead(i, StreamResult::from(j))
+    StreamWrite => StreamWrite(i, StreamResult::from(j))
+    FutureRead => FutureRead(i, FutureReadResult::from(j))
+    FutureWrite => FutureWrite(i, FutureWriteResult::from(j))
+    TaskCancelled => TaskCancelled
+  }
+}
+
+// #endregion
+
+// #region waitable set
+
+///|
+struct WaitableSet(Int) derive(Eq, Show, Hash)
+
+///|
+fn WaitableSet::new() -> WaitableSet {
+  WaitableSet(waitable_set_new())
+}
+
+///|
+fn WaitableSet::drop(self : Self) -> Unit {
+  waitable_set_drop(self.0)
+}
+
+// #endregion
+
+// #region Future
+
+///|
+priv enum FutureReadResult {
+  Completed = 0
+  Cancelled = 2
+}
+
+///|
+fn FutureReadResult::from(int : Int) -> FutureReadResult {
+  match int {
+    0 => Completed
+    2 => Cancelled
+    _ => panic()
+  }
+}
+
+///|
+priv enum FutureWriteResult {
+  Completed = 0
+  Dropped = 1
+  Cancelled = 2
+}
+
+///|
+fn FutureWriteResult::from(int : Int) -> FutureWriteResult {
+  match int {
+    0 => Completed
+    1 => Dropped
+    2 => Cancelled
+    _ => panic()
+  }
+}
+
+// #endregion
+
+// #region Stream
+
+///|
+priv struct StreamResult {
+  progress : Int
+  copy_result : CopyResult
+}
+
+///|
+fn StreamResult::from(int : Int) -> StreamResult {
+  let progress = int >> 4
+  let copy_result = CopyResult::from(int & 0xf)
+  { progress, copy_result }
+}
+
+///|
+priv enum CopyResult {
+  Completed = 0
+  Dropped = 1
+  Cancelled = 2
+}
+
+///|
+fn CopyResult::from(int : Int) -> CopyResult {
+  match int {
+    0 => Completed
+    1 => Dropped
+    2 => Cancelled
+    _ => panic()
+  }
+}
+
+// #endregion
+
+// #region callback code
+
+///|
+/// Code to let the runtime know what to do in the callback
+priv enum CallbackCode {
+  Completed
+  Yield
+  Wait(WaitableSet)
+  Poll(WaitableSet)
+}
+
+///|
+fn CallbackCode::encode(self : Self) -> Int {
+  match self {
+    Completed => 0
+    Yield => 1
+    Wait(id) => 2 | (id.0 << 4)
+    Poll(id) => 3 | (id.0 << 4)
+  }
+}
+
+///|
+fn CallbackCode::_decode(int : Int) -> CallbackCode {
+  let id = int >> 4
+  match int & 0xf {
+    0 => Completed
+    1 => Yield
+    2 => Wait(id)
+    3 => Poll(id)
+    _ => panic()
+  }
+}
+
+// #endregion
+
+// #region Component async primitives
+
+///|
+/// Return whether is cancelled.
+/// Use for non-callback implementation.
+fn _yield() -> Bool = "$root" "[cancellable][yield]"
+
+///|
+pub fn backpressure_inc() = "$root" "[backpressure-inc]"
+
+///|
+pub fn backpressure_dec() = "$root" "[backpressure-dec]"
+
+///|
+fn subtask_cancel(id : Int) -> Int = "$root" "[subtask-cancel]"
+
+///|
+fn subtask_drop(id : Int) = "$root" "[subtask-drop]"
+
+///|
+pub fn context_set(task : Int) = "$root" "[context-set-0]"
+
+///|
+pub fn context_get() -> Int = "$root" "[context-get-0]"
+
+///|
+fn tls_set(tls : Int) = "$root" "[context-set-0]"
+
+///|
+fn tls_get() -> Int = "$root" "[context-get-0]"
+
+///|
+pub fn task_cancel() = "[export]$root" "[task-cancel]"
+
+///|
+fn waitable_set_new() -> Int = "$root" "[waitable-set-new]"
+
+///|
+fn waitable_set_drop(set : Int) = "$root" "[waitable-set-drop]"
+
+///|
+fn waitable_join(waitable : Int, set : Int) = "$root" "[waitable-join]"
+
+// #endregion

crates/moonbit/src/async/async_primitive.mbt

@@ -0,0 +1,21 @@
+// Copyright 2025 International Digital Economy Academy
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+///|
+async fn[T] async_suspend(
+  cb : ((T) -> Unit, (Cancelled) -> Unit) -> Unit,
+) -> T raise Cancelled = "%async.suspend"
+
+///|
+fn run_async(f : async () -> Unit noraise) = "%async.run"