;; This test contains two components $C and $D that test cancelling reads
;; and writes in the presence and absence of partial reads/writes.
;;
;; $C exports a function 'start-stream' that creates and holds onto a writable
;; stream in the global $sw as well as various operations that operate on $sw.
;; $D calls $C.start-stream to get the readable end and then drives the test.
(component
(component $C
(core module $Memory (memory (export "mem") 1))
(core instance $memory (instantiate $Memory))
(core module $CM
(import "" "mem" (memory 1))
(import "" "task.return" (func $task.return (param i32)))
(import "" "stream.new" (func $stream.new (result i64)))
(import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32)))
(import "" "stream.cancel-write" (func $stream.cancel-write (param i32) (result i32)))
(import "" "stream.drop-writable" (func $stream.drop-writable (param i32)))
(global $sw (mut i32) (i32.const 0))
(func $start-stream (export "start-stream") (result i32)
;; create a new stream, return the readable end to the caller
(local $ret64 i64)
(local.set $ret64 (call $stream.new))
(global.set $sw (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32))))
(i32.wrap_i64 (local.get $ret64))
)
(func $write4 (export "write4")
;; write 6 bytes into the stream, expecting to rendezvous with a stream.read
(local $ret i32)
(i32.store (i32.const 8) (i32.const 0xabcd))
(local.set $ret (call $stream.write (global.get $sw) (i32.const 8) (i32.const 4)))
(if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret))
(then unreachable))
)
(func $write4-and-drop (export "write4-and-drop")
(call $write4)
(call $stream.drop-writable (global.get $sw))
)
(func $start-blocking-write (export "start-blocking-write")
(local $ret i32)
;; prepare the write buffer
(i64.store (i32.const 8) (i64.const 0x123456789abcdef))
;; start one blocking write and immediately cancel it
(local.set $ret (call $stream.write (global.get $sw) (i32.const 8) (i32.const 8)))
(if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret))
(then unreachable))
(local.set $ret (call $stream.cancel-write (global.get $sw)))
(if (i32.ne (i32.const 0x2 (; CANCELLED ;)) (local.get $ret))
(then unreachable))
;; start a second blockign write and leave it pending
(local.set $ret (call $stream.write (global.get $sw) (i32.const 8) (i32.const 8)))
(if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret))
(then unreachable))
)
(func $cancel-after-read4 (export "cancel-after-read4")
(local $ret i32)
(local.set $ret (call $stream.cancel-write (global.get $sw)))
(if (i32.ne (i32.const 0x42 (; CANCELLED=2 | (4<<4) ;)) (local.get $ret))
(then unreachable))
)
)
(type $ST (stream u8))
(canon task.return (result u32) (core func $task.return))
(canon stream.new $ST (core func $stream.new))
(canon stream.write $ST async (memory $memory "mem") (core func $stream.write))
(canon stream.cancel-write $ST (core func $stream.cancel-write))
(canon stream.drop-writable $ST (core func $stream.drop-writable))
(core instance $cm (instantiate $CM (with "" (instance
(export "mem" (memory $memory "mem"))
(export "task.return" (func $task.return))
(export "stream.new" (func $stream.new))
(export "stream.write" (func $stream.write))
(export "stream.cancel-write" (func $stream.cancel-write))
(export "stream.drop-writable" (func $stream.drop-writable))
))))
(func (export "start-stream") async (result (stream u8)) (canon lift (core func $cm "start-stream")))
(func (export "write4") async (canon lift (core func $cm "write4")))
(func (export "write4-and-drop") async (canon lift (core func $cm "write4-and-drop")))
(func (export "start-blocking-write") async (canon lift (core func $cm "start-blocking-write")))
(func (export "cancel-after-read4") async (canon lift (core func $cm "cancel-after-read4")))
)
(component $D
(import "c" (instance $c
(export "start-stream" (func async (result (stream u8))))
(export "write4" (func async))
(export "write4-and-drop" (func async))
(export "start-blocking-write" (func async))
(export "cancel-after-read4" (func async))
))
(core module $Memory (memory (export "mem") 1))
(core instance $memory (instantiate $Memory))
(core module $DM
(import "" "mem" (memory 1))
(import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32)))
(import "" "stream.cancel-read" (func $stream.cancel-read (param i32) (result i32)))
(import "" "stream.drop-readable" (func $stream.drop-readable (param i32)))
(import "" "start-stream" (func $start-stream (result i32)))
(import "" "write4" (func $write4))
(import "" "write4-and-drop" (func $write4-and-drop))
(import "" "start-blocking-write" (func $start-blocking-write))
(import "" "cancel-after-read4" (func $cancel-after-read4))
(func $run (export "run") (result i32)
(local $ret i32)
(local $sr i32)
;; call 'start-stream' to get the stream we'll be working with
(local.set $sr (call $start-stream))
(if (i32.ne (i32.const 1) (local.get $sr))
(then unreachable))
;; start read that will block
(local.set $ret (call $stream.read (local.get $sr) (i32.const 8) (i32.const 100)))
(if (i32.ne (i32.const -1 (; BLOCKED;)) (local.get $ret))
(then unreachable))
;; cancelling it will finish without anything having been written
(local.set $ret (call $stream.cancel-read (local.get $sr)))
(if (i32.ne (i32.const 0x2 (; CANCELLED ;)) (local.get $ret))
(then unreachable))
;; read, block, call $C to write 4 bytes into the buffer,
;; then cancel, which should show "4+cancelled"
(local.set $ret (call $stream.read (local.get $sr) (i32.const 8) (i32.const 100)))
(if (i32.ne (i32.const -1 (; BLOCKED;)) (local.get $ret))
(then unreachable))
(call $write4)
(local.set $ret (call $stream.cancel-read (local.get $sr)))
(if (i32.ne (i32.const 0x42 (; CANCELLED=2 | (4<<4) ;)) (local.get $ret))
(then unreachable))
(if (i32.ne (i32.const 0xabcd) (i32.load (i32.const 8)))
(then unreachable))
;; read, block, call $C to write 4 bytes into the buffer and drop, then cancel
(local.set $ret (call $stream.read (local.get $sr) (i32.const 8) (i32.const 100)))
(if (i32.ne (i32.const -1 (; BLOCKED;)) (local.get $ret))
(then unreachable))
(call $write4-and-drop)
(local.set $ret (call $stream.cancel-read (local.get $sr)))
(if (i32.ne (i32.const 0x41 (; DROPPED=1 | (4<<4) ;)) (local.get $ret))
(then unreachable))
(if (i32.ne (i32.const 0xabcd) (i32.load (i32.const 8)))
(then unreachable))
(call $stream.drop-readable (local.get $sr))
;; get a new $sr
(local.set $sr (call $start-stream))
(if (i32.ne (i32.const 1) (local.get $sr))
(then unreachable))
;; start outstanding write in $C, read 4 of it, then call back into $C
;; which will cancel and see 4 written.
(call $start-blocking-write)
(local.set $ret (call $stream.read (local.get $sr) (i32.const 8) (i32.const 4)))
(if (i32.ne (i32.const 0x40 (; COMPLETED=0 | (4<<4) ;)) (local.get $ret))
(then unreachable))
(if (i32.ne (i32.const 0x89abcdef) (i32.load (i32.const 8)))
(then unreachable))
(call $cancel-after-read4)
;; return 42 to the top-level assert_return
(i32.const 42)
)
)
(type $ST (stream u8))
(canon stream.read $ST async (memory $memory "mem") (core func $stream.read))
(canon stream.cancel-read $ST (core func $stream.cancel-read))
(canon stream.drop-readable $ST (core func $stream.drop-readable))
(canon lower (func $c "start-stream") (core func $start-stream'))
(canon lower (func $c "write4") (core func $write4'))
(canon lower (func $c "write4-and-drop") (core func $write4-and-drop'))
(canon lower (func $c "start-blocking-write") (core func $start-blocking-write'))
(canon lower (func $c "cancel-after-read4") (core func $cancel-after-read4'))
(core instance $dm (instantiate $DM (with "" (instance
(export "mem" (memory $memory "mem"))
(export "stream.read" (func $stream.read))
(export "stream.cancel-read" (func $stream.cancel-read))
(export "stream.drop-readable" (func $stream.drop-readable))
(export "start-stream" (func $start-stream'))
(export "write4" (func $write4'))
(export "write4-and-drop" (func $write4-and-drop'))
(export "start-blocking-write" (func $start-blocking-write'))
(export "cancel-after-read4" (func $cancel-after-read4'))
))))
(func (export "run") async (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"))
)
(assert_return (invoke "run") (u32.const 42))