Add {stream,future}.forward built-in

Co-authored-by: Roman Volosatovs <rvolosatovs@riseup.net>

Author: lukewagner

design/mvp/Binary.md

@@ -316,13 +316,15 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 0x12 t:<typeidx> async?:<async?>                    => (canon stream.cancel-write t async? (core func)) 🔀
            | 0x13 t:<typeidx>                                    => (canon stream.drop-readable t (core func)) 🔀
            | 0x14 t:<typeidx>                                    => (canon stream.drop-writable t (core func)) 🔀
+           | 0x2e t:<typeidx>                                    => (canon stream.forward 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 t:<typeidx> async?:<async?>                    => (canon future.cancel-read t async? (core func)) 🔀
            | 0x19 t:<typeidx> async?:<async?>                    => (canon future.cancel-write t async? (core func)) 🔀
            | 0x1a t:<typeidx>                                    => (canon future.drop-readable t (core func)) 🔀
            | 0x1b t:<typeidx>                                    => (canon future.drop-writable t (core func)) 🔀
+           | 0x2f t:<typeidx>                                    => (canon future.forward t (core func)) ➡️
            | 0x1c opts:<opts>                                    => (canon error-context.new opts (core func)) 📝
            | 0x1d opts:<opts>                                    => (canon error-context.debug-message opts (core func)) 📝
            | 0x1e                                                => (canon error-context.drop (core func)) 📝

design/mvp/CanonicalABI.md

@@ -56,6 +56,7 @@ specified here.
   * [`canon {stream,future}.{read,write}`](#-canon-streamfuturereadwrite) 🔀
   * [`canon {stream,future}.cancel-{read,write}`](#-canon-streamfuturecancel-readwrite) 🔀
   * [`canon {stream,future}.drop-{readable,writable}`](#-canon-streamfuturedrop-readablewritable) 🔀
+  * [`canon {stream,future}.forward`](#-canon-streamfutureforward) ➡️
   * [`canon thread.index`](#-canon-threadindex) 🧵
   * [`canon thread.new-indirect`](#-canon-threadnew-indirect) 🧵
   * [`canon thread.resume-later`](#-canon-threadresume-later) 🧵
@@ -1719,9 +1720,10 @@ cancellation. In the future, guest components may be given the same capability.
       self.notify(End.Result.CANCELLED)
 ```
 
-The `End.drop` method is called by `{stream,future}.drop-{readable,writable}` to
-update the `other` end's state and possibly set a pending notification for the
-other end, if doing so wouldn't clobber an already-pending notification.
+The `End.drop` method is called by `{stream,future}.drop-{readable,writable}`
+and `End.forward` to update the `other` end's state and possibly set a pending
+notification for the other end, if doing so wouldn't clobber an already-pending
+notification.
 ```python
   def drop(self):
     assert(not self.copying_or_cancelling())
@@ -1737,6 +1739,37 @@ other end, if doing so wouldn't clobber an already-pending notification.
     return self.state in { End.State.COPYING, End.State.CANCELLING_COPY }
 ```
 
+The `End.forward` function is called by `{stream,future}.forward` to efficiently
+forward all the values from a given readable end into a given writable end,
+propagating `DROPPED` results in both directions. This causes the given readable
+end and writable end to disappear, leaving only the `other` writable and
+readable ends, which are now linked together directly to form a single stream or
+future. If *both* of these formerly-separate ends have pending copy operations
+with pending `buffer`s, the forwarding operation performs a synchronous copy,
+leaving at most one pending buffer (the bigger of the two) and notifying one or
+both sides of the progress made. In the corner case where a stream or future's
+readable end is forwarded to its own writable end (creating a trivial recursive
+loop), there is no trap since the whole thing simply disappears.
+```python
+  def forward(src: End, dst: End):
+    if src.other is dst or src.other is None or dst.other is None:
+      src.drop()
+      dst.drop()
+    else:
+      writable_end = src.other
+      readable_end = dst.other
+      writable_end.other = readable_end
+      readable_end.other = writable_end
+      if writable_end.buffer is not None and readable_end.buffer is not None:
+        if readable_end.buffer.remain() > writable_end.buffer.remain():
+          bigger_end, smaller_end = readable_end, writable_end
+        else:
+          bigger_end, smaller_end = writable_end, readable_end
+        buffer = smaller_end.buffer
+        smaller_end.buffer = None
+        smaller_end.copy(buffer)
+```
+
 Next, the intermediate `{Stream,Future}End` base classes are defined to implement
 the `{Stream,Future}End.notify` methods that are called by the
 `End.{copy,cancel,drop}` methods above. `notify`'s behavior does not exhibit the
@@ -4451,6 +4484,46 @@ def drop(EndT, stream_or_future_t, i):
 ```
 
 
+### ➡️ `canon {stream,future}.forward`
+
+For canonical definitions:
+```wat
+(canon stream.forward $stream_t (core func $forward))
+(canon future.forward $future_t (core func $forward))
+```
+validation specifies:
+* `$forward` is given type `(func (param $ri i32) (param $wi i32))`
+* `$stream_t`/`$future_t` must be a type of the form `(stream $t?)`/`(future $t?)`
+
+Calling `$forward` removes the readable and writable ends at the given indices,
+after checking that all the types match, the ends are in the `IDLE` state, and
+the ends are not currently part of a waitable set. Then the readable end is
+forwarded into the writable end as defined by `End.forward` above.
+```python
+def canon_stream_forward(stream_t, ri, wi):
+  return forward(ReadableStreamEnd, WritableStreamEnd, stream_t, ri, wi)
+
+def canon_future_forward(future_t, ri, wi):
+  return forward(ReadableFutureEnd, WritableFutureEnd, future_t, ri, wi)
+
+def forward(ReadableEndT, WritableEndT, stream_or_future_t, ri, wi):
+  inst = current_instance()
+  trap_if(not inst.may_leave)
+  readable_end = inst.handles.remove(ri)
+  trap_if(not isinstance(readable_end, ReadableEndT))
+  trap_if(readable_end.t != stream_or_future_t.t)
+  trap_if(readable_end.state != End.State.IDLE)
+  trap_if(readable_end.in_waitable_set())
+  writable_end = inst.handles.remove(wi)
+  trap_if(not isinstance(writable_end, WritableEndT))
+  trap_if(writable_end.t != stream_or_future_t.t)
+  trap_if(writable_end.state != End.State.IDLE)
+  trap_if(writable_end.in_waitable_set())
+  End.forward(readable_end, writable_end)
+  return []
+```
+
+
 ### 🧵 `canon thread.index`
 
 For a canonical definition:

design/mvp/Concurrency.md

@@ -577,7 +577,7 @@ or `future`. When *producing* a `stream` or `future` value as a parameter (of
 an import call) or result (of an export call), the producer can *transfer
 ownership* of a readable end that it has either been given by the outside world
 or freshly created via `{stream,future}.new` (which also return a fresh paired
-writable end that is permanently owned by the calling component instance).
+writable end).
 
 Based on this, `stream<T>` and `future<T>` values can be passed between
 functions as if they were synchronous `list<T>` and `T` values, resp. For
@@ -619,6 +619,13 @@ without requiring an explicit `future` return type. Thus, a function like
 which point the caller receives the readable end of a `future` that, when
 successfully read, conveys the completion of a second event.
 
+Given the readable end of one stream/future and the writable end of another, the
+`{stream,future}.forward` built-ins can be called to efficiently forward all
+remaining values from the readable end into the writable end, avoiding any
+intermediate copies. Doing so relinquishes ownership of both handles, allowing
+the calling component instance to be eagerly torn down while the forwarding is
+in progress.
+
 The [Stream and Future State] section describes the runtime state maintained for
 streams and futures by the Canonical ABI.
 
@@ -1478,7 +1485,6 @@ specified, the following features are being considered for addition to complete
 the concurrency story:
 * remove the temporary trap mentioned above that occurs when a `read` and
   `write` of a stream/future happen from within the same component instance
-* zero-copy forwarding/splicing
 * allow `async` functions using the stackful ABI to be notified of
   cancellation
 * allow the `stream<char>` type to validate; make it use `string-encoding`

design/mvp/Explainer.md

@@ -75,6 +75,7 @@ shipped as part of a future WASI Developer Preview release:
 * 📝: the `error-context` type
 * 🔗: canonical interface names
 * 🐘: [memory64]
+* ➡️: `stream.forward` and `future.forward` built-ins
 
 
 ## Grammar
@@ -1573,13 +1574,15 @@ canon ::= ...
         | (canon stream.cancel-write <typeidx> async? (core func <id>?)) 🔀
         | (canon stream.drop-readable <typeidx> (core func <id>?)) 🔀
         | (canon stream.drop-writable <typeidx> (core func <id>?)) 🔀
+        | (canon stream.forward <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 <typeidx> async? (core func <id>?)) 🔀
         | (canon future.cancel-write <typeidx> async? (core func <id>?)) 🔀
         | (canon future.drop-readable <typeidx> (core func <id>?)) 🔀
         | (canon future.drop-writable <typeidx> (core func <id>?)) 🔀
+        | (canon future.forward <typeidx> (core func <id>?)) ➡️
         | (canon thread.index (core func <id>?)) 🧵
         | (canon thread.new-indirect core-prefix(<core:typeidx>) core-prefix(<core:tableidx>) (core func <id>?)) 🧵
         | (canon thread.resume-later (core func <id>?)) 🧵
@@ -2145,6 +2148,25 @@ already been dropped.
 For details, see [Streams and Futures] in the concurrency explainer and
 [`canon_stream_drop_readable`] in the Canonical ABI explainer.
 
+###### ➡️ `stream.forward` and `future.forward`
+
+| Synopsis                                       |                                                                            |
+| ---------------------------------------------- | -------------------------------------------------------------------------- |
+| Approximate WIT signature for `stream.forward` | `func<stream<T?>>(r: readable-stream-end<T?>, w: writable-stream-end<T?>)` |
+| Approximate WIT signature for `future.forward` | `func<future<T?>>(r: readable-future-end<T?>, w: writable-future-end<T?>)` |
+| Canonical ABI signature                        | `[ri:i32 wi:i32] -> []`                                                    |
+
+The `{stream,future}.forward` built-ins remove the given readable and writable
+ends from the caller's handle table and logically forward everything from the
+readable end into the writable end (propagating drops in both directions), but
+do so without an intermediate copy. The call traps if either end has a
+mismatched direction or element type, is in the middle of a read or write, is a
+member of a waitable set, or has received its final `dropped` or, for futures,
+`completed` result.
+
+For details, see [Streams and Futures] in the concurrency explainer and
+[`canon_stream_forward`] in the Canonical ABI explainer.
+
 ###### 🧵 `thread.index`
 
 | Synopsis                   |                 |
@@ -3377,6 +3399,7 @@ For some use-case-focused, worked examples, see:
 [`canon_future_read`]: CanonicalABI.md#-canon-streamfuturereadwrite
 [`canon_stream_cancel_read`]: CanonicalABI.md#-canon-streamfuturecancel-readwrite
 [`canon_stream_drop_readable`]: CanonicalABI.md#-canon-streamfuturedrop-readablewritable
+[`canon_stream_forward`]: CanonicalABI.md#-canon-streamfutureforward
 [`canon_subtask_cancel`]: CanonicalABI.md#-canon-subtaskcancel
 [`canon_subtask_drop`]: CanonicalABI.md#-canon-subtaskdrop
 [`canon_resource_new`]: CanonicalABI.md#canon-resourcenew

design/mvp/canonical-abi/definitions.py

@@ -4,9 +4,8 @@
 
 ### Boilerplate
 
-from __future__ import annotations
 from dataclasses import dataclass
-from typing import Any, Optional, Callable, TypeVar, Generic, Literal
+from typing import Optional, Callable, Literal
 from enum import Enum, IntEnum
 import math
 import struct
@@ -988,6 +987,24 @@ def drop(self):
   def copying_or_cancelling(self):
     return self.state in { End.State.COPYING, End.State.CANCELLING_COPY }
 
+  def forward(src: End, dst: End):
+    if src.other is dst or src.other is None or dst.other is None:
+      src.drop()
+      dst.drop()
+    else:
+      writable_end = src.other
+      readable_end = dst.other
+      writable_end.other = readable_end
+      readable_end.other = writable_end
+      if writable_end.buffer is not None and readable_end.buffer is not None:
+        if readable_end.buffer.remain() > writable_end.buffer.remain():
+          bigger_end, smaller_end = readable_end, writable_end
+        else:
+          bigger_end, smaller_end = writable_end, readable_end
+        buffer = smaller_end.buffer
+        smaller_end.buffer = None
+        smaller_end.copy(buffer)
+
 class StreamEnd(End):
   def notify(self, result: End.Result, progress = 0):
     def stream_event():
@@ -2498,6 +2515,30 @@ def drop(EndT, stream_or_future_t, i):
   end.drop()
   return []
 
+### ➡️ `canon {stream,future}.forward`
+
+def canon_stream_forward(stream_t, ri, wi):
+  return forward(ReadableStreamEnd, WritableStreamEnd, stream_t, ri, wi)
+
+def canon_future_forward(future_t, ri, wi):
+  return forward(ReadableFutureEnd, WritableFutureEnd, future_t, ri, wi)
+
+def forward(ReadableEndT, WritableEndT, stream_or_future_t, ri, wi):
+  inst = current_instance()
+  trap_if(not inst.may_leave)
+  readable_end = inst.handles.remove(ri)
+  trap_if(not isinstance(readable_end, ReadableEndT))
+  trap_if(readable_end.t != stream_or_future_t.t)
+  trap_if(readable_end.state != End.State.IDLE)
+  trap_if(readable_end.in_waitable_set())
+  writable_end = inst.handles.remove(wi)
+  trap_if(not isinstance(writable_end, WritableEndT))
+  trap_if(writable_end.t != stream_or_future_t.t)
+  trap_if(writable_end.state != End.State.IDLE)
+  trap_if(writable_end.in_waitable_set())
+  End.forward(readable_end, writable_end)
+  return []
+
 ### 🧵 `canon thread.index`
 
 def canon_thread_index():

design/mvp/canonical-abi/run_tests.py

@@ -1651,7 +1651,7 @@ def core_func(args):
   assert(host_reader2.received == [11,12,13,14,15,16,17,18])
 
 
-def test_stream_forward():
+def test_transfer_readable_end():
   host_writer = HostWriter(U8Type(), [1,2,3,4], chunk=4)
   def on_start():
     return [host_writer.readable_end]
@@ -1674,6 +1674,97 @@ def core_func(args):
   assert(host_writer.readable_end is return_value)
 
 
+def test_forward():
+  store = Store()
+  mem = bytearray(32)
+  opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True)
+  inst = ComponentInstance(store)
+  st = StreamType(U8Type())
+  ft = FutureType(U8Type())
+
+  def core_func(args):
+    def new_stream(t = st):
+      [packed] = canon_stream_new(t)
+      return unpack_new_ends(packed)
+    def new_future():
+      [packed] = canon_future_new(ft)
+      return unpack_new_ends(packed)
+
+    rsi,wsi = new_stream()
+    [] = canon_stream_forward(st, rsi, wsi)
+
+    rsi1,wsi1 = new_stream()
+    rsi2,wsi2 = new_stream()
+    [] = canon_stream_drop_readable(st, rsi2)
+    [] = canon_stream_forward(st, rsi1, wsi2)
+    [ret] = canon_stream_write(st, opts, wsi1, 0, 4)
+    result,n = unpack_result(ret)
+    assert(n == 0 and result == End.Result.DROPPED)
+    [] = canon_stream_drop_writable(st, wsi1)
+
+    rsi1,wsi1 = new_stream()
+    rsi2,wsi2 = new_stream()
+    [] = canon_stream_forward(st, rsi1, wsi2)
+    mem[0:4] = b'\x01\x02\x03\x04'
+    [ret] = canon_stream_write(st, opts, wsi1, 0, 4)
+    assert(ret == definitions.BLOCKED)
+    [ret] = canon_stream_read(st, opts, rsi2, 8, 4)
+    result,n = unpack_result(ret)
+    assert(n == 4 and result == End.Result.COMPLETED)
+    assert(mem[8:12] == b'\x01\x02\x03\x04')
+
+    rsi1,wsi1 = new_stream()
+    rsi2,wsi2 = new_stream()
+    [ret] = canon_stream_read(st, opts, rsi2, 8, 4)
+    assert(ret == definitions.BLOCKED)
+    [] = canon_stream_forward(st, rsi1, wsi2)
+    mem[0:4] = b'\x05\x06\x07\x08'
+    [ret] = canon_stream_write(st, opts, wsi1, 0, 4)
+    result,n = unpack_result(ret)
+    assert(n == 4 and result == End.Result.COMPLETED)
+    assert(mem[8:12] == b'\x05\x06\x07\x08')
+    [seti] = canon_waitable_set_new()
+    [] = canon_waitable_join(rsi2, seti)
+    [event] = canon_waitable_set_wait(MemInst(mem, 'i32'), seti, 16)
+    assert(event == EventCode.STREAM_READ)
+    assert(mem[16] == rsi2)
+    result,n = unpack_result(mem[20])
+    assert(n == 4 and result == End.Result.COMPLETED)
+
+    rfi1,wfi1 = new_future()
+    rfi2,wfi2 = new_future()
+    [ret] = canon_future_read(ft, opts, rfi2, 8)
+    assert(ret == definitions.BLOCKED)
+    [] = canon_future_forward(ft, rfi1, wfi2)
+    mem[0] = 42
+    [ret] = canon_future_write(ft, opts, wfi1, 0)
+    assert(ret == End.Result.COMPLETED)
+    assert(mem[8] == 42)
+
+    rfi1,wfi1 = new_future()
+    rfi2,wfi2 = new_future()
+    mem[0] = 43
+    [ret] = canon_future_write(ft, opts, wfi1, 0)
+    assert(ret == definitions.BLOCKED)
+    [ret] = canon_future_read(ft, opts, rfi2, 8)
+    assert(ret == definitions.BLOCKED)
+    [] = canon_future_forward(ft, rfi1, wfi2)
+    assert(mem[8] == 43)
+
+    rfi1,wfi1 = new_future()
+    rfi2,wfi2 = new_future()
+    [] = canon_future_drop_readable(ft, rfi2)
+    [] = canon_future_forward(ft, rfi1, wfi2)
+    [ret] = canon_future_write(ft, opts, wfi1, 0)
+    assert(ret == End.Result.DROPPED)
+    [] = canon_future_drop_writable(ft, wfi1)
+
+    return []
+
+  caller_ft = FuncType([], [], async_ = True)
+  lift_and_run(mk_opts(), inst, caller_ft, core_func, lambda:[], lambda _:())
+
+
 def test_receive_own_stream():
   store = Store()
   inst = ComponentInstance(store)
@@ -2974,7 +3065,8 @@ def on_resolve(v):
 test_sync_using_wait()
 test_eager_stream_completion()
 test_async_stream_ops()
-test_stream_forward()
+test_transfer_readable_end()
+test_forward()
 test_receive_own_stream()
 test_host_partial_reads_writes()
 test_wasm_to_wasm_stream()