Break waitable.drop into subtask.drop and {stream,future}.close-{readable,writable}

Author: lukewagner

design/mvp/Binary.md

@@ -292,17 +292,21 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 0x0a async?:<async>? m:<core:memdix>                => (canon task.wait async? (memory m) (core func)) 🔀
            | 0x0b async?:<async>? m:<core:memidx>                => (canon task.poll async? (memory m) (core func)) 🔀
            | 0x0c async?:<async>?                                => (canon task.yield async? (core func)) 🔀
-           | 0x0d                                                => (canon waitable.drop (core func)) 🔀
+           | 0x0d                                                => (canon subtask.drop (core func)) 🔀
            | 0x0e t:<typeidx>                                    => (canon stream.new t (core func)) 🔀
            | 0x0f t:<typeidx> opts:<opts>                        => (canon stream.read t opts (core func)) 🔀
            | 0x10 t:<typeidx> opts:<opts>                        => (canon stream.write t opts (core func)) 🔀
            | 0x11 async?:<async?>                                => (canon stream.cancel-read async? (core func)) 🔀
            | 0x12 async?:<async?>                                => (canon stream.cancel-write async? (core func)) 🔀
-           | 0x13 t:<typeidx>                                    => (canon future.new t (core func)) 🔀
-           | 0x14 t:<typeidx> opts:<opts>                        => (canon future.read t opts (core func)) 🔀
-           | 0x15 t:<typeidx> opts:<opts>                        => (canon future.write t opts (core func)) 🔀
-           | 0x16 async?:<async?>                                => (canon future.cancel-read async? (core func)) 🔀
-           | 0x17 async?:<async?>                                => (canon future.cancel-write async? (core func)) 🔀
+           | 0x13 t:<typeidx>                                    => (canon stream.close-readable t (core func)) 🔀
+           | 0x14 t:<typeidx>                                    => (canon stream.close-writable t (core func)) 🔀
+           | 0x15 t:<typeidx>                                    => (canon future.new t (core func)) 🔀
+           | 0x16 t:<typeidx> opts:<opts>                        => (canon future.read t opts (core func)) 🔀
+           | 0x17 t:<typeidx> opts:<opts>                        => (canon future.write t opts (core func)) 🔀
+           | 0x18 async?:<async?>                                => (canon future.cancel-read async? (core func)) 🔀
+           | 0x19 async?:<async?>                                => (canon future.cancel-write async? (core func)) 🔀
+           | 0x1a t:<typeidx>                                    => (canon future.close-readable t (core func)) 🔀
+           | 0x1b t:<typeidx>                                    => (canon future.close-writable t (core func)) 🔀
 async?   ::= 0x00                                                =>
            | 0x01                                                => async
 opts     ::= opt*:vec(<canonopt>)                                => opt*

design/mvp/CanonicalABI.md

@@ -791,8 +791,8 @@ counter is used below to record the number of unmet obligations to drop the
 streams and futures connected to this `Subtask`.
 ```python
   def drop(self):
-    trap_if(self.enqueued)
     trap_if(self.state != CallState.DONE)
+    assert(not self.enqueued)
     self.supertask.todo -= 1
 ```
 
@@ -1058,24 +1058,29 @@ and ensures that streams-of-borrows are dropped before the end of the call,
 just like normal `borrow` handles.
 
 Given the above logic, the [readable and writable ends] of a stream can be
-concretely implemented by the following two classes. The `copy`, `cancel_copy`
-and `drop` methods are called polymorphically by the common `async_copy`
-routine shared by the `stream.read` and `stream.write` built-ins below.
+concretely implemented by the following two classes. The readable end
+inherits `StreamHandle`'s constructor, which takes an already-created abstract
+`ReadableStream` passed into the component. In contrast, constructing a
+writable end constructs a fresh `ReadableStreamGuestImpl` that will later
+be given to the readable end paired with this writable end. The `copy`,
+`cancel_copy` and `drop` methods are called polymorphically by the common
+`async_copy` routine shared by the `stream.read` and `stream.write` built-ins
+below.
 ```python
 class ReadableStreamHandle(StreamHandle):
   async def copy(self, dst, on_block):
     await self.stream.read(dst, on_block)
   async def cancel_copy(self, dst, on_block):
     await self.stream.cancel_read(dst, on_block)
 
-class WritableStreamHandle(ReadableStreamGuestImpl, StreamHandle):
+class WritableStreamHandle(StreamHandle):
   def __init__(self, t, inst):
-    ReadableStreamGuestImpl.__init__(self, inst)
-    StreamHandle.__init__(self, self, t)
+    stream = ReadableStreamGuestImpl(inst)
+    StreamHandle.__init__(self, stream, t)
   async def copy(self, src, on_block):
-    await self.write(src, on_block)
+    await self.stream.write(src, on_block)
   async def cancel_copy(self, src, on_block):
-    await self.cancel_write(src, on_block)
+    await self.stream.cancel_write(src, on_block)
 ```
 
 Given the above definitions of how `stream` works, a `future` can simply be
@@ -1097,24 +1102,24 @@ class ReadableFutureHandle(FutureHandle):
     if dst.remain() == 0:
       self.stream.close()
 
-class WritableFutureHandle(ReadableStreamGuestImpl, FutureHandle):
+class WritableFutureHandle(FutureHandle):
   def __init__(self, t, inst):
-    ReadableStreamGuestImpl.__init__(self, inst)
-    FutureHandle.__init__(self, self, t)
+    stream = ReadableStreamGuestImpl(inst)
+    FutureHandle.__init__(self, stream, t)
 
   async def copy(self, src, on_block):
     assert(src.remain() == 1)
-    await self.write(src, on_block)
+    await self.stream.write(src, on_block)
     if src.remain() == 0:
-      self.close()
+      self.stream.close()
 
   async def cancel_copy(self, src, on_block):
     await self.cancel_write(src, on_block)
     if src.remain() == 0:
-      self.close()
+      self.stream.close()
 
   def drop(self):
-    trap_if(not self.closed())
+    trap_if(not self.stream.closed())
     FutureHandle.drop(self)
 ```
 The overridden `WritableFutureHandle.drop` method traps if the internal stream
@@ -2061,12 +2066,13 @@ def lower_future(cx, v, t):
 
 def lower_async_value(ReadableHandleT, WritableHandleT, cx, v, t):
   assert(isinstance(v, ReadableStream))
-  if isinstance(v, WritableHandleT) and cx.inst is v.impl:
-    i = cx.inst.waitables.array.index(v)
-    assert(v.paired)
-    v.paired = False
+  if isinstance(v, ReadableStreamGuestImpl) and cx.inst is v.impl:
+    [h] = [h for h in cx.inst.waitables.array if h and h.stream is v]
+    assert(h.paired)
+    h.paired = False
     if contains_borrow(t):
-      v.borrow_scope = None
+      h.borrow_scope = None
+    i = cx.inst.waitables.array.index(h)
     assert(2**31 > Table.MAX_LENGTH >= i)
     return i | (2**31)
   else:
@@ -2091,12 +2097,14 @@ it itself holds the `WritableStreamHandle` for. Without specially handling
 this case, this would lead to copies from a single linear memory into itself
 which is both inefficient and raises subtle semantic interleaving questions
 that we would rather avoid. To avoid both, this case is detected and the
-`ReadableStream` is "unwrapped" to writable handle, returning the existing
+`ReadableStream` is "unwrapped" to the writable handle, returning the existing
 index of it in the `waitables` table, setting the high bit to signal this fact
 to guest code. Guest code must therefore handle this special case by
 collapsing the two ends of the stream to work fully without guest code (since
 the Canonical ABI is now wholly unnecessary to pass values from writer to
-reader).
+reader). The O(N) searches through the `waitables` table are expected to be
+optimized away by instead storing a pointer or index of the writable handle in
+the stream itself (alongside the `impl` field).
 
 
 ### Flattening
@@ -2969,6 +2977,27 @@ execute, however tasks in *other* component instances may execute. This allows
 a long-running task in one component to avoid starving other components
 without needing support full reentrancy.
 
+### 🔀 `canon subtask.drop`
+
+For a canonical definition:
+```wasm
+(canon subtask.drop (core func $f))
+```
+validation specifies:
+* `$f` is given type `(func (param i32))`
+
+Calling `$f` removes the subtask at the given index from the current
+component instance's `watiable` table, performing the guards and bookkeeping
+defined by `Subtask.drop()`.
+```python
+async def canon_subtask_drop(task, i):
+  trap_if(not task.inst.may_leave)
+  h = task.inst.waitables.remove(i)
+  trap_if(not isinstance(h, Subtask))
+  h.drop()
+  return []
+```
+
 ### 🔀 `canon {stream,future}.new`
 
 For canonical definitions:
@@ -3177,22 +3206,41 @@ new `waitable` element allocated; the cancellation is simply reported as a
 normal `{STREAM,FUTURE}_{READ,WRITE}` event by the original, now-unblocked
 `read` or `write`.
 
-### 🔀 `canon waitable.drop`
+### 🔀 `canon {stream,future}.close-{readable,writable}`
 
-For a canonical definition:
+For canonical definitions:
 ```wasm
-(canon waitable.drop (core func $f))
+(canon stream.close-readable $t (core func $f))
+(canon stream.close-writable $t (core func $f))
+(canon future.close-readable $t (core func $f))
+(canon future.close-writable $t (core func $f))
 ```
 validation specifies:
 * `$f` is given type `(func (param i32))`
 
-Calling `$f` removes the indicated waitable (subtask, stream or future) from
-the instance's table, trapping if various conditions aren't met in the
-waitable's `drop()` method.
+Calling `$f` removes the readable or writable end of the stream or future at
+the given index from the current component instance's `waitable` table,
+performing the guards and bookkeeping defined by
+`{Readable,Writable}{Stream,Future}Handle.drop()` above.
 ```python
-async def canon_waitable_drop(task, i):
+async def canon_stream_close_readable(t, task, i):
+  return await close_async_value(ReadableStreamHandle, t, task, i)
+
+async def canon_stream_close_writable(t, task, i):
+  return await close_async_value(WritableStreamHandle, t, task, i)
+
+async def canon_future_close_readable(t, task, i):
+  return await close_async_value(ReadableFutureHandle, t, task, i)
+
+async def canon_future_close_writable(t, task, i):
+  return await close_async_value(WritableFutureHandle, t, task, i)
+
+async def close_async_value(HandleT, t, task, i):
   trap_if(not task.inst.may_leave)
-  task.inst.waitables.remove(i).drop()
+  h = task.inst.waitables.remove(i)
+  trap_if(not isinstance(h, HandleT))
+  trap_if(h.t != t)
+  h.drop()
   return []
 ```
 

design/mvp/Explainer.md

@@ -1372,16 +1372,21 @@ canon ::= ...
         | (canon task.wait async? (memory <core:memidx>) (core func <id>?)) 🔀
         | (canon task.poll async? (memory <core:memidx>) (core func <id>?)) 🔀
         | (canon task.yield async? (core func <id>?)) 🔀
+        | (canon subtask.drop (core func <id>?)) 🔀
         | (canon stream.new <typeidx> (core func <id>?)) 🔀
         | (canon stream.read <typeidx> <canonopt>* (core func <id>?)) 🔀
         | (canon stream.write <typeidx> <canonopt>* (core func <id>?)) 🔀
         | (canon stream.cancel-read async? (core func <id>?)) 🔀
         | (canon stream.cancel-write async? (core func <id>?)) 🔀
+        | (canon stream.close-readable <typeidx> (core func <id>?)) 🔀
+        | (canon stream.close-writable <typeidx> (core func <id>?)) 🔀
         | (canon future.new <typeidx> (core func <id>?)) 🔀
         | (canon future.read <typeidx> <canonopt>* (core func <id>?)) 🔀
         | (canon future.write <typeidx> <canonopt>* (core func <id>?)) 🔀
         | (canon future.cancel-read async? (core func <id>?)) 🔀
         | (canon future.cancel-write async? (core func <id>?)) 🔀
+        | (canon future.close-readable <typeidx> (core func <id>?)) 🔀
+        | (canon future.close-writable <typeidx> (core func <id>?)) 🔀
         | (canon waitable.drop (core func <id>?)) 🔀
         | (canon thread.spawn <typeidx> (core func <id>?)) 🧵
         | (canon thread.hw_concurrency (core func <id>?)) 🧵
@@ -1470,6 +1475,10 @@ switch to another task, allowing a long-running computation to cooperatively
 interleave with other tasks. (See also [`canon_task_yield`] in the Canonical
 ABI explainer.)
 
+The `subtask.drop` built-in has type `[i32] -> []` and removes the indicated
+[subtask](Async.md#subtask-and-supertask) from the current instance's subtask
+table, trapping if the subtask isn't done.
+
 The `{stream,future}.new` built-ins have type `[] -> [i32]` and return a new
 [writable end](Async.md#streams-and-futures) of a stream or future. (See
 [`canon_stream_new`] in the Canonical ABI explainer for details.)
@@ -1500,11 +1509,11 @@ blocks, the return value is the sentinel "`BLOCKED`" value and the caller must
 or `write`. (See [`canon_stream_cancel_read`] in the Canonical ABI explainer
 for details.)
 
-The `waitable.drop` built-in has type `[i32] -> []` and removes the indicated
-[subtask](Async.md#subtask-and-supertask) or [stream or future](Async.md#streams-and-futures)
-from the current instance's [waitables](Async.md#waiting) table, trapping if
-the subtask isn't done or the stream or future is in the middle of reading
-or writing.
+The `{stream,future}.close-{readable,writable}` built-ins have type
+`[i32] -> []` and removes the indicated [stream or future](Async.md#streams-and-futures)
+from the current component instance's [waitables](Async.md#waiting) table,
+trapping if the stream or future has a mismatched direction or type or are in
+the middle of a `read` or `write`.
 
 ##### 🧵 Threading built-ins