diff --git a/test/async/dtor-context-is-zero.wast b/test/async/dtor-context-is-zero.wast new file mode 100644 index 00000000..7c86473b --- /dev/null +++ b/test/async/dtor-context-is-zero.wast @@ -0,0 +1,158 @@ +;; A destructor called by `resource.drop` runs on a fresh thread, so its +;; `context.get` is always zero on entry, even when the destructor is +;; reentered recursively or another thread of its instance is blocked. + +;; The destructor recursively reenters itself via donut wrapping +(component + (core module $Indirect + (table (export "tbl") 1 funcref) + (func (export "dtor") (param i32) + (call_indirect (param i32) (local.get 0) (i32.const 0)))) + (core instance $indirect (instantiate $Indirect)) + (type $R (resource (rep i32) (dtor (core func $indirect "dtor")))) + (canon resource.new $R (core func $resource.new)) + (canon context.get i32 0 (core func $context.get)) + (canon context.set i32 0 (core func $context.set)) + + (component $C + (import "r" (type $R (sub resource))) + (canon resource.drop $R (core func $resource.drop)) + (func (export "drop") (param "r" (own $R)) (canon lift (core func $resource.drop)))) + (instance $c (instantiate $C (with "r" (type $R)))) + (canon lower (func $c "drop") (core func $drop)) + + (core module $M + (import "" "tbl" (table 1 funcref)) + (import "" "resource.new" (func $resource.new (param i32) (result i32))) + (import "" "drop" (func $drop (param i32))) + (import "" "context.get" (func $context.get (result i32))) + (import "" "context.set" (func $context.set (param i32))) + ;; dtor(n) sets its context to n+1, then (if n > 0) has $C drop a new + ;; resource with rep n-1, which calls back into dtor(n-1) + (func $dtor (param $n i32) + (if (call $context.get) (then unreachable)) + (call $context.set (i32.add (local.get $n) (i32.const 1))) + (if (local.get $n) (then + (call $drop (call $resource.new (i32.sub (local.get $n) (i32.const 1)))))) + (if (i32.ne (call $context.get) (i32.add (local.get $n) (i32.const 1))) + (then unreachable))) + (elem (i32.const 0) $dtor) + (func (export "run") (param $n i32) + (call $context.set (i32.const 100)) + (call $drop (call $resource.new (local.get $n))) + (if (i32.ne (call $context.get) (i32.const 100)) (then unreachable))) + ) + (core instance $m (instantiate $M (with "" (instance + (export "tbl" (table $indirect "tbl")) + (export "resource.new" (func $resource.new)) + (export "drop" (func $drop)) + (export "context.get" (func $context.get)) + (export "context.set" (func $context.set)))))) + (func (export "run") (param "n" u32) (canon lift (core func $m "run"))) +) +(assert_return (invoke "run" (u32.const 3))) + +;; The destructor is called while another thread of its instance is blocked +;; inside an async callback, holding the instance's exclusive lock +(component + (component $P + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (type $FT (future)) + (canon context.get i32 0 (core func $context.get)) + (canon context.set i32 0 (core func $context.set)) + (canon future.read $FT (memory (core memory $memory "mem")) (core func $future.read)) + (canon task.return (core func $task.return)) + (core module $M + (import "" "context.get" (func $context.get (result i32))) + (import "" "context.set" (func $context.set (param i32))) + (import "" "future.read" (func $future.read (param i32 i32) (result i32))) + (import "" "task.return" (func $task.return)) + (func (export "dtor") (param i32) + (if (call $context.get) (then unreachable)) + (call $context.set (i32.const 2))) + ;; blocks (holding the exclusive lock) until the caller writes $fut + (func (export "hold") (param $fut i32) (result i32) + (call $context.set (i32.const 1)) + (if (call $future.read (local.get $fut) (i32.const 0)) (then unreachable)) + (if (i32.ne (call $context.get) (i32.const 1)) (then unreachable)) + (call $task.return) + (i32.const 0 (; EXIT ;))) + (func (export "cb") (param i32 i32 i32) (result i32) unreachable) + ) + (core instance $m (instantiate $M (with "" (instance + (export "context.get" (func $context.get)) + (export "context.set" (func $context.set)) + (export "future.read" (func $future.read)) + (export "task.return" (func $task.return)))))) + (type $R (resource (rep i32) (dtor (core func $m "dtor")))) + (export $R' "r" (type $R)) + (canon resource.new $R (core func $resource.new)) + (func (export "make") (param "rep" u32) (result (own $R')) + (canon lift (core func $resource.new))) + (func (export "hold") async (param "fut" $FT) + (canon lift (core func $m "hold") async (callback (core func $m "cb")))) + ) + + (component $D + (import "r" (type $R (sub resource))) + (type $FT (future)) + (import "make" (func $make (param "rep" u32) (result (own $R)))) + (import "hold" (func $hold async (param "fut" $FT))) + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (canon lower (func $make) (core func $make)) + (canon lower (func $hold) async (core func $hold)) + (canon resource.drop $R (core func $resource.drop)) + (canon future.new $FT (core func $future.new)) + (canon future.write $FT (memory (core memory $memory "mem")) (core func $future.write)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable.join (core func $waitable.join)) + (canon waitable-set.wait (memory (core memory $memory "mem")) (core func $waitable-set.wait)) + (core module $M + (import "" "make" (func $make (param i32) (result i32))) + (import "" "hold" (func $hold (param i32) (result i32))) + (import "" "resource.drop" (func $resource.drop (param i32))) + (import "" "future.new" (func $future.new (result i64))) + (import "" "future.write" (func $future.write (param i32 i32) (result i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32))) + (func (export "run") (result i32) + (local $r i32) (local $fut i64) (local $ret i32) (local $ws i32) + (local.set $r (call $make (i32.const 0))) + (local.set $fut (call $future.new)) + (local.set $ret (call $hold (i32.wrap_i64 (local.get $fut)))) + (if (i32.ne (i32.and (local.get $ret) (i32.const 0xf)) (i32.const 1 (; STARTED ;))) + (then unreachable)) + ;; "hold" is now blocked, so this dtor call overlaps it + (call $resource.drop (local.get $r)) + ;; unblock "hold" and wait for it to finish its checks + (if (call $future.write (i32.wrap_i64 (i64.shr_u (local.get $fut) (i64.const 32))) (i32.const 0)) + (then unreachable)) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (i32.shr_u (local.get $ret) (i32.const 4)) (local.get $ws)) + (if (i32.ne (call $waitable-set.wait (local.get $ws) (i32.const 0)) (i32.const 1 (; SUBTASK ;))) + (then unreachable)) + (i32.const 42)) + ) + (core instance $m (instantiate $M (with "" (instance + (export "make" (func $make)) + (export "hold" (func $hold)) + (export "resource.drop" (func $resource.drop)) + (export "future.new" (func $future.new)) + (export "future.write" (func $future.write)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable.join" (func $waitable.join)) + (export "waitable-set.wait" (func $waitable-set.wait)))))) + (func (export "run") async (result u32) (canon lift (core func $m "run"))) + ) + + (instance $p (instantiate $P)) + (instance $d (instantiate $D + (with "r" (type $p "r")) + (with "make" (func $p "make")) + (with "hold" (func $p "hold")))) + (func (export "run") (alias export $d "run")) +) +(assert_return (invoke "run") (u32.const 42)) diff --git a/test/async/during-sync-call-blocking-ops.wast b/test/async/during-sync-call-blocking-ops.wast new file mode 100644 index 00000000..22903575 --- /dev/null +++ b/test/async/during-sync-call-blocking-ops.wast @@ -0,0 +1,247 @@ +;; When the implicit thread of a non-async-typed call blocks before returning, +;; the runtime resumes other ready threads of the same component instance until +;; the call returns. Other during-sync-* tests block with thread.suspend and +;; thread.yield; this test checks, one export per operation, that the other +;; blocking operations can also switch to a ready thread. +(component + (component $C + (core module $Core + (import "" "thread.index" (func $thread.index (result i32))) + (import "" "thread.suspend" (func $thread.suspend (result i32))) + (import "" "thread.yield-then-resume" (func $thread.yield-then-resume (param i32) (result i32))) + + (global $blocked (mut i32) (i32.const 0)) + + (func (export "block") + (global.set $blocked (call $thread.index)) + (drop (call $thread.suspend))) + + (func (export "unblock") + (drop (call $thread.yield-then-resume (global.get $blocked)))) + ) + (canon thread.index (core func $thread.index)) + (canon thread.suspend (core func $thread.suspend)) + (canon thread.yield-then-resume (core func $thread.yield-then-resume)) + (core instance $core (instantiate $Core (with "" (instance + (export "thread.index" (func $thread.index)) + (export "thread.suspend" (func $thread.suspend)) + (export "thread.yield-then-resume" (func $thread.yield-then-resume)) + )))) + (func (export "block") async + (canon lift (core func $core "block"))) + (func (export "unblock") + (canon lift (core func $core "unblock"))) + ) + (instance $c (instantiate $C)) + + (core module $Memory (memory (export "mem") 1)) + (core instance $memory (instantiate $Memory)) + (core module $Table (table (export "tbl") 8 funcref)) + (core instance $table (instantiate $Table)) + (core module $Core + (import "" "mem" (memory 1)) + (import "" "tbl" (table 8 funcref)) + (import "" "block" (func $block)) + (import "" "block-async" (func $block-async (result i32))) + (import "" "unblock" (func $unblock)) + (import "" "thread.index" (func $thread.index (result i32))) + (import "" "thread.new-indirect" (func $thread.new-indirect (param i32 i32) (result i32))) + (import "" "thread.resume-later" (func $thread.resume-later (param i32))) + (import "" "thread.suspend-then-promote" (func $thread.suspend-then-promote (param i32) (result i32))) + (import "" "thread.yield-then-promote" (func $thread.yield-then-promote (param i32) (result i32))) + (import "" "subtask.cancel" (func $subtask.cancel (param i32) (result i32))) + (import "" "waitable-set.new" (func $waitable-set.new (result i32))) + (import "" "waitable.join" (func $waitable.join (param i32 i32))) + (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32))) + (import "" "future.new" (func $future.new (result i64))) + (import "" "future.read" (func $future.read (param i32 i32) (result i32))) + (import "" "future.read-async" (func $future.read-async (param i32 i32) (result i32))) + (import "" "future.write" (func $future.write (param i32 i32) (result 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))) + + (func $expect (param $actual i32) (param $expected i32) + (if (i32.ne (local.get $actual) (local.get $expected)) + (then unreachable))) + (func $readable (param $ends i64) (result i32) + (i32.wrap_i64 (local.get $ends))) + (func $writable (param $ends i64) (result i32) + (i32.wrap_i64 (i64.shr_u (local.get $ends) (i64.const 32)))) + + ;; helper start functions, which record that they ran and then use their + ;; argument (if any) to unblock the operation + (global $helper-ran (mut i32) (i32.const 0)) + (func $call-unblock (param i32) + (global.set $helper-ran (i32.const 1)) + (call $unblock)) + (func $write-future (param $writable i32) + (global.set $helper-ran (i32.const 1)) + (call $expect (call $future.write (local.get $writable) (i32.const 0)) (i32.const 0 (; COMPLETED ;)))) + (func $read-future (param $readable i32) + (global.set $helper-ran (i32.const 1)) + (call $expect (call $future.read (local.get $readable) (i32.const 0)) (i32.const 0 (; COMPLETED ;)))) + (func $write-stream (param $writable i32) + (global.set $helper-ran (i32.const 1)) + (call $expect (call $stream.write (local.get $writable) (i32.const 0) (i32.const 1)) (i32.const 0x10 (; COMPLETED, 1 ;)))) + (func $read-stream (param $readable i32) + (global.set $helper-ran (i32.const 1)) + (call $expect (call $stream.read (local.get $readable) (i32.const 0) (i32.const 1)) (i32.const 0x10 (; COMPLETED, 1 ;)))) + (func $resume-later (param $thread i32) + (global.set $helper-ran (i32.const 1)) + (call $thread.resume-later (local.get $thread))) + (func $nothing (param i32) + (global.set $helper-ran (i32.const 1))) + (func $unreachable (param i32) + unreachable) + (elem (i32.const 0) func $call-unblock $write-future $read-future $write-stream $read-stream $resume-later $nothing $unreachable) + + (func $spawn (param $helper i32) (param $arg i32) + (global.set $helper-ran (i32.const 0)) + (call $thread.resume-later (call $thread.new-indirect (local.get $helper) (local.get $arg)))) + (func $done (result i32) + (call $expect (global.get $helper-ran) (i32.const 1)) + (i32.const 42)) + + ;; a thread that is never made ready, and so never runs + (func $never-ready (result i32) + (call $thread.new-indirect (i32.const 7 (; $unreachable ;)) (i32.const 0))) + + (func (export "sync-call") (result i32) + (call $spawn (i32.const 0 (; $call-unblock ;)) (i32.const 0)) + (call $block) + (call $done)) + + (func (export "waitable-set-wait") (result i32) + (local $ends i64) (local $ws i32) + (local.set $ends (call $future.new)) + (call $expect (call $future.read-async (call $readable (local.get $ends)) (i32.const 0)) (i32.const -1 (; BLOCKED ;))) + (local.set $ws (call $waitable-set.new)) + (call $waitable.join (call $readable (local.get $ends)) (local.get $ws)) + (call $spawn (i32.const 1 (; $write-future ;)) (call $writable (local.get $ends))) + (call $expect (call $waitable-set.wait (local.get $ws) (i32.const 0)) (i32.const 4 (; FUTURE_READ ;))) + (call $done)) + + (func (export "future-read") (result i32) + (local $ends i64) + (local.set $ends (call $future.new)) + (call $spawn (i32.const 1 (; $write-future ;)) (call $writable (local.get $ends))) + (call $expect (call $future.read (call $readable (local.get $ends)) (i32.const 0)) (i32.const 0 (; COMPLETED ;))) + (call $done)) + + (func (export "future-write") (result i32) + (local $ends i64) + (local.set $ends (call $future.new)) + (call $spawn (i32.const 2 (; $read-future ;)) (call $readable (local.get $ends))) + (call $expect (call $future.write (call $writable (local.get $ends)) (i32.const 0)) (i32.const 0 (; COMPLETED ;))) + (call $done)) + + (func (export "stream-read") (result i32) + (local $ends i64) + (local.set $ends (call $stream.new)) + (call $spawn (i32.const 3 (; $write-stream ;)) (call $writable (local.get $ends))) + (call $expect (call $stream.read (call $readable (local.get $ends)) (i32.const 0) (i32.const 1)) (i32.const 0x10 (; COMPLETED, 1 ;))) + (call $done)) + + (func (export "stream-write") (result i32) + (local $ends i64) + (local.set $ends (call $stream.new)) + (call $spawn (i32.const 4 (; $read-stream ;)) (call $readable (local.get $ends))) + (call $expect (call $stream.write (call $writable (local.get $ends)) (i32.const 0) (i32.const 1)) (i32.const 0x10 (; COMPLETED, 1 ;))) + (call $done)) + + ;; "block" is suspended inside core wasm (not in a callback event loop), so + ;; it can't receive the cancellation request, and the cancellation blocks + ;; until "block" returns + (func (export "subtask-cancel") (result i32) + (local $ret i32) + (local.set $ret (call $block-async)) + (call $expect (i32.and (local.get $ret) (i32.const 0xf)) (i32.const 1 (; STARTED ;))) + (call $spawn (i32.const 0 (; $call-unblock ;)) (i32.const 0)) + (call $expect (call $subtask.cancel (i32.shr_u (local.get $ret) (i32.const 4))) (i32.const 2 (; RETURNED ;))) + (call $done)) + + ;; *-then-promote of a thread that isn't ready falls back to plain + ;; thread.{suspend,yield} + (func (export "suspend-then-promote") (result i32) + (call $spawn (i32.const 5 (; $resume-later ;)) (call $thread.index)) + (drop (call $thread.suspend-then-promote (call $never-ready))) + (call $done)) + + ;; Since a yield may complete without switching, keep yielding until the + ;; helper has run. + (func (export "yield-then-promote") (result i32) + (local $target i32) + (local.set $target (call $never-ready)) + (call $spawn (i32.const 6 (; $nothing ;)) (i32.const 0)) + (loop $again + (drop (call $thread.yield-then-promote (local.get $target))) + (br_if $again (i32.eqz (global.get $helper-ran)))) + (call $done)) + ) + (type $FT (future)) + (type $ST (stream)) + (alias core export $memory "mem" (core memory $mem)) + (alias core export $table "tbl" (core table $tbl)) + (core type $start-func-ty (func (param i32))) + (core func $thread.new-indirect (canon thread.new-indirect $start-func-ty (core table $tbl))) + (canon lower (func $c "block") (core func $block)) + (canon lower (func $c "block") async (core func $block-async)) + (canon lower (func $c "unblock") (core func $unblock)) + (canon thread.index (core func $thread.index)) + (canon thread.resume-later (core func $thread.resume-later)) + (canon thread.suspend-then-promote (core func $thread.suspend-then-promote)) + (canon thread.yield-then-promote (core func $thread.yield-then-promote)) + (canon subtask.cancel (core func $subtask.cancel)) + (canon waitable-set.new (core func $waitable-set.new)) + (canon waitable.join (core func $waitable.join)) + (canon waitable-set.wait (memory (core memory $mem)) (core func $waitable-set.wait)) + (canon future.new $FT (core func $future.new)) + (canon future.read $FT (core func $future.read)) + (canon future.read $FT async (core func $future.read-async)) + (canon future.write $FT (core func $future.write)) + (canon stream.new $ST (core func $stream.new)) + (canon stream.read $ST (core func $stream.read)) + (canon stream.write $ST (core func $stream.write)) + (core instance $core (instantiate $Core (with "" (instance + (export "mem" (memory $mem)) + (export "tbl" (table $tbl)) + (export "block" (func $block)) + (export "block-async" (func $block-async)) + (export "unblock" (func $unblock)) + (export "thread.index" (func $thread.index)) + (export "thread.new-indirect" (func $thread.new-indirect)) + (export "thread.resume-later" (func $thread.resume-later)) + (export "thread.suspend-then-promote" (func $thread.suspend-then-promote)) + (export "thread.yield-then-promote" (func $thread.yield-then-promote)) + (export "subtask.cancel" (func $subtask.cancel)) + (export "waitable-set.new" (func $waitable-set.new)) + (export "waitable.join" (func $waitable.join)) + (export "waitable-set.wait" (func $waitable-set.wait)) + (export "future.new" (func $future.new)) + (export "future.read" (func $future.read)) + (export "future.read-async" (func $future.read-async)) + (export "future.write" (func $future.write)) + (export "stream.new" (func $stream.new)) + (export "stream.read" (func $stream.read)) + (export "stream.write" (func $stream.write)) + )))) + (func (export "sync-call") (result u32) (canon lift (core func $core "sync-call"))) + (func (export "waitable-set-wait") (result u32) (canon lift (core func $core "waitable-set-wait"))) + (func (export "future-read") (result u32) (canon lift (core func $core "future-read"))) + (func (export "future-write") (result u32) (canon lift (core func $core "future-write"))) + (func (export "stream-read") (result u32) (canon lift (core func $core "stream-read"))) + (func (export "stream-write") (result u32) (canon lift (core func $core "stream-write"))) + (func (export "subtask-cancel") (result u32) (canon lift (core func $core "subtask-cancel"))) + (func (export "suspend-then-promote") (result u32) (canon lift (core func $core "suspend-then-promote"))) + (func (export "yield-then-promote") (result u32) (canon lift (core func $core "yield-then-promote"))) +) +(assert_return (invoke "sync-call") (u32.const 42)) +(assert_return (invoke "waitable-set-wait") (u32.const 42)) +(assert_return (invoke "future-read") (u32.const 42)) +(assert_return (invoke "future-write") (u32.const 42)) +(assert_return (invoke "stream-read") (u32.const 42)) +(assert_return (invoke "stream-write") (u32.const 42)) +(assert_return (invoke "subtask-cancel") (u32.const 42)) +(assert_return (invoke "suspend-then-promote") (u32.const 42)) +(assert_return (invoke "yield-then-promote") (u32.const 42)) diff --git a/test/async/during-sync-call-exclusive-resume.wast b/test/async/during-sync-call-exclusive-resume.wast index 40d2b8e2..3fc2b438 100644 --- a/test/async/during-sync-call-exclusive-resume.wast +++ b/test/async/during-sync-call-exclusive-resume.wast @@ -101,3 +101,69 @@ ) (assert_return (invoke "arm") (u32.const 1)) (assert_trap (invoke "sync-block") "cannot block a synchronous task before returning") + +;; Same as above, except the async callback task has *not* resolved when its +;; callback returns EXIT during the sync call, which traps. +(component + (component $C + (core module $Core + (import "" "thread.suspend" (func $thread.suspend (result i32))) + + (global $in-sync-call (mut i32) (i32.const 0)) + + ;; async callback task parks its implicit thread, ready, in its event + ;; loop *without* calling task.return + (func (export "arm") (result i32) + (i32.const 1 (; YIELD ;))) + + ;; Park again on a spurious wake-up outside the sync call; exit (without + ;; having resolved the task) when resumed during it. + (func (export "exit-cb") (param i32 i32 i32) (result i32) + (if (i32.eqz (global.get $in-sync-call)) + (then (return (i32.const 1 (; YIELD ;))))) + (i32.const 0 (; EXIT ;))) + + ;; non-async-typed: suspend, leaving the parked thread as the only ready + ;; thread in this instance + (func (export "sync-block") + (global.set $in-sync-call (i32.const 1)) + (drop (call $thread.suspend)) + unreachable) + ) + (canon thread.suspend (core func $thread.suspend)) + (core instance $core (instantiate $Core (with "" (instance + (export "thread.suspend" (func $thread.suspend)) + )))) + (func (export "arm") async + (canon lift (core func $core "arm") async (callback (core func $core "exit-cb")))) + (func (export "sync-block") + (canon lift (core func $core "sync-block"))) + ) + (component $D + (import "c" (instance $c + (export "arm" (func async)) + (export "sync-block" (func)) + )) + (core module $Core + (import "" "arm" (func $arm (result i32))) + (import "" "sync-block" (func $sync-block)) + ;; start 'arm' without waiting for it to resolve, then call 'sync-block' + (func (export "run") + (if (i32.ne (i32.and (call $arm) (i32.const 0xf)) (i32.const 1 (; STARTED ;))) + (then unreachable)) + (call $sync-block)) + ) + (canon lower (func $c "arm") async (core func $arm)) + (canon lower (func $c "sync-block") (core func $sync-block)) + (core instance $core (instantiate $Core (with "" (instance + (export "arm" (func $arm)) + (export "sync-block" (func $sync-block)) + )))) + (func (export "run") + (canon lift (core func $core "run"))) + ) + (instance $c (instantiate $C)) + (instance $d (instantiate $D (with "c" (instance $c)))) + (func (export "run") (alias export $d "run")) +) +(assert_trap (invoke "run") "cannot block a synchronous task before returning")