CABI: fix may_block to not use the current task

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -198,7 +198,7 @@ def invoke(self, f: FuncInst, caller: Optional[Supertask], on_start, on_resolve)
   def tick(self):
     random.shuffle(self.waiting)
     for thread in self.waiting:
-      if thread.ready():
+      if thread.ready() and thread.task.inst.may_enter:
         thread.resume(Cancelled.FALSE)
         return
 
@@ -285,7 +285,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 +298,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 +493,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 +511,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 +524,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 +535,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 +690,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 +745,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 +2048,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)
+
   task = Task(opts, inst, ft, caller, on_resolve)
   def thread_func(thread):
     if not task.enter(thread):
@@ -2082,7 +2094,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 +2110,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 +2137,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 +2317,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 +2368,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 +2425,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 +2479,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 +2531,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 +2608,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]