Add cooperative threads

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -358,22 +358,6 @@ def write(self, vs):
       assert(all(v == () for v in vs))
     self.progress += len(vs)
 
-#### Context-Local Storage
-
-class ContextLocalStorage:
-  LENGTH = 1
-  array: list[int]
-
-  def __init__(self):
-    self.array = [0] * ContextLocalStorage.LENGTH
-
-  def set(self, i, v):
-    assert(types_match_values(['i32'], [v]))
-    self.array[i] = v
-
-  def get(self, i):
-    return self.array[i]
-
 #### Waitable State
 
 class EventCode(IntEnum):
@@ -475,7 +459,6 @@ class State(Enum):
   supertask: Optional[Task]
   on_resolve: Callable[[Optional[list[any]]], None]
   num_borrows: int
-  context: ContextLocalStorage
 
   def __init__(self, opts, inst, ft, supertask, on_resolve):
     self.state = Task.State.INITIAL
@@ -485,7 +468,6 @@ def __init__(self, opts, inst, ft, supertask, on_resolve):
     self.supertask = supertask
     self.on_resolve = on_resolve
     self.num_borrows = 0
-    self.context = ContextLocalStorage()
 
   async def enter(self, thread):
     self.trap_if_on_the_stack(self.inst)
@@ -889,12 +871,16 @@ def drop(self):
 
 class Thread:
   task: Task
+  context: list[int]
   awaitable: Optional[Awaitable]
   on_resume: Optional[asyncio.Future]
   on_suspend_or_exit: Optional[asyncio.Future]
 
+  CONTEXT_LENGTH = 1
+
   def __init__(self, task, coro):
     self.task = task
+    self.context = [0] * Thread.CONTEXT_LENGTH
     self.awaitable = asyncio.Future()
     self.on_resume = asyncio.Future()
     self.on_suspend_or_exit = None
@@ -930,6 +916,18 @@ async def suspend(self, awaitable) -> Cancelled:
     self.on_resume = None
     return cancelled
 
+  async def switch(self, other: Thread) -> Cancelled:
+    assert(not self.awaitable and not other.awaitable)
+    assert(self.on_suspend_or_exit and not other.on_suspend_or_exit)
+    other.on_suspend_or_exit = self.on_suspend_or_exit
+    self.on_suspend_or_exit = None
+    other.on_resume.set_result(Cancelled.FALSE)
+    assert(not self.on_resume)
+    self.on_resume = asyncio.Future()
+    cancelled = await self.on_resume
+    self.on_resume = None
+    return cancelled
+
 #### Store State / Embedding API
 
 class Store:
@@ -2108,19 +2106,31 @@ async def canon_resource_rep(rt, thread, i):
   trap_if(h.rt is not rt)
   return [h.rep]
 
+### 🧵 `canon thread.new_indirect`
+
+async def canon_thread_new_indirect(shared, ft, ftbl, thread, i, c):
+  inst = thread.task.inst
+  trap_if(not inst.may_leave)
+  assert(not shared)
+  f = ftbl.get(i)
+  trap_if(f is None)
+  trap_if(f.type != ft)
+  new_thread = Thread(thread.task, f(c))
+  return [inst.table.add(thread)]
+
 ### 🔀 `canon context.get`
 
 async def canon_context_get(t, i, thread):
   assert(t == 'i32')
-  assert(i < ContextLocalStorage.LENGTH)
-  return [thread.task.context.get(i)]
+  assert(i < Thread.CONTEXT_LENGTH)
+  return [thread.context[i]]
 
 ### 🔀 `canon context.set`
 
 async def canon_context_set(t, i, thread, v):
   assert(t == 'i32')
-  assert(i < ContextLocalStorage.LENGTH)
-  thread.task.context.set(i, v)
+  assert(i < Thread.CONTEXT_LENGTH)
+  thread.context[i] = v
   return []
 
 ### 🔀 `canon backpressure.set`