diff --git a/design/mvp/Binary.md b/design/mvp/Binary.md index 9a022d72..80903f5a 100644 --- a/design/mvp/Binary.md +++ b/design/mvp/Binary.md @@ -316,6 +316,7 @@ canon ::= 0x00 0x00 f: opts: ft: => (canon lift | 0x12 t: async?: => (canon stream.cancel-write t async? (core func)) ๐Ÿ”€ | 0x13 t: => (canon stream.drop-readable t (core func)) ๐Ÿ”€ | 0x14 t: => (canon stream.drop-writable t (core func)) ๐Ÿ”€ + | 0x2e t: => (canon stream.forward t (core func)) โžก๏ธ | 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)) ๐Ÿ”€ @@ -323,6 +324,7 @@ canon ::= 0x00 0x00 f: opts: ft: => (canon lift | 0x19 t: async?: => (canon future.cancel-write t async? (core func)) ๐Ÿ”€ | 0x1a t: => (canon future.drop-readable t (core func)) ๐Ÿ”€ | 0x1b t: => (canon future.drop-writable t (core func)) ๐Ÿ”€ + | 0x2f t: => (canon future.forward t (core func)) โžก๏ธ | 0x1c opts: => (canon error-context.new opts (core func)) ๐Ÿ“ | 0x1d opts: => (canon error-context.debug-message opts (core func)) ๐Ÿ“ | 0x1e => (canon error-context.drop (core func)) ๐Ÿ“ diff --git a/design/mvp/CanonicalABI.md b/design/mvp/CanonicalABI.md index 2b05c97b..58681669 100644 --- a/design/mvp/CanonicalABI.md +++ b/design/mvp/CanonicalABI.md @@ -56,6 +56,7 @@ specified here. * [`canon {stream,future}.{read,write}`](#-canon-streamfuturereadwrite) ๐Ÿ”€ * [`canon {stream,future}.cancel-{read,write}`](#-canon-streamfuturecancel-readwrite) ๐Ÿ”€ * [`canon {stream,future}.drop-{readable,writable}`](#-canon-streamfuturedrop-readablewritable) ๐Ÿ”€ + * [`canon {stream,future}.forward`](#-canon-streamfutureforward) โžก๏ธ * [`canon thread.index`](#-canon-threadindex) ๐Ÿงต * [`canon thread.new-indirect`](#-canon-threadnew-indirect) ๐Ÿงต * [`canon thread.resume-later`](#-canon-threadresume-later) ๐Ÿงต @@ -1696,10 +1697,41 @@ cancellation. In the future, guest components may be given the same capability. self.notify(progress = 0) ``` -The `End.drop` method is called by `{stream,future}.drop-{readable,writable}` to -update the `other` end's state and possibly set a pending notification for the -other end, if the other end isn't already `DONE` and doing so wouldn't clobber -an already-pending notification. +The `End.forward` function is called by `{stream,future}.forward` to efficiently +forward all the values from a given readable end into a given writable end, +propagating `DROPPED` results in both directions. This causes the given readable +end and writable end to disappear, leaving only the `other` writable and +readable ends, which are now linked together directly to form a single stream or +future. If *both* of these formerly-separate ends have pending copy operations +with pending `buffer`s, the forwarding operation performs a synchronous copy, +leaving at most one pending buffer (the bigger of the two) and notifying one or +both sides of the progress made. In the corner case where a stream or future's +readable end is forwarded to its own writable end (creating a trivial recursive +loop), there is no trap since the whole thing simply disappears. +```python + def forward(src: End, dst: End): + if src.other is dst or src.other is None or dst.other is None: + src.drop() + dst.drop() + else: + writable_end = src.other + readable_end = dst.other + writable_end.other = readable_end + readable_end.other = writable_end + if writable_end.buffer is not None and readable_end.buffer is not None: + if readable_end.buffer.remain() > writable_end.buffer.remain(): + bigger_end, smaller_end = readable_end, writable_end + else: + bigger_end, smaller_end = writable_end, readable_end + buffer = smaller_end.buffer + smaller_end.buffer = None + smaller_end.copy(buffer) +``` + +The `End.drop` method is called by `{stream,future}.drop-{readable,writable}` +and `End.forward` to update the `other` end's state and possibly set a pending +notification for the other end, if the other end isn't already `DONE` and doing +so wouldn't clobber an already-pending notification. ```python def drop(self): assert(not self.copying_or_cancelling()) @@ -4446,6 +4478,46 @@ def drop(EndT, stream_or_future_t, i): ``` +### โžก๏ธ `canon {stream,future}.forward` + +For canonical definitions: +```wat +(canon stream.forward $stream_t (core func $forward)) +(canon future.forward $future_t (core func $forward)) +``` +validation specifies: +* `$forward` is given type `(func (param $ri i32) (param $wi i32))` +* `$stream_t`/`$future_t` must be a type of the form `(stream $t?)`/`(future $t?)` + +Calling `$forward` removes the readable and writable ends at the given indices, +after checking that all the types match, the ends are in the `IDLE` state, and +the ends are not currently part of a waitable set. Then the readable end is +forwarded into the writable end as defined by `End.forward` above. +```python +def canon_stream_forward(stream_t, ri, wi): + return forward(ReadableStreamEnd, WritableStreamEnd, stream_t, ri, wi) + +def canon_future_forward(future_t, ri, wi): + return forward(ReadableFutureEnd, WritableFutureEnd, future_t, ri, wi) + +def forward(ReadableEndT, WritableEndT, stream_or_future_t, ri, wi): + inst = current_instance() + trap_if(not inst.may_leave) + readable_end = inst.handles.remove(ri) + trap_if(not isinstance(readable_end, ReadableEndT)) + trap_if(readable_end.t != stream_or_future_t.t) + trap_if(readable_end.state != End.State.IDLE) + trap_if(readable_end.in_waitable_set()) + writable_end = inst.handles.remove(wi) + trap_if(not isinstance(writable_end, WritableEndT)) + trap_if(writable_end.t != stream_or_future_t.t) + trap_if(writable_end.state != End.State.IDLE) + trap_if(writable_end.in_waitable_set()) + End.forward(readable_end, writable_end) + return [] +``` + + ### ๐Ÿงต `canon thread.index` For a canonical definition: diff --git a/design/mvp/Concurrency.md b/design/mvp/Concurrency.md index 08b9282b..282e441c 100644 --- a/design/mvp/Concurrency.md +++ b/design/mvp/Concurrency.md @@ -577,7 +577,7 @@ or `future`. When *producing* a `stream` or `future` value as a parameter (of an import call) or result (of an export call), the producer can *transfer ownership* of a readable end that it has either been given by the outside world or freshly created via `{stream,future}.new` (which also return a fresh paired -writable end that is permanently owned by the calling component instance). +writable end). Based on this, `stream` and `future` values can be passed between functions as if they were synchronous `list` and `T` values, resp. For @@ -619,6 +619,13 @@ without requiring an explicit `future` return type. Thus, a function like which point the caller receives the readable end of a `future` that, when successfully read, conveys the completion of a second event. +Given the readable end of one stream/future and the writable end of another, the +`{stream,future}.forward` built-ins can be called to efficiently forward all +remaining values from the readable end into the writable end, avoiding any +intermediate copies. Doing so relinquishes ownership of both handles, allowing +the calling component instance to be eagerly torn down while the forwarding is +in progress. + The [Stream and Future State] section describes the runtime state maintained for streams and futures by the Canonical ABI. @@ -1478,7 +1485,6 @@ specified, the following features are being considered for addition to complete the concurrency story: * remove the temporary trap mentioned above that occurs when a `read` and `write` of a stream/future happen from within the same component instance -* zero-copy forwarding/splicing * allow `async` functions using the stackful ABI to be notified of cancellation * allow the `stream` type to validate; make it use `string-encoding` diff --git a/design/mvp/Explainer.md b/design/mvp/Explainer.md index b0cee058..3065f311 100644 --- a/design/mvp/Explainer.md +++ b/design/mvp/Explainer.md @@ -76,6 +76,7 @@ shipped as part of a future WASI Developer Preview release: * ๐Ÿ”—: canonical interface names * ๐Ÿ˜: [memory64] * ๐Ÿ“ก: getters and setters +* โžก๏ธ: `stream.forward` and `future.forward` built-ins ## Grammar @@ -1574,6 +1575,7 @@ canon ::= ... | (canon stream.cancel-write async? (core func ?)) ๐Ÿ”€ | (canon stream.drop-readable (core func ?)) ๐Ÿ”€ | (canon stream.drop-writable (core func ?)) ๐Ÿ”€ + | (canon stream.forward (core func ?)) โžก๏ธ | (canon future.new (core func ?)) ๐Ÿ”€ | (canon future.read * (core func ?)) ๐Ÿ”€ | (canon future.write * (core func ?)) ๐Ÿ”€ @@ -1581,6 +1583,7 @@ canon ::= ... | (canon future.cancel-write async? (core func ?)) ๐Ÿ”€ | (canon future.drop-readable (core func ?)) ๐Ÿ”€ | (canon future.drop-writable (core func ?)) ๐Ÿ”€ + | (canon future.forward (core func ?)) โžก๏ธ | (canon thread.index (core func ?)) ๐Ÿงต | (canon thread.new-indirect core-prefix() core-prefix() (core func ?)) ๐Ÿงต | (canon thread.resume-later (core func ?)) ๐Ÿงต @@ -2146,6 +2149,25 @@ already been dropped. For details, see [Streams and Futures] in the concurrency explainer and [`canon_stream_drop_readable`] in the Canonical ABI explainer. +###### โžก๏ธ `stream.forward` and `future.forward` + +| Synopsis | | +| ---------------------------------------------- | -------------------------------------------------------------------------- | +| Approximate WIT signature for `stream.forward` | `func>(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 | `[ri:i32 wi:i32] -> []` | + +The `{stream,future}.forward` built-ins remove the given readable and writable +ends from the caller's handle table and logically forward everything from the +readable end into the writable end (propagating drops in both directions), but +do so without an intermediate copy. The call traps if either end has a +mismatched direction or element type, is in the middle of a read or write, is a +member of a waitable set, or has received its final `dropped` or, for futures, +`completed` result. + +For details, see [Streams and Futures] in the concurrency explainer and +[`canon_stream_forward`] in the Canonical ABI explainer. + ###### ๐Ÿงต `thread.index` | Synopsis | | @@ -3423,6 +3445,7 @@ For some use-case-focused, worked examples, see: [`canon_future_read`]: CanonicalABI.md#-canon-streamfuturereadwrite [`canon_stream_cancel_read`]: CanonicalABI.md#-canon-streamfuturecancel-readwrite [`canon_stream_drop_readable`]: CanonicalABI.md#-canon-streamfuturedrop-readablewritable +[`canon_stream_forward`]: CanonicalABI.md#-canon-streamfutureforward [`canon_subtask_cancel`]: CanonicalABI.md#-canon-subtaskcancel [`canon_subtask_drop`]: CanonicalABI.md#-canon-subtaskdrop [`canon_resource_new`]: CanonicalABI.md#canon-resourcenew diff --git a/design/mvp/canonical-abi/definitions.py b/design/mvp/canonical-abi/definitions.py index edcd26ba..1448986c 100644 --- a/design/mvp/canonical-abi/definitions.py +++ b/design/mvp/canonical-abi/definitions.py @@ -4,9 +4,8 @@ ### Boilerplate -from __future__ import annotations from dataclasses import dataclass -from typing import Any, Optional, Callable, TypeVar, Generic, Literal +from typing import Optional, Callable, Literal from enum import Enum, IntEnum import math import struct @@ -969,6 +968,24 @@ def cancel(self): or random.randint(0,1))): self.notify(progress = 0) + def forward(src: End, dst: End): + if src.other is dst or src.other is None or dst.other is None: + src.drop() + dst.drop() + else: + writable_end = src.other + readable_end = dst.other + writable_end.other = readable_end + readable_end.other = writable_end + if writable_end.buffer is not None and readable_end.buffer is not None: + if readable_end.buffer.remain() > writable_end.buffer.remain(): + bigger_end, smaller_end = readable_end, writable_end + else: + bigger_end, smaller_end = writable_end, readable_end + buffer = smaller_end.buffer + smaller_end.buffer = None + smaller_end.copy(buffer) + def drop(self): assert(not self.copying_or_cancelling()) if self.other is not None: @@ -2501,6 +2518,30 @@ def drop(EndT, stream_or_future_t, i): end.drop() return [] +### โžก๏ธ `canon {stream,future}.forward` + +def canon_stream_forward(stream_t, ri, wi): + return forward(ReadableStreamEnd, WritableStreamEnd, stream_t, ri, wi) + +def canon_future_forward(future_t, ri, wi): + return forward(ReadableFutureEnd, WritableFutureEnd, future_t, ri, wi) + +def forward(ReadableEndT, WritableEndT, stream_or_future_t, ri, wi): + inst = current_instance() + trap_if(not inst.may_leave) + readable_end = inst.handles.remove(ri) + trap_if(not isinstance(readable_end, ReadableEndT)) + trap_if(readable_end.t != stream_or_future_t.t) + trap_if(readable_end.state != End.State.IDLE) + trap_if(readable_end.in_waitable_set()) + writable_end = inst.handles.remove(wi) + trap_if(not isinstance(writable_end, WritableEndT)) + trap_if(writable_end.t != stream_or_future_t.t) + trap_if(writable_end.state != End.State.IDLE) + trap_if(writable_end.in_waitable_set()) + End.forward(readable_end, writable_end) + return [] + ### ๐Ÿงต `canon thread.index` def canon_thread_index(): diff --git a/design/mvp/canonical-abi/run_tests.py b/design/mvp/canonical-abi/run_tests.py index 1d1e9fbc..926f1ea0 100644 --- a/design/mvp/canonical-abi/run_tests.py +++ b/design/mvp/canonical-abi/run_tests.py @@ -1651,7 +1651,7 @@ def core_func(args): assert(host_reader2.received == [11,12,13,14,15,16,17,18]) -def test_stream_forward(): +def test_transfer_readable_end(): host_writer = HostWriter(U8Type(), [1,2,3,4], chunk=4) def on_start(): return [host_writer.readable_end] @@ -1683,6 +1683,97 @@ def core_func(args): assert(host_reader.dropped and host_reader.take() == []) +def test_forward(): + store = Store() + mem = bytearray(32) + opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True) + inst = ComponentInstance(store) + st = StreamType(U8Type()) + ft = FutureType(U8Type()) + + def core_func(args): + def new_stream(t = st): + [packed] = canon_stream_new(t) + return unpack_new_ends(packed) + def new_future(): + [packed] = canon_future_new(ft) + return unpack_new_ends(packed) + + rsi,wsi = new_stream() + [] = canon_stream_forward(st, rsi, wsi) + + rsi1,wsi1 = new_stream() + rsi2,wsi2 = new_stream() + [] = canon_stream_drop_readable(st, rsi2) + [] = canon_stream_forward(st, rsi1, wsi2) + [ret] = canon_stream_write(st, opts, wsi1, 0, 4) + result,n = unpack_result(ret) + assert(n == 0 and result == CopyResult.DROPPED) + [] = canon_stream_drop_writable(st, wsi1) + + rsi1,wsi1 = new_stream() + rsi2,wsi2 = new_stream() + [] = canon_stream_forward(st, rsi1, wsi2) + mem[0:4] = b'\x01\x02\x03\x04' + [ret] = canon_stream_write(st, opts, wsi1, 0, 4) + assert(ret == definitions.BLOCKED) + [ret] = canon_stream_read(st, opts, rsi2, 8, 4) + result,n = unpack_result(ret) + assert(n == 4 and result == CopyResult.COMPLETED) + assert(mem[8:12] == b'\x01\x02\x03\x04') + + rsi1,wsi1 = new_stream() + rsi2,wsi2 = new_stream() + [ret] = canon_stream_read(st, opts, rsi2, 8, 4) + assert(ret == definitions.BLOCKED) + [] = canon_stream_forward(st, rsi1, wsi2) + mem[0:4] = b'\x05\x06\x07\x08' + [ret] = canon_stream_write(st, opts, wsi1, 0, 4) + result,n = unpack_result(ret) + assert(n == 4 and result == CopyResult.COMPLETED) + assert(mem[8:12] == b'\x05\x06\x07\x08') + [seti] = canon_waitable_set_new() + [] = canon_waitable_join(rsi2, seti) + [event] = canon_waitable_set_wait(MemInst(mem, 'i32'), seti, 16) + assert(event == EventCode.STREAM_READ) + assert(mem[16] == rsi2) + result,n = unpack_result(mem[20]) + assert(n == 4 and result == CopyResult.COMPLETED) + + rfi1,wfi1 = new_future() + rfi2,wfi2 = new_future() + [ret] = canon_future_read(ft, opts, rfi2, 8) + assert(ret == definitions.BLOCKED) + [] = canon_future_forward(ft, rfi1, wfi2) + mem[0] = 42 + [ret] = canon_future_write(ft, opts, wfi1, 0) + assert(ret == CopyResult.COMPLETED) + assert(mem[8] == 42) + + rfi1,wfi1 = new_future() + rfi2,wfi2 = new_future() + mem[0] = 43 + [ret] = canon_future_write(ft, opts, wfi1, 0) + assert(ret == definitions.BLOCKED) + [ret] = canon_future_read(ft, opts, rfi2, 8) + assert(ret == definitions.BLOCKED) + [] = canon_future_forward(ft, rfi1, wfi2) + assert(mem[8] == 43) + + rfi1,wfi1 = new_future() + rfi2,wfi2 = new_future() + [] = canon_future_drop_readable(ft, rfi2) + [] = canon_future_forward(ft, rfi1, wfi2) + [ret] = canon_future_write(ft, opts, wfi1, 0) + assert(ret == CopyResult.DROPPED) + [] = canon_future_drop_writable(ft, wfi1) + + return [] + + caller_ft = FuncType([], [], async_ = True) + lift_and_run(mk_opts(), inst, caller_ft, core_func, lambda:[], lambda _:()) + + def test_receive_own_stream(): store = Store() inst = ComponentInstance(store) @@ -3043,7 +3134,8 @@ def on_resolve(v): test_sync_using_wait() test_eager_stream_completion() test_async_stream_ops() -test_stream_forward() +test_transfer_readable_end() +test_forward() test_receive_own_stream() test_host_partial_reads_writes() test_wasm_to_wasm_stream() diff --git a/test/async/big-interleaving-test.wast b/test/async/big-interleaving-test.wast index 5ca9bf85..71ee132e 100644 --- a/test/async/big-interleaving-test.wast +++ b/test/async/big-interleaving-test.wast @@ -168,6 +168,7 @@ (import "" "stream.cancel-write" (func $stream.cancel-write (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 "" "stream.forward" (func $stream.forward (param i32 i32))) (import "" "future.new" (func $future.new (result i64))) (import "" "future.read" (func $future.read (param i32 i32) (result i32))) (import "" "future.write" (func $future.write (param i32 i32) (result i32))) @@ -175,6 +176,7 @@ (import "" "future.cancel-write" (func $future.cancel-write (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 "" "future.forward" (func $future.forward (param i32 i32))) (import "" "task.return" (func $task.return)) (import "" "waitable.join" (func $waitable.join (param i32 i32))) (import "" "waitable-set.new" (func $waitable-set.new (result i32))) @@ -314,6 +316,11 @@ (func (export "drop-writable") (param $slot i32) (call $stream.drop-writable (call $tx (local.get $slot)))) + (func (export "stream-forward") (param $src i32) (param $dst i32) + (call $stream.forward (call $rx (local.get $src)) (call $tx (local.get $dst)))) + (func (export "future-forward") (param $src i32) (param $dst i32) + (call $future.forward (call $rx (local.get $src)) (call $tx (local.get $dst)))) + (func (export "poll") (param $slot i32) (param $expected-event i32) (param $expected-payload i32) (local $ws i32) (local $event i32) (local.set $ws (call $waitable-set.new)) @@ -373,6 +380,7 @@ (canon stream.cancel-write $ST async (core func $stream.cancel-write)) (canon stream.drop-readable $ST (core func $stream.drop-readable)) (canon stream.drop-writable $ST (core func $stream.drop-writable)) + (canon stream.forward $ST (core func $stream.forward)) (canon future.new $FT (core func $future.new)) (canon future.read $FT async (memory (core memory $memory "mem")) (core func $future.read)) (canon future.write $FT async (memory (core memory $memory "mem")) (core func $future.write)) @@ -380,6 +388,7 @@ (canon future.cancel-write $FT async (core func $future.cancel-write)) (canon future.drop-readable $FT (core func $future.drop-readable)) (canon future.drop-writable $FT (core func $future.drop-writable)) + (canon future.forward $FT (core func $future.forward)) (canon task.return (core func $task.return)) (canon waitable.join (core func $waitable.join)) (canon waitable-set.new (core func $waitable-set.new)) @@ -411,6 +420,7 @@ (export "stream.cancel-write" (func $stream.cancel-write)) (export "stream.drop-readable" (func $stream.drop-readable)) (export "stream.drop-writable" (func $stream.drop-writable)) + (export "stream.forward" (func $stream.forward)) (export "future.new" (func $future.new)) (export "future.read" (func $future.read)) (export "future.write" (func $future.write)) @@ -418,6 +428,7 @@ (export "future.cancel-write" (func $future.cancel-write)) (export "future.drop-readable" (func $future.drop-readable)) (export "future.drop-writable" (func $future.drop-writable)) + (export "future.forward" (func $future.forward)) (export "task.return" (func $task.return)) (export "waitable.join" (func $waitable.join)) (export "waitable-set.new" (func $waitable-set.new)) @@ -453,6 +464,8 @@ (func (export "future-drop-writable") (param "slot" u8) (canon lift (core func $tm "future-drop-writable"))) (func (export "drop-readable") (param "slot" u8) (canon lift (core func $tm "drop-readable"))) (func (export "drop-writable") (param "slot" u8) (canon lift (core func $tm "drop-writable"))) + (func (export "stream-forward") (param "src" u8) (param "dst" u8) (canon lift (core func $tm "stream-forward"))) + (func (export "future-forward") (param "src" u8) (param "dst" u8) (canon lift (core func $tm "future-forward"))) (func (export "poll") (param "slot" u8) (param "event" u8) (param "payload" u32) (canon lift (core func $tm "poll"))) (func (export "await") async (param "slot" u8) (param "event" u8) (param "payload" u32) (canon lift (core func $tm "await") async (memory (core memory $memory "mem")))) @@ -469,6 +482,8 @@ (export $sub-args-e "sub-args" (type $sub-args)) (type $sub-expect (record (field "sub" u8) (field "state" u8))) (export $sub-expect-e "sub-expect" (type $sub-expect)) + (type $forward-args (record (field "src" u8) (field "dst" u8))) + (export $forward-args-e "forward-args" (type $forward-args)) (type $command (variant (case "stream-new" u8) (case "future-new" u8) @@ -501,7 +516,9 @@ (case "await-subtask" $sub-expect-e) (case "mock-bp-inc") (case "mock-bp-dec") - (case "subtask-cancel-await" $sub-expect-e))) + (case "subtask-cancel-await" $sub-expect-e) + (case "stream-forward" $forward-args-e) + (case "future-forward" $forward-args-e))) (export $command-e "command" (type $command)) (import "call-import" (func $call-import (param "slot" u8) (result s32))) (import "stream-new" (func $stream-new (param "slot" u8))) @@ -509,6 +526,8 @@ (import "testee-read" (func $testee-read (param "handle" u8) (param "bytes" u32) (result s32))) (import "drop-readable" (func $drop-readable (param "slot" u8))) (import "drop-writable" (func $drop-writable (param "slot" u8))) + (import "stream-forward" (func $stream-forward (param "src" u8) (param "dst" u8))) + (import "future-forward" (func $future-forward (param "src" u8) (param "dst" u8))) (import "poll" (func $poll (param "slot" u8) (param "event" u8) (param "payload" u32))) (import "await" (func $await async (param "slot" u8) (param "event" u8) (param "payload" u32))) (import "future-new" (func $future-new (param "slot" u8))) @@ -545,6 +564,8 @@ (import "" "call-import-future" (func $call-import-future (param i32) (result i32))) (import "" "drop-readable" (func $drop-readable (param i32))) (import "" "drop-writable" (func $drop-writable (param i32))) + (import "" "stream-forward" (func $stream-forward (param i32 i32))) + (import "" "future-forward" (func $future-forward (param i32 i32))) (import "" "poll" (func $poll (param i32 i32 i32))) (import "" "await" (func $await (param i32 i32 i32) (result i32))) (import "" "poll-readable" (func $poll-readable (param i32 i32 i32))) @@ -603,6 +624,8 @@ (global $MOCK_BP_INC i32 (i32.const 29)) (global $MOCK_BP_DEC i32 (i32.const 30)) (global $SUBTASK_CANCEL_AWAIT i32 (i32.const 31)) + (global $STREAM_FORWARD i32 (i32.const 32)) + (global $FUTURE_FORWARD i32 (i32.const 33)) (global $last (mut i32) (i32.const 0)) (global $VOID_OK i32 (i32.const 1337)) @@ -661,6 +684,19 @@ (call $drop-writable (i32.load8_u offset=4 (local.get $insn))) (global.set $last (global.get $VOID_OK)))) + (if (i32.eq (local.get $op) (global.get $STREAM_FORWARD)) + (then + (call $stream-forward + (i32.load8_u offset=4 (local.get $insn)) + (i32.load8_u offset=5 (local.get $insn))) + (global.set $last (global.get $VOID_OK)))) + (if (i32.eq (local.get $op) (global.get $FUTURE_FORWARD)) + (then + (call $future-forward + (i32.load8_u offset=4 (local.get $insn)) + (i32.load8_u offset=5 (local.get $insn))) + (global.set $last (global.get $VOID_OK)))) + (if (i32.eq (local.get $op) (global.get $CALL_IMPORT)) (then (global.set $last (call $call-import (i32.load8_u offset=4 (local.get $insn)))))) @@ -762,6 +798,8 @@ (canon lower (func $future-drop-writable) (core func $future-drop-writable')) (canon lower (func $drop-readable) (core func $drop-readable')) (canon lower (func $drop-writable) (core func $drop-writable')) + (canon lower (func $stream-forward) (core func $stream-forward')) + (canon lower (func $future-forward) (core func $future-forward')) (canon lower (func $poll) (core func $poll')) (canon lower (func $await) async (core func $await')) (canon lower (func $poll-readable) (core func $poll-readable')) @@ -791,6 +829,8 @@ (export "future-drop-writable" (func $future-drop-writable')) (export "drop-readable" (func $drop-readable')) (export "drop-writable" (func $drop-writable')) + (export "stream-forward" (func $stream-forward')) + (export "future-forward" (func $future-forward')) (export "poll" (func $poll')) (export "await" (func $await')) (export "poll-readable" (func $poll-readable')) @@ -831,6 +871,8 @@ (with "future-drop-writable" (func $testee "future-drop-writable")) (with "drop-readable" (func $testee "drop-readable")) (with "drop-writable" (func $testee "drop-writable")) + (with "stream-forward" (func $testee "stream-forward")) + (with "future-forward" (func $testee "future-forward")) (with "poll" (func $testee "poll")) (with "await" (func $testee "await")) (with "poll-readable" (func $testee "poll-readable")) @@ -853,6 +895,7 @@ (export "poll-expect" (type $driver "poll-expect")) (export "sub-args" (type $driver "sub-args")) (export "sub-expect" (type $driver "sub-expect")) + (export "forward-args" (type $driver "forward-args")) (export "command" (type $driver "command"))) (export "types" (instance $types)) (alias export $driver "run" (func $run)) @@ -1689,3 +1732,49 @@ (variant.const "expect-code" (s32.const 3)) (variant.const "subtask-drop" (u8.const 1)) (variant.const "mock-bp-dec")))) + +(component instance $i $Tester) + +(assert_return + (invoke "run" + (list.const + (variant.const "stream-new" (u8.const 0)) + (variant.const "stream-new" (u8.const 1)) + (variant.const "future-new" (u8.const 2)) + (variant.const "future-new" (u8.const 3)) + (variant.const "stream-new" (u8.const 4)) + (variant.const "stream-new" (u8.const 5)) + (variant.const "future-new" (u8.const 6)) + (variant.const "future-new" (u8.const 7)) + (variant.const "stream-forward" (record.const (field "src" u8.const 0) (field "dst" u8.const 1))) + (variant.const "future-forward" (record.const (field "src" u8.const 2) (field "dst" u8.const 3))) + (variant.const "stream-forward" (record.const (field "src" u8.const 4) (field "dst" u8.const 5))) + (variant.const "future-forward" (record.const (field "src" u8.const 6) (field "dst" u8.const 7))) + (variant.const "testee-write" (record.const (field "handle" u8.const 0) (field "bytes" u32.const 4))) + (variant.const "expect-code" (s32.const -1)) + (variant.const "future-write" (u8.const 2)) + (variant.const "expect-code" (s32.const -1)) + (variant.const "testee-write" (record.const (field "handle" u8.const 4) (field "bytes" u32.const 4))) + (variant.const "expect-code" (s32.const -1)) + (variant.const "future-write" (u8.const 6)) + (variant.const "expect-code" (s32.const -1)) + (variant.const "future-read" (u8.const 7)) + (variant.const "expect-code" (s32.const 0)) + (variant.const "testee-read" (record.const (field "handle" u8.const 1) (field "bytes" u32.const 4))) + (variant.const "expect-code" (s32.const 0x40)) + (variant.const "future-read" (u8.const 3)) + (variant.const "expect-code" (s32.const 0)) + (variant.const "testee-read" (record.const (field "handle" u8.const 5) (field "bytes" u32.const 4))) + (variant.const "expect-code" (s32.const 0x40)) + (variant.const "poll" (record.const (field "slot" u8.const 0) (field "event" enum.const "stream-write") (field "payload" u32.const 0x40))) + (variant.const "poll" (record.const (field "slot" u8.const 2) (field "event" enum.const "future-write") (field "payload" u32.const 0))) + (variant.const "poll" (record.const (field "slot" u8.const 4) (field "event" enum.const "stream-write") (field "payload" u32.const 0x40))) + (variant.const "poll" (record.const (field "slot" u8.const 6) (field "event" enum.const "future-write") (field "payload" u32.const 0))) + (variant.const "drop-writable" (u8.const 0)) + (variant.const "drop-readable" (u8.const 1)) + (variant.const "future-drop-writable" (u8.const 2)) + (variant.const "future-drop-readable" (u8.const 3)) + (variant.const "drop-writable" (u8.const 4)) + (variant.const "drop-readable" (u8.const 5)) + (variant.const "future-drop-writable" (u8.const 6)) + (variant.const "future-drop-readable" (u8.const 7))))) diff --git a/test/async/forward.wast b/test/async/forward.wast new file mode 100644 index 00000000..cb15ce0e --- /dev/null +++ b/test/async/forward.wast @@ -0,0 +1,1455 @@ +;; Tests the `stream.forward` and `future.forward` built-ins. + +(component definition $Tester + (core module $Memory + (memory (export "mem") 1) + (func (export "realloc") (param i32 i32 i32 i32) (result i32) + unreachable) + ) + (core instance $memory (instantiate $Memory)) + (core module $M + (import "" "mem" (memory 1)) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.poll" (func $waitable-set.poll (param i32 i32) (result i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (import "" "stream.cancel-read" (func $stream.cancel-read (param i32) (result i32))) + (import "" "stream.cancel-write" (func $stream.cancel-write (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 "" "stream.forward" (func $stream.forward (param i32 i32))) + (import "" "stream.forward-u32" (func $stream.forward-u32 (param i32 i32))) + (import "" "stream.new-e" (func $stream.new-e (result i64))) + (import "" "stream.read-e" (func $stream.read-e (param i32 i32 i32) (result i32))) + (import "" "stream.write-e" (func $stream.write-e (param i32 i32 i32) (result i32))) + (import "" "stream.forward-e" (func $stream.forward-e (param i32 i32))) + (import "" "stream.new-s" (func $stream.new-s (result i64))) + (import "" "stream.read-s" (func $stream.read-s (param i32 i32 i32) (result i32))) + (import "" "stream.write-s" (func $stream.write-s (param i32 i32 i32) (result i32))) + (import "" "stream.forward-s" (func $stream.forward-s (param i32 i32))) + (import "" "future.new" (func $future.new (result i64))) + (import "" "future.read" (func $future.read (param i32 i32) (result i32))) + (import "" "future.write" (func $future.write (param i32 i32) (result i32))) + (import "" "future.drop-readable" (func $future.drop-readable (param i32))) + (import "" "future.drop-writable" (func $future.drop-writable (param i32))) + (import "" "future.forward" (func $future.forward (param i32 i32))) + (import "" "future.new-s" (func $future.new-s (result i64))) + (import "" "future.read-s" (func $future.read-s (param i32 i32) (result i32))) + (import "" "future.write-s" (func $future.write-s (param i32 i32) (result i32))) + (import "" "future.forward-s" (func $future.forward-s (param i32 i32))) + + ;; The three streams/futures used by the tests below are named A, B and C + ;; and their ends are kept in globals so that each test can stay readable. + ;; A fresh $Tester instance is created before every invoke, so the handle + ;; indices allocated by the $new-* helpers are deterministic: 1,2 for A, + ;; 3,4 for B and 5,6 for C. + (global $rA (mut i32) (i32.const 0)) + (global $wA (mut i32) (i32.const 0)) + (global $rB (mut i32) (i32.const 0)) + (global $wB (mut i32) (i32.const 0)) + (global $rC (mut i32) (i32.const 0)) + (global $wC (mut i32) (i32.const 0)) + + (func $new-a + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $rA (i32.wrap_i64 (local.get $ret64))) + (global.set $wA (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $new-b + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $rB (i32.wrap_i64 (local.get $ret64))) + (global.set $wB (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $new-c + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $rC (i32.wrap_i64 (local.get $ret64))) + (global.set $wC (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $fnew-a + (local $ret64 i64) + (local.set $ret64 (call $future.new)) + (global.set $rA (i32.wrap_i64 (local.get $ret64))) + (global.set $wA (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $fnew-b + (local $ret64 i64) + (local.set $ret64 (call $future.new)) + (global.set $rB (i32.wrap_i64 (local.get $ret64))) + (global.set $wB (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $fnew-c + (local $ret64 i64) + (local.set $ret64 (call $future.new)) + (global.set $rC (i32.wrap_i64 (local.get $ret64))) + (global.set $wC (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $enew-a + (local $ret64 i64) + (local.set $ret64 (call $stream.new-e)) + (global.set $rA (i32.wrap_i64 (local.get $ret64))) + (global.set $wA (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $enew-b + (local $ret64 i64) + (local.set $ret64 (call $stream.new-e)) + (global.set $rB (i32.wrap_i64 (local.get $ret64))) + (global.set $wB (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $snew-a + (local $ret64 i64) + (local.set $ret64 (call $stream.new-s)) + (global.set $rA (i32.wrap_i64 (local.get $ret64))) + (global.set $wA (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $snew-b + (local $ret64 i64) + (local.set $ret64 (call $stream.new-s)) + (global.set $rB (i32.wrap_i64 (local.get $ret64))) + (global.set $wB (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $fsnew-a + (local $ret64 i64) + (local.set $ret64 (call $future.new-s)) + (global.set $rA (i32.wrap_i64 (local.get $ret64))) + (global.set $wA (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $fsnew-b + (local $ret64 i64) + (local.set $ret64 (call $future.new-s)) + (global.set $rB (i32.wrap_i64 (local.get $ret64))) + (global.set $wB (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + + ;; the merged stream is usable in both arrival orders + (func (export "forward-then-write-then-read") (result i32) + (local $ret i32) + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read (global.get $rB) (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 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + (func (export "forward-then-read-then-write") (result i32) + (local $ret i32) + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + + ;; a copy that was already pending when the forward happened is carried + ;; over onto the merged stream + (func (export "read-then-forward-then-write") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $rB) (local.get $ws)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 2 (; STREAM_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (i32.const 42) + ) + + ;; both sides are blocked before the forward, so the forward itself + ;; performs the rendezvous + (func (export "rendezvous-during-forward") (result i32) + (local $ret i32) (local $ws1 i32) (local $ws2 i32) + (call $new-a) + (call $new-b) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (local.set $ws1 (call $waitable-set.new)) + (local.set $ws2 (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws1)) + (call $waitable.join (global.get $rB) (local.get $ws2)) + (local.set $ret (call $waitable-set.poll (local.get $ws1) (i32.const 32))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws2) (i32.const 40))) + (if (i32.ne (i32.const 2 (; STREAM_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 40))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 44))) + (then unreachable)) + (i32.const 42) + ) + + ;; the destination's readable end may already be in a waitable set when + ;; the forward happens, and the carried-over read must still be reported + ;; through it + (func (export "pending-read-in-set-before-forward") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $rB) (local.get $ws)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 2 (; STREAM_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + + ;; discarding A's readable end and B's writable end must not look like a + ;; drop to the two ends that survive + (func (export "no-event-from-forward") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws)) + (call $waitable.join (global.get $rB) (local.get $ws)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 0 (; NONE ;)) (local.get $ret)) + (then unreachable)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 0 (; NONE ;)) (local.get $ret)) + (then unreachable)) + (i32.const 42) + ) + + ;; a zero-length readiness read or write is carried over just like a real + ;; one and completes with 0 elements once the other end shows up + (func (export "zero-length-read-carried-over") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 0) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $rB) (local.get $ws)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 2 (; STREAM_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x0 (; COMPLETED=0 | (0<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (local.set $ret (call $stream.read (global.get $rB) (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 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + (func (export "zero-length-write-then-forward") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x0 (; COMPLETED=0 | (0<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + + ;; drop propagation in both directions + (func (export "dest-reader-dropped-before-forward") (result i32) + (local $ret i32) + (call $new-a) + (call $new-b) + (call $stream.drop-readable (global.get $rB)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-writable (global.get $wA)) + (i32.const 42) + ) + (func (export "dest-reader-dropped-before-forward-blocked-writer") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $rB)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (call $stream.drop-writable (global.get $wA)) + (i32.const 42) + ) + (func (export "dest-reader-dropped-after-forward") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws)) + (call $stream.drop-readable (global.get $rB)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 3 (; STREAM_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (i32.const 42) + ) + (func (export "src-writer-dropped-before-forward") (result i32) + (local $ret i32) + (call $new-a) + (call $new-b) + (call $stream.drop-writable (global.get $wA)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.drop-readable (global.get $rB)) + (i32.const 42) + ) + (func (export "src-writer-dropped-after-forward") (result i32) + (local $ret i32) (local $ws i32) + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $rB) (local.get $ws)) + (call $stream.drop-writable (global.get $wA)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 2 (; STREAM_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + (i32.const 42) + ) + + ;; cancellation after forward + (func (export "cancel-carried-over-read") (result i32) + (local $ret i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.cancel-read (global.get $rB))) + (if (i32.ne (i32.const 0x2 (; CANCELLED=2 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read (global.get $rB) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + + ;; self-loop + (func (export "self-forward") (result i32) + (call $new-a) + (if (i32.ne (i32.const 1) (global.get $rA)) + (then unreachable)) + (if (i32.ne (i32.const 2) (global.get $wA)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wA)) + (call $new-b) + (if (i32.ne (i32.const 2) (global.get $rB)) + (then unreachable)) + (if (i32.ne (i32.const 1) (global.get $wB)) + (then unreachable)) + (i32.const 42) + ) + + ;; A absorbs B, then C absorbs A: one stream from C's writer to B's reader + (func (export "forward-chain") (result i32) + (local $ret i32) + (call $new-a) + (call $new-b) + (call $new-c) + (call $stream.forward (global.get $rA) (global.get $wB)) + (call $stream.forward (global.get $rC) (global.get $wA)) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (local.set $ret (call $stream.write (global.get $wC) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read (global.get $rB) (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 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + + ;; forwarding A into B and then B's (now-shared) readable end back into + ;; A's writable end degenerates into the self-loop case, so no cycle can + ;; be built and there is nothing left to trap on + (func (export "forward-cycle") (result i32) + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + (call $stream.forward (global.get $rB) (global.get $wA)) + (call $new-c) + (if (i32.ne (i32.const 2) (global.get $rC)) + (then unreachable)) + (if (i32.ne (i32.const 3) (global.get $wC)) + (then unreachable)) + (i32.const 42) + ) + + ;; future versions of all of the above + (func (export "future-forward-then-write-then-read") (result i32) + (local $ret i32) + (call $fnew-a) + (call $fnew-b) + (call $future.forward (global.get $rA) (global.get $wB)) + (i32.store8 (i32.const 0) (i32.const 0x5a)) + (local.set $ret (call $future.write (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $future.read (global.get $rB) (i32.const 8))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x5a) (i32.load8_u (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + (func (export "future-forward-read-then-write") (result i32) + (local $ret i32) (local $ws i32) + (call $fnew-a) + (call $fnew-b) + (local.set $ret (call $future.read (global.get $rB) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $future.forward (global.get $rA) (global.get $wB)) + (i32.store8 (i32.const 0) (i32.const 0x5a)) + (local.set $ret (call $future.write (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x5a) (i32.load8_u (i32.const 8))) + (then unreachable)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $rB) (local.get $ws)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 4 (; FUTURE_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (i32.load (i32.const 36))) + (then unreachable)) + (i32.const 42) + ) + (func (export "future-rendezvous-during-forward") (result i32) + (local $ret i32) (local $ws1 i32) (local $ws2 i32) + (call $fnew-a) + (call $fnew-b) + (i32.store8 (i32.const 0) (i32.const 0x5a)) + (local.set $ret (call $future.write (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $future.read (global.get $rB) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $future.forward (global.get $rA) (global.get $wB)) + (if (i32.ne (i32.const 0x5a) (i32.load8_u (i32.const 8))) + (then unreachable)) + (local.set $ws1 (call $waitable-set.new)) + (local.set $ws2 (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws1)) + (call $waitable.join (global.get $rB) (local.get $ws2)) + (local.set $ret (call $waitable-set.poll (local.get $ws1) (i32.const 32))) + (if (i32.ne (i32.const 5 (; FUTURE_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (i32.load (i32.const 36))) + (then unreachable)) + (local.set $ret (call $waitable-set.poll (local.get $ws2) (i32.const 40))) + (if (i32.ne (i32.const 4 (; FUTURE_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $rB) (i32.load (i32.const 40))) + (then unreachable)) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (i32.load (i32.const 44))) + (then unreachable)) + (call $future.drop-writable (global.get $wA)) + (i32.const 42) + ) + (func (export "future-forward-dest-reader-dropped") (result i32) + (local $ret i32) + (call $fnew-a) + (call $fnew-b) + (call $future.drop-readable (global.get $rB)) + (call $future.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $future.write (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const 1 (; DROPPED ;)) (local.get $ret)) + (then unreachable)) + (call $future.drop-writable (global.get $wA)) + (i32.const 42) + ) + (func (export "future-forward-dest-reader-dropped-blocked-writer") (result i32) + (local $ret i32) (local $ws i32) + (call $fnew-a) + (call $fnew-b) + (i32.store8 (i32.const 0) (i32.const 0x5a)) + (local.set $ret (call $future.write (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $future.drop-readable (global.get $rB)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (global.get $wA) (local.get $ws)) + (call $future.forward (global.get $rA) (global.get $wB)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 5 (; FUTURE_WRITE ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (global.get $wA) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 1 (; DROPPED ;)) (i32.load (i32.const 36))) + (then unreachable)) + (call $future.drop-writable (global.get $wA)) + (i32.const 42) + ) + (func (export "future-forward-discards-unwritten-writer") (result i32) + (local $ret i32) + (call $fnew-a) + (call $fnew-b) + (call $future.forward (global.get $rA) (global.get $wB)) + (i32.store8 (i32.const 0) (i32.const 0x5a)) + (local.set $ret (call $future.write (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $future.read (global.get $rB) (i32.const 8))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x5a) (i32.load8_u (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + (func (export "future-self-forward") (result i32) + (call $fnew-a) + (if (i32.ne (i32.const 1) (global.get $rA)) + (then unreachable)) + (if (i32.ne (i32.const 2) (global.get $wA)) + (then unreachable)) + (call $future.forward (global.get $rA) (global.get $wA)) + (call $fnew-b) + (if (i32.ne (i32.const 2) (global.get $rB)) + (then unreachable)) + (if (i32.ne (i32.const 1) (global.get $wB)) + (then unreachable)) + (i32.const 42) + ) + (func (export "future-forward-chain") (result i32) + (local $ret i32) + (call $fnew-a) + (call $fnew-b) + (call $fnew-c) + (call $future.forward (global.get $rA) (global.get $wB)) + (call $future.forward (global.get $rC) (global.get $wA)) + (i32.store8 (i32.const 0) (i32.const 0x5a)) + (local.set $ret (call $future.write (global.get $wC) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $future.read (global.get $rB) (i32.const 8))) + (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x5a) (i32.load8_u (i32.const 8))) + (then unreachable)) + (i32.const 42) + ) + + ;; an empty element type forwards just like any other + (func (export "empty-element-type") (result i32) + (local $ret i32) + (call $enew-a) + (call $enew-b) + (call $stream.forward-e (global.get $rA) (global.get $wB)) + (local.set $ret (call $stream.write-e (global.get $wA) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read-e (global.get $rB) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (i32.const 42) + ) + + ;; trap conditions + + (func (export "trap-swapped-args") + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $wA) (global.get $rB)) + unreachable + ) + (func (export "trap-type-mismatch") + (call $new-a) + (call $new-b) + (call $stream.forward-u32 (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-same-index") + (call $new-a) + (call $stream.forward (global.get $rA) (global.get $rA)) + unreachable + ) + (func (export "trap-src-reader-copying") + (local $ret i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.read (global.get $rA) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-dest-writer-copying") + (local $ret i32) + (call $new-a) + (call $new-b) + (local.set $ret (call $stream.write (global.get $wB) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-src-reader-in-set") + (call $new-a) + (call $new-b) + (call $waitable.join (global.get $rA) (call $waitable-set.new)) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-dest-writer-in-set") + (call $new-a) + (call $new-b) + (call $waitable.join (global.get $wB) (call $waitable-set.new)) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-src-reader-done") + (local $ret i32) + (call $new-a) + (call $new-b) + (call $stream.drop-writable (global.get $wA)) + (local.set $ret (call $stream.read (global.get $rA) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-dest-writer-done") + (local $ret i32) + (call $new-a) + (call $new-b) + (call $stream.drop-readable (global.get $rB)) + (local.set $ret (call $stream.write (global.get $wB) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-use-after-forward") + (call $new-a) + (call $new-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + (call $stream.drop-readable (global.get $rA)) + unreachable + ) + (func (export "trap-stream-forward-of-future-handles") + (call $fnew-a) + (call $fnew-b) + (call $stream.forward (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-future-swapped-args") + (call $fnew-a) + (call $fnew-b) + (call $future.forward (global.get $wA) (global.get $rB)) + unreachable + ) + (func (export "trap-future-reader-copying") + (local $ret i32) + (call $fnew-a) + (call $fnew-b) + (local.set $ret (call $future.read (global.get $rA) (i32.const 8))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $future.forward (global.get $rA) (global.get $wB)) + unreachable + ) + + ;; trap (in the temporary same-instance trap) if forwarding triggers a + ;; same-instance copy of a complex value + (func (export "trap-same-instance-rendezvous") + (local $ret i32) + (call $snew-a) + (call $snew-b) + (i32.store (i32.const 0) (i32.const 64)) ;; the one string's pointer ... + (i32.store (i32.const 4) (i32.const 2)) ;; ... and its length + (i32.store16 (i32.const 64) (i32.const 0x6968 (; "hi" ;))) + (local.set $ret (call $stream.write-s (global.get $wA) (i32.const 0) (i32.const 1))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read-s (global.get $rB) (i32.const 16) (i32.const 1))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $stream.forward-s (global.get $rA) (global.get $wB)) + unreachable + ) + (func (export "trap-future-same-instance-rendezvous") + (local $ret i32) + (call $fsnew-a) + (call $fsnew-b) + (i32.store (i32.const 0) (i32.const 64)) ;; the string's pointer ... + (i32.store (i32.const 4) (i32.const 2)) ;; ... and its length + (i32.store16 (i32.const 64) (i32.const 0x6968 (; "hi" ;))) + (local.set $ret (call $future.write-s (global.get $wA) (i32.const 0))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $future.read-s (global.get $rB) (i32.const 16))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (call $future.forward-s (global.get $rA) (global.get $wB)) + unreachable + ) + ) + (type $ST (stream u8)) + (type $ST32 (stream u32)) + (type $STE (stream)) + (type $STS (stream string)) + (type $FT (future u8)) + (type $FTS (future string)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.poll (memory (core memory $memory "mem")) (core func $waitable-set.poll)) + (canon waitable.join (core func $waitable.join)) + (canon stream.new $ST (core func $stream.new)) + (canon stream.read $ST async (memory (core memory $memory "mem")) (core func $stream.read)) + (canon stream.write $ST async (memory (core memory $memory "mem")) (core func $stream.write)) + (canon stream.cancel-read $ST (core func $stream.cancel-read)) + (canon stream.cancel-write $ST (core func $stream.cancel-write)) + (canon stream.drop-readable $ST (core func $stream.drop-readable)) + (canon stream.drop-writable $ST (core func $stream.drop-writable)) + (canon stream.forward $ST (core func $stream.forward)) + (canon stream.forward $ST32 (core func $stream.forward-u32)) + (canon stream.new $STE (core func $stream.new-e)) + (canon stream.read $STE async (core func $stream.read-e)) + (canon stream.write $STE async (core func $stream.write-e)) + (canon stream.forward $STE (core func $stream.forward-e)) + (canon stream.new $STS (core func $stream.new-s)) + (canon stream.read $STS async (memory (core memory $memory "mem")) (realloc (core func $memory "realloc")) (core func $stream.read-s)) + (canon stream.write $STS async (memory (core memory $memory "mem")) (core func $stream.write-s)) + (canon stream.forward $STS (core func $stream.forward-s)) + (canon future.new $FT (core func $future.new)) + (canon future.read $FT async (memory (core memory $memory "mem")) (core func $future.read)) + (canon future.write $FT async (memory (core memory $memory "mem")) (core func $future.write)) + (canon future.drop-readable $FT (core func $future.drop-readable)) + (canon future.drop-writable $FT (core func $future.drop-writable)) + (canon future.forward $FT (core func $future.forward)) + (canon future.new $FTS (core func $future.new-s)) + (canon future.read $FTS async (memory (core memory $memory "mem")) (realloc (core func $memory "realloc")) (core func $future.read-s)) + (canon future.write $FTS async (memory (core memory $memory "mem")) (core func $future.write-s)) + (canon future.forward $FTS (core func $future.forward-s)) + (core instance $m (instantiate $M (with "" (instance + (export "mem" (memory $memory "mem")) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.poll" (func $waitable-set.poll)) + (export "waitable.join" (func $waitable.join)) + (export "stream.new" (func $stream.new)) + (export "stream.read" (func $stream.read)) + (export "stream.write" (func $stream.write)) + (export "stream.cancel-read" (func $stream.cancel-read)) + (export "stream.cancel-write" (func $stream.cancel-write)) + (export "stream.drop-readable" (func $stream.drop-readable)) + (export "stream.drop-writable" (func $stream.drop-writable)) + (export "stream.forward" (func $stream.forward)) + (export "stream.forward-u32" (func $stream.forward-u32)) + (export "stream.new-e" (func $stream.new-e)) + (export "stream.read-e" (func $stream.read-e)) + (export "stream.write-e" (func $stream.write-e)) + (export "stream.forward-e" (func $stream.forward-e)) + (export "stream.new-s" (func $stream.new-s)) + (export "stream.read-s" (func $stream.read-s)) + (export "stream.write-s" (func $stream.write-s)) + (export "stream.forward-s" (func $stream.forward-s)) + (export "future.new" (func $future.new)) + (export "future.read" (func $future.read)) + (export "future.write" (func $future.write)) + (export "future.drop-readable" (func $future.drop-readable)) + (export "future.drop-writable" (func $future.drop-writable)) + (export "future.forward" (func $future.forward)) + (export "future.new-s" (func $future.new-s)) + (export "future.read-s" (func $future.read-s)) + (export "future.write-s" (func $future.write-s)) + (export "future.forward-s" (func $future.forward-s)) + )))) + (func (export "forward-then-write-then-read") (result u32) (canon lift (core func $m "forward-then-write-then-read"))) + (func (export "forward-then-read-then-write") (result u32) (canon lift (core func $m "forward-then-read-then-write"))) + (func (export "read-then-forward-then-write") (result u32) (canon lift (core func $m "read-then-forward-then-write"))) + (func (export "rendezvous-during-forward") (result u32) (canon lift (core func $m "rendezvous-during-forward"))) + (func (export "pending-read-in-set-before-forward") (result u32) (canon lift (core func $m "pending-read-in-set-before-forward"))) + (func (export "no-event-from-forward") (result u32) (canon lift (core func $m "no-event-from-forward"))) + (func (export "zero-length-read-carried-over") (result u32) (canon lift (core func $m "zero-length-read-carried-over"))) + (func (export "zero-length-write-then-forward") (result u32) (canon lift (core func $m "zero-length-write-then-forward"))) + (func (export "dest-reader-dropped-before-forward") (result u32) (canon lift (core func $m "dest-reader-dropped-before-forward"))) + (func (export "dest-reader-dropped-before-forward-blocked-writer") (result u32) (canon lift (core func $m "dest-reader-dropped-before-forward-blocked-writer"))) + (func (export "dest-reader-dropped-after-forward") (result u32) (canon lift (core func $m "dest-reader-dropped-after-forward"))) + (func (export "src-writer-dropped-before-forward") (result u32) (canon lift (core func $m "src-writer-dropped-before-forward"))) + (func (export "src-writer-dropped-after-forward") (result u32) (canon lift (core func $m "src-writer-dropped-after-forward"))) + (func (export "cancel-carried-over-read") (result u32) (canon lift (core func $m "cancel-carried-over-read"))) + (func (export "self-forward") (result u32) (canon lift (core func $m "self-forward"))) + (func (export "forward-chain") (result u32) (canon lift (core func $m "forward-chain"))) + (func (export "forward-cycle") (result u32) (canon lift (core func $m "forward-cycle"))) + (func (export "future-forward-then-write-then-read") (result u32) (canon lift (core func $m "future-forward-then-write-then-read"))) + (func (export "future-forward-read-then-write") (result u32) (canon lift (core func $m "future-forward-read-then-write"))) + (func (export "future-rendezvous-during-forward") (result u32) (canon lift (core func $m "future-rendezvous-during-forward"))) + (func (export "future-forward-dest-reader-dropped") (result u32) (canon lift (core func $m "future-forward-dest-reader-dropped"))) + (func (export "future-forward-dest-reader-dropped-blocked-writer") (result u32) (canon lift (core func $m "future-forward-dest-reader-dropped-blocked-writer"))) + (func (export "future-forward-discards-unwritten-writer") (result u32) (canon lift (core func $m "future-forward-discards-unwritten-writer"))) + (func (export "future-self-forward") (result u32) (canon lift (core func $m "future-self-forward"))) + (func (export "future-forward-chain") (result u32) (canon lift (core func $m "future-forward-chain"))) + (func (export "empty-element-type") (result u32) (canon lift (core func $m "empty-element-type"))) + (func (export "trap-swapped-args") (canon lift (core func $m "trap-swapped-args"))) + (func (export "trap-type-mismatch") (canon lift (core func $m "trap-type-mismatch"))) + (func (export "trap-same-index") (canon lift (core func $m "trap-same-index"))) + (func (export "trap-src-reader-copying") (canon lift (core func $m "trap-src-reader-copying"))) + (func (export "trap-dest-writer-copying") (canon lift (core func $m "trap-dest-writer-copying"))) + (func (export "trap-src-reader-in-set") (canon lift (core func $m "trap-src-reader-in-set"))) + (func (export "trap-dest-writer-in-set") (canon lift (core func $m "trap-dest-writer-in-set"))) + (func (export "trap-src-reader-done") (canon lift (core func $m "trap-src-reader-done"))) + (func (export "trap-dest-writer-done") (canon lift (core func $m "trap-dest-writer-done"))) + (func (export "trap-use-after-forward") (canon lift (core func $m "trap-use-after-forward"))) + (func (export "trap-stream-forward-of-future-handles") (canon lift (core func $m "trap-stream-forward-of-future-handles"))) + (func (export "trap-future-swapped-args") (canon lift (core func $m "trap-future-swapped-args"))) + (func (export "trap-future-reader-copying") (canon lift (core func $m "trap-future-reader-copying"))) + (func (export "trap-same-instance-rendezvous") (canon lift (core func $m "trap-same-instance-rendezvous"))) + (func (export "trap-future-same-instance-rendezvous") (canon lift (core func $m "trap-future-same-instance-rendezvous"))) +) + +(component instance $i $Tester) +(assert_return (invoke "forward-then-write-then-read") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "forward-then-read-then-write") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "read-then-forward-then-write") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "rendezvous-during-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "pending-read-in-set-before-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "no-event-from-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "zero-length-read-carried-over") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "zero-length-write-then-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "dest-reader-dropped-before-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "dest-reader-dropped-before-forward-blocked-writer") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "dest-reader-dropped-after-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "src-writer-dropped-before-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "src-writer-dropped-after-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "cancel-carried-over-read") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "self-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "forward-chain") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "forward-cycle") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-forward-then-write-then-read") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-forward-read-then-write") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-rendezvous-during-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-forward-dest-reader-dropped") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-forward-dest-reader-dropped-blocked-writer") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-forward-discards-unwritten-writer") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-self-forward") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "future-forward-chain") (u32.const 42)) +(component instance $i $Tester) +(assert_return (invoke "empty-element-type") (u32.const 42)) +(component instance $i $Tester) +(assert_trap (invoke "trap-swapped-args") "handle index 2 used with the wrong type") +(component instance $i $Tester) +(assert_trap (invoke "trap-type-mismatch") "handle index 1 used with the wrong type") +(component instance $i $Tester) +(assert_trap (invoke "trap-same-index") "unknown handle index 1") +(component instance $i $Tester) +(assert_trap (invoke "trap-src-reader-copying") "cannot forward busy stream") +(component instance $i $Tester) +(assert_trap (invoke "trap-dest-writer-copying") "cannot forward busy stream") +(component instance $i $Tester) +(assert_trap (invoke "trap-src-reader-in-set") "cannot forward stream while it's in a waitable set") +(component instance $i $Tester) +(assert_trap (invoke "trap-dest-writer-in-set") "cannot forward stream while it's in a waitable set") +(component instance $i $Tester) +(assert_trap (invoke "trap-src-reader-done") "cannot forward stream after being notified that the writable end dropped") +(component instance $i $Tester) +(assert_trap (invoke "trap-dest-writer-done") "cannot forward stream after being notified that the readable end dropped") +(component instance $i $Tester) +(assert_trap (invoke "trap-use-after-forward") "unknown handle index 1") +(component instance $i $Tester) +(assert_trap (invoke "trap-stream-forward-of-future-handles") "handle index 1 used with the wrong type") +(component instance $i $Tester) +(assert_trap (invoke "trap-future-swapped-args") "handle index 2 used with the wrong type") +(component instance $i $Tester) +(assert_trap (invoke "trap-future-reader-copying") "cannot forward busy future") +(component instance $i $Tester) +(assert_trap (invoke "trap-same-instance-rendezvous") "cannot read from and write to intra-component future/stream with non-numeric payload") +(component instance $i $Tester) +(assert_trap (invoke "trap-future-same-instance-rendezvous") "cannot read from and write to intra-component future/stream with non-numeric payload") + +;; A proxying use case: $P produces a stream, $X splices it into a stream of its +;; own and hands the readable end on, and $D consumes it. Because $X gives up +;; both handles in the same call, the copy from $P to $D runs with $X holding no +;; stream ends at all. +(component definition $ProxyTester + (component $P + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $PM + (import "" "mem" (memory 1)) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (global $w (mut i32) (i32.const 0)) + (func (export "start") (result i32) + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $w (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (i32.wrap_i64 (local.get $ret64)) + ) + (func (export "write4") (result i32) + (i32.store (i32.const 0) (i32.const 0x04030201)) + (call $stream.write (global.get $w) (i32.const 0) (i32.const 4)) + ) + ) + (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)) + (core instance $pm (instantiate $PM (with "" (instance + (export "mem" (memory $memory "mem")) + (export "stream.new" (func $stream.new)) + (export "stream.write" (func $stream.write)) + )))) + (func (export "start") (result (stream u8)) (canon lift (core func $pm "start"))) + (func (export "write4") (result u32) (canon lift (core func $pm "write4"))) + ) + (component $X + (core module $XM + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.forward" (func $stream.forward (param i32 i32))) + (func (export "proxy") (param $in i32) (result i32) + (local $ret64 i64) (local $r i32) (local $w i32) + (local.set $ret64 (call $stream.new)) + (local.set $r (i32.wrap_i64 (local.get $ret64))) + (local.set $w (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ;; splice the incoming stream into the outgoing one, giving up both the + ;; incoming readable end and the outgoing writable end + (call $stream.forward (local.get $in) (local.get $w)) + (local.get $r) + ) + ) + (type $ST (stream u8)) + (canon stream.new $ST (core func $stream.new)) + (canon stream.forward $ST (core func $stream.forward)) + (core instance $xm (instantiate $XM (with "" (instance + (export "stream.new" (func $stream.new)) + (export "stream.forward" (func $stream.forward)) + )))) + (func (export "proxy") (param "in" (stream u8)) (result (stream u8)) + (canon lift (core func $xm "proxy"))) + ) + (component $D + (import "p" (instance $p + (export "start" (func (result (stream u8)))) + (export "write4" (func (result u32))) + )) + (import "x" (instance $x + (export "proxy" (func (param "in" (stream u8)) (result (stream u8)))) + )) + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $DM + (import "" "mem" (memory 1)) + (import "" "start" (func $start (result i32))) + (import "" "write4" (func $write4 (result i32))) + (import "" "proxy" (func $proxy (param i32) (result i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.poll" (func $waitable-set.poll (param i32 i32) (result i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.drop-readable" (func $stream.drop-readable (param i32))) + + (func (export "proxy-read-then-write") (result i32) + (local $ret i32) (local $rq i32) (local $ws i32) + (local.set $rq (call $proxy (call $start))) + (local.set $ret (call $stream.read (local.get $rq) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + + ;; $P's write rendezvouses with $D's read even though the stream $D is + ;; reading from was created by $X + (local.set $ret (call $write4)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (local.get $rq) (local.get $ws)) + (local.set $ret (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (if (i32.ne (i32.const 2 (; STREAM_READ ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (local.get $rq) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (i32.load (i32.const 36))) + (then unreachable)) + + (i32.const 42) + ) + (func (export "proxy-write-then-read") (result i32) + (local $ret i32) (local $rq i32) + (local.set $rq (call $proxy (call $start))) + (local.set $ret (call $write4)) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (local.set $ret (call $stream.read (local.get $rq) (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 0x04030201) (i32.load (i32.const 8))) + (then unreachable)) + (i32.const 43) + ) + (func (export "proxy-drop-reader") (result i32) + (local $ret i32) (local $rq i32) + (local.set $rq (call $proxy (call $start))) + ;; dropping the far end of the spliced stream is visible to $P + (call $stream.drop-readable (local.get $rq)) + (local.set $ret (call $write4)) + (if (i32.ne (i32.const 0x1 (; DROPPED=1 | (0<<4) ;)) (local.get $ret)) + (then unreachable)) + (i32.const 44) + ) + ) + (type $ST (stream u8)) + (canon lower (func $p "start") (memory (core memory $memory "mem")) (core func $start)) + (canon lower (func $p "write4") (memory (core memory $memory "mem")) (core func $write4)) + (canon lower (func $x "proxy") (memory (core memory $memory "mem")) (core func $proxy)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.poll (memory (core memory $memory "mem")) (core func $waitable-set.poll)) + (canon waitable.join (core func $waitable.join)) + (canon stream.read $ST async (memory (core memory $memory "mem")) (core func $stream.read)) + (canon stream.drop-readable $ST (core func $stream.drop-readable)) + (core instance $dm (instantiate $DM (with "" (instance + (export "mem" (memory $memory "mem")) + (export "start" (func $start)) + (export "write4" (func $write4)) + (export "proxy" (func $proxy)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.poll" (func $waitable-set.poll)) + (export "waitable.join" (func $waitable.join)) + (export "stream.read" (func $stream.read)) + (export "stream.drop-readable" (func $stream.drop-readable)) + )))) + (func (export "proxy-read-then-write") (result u32) (canon lift (core func $dm "proxy-read-then-write"))) + (func (export "proxy-write-then-read") (result u32) (canon lift (core func $dm "proxy-write-then-read"))) + (func (export "proxy-drop-reader") (result u32) (canon lift (core func $dm "proxy-drop-reader"))) + ) + (instance $p (instantiate $P)) + (instance $x (instantiate $X)) + (instance $d (instantiate $D (with "p" (instance $p)) (with "x" (instance $x)))) + (func (export "proxy-read-then-write") (alias export $d "proxy-read-then-write")) + (func (export "proxy-write-then-read") (alias export $d "proxy-write-then-read")) + (func (export "proxy-drop-reader") (alias export $d "proxy-drop-reader")) +) + +(component instance $i $ProxyTester) +(assert_return (invoke "proxy-read-then-write") (u32.const 42)) +(component instance $i $ProxyTester) +(assert_return (invoke "proxy-write-then-read") (u32.const 43)) +(component instance $i $ProxyTester) +(assert_return (invoke "proxy-drop-reader") (u32.const 44)) + + +;; One stream forwarded five times, ping-ponging between two components: $C and +;; $D take turns splicing the chain into a stream of their own, and after every +;; splice the surviving writer writes and the surviving reader reads a distinct +;; 4-byte pattern all the way through. +(component definition $PingPongTester + (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.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (import "" "stream.forward" (func $stream.forward (param i32 i32))) + (import "" "stream.drop-writable" (func $stream.drop-writable (param i32))) + + (global $r (mut i32) (i32.const 0)) ;; the chain's readable end, while $C holds it + (global $w (mut i32) (i32.const 0)) ;; the writable end of the stream made by "fresh" + + ;; splice the incoming readable end into a fresh local stream, keeping the + ;; local readable end: $C becomes the reader of the merged stream + (func (export "splice") (param $in i32) + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $r (i32.wrap_i64 (local.get $ret64))) + (call $stream.forward (local.get $in) + (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func (export "read4") (param $expected i32) (result i32) + (local $ret i32) + (local.set $ret (call $stream.read (global.get $r) (i32.const 8) (i32.const 4))) + (if (i32.eq (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret)) + (then + (if (i32.ne (i32.load (i32.const 8)) (local.get $expected)) + (then unreachable)))) + (local.get $ret) + ) + ;; give the chain's readable end back to $D so that it can take the next turn + (func (export "hand-back") (result i32) + (global.get $r) + ) + ;; a brand new stream whose readable end $D can splice in front of the chain + (func (export "fresh") (result i32) + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $w (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + (i32.wrap_i64 (local.get $ret64)) + ) + (func (export "write4") (param $val i32) (result i32) + (i32.store (i32.const 0) (local.get $val)) + (call $stream.write (global.get $w) (i32.const 0) (i32.const 4)) + ) + (func (export "drop-writer") + (call $stream.drop-writable (global.get $w)) + ) + ) + (type $ST (stream u8)) + (canon stream.new $ST (core func $stream.new)) + (canon stream.read $ST async (memory (core memory $memory "mem")) (core func $stream.read)) + (canon stream.write $ST async (memory (core memory $memory "mem")) (core func $stream.write)) + (canon stream.forward $ST (core func $stream.forward)) + (canon stream.drop-writable $ST (core func $stream.drop-writable)) + (core instance $cm (instantiate $CM (with "" (instance + (export "mem" (memory $memory "mem")) + (export "stream.new" (func $stream.new)) + (export "stream.read" (func $stream.read)) + (export "stream.write" (func $stream.write)) + (export "stream.forward" (func $stream.forward)) + (export "stream.drop-writable" (func $stream.drop-writable)) + )))) + (func (export "splice") (param "in" (stream u8)) (canon lift (core func $cm "splice"))) + (func (export "read4") (param "expected" u32) (result s32) (canon lift (core func $cm "read4"))) + (func (export "hand-back") (result (stream u8)) (canon lift (core func $cm "hand-back"))) + (func (export "fresh") (result (stream u8)) (canon lift (core func $cm "fresh"))) + (func (export "write4") (param "val" u32) (result s32) (canon lift (core func $cm "write4"))) + (func (export "drop-writer") (canon lift (core func $cm "drop-writer"))) + ) + (component $D + (import "c" (instance $c + (export "splice" (func (param "in" (stream u8)))) + (export "read4" (func (param "expected" u32) (result s32))) + (export "hand-back" (func (result (stream u8)))) + (export "fresh" (func (result (stream u8)))) + (export "write4" (func (param "val" u32) (result s32))) + (export "drop-writer" (func)) + )) + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $DM + (import "" "mem" (memory 1)) + (import "" "c-splice" (func $c-splice (param i32))) + (import "" "c-read4" (func $c-read4 (param i32) (result i32))) + (import "" "c-hand-back" (func $c-hand-back (result i32))) + (import "" "c-fresh" (func $c-fresh (result i32))) + (import "" "c-write4" (func $c-write4 (param i32) (result i32))) + (import "" "c-drop-writer" (func $c-drop-writer)) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (import "" "stream.forward" (func $stream.forward (param i32 i32))) + (import "" "stream.drop-readable" (func $stream.drop-readable (param i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable-set.poll" (func $waitable-set.poll (param i32 i32) (result i32))) + (import "" "waitable-set.drop" (func $waitable-set.drop (param i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + + (global $r (mut i32) (i32.const 0)) ;; the chain's readable end, while $D holds it + (global $w (mut i32) (i32.const 0)) ;; the chain's writable end, while $D holds it + + ;; the mirror image of $C's "splice": $D becomes the reader + (func $d-splice (param $in i32) + (local $ret64 i64) + (local.set $ret64 (call $stream.new)) + (global.set $r (i32.wrap_i64 (local.get $ret64))) + (call $stream.forward (local.get $in) + (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + ) + (func $d-write4 (param $val i32) + (local $ret i32) + (i32.store (i32.const 0) (local.get $val)) + (local.set $ret (call $stream.write (global.get $w) (i32.const 0) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + ) + (func $d-read4 (param $expected i32) + (local $ret i32) + (local.set $ret (call $stream.read (global.get $r) (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.load (i32.const 8)) (local.get $expected)) + (then unreachable)) + ) + (func $expect-c-read4 (param $expected i32) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) + (call $c-read4 (local.get $expected))) + (then unreachable)) + ) + ;; the blocked write is completed by the far side's read, so its event has + ;; to be taken before the same writable end can be used for the next hop + (func $poll (param $waitable i32) (param $event i32) (param $payload i32) + (local $ws i32) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (local.get $waitable) (local.get $ws)) + (if (i32.ne (local.get $event) (call $waitable-set.poll (local.get $ws) (i32.const 32))) + (then unreachable)) + (if (i32.ne (local.get $waitable) (i32.load (i32.const 32))) + (then unreachable)) + (if (i32.ne (local.get $payload) (i32.load (i32.const 36))) + (then unreachable)) + (call $waitable.join (local.get $waitable) (i32.const 0)) + (call $waitable-set.drop (local.get $ws)) + ) + + (func (export "run") (result i32) + (local $ret i32) (local $ret64 i64) + + ;; $D makes the stream and keeps the writable end for the first four hops + (local.set $ret64 (call $stream.new)) + (global.set $r (i32.wrap_i64 (local.get $ret64))) + (global.set $w (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32)))) + + ;; hop 1: $C splices, so the merged stream runs from $D's writer to $C's + ;; reader and the copy crosses a component boundary + (call $c-splice (global.get $r)) + (call $d-write4 (i32.const 0x11111111)) + (call $expect-c-read4 (i32.const 0x11111111)) + (call $poll (global.get $w) (i32.const 3 (; STREAM_WRITE ;)) (i32.const 0x40)) + + ;; hop 2: $C hands the reader back and $D splices, so this time both ends + ;; of the merged stream are $D's own + (call $d-splice (call $c-hand-back)) + (call $d-write4 (i32.const 0x22222222)) + (call $d-read4 (i32.const 0x22222222)) + (call $poll (global.get $w) (i32.const 3 (; STREAM_WRITE ;)) (i32.const 0x40)) + + ;; hop 3: and back over to $C + (call $c-splice (global.get $r)) + (call $d-write4 (i32.const 0x33333333)) + (call $expect-c-read4 (i32.const 0x33333333)) + (call $poll (global.get $w) (i32.const 3 (; STREAM_WRITE ;)) (i32.const 0x40)) + + ;; hop 4: and back to $D + (call $d-splice (call $c-hand-back)) + (call $d-write4 (i32.const 0x44444444)) + (call $d-read4 (i32.const 0x44444444)) + (call $poll (global.get $w) (i32.const 3 (; STREAM_WRITE ;)) (i32.const 0x40)) + + ;; hop 5: $D splices a stream made by $C in front of the chain, which is + ;; the one way the writer can end up in the other component. $D keeps the + ;; reader it already had, so the copy now runs $C -> $D. + (call $stream.forward (call $c-fresh) (global.get $w)) + (local.set $ret (call $stream.read (global.get $r) (i32.const 8) (i32.const 4))) + (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret)) + (then unreachable)) + (if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) + (call $c-write4 (i32.const 0x55555555))) + (then unreachable)) + (call $poll (global.get $r) (i32.const 2 (; STREAM_READ ;)) (i32.const 0x40)) + (if (i32.ne (i32.load (i32.const 8)) (i32.const 0x55555555)) + (then unreachable)) + + (call $stream.drop-readable (global.get $r)) + (call $c-drop-writer) + (i32.const 42) + ) + ) + (type $ST (stream u8)) + (canon lower (func $c "splice") (memory (core memory $memory "mem")) (core func $c-splice)) + (canon lower (func $c "read4") (memory (core memory $memory "mem")) (core func $c-read4)) + (canon lower (func $c "hand-back") (memory (core memory $memory "mem")) (core func $c-hand-back)) + (canon lower (func $c "fresh") (memory (core memory $memory "mem")) (core func $c-fresh)) + (canon lower (func $c "write4") (memory (core memory $memory "mem")) (core func $c-write4)) + (canon lower (func $c "drop-writer") (core func $c-drop-writer)) + (canon stream.new $ST (core func $stream.new)) + (canon stream.read $ST async (memory (core memory $memory "mem")) (core func $stream.read)) + (canon stream.write $ST async (memory (core memory $memory "mem")) (core func $stream.write)) + (canon stream.forward $ST (core func $stream.forward)) + (canon stream.drop-readable $ST (core func $stream.drop-readable)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable-set.poll (memory (core memory $memory "mem")) (core func $waitable-set.poll)) + (canon waitable-set.drop (core func $waitable-set.drop)) + (canon waitable.join (core func $waitable.join)) + (core instance $dm (instantiate $DM (with "" (instance + (export "mem" (memory $memory "mem")) + (export "c-splice" (func $c-splice)) + (export "c-read4" (func $c-read4)) + (export "c-hand-back" (func $c-hand-back)) + (export "c-fresh" (func $c-fresh)) + (export "c-write4" (func $c-write4)) + (export "c-drop-writer" (func $c-drop-writer)) + (export "stream.new" (func $stream.new)) + (export "stream.read" (func $stream.read)) + (export "stream.write" (func $stream.write)) + (export "stream.forward" (func $stream.forward)) + (export "stream.drop-readable" (func $stream.drop-readable)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable-set.poll" (func $waitable-set.poll)) + (export "waitable-set.drop" (func $waitable-set.drop)) + (export "waitable.join" (func $waitable.join)) + )))) + (func (export "run") (result u32) (canon lift (core func $dm "run"))) + ) + (instance $c (instantiate $C)) + (instance $d (instantiate $D (with "c" (instance $c)))) + (func (export "run") (alias export $d "run")) +) + +(component instance $i $PingPongTester) +(assert_return (invoke "run") (u32.const 42)) + + +(assert_invalid + (component + (type $FT (future u8)) + (canon stream.forward $FT (core func $f)) + ) + "`stream.forward` requires a stream type") +(assert_invalid + (component + (type $ST (stream u8)) + (canon future.forward $ST (core func $f)) + ) + "`future.forward` requires a future type") diff --git a/test/nyi.txt b/test/nyi.txt index cdc233f7..3efe95b6 100644 --- a/test/nyi.txt +++ b/test/nyi.txt @@ -1,3 +1,6 @@ # See README.md ./validation/kebab.wast ./binary/binary.wast +./values/post-return.wast +./async/big-interleaving-test.wast +./async/forward.wast diff --git a/test/values/post-return.wast b/test/values/post-return.wast index 936dd457..62c0350d 100644 --- a/test/values/post-return.wast +++ b/test/values/post-return.wast @@ -37,6 +37,7 @@ (canon stream.cancel-write $ST (core func $stream.cancel-write)) (canon stream.drop-readable $ST (core func $stream.drop-readable)) (canon stream.drop-writable $ST (core func $stream.drop-writable)) + (canon stream.forward $ST (core func $stream.forward)) (canon future.new $FT (core func $future.new)) (canon future.read $FT (memory (core memory $memory "mem")) (core func $future.read)) (canon future.write $FT (memory (core memory $memory "mem")) (core func $future.write)) @@ -44,6 +45,7 @@ (canon future.cancel-write $FT (core func $future.cancel-write)) (canon future.drop-readable $FT (core func $future.drop-readable)) (canon future.drop-writable $FT (core func $future.drop-writable)) + (canon future.forward $FT (core func $future.forward)) (core module $DM (import "" "mem" (memory 1)) (import "" "import" (func $import)) @@ -67,6 +69,7 @@ (import "" "stream.cancel-write" (func $stream.cancel-write (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 "" "stream.forward" (func $stream.forward (param i32 i32))) (import "" "future.new" (func $future.new (result i64))) (import "" "future.read" (func $future.read (param i32 i32) (result i32))) (import "" "future.write" (func $future.write (param i32 i32) (result i32))) @@ -74,6 +77,7 @@ (import "" "future.cancel-write" (func $future.cancel-write (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 "" "future.forward" (func $future.forward (param i32 i32))) (func (export "noop")) (func (export "trap-calling-import") (call $import)) @@ -97,6 +101,7 @@ (func (export "trap-calling-stream-cancel-write") (drop (call $stream.cancel-write (i32.const 0)))) (func (export "trap-calling-stream-drop-readable") (call $stream.drop-readable (i32.const 0))) (func (export "trap-calling-stream-drop-writable") (call $stream.drop-writable (i32.const 0))) + (func (export "trap-calling-stream-forward") (call $stream.forward (i32.const 0) (i32.const 0))) (func (export "trap-calling-future-new") (drop (call $future.new))) (func (export "trap-calling-future-read") (drop (call $future.read (i32.const 0) (i32.const 0)))) (func (export "trap-calling-future-write") (drop (call $future.write (i32.const 0) (i32.const 0)))) @@ -104,6 +109,7 @@ (func (export "trap-calling-future-cancel-write") (drop (call $future.cancel-write (i32.const 0)))) (func (export "trap-calling-future-drop-readable") (call $future.drop-readable (i32.const 0))) (func (export "trap-calling-future-drop-writable") (call $future.drop-writable (i32.const 0))) + (func (export "trap-calling-future-forward") (call $future.forward (i32.const 0) (i32.const 0))) ) (canon lower (func $import) (core func $import')) (core instance $dm (instantiate $DM (with "" (instance @@ -129,6 +135,7 @@ (export "stream.cancel-write" (func $stream.cancel-write)) (export "stream.drop-readable" (func $stream.drop-readable)) (export "stream.drop-writable" (func $stream.drop-writable)) + (export "stream.forward" (func $stream.forward)) (export "future.new" (func $future.new)) (export "future.read" (func $future.read)) (export "future.write" (func $future.write)) @@ -136,6 +143,7 @@ (export "future.cancel-write" (func $future.cancel-write)) (export "future.drop-readable" (func $future.drop-readable)) (export "future.drop-writable" (func $future.drop-writable)) + (export "future.forward" (func $future.forward)) )))) (func (export "trap-calling-import") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-import")))) (func (export "trap-calling-resource-new") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-resource-new")))) @@ -158,6 +166,7 @@ (func (export "trap-calling-stream-cancel-write") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-stream-cancel-write")))) (func (export "trap-calling-stream-drop-readable") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-stream-drop-readable")))) (func (export "trap-calling-stream-drop-writable") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-stream-drop-writable")))) + (func (export "trap-calling-stream-forward") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-stream-forward")))) (func (export "trap-calling-future-new") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-new")))) (func (export "trap-calling-future-read") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-read")))) (func (export "trap-calling-future-write") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-write")))) @@ -165,6 +174,7 @@ (func (export "trap-calling-future-cancel-write") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-cancel-write")))) (func (export "trap-calling-future-drop-readable") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-drop-readable")))) (func (export "trap-calling-future-drop-writable") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-drop-writable")))) + (func (export "trap-calling-future-forward") (canon lift (core func $dm "noop") (post-return (core func $dm "trap-calling-future-forward")))) ) (instance $c (instantiate $C)) (instance $d (instantiate $D (with "import" (func $c "import")))) @@ -189,6 +199,7 @@ (func (export "trap-calling-stream-cancel-write") (alias export $d "trap-calling-stream-cancel-write")) (func (export "trap-calling-stream-drop-readable") (alias export $d "trap-calling-stream-drop-readable")) (func (export "trap-calling-stream-drop-writable") (alias export $d "trap-calling-stream-drop-writable")) + (func (export "trap-calling-stream-forward") (alias export $d "trap-calling-stream-forward")) (func (export "trap-calling-future-new") (alias export $d "trap-calling-future-new")) (func (export "trap-calling-future-read") (alias export $d "trap-calling-future-read")) (func (export "trap-calling-future-write") (alias export $d "trap-calling-future-write")) @@ -196,6 +207,7 @@ (func (export "trap-calling-future-cancel-write") (alias export $d "trap-calling-future-cancel-write")) (func (export "trap-calling-future-drop-readable") (alias export $d "trap-calling-future-drop-readable")) (func (export "trap-calling-future-drop-writable") (alias export $d "trap-calling-future-drop-writable")) + (func (export "trap-calling-future-forward") (alias export $d "trap-calling-future-forward")) ) (component instance $i1 $Tester) @@ -254,6 +266,10 @@ (assert_trap (invoke "trap-calling-future-drop-readable") "cannot leave component instance") (component instance $i27 $Tester) (assert_trap (invoke "trap-calling-future-drop-writable") "cannot leave component instance") +(component instance $i28 $Tester) +(assert_trap (invoke "trap-calling-stream-forward") "cannot leave component instance") +(component instance $i29 $Tester) +(assert_trap (invoke "trap-calling-future-forward") "cannot leave component instance") ;; built-ins that don't trap: