Add <typeidx> to {stream,future}.cancel-{read,write}

Author: lukewagner

design/mvp/Binary.md

@@ -297,15 +297,15 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 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)) 🔀
+           | 0x11 t:<typeidx> async?:<async?>                    => (canon stream.cancel-read async? (core func)) 🔀
+           | 0x12 t:<typeidx> async?:<async?>                    => (canon stream.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)) 🔀
+           | 0x18 t:<typeidx> async?:<async?>                    => (canon future.cancel-read async? (core func)) 🔀
+           | 0x19 t:<typeidx> 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)) 🔀
            | 0x1c opts:<opts>                                    => (canon error.new opts (core func)) 🔀

design/mvp/CanonicalABI.md

@@ -3198,10 +3198,10 @@ an optional last value of the stream or future.
 
 For canonical definitions:
 ```wasm
-(canon stream.cancel-read $async? (core func $f))
-(canon stream.cancel-write $async? (core func $f))
-(canon future.cancel-read $async? (core func $f))
-(canon future.cancel-write $async? (core func $f))
+(canon stream.cancel-read $t $async? (core func $f))
+(canon stream.cancel-write $t $async? (core func $f))
+(canon future.cancel-read $t $async? (core func $f))
+(canon future.cancel-write $t $async? (core func $f))
 ```
 validation specifies:
 * `$f` is given type `(func (param i32) (result i32))`
@@ -3217,22 +3217,23 @@ cancel a `read` or `write` (and regain ownership of the passed buffer) is
 crucial since some languages will need to cancel reading or writing from
 within the synchronous context of a destructor.
 ```python
-async def canon_stream_cancel_read(sync, task, i):
-  return await cancel_async_copy(ReadableStreamHandle, sync, task, i)
+async def canon_stream_cancel_read(t, sync, task, i):
+  return await cancel_async_copy(ReadableStreamHandle, t, sync, task, i)
 
-async def canon_stream_cancel_write(sync, task, i):
-  return await cancel_async_copy(WritableStreamHandle, sync, task, i)
+async def canon_stream_cancel_write(t, sync, task, i):
+  return await cancel_async_copy(WritableStreamHandle, t, sync, task, i)
 
-async def canon_future_cancel_read(sync, task, i):
-  return await cancel_async_copy(ReadableFutureHandle, sync, task, i)
+async def canon_future_cancel_read(t, sync, task, i):
+  return await cancel_async_copy(ReadableFutureHandle, t, sync, task, i)
 
-async def canon_future_cancel_write(sync, task, i):
-  return await cancel_async_copy(WritableFutureHandle, sync, task, i)
+async def canon_future_cancel_write(t, sync, task, i):
+  return await cancel_async_copy(WritableFutureHandle, t, sync, task, i)
 
-async def cancel_async_copy(HandleT, sync, task, i):
+async def cancel_async_copy(HandleT, t, sync, task, i):
   trap_if(not task.inst.may_leave)
   h = task.inst.waitables.get(i)
   trap_if(not isinstance(h, HandleT))
+  trap_if(h.t != t)
   trap_if(not h.copying_buffer)
   if h.stream.closed():
     flat_results = [pack_async_copy_result(task, h.copying_buffer, h)]

design/mvp/Explainer.md

@@ -1403,15 +1403,15 @@ canon ::= ...
         | (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.cancel-read <typeidx> async? (core func <id>?)) 🔀
+        | (canon stream.cancel-write <typeidx> 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.cancel-read <typeidx> async? (core func <id>?)) 🔀
+        | (canon future.cancel-write <typeidx> async? (core func <id>?)) 🔀
         | (canon future.close-readable <typeidx> (core func <id>?)) 🔀
         | (canon future.close-writable <typeidx> (core func <id>?)) 🔀
         | (canon error.new <canonopt>* (core func <id>?))

design/mvp/canonical-abi/definitions.py

@@ -2029,22 +2029,23 @@ def pack_async_copy_result(task, buffer, h):
 
 ### 🔀 `canon {stream,future}.cancel-{read,write}`
 
-async def canon_stream_cancel_read(sync, task, i):
-  return await cancel_async_copy(ReadableStreamHandle, sync, task, i)
+async def canon_stream_cancel_read(t, sync, task, i):
+  return await cancel_async_copy(ReadableStreamHandle, t, sync, task, i)
 
-async def canon_stream_cancel_write(sync, task, i):
-  return await cancel_async_copy(WritableStreamHandle, sync, task, i)
+async def canon_stream_cancel_write(t, sync, task, i):
+  return await cancel_async_copy(WritableStreamHandle, t, sync, task, i)
 
-async def canon_future_cancel_read(sync, task, i):
-  return await cancel_async_copy(ReadableFutureHandle, sync, task, i)
+async def canon_future_cancel_read(t, sync, task, i):
+  return await cancel_async_copy(ReadableFutureHandle, t, sync, task, i)
 
-async def canon_future_cancel_write(sync, task, i):
-  return await cancel_async_copy(WritableFutureHandle, sync, task, i)
+async def canon_future_cancel_write(t, sync, task, i):
+  return await cancel_async_copy(WritableFutureHandle, t, sync, task, i)
 
-async def cancel_async_copy(HandleT, sync, task, i):
+async def cancel_async_copy(HandleT, t, sync, task, i):
   trap_if(not task.inst.may_leave)
   h = task.inst.waitables.get(i)
   trap_if(not isinstance(h, HandleT))
+  trap_if(h.t != t)
   trap_if(not h.copying_buffer)
   if h.stream.closed():
     flat_results = [pack_async_copy_result(task, h.copying_buffer, h)]

design/mvp/canonical-abi/run_tests.py

@@ -1245,7 +1245,7 @@ async def core_func(task, args):
     assert(ret == 0)
     result = int.from_bytes(mem[retp : retp+4], 'little', signed=False)
     assert(result == (wsi | 2**31))
-    [ret] = await canon_stream_cancel_write(True, task, wsi)
+    [ret] = await canon_stream_cancel_write(U8Type(), True, task, wsi)
     assert(ret == 0)
     [] = await canon_stream_close_writable(U8Type(), task, wsi, 0)
     return []
@@ -1523,7 +1523,7 @@ async def core_func(task, args):
     host_sink.set_remain(2)
     got = await host_sink.consume(2)
     assert(got == [0xa, 0xb])
-    [ret] = await canon_stream_cancel_write(True, task, wsi)
+    [ret] = await canon_stream_cancel_write(U8Type(), True, task, wsi)
     assert(ret == 2)
     [] = await canon_stream_close_writable(U8Type(), task, wsi, 0)
     host_sink.set_remain(100)
@@ -1538,7 +1538,7 @@ async def core_func(task, args):
     host_sink.set_remain(2)
     got = await host_sink.consume(2)
     assert(got == [1, 2])
-    [ret] = await canon_stream_cancel_write(False, task, wsi)
+    [ret] = await canon_stream_cancel_write(U8Type(), False, task, wsi)
     assert(ret == 2)
     [] = await canon_stream_close_writable(U8Type(), task, wsi, 0)
     host_sink.set_remain(100)
@@ -1550,7 +1550,7 @@ async def core_func(task, args):
     rsi = mem[retp]
     [ret] = await canon_stream_read(U8Type(), lower_opts, task, rsi, 0, 4)
     assert(ret == definitions.BLOCKED)
-    [ret] = await canon_stream_cancel_read(True, task, rsi)
+    [ret] = await canon_stream_cancel_read(U8Type(), True, task, rsi)
     assert(ret == 0)
     [] = await canon_stream_close_readable(U8Type(), task, rsi)
 
@@ -1561,7 +1561,7 @@ async def core_func(task, args):
     [ret] = await canon_stream_read(U8Type(), lower_opts, task, rsi, 0, 4)
     assert(ret == definitions.BLOCKED)
     host_source.eager_cancel.clear()
-    [ret] = await canon_stream_cancel_read(False, task, rsi)
+    [ret] = await canon_stream_cancel_read(U8Type(), False, task, rsi)
     assert(ret == definitions.BLOCKED)
     host_source.write([7,8])
     await asyncio.sleep(0)
@@ -1682,7 +1682,7 @@ async def core_func(task, args):
     assert(p1 == subi)
 
     await task.yield_(sync = False)
-    [ret] = await canon_future_cancel_read(True, task, rfi)
+    [ret] = await canon_future_cancel_read(U8Type(), True, task, rfi)
     assert(ret == 1)
     assert(mem[readp] == 43)