Rebase CABI onto explicit stack-switching interface (no behavior change)

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -362,6 +362,52 @@ def __init__(self, impl, dtor = None, dtor_async = False, dtor_callback = None):
     self.dtor_async = dtor_async
     self.dtor_callback = dtor_callback
 
+#### Stack Switching Support
+
+class Continuation:
+  lock: threading.Lock
+  handler: Handler
+
+class Handler:
+  tls = threading.local()
+  lock: threading.Lock
+  result: Optional[tuple[Continuation, any]]
+
+def cont_new(f: Callable[[], None]) -> Continuation:
+  cont = Continuation()
+  cont.lock = threading.Lock()
+  cont.lock.acquire()
+  def wrapper():
+    cont.lock.acquire()
+    Handler.tls.value = cont.handler
+    f()
+    handler = Handler.tls.value
+    del Handler.tls.value
+    handler.result = None
+    handler.lock.release()
+  threading.Thread(target = wrapper).start()
+  return cont
+
+def resume(cont: Continuation) -> Optional[tuple[Continuation, any]]:
+  handler = Handler()
+  handler.lock = threading.Lock()
+  handler.lock.acquire()
+  cont.handler = handler
+  cont.lock.release()
+  handler.lock.acquire()
+  return handler.result
+
+def suspend(v: any):
+  handler = Handler.tls.value
+  del Handler.tls.value
+  cont = Continuation()
+  cont.lock = threading.Lock()
+  cont.lock.acquire()
+  handler.result = (cont, v)
+  handler.lock.release()
+  cont.lock.acquire()
+  Handler.tls.value = cont.handler
+
 #### Thread State
 
 class SuspendResult(IntEnum):
@@ -370,9 +416,7 @@ class SuspendResult(IntEnum):
 
 class Thread:
   task: Task
-  fiber: threading.Thread
-  fiber_lock: threading.Lock
-  parent_lock: Optional[threading.Lock]
+  cont: Optional[Continuation]
   ready_func: Optional[Callable[[], bool]]
   cancellable: bool
   suspend_result: Optional[SuspendResult]
@@ -383,7 +427,7 @@ class Thread:
   CONTEXT_LENGTH = 2
 
   def running(self):
-    return self.parent_lock is not None
+    return self.cont is None
 
   def suspended(self):
     return not self.running() and self.ready_func is None
@@ -397,51 +441,53 @@ def ready(self):
 
   def __init__(self, task, thread_func):
     self.task = task
-    self.fiber_lock = threading.Lock()
-    self.fiber_lock.acquire()
-    self.parent_lock = None
+    self.cont = None
     self.ready_func = None
     self.cancellable = False
     self.suspend_result = None
     self.in_event_loop = False
     self.index = None
     self.context = [0] * Thread.CONTEXT_LENGTH
-    def fiber_func():
-      self.fiber_lock.acquire()
+    def wrapper():
       assert(self.running() and self.suspend_result == SuspendResult.NOT_CANCELLED)
       self.suspend_result = None
       thread_func(self)
       assert(self.running())
       self.task.thread_stop(self)
       if self.index is not None:
         self.task.inst.threads.remove(self.index)
-      self.parent_lock.release()
-    self.fiber = threading.Thread(target = fiber_func)
-    self.fiber.start()
+        self.index = None
+    self.cont = cont_new(wrapper)
     self.task.thread_start(self)
     assert(self.suspended())
 
   def resume(self, suspend_result = SuspendResult.NOT_CANCELLED):
-    assert(not self.running() and self.suspend_result is None)
+    assert(not self.running())
     if self.ready_func:
       assert(suspend_result == SuspendResult.CANCELLED or self.ready_func())
       self.ready_func = None
       self.task.inst.store.pending.remove(self)
     assert(self.cancellable or suspend_result == SuspendResult.NOT_CANCELLED)
-    self.suspend_result = suspend_result
-    self.parent_lock = threading.Lock()
-    self.parent_lock.acquire()
-    self.fiber_lock.release()
-    self.parent_lock.acquire()
-    self.parent_lock = None
-    assert(not self.running())
+    thread = self
+    while True:
+      assert(not thread.running() and thread.suspend_result is None)
+      cont = thread.cont
+      thread.cont = None
+      thread.suspend_result = suspend_result
+      if not (resume_result := resume(cont)):
+        assert(thread.index is None and thread not in self.task.threads)
+        return
+      thread.cont, switch_to_thread = resume_result
+      if switch_to_thread is None:
+        return
+      thread = switch_to_thread
+      suspend_result = SuspendResult.NOT_CANCELLED
 
   def suspend(self, cancellable) -> SuspendResult:
     assert(self.task.may_block())
     assert(self.running() and not self.cancellable and self.suspend_result is None)
     self.cancellable = cancellable
-    self.parent_lock.release()
-    self.fiber_lock.acquire()
+    suspend(None)
     assert(self.running())
     self.cancellable = False
     suspend_result = self.suspend_result
@@ -465,16 +511,9 @@ def suspend_until(self, ready_func, cancellable = False) -> SuspendResult:
     return self.suspend(cancellable)
 
   def switch_to(self, cancellable, other: Thread) -> SuspendResult:
-    assert(self.running() and not self.cancellable and self.suspend_result is None)
-    assert(other.suspended() and other.suspend_result is None)
+    assert(self.running() and not self.cancellable)
     self.cancellable = cancellable
-    other.suspend_result = SuspendResult.NOT_CANCELLED
-    assert(self.parent_lock and not other.parent_lock)
-    other.parent_lock = self.parent_lock
-    self.parent_lock = None
-    assert(not self.running() and other.running())
-    other.fiber_lock.release()
-    self.fiber_lock.acquire()
+    suspend(other)
     assert(self.running())
     self.cancellable = False
     suspend_result = self.suspend_result