WIP: instead, remove may_block

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -196,11 +196,16 @@ def invoke(self, f: FuncInst, caller: Optional[Supertask], on_start, on_resolve)
     return f(host_caller, on_start, on_resolve)
 
   def tick(self):
+    if (thread := self.find_ready_thread()):
+      thread.resume(Cancelled.FALSE)
+
+  def find_ready_thread(self) -> Optional[Thread]:
     random.shuffle(self.waiting)
     for thread in self.waiting:
       if thread.ready():
-        thread.resume(Cancelled.FALSE)
-        return
+        return thread
+    return None
+
 
 FuncInst: Callable[[Optional[Supertask], OnStart, OnResolve], Call]
 
@@ -286,6 +291,7 @@ class ComponentInstance:
   handles: Table[ResourceHandle | Waitable | WaitableSet | ErrorContext]
   threads: Table[Thread]
   may_leave: bool
+  may_enter: bool
   backpressure: int
   exclusive: Optional[Task]
   num_waiting_to_enter: int
@@ -297,6 +303,7 @@ def __init__(self, store, parent = None):
     self.handles = Table()
     self.threads = Table()
     self.may_leave = True
+    self.may_enter = True
     self.backpressure = 0
     self.exclusive = None
     self.num_waiting_to_enter = 0
@@ -503,12 +510,16 @@ def resume(self, cancelled):
     while thread is not None:
       cont = thread.cont
       thread.cont = None
+      assert(thread.task.inst.may_enter)
+      thread.task.inst.may_enter = False
       (thread.cont, switch_to) = resume(cont, cancelled, thread)
+      assert(not thread.task.inst.may_enter)
+      thread.task.inst.may_enter = True
       thread = switch_to
       cancelled = Cancelled.FALSE
 
   def suspend(self, cancellable) -> Cancelled:
-    assert(self.running() and self.task.may_block())
+    assert(self.running())
     if self.task.deliver_pending_cancel(cancellable):
       return Cancelled.TRUE
     self.cancellable = cancellable
@@ -517,7 +528,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())
     if self.task.deliver_pending_cancel(cancellable):
       return Cancelled.TRUE
     if ready_func() and not DETERMINISTIC_PROFILE and random.randint(0,1):
@@ -526,16 +537,8 @@ def wait_until(self, ready_func, cancellable = False) -> Cancelled:
     self.task.inst.store.waiting.append(self)
     return self.suspend(cancellable)
 
-  def yield_until(self, ready_func, cancellable) -> Cancelled:
-    assert(self.running())
-    if self.task.may_block():
-      return self.wait_until(ready_func, cancellable)
-    else:
-      assert(ready_func())
-      return Cancelled.FALSE
-
   def yield_(self, cancellable) -> Cancelled:
-    return self.yield_until(lambda: True, cancellable)
+    return self.wait_until(lambda: True, cancellable)
 
   def switch_to(self, cancellable, other: Thread) -> Cancelled:
     assert(self.running() and other.suspended())
@@ -673,26 +676,22 @@ def __init__(self, opts, inst, ft, supertask, on_resolve):
   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 = current_thread()
-    if self.ft.async_:
-      def has_backpressure():
-        return self.inst.backpressure > 0 or (self.needs_exclusive() and bool(self.inst.exclusive))
-      if has_backpressure() or self.inst.num_waiting_to_enter > 0:
-        self.inst.num_waiting_to_enter += 1
-        self.waiting_to_enter = thread
-        cancelled = thread.wait_until(lambda: not has_backpressure(), cancellable = True)
-        self.waiting_to_enter = None
-        self.inst.num_waiting_to_enter -= 1
-        if cancelled:
-          self.cancel()
-          return False
-      if self.needs_exclusive():
-        assert(self.inst.exclusive is None)
-        self.inst.exclusive = self
+    def has_backpressure():
+      return self.inst.backpressure > 0 or (self.needs_exclusive() and bool(self.inst.exclusive))
+    if has_backpressure() or self.inst.num_waiting_to_enter > 0:
+      self.inst.num_waiting_to_enter += 1
+      self.waiting_to_enter = thread
+      cancelled = thread.wait_until(lambda: not has_backpressure(), cancellable = True)
+      self.waiting_to_enter = None
+      self.inst.num_waiting_to_enter -= 1
+      if cancelled:
+        self.cancel()
+        return False
+    if self.needs_exclusive():
+      assert(self.inst.exclusive is None)
+      self.inst.exclusive = self
     self.register_thread(thread)
     return True
 
@@ -2041,6 +2040,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) # true?
+
   task = Task(opts, inst, ft, caller, on_resolve)
   def thread_func():
     if not task.enter():
@@ -2077,12 +2078,11 @@ def thread_func():
       inst.exclusive = None
       match code:
         case CallbackCode.YIELD:
-          if thread.yield_until(lambda: not inst.exclusive, cancellable = True) == Cancelled.TRUE:
+          if thread.wait_until(lambda: not inst.exclusive, cancellable = True) == Cancelled.TRUE:
             event = (EventCode.TASK_CANCELLED, 0, 0)
           else:
             event = (EventCode.NONE, 0, 0)
         case CallbackCode.WAIT:
-          trap_if(not task.may_block())
           wset = inst.handles.get(si)
           trap_if(not isinstance(wset, WaitableSet))
           event = wset.wait_until(lambda: not inst.exclusive, cancellable = True)
@@ -2098,6 +2098,11 @@ def thread_func():
 
   thread = Thread(task, thread_func)
   thread.resume(Cancelled.FALSE)
+  if not task.ft.async_:
+    while task.state != Task.State.RESOLVED:
+      other = inst.store.find_ready_thread()
+      trap_if(other is None)
+      other.resume(Cancelled.FALSE)
   return task
 
 class CallbackCode(IntEnum):
@@ -2125,7 +2130,6 @@ def call_and_trap_on_throw(callee, args):
 def canon_lower(opts, ft, callee: FuncInst, flat_args):
   thread = current_thread()
   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)
@@ -2311,13 +2315,12 @@ def canon_waitable_set_new():
 ### 🔀 `canon waitable-set.wait`
 
 def canon_waitable_set_wait(cancellable, mem, si, ptr):
-  task = current_thread().task
-  trap_if(not task.inst.may_leave)
-  trap_if(not task.may_block())
-  wset = task.inst.handles.get(si)
+  inst = current_thread().task.inst
+  trap_if(not inst.may_leave)
+  wset = inst.handles.get(si)
   trap_if(not isinstance(wset, WaitableSet))
   event = wset.wait(cancellable)
-  return unpack_event(mem, task.inst, ptr, event)
+  return unpack_event(mem, inst, ptr, event)
 
 def unpack_event(mem, inst, ptr, e: EventTuple):
   event, p1, p2 = e
@@ -2368,7 +2371,6 @@ def canon_waitable_join(wi, si):
 def canon_subtask_cancel(async_, i):
   thread = current_thread()
   trap_if(not thread.task.inst.may_leave)
-  trap_if(not thread.task.may_block() and not async_)
   subtask = thread.task.inst.handles.get(i)
   trap_if(not isinstance(subtask, Subtask))
   trap_if(subtask.resolve_delivered())
@@ -2429,7 +2431,6 @@ def canon_stream_write(stream_t, opts, i, ptr, n):
 def stream_copy(EndT, BufferT, event_code, stream_t, opts, i, ptr, n):
   thread = current_thread()
   trap_if(not thread.task.inst.may_leave)
-  trap_if(not thread.task.may_block() and not opts.async_)
 
   e = thread.task.inst.handles.get(i)
   trap_if(not isinstance(e, EndT))
@@ -2484,7 +2485,6 @@ def canon_future_write(future_t, opts, i, ptr):
 def future_copy(EndT, BufferT, event_code, future_t, opts, i, ptr):
   thread = current_thread()
   trap_if(not thread.task.inst.may_leave)
-  trap_if(not thread.task.may_block() and not opts.async_)
 
   e = thread.task.inst.handles.get(i)
   trap_if(not isinstance(e, EndT))
@@ -2537,7 +2537,6 @@ def canon_future_cancel_write(future_t, async_, i):
 def cancel_copy(EndT, event_code, stream_or_future_t, async_, i):
   thread = current_thread()
   trap_if(not thread.task.inst.may_leave)
-  trap_if(not thread.task.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)
@@ -2620,7 +2619,6 @@ def canon_thread_switch_to(cancellable, i):
 def canon_thread_suspend(cancellable):
   thread = current_thread()
   trap_if(not thread.task.inst.may_leave)
-  trap_if(not thread.task.may_block())
   cancelled = thread.suspend(cancellable)
   return [cancelled]
 

design/mvp/canonical-abi/run_tests.py

@@ -2846,7 +2846,7 @@ def mk_task(supertask, inst):
 test_async_to_async()
 test_async_callback()
 test_callback_interleaving()
-test_sync_ignores_backpressure()
+#test_sync_ignores_backpressure()
 test_async_to_sync()
 test_async_backpressure()
 test_sync_using_wait()