Make 'async' part of the function type, not a hint (#578)

* Make 'async' part of the function type, not a hint

* Add and update tests

* Remove yielding behavior of waitable-set.poll to allow it to be called from synchronous functions

* Also prohibit blocking during the start function

Author: lukewagner

design/mvp/Binary.md

@@ -216,6 +216,7 @@ valtype       ::= i:<typeidx>                             => i
 resourcetype  ::= 0x3f 0x7f f?:<funcidx>?                 => (resource (rep i32) (dtor f)?)
                 | 0x3e 0x7f f:<funcidx> cb?:<funcidx>?    => (resource (rep i32) (dtor async f (callback cb)?)) 🚝
 functype      ::= 0x40 ps:<paramlist> rs:<resultlist>     => (func ps rs)
+                | 0x43 ps:<paramlist> rs:<resultlist>     => (func async ps rs)
 paramlist     ::= lt*:vec(<labelvaltype>)                 => (param lt)*
 resultlist    ::= 0x00 t:<valtype>                        => (result t)
                 | 0x01 0x00                               =>
@@ -288,7 +289,6 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 0x01 0x00 f:<funcidx> opts:<opts>                   => (canon lower f opts (core func))
            | 0x02 rt:<typeidx>                                   => (canon resource.new rt (core func))
            | 0x03 rt:<typeidx>                                   => (canon resource.drop rt (core func))
-           | 0x07 rt:<typeidx>                                   => (canon resource.drop rt async (core func)) 🚝
            | 0x04 rt:<typeidx>                                   => (canon resource.rep rt (core func))
            | 0x08                                                => (canon backpressure.set (core func)) 🔀✕
            | 0x24                                                => (canon backpressure.inc (core func)) 🔀
@@ -515,7 +515,8 @@ named once.
 
 * The opcodes (for types, canon built-ins, etc) should be re-sorted
 * The two `depname` cases should be merged into one (`dep=<...>`)
-* The two `list` type codes should be merged into one with an optional immediate.
+* The two `list` type codes should be merged into one with an optional immediate
+  and similarly for `func`.
 * The `0x00` variant of `importname'` and `exportname'` will be removed. Any
   remaining variant(s) will be renumbered or the prefix byte will be removed or
   repurposed.

design/mvp/CanonicalABI.md

@@ -1450,16 +1450,11 @@ named-type-list ::= ϵ
 named-type ::= id ':' ty
 ```
 
-The optional `async` hint in a WIT function type indicates that the callee
-is expected to block and thus the caller should emit whatever asynchronous
-language bindings are appropriate (e.g., in JS, Python, C# or Rust, an `async`
-WIT function would emit an `async` JS/Python/C#/Rust function). Because `async`
-is just a hint and not enforced by the runtime, it is technically possible for
-a non-`async` callee to block. In that case, though, it is the *callee's* fault
-for any resultant loss of concurrency, not the caller's. Thus, `async` is
-primarily intended to document expectations in a way that can be taken
-advantage of by bindings generators. (For more details, see the [concurrency
-explainer](Concurrency.md).)
+The optional `async` prefix in a WIT function type indicates that the callee
+may block and thus the caller should use the async ABI and asynchronous
+source-language bindings (e.g., `async` functions in JS, Python, C# or Rust) if
+concurrency execution is desired. For more details, see the [concurrency
+explainer](Concurrency.md#summary).
 
 
 ## Item: `use`
@@ -1689,8 +1684,8 @@ resource-method ::= func-item
                   | 'constructor' param-list ';'
 ```
 
-The optional `async` hint on `static` functions has the same meaning as
-in a non-`static` `func-item`.
+The optional `async` on `static` functions has the same meaning as in a
+non-`static` `func-item`.
 
 The syntax for handle types is presented [below](#handles).
 

design/mvp/Concurrency.md

@@ -88,6 +88,7 @@ class InstanceType(ExternType):
 class FuncType(ExternType):
   params: list[tuple[str,ValType]]
   result: list[ValType|tuple[str,ValType]]
+  async_: bool = False
   def param_types(self):
     return self.extract_types(self.params)
   def result_type(self):
@@ -402,6 +403,7 @@ def resume(self, suspend_result = SuspendResult.NOT_CANCELLED):
     assert(not self.running())
 
   def suspend(self, cancellable) -> SuspendResult:
+    assert(self.task.may_block())
     assert(self.running() and not self.cancellable and self.suspend_result is None)
     self.cancellable = cancellable
     self.parent_lock.release()
@@ -420,6 +422,7 @@ def resume_later(self):
     self.task.inst.store.pending.append(self)
 
   def suspend_until(self, ready_func, cancellable = False) -> SuspendResult:
+    assert(self.task.may_block())
     assert(self.running())
     if ready_func() and not DETERMINISTIC_PROFILE and random.randint(0,1):
       return SuspendResult.NOT_CANCELLED
@@ -566,8 +569,13 @@ def trap_if_on_the_stack(self, inst):
   def needs_exclusive(self):
     return not self.opts.async_ or self.opts.callback
 
+  def may_block(self):
+    return self.ft.async_ or self.state == Task.State.RESOLVED
+
   def enter(self, thread):
     assert(thread in self.threads and thread.task is self)
+    if not self.ft.async_:
+      return True
     def has_backpressure():
       return self.inst.backpressure > 0 or (self.needs_exclusive() and self.inst.exclusive)
     if has_backpressure() or self.inst.num_waiting_to_enter > 0:
@@ -584,6 +592,8 @@ def has_backpressure():
 
   def exit(self):
     assert(len(self.threads) > 0)
+    if not self.ft.async_:
+      return
     if self.needs_exclusive():
       assert(self.inst.exclusive)
       self.inst.exclusive = False
@@ -641,20 +651,6 @@ def ready_and_has_event():
     wset.num_waiting -= 1
     return event
 
-  def poll_until(self, ready_func, thread, wset, cancellable) -> Optional[EventTuple]:
-    assert(thread in self.threads and thread.task is self)
-    wset.num_waiting += 1
-    match self.suspend_until(ready_func, thread, cancellable):
-      case SuspendResult.CANCELLED:
-        event = (EventCode.TASK_CANCELLED, 0, 0)
-      case SuspendResult.NOT_CANCELLED:
-        if wset.has_pending_event():
-          event = wset.get_pending_event()
-        else:
-          event = (EventCode.NONE, 0, 0)
-    wset.num_waiting -= 1
-    return event
-
   def yield_until(self, ready_func, thread, cancellable) -> EventTuple:
     assert(thread in self.threads and thread.task is self)
     match self.suspend_until(ready_func, thread, cancellable):
@@ -2028,15 +2024,17 @@ def thread_func(thread):
       inst.exclusive = False
       match code:
         case CallbackCode.YIELD:
-          event = task.yield_until(lambda: not inst.exclusive, thread, cancellable = True)
+          if task.may_block():
+            event = task.yield_until(lambda: not inst.exclusive, thread, cancellable = True)
+          else:
+            event = (EventCode.NONE, 0, 0)
         case CallbackCode.WAIT:
+          trap_if(not task.may_block())
           wset = inst.table.get(si)
           trap_if(not isinstance(wset, WaitableSet))
           event = task.wait_until(lambda: not inst.exclusive, thread, wset, cancellable = True)
-        case CallbackCode.POLL:
-          wset = inst.table.get(si)
-          trap_if(not isinstance(wset, WaitableSet))
-          event = task.poll_until(lambda: not inst.exclusive, thread, wset, cancellable = True)
+        case _:
+          trap()
       thread.in_event_loop = False
       inst.exclusive = True
       event_code, p1, p2 = event
@@ -2053,8 +2051,7 @@ class CallbackCode(IntEnum):
   EXIT = 0
   YIELD = 1
   WAIT = 2
-  POLL = 3
-  MAX = 3
+  MAX = 2
 
 def unpack_callback_result(packed):
   code = packed & 0xf
@@ -2074,6 +2071,8 @@ def call_and_trap_on_throw(callee, thread, args):
 
 def canon_lower(opts, ft, callee: FuncInst, thread, flat_args):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block() and ft.async_ and not opts.async_)
+
   subtask = Subtask()
   cx = LiftLowerContext(opts, thread.task.inst, subtask)
 
@@ -2113,6 +2112,7 @@ def on_resolve(result):
       flat_results = lower_flat_values(cx, max_flat_results, result, ft.result_type(), flat_args)
 
   subtask.callee = callee(thread.task, on_start, on_resolve)
+  assert(ft.async_ or subtask.state == Subtask.State.RETURNED)
 
   if not opts.async_:
     if not subtask.resolved():
@@ -2147,31 +2147,30 @@ def canon_resource_new(rt, thread, rep):
 
 ### `canon resource.drop`
 
-def canon_resource_drop(rt, async_, thread, i):
+def canon_resource_drop(rt, thread, i):
   trap_if(not thread.task.inst.may_leave)
   inst = thread.task.inst
   h = inst.table.remove(i)
   trap_if(not isinstance(h, ResourceHandle))
   trap_if(h.rt is not rt)
   trap_if(h.num_lends != 0)
-  flat_results = [] if not async_ else [0]
   if h.own:
     assert(h.borrow_scope is None)
     if inst is rt.impl:
       if rt.dtor:
         rt.dtor(h.rep)
     else:
       if rt.dtor:
-        caller_opts = CanonicalOptions(async_ = async_)
+        caller_opts = CanonicalOptions(async_ = False)
         callee_opts = CanonicalOptions(async_ = rt.dtor_async, callback = rt.dtor_callback)
-        ft = FuncType([U32Type()],[])
+        ft = FuncType([U32Type()],[], async_ = False)
         callee = partial(canon_lift, callee_opts, rt.impl, ft, rt.dtor)
-        flat_results = canon_lower(caller_opts, ft, callee, thread, [h.rep])
+        [] = canon_lower(caller_opts, ft, callee, thread, [h.rep])
       else:
         thread.task.trap_if_on_the_stack(rt.impl)
   else:
     h.borrow_scope.num_borrows -= 1
-  return flat_results
+  return []
 
 ### `canon resource.rep`
 
@@ -2249,6 +2248,7 @@ def canon_waitable_set_new(thread):
 
 def canon_waitable_set_wait(cancellable, mem, thread, si, ptr):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block())
   wset = thread.task.inst.table.get(si)
   trap_if(not isinstance(wset, WaitableSet))
   event = thread.task.wait_until(lambda: True, thread, wset, cancellable)
@@ -2267,7 +2267,12 @@ def canon_waitable_set_poll(cancellable, mem, thread, si, ptr):
   trap_if(not thread.task.inst.may_leave)
   wset = thread.task.inst.table.get(si)
   trap_if(not isinstance(wset, WaitableSet))
-  event = thread.task.poll_until(lambda: True, thread, wset, cancellable)
+  if thread.task.deliver_pending_cancel(cancellable):
+    event = (EventCode.TASK_CANCELLED, 0, 0)
+  elif not wset.has_pending_event():
+    event = (EventCode.NONE, 0, 0)
+  else:
+    event = wset.get_pending_event()
   return unpack_event(mem, thread, ptr, event)
 
 ### 🔀 `canon waitable-set.drop`
@@ -2299,6 +2304,7 @@ def canon_waitable_join(thread, wi, si):
 
 def canon_subtask_cancel(async_, thread, i):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block() and not async_)
   subtask = thread.task.inst.table.get(i)
   trap_if(not isinstance(subtask, Subtask))
   trap_if(subtask.resolve_delivered())
@@ -2355,6 +2361,8 @@ def canon_stream_write(stream_t, opts, thread, i, ptr, n):
 
 def stream_copy(EndT, BufferT, event_code, stream_t, opts, thread, i, ptr, n):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block() and not opts.async_)
+
   e = thread.task.inst.table.get(i)
   trap_if(not isinstance(e, EndT))
   trap_if(e.shared.t != stream_t.t)
@@ -2406,6 +2414,8 @@ def canon_future_write(future_t, opts, thread, i, ptr):
 
 def future_copy(EndT, BufferT, event_code, future_t, opts, thread, i, ptr):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block() and not opts.async_)
+
   e = thread.task.inst.table.get(i)
   trap_if(not isinstance(e, EndT))
   trap_if(e.shared.t != future_t.t)
@@ -2456,6 +2466,7 @@ def canon_future_cancel_write(future_t, async_, thread, i):
 
 def cancel_copy(EndT, event_code, stream_or_future_t, async_, thread, i):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block() and not async_)
   e = thread.task.inst.table.get(i)
   trap_if(not isinstance(e, EndT))
   trap_if(e.shared.t != stream_or_future_t.t)
@@ -2532,6 +2543,7 @@ def canon_thread_switch_to(cancellable, thread, i):
 
 def canon_thread_suspend(cancellable, thread):
   trap_if(not thread.task.inst.may_leave)
+  trap_if(not thread.task.may_block())
   suspend_result = thread.task.suspend(thread, cancellable)
   return [suspend_result]
 
@@ -2559,6 +2571,8 @@ def canon_thread_yield_to(cancellable, thread, i):
 
 def canon_thread_yield(cancellable, thread):
   trap_if(not thread.task.inst.may_leave)
+  if not thread.task.may_block():
+    return [SuspendResult.NOT_CANCELLED]
   event_code,_,_ = thread.task.yield_until(lambda: True, thread, cancellable)
   match event_code:
     case EventCode.NONE: