diff --git a/Cargo.lock b/Cargo.lock index d293387..1c8cd0b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -52,6 +52,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "arrayvec" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" + [[package]] name = "async-broadcast" version = "0.7.2" @@ -76,6 +82,113 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-executor" +version = "1.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a" +dependencies = [ + "async-task", + "concurrent-queue", + "fastrand", + "futures-lite", + "pin-project-lite", + "slab", +] + +[[package]] +name = "async-fs" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5" +dependencies = [ + "async-lock", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-io" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" +dependencies = [ + "autocfg", + "cfg-if", + "concurrent-queue", + "futures-io", + "futures-lite", + "parking", + "polling", + "rustix", + "slab", + "windows-sys 0.61.2", +] + +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + +[[package]] +name = "async-net" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b948000fad4873c1c9339d60f2623323a0cfd3816e5181033c6a5cb68b2accf7" +dependencies = [ + "async-io", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-process" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" +dependencies = [ + "async-channel", + "async-io", + "async-lock", + "async-signal", + "async-task", + "blocking", + "cfg-if", + "event-listener", + "futures-lite", + "rustix", +] + +[[package]] +name = "async-signal" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485" +dependencies = [ + "async-io", + "async-lock", + "atomic-waker", + "cfg-if", + "futures-core", + "futures-io", + "rustix", + "signal-hook-registry", + "slab", + "windows-sys 0.61.2", +] + +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + [[package]] name = "asyncband" version = "0.6.7" @@ -84,6 +197,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + [[package]] name = "base64" version = "0.22.1" @@ -98,7 +223,7 @@ dependencies = [ "async-channel", "asyncband", "divan", - "flume", + "flume 0.12.0", "pollster", "tokio", ] @@ -109,6 +234,19 @@ version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +[[package]] +name = "blocking" +version = "1.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" +dependencies = [ + "async-channel", + "async-task", + "futures-io", + "futures-lite", + "piper", +] + [[package]] name = "bytes" version = "1.12.1" @@ -163,6 +301,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + [[package]] name = "clap" version = "4.6.5" @@ -210,6 +354,146 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "compio" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6829f76b635c05ef91f97dc44c3b27eb72f346179bd47c5018cd033d913420d" +dependencies = [ + "compio-buf", + "compio-dispatcher", + "compio-driver", + "compio-fs", + "compio-io", + "compio-log", + "compio-net", + "compio-runtime", +] + +[[package]] +name = "compio-buf" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ebb4036bf394915196c09362e4fd5581ee8bf0f3302ab598bff9d646aea2061" +dependencies = [ + "arrayvec", + "bytes", + "libc", +] + +[[package]] +name = "compio-dispatcher" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ff5d99ec2303ea61b4f4d2e2ebde551fce16462a479c59da0ff637bfb1951c9" +dependencies = [ + "compio-driver", + "compio-runtime", + "flume 0.11.1", + "futures-channel", +] + +[[package]] +name = "compio-driver" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f8915a1e560ab8be655c23771502a107d2802941a83fbf142734bed60a11441" +dependencies = [ + "cfg-if", + "cfg_aliases", + "compio-buf", + "compio-log", + "crossbeam-queue", + "flume 0.11.1", + "futures-util", + "libc", + "once_cell", + "paste", + "polling", + "socket2", + "windows-sys 0.61.2", +] + +[[package]] +name = "compio-fs" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea585c274239b9fd350a484c75a31fd0df5f031805b6664d83a98f5e8b019e2f" +dependencies = [ + "cfg-if", + "cfg_aliases", + "compio-buf", + "compio-driver", + "compio-io", + "compio-runtime", + "libc", + "os_pipe", + "widestring", + "windows-sys 0.61.2", +] + +[[package]] +name = "compio-io" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1e64c6d723589492a4f5041394301e9903466a606f6d9bcc11e406f9f07e9ec" +dependencies = [ + "compio-buf", + "futures-util", + "paste", +] + +[[package]] +name = "compio-log" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc4e560213c1996b618da369b7c9109564b41af9033802ae534465c4ee4e132f" +dependencies = [ + "tracing", +] + +[[package]] +name = "compio-net" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e6ea7aa4a9f38d68dd0098a11232236cb3efd0ef3cab50ecdada3d745f1f776" +dependencies = [ + "cfg-if", + "compio-buf", + "compio-driver", + "compio-io", + "compio-runtime", + "either", + "libc", + "once_cell", + "socket2", + "widestring", + "windows-sys 0.61.2", +] + +[[package]] +name = "compio-runtime" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b467b43c646a60bae3bb3a55c2c200ea7c4416a63cc5a6070450aa6771b3c62e" +dependencies = [ + "async-task", + "cfg-if", + "compio-buf", + "compio-driver", + "compio-log", + "core_affinity", + "crossbeam-queue", + "futures-util", + "libc", + "once_cell", + "pin-project-lite", + "scoped-tls", + "slab", + "socket2", + "windows-sys 0.61.2", +] + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -254,6 +538,26 @@ dependencies = [ "url", ] +[[package]] +name = "core_affinity" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a034b3a7b624016c6e13f5df875747cc25f884156aad2abd12b6c46797971342" +dependencies = [ + "libc", + "num_cpus", + "winapi", +] + +[[package]] +name = "crossbeam-queue" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.22" @@ -311,6 +615,12 @@ dependencies = [ "litrs", ] +[[package]] +name = "either" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" + [[package]] name = "equivalent" version = "1.0.2" @@ -324,7 +634,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -355,12 +665,29 @@ dependencies = [ "tokio", ] +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + [[package]] name = "find-msvc-tools" version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "flume" version = "0.12.0" @@ -381,11 +708,50 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + [[package]] name = "futures-core" -version = "0.3.33" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-lite" +version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "parking", + "pin-project-lite", +] + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] [[package]] name = "futures-sink" @@ -393,6 +759,26 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-macro", + "futures-sink", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -416,6 +802,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "http" version = "1.5.0" @@ -619,6 +1011,16 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi", + "libc", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -631,6 +1033,16 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "os_pipe" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "parking" version = "2.2.1" @@ -660,6 +1072,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + [[package]] name = "percent-encoding" version = "2.3.2" @@ -672,6 +1090,31 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "piper" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" +dependencies = [ + "atomic-waker", + "fastrand", + "futures-io", +] + +[[package]] +name = "polling" +version = "3.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" +dependencies = [ + "cfg-if", + "concurrent-queue", + "hermit-abi", + "pin-project-lite", + "rustix", + "windows-sys 0.61.2", +] + [[package]] name = "pollster" version = "1.0.1" @@ -764,7 +1207,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -802,6 +1245,12 @@ dependencies = [ "untrusted", ] +[[package]] +name = "scoped-tls" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" + [[package]] name = "scopeguard" version = "1.2.0" @@ -877,12 +1326,35 @@ dependencies = [ "libc", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +[[package]] +name = "smol" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a33bd3e260892199c3ccfc487c88b2da2265080acb316cd920da72fdfd7c599f" +dependencies = [ + "async-channel", + "async-executor", + "async-fs", + "async-io", + "async-lock", + "async-net", + "async-process", + "blocking", + "futures-lite", +] + [[package]] name = "socket2" version = "0.6.5" @@ -968,11 +1440,24 @@ name = "tests-integration" version = "0.0.0" dependencies = [ "asyncband", + "compio", + "futures-lite", "pollster", + "smol", + "tests-integration-macros", "tokio", "tokio-test", ] +[[package]] +name = "tests-integration-macros" +version = "0.0.0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "thiserror" version = "2.0.20" @@ -1083,6 +1568,22 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1186,6 +1687,34 @@ dependencies = [ "libc", ] +[[package]] +name = "widestring" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" + +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 11086fb..f511464 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,14 @@ # under the License. [workspace] -members = ["asyncband", "benchmarks", "examples", "tests-integration", "xtask"] +members = [ + "asyncband", + "benchmarks", + "examples", + "tests-integration", + "tests-integration-macros", + "xtask", +] resolver = "3" [workspace.package] @@ -39,11 +46,19 @@ async-broadcast = { version = "0.7.2" } async-channel = { version = "2.5.0" } cargo_metadata = { version = "0.23.1" } clap = { version = "4.6.5" } +# Compio 0.17+ uses Rust features newer than the workspace MSRV. +compio = { version = "0.16.1", default-features = false, features = [ + "dispatcher", + "runtime", + "time", +] } divan = { version = "0.1.21" } flume = { version = "0.12.0", default-features = false } +futures-lite = { version = "2.6.1" } pollster = { version = "1.0.1" } semver = { version = "1.0.28" } serde = { version = "1.0.229" } +smol = { version = "2.0.2" } tokio = { version = "1.53.1" } tokio-test = { version = "0.4.5" } ureq = { version = "3.3.0", default-features = false } diff --git a/tests-integration-macros/Cargo.toml b/tests-integration-macros/Cargo.toml new file mode 100644 index 0000000..3d4a84b --- /dev/null +++ b/tests-integration-macros/Cargo.toml @@ -0,0 +1,34 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "tests-integration-macros" +publish = false + +edition.workspace = true +rust-version.workspace = true + +[lib] +proc-macro = true + +[dependencies] +proc-macro2 = "1.0.106" +quote = "1.0.45" +syn = { version = "2.0.117", features = ["full"] } + +[lints] +workspace = true diff --git a/tests-integration-macros/src/lib.rs b/tests-integration-macros/src/lib.rs new file mode 100644 index 0000000..e4389cd --- /dev/null +++ b/tests-integration-macros/src/lib.rs @@ -0,0 +1,206 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::BTreeSet; + +use proc_macro::TokenStream; +use quote::format_ident; +use quote::quote; +use syn::Ident; +use syn::ItemFn; +use syn::Meta; +use syn::Token; +use syn::parse::Parser; +use syn::parse_macro_input; +use syn::punctuated::Punctuated; + +#[derive(Clone, Copy)] +struct RuntimeSpec { + name: &'static str, + type_name: &'static str, + capabilities: &'static [&'static str], +} + +const RUNTIMES: &[RuntimeSpec] = &[ + RuntimeSpec { + name: "tokio", + type_name: "Tokio", + capabilities: &["time"], + }, + RuntimeSpec { + name: "smol", + type_name: "Smol", + capabilities: &["time"], + }, + RuntimeSpec { + name: "compio", + type_name: "Compio", + capabilities: &["time"], + }, +]; + +#[derive(Default)] +struct RuntimeTestArgs { + only: Option>, + required: BTreeSet, +} + +/// Runs an async test on every configured runtime which has the requested +/// capabilities. +/// +/// Supported forms are `#[runtime_test]`, +/// `#[runtime_test(require(time))]`, and +/// `#[runtime_test(only(compio))]`. +/// +/// Each generated test body receives a `runtime` type alias whose operations +/// are statically bound to that wrapper's runtime. The annotated function does +/// not need a runtime generic parameter. +#[proc_macro_attribute] +pub fn runtime_test(attribute: TokenStream, item: TokenStream) -> TokenStream { + let args = match parse_args(attribute) { + Ok(args) => args, + Err(error) => return error.into_compile_error().into(), + }; + let function = parse_macro_input!(item as ItemFn); + + expand_runtime_test(args, function) + .unwrap_or_else(syn::Error::into_compile_error) + .into() +} + +fn parse_args(attribute: TokenStream) -> syn::Result { + let attributes = Punctuated::::parse_terminated.parse(attribute)?; + let mut args = RuntimeTestArgs::default(); + + for attribute in attributes { + let Meta::List(list) = attribute else { + return Err(syn::Error::new_spanned( + attribute, + "expected `require(...)` or `only(...)`", + )); + }; + let values = list.parse_args_with(Punctuated::::parse_terminated)?; + + if list.path.is_ident("require") { + for value in values { + let capability = value.to_string(); + if capability != "time" { + return Err(syn::Error::new_spanned(value, "unknown runtime capability")); + } + args.required.insert(capability); + } + } else if list.path.is_ident("only") { + if args.only.is_some() { + return Err(syn::Error::new_spanned(list, "duplicate `only(...)`")); + } + let mut runtimes = BTreeSet::new(); + for value in values { + let runtime = value.to_string(); + if !RUNTIMES.iter().any(|spec| spec.name == runtime) { + return Err(syn::Error::new_spanned(value, "unknown runtime")); + } + runtimes.insert(runtime); + } + args.only = Some(runtimes); + } else { + return Err(syn::Error::new_spanned( + list.path, + "expected `require(...)` or `only(...)`", + )); + } + } + + Ok(args) +} + +fn expand_runtime_test( + args: RuntimeTestArgs, + function: ItemFn, +) -> syn::Result { + if function.sig.asyncness.is_none() { + return Err(syn::Error::new_spanned( + function.sig.fn_token, + "`runtime_test` requires an async function", + )); + } + if !function.sig.inputs.is_empty() { + return Err(syn::Error::new_spanned( + &function.sig.inputs, + "`runtime_test` functions cannot take arguments", + )); + } + + if !function.sig.generics.params.is_empty() || function.sig.generics.where_clause.is_some() { + return Err(syn::Error::new_spanned( + &function.sig.generics, + "`runtime_test` functions cannot declare generics", + )); + } + + let selected = RUNTIMES + .iter() + .copied() + .filter(|runtime| { + args.only + .as_ref() + .is_none_or(|only| only.contains(runtime.name)) + }) + .filter(|runtime| { + args.required + .iter() + .all(|required| runtime.capabilities.contains(&required.as_str())) + }) + .collect::>(); + + if selected.is_empty() { + return Err(syn::Error::new_spanned( + &function.sig.ident, + "no configured runtime satisfies this test", + )); + } + + let attributes = &function.attrs; + let body = &function.block; + let function_name = &function.sig.ident; + let wrappers = selected.into_iter().map(|runtime| { + let runtime_type = format_ident!("{}", runtime.type_name); + let wrapper_name = format_ident!("{}_{}", function_name, runtime.name); + + quote! { + #(#attributes)* + #[test] + fn #wrapper_name() { + ::tests_integration::runtime::run::< + ::tests_integration::runtime::#runtime_type, + _, + _, + >(|| async move { + #[allow(non_camel_case_types)] + type runtime = ::tests_integration::runtime::RuntimeOps< + ::tests_integration::runtime::#runtime_type + >; + + #body + }); + } + } + }); + + Ok(quote! { + #(#wrappers)* + }) +} diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index b66fda0..2e74443 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -23,6 +23,10 @@ edition.workspace = true rust-version.workspace = true [dependencies] +compio = { workspace = true } +futures-lite = { workspace = true } +smol = { workspace = true } +tests-integration-macros = { path = "../tests-integration-macros" } tokio = { workspace = true, features = ["full"] } [dev-dependencies] diff --git a/tests-integration/src/lib.rs b/tests-integration/src/lib.rs index 95f6cdb..7ebb070 100644 --- a/tests-integration/src/lib.rs +++ b/tests-integration/src/lib.rs @@ -24,7 +24,14 @@ use std::task::Waker; use tokio::runtime::Runtime; +pub mod runtime; +pub use tests_integration_macros::runtime_test; + /// Polls a pinned future once with a no-op waker. +/// +/// This legacy helper is useful for synchronous, manually driven tests. New +/// cross-runtime async tests should use [`runtime::poll_once`], which polls with +/// the current runtime's waker. pub fn poll_once(future: Pin<&mut F>) -> Poll { future.poll(&mut Context::from_waker(Waker::noop())) } diff --git a/tests-integration/src/runtime.rs b/tests-integration/src/runtime.rs new file mode 100644 index 0000000..1db7f5a --- /dev/null +++ b/tests-integration/src/runtime.rs @@ -0,0 +1,520 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::future::Future; +use std::num::NonZeroUsize; +use std::panic::AssertUnwindSafe; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::OnceLock; +use std::task::Poll; +use std::time::Duration; + +use futures_lite::FutureExt; + +/// A normalized handle for a `Send` task. +pub type Task = Pin + Send + 'static>>; + +/// A normalized handle for a thread-local task. +pub type LocalTask = Pin + 'static>>; + +#[derive(Debug)] +pub enum JoinError { + Cancelled, + Panicked, +} + +/// The executor operations available to every shared test. +pub trait Runtime: 'static { + const NAME: &'static str; + + /// Drives a root future to completion on this runtime. + fn block_on(future: F) -> F::Output; + + /// Spawns a `Send` task on this runtime's worker facility. + /// + /// The scheduler is free to run different tasks on different threads. It + /// does not guarantee that an individual task will migrate between polls. + fn spawn(future: F) -> Task> + where + F: Future + Send + 'static, + T: Send + 'static; + + /// Spawns a `Send` task without retaining its join handle. + fn spawn_detached(future: F) + where + F: Future + Send + 'static; + + /// Spawns a task which is permanently bound to the current runtime thread. + fn spawn_local(future: F) -> LocalTask> + where + F: Future + 'static, + T: 'static; + + /// Spawns a thread-local task without retaining its join handle. + fn spawn_local_detached(future: F) + where + F: Future + 'static; + + /// Executes one test case. + /// + /// The factory is deliberately reusable. A normal runtime invokes it once; + /// a Loom adapter can invoke it once per explored schedule. + fn run_test(test: Test) + where + Self: Sized, + Test: Fn() -> Fut + Send + Sync + 'static, + Fut: Future + 'static, + { + Self::block_on(test()); + } +} + +/// Optional wall-clock capability for real runtimes. +/// +/// A model runtime does not have to implement this trait: Loom bounds an +/// exploration by branches, permutations, or duration rather than by racing a +/// future against a wall-clock timer. +pub trait TimeRuntime: Runtime { + fn sleep(duration: Duration) -> LocalTask<()>; + + fn timeout<'a, F, T>(duration: Duration, future: F) -> TaskRef<'a, Result> + where + F: Future + 'a, + T: 'a; +} + +pub type TaskRef<'a, T> = Pin + 'a>>; + +#[derive(Debug)] +pub struct Elapsed; + +const TEST_TIMEOUT: Duration = Duration::from_secs(10); + +/// Runs one fresh test future with the selected real runtime. +/// +/// Taking a factory rather than a future is intentional: a future Loom adapter +/// can invoke the factory once for each explored schedule. +pub fn run(test: Test) +where + R: Runtime, + Test: Fn() -> Fut + Send + Sync + 'static, + Fut: Future + 'static, +{ + R::run_test(test); +} + +fn run_with_timeout(test: Test) +where + R: Runtime + TimeRuntime, + Test: Fn() -> Fut, + Fut: Future, +{ + R::block_on(async { + if R::timeout(TEST_TIMEOUT, test()).await.is_err() { + panic!("{} test timed out after {TEST_TIMEOUT:?}", R::NAME); + } + }); +} + +/// Polls a future once using the current task's real waker. +/// +/// Returning `None` means the inner future returned `Poll::Pending`. The +/// wrapper itself still completes immediately, so this operation is executor +/// independent and also usable under a model runtime. +pub async fn poll_once(mut future: Pin<&mut F>) -> Option { + std::future::poll_fn(move |context| { + let output = match future.as_mut().poll(context) { + Poll::Ready(output) => Some(output), + Poll::Pending => None, + }; + Poll::Ready(output) + }) + .await +} + +/// Yields exactly one poll turn without depending on a runtime-specific API. +pub async fn yield_once() { + let mut yielded = false; + std::future::poll_fn(move |context| { + if yielded { + Poll::Ready(()) + } else { + yielded = true; + context.waker().wake_by_ref(); + Poll::Pending + } + }) + .await +} + +/// Statically bound runtime operations injected by [`runtime_test`]. +/// +/// The attribute macro aliases this type to `runtime` inside each generated +/// test, keeping the concrete runtime out of the test function's signature. +pub struct RuntimeOps(std::marker::PhantomData R>); + +impl RuntimeOps { + pub const NAME: &'static str = R::NAME; + + pub fn block_on(future: F) -> F::Output { + R::block_on(future) + } + + pub fn spawn(future: F) -> Task> + where + F: Future + Send + 'static, + T: Send + 'static, + { + R::spawn(future) + } + + pub fn spawn_detached(future: F) + where + F: Future + Send + 'static, + { + R::spawn_detached(future); + } + + pub fn spawn_local(future: F) -> LocalTask> + where + F: Future + 'static, + T: 'static, + { + R::spawn_local(future) + } + + pub fn spawn_local_detached(future: F) + where + F: Future + 'static, + { + R::spawn_local_detached(future); + } + + pub async fn poll_once(future: Pin<&mut F>) -> Option { + crate::runtime::poll_once(future).await + } + + pub async fn yield_once() { + crate::runtime::yield_once().await; + } +} + +impl RuntimeOps { + pub fn sleep(duration: Duration) -> LocalTask<()> { + R::sleep(duration) + } + + pub fn timeout<'a, F, T>(duration: Duration, future: F) -> TaskRef<'a, Result> + where + F: Future + 'a, + T: 'a, + { + R::timeout(duration, future) + } +} + +pub struct Tokio; + +impl Runtime for Tokio { + const NAME: &'static str = "tokio"; + + fn block_on(future: F) -> F::Output { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_time() + .build() + .expect("failed to build Tokio runtime"); + let local = tokio::task::LocalSet::new(); + runtime.block_on(local.run_until(future)) + } + + fn spawn(future: F) -> Task> + where + F: Future + Send + 'static, + T: Send + 'static, + { + let handle = tokio::spawn(AssertUnwindSafe(future).catch_unwind()); + Box::pin(async move { + match handle.await { + Ok(Ok(output)) => Ok(output), + Ok(Err(_)) => Err(JoinError::Panicked), + Err(_) => Err(JoinError::Cancelled), + } + }) + } + + fn spawn_detached(future: F) + where + F: Future + Send + 'static, + { + drop(tokio::spawn(AssertUnwindSafe(future).catch_unwind())); + } + + fn spawn_local(future: F) -> LocalTask> + where + F: Future + 'static, + T: 'static, + { + let handle = tokio::task::spawn_local(AssertUnwindSafe(future).catch_unwind()); + Box::pin(async move { + match handle.await { + Ok(Ok(output)) => Ok(output), + Ok(Err(_)) => Err(JoinError::Panicked), + Err(_) => Err(JoinError::Cancelled), + } + }) + } + + fn spawn_local_detached(future: F) + where + F: Future + 'static, + { + drop(tokio::task::spawn_local( + AssertUnwindSafe(future).catch_unwind(), + )); + } + + fn run_test(test: Test) + where + Test: Fn() -> Fut + Send + Sync + 'static, + Fut: Future + 'static, + { + run_with_timeout::(test); + } +} + +impl TimeRuntime for Tokio { + fn sleep(duration: Duration) -> LocalTask<()> { + Box::pin(tokio::time::sleep(duration)) + } + + fn timeout<'a, F, T>(duration: Duration, future: F) -> TaskRef<'a, Result> + where + F: Future + 'a, + T: 'a, + { + Box::pin(async move { + tokio::time::timeout(duration, future) + .await + .map_err(|_| Elapsed) + }) + } +} + +pub struct Smol; + +const SMOL_WORKER_THREADS: usize = 2; + +fn smol_executor() -> &'static smol::Executor<'static> { + static EXECUTOR: OnceLock>> = OnceLock::new(); + + EXECUTOR.get_or_init(|| { + let executor = Arc::new(smol::Executor::new()); + for worker in 0..SMOL_WORKER_THREADS { + let executor = executor.clone(); + std::thread::Builder::new() + .name(format!("asyncband-smol-{worker}")) + .spawn(move || smol::block_on(executor.run(std::future::pending::<()>()))) + .expect("failed to spawn Smol executor worker"); + } + executor + }) +} + +thread_local! { + static SMOL_LOCAL_EXECUTOR: smol::LocalExecutor<'static> = const { smol::LocalExecutor::new() }; +} + +impl Runtime for Smol { + const NAME: &'static str = "smol"; + + fn block_on(future: F) -> F::Output { + SMOL_LOCAL_EXECUTOR.with(|executor| smol::block_on(executor.run(future))) + } + + fn spawn(future: F) -> Task> + where + F: Future + Send + 'static, + T: Send + 'static, + { + let handle = smol_executor().spawn(AssertUnwindSafe(future).catch_unwind()); + Box::pin(async move { handle.await.map_err(|_| JoinError::Panicked) }) + } + + fn spawn_detached(future: F) + where + F: Future + Send + 'static, + { + smol_executor() + .spawn(AssertUnwindSafe(future).catch_unwind()) + .detach(); + } + + fn spawn_local(future: F) -> LocalTask> + where + F: Future + 'static, + T: 'static, + { + let handle = SMOL_LOCAL_EXECUTOR + .with(|executor| executor.spawn(AssertUnwindSafe(future).catch_unwind())); + Box::pin(async move { handle.await.map_err(|_| JoinError::Panicked) }) + } + + fn spawn_local_detached(future: F) + where + F: Future + 'static, + { + SMOL_LOCAL_EXECUTOR.with(|executor| { + executor + .spawn(AssertUnwindSafe(future).catch_unwind()) + .detach(); + }); + } + + fn run_test(test: Test) + where + Test: Fn() -> Fut + Send + Sync + 'static, + Fut: Future + 'static, + { + run_with_timeout::(test); + } +} + +impl TimeRuntime for Smol { + fn sleep(duration: Duration) -> LocalTask<()> { + Box::pin(async move { + smol::Timer::after(duration).await; + }) + } + + fn timeout<'a, F, T>(duration: Duration, future: F) -> TaskRef<'a, Result> + where + F: Future + 'a, + T: 'a, + { + Box::pin(futures_lite::future::race( + async move { Ok(future.await) }, + async move { + smol::Timer::after(duration).await; + Err(Elapsed) + }, + )) + } +} + +pub struct Compio; + +const COMPIO_WORKER_THREADS: usize = 2; + +fn compio_dispatcher() -> &'static compio::dispatcher::Dispatcher { + static DISPATCHER: OnceLock = OnceLock::new(); + + DISPATCHER.get_or_init(|| { + compio::dispatcher::Dispatcher::builder() + .worker_threads( + NonZeroUsize::new(COMPIO_WORKER_THREADS) + .expect("Compio worker count must be non-zero"), + ) + .build() + .expect("failed to build Compio dispatcher") + }) +} + +impl Runtime for Compio { + const NAME: &'static str = "compio"; + + fn block_on(future: F) -> F::Output { + compio::runtime::Runtime::new() + .expect("failed to build Compio runtime") + .block_on(future) + } + + fn spawn(future: F) -> Task> + where + F: Future + Send + 'static, + T: Send + 'static, + { + let receiver = compio_dispatcher() + .dispatch(move || AssertUnwindSafe(future).catch_unwind()) + .unwrap_or_else(|_| panic!("failed to dispatch Compio task")); + Box::pin(async move { + match receiver.await { + Ok(Ok(output)) => Ok(output), + Ok(Err(_)) => Err(JoinError::Panicked), + Err(_) => Err(JoinError::Cancelled), + } + }) + } + + fn spawn_detached(future: F) + where + F: Future + Send + 'static, + { + let receiver = compio_dispatcher() + .dispatch(move || AssertUnwindSafe(future).catch_unwind()) + .unwrap_or_else(|_| panic!("failed to dispatch Compio task")); + drop(receiver); + } + + fn spawn_local(future: F) -> LocalTask> + where + F: Future + 'static, + T: 'static, + { + let handle = compio::runtime::spawn(AssertUnwindSafe(future).catch_unwind()); + Box::pin(async move { + match handle.await { + Ok(Ok(output)) => Ok(output), + Ok(Err(_)) | Err(_) => Err(JoinError::Panicked), + } + }) + } + + fn spawn_local_detached(future: F) + where + F: Future + 'static, + { + compio::runtime::spawn(AssertUnwindSafe(future).catch_unwind()).detach(); + } + + fn run_test(test: Test) + where + Test: Fn() -> Fut + Send + Sync + 'static, + Fut: Future + 'static, + { + run_with_timeout::(test); + } +} + +impl TimeRuntime for Compio { + fn sleep(duration: Duration) -> LocalTask<()> { + Box::pin(compio::runtime::time::sleep(duration)) + } + + fn timeout<'a, F, T>(duration: Duration, future: F) -> TaskRef<'a, Result> + where + F: Future + 'a, + T: 'a, + { + Box::pin(async move { + compio::runtime::time::timeout(duration, future) + .await + .map_err(|_| Elapsed) + }) + } +} diff --git a/tests-integration/tests/mutex_test.rs b/tests-integration/tests/mutex_test.rs index 7f7e27f..a58284b 100644 --- a/tests-integration/tests/mutex_test.rs +++ b/tests-integration/tests/mutex_test.rs @@ -18,6 +18,7 @@ use std::sync::Arc; use asyncband::mutex::*; +use tests_integration::runtime_test; #[test] fn test_try_lock_never_blocks() { @@ -46,7 +47,7 @@ fn test_get_mut_provides_exclusive_access() { assert_eq!(inner, 100); } -#[tokio::test] +#[runtime_test] async fn test_guard_map_preserves_lock() { let data = (99i32, vec![1, 2, 3]); let mutex = Mutex::new(data); @@ -64,7 +65,7 @@ async fn test_guard_map_preserves_lock() { assert_eq!(guard.0, 100); } -#[tokio::test] +#[runtime_test] async fn test_mapped_guard_holds_lock() { // Test that MappedMutexGuard properly holds the lock even after the original guard is moved let mutex = Arc::new(Mutex::new((10, 20))); @@ -87,7 +88,7 @@ async fn test_mapped_guard_holds_lock() { ); } -#[tokio::test] +#[runtime_test] async fn test_owned_mapped_guard_holds_lock() { // Test that mapped owned guard properly holds the lock let mutex = Arc::new(Mutex::new((30, 40))); @@ -111,7 +112,7 @@ async fn test_owned_mapped_guard_holds_lock() { ); } -#[tokio::test] +#[runtime_test] async fn test_guard_filter_map_failure() { let data: Vec = vec![]; let mutex = Mutex::new(data); @@ -128,7 +129,7 @@ async fn test_guard_filter_map_failure() { } } -#[tokio::test] +#[runtime_test] async fn test_owned_guard_filter_map_failure() { let data: Vec = vec![]; let mutex = Arc::new(Mutex::new(data)); @@ -145,7 +146,7 @@ async fn test_owned_guard_filter_map_failure() { } } -#[tokio::test] +#[runtime_test] async fn test_multiple_map_operations() { // Test multiple consecutive map operations let data = vec![vec![1, 2], vec![3, 4]]; @@ -164,7 +165,7 @@ async fn test_multiple_map_operations() { assert_eq!(guard[1][0], 3); } -#[tokio::test] +#[runtime_test] async fn test_stress() { let mutex = Arc::new(Mutex::new(0)); let mut handles = Vec::new(); @@ -172,11 +173,11 @@ async fn test_stress() { // Create many concurrent tasks for i in 0..1000 { let mutex = mutex.clone(); - handles.push(tokio::spawn(async move { + handles.push(runtime::spawn(async move { let mut guard = mutex.lock().await; *guard += 1; if i % 10 == 0 { - tokio::task::yield_now().await; + runtime::yield_once().await; } })); } @@ -189,7 +190,7 @@ async fn test_stress() { assert_eq!(final_value, 1000); } -#[tokio::test] +#[runtime_test] async fn test_guard_prevents_concurrent_access() { // Test that holding a guard prevents other tasks from acquiring the lock let mutex = Arc::new(Mutex::new(0)); @@ -202,12 +203,12 @@ async fn test_guard_prevents_concurrent_access() { "Lock should be held by the first guard" ); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let _guard2 = mutex_clone.lock().await; 123 }); - tokio::task::yield_now().await; + runtime::yield_once().await; assert!( mutex.try_lock().is_none(), @@ -242,13 +243,13 @@ fn test_lock_panic_safety() { assert!(mutex.try_lock().is_some()); } -#[tokio::test] +#[runtime_test] async fn test_async_lock_panic_safety() { // Test panic safety with async locks let mutex = Arc::new(Mutex::new(0)); let mutex_clone = mutex.clone(); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let _guard = mutex_clone.lock().await; panic!("async test panic"); }); @@ -260,12 +261,12 @@ async fn test_async_lock_panic_safety() { assert!(guard.is_some()); } -#[tokio::test] +#[runtime_test] async fn test_owned_guard_panic_safety() { let mutex = Arc::new(Mutex::new(0)); let mutex_clone = mutex.clone(); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let _guard = mutex_clone.clone().lock_owned().await; panic!("owned guard panic"); }); @@ -277,13 +278,13 @@ async fn test_owned_guard_panic_safety() { assert!(guard.is_some()); } -#[tokio::test] +#[runtime_test] async fn test_mapped_guard_panic_safety() { // Test panic safety with mapped guards let mutex = Arc::new(Mutex::new((66, vec![1, 2, 3]))); let mutex_clone = mutex.clone(); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let guard = mutex_clone.lock().await; let _mapped = MutexGuard::map(guard, |data| &mut data.0); panic!("mapped guard panic"); @@ -295,7 +296,7 @@ async fn test_mapped_guard_panic_safety() { assert!(guard.is_some()); } -#[tokio::test] +#[runtime_test] async fn test_memory_ordering_correctness() { // Test that mutex provides proper memory ordering guarantees // When one task modifies data under mutex protection, @@ -303,7 +304,7 @@ async fn test_memory_ordering_correctness() { let mutex = Arc::new(Mutex::new(vec![1, 2, 3])); let mutex_clone = mutex.clone(); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let mut guard = mutex_clone.lock().await; guard.push(4); guard[0] = 100; @@ -319,13 +320,13 @@ async fn test_memory_ordering_correctness() { // to subsequent lock acquisitions } -#[tokio::test] +#[runtime_test] async fn test_mutex_zst() { // Test that Mutex works correctly with Zero-Sized Types let mutex = Arc::new(Mutex::new(())); let mutex_clone = mutex.clone(); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let guard = mutex_clone.lock().await; *guard; }); @@ -341,7 +342,7 @@ async fn test_mutex_zst() { *guard; } -#[tokio::test] +#[runtime_test] async fn test_mapped_mutex_guard_send() { // Test that MappedMutexGuard can be sent across await points #[derive(Debug)] @@ -356,11 +357,11 @@ async fn test_mapped_mutex_guard_send() { })); let mutex_clone = mutex.clone(); - let handle = tokio::spawn(async move { + let handle = runtime::spawn(async move { let guard = mutex_clone.lock().await; let mapped_guard = MutexGuard::map(guard, |data| &mut data.field1); - tokio::task::yield_now().await; + runtime::yield_once().await; *mapped_guard }); diff --git a/tests-integration/tests/runtime_test_macro_test.rs b/tests-integration/tests/runtime_test_macro_test.rs new file mode 100644 index 0000000..062cdf5 --- /dev/null +++ b/tests-integration/tests/runtime_test_macro_test.rs @@ -0,0 +1,48 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use tests_integration::runtime_test; + +#[runtime_test(only(compio))] +async fn runtime_specific_test() { + runtime::spawn(async {}).await.unwrap(); +} + +#[runtime_test] +async fn spawn_uses_a_worker_thread() { + let caller = std::thread::current().id(); + let worker = runtime::spawn(async { std::thread::current().id() }) + .await + .unwrap(); + + assert_ne!(caller, worker); +} + +#[runtime_test] +async fn spawn_local_stays_on_the_current_thread() { + let caller = std::thread::current().id(); + let local_value = std::rc::Rc::new(()); + let worker = runtime::spawn_local(async move { + runtime::yield_once().await; + assert_eq!(std::rc::Rc::strong_count(&local_value), 1); + std::thread::current().id() + }) + .await + .unwrap(); + + assert_eq!(caller, worker); +} diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index 4adbef7..12509cc 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -25,6 +25,7 @@ use std::task::Waker; use std::vec::Vec; use asyncband::semaphore::Semaphore; +use tests_integration::runtime_test; #[test] fn no_permits() { @@ -45,23 +46,23 @@ fn try_acquire() { assert!(p3.is_some()); } -#[tokio::test] +#[runtime_test] async fn acquire() { let sem = Arc::new(Semaphore::new(1)); let p1 = sem.try_acquire(1).unwrap(); let sem_clone = sem.clone(); - let j = tokio::spawn(async move { + let j = runtime::spawn(async move { let _p2 = sem_clone.acquire(1).await; }); drop(p1); j.await.unwrap(); } -#[tokio::test] +#[runtime_test] async fn add_permits() { let sem = Arc::new(Semaphore::new(0)); let sem_clone = sem.clone(); - let j = tokio::spawn(async move { + let j = runtime::spawn(async move { let _p2 = sem_clone.acquire(1).await; }); sem.release(1); @@ -81,15 +82,17 @@ fn forget() { assert!(sem.try_acquire(1).is_none()); } -#[tokio::test] +#[runtime_test] async fn stress_test() { let sem = Arc::new(Semaphore::new(5)); let mut join_handles = Vec::new(); for i in 0..100 { let sem_clone = sem.clone(); - join_handles.push(tokio::spawn(async move { + join_handles.push(runtime::spawn(async move { let _p = sem_clone.acquire(1).await; - tokio::time::sleep(std::time::Duration::from_millis(100 - i)).await; + for _ in i..100 { + runtime::yield_once().await; + } })); } for j in join_handles { @@ -222,6 +225,32 @@ fn cancellation_restores_permits_before_dropping_waker() { assert_eq!(Arc::strong_count(&semaphore), 1); } +#[allow(unused_mut)] +#[runtime_test] +async fn acquire_then_reduce_permits_exact() { + let s = Arc::new(Semaphore::new(5)); + s.reduce_permits(3); + assert_eq!(s.available_permits(), 2); + + let (acquired_tx, mut acquired_rx) = asyncband::oneshot::channel(); + + let s_clone = s.clone(); + runtime::spawn_detached(async move { + let permit = s_clone.acquire(3).await; + drop(permit); + acquired_tx.send(()).unwrap(); + }); + assert!(acquired_rx.try_recv().is_err()); + + s.reduce_permits(2); + s.release(2); + assert!(acquired_rx.try_recv().is_err()); + + s.release(1); + acquired_rx.await.unwrap(); + assert_eq!(s.available_permits(), 3); +} + #[test] fn reduce_permits_takes_priority_over_pending_acquires() { let s = Semaphore::new(0);