feat: async support in MoonBit (#1393)

* feat: moonbit support async component model & use include_str for embedded moonbit code

* test: add moonbit tests for async component model

* style: cargo fmt

* chore: add features section in Cargo.toml

* refactor: rename WaitableTrait to Waitable and update related implementations

* style: clippy fix

* test: fix stub.mbt function name

* refactor: use FutureReader instead of Future

* refactor: remove defer from Waitable trait & fix callback Event handling

* feat: update future/stream import function signatures to include kind

* fix: enhance task cancel logic & refactor Waitable cancel method

* refactor: async-wasm module for improved task management and error handling
feat: support async component model `Stream` methods

* refactor: remove unused WASIP1

* test: add moonbit async tests

* refactor: use `_async_debug` instead of `print` function

* refactor: remove unused `SubtaskStatus::encode` function and clean up test case

* feat: implement indirect parameter handling and clean up unused type methods

* feat: enhance waitable task management with resource tracking and improved drop semantics

* test: refactor tests using updated syntax

* fix: improve resource management

* refactor: improve async handling

* chore: fix type declarations

* chore: remove extra debug function

* Revert "refactor: improve async handling"

This reverts commit 0f2a664.

* chore: add back waitable_task change

* fix: make future/stream type vtable generate correctly

* chore: clean up unused function generation

---------

Co-authored-by: BigOrangeQWQ <2284086963@qq.com>

Author: peter-jerry-ye

crates/moonbit/Cargo.toml

@@ -17,3 +17,6 @@ anyhow = { workspace = true }
 wit-bindgen-core = { workspace = true }
 heck = { workspace = true }
 clap = { workspace = true, optional = true }
+
+[features]
+clap = ['dep:clap', 'wit-bindgen-core/clap']

crates/moonbit/src/async-wasm/async_primitive.mbt

@@ -0,0 +1,231 @@
+///|
+async fn[T, E : Error] async_suspend(
+  cb : ((T) -> Unit, (E) -> Unit) -> Unit,
+) -> T raise E = "%async.suspend"
+
+///|
+fn run_async(f : async () -> Unit noraise) = "%async.run"
+
+
+///|
+priv enum State {
+  Done
+  Fail(Error)
+  Running
+  Suspend(ok_cont~ : (Unit) -> Unit, err_cont~ : (Error) -> Unit)
+}
+
+///|
+struct Coroutine {
+  coro_id : Int
+  mut state : State
+  mut shielded : Bool
+  mut cancelled : Bool
+  mut ready : Bool
+  downstream : Map[Int, Coroutine]
+}
+
+///|
+pub impl Eq for Coroutine with equal(c1, c2) {
+  c1.coro_id == c2.coro_id
+}
+
+///|
+pub impl Hash for Coroutine with hash_combine(self, hasher) {
+  self.coro_id.hash_combine(hasher)
+}
+
+///|
+pub fn Coroutine::wake(self : Coroutine) -> Unit {
+  self.ready = true
+  scheduler.run_later.push_back(self)
+}
+
+///|
+pub fn Coroutine::run(self : Coroutine) -> Unit {
+  self.ready = true
+  scheduler.run_later.push_front(self)
+}
+
+///|
+pub fn Coroutine::is_done(self : Coroutine) -> Bool {
+  match self.state {
+    Done => true
+    Fail(_) => true
+    Running | Suspend(_) => false
+  }
+}
+
+///|
+pub fn is_being_cancelled() -> Bool {
+  current_coroutine().cancelled
+}
+
+///|
+pub fn current_coroutine_done() -> Bool {
+  guard scheduler.curr_coro is Some(coro) else { return true }
+  coro.is_done()
+}
+
+///|
+pub(all) suberror Cancelled derive(Show)
+
+///|
+pub fn Coroutine::cancel(self : Coroutine) -> Unit {
+  self.cancelled = true
+  if not(self.shielded || self.ready) {
+    self.wake()
+  }
+}
+
+///|
+pub async fn pause() -> Unit {
+  guard scheduler.curr_coro is Some(coro)
+  if coro.cancelled && not(coro.shielded) {
+    raise Cancelled::Cancelled
+  }
+  async_suspend(fn(ok_cont, err_cont) {
+    guard coro.state is Running
+    coro.state = Suspend(ok_cont~, err_cont~)
+    coro.ready = true
+    scheduler.run_later.push_back(coro)
+  })
+}
+
+///|
+pub async fn suspend() -> Unit raise {
+  guard scheduler.curr_coro is Some(coro)
+  if coro.cancelled && not(coro.shielded) {
+    raise Cancelled::Cancelled
+  }
+  scheduler.blocking += 1
+  defer {
+    scheduler.blocking -= 1
+  }
+  async_suspend(fn(ok_cont, err_cont) {
+    guard coro.state is Running
+    coro.state = Suspend(ok_cont~, err_cont~)
+  })
+}
+
+///|
+pub fn spawn(f : async () -> Unit raise) -> Coroutine {
+  scheduler.coro_id += 1
+  let coro = {
+    state: Running,
+    ready: true,
+    shielded: false,
+    downstream: {},
+    coro_id: scheduler.coro_id,
+    cancelled: false,
+  }
+  fn run(_) {
+    run_async(fn() {
+      coro.shielded = false
+      try f() catch {
+        err => coro.state = Fail(err)
+      } noraise {
+        _ => coro.state = Done
+      }
+      for _, coro in coro.downstream {
+        coro.wake()
+      }
+      coro.downstream.clear()
+    })
+  }
+
+  coro.state = Suspend(ok_cont=run, err_cont=_ => ())
+  scheduler.run_later.push_back(coro)
+  coro
+}
+
+///|
+pub fn Coroutine::unwrap(self : Coroutine) -> Unit raise {
+  match self.state {
+    Done => ()
+    Fail(err) => raise err
+    Running | Suspend(_) => panic()
+  }
+}
+
+///|
+pub async fn Coroutine::wait(target : Coroutine) -> Unit raise {
+  guard scheduler.curr_coro is Some(coro)
+  guard not(physical_equal(coro, target))
+  match target.state {
+    Done => return
+    Fail(err) => raise err
+    Running | Suspend(_) => ()
+  }
+  target.downstream[coro.coro_id] = coro
+  try suspend() catch {
+    err => {
+      target.downstream.remove(coro.coro_id)
+      raise err
+    }
+  } noraise {
+    _ => target.unwrap()
+  }
+}
+
+///|
+pub async fn protect_from_cancel(f : async () -> Unit raise) -> Unit raise {
+  guard scheduler.curr_coro is Some(coro)
+  if coro.shielded {
+    // already in a shield, do nothing
+    f()
+  } else {
+    coro.shielded = true
+    defer {
+      coro.shielded = false
+    }
+    f()
+    if coro.cancelled {
+      raise Cancelled::Cancelled
+    }
+  }
+}
+
+
+///|
+priv struct Scheduler {
+  mut coro_id : Int
+  mut curr_coro : Coroutine?
+  mut blocking : Int
+  run_later : @deque.Deque[Coroutine]
+}
+
+///|
+let scheduler : Scheduler = {
+  coro_id: 0,
+  curr_coro: None,
+  blocking: 0,
+  run_later: @deque.new(),
+}
+
+///|
+pub fn current_coroutine() -> Coroutine {
+  scheduler.curr_coro.unwrap()
+}
+
+///|
+pub fn no_more_work() -> Bool {
+  scheduler.blocking == 0 && scheduler.run_later.is_empty()
+}
+
+///|
+pub fn rschedule() -> Unit {
+  while scheduler.run_later.pop_front() is Some(coro) {
+    coro.ready = false
+    guard coro.state is Suspend(ok_cont~, err_cont~) else {  }
+    coro.state = Running
+    let last_coro = scheduler.curr_coro
+    scheduler.curr_coro = Some(coro)
+    if coro.cancelled && !coro.shielded {
+      err_cont(Cancelled::Cancelled)
+    } else {
+      ok_cont(())
+    }
+    scheduler.curr_coro = last_coro
+  }
+}