Add cooperative threads

Author: lukewagner

design/mvp/CanonicalABI.md

@@ -37,6 +37,7 @@ being specified here.
   * [`canon resource.new`](#canon-resourcenew)
   * [`canon resource.drop`](#canon-resourcedrop)
   * [`canon resource.rep`](#canon-resourcerep)
+  * [`canon thread.new_indirect`](#-canon-threadnew_indirect) 🧵
   * [`canon context.get`](#-canon-contextget) 🔀
   * [`canon context.set`](#-canon-contextset) 🔀
   * [`canon backpressure.set`](#-canon-backpressureset) 🔀
@@ -58,9 +59,7 @@ being specified here.
   * [`canon error-context.new`](#-canon-error-contextnew) 📝
   * [`canon error-context.debug-message`](#-canon-error-contextdebug-message) 📝
   * [`canon error-context.drop`](#-canon-error-contextdrop) 📝
-  * [`canon thread.spawn_ref`](#-canon-threadspawn_ref) 🧵
-  * [`canon thread.spawn_indirect`](#-canon-threadspawn_indirect) 🧵
-  * [`canon thread.available_parallelism`](#-canon-threadavailable_parallelism) 🧵
+  * [`canon thread.available_parallelism`](#-canon-threadavailable_parallelism) ⦷
 
 ## Supporting definitions
 
@@ -3419,6 +3418,46 @@ Note that the "locally-defined" requirement above ensures that only the
 component instance defining a resource can access its representation.
 
 
+### 🧵 `canon thread.new_indirect`
+
+TODO: rewrite
+
+For a canonical definition:
+```wat
+(canon thread.new_indirect $shared? $ft $tbl (core func $f))
+```
+validation specifies:
+* `$ft` must refer to a `shared` function type; initially, only the type
+  `(shared (func (param $c i32)))` is allowed (see explanation in
+  `thread.spawn_ref` above)
+* `$tbl` must refer to a shared table whose element type matches `(ref null
+  (shared func))`
+* `$spawn_indirect` is given type `(func (param $i i32) (param $c i32) (result
+  $e i32))`.
+
+Calling `$spawn_indirect` retrieves a reference to function `$f` from table
+`$tbl` and checks that `$f` is of type `$ft`. If that succeeds, it spawns a
+thread which:
+  - invokes `$f` with `$c`
+  - executes `$f` until completion or trap in a `shared` context as described by
+    the [shared-everything threads] proposal.
+
+In pseudocode, `$spawn_indirect` looks like:
+
+```python
+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)]
+```
+TODO
+
+
 ### 🔀 `canon context.get`
 
 For a canonical definition:
@@ -4226,6 +4265,7 @@ async def canon_error_context_debug_message(opts, task, i, ptr):
 Note that `ptr` points to an 8-byte region of memory into which will be stored
 the pointer and length of the debug string (allocated via `opts.realloc`).
 
+
 ### 📝 `canon error-context.drop`
 
 For a canonical definition:
@@ -4246,99 +4286,7 @@ async def canon_error_context_drop(task, i):
 ```
 
 
-### 🧵 `canon thread.spawn_ref`
-
-For a canonical definition:
-```wat
-(canon thread.spawn_ref $ft (core func $spawn_ref))
-```
-validation specifies:
-* `$ft` must refer to a `shared` function type; initially, only the type
-  `(shared (func (param $c i32)))` is allowed (see explanation below)
-* `$spawn_ref` is given type `(func (param $f (ref null $ft)) (param $c i32)
-  (result $e i32))`.
-
-> Note: ideally, a thread could be spawned with [arbitrary thread parameters].
-> Currently, that would require additional work in the toolchain to support so,
-> for simplicity, the current proposal simply fixes a single `i32` parameter
-> type. However, `thread.spawn_ref` could be extended to allow arbitrary thread
-> parameters in the future, once it's concretely beneficial to the toolchain.
-> The inclusion of `$ft` ensures backwards compatibility for when arbitrary
-> parameters are allowed.
-
-Calling `$spawn_ref` checks that the reference `$f` is not null. Then, it spawns
-a thread which:
-  - invokes `$f` with `$c`
-  - executes `$f` until completion or trap in a `shared` context as described by
-    the [shared-everything threads] proposal.
-
-In pseudocode, `$spawn_ref` looks like:
-
-```python
-def canon_thread_spawn_ref(f, c):
-  trap_if(f is None)
-  if DETERMINISTIC_PROFILE:
-    return [-1]
-
-  def thread_start():
-    try:
-      f(c)
-    except CoreWebAssemblyException:
-      trap()
-
-  if spawn(thread_start):
-    return [0]
-  else:
-    return [-1]
-```
-
-
-### 🧵 `canon thread.spawn_indirect`
-
-For a canonical definition:
-```wat
-(canon thread.spawn_indirect $ft $tbl (core func $spawn_indirect))
-```
-validation specifies:
-* `$ft` must refer to a `shared` function type; initially, only the type
-  `(shared (func (param $c i32)))` is allowed (see explanation in
-  `thread.spawn_ref` above)
-* `$tbl` must refer to a shared table whose element type matches `(ref null
-  (shared func))`
-* `$spawn_indirect` is given type `(func (param $i i32) (param $c i32) (result
-  $e i32))`.
-
-Calling `$spawn_indirect` retrieves a reference to function `$f` from table
-`$tbl` and checks that `$f` is of type `$ft`. If that succeeds, it spawns a
-thread which:
-  - invokes `$f` with `$c`
-  - executes `$f` until completion or trap in a `shared` context as described by
-    the [shared-everything threads] proposal.
-
-In pseudocode, `$spawn_indirect` looks like:
-
-```python
-def canon_thread_spawn_indirect(ft, tbl, i, c):
-  f = tbl[i]
-  trap_if(f is None)
-  trap_if(f.type != ft)
-  if DETERMINISTIC_PROFILE:
-    return [-1]
-
-  def thread_start():
-    try:
-      f(c)
-    except CoreWebAssemblyException:
-      trap()
-
-  if spawn(thread_start):
-    return [0]
-  else:
-    return [-1]
-```
-
-
-### 🧵 `canon thread.available_parallelism`
+### ⦷ `canon thread.available_parallelism`
 
 For a canonical definition:
 ```wat

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`