Allow same-instance stream/future writes when element is empty

Author: lukewagner

design/mvp/CanonicalABI.md

@@ -1213,7 +1213,8 @@ but in the opposite direction. Both are implemented by a single underlying
       self.pending_on_copy_done = on_copy_done
       return 'blocked'
     else:
-      trap_if(inst is self.pending_inst) # temporary
+      assert(self.t == src.t == dst.t)
+      trap_if(inst is self.pending_inst and self.t is not None) # temporary
       ncopy = min(src.remain(), dst.remain())
       assert(ncopy > 0)
       dst.write(src.read(ncopy))
@@ -1224,10 +1225,11 @@ but in the opposite direction. Both are implemented by a single underlying
       return 'done'
 ```
 Currently, there is a trap when both the `read` and `write` come from the same
-component instance, but this trapping condition will be removed in a subsequent
-release. The reason for this trap is that when lifting and lowering can alias
-the same memory, interleaving must be handled carefully. Future improvements to
-the Canonical ABI ([lazy lowering]) can greatly simplify this interleaving.
+component instance and there is a non-empty element type. This trap will be
+removed in a subsequent release; the reason for the trap is that when lifting
+and lowering can alias the same memory, interleavings can be complex and must
+be handled carefully. Future improvements to the Canonical ABI ([lazy lowering])
+can greatly simplify this interleaving and be more practical to implement.
 
 Given the above, we can define the `{Readable,Writable}StreamEnd` classes that
 are actually stored in the `waitables` table. The classes are almost entirely

design/mvp/canonical-abi/definitions.py

@@ -716,7 +716,8 @@ def copy(self, inst, buffer, on_partial_copy, on_copy_done, src, dst):
       self.pending_on_copy_done = on_copy_done
       return 'blocked'
     else:
-      trap_if(inst is self.pending_inst) # temporary
+      assert(self.t == src.t == dst.t)
+      trap_if(inst is self.pending_inst and self.t is not None) # temporary
       ncopy = min(src.remain(), dst.remain())
       assert(ncopy > 0)
       dst.write(src.read(ncopy))

design/mvp/canonical-abi/run_tests.py

@@ -1058,6 +1058,7 @@ def write(self, vs):
 
 class HostSink:
   stream: ReadableStream
+  t: ValType
   received: list[int]
   chunk: int
   write_remain: int
@@ -1066,6 +1067,7 @@ class HostSink:
 
   def __init__(self, stream, chunk, remain = 2**64):
     self.stream = stream
+    self.t = stream.t
     self.received = []
     self.chunk = chunk
     self.write_remain = remain
@@ -1756,10 +1758,12 @@ async def core_func(task, args):
 
 
 class HostFutureSink:
+  t: ValType
   v: Optional[any]
   has_v: asyncio.Event
 
-  def __init__(self):
+  def __init__(self, t):
+    self.t = t
     self.v = None
     self.has_v = asyncio.Event()
 
@@ -1824,7 +1828,7 @@ async def host_func(task, on_start, on_return, on_block):
     [future] = on_start()
     outgoing = HostFutureSource(U8Type())
     on_return([outgoing])
-    incoming = HostFutureSink()
+    incoming = HostFutureSink(U8Type())
     future.read(None, incoming, lambda:(), lambda why:())
     await on_block(incoming.has_v.wait())
     assert(incoming.v == 42)
@@ -1894,6 +1898,63 @@ async def core_func(task, args):
 
   await canon_lift(lift_opts, inst, FuncType([],[]), core_func, None, lambda:[], lambda _:(), host_on_block)
 
+async def test_self_empty():
+  inst = ComponentInstance()
+  mem = bytearray(24)
+  sync_opts = mk_opts(memory=mem, sync=True)
+  async_opts = mk_opts(memory=mem, sync=False)
+
+  ft = FuncType([],[])
+  async def core_func(task, args):
+    [seti] = await canon_waitable_set_new(task)
+
+    [packed] = await canon_future_new(None, task)
+    rfi,wfi = unpack_new_ends(packed)
+
+    [ret] = await canon_future_write(None, async_opts, task, wfi, 10000)
+    assert(ret == definitions.BLOCKED)
+
+    [ret] = await canon_future_read(None, async_opts, task, rfi, 20000)
+    result,n = unpack_result(ret)
+    assert(n == 1 and result == definitions.CLOSED)
+    [] = await canon_future_close_readable(None, task, rfi)
+
+    [] = await canon_waitable_join(task, wfi, seti)
+    [event] = await canon_waitable_set_wait(True, mem, task, seti, 0)
+    assert(event == EventCode.FUTURE_WRITE)
+    assert(mem[0] == wfi)
+    result,n = unpack_result(mem[4])
+    assert(result == definitions.CLOSED)
+    assert(n == 1)
+    [] = await canon_future_close_writable(None, task, wfi)
+
+    [packed] = await canon_stream_new(None, task)
+    rsi,wsi = unpack_new_ends(packed)
+    [ret] = await canon_stream_write(None, async_opts, task, wsi, 10000, 3)
+    assert(ret == definitions.BLOCKED)
+
+    [ret] = await canon_stream_read(None, async_opts, task, rsi, 2000, 1)
+    result,n = unpack_result(ret)
+    assert(n == 1 and result == definitions.COMPLETED)
+    [ret] = await canon_stream_read(None, async_opts, task, rsi, 2000, 4)
+    result,n = unpack_result(ret)
+    assert(n == 2 and result == definitions.COMPLETED)
+    [] = await canon_stream_close_readable(None, task, rsi)
+
+    [] = await canon_waitable_join(task, wsi, seti)
+    [event] = await canon_waitable_set_wait(True, mem, task, seti, 0)
+    assert(event == EventCode.STREAM_WRITE)
+    assert(mem[0] == wsi)
+    result,n = unpack_result(mem[4])
+    assert(result == definitions.CLOSED)
+    assert(n == 3)
+    [] = await canon_stream_close_writable(None, task, wsi)
+
+    [] = await canon_waitable_set_drop(task, seti)
+    return []
+
+  await canon_lift(sync_opts, inst, ft, core_func, None, lambda:[], lambda _:(), host_on_block)
+
 
 async def run_async_tests():
   await test_roundtrips()
@@ -1912,6 +1973,7 @@ async def run_async_tests():
   await test_wasm_to_wasm_stream_empty()
   await test_cancel_copy()
   await test_futures()
+  await test_self_empty()
 
 asyncio.run(run_async_tests())