From ebf3fd644ffcfdf5da27da57ee10e370efb97b36 Mon Sep 17 00:00:00 2001 From: Luke Wagner Date: Wed, 12 Aug 2026 18:33:40 -0500 Subject: [PATCH] CABI: when donut-wrapping, there needs to be implicit backpressure for recursive sync-to-async calls --- design/mvp/CanonicalABI.md | 36 +++++-- design/mvp/Concurrency.md | 6 +- design/mvp/canonical-abi/definitions.py | 28 +++-- design/mvp/canonical-abi/run_tests.py | 130 ++++++++++++++++++++++++ 4 files changed, 181 insertions(+), 19 deletions(-) diff --git a/design/mvp/CanonicalABI.md b/design/mvp/CanonicalABI.md index 621296b5..7962d1ed 100644 --- a/design/mvp/CanonicalABI.md +++ b/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 diff --git a/design/mvp/Concurrency.md b/design/mvp/Concurrency.md index 44cb269d..e3d3b2ed 100644 --- a/design/mvp/Concurrency.md +++ b/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` diff --git a/design/mvp/canonical-abi/definitions.py b/design/mvp/canonical-abi/definitions.py index 8f04f093..d963175d 100644 --- a/design/mvp/canonical-abi/definitions.py +++ b/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) diff --git a/design/mvp/canonical-abi/run_tests.py b/design/mvp/canonical-abi/run_tests.py index 619e5f42..ab5053be 100644 --- a/design/mvp/canonical-abi/run_tests.py +++ b/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")