add stream.forward and future.forward builtins

design/mvp/Binary.md

@@ -312,13 +312,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)) 🔀
+           | 0x2e t:<typeidx>                                    => (canon stream.forward t (core func)) ⏩
            | 0x11 t:<typeidx> async?:<async?>                    => (canon stream.cancel-read t async? (core func)) 🔀
            | 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)) 🔀
            | 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)) 🔀
+           | 0x2f t:<typeidx>                                    => (canon future.forward t (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)) 🔀

design/mvp/CanonicalABI.md

@@ -56,6 +56,7 @@ specified here.
   * [`canon {stream,future}.new`](#-canon-streamfuturenew) 🔀
   * [`canon stream.{read,write}`](#-canon-streamreadwrite) 🔀
   * [`canon future.{read,write}`](#-canon-futurereadwrite) 🔀
+  * [`canon {stream,future}.forward`](#-canon-streamfutureforward) ⏩
   * [`canon {stream,future}.cancel-{read,write}`](#-canon-streamfuturecancel-readwrite) 🔀
   * [`canon {stream,future}.drop-{readable,writable}`](#-canon-streamfuturedrop-readablewritable) 🔀
   * [`canon thread.index`](#-canon-threadindex) 🧵
@@ -1751,7 +1752,7 @@ class BufferGuestImpl(Buffer):
   def is_zero_length(self):
     return self.length == 0
 
-class ReadableBufferGuestImpl(BufferGuestImpl):
+class ReadableBufferGuestImpl(BufferGuestImpl, ReadableBuffer):
   def read(self, n):
     assert(n <= self.remain())
     if self.t:
@@ -1878,6 +1879,7 @@ Introducing `SharedStreamImpl` in chunks, starting with the fields and initializ
 ```python
 class SharedStreamImpl(ReadableStream, WritableStream):
   dropped: bool
+  forward: Optional[ReadableStream]
   pending_inst: Optional[ComponentInstance]
   pending_buffer: Optional[Buffer]
   pending_on_copy: Optional[OnCopy]
@@ -1886,6 +1888,7 @@ class SharedStreamImpl(ReadableStream, WritableStream):
   def __init__(self, t):
     self.t = t
     self.dropped = False
+    self.forward = None
     self.reset_pending()
 
   def reset_pending(self):
@@ -1896,23 +1899,39 @@ class SharedStreamImpl(ReadableStream, WritableStream):
     self.pending_buffer = buffer
     self.pending_on_copy = on_copy
     self.pending_on_copy_done = on_copy_done
+
+  def take_pending(self):
+    pending = (self.pending_inst, self.pending_buffer,
+               self.pending_on_copy, self.pending_on_copy_done)
+    self.reset_pending()
+    return pending
 ```
 If set, the `pending_*` fields record the `Buffer` and `OnCopy*` callbacks of a
 `read` or `write` that is waiting to rendezvous with a complementary `write` or
-`read`. Dropping the readable or writable end of a stream or cancelling a
-`read` or `write` notifies any pending `read` or `write` via its `OnCopyDone`
-callback:
+
+The ⏩ `forward` field is set once this stream's *writable* end has been passed
+to [`stream.forward`](Explainer.md#-streamforward-and-futureforward), in which case this stream
+has been permanently replaced by the source stream given to `stream.forward`
+and all subsequent operations are simply delegated to that stream.
+
+Dropping the readable or writable end of a stream or cancelling a `read` or
+`write` notifies any pending `read` or `write` via its `OnCopyDone` callback:
 ```python
   def reset_and_notify_pending(self, result):
     pending_on_copy_done = self.pending_on_copy_done
     self.reset_pending()
     pending_on_copy_done(result)
 
   def cancel(self):
-    self.reset_and_notify_pending(CopyResult.CANCELLED)
+    if self.forward:
+      self.forward.cancel()
+    else:
+      self.reset_and_notify_pending(CopyResult.CANCELLED)
 
   def drop(self):
-    if not self.dropped:
+    if self.forward:
+      self.forward.drop()
+    elif not self.dropped:
       self.dropped = True
       if self.pending_buffer:
         self.reset_and_notify_pending(CopyResult.DROPPED)
@@ -1940,7 +1959,9 @@ copy without blocking. In the final special case where the pending writer has a
 zero-length buffer, the writer is notified, but the reader remains blocked:
 ```python
   def read(self, inst, dst_buffer, on_copy, on_copy_done):
-    if self.dropped:
+    if self.forward:
+      self.forward.read(inst, dst_buffer, on_copy, on_copy_done)
+    elif self.dropped:
       on_copy_done(CopyResult.DROPPED)
     elif not self.pending_buffer:
       self.set_pending(inst, dst_buffer, on_copy, on_copy_done)
@@ -2087,17 +2108,25 @@ progress (since at most 1 value is copied) and the given `Buffer` must have
 `remain() == 1`.
 
 Introducing `SharedFutureImpl` in chunks, the first part is exactly
-symmetric to `SharedStreamImpl` in how initialization and cancellation work:
+symmetric to `SharedStreamImpl` in how initialization and cancellation work.
+In particular, the ⏩ `forward` field works just like its stream counterpart
+(described in [Stream State](#stream-state) above): once set (by
+[`future.forward`](Explainer.md#-streamforward-and-futureforward)), this
+future has been permanently replaced by the source future given to
+`future.forward` and all subsequent operations are simply delegated to that
+future:
 ```python
 class SharedFutureImpl(ReadableFuture, WritableFuture):
   dropped: bool
+  forward: Optional[ReadableFuture]
   pending_inst: Optional[ComponentInstance]
   pending_buffer: Optional[Buffer]
   pending_on_copy_done: Optional[OnCopyDone]
 
   def __init__(self, t):
     self.t = t
     self.dropped = False
+    self.forward = None
     self.reset_pending()
 
   def reset_pending(self):
@@ -2108,23 +2137,33 @@ class SharedFutureImpl(ReadableFuture, WritableFuture):
     self.pending_buffer = buffer
     self.pending_on_copy_done = on_copy_done
 
+  def take_pending(self):
+    pending = (self.pending_inst, self.pending_buffer, self.pending_on_copy_done)
+    self.reset_pending()
+    return pending
+
   def reset_and_notify_pending(self, result):
     pending_on_copy_done = self.pending_on_copy_done
     self.reset_pending()
     pending_on_copy_done(result)
 
   def cancel(self):
-    self.reset_and_notify_pending(CopyResult.CANCELLED)
+    if self.forward:
+      self.forward.cancel()
+    else:
+      self.reset_and_notify_pending(CopyResult.CANCELLED)
 ```
 Dropping works the same in futures as in streams, except that a future
 writable end cannot be dropped without having written a value. This is guarded
 by `WritableFutureEnd.drop` so it can be asserted here:
 ```python
   def drop(self):
-    if not self.dropped:
+    if self.forward:
+      self.forward.drop()
+    elif not self.dropped:
       self.dropped = True
       if self.pending_buffer:
-        assert(isinstance(self.pending_buffer, WritableBuffer))
+        assert(isinstance(self.pending_buffer, ReadableBuffer))
         self.reset_and_notify_pending(CopyResult.DROPPED)
 ```
 Lastly, `read` and `write` work mostly like streams, but simplified based on
@@ -2133,6 +2172,9 @@ that, as mentioned above, only the writable end can observe that the readable
 end was dropped before receiving a value.
 ```python
   def read(self, inst, dst_buffer, on_copy_done):
+    if self.forward:
+      self.forward.read(inst, dst_buffer, on_copy_done)
+      return
     assert(not self.dropped and dst_buffer.remain() == 1)
     if not self.pending_buffer:
       self.set_pending(inst, dst_buffer, on_copy_done)
@@ -4762,6 +4804,97 @@ synchronously and returning either the progress made or `BLOCKED`.
 ```
 
 
+### ⏩ `canon {stream,future}.forward`
+
+For canonical definitions:
+```wat
+(canon stream.forward $stream_t (core func $f))
+(canon future.forward $future_t (core func $f))
+```
+validation specifies:
+* `$f` is given type `(func (param i32 i32))`
+* `$stream_t` must be a type of the form `(stream $t?)`
+* `$future_t` must be a type of the form `(future $t?)`
+
+Calling `$f` forwards *all* remaining elements of the readable stream end at
+index `$ri` (resp., the value of the readable future end at index `$ri`) into
+the writable end at index `$wi` of another stream (resp., future) of the same
+type. `{stream,future}.forward` *transfers* both ends out of the calling
+component instance: the ends are removed from the `handles` table, `$f`
+returns immediately with no result and no event is ever delivered.
+
+The `canon_stream_forward` and `canon_future_forward` functions share a single
+implementation, `forward`, that sets the destination's `forward` field
+(defined in [Stream State](#stream-state) and [Future State](#future-state)
+above), permanently delegating every subsequent `read`, `cancel` and `drop` of
+the destination to the source, and transfers any consumer `read` already
+blocked on the destination to the source (performing rendezvous immediately if
+the source has a pending `write`). If the destination's readable end has
+already been dropped, there is nothing left to forward into, and so the source
+is simply dropped, notifying the source's writer:
+```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_source(shared):
+  while isinstance(shared, SharedStreamImpl | SharedFutureImpl) and shared.forward:
+    shared = shared.forward
+  return shared
+
+def forward(ReadableEndT, WritableEndT, stream_or_future_t, ri, wi):
+  inst = current_instance()
+  trap_if(not inst.may_leave)
+  r = inst.handles.get(ri)
+  trap_if(not isinstance(r, ReadableEndT))
+  trap_if(r.shared.t != stream_or_future_t.t)
+  trap_if(r.state != CopyState.IDLE)
+  trap_if(r.in_waitable_set())
+  w = inst.handles.get(wi)
+  trap_if(not isinstance(w, WritableEndT))
+  trap_if(w.shared.t != stream_or_future_t.t)
+  trap_if(w.state != CopyState.IDLE)
+  trap_if(w.in_waitable_set())
+  trap_if(forward_source(r.shared) is w.shared)
+  assert(not contains_borrow(stream_or_future_t))
+
+  inst.handles.remove(ri)
+  inst.handles.remove(wi)
+
+  if w.shared.dropped:
+    r.shared.drop()
+  else:
+    w.shared.forward = r.shared
+    if w.shared.pending_buffer:
+      pending = w.shared.take_pending()
+      r.shared.read(*pending)
+  return []
+```
+Since `{stream,future}.forward` transfers both ends out of the component
+instance, it traps if either end is currently in a waitable set, exactly as
+when a readable end is transferred to another component (in `lift_async_value`
+above). Forwards may be chained, but a forward that would (transitively) make
+a stream or future its own source traps; the `forward_source` helper follows
+an existing chain of forwards to the stream or future that ultimately serves
+as the source of a given stream or future.
+
+Once `{stream,future}.forward` returns, the source and destination have been
+fused into one: the producer and consumer on either side observe exactly what
+they would observe if the source's readable end had been transferred directly
+to the consumer in place of the destination's readable end. Elements
+rendezvous end-to-end with no intermediate buffering (as described in
+[Stream State](#stream-state) above) and drops propagate in both directions:
+when the source stream reaches its end (resp., the source future's value is
+written), the consumer observes it and, when the consumer drops the
+destination's readable end, the drop is delegated to the source where the
+producer observes it. This gives a host the opportunity to remove the calling
+component from the copy path entirely which, in chains of forwarding
+components, allows forwarding state (and even whole component instances that
+are no longer otherwise reachable) to be eagerly torn down.
+
+
 ### 🔀 `canon {stream,future}.cancel-{read,write}`
 
 For canonical definitions:

design/mvp/Concurrency.md

@@ -606,6 +606,18 @@ write in the future) can be queried and signalled by performing a `0`-length
 read or write (see the [Stream State] section in the Canonical ABI explainer
 for details).
 
+⏩ When a component wants to pipe the contents of one stream (or the value of
+one future) into another without an intermediate copy, it can use the
+`stream.forward` (resp., `future.forward`) built-in, which is given the
+readable end of a source and the writable end of a destination of the same
+type and *transfers* both ends out of the calling component, returning
+immediately and forwarding elements until the source stream ends or the
+destination's readable end is dropped, with the resulting end of the stream
+or drop propagated in both directions. Since the calling component keeps no
+handle on (and receives no notification of) the outcome, the runtime can fuse
+the source and destination together and remove the calling component from the
+path entirely.
+
 As a temporary limitation, if a `read` and `write` for a single stream or
 future occur from within the same component and the element type is a
 non-empty, non-number type, there is a trap. In the future this limitation will
@@ -1501,7 +1513,9 @@ 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
+* a built-in complementing ⏩ `{stream,future}.forward` that forwards a
+  bounded number of stream elements with zero copies while reporting the
+  outcome back to the caller
 * allow the `stream<char>` type to validate; make it use `string-encoding`
   and not split code points
 * add built-ins providing guest code more control over its containing

design/mvp/Explainer.md

@@ -69,6 +69,7 @@ shipped as part of a future WASI Developer Preview release:
 * 🪺: nested namespaces and packages in import/export names
 * 🚝: enabling more canonical ABI options on more async-related builtins
 * 🚟: using `async` with `canon lift` without `callback` (stackful lift)
+* ⏩: the `stream.forward` and `future.forward` built-ins
 * 🧵: threading built-ins
   * 🧵②: [shared-everything-threads]-based threading built-ins
 * 🔧: fixed-length lists
@@ -1569,13 +1570,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.forward <typeidx> (core func <id>?)) ⏩
         | (canon stream.cancel-read <typeidx> async? (core func <id>?)) 🔀
         | (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 future.new <typeidx> (core func <id>?)) 🔀
         | (canon future.read <typeidx> <canonopt>* (core func <id>?)) 🔀
         | (canon future.write <typeidx> <canonopt>* (core func <id>?)) 🔀
+        | (canon future.forward <typeidx> (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>?)) 🔀
@@ -2104,6 +2107,50 @@ by the `enum` definition above.
 For details, see [Streams and Futures] in the concurrency explainer and
 [`canon_future_read`] 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                        | `[readable-end:i32 writable-end:i32] -> []`                                 |
+
+The `stream.forward` built-in forwards *all* remaining elements from the
+readable end `r` of one stream into the writable end `w` of another stream and
+then propagates the resulting end of the stream or drop in both directions.
+Analogously, the `future.forward` built-in forwards the value of the future
+with readable end `r` into the future with writable end `w`, propagating
+drops in both directions. `{stream,future}.forward` *transfers* both ends
+out of the calling component: `r` and `w` are removed from the component
+instance's table, `{stream,future}.forward` returns immediately without a
+result, and no event is ever delivered. As with
+other transfers of stream and future ends, `{stream,future}.forward` traps if
+`r` or `w` is currently in a [waitable set]. A `{stream,future}.forward`
+cannot be cancelled and, since it never blocks the caller, has no `async`
+immediate. `{stream,future}.forward` takes no `canonopt`s, since no elements
+pass through the calling component's linear memory.
+
+`{stream,future}.forward` fuses the source and destination into one: the
+destination is replaced by the source, as if `r` had been transferred directly
+to the consumer in place of the destination's readable end. Elements flow
+directly from the source's producer to the destination's consumer: there is
+no intermediate buffering, backpressure is end-to-end and elements are copied
+directly from the producer's buffer into the consumer's buffer. When the
+source stream reaches its end (or the source future's value is written), the
+consumer observes it after receiving all forwarded elements. Symmetrically,
+if the consumer drops the destination's readable end, the producer observes
+`dropped`. Since the calling component gives up both ends and receives no
+notification of the outcome, the host can remove the calling component from
+the path entirely (in chains of forwarding components, this allows forwarding
+state and even whole component instances that are no longer otherwise
+reachable to be eagerly torn down).
+
+Forwards may be chained, and a `{stream,future}.forward` that would make a
+stream or future (transitively) its own source traps.
+
+For details, see [Streams and Futures] in the concurrency explainer and
+[`canon_stream_forward`] in the Canonical ABI explainer.
+
 ###### 🔀 `stream.cancel-read`, `stream.cancel-write`, `future.cancel-read`, and `future.cancel-write`
 
 | Synopsis                                            |                                                                               |
@@ -3368,6 +3415,7 @@ For some use-case-focused, worked examples, see:
 [`canon_stream_read`]: CanonicalABI.md#-canon-streamreadwrite
 [`canon_future_read`]: CanonicalABI.md#-canon-futurereadwrite
 [`canon_future_write`]: CanonicalABI.md#-canon-futurereadwrite
+[`canon_stream_forward`]: CanonicalABI.md#-canon-streamfutureforward
 [`canon_stream_cancel_read`]: CanonicalABI.md#-canon-streamfuturecancel-readwrite
 [`canon_stream_drop_readable`]: CanonicalABI.md#-canon-streamfuturedrop-readablewritable
 [`canon_subtask_cancel`]: CanonicalABI.md#-canon-subtaskcancel