From 574e9c16d3ee4711a68b1c44b5314298b7f514a2 Mon Sep 17 00:00:00 2001 From: Roman Volosatovs Date: Mon, 24 Aug 2026 16:29:53 +0200 Subject: [PATCH 1/2] CABI: allow dropping a readable future end with a pending write The Explainer specifies that a component may call `future.drop-readable` before reading a value to signal loss of interest, in which case a blocked `future.write` completes with `future-write-result.dropped`. But `SharedFutureImpl.drop` asserted that any pending buffer is a `WritableBuffer`, i.e. a pending *read*, while the only pending buffer actually possible at this point is the buffer of a pending *write*: a reader cannot drop its end while its own read is pending (`CopyEnd.drop` traps while copying) and a writer cannot drop its end before the write resolves (`WritableFutureEnd.drop` traps unless DONE). Exercising this spec-legal path failed the assert instead of notifying the writer with DROPPED. Flip the assert to expect a `ReadableBuffer` (the source buffer of the pending write), make `ReadableBufferGuestImpl` actually subclass `ReadableBuffer` (it previously only derived `BufferGuestImpl`, unlike `WritableBufferGuestImpl`, which does mix in `WritableBuffer`), and add a test reproducing the scenario. Signed-off-by: Roman Volosatovs Assisted-by: claude:claude-fable-5 --- design/mvp/CanonicalABI.md | 4 +-- design/mvp/canonical-abi/definitions.py | 4 +-- design/mvp/canonical-abi/run_tests.py | 37 +++++++++++++++++++++++++ 3 files changed, 41 insertions(+), 4 deletions(-) diff --git a/design/mvp/CanonicalABI.md b/design/mvp/CanonicalABI.md index 0c222b79..1cb97a3f 100644 --- a/design/mvp/CanonicalABI.md +++ b/design/mvp/CanonicalABI.md @@ -1751,7 +1751,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: @@ -2124,7 +2124,7 @@ by `WritableFutureEnd.drop` so it can be asserted here: if 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 diff --git a/design/mvp/canonical-abi/definitions.py b/design/mvp/canonical-abi/definitions.py index 99fb6fc6..807cfae8 100644 --- a/design/mvp/canonical-abi/definitions.py +++ b/design/mvp/canonical-abi/definitions.py @@ -962,7 +962,7 @@ def remain(self): 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: @@ -1158,7 +1158,7 @@ def drop(self): if 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) def read(self, inst, dst_buffer, on_copy_done): diff --git a/design/mvp/canonical-abi/run_tests.py b/design/mvp/canonical-abi/run_tests.py index b4bd0c34..30573d41 100644 --- a/design/mvp/canonical-abi/run_tests.py +++ b/design/mvp/canonical-abi/run_tests.py @@ -2308,6 +2308,42 @@ def core_func(args): lift_and_run(lift_opts, inst, caller_ft, core_func, lambda:[], lambda _:()) +def test_future_drop_readable_with_pending_write(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + future_t = FutureType(U8Type()) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_future_new(future_t) + rfi,wfi = unpack_new_ends(packed) + + mem[0] = 42 + [ret] = canon_future_write(future_t, opts, wfi, 0) + assert(ret == definitions.BLOCKED) + + # The reader may drop its end before reading a value; the blocked write + # is notified that the readable end was dropped. + [] = canon_future_drop_readable(future_t, rfi) + retp = 16 + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(wfi, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.FUTURE_WRITE) + assert(mem[retp+0] == wfi) + assert(mem[retp+4] == CopyResult.DROPPED) + [] = canon_waitable_join(wfi, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_future_drop_writable(future_t, wfi) + return [] + + caller_ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, caller_ft, core_func, lambda:[], lambda _:()) + + def test_cancel_subtask(): store = Store() root_inst = ComponentInstance(store) @@ -3039,6 +3075,7 @@ def core_consumer(args): test_wasm_to_wasm_stream_empty() test_cancel_copy() test_futures() +test_future_drop_readable_with_pending_write() test_cancel_subtask() test_self_copy(None) test_self_copy(U8Type()) From 17dffe0374d8e64060791b63160187fed7eca004 Mon Sep 17 00:00:00 2001 From: Roman Volosatovs Date: Mon, 24 Aug 2026 14:14:16 +0200 Subject: [PATCH 2/2] add `stream.forward` and `future.forward` canon builtins MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce the โฉ-gated `{stream,future}.forward` built-ins for zero-copy forwarding: given the readable end of a source stream (resp., future) and the writable end of a destination of the same type, both ends are transferred out of the calling component and the destination is fused with the source, delegating all subsequent reads, cancels and drops to it. Elements rendezvous directly between the outer producer and consumer with end-to-end backpressure and no read-ahead, and end-of-stream and drops propagate in both directions, allowing hosts to remove the calling component from the copy path entirely. Forwards may be chained; a forward that would (transitively) make a stream or future its own source traps. In the binary format, `stream.forward` is assigned opcode 0x2e and `future.forward` opcode 0x2f. The binary decode test is left to a follow-up commit so that test/binary/binary.wast keeps passing against engines that do not decode these opcodes yet. Signed-off-by: Roman Volosatovs Assisted-by: claude:claude-fable-5 --- design/mvp/Binary.md | 2 + design/mvp/CanonicalABI.md | 151 +++- design/mvp/Concurrency.md | 16 +- design/mvp/Explainer.md | 48 ++ design/mvp/canonical-abi/definitions.py | 81 ++- design/mvp/canonical-abi/run_tests.py | 869 +++++++++++++++++++++++- test/async/forward-future.wast | 576 ++++++++++++++++ test/async/forward-stream.wast | 859 +++++++++++++++++++++++ test/nyi.txt | 2 + 9 files changed, 2587 insertions(+), 17 deletions(-) create mode 100644 test/async/forward-future.wast create mode 100644 test/async/forward-stream.wast diff --git a/design/mvp/Binary.md b/design/mvp/Binary.md index 3915bb75..4a3141d9 100644 --- a/design/mvp/Binary.md +++ b/design/mvp/Binary.md @@ -312,6 +312,7 @@ canon ::= 0x00 0x00 f: opts: ft: => (canon lift | 0x0e t: => (canon stream.new t (core func)) ๐Ÿ”€ | 0x0f t: opts: => (canon stream.read t opts (core func)) ๐Ÿ”€ | 0x10 t: opts: => (canon stream.write t opts (core func)) ๐Ÿ”€ + | 0x2e t: => (canon stream.forward t (core func)) โฉ | 0x11 t: async?: => (canon stream.cancel-read t async? (core func)) ๐Ÿ”€ | 0x12 t: async?: => (canon stream.cancel-write t async? (core func)) ๐Ÿ”€ | 0x13 t: => (canon stream.drop-readable t (core func)) ๐Ÿ”€ @@ -319,6 +320,7 @@ canon ::= 0x00 0x00 f: opts: ft: => (canon lift | 0x15 t: => (canon future.new t (core func)) ๐Ÿ”€ | 0x16 t: opts: => (canon future.read t opts (core func)) ๐Ÿ”€ | 0x17 t: opts: => (canon future.write t opts (core func)) ๐Ÿ”€ + | 0x2f t: => (canon future.forward t (core func)) โฉ | 0x18 t: async?: => (canon future.cancel-read t async? (core func)) ๐Ÿ”€ | 0x19 t: async?: => (canon future.cancel-write t async? (core func)) ๐Ÿ”€ | 0x1a t: => (canon future.drop-readable t (core func)) ๐Ÿ”€ diff --git a/design/mvp/CanonicalABI.md b/design/mvp/CanonicalABI.md index 1cb97a3f..37399d34 100644 --- a/design/mvp/CanonicalABI.md +++ b/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) ๐Ÿงต @@ -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,12 +1899,23 @@ 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 @@ -1909,10 +1923,15 @@ callback: 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,10 +2108,17 @@ 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] @@ -2098,6 +2126,7 @@ class SharedFutureImpl(ReadableFuture, WritableFuture): def __init__(self, t): self.t = t self.dropped = False + self.forward = None self.reset_pending() def reset_pending(self): @@ -2108,20 +2137,30 @@ 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, ReadableBuffer)) @@ -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: diff --git a/design/mvp/Concurrency.md b/design/mvp/Concurrency.md index 2419a08d..df29b406 100644 --- a/design/mvp/Concurrency.md +++ b/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` type to validate; make it use `string-encoding` and not split code points * add built-ins providing guest code more control over its containing diff --git a/design/mvp/Explainer.md b/design/mvp/Explainer.md index 3cab3bb2..dc753aaa 100644 --- a/design/mvp/Explainer.md +++ b/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,6 +1570,7 @@ canon ::= ... | (canon stream.new (core func ?)) ๐Ÿ”€ | (canon stream.read * (core func ?)) ๐Ÿ”€ | (canon stream.write * (core func ?)) ๐Ÿ”€ + | (canon stream.forward (core func ?)) โฉ | (canon stream.cancel-read async? (core func ?)) ๐Ÿ”€ | (canon stream.cancel-write async? (core func ?)) ๐Ÿ”€ | (canon stream.drop-readable (core func ?)) ๐Ÿ”€ @@ -1576,6 +1578,7 @@ canon ::= ... | (canon future.new (core func ?)) ๐Ÿ”€ | (canon future.read * (core func ?)) ๐Ÿ”€ | (canon future.write * (core func ?)) ๐Ÿ”€ + | (canon future.forward (core func ?)) โฉ | (canon future.cancel-read async? (core func ?)) ๐Ÿ”€ | (canon future.cancel-write async? (core func ?)) ๐Ÿ”€ | (canon future.drop-readable (core func ?)) ๐Ÿ”€ @@ -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>(r: readable-stream-end, w: writable-stream-end)` | +| Approximate WIT signature for `future.forward` | `func>(r: readable-future-end, w: writable-future-end)` | +| 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 diff --git a/design/mvp/canonical-abi/definitions.py b/design/mvp/canonical-abi/definitions.py index 807cfae8..ef906919 100644 --- a/design/mvp/canonical-abi/definitions.py +++ b/design/mvp/canonical-abi/definitions.py @@ -1007,6 +1007,7 @@ class WritableStream(SharedBase): class SharedStreamImpl(ReadableStream, WritableStream): dropped: bool + forward: Optional[ReadableStream] pending_inst: Optional[ComponentInstance] pending_buffer: Optional[Buffer] pending_on_copy: Optional[OnCopy] @@ -1015,6 +1016,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): @@ -1026,22 +1028,35 @@ def set_pending(self, inst, buffer, on_copy, on_copy_done): 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 + 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) 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) @@ -1129,6 +1144,7 @@ class WritableFuture(SharedBase): class SharedFutureImpl(ReadableFuture, WritableFuture): dropped: bool + forward: Optional[ReadableFuture] pending_inst: Optional[ComponentInstance] pending_buffer: Optional[Buffer] pending_on_copy_done: Optional[OnCopyDone] @@ -1136,6 +1152,7 @@ class SharedFutureImpl(ReadableFuture, WritableFuture): def __init__(self, t): self.t = t self.dropped = False + self.forward = None self.reset_pending() def reset_pending(self): @@ -1146,22 +1163,35 @@ def set_pending(self, inst, buffer, on_copy_done): 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) 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, ReadableBuffer)) self.reset_and_notify_pending(CopyResult.DROPPED) 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) @@ -2632,6 +2662,47 @@ def on_copy_done(result): assert(code == event_code and index == i) return [payload] +### โฉ `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_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 [] + ### ๐Ÿ”€ `canon {stream,future}.cancel-{read,write}` def canon_stream_cancel_read(stream_t, async_, i): diff --git a/design/mvp/canonical-abi/run_tests.py b/design/mvp/canonical-abi/run_tests.py index 30573d41..74277f3d 100644 --- a/design/mvp/canonical-abi/run_tests.py +++ b/design/mvp/canonical-abi/run_tests.py @@ -1683,7 +1683,7 @@ def core_func(args): assert(dst_stream.received == [11,12,13,14,15,16,17,18]) -def test_stream_forward(): +def test_stream_passthrough(): src_stream = HostSource(U8Type(), [1,2,3,4], chunk=4) def on_start(): return [src_stream] @@ -1707,6 +1707,856 @@ def core_func(args): assert(src_stream is dst_stream) +def test_stream_forward_builtin(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + stream_t = StreamType(U8Type()) + + src_stream = HostSource(U8Type(), [1,2,3,4], chunk=4, destroy_if_empty=False) + def on_start(): + return [src_stream] + + dst_stream = None + def on_resolve(results): + assert(len(results) == 1) + nonlocal dst_stream + dst_stream = HostSink(results[0], chunk=4) + + def core_func(args): + [rsi1] = args + [packed] = canon_stream_new(stream_t) + rsi2,wsi2 = unpack_new_ends(packed) + [] = canon_task_return([stream_t], opts, [rsi2]) + [] = canon_stream_forward(stream_t, rsi1, wsi2) + + # Both ends were transferred out of the table. + try: + canon_stream_drop_readable(stream_t, rsi1) + assert(False) + except Trap: + pass + try: + canon_stream_drop_writable(stream_t, wsi2) + assert(False) + except Trap: + pass + + src_stream.write([5,6,7,8]) + assert(dst_stream.consume(8) == [1,2,3,4,5,6,7,8]) + + # The end of the source stream propagates to the destination's reader. + src_stream.destroy_once_empty() + assert(dst_stream.consume(1) is None) + assert(dst_stream.closed) + return [] + + ft = FuncType([stream_t], [stream_t], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_chained(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(32) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + stream_t = StreamType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_stream_new(stream_t) + rsia,wsia = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsib,wsib = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsic,wsic = unpack_new_ends(packed) + retp = 16 + [seti] = canon_waitable_set_new() + + [] = canon_stream_forward(stream_t, rsia, wsib) + [] = canon_stream_forward(stream_t, rsib, wsic) + + # A write to the head of the chain rendezvous directly with a read from + # the tail of the chain. + mem[0:4] = b'\x01\x02\x03\x04' + [ret] = canon_stream_write(stream_t, opts, wsia, 0, 4) + assert(ret == definitions.BLOCKED) + [ret] = canon_stream_read(stream_t, sync_opts, rsic, 8, 4) + result,n = unpack_result(ret) + assert(n == 4 and result == CopyResult.COMPLETED) + assert(mem[8:12] == b'\x01\x02\x03\x04') + + [] = canon_waitable_join(wsia, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.STREAM_WRITE) + assert(mem[retp+0] == wsia) + result,n = unpack_result(mem[retp+4]) + assert(n == 4 and result == CopyResult.COMPLETED) + [] = canon_waitable_join(wsia, 0) + + # Dropping the head's writable end propagates the end of the stream + # through both forwards to the tail's reader. + [] = canon_stream_drop_writable(stream_t, wsia) + [ret] = canon_stream_read(stream_t, sync_opts, rsic, 8, 4) + assert(ret == CopyResult.DROPPED) + + [] = canon_waitable_set_drop(seti) + [] = canon_stream_drop_readable(stream_t, rsic) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_dest_dropped(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + stream_t = StreamType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_stream_new(stream_t) + rsia,wsia = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsib,wsib = unpack_new_ends(packed) + + [] = canon_stream_forward(stream_t, rsia, wsib) + mem[0:2] = b'\x01\x02' + [ret] = canon_stream_write(stream_t, opts, wsia, 0, 2) + assert(ret == definitions.BLOCKED) + + # Dropping the destination's readable end propagates back to the source's + # blocked writer. + [] = canon_stream_drop_readable(stream_t, rsib) + retp = 16 + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(wsia, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.STREAM_WRITE) + assert(mem[retp+0] == wsia) + result,n = unpack_result(mem[retp+4]) + assert(n == 0 and result == CopyResult.DROPPED) + [] = canon_waitable_join(wsia, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_stream_drop_writable(stream_t, wsia) + + # An already-dropped destination is propagated to the source when the + # forward starts. + [packed] = canon_stream_new(stream_t) + rsic,wsic = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsid,wsid = unpack_new_ends(packed) + [] = canon_stream_drop_readable(stream_t, rsic) + [] = canon_stream_forward(stream_t, rsid, wsic) + [ret] = canon_stream_write(stream_t, sync_opts, wsid, 0, 2) + assert(ret == CopyResult.DROPPED) + [] = canon_stream_drop_writable(stream_t, wsid) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_source_dropped(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + stream_t = StreamType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + + # An already-dropped source is propagated to the destination's reader, + # without any special case in stream.forward: reads of the destination are + # delegated to the dropped source stream. + [packed] = canon_stream_new(stream_t) + rsia,wsia = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsib,wsib = unpack_new_ends(packed) + [] = canon_stream_drop_writable(stream_t, wsia) + [] = canon_stream_forward(stream_t, rsia, wsib) + [ret] = canon_stream_read(stream_t, sync_opts, rsib, 0, 4) + result,n = unpack_result(ret) + assert(n == 0 and result == CopyResult.DROPPED) + [] = canon_stream_drop_readable(stream_t, rsib) + + # The same holds for a read that was already blocked on the destination + # when the forward started. + [packed] = canon_stream_new(stream_t) + rsic,wsic = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsid,wsid = unpack_new_ends(packed) + [] = canon_stream_drop_writable(stream_t, wsic) + [ret] = canon_stream_read(stream_t, opts, rsid, 0, 4) + assert(ret == definitions.BLOCKED) + [] = canon_stream_forward(stream_t, rsic, wsid) + retp = 16 + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(rsid, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.STREAM_READ) + assert(mem[retp+0] == rsid) + result,n = unpack_result(mem[retp+4]) + assert(n == 0 and result == CopyResult.DROPPED) + [] = canon_waitable_join(rsid, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_stream_drop_readable(stream_t, rsid) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_pending_copies(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + stream_t = StreamType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + retp = 16 + readp = 8 + [seti] = canon_waitable_set_new() + + # A read that was already blocked on the destination when the forward + # started is transferred to the source. + [packed] = canon_stream_new(stream_t) + rsia,wsia = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsib,wsib = unpack_new_ends(packed) + [ret] = canon_stream_read(stream_t, opts, rsib, readp, 4) + assert(ret == definitions.BLOCKED) + [] = canon_stream_forward(stream_t, rsia, wsib) + mem[0:4] = b'\x01\x02\x03\x04' + [ret] = canon_stream_write(stream_t, sync_opts, wsia, 0, 4) + result,n = unpack_result(ret) + assert(n == 4 and result == CopyResult.COMPLETED) + [] = canon_waitable_join(rsib, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.STREAM_READ) + assert(mem[retp+0] == rsib) + result,n = unpack_result(mem[retp+4]) + assert(n == 4 and result == CopyResult.COMPLETED) + assert(mem[readp:readp+4] == b'\x01\x02\x03\x04') + [] = canon_waitable_join(rsib, 0) + [] = canon_stream_drop_writable(stream_t, wsia) + [] = canon_stream_drop_readable(stream_t, rsib) + + # A write that was already blocked on the source when the forward started + # rendezvous with a later read from the destination. + [packed] = canon_stream_new(stream_t) + rsic,wsic = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsid,wsid = unpack_new_ends(packed) + mem[0:4] = b'\x05\x06\x07\x08' + [ret] = canon_stream_write(stream_t, opts, wsic, 0, 4) + assert(ret == definitions.BLOCKED) + [] = canon_stream_forward(stream_t, rsic, wsid) + [ret] = canon_stream_read(stream_t, sync_opts, rsid, readp, 4) + result,n = unpack_result(ret) + assert(n == 4 and result == CopyResult.COMPLETED) + assert(mem[readp:readp+4] == b'\x05\x06\x07\x08') + [] = canon_waitable_join(wsic, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.STREAM_WRITE) + assert(mem[retp+0] == wsic) + result,n = unpack_result(mem[retp+4]) + assert(n == 4 and result == CopyResult.COMPLETED) + [] = canon_waitable_join(wsic, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_stream_drop_writable(stream_t, wsic) + [] = canon_stream_drop_readable(stream_t, rsid) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_waitable_set_traps(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + stream_t = StreamType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_stream_new(stream_t) + rsia,wsia = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_t) + rsib,wsib = unpack_new_ends(packed) + [seti] = canon_waitable_set_new() + + # Like transferring an end to another component, forwarding an end that + # is in a waitable set traps. + [] = canon_waitable_join(rsia, seti) + try: + canon_stream_forward(stream_t, rsia, wsib) + assert(False) + except Trap: + pass + [] = canon_waitable_join(rsia, 0) + + [] = canon_waitable_join(wsib, seti) + try: + canon_stream_forward(stream_t, rsia, wsib) + assert(False) + except Trap: + pass + [] = canon_waitable_join(wsib, 0) + + [] = canon_stream_forward(stream_t, rsia, wsib) + [] = canon_stream_drop_writable(stream_t, wsia) + [ret] = canon_stream_read(stream_t, opts, rsib, 8, 4) + result,n = unpack_result(ret) + assert(n == 0 and result == CopyResult.DROPPED) + + [] = canon_waitable_set_drop(seti) + [] = canon_stream_drop_readable(stream_t, rsib) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_type_mismatch_traps(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + stream_u8_t = StreamType(U8Type()) + stream_u16_t = StreamType(U16Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_stream_new(stream_u8_t) + rsi8,wsi8 = unpack_new_ends(packed) + [packed] = canon_stream_new(stream_u16_t) + rsi16,wsi16 = unpack_new_ends(packed) + + # The element type of the readable end must match the type immediate. + try: + canon_stream_forward(stream_u8_t, rsi16, wsi8) + assert(False) + except Trap: + pass + + # So must the element type of the writable end. + try: + canon_stream_forward(stream_u8_t, rsi8, wsi16) + assert(False) + except Trap: + pass + + [] = canon_stream_drop_readable(stream_u8_t, rsi8) + [] = canon_stream_drop_writable(stream_u8_t, wsi8) + [] = canon_stream_drop_readable(stream_u16_t, rsi16) + [] = canon_stream_drop_writable(stream_u16_t, wsi16) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_stream_forward_builtin_cycle_traps(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + stream_t = StreamType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + + # Forwarding a stream directly into itself traps. + [packed] = canon_stream_new(stream_t) + rsia,wsia = unpack_new_ends(packed) + try: + canon_stream_forward(stream_t, rsia, wsia) + assert(False) + except Trap: + pass + + # So does transitively making a stream its own source through a chain of + # forwards. + [packed] = canon_stream_new(stream_t) + rsib,wsib = unpack_new_ends(packed) + [] = canon_stream_forward(stream_t, rsia, wsib) + try: + canon_stream_forward(stream_t, rsib, wsia) + assert(False) + except Trap: + pass + + [] = canon_stream_drop_writable(stream_t, wsia) + [ret] = canon_stream_read(stream_t, mk_opts(memory=MemInst(mem, 'i32')), rsib, 0, 4) + result,n = unpack_result(ret) + assert(n == 0 and result == CopyResult.DROPPED) + [] = canon_stream_drop_readable(stream_t, rsib) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_t = FutureType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_future_new(future_t) + rfia,wfia = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfib,wfib = unpack_new_ends(packed) + + [] = canon_future_forward(future_t, rfia, wfib) + + # Both ends were transferred out of the table. + try: + canon_future_drop_readable(future_t, rfia) + assert(False) + except Trap: + pass + try: + canon_future_drop_writable(future_t, wfib) + assert(False) + except Trap: + pass + + # A write to the source rendezvous directly with a read from the + # destination. + writep = 0 + readp = 8 + mem[writep] = 42 + [ret] = canon_future_write(future_t, opts, wfia, writep) + assert(ret == definitions.BLOCKED) + [ret] = canon_future_read(future_t, sync_opts, rfib, readp) + assert(ret == CopyResult.COMPLETED) + assert(mem[readp] == 42) + + retp = 16 + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(wfia, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.FUTURE_WRITE) + assert(mem[retp+0] == wfia) + assert(mem[retp+4] == CopyResult.COMPLETED) + [] = canon_waitable_join(wfia, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_future_drop_writable(future_t, wfia) + [] = canon_future_drop_readable(future_t, rfib) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin_chained(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_t = FutureType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_future_new(future_t) + rfia,wfia = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfib,wfib = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfic,wfic = unpack_new_ends(packed) + + [] = canon_future_forward(future_t, rfia, wfib) + [] = canon_future_forward(future_t, rfib, wfic) + + # A write to the head of the chain rendezvous directly with a read from + # the tail of the chain. + writep = 0 + readp = 8 + mem[writep] = 42 + [ret] = canon_future_write(future_t, opts, wfia, writep) + assert(ret == definitions.BLOCKED) + [ret] = canon_future_read(future_t, sync_opts, rfic, readp) + assert(ret == CopyResult.COMPLETED) + assert(mem[readp] == 42) + + retp = 16 + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(wfia, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.FUTURE_WRITE) + assert(mem[retp+0] == wfia) + assert(mem[retp+4] == CopyResult.COMPLETED) + [] = canon_waitable_join(wfia, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_future_drop_writable(future_t, wfia) + [] = canon_future_drop_readable(future_t, rfic) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin_dest_dropped(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_t = FutureType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + + # Dropping the destination's readable end propagates back to the source's + # blocked writer. + [packed] = canon_future_new(future_t) + rfia,wfia = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfib,wfib = unpack_new_ends(packed) + [] = canon_future_forward(future_t, rfia, wfib) + mem[0] = 42 + [ret] = canon_future_write(future_t, opts, wfia, 0) + assert(ret == definitions.BLOCKED) + [] = canon_future_drop_readable(future_t, rfib) + retp = 16 + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(wfia, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.FUTURE_WRITE) + assert(mem[retp+0] == wfia) + assert(mem[retp+4] == CopyResult.DROPPED) + [] = canon_waitable_join(wfia, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_future_drop_writable(future_t, wfia) + + # An already-dropped destination is propagated to the source when the + # forward starts. + [packed] = canon_future_new(future_t) + rfic,wfic = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfid,wfid = unpack_new_ends(packed) + [] = canon_future_drop_readable(future_t, rfic) + [] = canon_future_forward(future_t, rfid, wfic) + [ret] = canon_future_write(future_t, sync_opts, wfid, 0) + assert(ret == CopyResult.DROPPED) + [] = canon_future_drop_writable(future_t, wfid) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin_pending_copies(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_t = FutureType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + retp = 16 + [seti] = canon_waitable_set_new() + + # A read that was already blocked on the destination when the forward + # started is transferred to the source. + [packed] = canon_future_new(future_t) + rfia,wfia = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfib,wfib = unpack_new_ends(packed) + readp = 8 + [ret] = canon_future_read(future_t, opts, rfib, readp) + assert(ret == definitions.BLOCKED) + [] = canon_future_forward(future_t, rfia, wfib) + mem[0] = 42 + [ret] = canon_future_write(future_t, sync_opts, wfia, 0) + assert(ret == CopyResult.COMPLETED) + [] = canon_waitable_join(rfib, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.FUTURE_READ) + assert(mem[retp+0] == rfib) + assert(mem[retp+4] == CopyResult.COMPLETED) + assert(mem[readp] == 42) + [] = canon_waitable_join(rfib, 0) + [] = canon_future_drop_writable(future_t, wfia) + [] = canon_future_drop_readable(future_t, rfib) + + # A write that was already blocked on the source when the forward started + # rendezvous with a later read from the destination. + [packed] = canon_future_new(future_t) + rfic,wfic = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfid,wfid = unpack_new_ends(packed) + mem[0] = 43 + [ret] = canon_future_write(future_t, opts, wfic, 0) + assert(ret == definitions.BLOCKED) + [] = canon_future_forward(future_t, rfic, wfid) + [ret] = canon_future_read(future_t, sync_opts, rfid, readp) + assert(ret == CopyResult.COMPLETED) + assert(mem[readp] == 43) + [] = canon_waitable_join(wfic, seti) + [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp) + assert(event == EventCode.FUTURE_WRITE) + assert(mem[retp+0] == wfic) + assert(mem[retp+4] == CopyResult.COMPLETED) + [] = canon_waitable_join(wfic, 0) + [] = canon_waitable_set_drop(seti) + [] = canon_future_drop_writable(future_t, wfic) + [] = canon_future_drop_readable(future_t, rfid) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin_waitable_set_traps(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_t = FutureType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_future_new(future_t) + rfia,wfia = unpack_new_ends(packed) + [packed] = canon_future_new(future_t) + rfib,wfib = unpack_new_ends(packed) + [seti] = canon_waitable_set_new() + + # Like transferring an end to another component, forwarding an end that + # is in a waitable set traps. + [] = canon_waitable_join(rfia, seti) + try: + canon_future_forward(future_t, rfia, wfib) + assert(False) + except Trap: + pass + [] = canon_waitable_join(rfia, 0) + + [] = canon_waitable_join(wfib, seti) + try: + canon_future_forward(future_t, rfia, wfib) + assert(False) + except Trap: + pass + [] = canon_waitable_join(wfib, 0) + + [] = canon_future_forward(future_t, rfia, wfib) + [] = canon_future_drop_readable(future_t, rfib) + [ret] = canon_future_write(future_t, sync_opts, wfia, 0) + assert(ret == CopyResult.DROPPED) + [] = canon_future_drop_writable(future_t, wfia) + [] = canon_waitable_set_drop(seti) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin_type_mismatch_traps(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_u8_t = FutureType(U8Type()) + future_u16_t = FutureType(U16Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + [packed] = canon_future_new(future_u8_t) + rfi8,wfi8 = unpack_new_ends(packed) + [packed] = canon_future_new(future_u16_t) + rfi16,wfi16 = unpack_new_ends(packed) + + # The element type of the readable end must match the type immediate. + try: + canon_future_forward(future_u8_t, rfi16, wfi8) + assert(False) + except Trap: + pass + + # So must the element type of the writable end. + try: + canon_future_forward(future_u8_t, rfi8, wfi16) + assert(False) + except Trap: + pass + + [] = canon_future_drop_readable(future_u8_t, rfi8) + [ret] = canon_future_write(future_u8_t, sync_opts, wfi8, 0) + assert(ret == CopyResult.DROPPED) + [] = canon_future_drop_writable(future_u8_t, wfi8) + [] = canon_future_drop_readable(future_u16_t, rfi16) + [ret] = canon_future_write(future_u16_t, sync_opts, wfi16, 0) + assert(ret == CopyResult.DROPPED) + [] = canon_future_drop_writable(future_u16_t, wfi16) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + +def test_future_forward_builtin_cycle_traps(): + store = Store() + inst = ComponentInstance(store) + mem = bytearray(24) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + sync_opts = mk_opts(memory=MemInst(mem, 'i32'), async_=False) + future_t = FutureType(U8Type()) + + def on_start(): + return [] + + def on_resolve(results): + assert(len(results) == 0) + + def core_func(args): + assert(len(args) == 0) + [] = canon_task_return([], opts, []) + + # Forwarding a future directly into itself traps. + [packed] = canon_future_new(future_t) + rfia,wfia = unpack_new_ends(packed) + try: + canon_future_forward(future_t, rfia, wfia) + assert(False) + except Trap: + pass + + # So does transitively making a future its own source through a chain of + # forwards. + [packed] = canon_future_new(future_t) + rfib,wfib = unpack_new_ends(packed) + [] = canon_future_forward(future_t, rfia, wfib) + try: + canon_future_forward(future_t, rfib, wfia) + assert(False) + except Trap: + pass + + [] = canon_future_drop_readable(future_t, rfib) + [ret] = canon_future_write(future_t, sync_opts, wfia, 0) + assert(ret == CopyResult.DROPPED) + [] = canon_future_drop_writable(future_t, wfia) + return [] + + ft = FuncType([], [], async_ = True) + lift_and_run(opts, inst, ft, core_func, on_start, on_resolve) + + def test_receive_own_stream(): store = Store() inst = ComponentInstance(store) @@ -3068,7 +3918,22 @@ def core_consumer(args): test_sync_using_wait() test_eager_stream_completion() test_async_stream_ops() -test_stream_forward() +test_stream_passthrough() +test_stream_forward_builtin() +test_stream_forward_builtin_chained() +test_stream_forward_builtin_dest_dropped() +test_stream_forward_builtin_source_dropped() +test_stream_forward_builtin_pending_copies() +test_stream_forward_builtin_waitable_set_traps() +test_stream_forward_builtin_type_mismatch_traps() +test_stream_forward_builtin_cycle_traps() +test_future_forward_builtin() +test_future_forward_builtin_chained() +test_future_forward_builtin_dest_dropped() +test_future_forward_builtin_pending_copies() +test_future_forward_builtin_waitable_set_traps() +test_future_forward_builtin_type_mismatch_traps() +test_future_forward_builtin_cycle_traps() test_receive_own_stream() test_host_partial_reads_writes() test_wasm_to_wasm_stream() diff --git a/test/async/forward-future.wast b/test/async/forward-future.wast new file mode 100644 index 00000000..a4f9f9d6 --- /dev/null +++ b/test/async/forward-future.wast @@ -0,0 +1,576 @@ +;; This test contains two components $C and $D where $C writes into a future +;; it created (the "source", ends $r.src/$w.src) and $D forwards that future +;; into a second future (the "destination", ends $r.dst/$w.dst) using the +;; future.forward built-in, reading the forwarded value back out of the +;; destination future itself. +;; +;; $D exports one function per scenario. Since traps take out their containing +;; instance, a fresh instance of $Tester is created for each invoke. +(component definition $Tester + (component $C + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $CM + (import "" "mem" (memory 1)) + (import "" "future.new" (func $future.new (result i64))) + (import "" "future.write" (func $future.write (param i32 i32) (result i32))) + (import "" "future.drop-writable" (func $future.drop-writable (param i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32))) + + (global $w.src (mut i32) (i32.const 0)) + + (func $start-future (export "start-future") (result i32) + ;; create a new future, return the readable end to the caller + (local $ret64 i64) + (local.set $ret64 (call $future.new)) + (global.set $w.src (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (i32.wrap_i64 (local.get $ret64)) + ) + (func $write1 (export "write1") + ;; write the value, expecting to rendezvous with a read + (local $ret i32) + (i32.store8 (i32.const 8) (i32.const 0xab)) + (local.set $ret (call $future.write (global.get $w.src) (i32.const 8))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + (call $future.drop-writable (global.get $w.src)) + ) + (func $write1-dropped (export "write1-dropped") + ;; write expecting to observe DROPPED synchronously + (local $ret i32) + (local.set $ret (call $future.write (global.get $w.src) (i32.const 8))) + (if (i32.ne (i32.const 1 (; DROPPED ;)) (local.get $ret)) + (then unreachable)) + (call $future.drop-writable (global.get $w.src)) + ) + (func $start-blocking-write (export "start-blocking-write") + (local $ret i32) + + ;; prepare the write buffer + (i32.store8 (i32.const 8) (i32.const 0xab)) + + ;; start a blocking write + (local.set $ret (call $future.write (global.get $w.src) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ) + (func $check-write-event (export "check-write-event") + ;; confirm the blocking write completed + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.src) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (i32.const 5 (; FUTURE_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $w.src) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (global.get $w.src) (i32.const 0)) + (call $future.drop-writable (global.get $w.src)) + ) + (func $check-write-dropped (export "check-write-dropped") + ;; confirm the blocking write observed DROPPED with nothing written + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.src) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (i32.const 5 (; FUTURE_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $w.src) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (i32.const 1 (; DROPPED ;)) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (global.get $w.src) (i32.const 0)) + (call $future.drop-writable (global.get $w.src)) + ) + ) + (type $FT (future u8)) + (canon future.new $FT (core func $future.new)) + (canon future.write $FT async (memory (core memory $memory "mem")) (core func $future.write)) + (canon future.drop-writable $FT (core func $future.drop-writable)) + (canon waitable.join (core func $waitable.join)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.wait (memory (core memory $memory "mem")) (core func $waitable-set.wait)) + (core instance $cm (instantiate $CM (with "" (instance + (export "mem" (memory $memory "mem")) + (export "future.new" (func $future.new)) + (export "future.write" (func $future.write)) + (export "future.drop-writable" (func $future.drop-writable)) + (export "waitable.join" (func $waitable.join)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.wait" (func $waitable-set.wait)) + )))) + (func (export "start-future") async (result (future u8)) (canon lift (core func $cm "start-future"))) + (func (export "write1") async (canon lift (core func $cm "write1"))) + (func (export "write1-dropped") async (canon lift (core func $cm "write1-dropped"))) + (func (export "start-blocking-write") async (canon lift (core func $cm "start-blocking-write"))) + (func (export "check-write-event") async (canon lift (core func $cm "check-write-event"))) + (func (export "check-write-dropped") async (canon lift (core func $cm "check-write-dropped"))) + ) + (component $D + (import "c" (instance $c + (export "start-future" (func async (result (future u8)))) + (export "write1" (func async)) + (export "write1-dropped" (func async)) + (export "start-blocking-write" (func async)) + (export "check-write-event" (func async)) + (export "check-write-dropped" (func async)) + )) + + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $Core + (import "" "mem" (memory 1)) + (import "" "future.new" (func $future.new (result i64))) + (import "" "future.new-u16" (func $future.new-u16 (result i64))) + (import "" "future.read" (func $future.read (param i32 i32) (result i32))) + (import "" "future.read-async" (func $future.read-async (param i32 i32) (result i32))) + (import "" "future.write-async" (func $future.write-async (param i32 i32) (result i32))) + (import "" "future.forward" (func $future.forward (param i32 i32))) + (import "" "future.cancel-read" (func $future.cancel-read (param i32) (result i32))) + (import "" "future.drop-readable" (func $future.drop-readable (param i32))) + (import "" "future.drop-writable" (func $future.drop-writable (param i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32))) + (import "" "start-future" (func $start-future (result i32))) + (import "" "write1" (func $write1)) + (import "" "write1-dropped" (func $write1-dropped)) + (import "" "start-blocking-write" (func $start-blocking-write)) + (import "" "check-write-event" (func $check-write-event)) + (import "" "check-write-dropped" (func $check-write-dropped)) + + (global $r.src (mut i32) (i32.const 0)) + (global $r.dst (mut i32) (i32.const 0)) + (global $w.dst (mut i32) (i32.const 0)) + + (func $setup + ;; get the source future from $C and create the destination future + (local $ret64 i64) + (global.set $r.src (call $start-future)) + (if (i32.ne (i32.const 1) (global.get $r.src)) + (then unreachable)) + (local.set $ret64 (call $future.new)) + (global.set $r.dst (i32.wrap_i64 (local.get $ret64))) + (global.set $w.dst (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (if (i32.ne (i32.const 2) (global.get $r.dst)) + (then unreachable)) + (if (i32.ne (i32.const 3) (global.get $w.dst)) + (then unreachable)) + ) + (func $expect-event (param $waitable i32) (param $event i32) (param $payload i32) + ;; wait for the given event on the given waitable with the given payload + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (local.get $waitable) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (local.get $event) (local.get $ret)) + (then unreachable)) + (if (i32.ne (local.get $waitable) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (local.get $payload) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (local.get $waitable) (i32.const 0)) + ) + (func $read1-dst + ;; synchronously read the value out of the destination future + (local $ret i32) + (local.set $ret (call $future.read (global.get $r.dst) (i32.const 8))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0xab) (i32.load8_u (i32.const 8))) + (then unreachable)) + (call $future.drop-readable (global.get $r.dst)) + ) + (func $read1-dst-async + ;; start a read on the destination future that will block + (local $ret i32) + (local.set $ret (call $future.read-async (global.get $r.dst) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ) + (func $expect-read1-dst + (call $expect-event (global.get $r.dst) (i32.const 4 (; FUTURE_READ ;)) (i32.const 0 (; COMPLETED ;))) + (if (i32.ne (i32.const 0xab) (i32.load8_u (i32.const 8))) + (then unreachable)) + (call $future.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-rendezvous") + (call $setup) + + ;; forward the source into the destination; this returns immediately + ;; and no event is ever delivered + (call $future.forward (global.get $r.src) (global.get $w.dst)) + + ;; with no read pending on the destination, $C's write blocks + (call $start-blocking-write) + + ;; the read is satisfied out of $C's write buffer, completing its write + (call $read1-dst) + (call $check-write-event) + ) + + (func (export "forward-pending-read") + (call $setup) + + ;; a read blocked on the destination when the forward starts is + ;; transferred to the source + (call $read1-dst-async) + (call $future.forward (global.get $r.src) (global.get $w.dst)) + (call $write1) + (call $expect-read1-dst) + ) + + (func (export "forward-pending-write") + (call $setup) + + ;; a write blocked on the source when the forward starts rendezvous + ;; with a later destination read + (call $start-blocking-write) + (call $future.forward (global.get $r.src) (global.get $w.dst)) + (call $read1-dst) + (call $check-write-event) + ) + + (func (export "forward-dst-dropped") + (call $setup) + (call $future.forward (global.get $r.src) (global.get $w.dst)) + (call $start-blocking-write) + + ;; dropping the destination's readable end drops the source's + ;; readable end, so $C's blocked write observes DROPPED + (call $future.drop-readable (global.get $r.dst)) + (call $check-write-dropped) + ) + + (func (export "forward-dst-already-dropped") + (call $setup) + + ;; forwarding into an already-dropped destination drops the source's + ;; readable end right away + (call $future.drop-readable (global.get $r.dst)) + (call $future.forward (global.get $r.src) (global.get $w.dst)) + (call $write1-dropped) + ) + + (func (export "forward-chained") + (local $ret64 i64) (local $r.mid i32) (local $w.mid i32) + (call $setup) + (local.set $ret64 (call $future.new)) + (local.set $r.mid (i32.wrap_i64 (local.get $ret64))) + (local.set $w.mid (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + + ;; chain two forwards through an intermediate future: a read at the + ;; end of the chain is satisfied directly out of $C's write + (call $future.forward (global.get $r.src) (local.get $w.mid)) + (call $future.forward (local.get $r.mid) (global.get $w.dst)) + (call $start-blocking-write) + (call $read1-dst) + (call $check-write-event) + ) + + (func (export "forward-cancel-read") + (local $ret i32) + (call $setup) + + ;; a read transferred to the source can still be cancelled through + ;; the destination's readable end, after which the future remains + ;; usable + (call $read1-dst-async) + (call $future.forward (global.get $r.src) (global.get $w.dst)) + (local.set $ret (call $future.cancel-read (global.get $r.dst))) + (if (i32.ne (i32.const 0x02 (; CANCELLED ;)) (local.get $ret)) + (then unreachable)) + (call $read1-dst-async) + (call $write1) + (call $expect-read1-dst) + ) + + (func (export "forward-after-value-read") + (local $ret i32) + (call $setup) + + ;; once the future's value has been read it can no longer be forwarded + (call $start-blocking-write) + (local.set $ret (call $future.read (global.get $r.src) (i32.const 8))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-after-value-written") + (local $ret i32) + (call $setup) + + ;; once a value has been written into the destination it can no longer + ;; be the target of a forward + (call $read1-dst-async) + (i32.store8 (i32.const 16) (i32.const 0xcd)) + (local.set $ret (call $future.write-async (global.get $w.dst) (i32.const 16))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-after-write-dropped") + (local $ret i32) + (call $setup) + + ;; a writable end that observed DROPPED can no longer be the target + ;; of a forward + (call $future.drop-readable (global.get $r.dst)) + (local.set $ret (call $future.write-async (global.get $w.dst) (i32.const 8))) + (if (i32.ne (i32.const 1 (; DROPPED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-removes-readable") + (call $setup) + ;; future.forward removes both ends from the table + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ;; boom + (call $future.drop-readable (global.get $r.src)) + ) + + (func (export "forward-removes-writable") + (call $setup) + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ;; boom + (call $future.drop-writable (global.get $w.dst)) + ) + + (func (export "forward-while-reading") + (local $ret i32) + (call $setup) + (local.set $ret (call $future.read-async (global.get $r.src) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-while-writing") + (local $ret i32) + (call $setup) + (local.set $ret (call $future.write-async (global.get $w.dst) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-readable-in-waitable-set") + (local $seti i32) + (call $setup) + ;; forwarding an end that is in a waitable set traps + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $r.src) (local.get $seti)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-writable-in-waitable-set") + (local $seti i32) + (call $setup) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.dst) (local.get $seti)) + ;; boom + (call $future.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-readable-as-writable") + (call $setup) + ;; boom + (call $future.forward (global.get $r.src) (global.get $r.dst)) + ) + + (func (export "forward-writable-as-readable") + (call $setup) + ;; boom + (call $future.forward (global.get $w.dst) (global.get $w.dst)) + ) + + (func (export "forward-readable-type-mismatch") + (local $ret64 i64) + (call $setup) + ;; the element type of the readable end must match the type immediate + (local.set $ret64 (call $future.new-u16)) + ;; boom + (call $future.forward (i32.wrap_i64 (local.get $ret64)) (global.get $w.dst)) + ) + + (func (export "forward-writable-type-mismatch") + (local $ret64 i64) + (call $setup) + ;; the element type of the writable end must match the type immediate + (local.set $ret64 (call $future.new-u16)) + ;; boom + (call $future.forward (global.get $r.src) (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + + (func (export "self-forward") + (local $ret64 i64) (local $r.self i32) (local $w.self i32) + ;; a future cannot be forwarded into itself + (local.set $ret64 (call $future.new)) + (local.set $r.self (i32.wrap_i64 (local.get $ret64))) + (local.set $w.self (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ;; boom + (call $future.forward (local.get $r.self) (local.get $w.self)) + ) + + (func (export "forward-cycle") + (local $ret64 i64) (local $r.a i32) (local $w.a i32) (local $r.b i32) (local $w.b i32) + (local.set $ret64 (call $future.new)) + (local.set $r.a (i32.wrap_i64 (local.get $ret64))) + (local.set $w.a (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (local.set $ret64 (call $future.new)) + (local.set $r.b (i32.wrap_i64 (local.get $ret64))) + (local.set $w.b (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + + (call $future.forward (local.get $r.a) (local.get $w.b)) + ;; forwarding $b back into $a would close a cycle + ;; boom + (call $future.forward (local.get $r.b) (local.get $w.a)) + ) + ) + (type $FT (future u8)) + (canon future.new $FT (core func $future.new)) + (type $FTU16 (future u16)) + (canon future.new $FTU16 (core func $future.new-u16)) + (canon future.read $FT (memory (core memory $memory "mem")) (core func $future.read)) + (canon future.read $FT async (memory (core memory $memory "mem")) (core func $future.read-async)) + (canon future.write $FT async (memory (core memory $memory "mem")) (core func $future.write-async)) + (canon future.forward $FT (core func $future.forward)) + (canon future.cancel-read $FT (core func $future.cancel-read)) + (canon future.drop-readable $FT (core func $future.drop-readable)) + (canon future.drop-writable $FT (core func $future.drop-writable)) + (canon waitable.join (core func $waitable.join)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.wait (memory (core memory $memory "mem")) (core func $waitable-set.wait)) + (canon lower (func $c "start-future") (core func $start-future')) + (canon lower (func $c "write1") (core func $write1')) + (canon lower (func $c "write1-dropped") (core func $write1-dropped')) + (canon lower (func $c "start-blocking-write") (core func $start-blocking-write')) + (canon lower (func $c "check-write-event") (core func $check-write-event')) + (canon lower (func $c "check-write-dropped") (core func $check-write-dropped')) + (core instance $core (instantiate $Core (with "" (instance + (export "mem" (memory $memory "mem")) + (export "future.new" (func $future.new)) + (export "future.new-u16" (func $future.new-u16)) + (export "future.read" (func $future.read)) + (export "future.read-async" (func $future.read-async)) + (export "future.write-async" (func $future.write-async)) + (export "future.forward" (func $future.forward)) + (export "future.cancel-read" (func $future.cancel-read)) + (export "future.drop-readable" (func $future.drop-readable)) + (export "future.drop-writable" (func $future.drop-writable)) + (export "waitable.join" (func $waitable.join)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.wait" (func $waitable-set.wait)) + (export "start-future" (func $start-future')) + (export "write1" (func $write1')) + (export "write1-dropped" (func $write1-dropped')) + (export "start-blocking-write" (func $start-blocking-write')) + (export "check-write-event" (func $check-write-event')) + (export "check-write-dropped" (func $check-write-dropped')) + )))) + (func (export "forward-rendezvous") async (canon lift (core func $core "forward-rendezvous"))) + (func (export "forward-pending-read") async (canon lift (core func $core "forward-pending-read"))) + (func (export "forward-pending-write") async (canon lift (core func $core "forward-pending-write"))) + (func (export "forward-dst-dropped") async (canon lift (core func $core "forward-dst-dropped"))) + (func (export "forward-dst-already-dropped") async (canon lift (core func $core "forward-dst-already-dropped"))) + (func (export "forward-chained") async (canon lift (core func $core "forward-chained"))) + (func (export "forward-cancel-read") async (canon lift (core func $core "forward-cancel-read"))) + (func (export "forward-after-value-read") async (canon lift (core func $core "forward-after-value-read"))) + (func (export "forward-after-value-written") async (canon lift (core func $core "forward-after-value-written"))) + (func (export "forward-after-write-dropped") async (canon lift (core func $core "forward-after-write-dropped"))) + (func (export "forward-removes-readable") async (canon lift (core func $core "forward-removes-readable"))) + (func (export "forward-removes-writable") async (canon lift (core func $core "forward-removes-writable"))) + (func (export "forward-while-reading") async (canon lift (core func $core "forward-while-reading"))) + (func (export "forward-while-writing") async (canon lift (core func $core "forward-while-writing"))) + (func (export "forward-readable-in-waitable-set") async (canon lift (core func $core "forward-readable-in-waitable-set"))) + (func (export "forward-writable-in-waitable-set") async (canon lift (core func $core "forward-writable-in-waitable-set"))) + (func (export "forward-readable-as-writable") async (canon lift (core func $core "forward-readable-as-writable"))) + (func (export "forward-writable-as-readable") async (canon lift (core func $core "forward-writable-as-readable"))) + (func (export "forward-readable-type-mismatch") async (canon lift (core func $core "forward-readable-type-mismatch"))) + (func (export "forward-writable-type-mismatch") async (canon lift (core func $core "forward-writable-type-mismatch"))) + (func (export "self-forward") async (canon lift (core func $core "self-forward"))) + (func (export "forward-cycle") async (canon lift (core func $core "forward-cycle"))) + ) + (instance $c (instantiate $C)) + (instance $d (instantiate $D (with "c" (instance $c)))) + (func (export "forward-rendezvous") (alias export $d "forward-rendezvous")) + (func (export "forward-pending-read") (alias export $d "forward-pending-read")) + (func (export "forward-pending-write") (alias export $d "forward-pending-write")) + (func (export "forward-dst-dropped") (alias export $d "forward-dst-dropped")) + (func (export "forward-dst-already-dropped") (alias export $d "forward-dst-already-dropped")) + (func (export "forward-chained") (alias export $d "forward-chained")) + (func (export "forward-cancel-read") (alias export $d "forward-cancel-read")) + (func (export "forward-after-value-read") (alias export $d "forward-after-value-read")) + (func (export "forward-after-value-written") (alias export $d "forward-after-value-written")) + (func (export "forward-after-write-dropped") (alias export $d "forward-after-write-dropped")) + (func (export "forward-removes-readable") (alias export $d "forward-removes-readable")) + (func (export "forward-removes-writable") (alias export $d "forward-removes-writable")) + (func (export "forward-while-reading") (alias export $d "forward-while-reading")) + (func (export "forward-while-writing") (alias export $d "forward-while-writing")) + (func (export "forward-readable-in-waitable-set") (alias export $d "forward-readable-in-waitable-set")) + (func (export "forward-writable-in-waitable-set") (alias export $d "forward-writable-in-waitable-set")) + (func (export "forward-readable-as-writable") (alias export $d "forward-readable-as-writable")) + (func (export "forward-writable-as-readable") (alias export $d "forward-writable-as-readable")) + (func (export "forward-readable-type-mismatch") (alias export $d "forward-readable-type-mismatch")) + (func (export "forward-writable-type-mismatch") (alias export $d "forward-writable-type-mismatch")) + (func (export "self-forward") (alias export $d "self-forward")) + (func (export "forward-cycle") (alias export $d "forward-cycle")) +) +(component instance $i $Tester) +(assert_return (invoke "forward-rendezvous")) +(component instance $i $Tester) +(assert_return (invoke "forward-pending-read")) +(component instance $i $Tester) +(assert_return (invoke "forward-pending-write")) +(component instance $i $Tester) +(assert_return (invoke "forward-dst-dropped")) +(component instance $i $Tester) +(assert_return (invoke "forward-dst-already-dropped")) +(component instance $i $Tester) +(assert_return (invoke "forward-chained")) +(component instance $i $Tester) +(assert_return (invoke "forward-cancel-read")) +(component instance $i $Tester) +(assert_trap (invoke "forward-after-value-read") "cannot forward future after previous read succeeded") +(component instance $i $Tester) +(assert_trap (invoke "forward-after-value-written") "cannot forward future after previous write succeeded") +(component instance $i $Tester) +(assert_trap (invoke "forward-after-write-dropped") "cannot forward future after being notified that the readable end dropped") +(component instance $i $Tester) +(assert_trap (invoke "forward-removes-readable") "unknown handle index 1") +(component instance $i $Tester) +(assert_trap (invoke "forward-removes-writable") "unknown handle index 3") +(component instance $i $Tester) +(assert_trap (invoke "forward-while-reading") "cannot remove busy future") +(component instance $i $Tester) +(assert_trap (invoke "forward-while-writing") "cannot remove busy future") +(component instance $i $Tester) +(assert_trap (invoke "forward-readable-in-waitable-set") "cannot forward future while it's in a waitable set") +(component instance $i $Tester) +(assert_trap (invoke "forward-writable-in-waitable-set") "cannot forward future while it's in a waitable set") +(component instance $i $Tester) +(assert_trap (invoke "forward-readable-as-writable") "expected writable future end") +(component instance $i $Tester) +(assert_trap (invoke "forward-writable-as-readable") "expected readable future end") +(component instance $i $Tester) +(assert_trap (invoke "forward-readable-type-mismatch") "handle is a future of a different type") +(component instance $i $Tester) +(assert_trap (invoke "forward-writable-type-mismatch") "handle is a future of a different type") +(component instance $i $Tester) +(assert_trap (invoke "self-forward") "cannot forward a future into itself") +(component instance $i $Tester) +(assert_trap (invoke "forward-cycle") "cannot forward a future into itself") diff --git a/test/async/forward-stream.wast b/test/async/forward-stream.wast new file mode 100644 index 00000000..92e09353 --- /dev/null +++ b/test/async/forward-stream.wast @@ -0,0 +1,859 @@ +;; This test contains two components $C and $D where $C writes into a stream +;; it created (the "source", ends $r.src/$w.src) and $D forwards that stream +;; into a second stream (the "destination", ends $r.dst/$w.dst) using the +;; stream.forward built-in, reading the forwarded elements back out of the +;; destination stream itself. +;; +;; $D exports one function per scenario. Since traps take out their containing +;; instance, a fresh instance of $Tester is created for each invoke. +(component definition $Tester + (component $C + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $CM + (import "" "mem" (memory 1)) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (import "" "stream.drop-writable" (func $stream.drop-writable (param i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32))) + + (global $w.src (mut i32) (i32.const 0)) + + (func $start-stream (export "start-stream") (result i32) + ;; create a new stream, return the readable end to the caller + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $w.src (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (i32.wrap_i64 (local.get $ret64)) + ) + (func $write4 (export "write4") + ;; write 4 bytes into the stream, expecting to rendezvous with a read + (local $ret i32) + (i32.store (i32.const 8) (i32.const 0x12345678)) + (local.set $ret (call $stream.write (global.get $w.src) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + ) + (func $write4-dropped (export "write4-dropped") + ;; write expecting to observe DROPPED synchronously + (local $ret i32) + (local.set $ret (call $stream.write (global.get $w.src) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + ) + (func $write0 (export "write0") + ;; zero-length write: completes with 0 elements, leaving a pending + ;; read pending + (local $ret i32) + (local.set $ret (call $stream.write (global.get $w.src) (i32.const 8) (i32.const 0))) + (if (i32.ne (i32.const 0x00 (; COMPLETED=0 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + ) + (func $start-blocking-write (export "start-blocking-write") + (local $ret i32) + + ;; prepare the write buffer + (i64.store (i32.const 8) (i64.const 0x123456789abcdef)) + + ;; start a blocking write + (local.set $ret (call $stream.write (global.get $w.src) (i32.const 8) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ) + (func $check-write-event (export "check-write-event") + ;; confirm the blocking write completed with all 8 elements accepted + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.src) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $w.src) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (i32.const 0x80 (; COMPLETED=0 | (8<<4) ;)) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (global.get $w.src) (i32.const 0)) + ) + (func $check-write-event4 (export "check-write-event4") + ;; confirm the blocking write completed with only 4 of its 8 elements + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.src) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $w.src) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (global.get $w.src) (i32.const 0)) + ) + (func $check-write-dropped (export "check-write-dropped") + ;; confirm the blocking write observed DROPPED with nothing written + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.src) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $w.src) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (global.get $w.src) (i32.const 0)) + ) + (func $check-write-dropped4 (export "check-write-dropped4") + ;; confirm the blocking write observed DROPPED after 4 of its 8 + ;; elements were accepted + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.src) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $w.src) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (i32.const 0x41 (; DROPPED=1 | (4<<4) ;)) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (global.get $w.src) (i32.const 0)) + ) + (func $drop-writable (export "drop-writable") + (call $stream.drop-writable (global.get $w.src)) + ) + ) + (type $ST (stream u8)) + (canon stream.new $ST (core func $stream.new)) + (canon stream.write $ST async (memory (core memory $memory "mem")) (core func $stream.write)) + (canon stream.drop-writable $ST (core func $stream.drop-writable)) + (canon waitable.join (core func $waitable.join)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.wait (memory (core memory $memory "mem")) (core func $waitable-set.wait)) + (core instance $cm (instantiate $CM (with "" (instance + (export "mem" (memory $memory "mem")) + (export "stream.new" (func $stream.new)) + (export "stream.write" (func $stream.write)) + (export "stream.drop-writable" (func $stream.drop-writable)) + (export "waitable.join" (func $waitable.join)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.wait" (func $waitable-set.wait)) + )))) + (func (export "start-stream") async (result (stream u8)) (canon lift (core func $cm "start-stream"))) + (func (export "write4") async (canon lift (core func $cm "write4"))) + (func (export "write4-dropped") async (canon lift (core func $cm "write4-dropped"))) + (func (export "write0") async (canon lift (core func $cm "write0"))) + (func (export "start-blocking-write") async (canon lift (core func $cm "start-blocking-write"))) + (func (export "check-write-event4") async (canon lift (core func $cm "check-write-event4"))) + (func (export "check-write-event") async (canon lift (core func $cm "check-write-event"))) + (func (export "check-write-dropped") async (canon lift (core func $cm "check-write-dropped"))) + (func (export "check-write-dropped4") async (canon lift (core func $cm "check-write-dropped4"))) + (func (export "drop-writable") async (canon lift (core func $cm "drop-writable"))) + ) + (component $D + (import "c" (instance $c + (export "start-stream" (func async (result (stream u8)))) + (export "write4" (func async)) + (export "write4-dropped" (func async)) + (export "write0" (func async)) + (export "start-blocking-write" (func async)) + (export "check-write-event" (func async)) + (export "check-write-event4" (func async)) + (export "check-write-dropped" (func async)) + (export "check-write-dropped4" (func async)) + (export "drop-writable" (func async)) + )) + + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $Core + (import "" "mem" (memory 1)) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.new-u16" (func $stream.new-u16 (result i64))) + (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.read-async" (func $stream.read-async (param i32 i32 i32) (result i32))) + (import "" "stream.write-async" (func $stream.write-async (param i32 i32 i32) (result i32))) + (import "" "stream.forward" (func $stream.forward (param i32 i32))) + (import "" "stream.cancel-read" (func $stream.cancel-read (param i32) (result i32))) + (import "" "stream.drop-readable" (func $stream.drop-readable (param i32))) + (import "" "stream.drop-writable" (func $stream.drop-writable (param i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32))) + (import "" "start-stream" (func $start-stream (result i32))) + (import "" "write4" (func $write4)) + (import "" "write4-dropped" (func $write4-dropped)) + (import "" "write0" (func $write0)) + (import "" "start-blocking-write" (func $start-blocking-write)) + (import "" "check-write-event" (func $check-write-event)) + (import "" "check-write-event4" (func $check-write-event4)) + (import "" "check-write-dropped" (func $check-write-dropped)) + (import "" "check-write-dropped4" (func $check-write-dropped4)) + (import "" "drop-writable" (func $drop-writable)) + + (global $r.src (mut i32) (i32.const 0)) + (global $r.dst (mut i32) (i32.const 0)) + (global $w.dst (mut i32) (i32.const 0)) + + (func $setup + ;; get the source stream from $C and create the destination stream + (local $ret64 i64) + (global.set $r.src (call $start-stream)) + (if (i32.ne (i32.const 1) (global.get $r.src)) + (then unreachable)) + (local.set $ret64 (call $stream.new)) + (global.set $r.dst (i32.wrap_i64 (local.get $ret64))) + (global.set $w.dst (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (if (i32.ne (i32.const 2) (global.get $r.dst)) + (then unreachable)) + (if (i32.ne (i32.const 3) (global.get $w.dst)) + (then unreachable)) + ) + (func $expect-event (param $waitable i32) (param $event i32) (param $payload i32) + ;; wait for the given event on the given waitable with the given payload + (local $ret i32) (local $seti i32) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (local.get $waitable) (local.get $seti)) + (local.set $ret (call $waitable-set.wait (local.get $seti) (i32.const 0))) + (if (i32.ne (local.get $event) (local.get $ret)) + (then unreachable)) + (if (i32.ne (local.get $waitable) (i32.load (i32.const 0))) + (then unreachable)) + (if (i32.ne (local.get $payload) (i32.load (i32.const 4))) + (then unreachable)) + (call $waitable.join (local.get $waitable) (i32.const 0)) + ) + (func $read4-dst (param $expected i32) + ;; synchronously read 4 bytes out of the destination stream + (local $ret i32) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (local.get $expected) (i32.load (i32.const 8))) + (then unreachable)) + ) + (func $read4-dst-async + ;; start a read on the destination stream that will block + (local $ret i32) + (local.set $ret (call $stream.read-async (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ) + (func $expect-read4-dst + (call $expect-event (global.get $r.dst) (i32.const 2 (; STREAM_READ ;)) (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;))) + (if (i32.ne (i32.const 0x12345678) (i32.load (i32.const 8))) + (then unreachable)) + ) + (func $pump4 + ;; read, block, write 4 bytes on the source, confirm the read completes + (call $read4-dst-async) + (call $write4) + (call $expect-read4-dst) + ) + + (func (export "forward-rendezvous") + (local $ret i32) + (call $setup) + + ;; forward the source into the destination; this returns immediately + ;; and no event is ever delivered + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + + ;; with no read pending on the destination, $C's 8-byte write blocks + (call $start-blocking-write) + + ;; two reads drain $C's write buffer, completing its write + (call $read4-dst (i32.const 0x89abcdef)) + (call $read4-dst (i32.const 0x01234567)) + (call $check-write-event) + + ;; a blocked read on the destination rendezvous with a later write + (call $pump4) + + ;; the end of the source stream propagates to the destination's reader + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-pending-read") + (local $ret i32) + (call $setup) + + ;; a read blocked on the destination when the forward starts is + ;; transferred to the source + (call $read4-dst-async) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $write4) + (call $expect-read4-dst) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-pending-write") + (local $ret i32) + (call $setup) + + ;; a write blocked on the source when the forward starts rendezvous + ;; with later destination reads + (call $start-blocking-write) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $read4-dst (i32.const 0x89abcdef)) + (call $read4-dst (i32.const 0x01234567)) + (call $check-write-event) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-eos-blocked-read") + (call $setup) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + + ;; a read blocked on the destination observes the end of the source + (call $read4-dst-async) + (call $drop-writable) + (call $expect-event (global.get $r.dst) (i32.const 2 (; STREAM_READ ;)) (i32.const 0x01 (; DROPPED=1 | (0<<4) ;))) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-dst-dropped") + (call $setup) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $start-blocking-write) + + ;; dropping the destination's readable end drops the source's + ;; readable end, so $C's blocked write observes DROPPED + (call $stream.drop-readable (global.get $r.dst)) + (call $check-write-dropped) + (call $drop-writable) + ) + + (func (export "forward-dst-already-dropped") + (call $setup) + + ;; forwarding into an already-dropped destination drops the source's + ;; readable end right away + (call $stream.drop-readable (global.get $r.dst)) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $write4-dropped) + (call $drop-writable) + ) + + (func (export "forward-chained") + (local $ret i32) (local $ret64 i64) (local $r.mid i32) (local $w.mid i32) + (call $setup) + (local.set $ret64 (call $stream.new)) + (local.set $r.mid (i32.wrap_i64 (local.get $ret64))) + (local.set $w.mid (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + + ;; chain two forwards through an intermediate stream: reads and the + ;; end of the source stream propagate through both + (call $stream.forward (global.get $r.src) (local.get $w.mid)) + (call $stream.forward (local.get $r.mid) (global.get $w.dst)) + (call $pump4) + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-zero-length-read") + (local $ret i32) + (call $setup) + + ;; a blocked zero-length read is transferred to the source and + ;; completes on the next write without consuming any elements + (local.set $ret (call $stream.read-async (global.get $r.dst) (i32.const 8) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $start-blocking-write) + (call $expect-event (global.get $r.dst) (i32.const 2 (; STREAM_READ ;)) (i32.const 0x00 (; COMPLETED=0 | (0<<4) ;))) + + ;; the write is still pending and is drained by ordinary reads + (call $read4-dst (i32.const 0x89abcdef)) + (call $read4-dst (i32.const 0x01234567)) + (call $check-write-event) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-zero-length-write") + (local $ret i32) + (call $setup) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + + ;; a zero-length write leaves the forwarded read pending + (call $read4-dst-async) + (call $write0) + (call $write4) + (call $expect-read4-dst) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-partially-completed-read") + (local $ret i32) + (call $setup) + + ;; leave a read pending on the destination with 4 of its 8 requested + ;; elements already written + (local.set $ret (call $stream.read-async (global.get $r.dst) (i32.const 8) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (i32.store (i32.const 16) (i32.const 0xdeadbeef)) + (local.set $ret (call $stream.write-async (global.get $w.dst) (i32.const 16) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + + ;; a partially completed read is completed (rather than transferred) + ;; by the forward + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $expect-event (global.get $r.dst) (i32.const 2 (; STREAM_READ ;)) (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;))) + (if (i32.ne (i32.const 0xdeadbeef) (i32.load (i32.const 8))) + (then unreachable)) + + ;; a fresh read is routed to the source as usual + (call $pump4) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-partially-completed-write") + (local $ret i32) + (call $setup) + + ;; leave a write pending on the source with 4 of its 8 elements + ;; already read + (call $start-blocking-write) + (local.set $ret (call $stream.read (global.get $r.src) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x89abcdef) (i32.load (i32.const 8))) + (then unreachable)) + + ;; a partially completed write is completed (rather than transferred) + ;; by the forward + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $check-write-event4) + + ;; a fresh write rendezvous with a read on the destination as usual + (call $pump4) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-partially-completed-read-dropped") + (local $ret i32) + (call $setup) + + ;; leave a read pending on the destination with 4 of its 8 requested + ;; elements already written + (local.set $ret (call $stream.read-async (global.get $r.dst) (i32.const 8) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (i32.store (i32.const 16) (i32.const 0xdeadbeef)) + (local.set $ret (call $stream.write-async (global.get $w.dst) (i32.const 16) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + + ;; the source's writer is already gone when the forward happens, so + ;; the queued COMPLETED(4) completion is merged with the drop + ;; notification into a single DROPPED(4) event + (call $drop-writable) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (local.set $ret (call $stream.cancel-read (global.get $r.dst))) + (if (i32.ne (i32.const 0x41 (; DROPPED=1 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0xdeadbeef) (i32.load (i32.const 8))) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-partially-completed-write-dropped") + (local $ret i32) + (call $setup) + + ;; leave a write pending on the source with 4 of its 8 elements + ;; already read + (call $start-blocking-write) + (local.set $ret (call $stream.read (global.get $r.src) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x89abcdef) (i32.load (i32.const 8))) + (then unreachable)) + + ;; the destination's reader is already gone when the forward happens, + ;; so $C's write observes its partial completion and the drop + ;; notification as a single DROPPED(4) event + (call $stream.drop-readable (global.get $r.dst)) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $check-write-dropped4) + (call $drop-writable) + ) + + (func (export "forward-cancel-read") + (local $ret i32) + (call $setup) + + ;; a read transferred to the source can still be cancelled through + ;; the destination's readable end, after which the stream remains + ;; usable + (call $read4-dst-async) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (local.set $ret (call $stream.cancel-read (global.get $r.dst))) + (if (i32.ne (i32.const 0x02 (; CANCELLED=2 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $pump4) + + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-src-writer-already-dropped") + (call $setup) + + ;; a source whose writer is already gone when the forward starts is + ;; observed as end-of-stream via the destination's readable end + (call $drop-writable) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + (call $expect-event (global.get $r.dst) (i32.const 2 (; STREAM_READ ;)) (i32.const 0x01 (; DROPPED=1 | (0<<4) ;))) + (call $stream.drop-readable (global.get $r.dst)) + ) + + (func (export "forward-after-eos") + (local $ret i32) + (call $setup) + + ;; a readable end that observed end-of-stream can no longer be + ;; forwarded + (call $drop-writable) + (local.set $ret (call $stream.read (global.get $r.src) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-after-write-dropped") + (local $ret i32) + (call $setup) + + ;; a writable end that observed DROPPED can no longer be the target + ;; of a forward + (call $stream.drop-readable (global.get $r.dst)) + (local.set $ret (call $stream.write-async (global.get $w.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x01 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-removes-readable") + (call $setup) + ;; stream.forward removes both ends from the table + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ;; boom + (call $stream.drop-readable (global.get $r.src)) + ) + + (func (export "forward-removes-writable") + (call $setup) + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ;; boom + (call $stream.drop-writable (global.get $w.dst)) + ) + + (func (export "forward-while-reading") + (local $ret i32) + (call $setup) + (local.set $ret (call $stream.read-async (global.get $r.src) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-while-writing") + (local $ret i32) + (call $setup) + (local.set $ret (call $stream.write-async (global.get $w.dst) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-readable-in-waitable-set") + (local $seti i32) + (call $setup) + ;; forwarding an end that is in a waitable set traps + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $r.src) (local.get $seti)) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-writable-in-waitable-set") + (local $seti i32) + (call $setup) + (local.set $seti (call $waitable-set.new)) + (call $waitable.join (global.get $w.dst) (local.get $seti)) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $w.dst)) + ) + + (func (export "forward-readable-as-writable") + (call $setup) + ;; boom + (call $stream.forward (global.get $r.src) (global.get $r.dst)) + ) + + (func (export "forward-writable-as-readable") + (call $setup) + ;; boom + (call $stream.forward (global.get $w.dst) (global.get $w.dst)) + ) + + (func (export "forward-readable-type-mismatch") + (local $ret64 i64) + (call $setup) + ;; the element type of the readable end must match the type immediate + (local.set $ret64 (call $stream.new-u16)) + ;; boom + (call $stream.forward (i32.wrap_i64 (local.get $ret64)) (global.get $w.dst)) + ) + + (func (export "forward-writable-type-mismatch") + (local $ret64 i64) + (call $setup) + ;; the element type of the writable end must match the type immediate + (local.set $ret64 (call $stream.new-u16)) + ;; boom + (call $stream.forward (global.get $r.src) (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + + (func (export "self-forward") + (local $ret64 i64) (local $r.self i32) (local $w.self i32) + ;; a stream cannot be forwarded into itself + (local.set $ret64 (call $stream.new)) + (local.set $r.self (i32.wrap_i64 (local.get $ret64))) + (local.set $w.self (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ;; boom + (call $stream.forward (local.get $r.self) (local.get $w.self)) + ) + + (func (export "forward-cycle") + (local $ret64 i64) (local $r.a i32) (local $w.a i32) (local $r.b i32) (local $w.b i32) + (local.set $ret64 (call $stream.new)) + (local.set $r.a (i32.wrap_i64 (local.get $ret64))) + (local.set $w.a (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (local.set $ret64 (call $stream.new)) + (local.set $r.b (i32.wrap_i64 (local.get $ret64))) + (local.set $w.b (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + + (call $stream.forward (local.get $r.a) (local.get $w.b)) + ;; forwarding $b back into $a would close a cycle + ;; boom + (call $stream.forward (local.get $r.b) (local.get $w.a)) + ) + ) + (type $ST (stream u8)) + (canon stream.new $ST (core func $stream.new)) + (type $STU16 (stream u16)) + (canon stream.new $STU16 (core func $stream.new-u16)) + (canon stream.read $ST (memory (core memory $memory "mem")) (core func $stream.read)) + (canon stream.read $ST async (memory (core memory $memory "mem")) (core func $stream.read-async)) + (canon stream.write $ST async (memory (core memory $memory "mem")) (core func $stream.write-async)) + (canon stream.forward $ST (core func $stream.forward)) + (canon stream.cancel-read $ST (core func $stream.cancel-read)) + (canon stream.drop-readable $ST (core func $stream.drop-readable)) + (canon stream.drop-writable $ST (core func $stream.drop-writable)) + (canon waitable.join (core func $waitable.join)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.wait (memory (core memory $memory "mem")) (core func $waitable-set.wait)) + (canon lower (func $c "start-stream") (core func $start-stream')) + (canon lower (func $c "write4") (core func $write4')) + (canon lower (func $c "write4-dropped") (core func $write4-dropped')) + (canon lower (func $c "write0") (core func $write0')) + (canon lower (func $c "start-blocking-write") (core func $start-blocking-write')) + (canon lower (func $c "check-write-event4") (core func $check-write-event4')) + (canon lower (func $c "check-write-event") (core func $check-write-event')) + (canon lower (func $c "check-write-dropped") (core func $check-write-dropped')) + (canon lower (func $c "check-write-dropped4") (core func $check-write-dropped4')) + (canon lower (func $c "drop-writable") (core func $drop-writable')) + (core instance $core (instantiate $Core (with "" (instance + (export "mem" (memory $memory "mem")) + (export "stream.new" (func $stream.new)) + (export "stream.new-u16" (func $stream.new-u16)) + (export "stream.read" (func $stream.read)) + (export "stream.read-async" (func $stream.read-async)) + (export "stream.write-async" (func $stream.write-async)) + (export "stream.forward" (func $stream.forward)) + (export "stream.cancel-read" (func $stream.cancel-read)) + (export "stream.drop-readable" (func $stream.drop-readable)) + (export "stream.drop-writable" (func $stream.drop-writable)) + (export "waitable.join" (func $waitable.join)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.wait" (func $waitable-set.wait)) + (export "start-stream" (func $start-stream')) + (export "write4" (func $write4')) + (export "write4-dropped" (func $write4-dropped')) + (export "write0" (func $write0')) + (export "start-blocking-write" (func $start-blocking-write')) + (export "check-write-event" (func $check-write-event')) + (export "check-write-event4" (func $check-write-event4')) + (export "check-write-dropped" (func $check-write-dropped')) + (export "check-write-dropped4" (func $check-write-dropped4')) + (export "drop-writable" (func $drop-writable')) + )))) + (func (export "forward-rendezvous") async (canon lift (core func $core "forward-rendezvous"))) + (func (export "forward-pending-read") async (canon lift (core func $core "forward-pending-read"))) + (func (export "forward-pending-write") async (canon lift (core func $core "forward-pending-write"))) + (func (export "forward-eos-blocked-read") async (canon lift (core func $core "forward-eos-blocked-read"))) + (func (export "forward-dst-dropped") async (canon lift (core func $core "forward-dst-dropped"))) + (func (export "forward-dst-already-dropped") async (canon lift (core func $core "forward-dst-already-dropped"))) + (func (export "forward-chained") async (canon lift (core func $core "forward-chained"))) + (func (export "forward-zero-length-read") async (canon lift (core func $core "forward-zero-length-read"))) + (func (export "forward-zero-length-write") async (canon lift (core func $core "forward-zero-length-write"))) + (func (export "forward-partially-completed-read") async (canon lift (core func $core "forward-partially-completed-read"))) + (func (export "forward-partially-completed-write") async (canon lift (core func $core "forward-partially-completed-write"))) + (func (export "forward-partially-completed-read-dropped") async (canon lift (core func $core "forward-partially-completed-read-dropped"))) + (func (export "forward-partially-completed-write-dropped") async (canon lift (core func $core "forward-partially-completed-write-dropped"))) + (func (export "forward-cancel-read") async (canon lift (core func $core "forward-cancel-read"))) + (func (export "forward-src-writer-already-dropped") async (canon lift (core func $core "forward-src-writer-already-dropped"))) + (func (export "forward-after-eos") async (canon lift (core func $core "forward-after-eos"))) + (func (export "forward-after-write-dropped") async (canon lift (core func $core "forward-after-write-dropped"))) + (func (export "forward-removes-readable") async (canon lift (core func $core "forward-removes-readable"))) + (func (export "forward-removes-writable") async (canon lift (core func $core "forward-removes-writable"))) + (func (export "forward-while-reading") async (canon lift (core func $core "forward-while-reading"))) + (func (export "forward-while-writing") async (canon lift (core func $core "forward-while-writing"))) + (func (export "forward-readable-in-waitable-set") async (canon lift (core func $core "forward-readable-in-waitable-set"))) + (func (export "forward-writable-in-waitable-set") async (canon lift (core func $core "forward-writable-in-waitable-set"))) + (func (export "forward-readable-as-writable") async (canon lift (core func $core "forward-readable-as-writable"))) + (func (export "forward-writable-as-readable") async (canon lift (core func $core "forward-writable-as-readable"))) + (func (export "forward-readable-type-mismatch") async (canon lift (core func $core "forward-readable-type-mismatch"))) + (func (export "forward-writable-type-mismatch") async (canon lift (core func $core "forward-writable-type-mismatch"))) + (func (export "self-forward") async (canon lift (core func $core "self-forward"))) + (func (export "forward-cycle") async (canon lift (core func $core "forward-cycle"))) + ) + (instance $c (instantiate $C)) + (instance $d (instantiate $D (with "c" (instance $c)))) + (func (export "forward-rendezvous") (alias export $d "forward-rendezvous")) + (func (export "forward-pending-read") (alias export $d "forward-pending-read")) + (func (export "forward-pending-write") (alias export $d "forward-pending-write")) + (func (export "forward-eos-blocked-read") (alias export $d "forward-eos-blocked-read")) + (func (export "forward-dst-dropped") (alias export $d "forward-dst-dropped")) + (func (export "forward-dst-already-dropped") (alias export $d "forward-dst-already-dropped")) + (func (export "forward-chained") (alias export $d "forward-chained")) + (func (export "forward-zero-length-read") (alias export $d "forward-zero-length-read")) + (func (export "forward-zero-length-write") (alias export $d "forward-zero-length-write")) + (func (export "forward-partially-completed-read") (alias export $d "forward-partially-completed-read")) + (func (export "forward-partially-completed-write") (alias export $d "forward-partially-completed-write")) + (func (export "forward-partially-completed-read-dropped") (alias export $d "forward-partially-completed-read-dropped")) + (func (export "forward-partially-completed-write-dropped") (alias export $d "forward-partially-completed-write-dropped")) + (func (export "forward-cancel-read") (alias export $d "forward-cancel-read")) + (func (export "forward-src-writer-already-dropped") (alias export $d "forward-src-writer-already-dropped")) + (func (export "forward-after-eos") (alias export $d "forward-after-eos")) + (func (export "forward-after-write-dropped") (alias export $d "forward-after-write-dropped")) + (func (export "forward-removes-readable") (alias export $d "forward-removes-readable")) + (func (export "forward-removes-writable") (alias export $d "forward-removes-writable")) + (func (export "forward-while-reading") (alias export $d "forward-while-reading")) + (func (export "forward-while-writing") (alias export $d "forward-while-writing")) + (func (export "forward-readable-in-waitable-set") (alias export $d "forward-readable-in-waitable-set")) + (func (export "forward-writable-in-waitable-set") (alias export $d "forward-writable-in-waitable-set")) + (func (export "forward-readable-as-writable") (alias export $d "forward-readable-as-writable")) + (func (export "forward-writable-as-readable") (alias export $d "forward-writable-as-readable")) + (func (export "forward-readable-type-mismatch") (alias export $d "forward-readable-type-mismatch")) + (func (export "forward-writable-type-mismatch") (alias export $d "forward-writable-type-mismatch")) + (func (export "self-forward") (alias export $d "self-forward")) + (func (export "forward-cycle") (alias export $d "forward-cycle")) +) +(component instance $i $Tester) +(assert_return (invoke "forward-rendezvous")) +(component instance $i $Tester) +(assert_return (invoke "forward-pending-read")) +(component instance $i $Tester) +(assert_return (invoke "forward-pending-write")) +(component instance $i $Tester) +(assert_return (invoke "forward-eos-blocked-read")) +(component instance $i $Tester) +(assert_return (invoke "forward-dst-dropped")) +(component instance $i $Tester) +(assert_return (invoke "forward-dst-already-dropped")) +(component instance $i $Tester) +(assert_return (invoke "forward-chained")) +(component instance $i $Tester) +(assert_return (invoke "forward-zero-length-read")) +(component instance $i $Tester) +(assert_return (invoke "forward-zero-length-write")) +(component instance $i $Tester) +(assert_return (invoke "forward-partially-completed-read")) +(component instance $i $Tester) +(assert_return (invoke "forward-partially-completed-write")) +(component instance $i $Tester) +(assert_return (invoke "forward-partially-completed-read-dropped")) +(component instance $i $Tester) +(assert_return (invoke "forward-partially-completed-write-dropped")) +(component instance $i $Tester) +(assert_return (invoke "forward-cancel-read")) +(component instance $i $Tester) +(assert_return (invoke "forward-src-writer-already-dropped")) +(component instance $i $Tester) +(assert_trap (invoke "forward-after-eos") "cannot forward stream after being notified that the writable end dropped") +(component instance $i $Tester) +(assert_trap (invoke "forward-after-write-dropped") "cannot forward stream after being notified that the readable end dropped") +(component instance $i $Tester) +(assert_trap (invoke "forward-removes-readable") "unknown handle index 1") +(component instance $i $Tester) +(assert_trap (invoke "forward-removes-writable") "unknown handle index 3") +(component instance $i $Tester) +(assert_trap (invoke "forward-while-reading") "cannot remove busy stream") +(component instance $i $Tester) +(assert_trap (invoke "forward-while-writing") "cannot remove busy stream") +(component instance $i $Tester) +(assert_trap (invoke "forward-readable-in-waitable-set") "cannot forward stream while it's in a waitable set") +(component instance $i $Tester) +(assert_trap (invoke "forward-writable-in-waitable-set") "cannot forward stream while it's in a waitable set") +(component instance $i $Tester) +(assert_trap (invoke "forward-readable-as-writable") "expected writable stream end") +(component instance $i $Tester) +(assert_trap (invoke "forward-writable-as-readable") "expected readable stream end") +(component instance $i $Tester) +(assert_trap (invoke "forward-readable-type-mismatch") "handle is a stream of a different type") +(component instance $i $Tester) +(assert_trap (invoke "forward-writable-type-mismatch") "handle is a stream of a different type") +(component instance $i $Tester) +(assert_trap (invoke "self-forward") "cannot forward a stream into itself") +(component instance $i $Tester) +(assert_trap (invoke "forward-cycle") "cannot forward a stream into itself") diff --git a/test/nyi.txt b/test/nyi.txt index 8f3e4250..2497197a 100644 --- a/test/nyi.txt +++ b/test/nyi.txt @@ -3,3 +3,5 @@ ./async/during-sync-call-may-block-if-other-ready-threads.wast ./async/during-sync-call-no-exclusive-resume.wast ./async/during-sync-call-no-sibling-resume.wast +./async/forward-stream.wast +./async/forward-future.wast