CABI: fix may_block to not use the current task

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -185,21 +185,29 @@ class FutureType(ValType):
 
 class Store:
   waiting: list[Thread]
+  nesting_depth: int
 
   def __init__(self):
     self.waiting = []
+    self.nesting_depth = 0
 
   def invoke(self, f: FuncInst, caller: Optional[Supertask], on_start, on_resolve) -> Call:
     host_caller = Supertask()
     host_caller.inst = None
     host_caller.supertask = caller
-    return f(host_caller, on_start, on_resolve)
+    self.nesting_depth += 1
+    assert(self.nesting_depth == host_caller.num_host_callers())
+    call = f(host_caller, on_start, on_resolve)
+    self.nesting_depth -= 1
 
   def tick(self):
+    assert(self.nesting_depth == 0)
     random.shuffle(self.waiting)
     for thread in self.waiting:
       if thread.ready():
+        self.nesting_depth = 1
         thread.resume(Cancelled.FALSE)
+        self.nesting_depth = 0
         return
 
 FuncInst: Callable[[Optional[Supertask], OnStart, OnResolve], Call]
@@ -211,6 +219,15 @@ class Supertask:
   inst: Optional[ComponentInstance]
   supertask: Optional[Supertask]
 
+  def num_host_callers(self):
+    n = 0
+    t = self
+    while t is not None:
+      if t.inst is None:
+        n += 1
+      t = t.supertask
+    return n
+
 class Call:
   request_cancellation: Callable[[], None]
 
@@ -285,7 +302,9 @@ class ComponentInstance:
   parent: Optional[ComponentInstance]
   handles: Table[ResourceHandle | Waitable | WaitableSet | ErrorContext]
   threads: Table[Thread]
+  may_enter: bool
   may_leave: bool
+  may_block: bool
   backpressure: int
   exclusive: Optional[Task]
   num_waiting_to_enter: int
@@ -296,7 +315,9 @@ def __init__(self, store, parent = None):
     self.parent = parent
     self.handles = Table()
     self.threads = Table()
+    self.may_enter = True
     self.may_leave = True
+    self.may_block = True
     self.backpressure = 0
     self.exclusive = None
     self.num_waiting_to_enter = 0
@@ -489,7 +510,8 @@ def resume_later(self):
 
   def resume(self, cancelled):
     assert(self.cancellable or not cancelled)
-    assert(not self.running())
+    assert(not self.running() and self.task.inst.may_enter)
+    self.task.inst.may_enter = False
     if self.waiting():
       assert(cancelled or self.ready())
       self.ready_func = None
@@ -506,9 +528,11 @@ def resume(self, cancelled):
         break
       thread = switch_to_thread
       cancelled = Cancelled.FALSE
+    assert(not self.task.inst.may_enter)
+    self.task.inst.may_enter = True
 
   def suspend(self, cancellable) -> Cancelled:
-    assert(self.running() and self.task.may_block())
+    assert(self.running() and self.task.inst.may_block)
     if self.task.deliver_pending_cancel(cancellable):
       return Cancelled.TRUE
     self.cancellable = cancellable
@@ -517,7 +541,7 @@ def suspend(self, cancellable) -> Cancelled:
     return cancelled
 
   def wait_until(self, ready_func, cancellable = False) -> Cancelled:
-    assert(self.running() and self.task.may_block())
+    assert(self.running() and self.task.inst.may_block)
     if self.task.deliver_pending_cancel(cancellable):
       return Cancelled.TRUE
     if ready_func() and not DETERMINISTIC_PROFILE and random.randint(0,1):
@@ -528,7 +552,7 @@ def wait_until(self, ready_func, cancellable = False) -> Cancelled:
 
   def yield_until(self, ready_func, cancellable) -> Cancelled:
     assert(self.running())
-    if self.task.may_block():
+    if self.task.inst.may_block:
       return self.wait_until(ready_func, cancellable)
     else:
       assert(ready_func())
@@ -683,12 +707,11 @@ def thread_stop(self, thread):
   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_:
+      assert(self.inst.may_block)
+      self.inst.may_block = False
       return True
     def has_backpressure():
       return self.inst.backpressure > 0 or (self.needs_exclusive() and bool(self.inst.exclusive))
@@ -739,13 +762,17 @@ def deliver_pending_cancel(self, cancellable) -> bool:
   def return_(self, result):
     trap_if(self.state == Task.State.RESOLVED)
     trap_if(self.num_borrows > 0)
+    if not self.ft.async_:
+      assert(not self.inst.may_block)
+      self.inst.may_block = True
     assert(result is not None)
     self.on_resolve(result)
     self.state = Task.State.RESOLVED
 
   def cancel(self):
     trap_if(self.state != Task.State.CANCEL_DELIVERED)
     trap_if(self.num_borrows > 0)
+    assert(self.ft.async_)
     self.on_resolve(None)
     self.state = Task.State.RESOLVED
 
@@ -2038,6 +2065,8 @@ def lower_flat_values(cx, max_flat, vs, ts, out_param = None):
 
 def canon_lift(opts, inst, ft, callee, caller, on_start, on_resolve) -> Call:
   trap_if(call_might_be_recursive(caller, inst))
+  assert(inst.may_enter) # is it actually guaranteed by `call_might_be_recursive`?
+
   task = Task(opts, inst, ft, caller, on_resolve)
   def thread_func(thread):
     if not task.enter(thread):
@@ -2082,7 +2111,7 @@ def thread_func(thread):
           else:
             event = (EventCode.NONE, 0, 0)
         case CallbackCode.WAIT:
-          trap_if(not task.may_block())
+          trap_if(not inst.may_block)
           wset = inst.handles.get(si)
           trap_if(not isinstance(wset, WaitableSet))
           event = wset.wait_until(lambda: not inst.exclusive, thread, cancellable = True)
@@ -2098,6 +2127,7 @@ def thread_func(thread):
 
   thread = Thread(task, thread_func)
   thread.resume(Cancelled.FALSE)
+  assert(ft.async_ or task.state == Task.State.RESOLVED)
   return task
 
 class CallbackCode(IntEnum):
@@ -2124,7 +2154,7 @@ 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_)
+  trap_if(not thread.task.inst.may_block and ft.async_ and not opts.async_)
 
   subtask = Subtask()
   cx = LiftLowerContext(opts, thread.task.inst, subtask)
@@ -2304,7 +2334,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())
+  trap_if(not thread.task.inst.may_block)
   wset = thread.task.inst.handles.get(si)
   trap_if(not isinstance(wset, WaitableSet))
   event = wset.wait(thread, cancellable)
@@ -2355,7 +2385,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_)
+  trap_if(not thread.task.inst.may_block and not async_)
   subtask = thread.task.inst.handles.get(i)
   trap_if(not isinstance(subtask, Subtask))
   trap_if(subtask.resolve_delivered())
@@ -2412,7 +2442,7 @@ 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_)
+  trap_if(not thread.task.inst.may_block and not opts.async_)
 
   e = thread.task.inst.handles.get(i)
   trap_if(not isinstance(e, EndT))
@@ -2466,7 +2496,7 @@ 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_)
+  trap_if(not thread.task.inst.may_block and not opts.async_)
 
   e = thread.task.inst.handles.get(i)
   trap_if(not isinstance(e, EndT))
@@ -2518,7 +2548,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_)
+  trap_if(not thread.task.inst.may_block and not async_)
   e = thread.task.inst.handles.get(i)
   trap_if(not isinstance(e, EndT))
   trap_if(e.shared.t != stream_or_future_t.t)
@@ -2595,7 +2625,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())
+  trap_if(not thread.task.inst.may_block)
   cancelled = thread.suspend(cancellable)
   return [cancelled]