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 0c222b79..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) ๐Ÿงต @@ -1751,7 +1752,7 @@ class BufferGuestImpl(Buffer): def is_zero_length(self): return self.length == 0 -class ReadableBufferGuestImpl(BufferGuestImpl): +class ReadableBufferGuestImpl(BufferGuestImpl, ReadableBuffer): def read(self, n): assert(n <= self.remain()) if self.t: @@ -1878,6 +1879,7 @@ Introducing `SharedStreamImpl` in chunks, starting with the fields and initializ ```python class SharedStreamImpl(ReadableStream, WritableStream): dropped: bool + forward: Optional[ReadableStream] pending_inst: Optional[ComponentInstance] pending_buffer: Optional[Buffer] pending_on_copy: Optional[OnCopy] @@ -1886,6 +1888,7 @@ class SharedStreamImpl(ReadableStream, WritableStream): def __init__(self, t): self.t = t self.dropped = False + self.forward = None self.reset_pending() def reset_pending(self): @@ -1896,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,23 +2137,33 @@ class SharedFutureImpl(ReadableFuture, WritableFuture): self.pending_buffer = buffer self.pending_on_copy_done = on_copy_done + def take_pending(self): + pending = (self.pending_inst, self.pending_buffer, self.pending_on_copy_done) + self.reset_pending() + return pending + def reset_and_notify_pending(self, result): pending_on_copy_done = self.pending_on_copy_done self.reset_pending() pending_on_copy_done(result) def cancel(self): - self.reset_and_notify_pending(CopyResult.CANCELLED) + if self.forward: + self.forward.cancel() + else: + self.reset_and_notify_pending(CopyResult.CANCELLED) ``` Dropping works the same in futures as in streams, except that a future writable end cannot be dropped without having written a value. This is guarded by `WritableFutureEnd.drop` so it can be asserted here: ```python def drop(self): - if not self.dropped: + if self.forward: + self.forward.drop() + elif not self.dropped: self.dropped = True if self.pending_buffer: - assert(isinstance(self.pending_buffer, WritableBuffer)) + assert(isinstance(self.pending_buffer, ReadableBuffer)) self.reset_and_notify_pending(CopyResult.DROPPED) ``` Lastly, `read` and `write` work mostly like streams, but simplified based on @@ -2133,6 +2172,9 @@ that, as mentioned above, only the writable end can observe that the readable end was dropped before receiving a value. ```python def read(self, inst, dst_buffer, on_copy_done): + if self.forward: + self.forward.read(inst, dst_buffer, on_copy_done) + return assert(not self.dropped and dst_buffer.remain() == 1) if not self.pending_buffer: self.set_pending(inst, dst_buffer, on_copy_done) @@ -4762,6 +4804,97 @@ synchronously and returning either the progress made or `BLOCKED`. ``` +### โฉ `canon {stream,future}.forward` + +For canonical definitions: +```wat +(canon stream.forward $stream_t (core func $f)) +(canon future.forward $future_t (core func $f)) +``` +validation specifies: +* `$f` is given type `(func (param i32 i32))` +* `$stream_t` must be a type of the form `(stream $t?)` +* `$future_t` must be a type of the form `(future $t?)` + +Calling `$f` forwards *all* remaining elements of the readable stream end at +index `$ri` (resp., the value of the readable future end at index `$ri`) into +the writable end at index `$wi` of another stream (resp., future) of the same +type. `{stream,future}.forward` *transfers* both ends out of the calling +component instance: the ends are removed from the `handles` table, `$f` +returns immediately with no result and no event is ever delivered. + +The `canon_stream_forward` and `canon_future_forward` functions share a single +implementation, `forward`, that sets the destination's `forward` field +(defined in [Stream State](#stream-state) and [Future State](#future-state) +above), permanently delegating every subsequent `read`, `cancel` and `drop` of +the destination to the source, and transfers any consumer `read` already +blocked on the destination to the source (performing rendezvous immediately if +the source has a pending `write`). If the destination's readable end has +already been dropped, there is nothing left to forward into, and so the source +is simply dropped, notifying the source's writer: +```python +def canon_stream_forward(stream_t, ri, wi): + return forward(ReadableStreamEnd, WritableStreamEnd, stream_t, ri, wi) + +def canon_future_forward(future_t, ri, wi): + return forward(ReadableFutureEnd, WritableFutureEnd, future_t, ri, wi) + +def forward_source(shared): + while isinstance(shared, SharedStreamImpl | SharedFutureImpl) and shared.forward: + shared = shared.forward + return shared + +def forward(ReadableEndT, WritableEndT, stream_or_future_t, ri, wi): + inst = current_instance() + trap_if(not inst.may_leave) + r = inst.handles.get(ri) + trap_if(not isinstance(r, ReadableEndT)) + trap_if(r.shared.t != stream_or_future_t.t) + trap_if(r.state != CopyState.IDLE) + trap_if(r.in_waitable_set()) + w = inst.handles.get(wi) + trap_if(not isinstance(w, WritableEndT)) + trap_if(w.shared.t != stream_or_future_t.t) + trap_if(w.state != CopyState.IDLE) + trap_if(w.in_waitable_set()) + trap_if(forward_source(r.shared) is w.shared) + assert(not contains_borrow(stream_or_future_t)) + + inst.handles.remove(ri) + inst.handles.remove(wi) + + if w.shared.dropped: + r.shared.drop() + else: + w.shared.forward = r.shared + if w.shared.pending_buffer: + pending = w.shared.take_pending() + r.shared.read(*pending) + return [] +``` +Since `{stream,future}.forward` transfers both ends out of the component +instance, it traps if either end is currently in a waitable set, exactly as +when a readable end is transferred to another component (in `lift_async_value` +above). Forwards may be chained, but a forward that would (transitively) make +a stream or future its own source traps; the `forward_source` helper follows +an existing chain of forwards to the stream or future that ultimately serves +as the source of a given stream or future. + +Once `{stream,future}.forward` returns, the source and destination have been +fused into one: the producer and consumer on either side observe exactly what +they would observe if the source's readable end had been transferred directly +to the consumer in place of the destination's readable end. Elements +rendezvous end-to-end with no intermediate buffering (as described in +[Stream State](#stream-state) above) and drops propagate in both directions: +when the source stream reaches its end (resp., the source future's value is +written), the consumer observes it and, when the consumer drops the +destination's readable end, the drop is delegated to the source where the +producer observes it. This gives a host the opportunity to remove the calling +component from the copy path entirely which, in chains of forwarding +components, allows forwarding state (and even whole component instances that +are no longer otherwise reachable) to be eagerly torn down. + + ### ๐Ÿ”€ `canon {stream,future}.cancel-{read,write}` For canonical definitions: 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 99fb6fc6..ef906919 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: @@ -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, WritableBuffer)) + 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 b4bd0c34..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) @@ -2308,6 +3158,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) @@ -3032,13 +3918,29 @@ 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() 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()) 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