Fill in missing case of implicit backpressure when donut-wrapping

design/mvp/CanonicalABI.md

@@ -126,6 +126,7 @@ class ComponentInstance:
   threads: Table[Thread]
   may_enter: bool
   may_leave: bool
+  sync_depth: int
   backpressure: int
   num_waiting_to_enter: int
   exclusive_thread: Optional[Thread]
@@ -138,6 +139,7 @@ class ComponentInstance:
     self.threads = Table()
     self.may_enter = True
     self.may_leave = True
+    self.sync_depth = 0
     self.backpressure = 0
     self.num_waiting_to_enter = 0
     self.exclusive_thread = None
@@ -819,7 +821,10 @@ of backpressure:
     `backpressure.{inc,dec}` which modify the `ComponentInstance.backpressure`
     counter.
  2. *Implicit backpressure* triggered when `Task.needs_exclusive()` is true and
-    the `ComponentInstance.exclusive_thread` lock is already held.
+    either the `ComponentInstance.exclusive_thread` lock is already held *or*,
+    in a [donut wrapping] scenario, a parent's `async` function is being called
+    by a child component's import while the parent has a non-`async` call
+    already on the stack.
  3. *Residual backpressure* triggered by explicit or implicit backpressure
     having been enabled then disabled, but there still being tasks waiting to
     enter that need to be given the chance to start without getting starved
@@ -837,8 +842,10 @@ exports.
     self.implicit_thread = current_thread()
     if self.ft.async_:
       def has_backpressure():
-        return (self.inst.backpressure > 0 or
-                (self.needs_exclusive() and self.inst.exclusive_thread is not None))
+        return (self.inst.backpressure > 0
+                or (self.needs_exclusive()
+                    and (self.inst.exclusive_thread is not None
+                         or self.inst.sync_depth > 0)))
       if has_backpressure() or self.inst.num_waiting_to_enter > 0:
         self.inst.num_waiting_to_enter += 1
         cancelled = self.implicit_thread.wait_until(lambda: not has_backpressure(), cancellable = True)
@@ -849,6 +856,8 @@ exports.
       if self.needs_exclusive():
         assert(self.inst.exclusive_thread is None)
         self.inst.exclusive_thread = self.implicit_thread
+    else:
+      self.inst.sync_depth += 1
     self.register_thread(self.implicit_thread)
     return True
 
@@ -884,9 +893,12 @@ returned a value to its caller.
   def exit_implicit_thread(self):
     assert(current_thread() is self.implicit_thread)
     self.unregister_thread(self.implicit_thread)
-    if self.ft.async_ and self.needs_exclusive():
-      assert(self.inst.exclusive_thread is self.implicit_thread)
-      self.inst.exclusive_thread = None
+    if self.ft.async_:
+      if self.needs_exclusive():
+        assert(self.inst.exclusive_thread is self.implicit_thread)
+        self.inst.exclusive_thread = None
+    else:
+      self.inst.sync_depth -= 1
 
   def unregister_thread(self, thread):
     assert(thread in self.threads and thread.task is self)
@@ -916,9 +928,12 @@ multiple), giving the thread the chance to handle cancellation promptly so that
       self.implicit_thread.resume(Cancelled.TRUE)
     else:
       assert(self.state == Task.State.STARTED)
-      candidates = { t for t in self.threads if t.cancellable }
-      if self.needs_exclusive() and self.inst.exclusive_thread not in { None, self.implicit_thread }:
-        candidates.discard(self.implicit_thread)
+      def exclusive_conflict(thread):
+        return (self.needs_exclusive()
+                and thread is self.implicit_thread
+                and (self.inst.exclusive_thread not in { None, self.implicit_thread }
+                     or self.inst.sync_depth > 0))
+      candidates = { t for t in self.threads if t.cancellable and not exclusive_conflict(t) }
       if candidates and self.inst.may_enter_from(caller):
         self.state = Task.State.CANCEL_DELIVERED
         self.inst.enter_from(caller)
@@ -932,7 +947,8 @@ thread when doing so would violate [Component Invariant] #2 or #3. In
 particular, invariant #2 requires not resuming any thread while the task's
 containing component instance may not be reentered and invariant #3 requires not
 resuming a `needs_exclusive` task's implicit thread while another task's
-implicit thread is running exclusively.
+`needs_exclusive` implicit thread is holding the `exclusive_thread` lock *or*
+there's a non-`async` call on the stack (which must execute in a LIFO manner).
 
 If cancellation cannot be immediately delivered by `Task.request_cancellation`,
 the request is remembered in `Task.state` and delivered at the next opportunity

design/mvp/Concurrency.md

@@ -711,7 +711,11 @@ the event loop after every event (instead of once at the end of the task),
 stackless async exports release the lock between every event, allowing a higher
 degree of concurrency than synchronous exports. Stackful async exports ignore
 the lock entirely and thus achieve the highest degree of (cooperative)
-concurrency.
+concurrency. Another source of implicit backpressure arises when, in a [donut
+wrapping] scenario, a recursive `async` call into the parent that requires the
+exclusive lock is attempted while the parent is actively executing a
+non-`async`-typed call (as this would otherwise allow non-LIFO execution that
+would break invariant #3).
 
 Since non-`async` functions are not allowed to block (including due to
 backpressure) and also don't pile up like `async` functions, non-`async`

design/mvp/canonical-abi/definitions.py

@@ -195,6 +195,7 @@ class ComponentInstance:
   threads: Table[Thread]
   may_enter: bool
   may_leave: bool
+  sync_depth: int
   backpressure: int
   num_waiting_to_enter: int
   exclusive_thread: Optional[Thread]
@@ -207,6 +208,7 @@ def __init__(self, store, parent = None):
     self.threads = Table()
     self.may_enter = True
     self.may_leave = True
+    self.sync_depth = 0
     self.backpressure = 0
     self.num_waiting_to_enter = 0
     self.exclusive_thread = None
@@ -479,8 +481,10 @@ def enter_implicit_thread(self):
     self.implicit_thread = current_thread()
     if self.ft.async_:
       def has_backpressure():
-        return (self.inst.backpressure > 0 or
-                (self.needs_exclusive() and self.inst.exclusive_thread is not None))
+        return (self.inst.backpressure > 0
+                or (self.needs_exclusive()
+                    and (self.inst.exclusive_thread is not None
+                         or self.inst.sync_depth > 0)))
       if has_backpressure() or self.inst.num_waiting_to_enter > 0:
         self.inst.num_waiting_to_enter += 1
         cancelled = self.implicit_thread.wait_until(lambda: not has_backpressure(), cancellable = True)
@@ -491,6 +495,8 @@ def has_backpressure():
       if self.needs_exclusive():
         assert(self.inst.exclusive_thread is None)
         self.inst.exclusive_thread = self.implicit_thread
+    else:
+      self.inst.sync_depth += 1
     self.register_thread(self.implicit_thread)
     return True
 
@@ -503,9 +509,12 @@ def register_thread(self, thread):
   def exit_implicit_thread(self):
     assert(current_thread() is self.implicit_thread)
     self.unregister_thread(self.implicit_thread)
-    if self.ft.async_ and self.needs_exclusive():
-      assert(self.inst.exclusive_thread is self.implicit_thread)
-      self.inst.exclusive_thread = None
+    if self.ft.async_:
+      if self.needs_exclusive():
+        assert(self.inst.exclusive_thread is self.implicit_thread)
+        self.inst.exclusive_thread = None
+    else:
+      self.inst.sync_depth -= 1
 
   def unregister_thread(self, thread):
     assert(thread in self.threads and thread.task is self)
@@ -522,9 +531,12 @@ def request_cancellation(self, caller: Optional[ComponentInstance]):
       self.implicit_thread.resume(Cancelled.TRUE)
     else:
       assert(self.state == Task.State.STARTED)
-      candidates = { t for t in self.threads if t.cancellable }
-      if self.needs_exclusive() and self.inst.exclusive_thread not in { None, self.implicit_thread }:
-        candidates.discard(self.implicit_thread)
+      def exclusive_conflict(thread):
+        return (self.needs_exclusive()
+                and thread is self.implicit_thread
+                and (self.inst.exclusive_thread not in { None, self.implicit_thread }
+                     or self.inst.sync_depth > 0))
+      candidates = { t for t in self.threads if t.cancellable and not exclusive_conflict(t) }
       if candidates and self.inst.may_enter_from(caller):
         self.state = Task.State.CANCEL_DELIVERED
         self.inst.enter_from(caller)

design/mvp/canonical-abi/run_tests.py

@@ -3020,6 +3020,134 @@ def core_consumer(args):
 
   lift_and_run(mk_opts(), consumer_inst, consumer_ft, core_consumer, lambda:[], lambda _:())
 
+def test_donut_sync_defers_async_enter():
+  store = Store()
+  parent_inst = ComponentInstance(store)
+  child_inst = ComponentInstance(store, parent_inst)
+
+  h_ran = RacyBool(False)
+  h_ft = FuncType([],[], async_ = True)
+  def h_core(args):
+    assert(len(args) == 0)
+    h_ran.set()
+    return []
+  h = store.lift(h_core, h_ft, mk_opts(), parent_inst)
+
+  child_mem = bytearray(16)
+  child_opts = mk_opts(memory = MemInst(child_mem, 'i32'), async_ = True)
+  subi = None
+  g_ft = FuncType([],[])
+  def g_core(args):
+    nonlocal subi
+    assert(parent_inst.sync_depth == 1)
+    [ret] = store.lower(h, h_ft, child_opts, child_inst)([])
+    state,subi = unpack_result(ret)
+    assert(state == Subtask.State.STARTING)
+    assert(h_ran.is_clear())
+    return []
+  g = store.lift(g_core, g_ft, mk_opts(), child_inst)
+
+  f_ft = FuncType([],[])
+  def f_core(args):
+    assert(len(args) == 0)
+    [] = store.lower(g, g_ft, mk_opts(), parent_inst)([])
+    assert(h_ran.is_clear())
+    return []
+
+  lift_and_run(mk_opts(), parent_inst, f_ft, f_core, lambda:[], lambda _:())
+  assert(h_ran.is_set())
+
+  def finish_core(args):
+    [seti] = canon_waitable_set_new()
+    [] = canon_waitable_join(subi, seti)
+    retp = 8
+    [event] = canon_waitable_set_wait(True, MemInst(child_mem, 'i32'), seti, retp)
+    assert(event == EventCode.SUBTASK)
+    assert(child_mem[retp+0] == subi)
+    assert(child_mem[retp+4] == Subtask.State.RETURNED)
+    [] = canon_subtask_drop(subi)
+    [] = canon_waitable_set_drop(seti)
+    return []
+  lift_and_run(mk_opts(), child_inst, FuncType([],[]), finish_core, lambda:[], lambda _:())
+
+def test_donut_sync_defers_cancellation():
+  store = Store()
+  parent_inst = ComponentInstance(store)
+  child_inst = ComponentInstance(store, parent_inst)
+
+  parent_mem = bytearray(16)
+  f_done = RacyBool(False)
+  h_ft = FuncType([FutureType(None)],[], async_ = True)
+  def h_core(args):
+    [rfut] = args
+    [ret] = canon_future_read(FutureType(None), mk_opts(MemInst(parent_mem, 'i32'), async_ = True), rfut, 0xdeadbeef)
+    assert(ret == definitions.BLOCKED)
+    [seti] = canon_waitable_set_new()
+    [] = canon_waitable_join(rfut, seti)
+    retp = 8
+    [event] = canon_waitable_set_wait(True, MemInst(parent_mem, 'i32'), seti, retp)
+    assert(event == EventCode.FUTURE_READ)
+    assert(parent_mem[retp+0] == rfut)
+    assert(parent_mem[retp+4] == CopyResult.COMPLETED)
+    assert(f_done.is_set())
+    [cancelled] = canon_thread_yield(True)
+    assert(cancelled == Cancelled.TRUE)
+    [] = canon_future_drop_readable(FutureType(None), rfut)
+    [] = canon_waitable_set_drop(seti)
+    return []
+  h = store.lift(h_core, h_ft, mk_opts(), parent_inst)
+
+  child_mem = bytearray(24)
+  child_async_opts = mk_opts(memory = MemInst(child_mem, 'i32'), async_ = True)
+  subi = None
+  wfut = None
+  setup_ft = FuncType([],[])
+  def setup_core(args):
+    nonlocal subi, wfut
+    [packed] = canon_future_new(FutureType(None))
+    rfut,wfut = unpack_new_ends(packed)
+    [ret] = store.lower(h, h_ft, child_async_opts, child_inst)([rfut])
+    state,subi = unpack_result(ret)
+    assert(state == Subtask.State.STARTED)
+    return []
+  _ = store.invoke(store.lift(setup_core, setup_ft, mk_opts(), child_inst), lambda:[], lambda _:())
+  assert(parent_inst.exclusive_thread is not None)
+
+  do_cancel_ft = FuncType([],[])
+  def do_cancel_core(args):
+    assert(parent_inst.sync_depth == 1)
+    [ret] = canon_subtask_cancel(True, subi)
+    assert(ret == definitions.BLOCKED)
+    [ret] = canon_future_write(FutureType(None), mk_opts(MemInst(child_mem, 'i32')), wfut, 0xdeadbeef)
+    assert(ret == CopyResult.COMPLETED)
+    [] = canon_future_drop_writable(FutureType(None), wfut)
+    return []
+  do_cancel = store.lift(do_cancel_core, do_cancel_ft, mk_opts(), child_inst)
+
+  f_ft = FuncType([],[])
+  def f_core(args):
+    assert(len(args) == 0)
+    [] = store.lower(do_cancel, do_cancel_ft, mk_opts(), parent_inst)([])
+    f_done.set()
+    return []
+  _ = store.invoke(store.lift(f_core, f_ft, mk_opts(), parent_inst), lambda:[], lambda _:())
+
+  while store.waiting:
+    store.tick()
+
+  def finish_core(args):
+    [seti] = canon_waitable_set_new()
+    [] = canon_waitable_join(subi, seti)
+    retp = 8
+    [event] = canon_waitable_set_wait(True, MemInst(child_mem, 'i32'), seti, retp)
+    assert(event == EventCode.SUBTASK)
+    assert(child_mem[retp+0] == subi)
+    assert(child_mem[retp+4] == Subtask.State.RETURNED)
+    [] = canon_subtask_drop(subi)
+    [] = canon_waitable_set_drop(seti)
+    return []
+  lift_and_run(mk_opts(), child_inst, FuncType([],[]), finish_core, lambda:[], lambda _:())
+
 test_roundtrips()
 test_cross_component_realloc()
 test_handles()
@@ -3047,5 +3175,7 @@ def core_consumer(args):
 test_threads()
 test_sync_threads()
 test_thread_cancel_callback()
+test_donut_sync_defers_async_enter()
+test_donut_sync_defers_cancellation()
 
 print("All tests passed")