From 6005e61b7538b4bfa18f66eabdfa8fc912c96f2b Mon Sep 17 00:00:00 2001 From: Paul Brackin Date: Fri, 19 Jun 2026 11:33:08 -0700 Subject: [PATCH 1/3] feat: trim async completion per-op overhead (Phases 1+3) and add dedicated drainer thread for wake-ups; bump pyo3 0.29; uvloop on by default with AEROSPIKE_NO_UVLOOP opt-out --- Cargo.lock | 383 +++++------------- Cargo.toml | 6 +- README.md | 50 ++- benchmarks/benchmark.py | 8 +- python/aerospike_async/__init__.py | 37 +- .../_aerospike_async_native.pyi | 3 - src/cdt.rs | 12 +- src/cluster.rs | 2 +- src/completion.rs | 196 ++++----- src/expressions.rs | 2 +- src/filter.rs | 8 +- src/lib.rs | 118 +++--- src/operations.rs | 10 +- src/policies.rs | 23 +- src/waker.rs | 170 ++++++++ 15 files changed, 535 insertions(+), 493 deletions(-) create mode 100644 src/waker.rs diff --git a/Cargo.lock b/Cargo.lock index f6f1def..fab71e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,7 +11,7 @@ checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" [[package]] name = "aerospike-core" version = "3.0.0-alpha.1" -source = "git+https://github.com/aerospike/aerospike-client-rust?branch=v3#6b7e6849a169f270e30640ec7277f724bf5f5675" +source = "git+https://github.com/aerospike/aerospike-client-rust?branch=v3#088e21d40058e9c049ce49c5d519a04959155797" dependencies = [ "aerospike-rt", "async-channel", @@ -36,7 +36,7 @@ dependencies = [ [[package]] name = "aerospike-rt" version = "3.0.0-alpha.1" -source = "git+https://github.com/aerospike/aerospike-client-rust?branch=v3#6b7e6849a169f270e30640ec7277f724bf5f5675" +source = "git+https://github.com/aerospike/aerospike-client-rust?branch=v3#088e21d40058e9c049ce49c5d519a04959155797" dependencies = [ "tokio", ] @@ -123,9 +123,9 @@ dependencies = [ [[package]] name = "autocfg" -version = "1.5.0" +version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-lc-rs" @@ -168,12 +168,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "bitflags" -version = "2.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" - [[package]] name = "block-buffer" version = "0.9.0" @@ -185,9 +179,9 @@ dependencies = [ [[package]] name = "block-buffer" -version = "0.12.0" +version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdd35008169921d80bc60d3d0ab416eecb028c4cd653352907921d95084790be" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" dependencies = [ "hybrid-array", ] @@ -205,9 +199,9 @@ dependencies = [ [[package]] name = "bumpalo" -version = "3.20.2" +version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" [[package]] name = "byteorder" @@ -217,15 +211,15 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" [[package]] name = "bytes" -version = "1.11.1" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593" [[package]] name = "cc" -version = "1.2.62" +version = "1.2.64" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" +checksum = "dad887fd958be91b5098c0248def011f4523ab786cd411be668777e55063501f" dependencies = [ "find-msvc-tools", "jobserver", @@ -252,9 +246,9 @@ dependencies = [ [[package]] name = "chrono" -version = "0.4.44" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ "iana-time-zone", "js-sys", @@ -343,9 +337,9 @@ checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" [[package]] name = "crypto-common" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77727bb15fa921304124b128af125e7e3b968275d1b108b379190264f4423710" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" dependencies = [ "hybrid-array", ] @@ -365,9 +359,6 @@ name = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" -dependencies = [ - "powerfmt", -] [[package]] name = "digest" @@ -384,7 +375,7 @@ version = "0.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ - "block-buffer 0.12.0", + "block-buffer 0.12.1", "const-oid", "crypto-common", ] @@ -397,9 +388,9 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "either" -version = "1.15.0" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" [[package]] name = "equivalent" @@ -444,12 +435,6 @@ dependencies = [ "miniz_oxide", ] -[[package]] -name = "foldhash" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" - [[package]] name = "fs_extra" version = "1.3.0" @@ -588,25 +573,14 @@ dependencies = [ [[package]] name = "getrandom" -version = "0.4.2" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", "rand_core 0.10.1", - "wasip2", - "wasip3", -] - -[[package]] -name = "hashbrown" -version = "0.15.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" -dependencies = [ - "foldhash", ] [[package]] @@ -679,12 +653,6 @@ dependencies = [ "cc", ] -[[package]] -name = "id-arena" -version = "2.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" - [[package]] name = "indexmap" version = "2.14.0" @@ -692,9 +660,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown 0.17.1", - "serde", - "serde_core", + "hashbrown", ] [[package]] @@ -754,13 +720,12 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.98" +version = "0.3.102" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67df7112613f8bfd9150013a0314e196f4800d3201ae742489d999db2f979f08" +checksum = "03d04c30968dffe80775bd4d7fb676131cd04a1fb46d2686dbffbaec2d9dfd31" dependencies = [ "cfg-if", "futures-util", - "once_cell", "wasm-bindgen", ] @@ -776,12 +741,6 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" -[[package]] -name = "leb128fmt" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" - [[package]] name = "libc" version = "0.2.186" @@ -790,18 +749,18 @@ checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" [[package]] name = "libmimalloc-sys" -version = "0.1.47" +version = "0.1.49" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d1eacfa31c33ec25e873c136ba5669f00f9866d0688bea7be4d3f7e43067df6" +checksum = "6a45a52f43e1c16f667ccfe4dd8c85b7f7c204fd5e3bf46c5b0db9a5c3c0b8e9" dependencies = [ "cc", ] [[package]] name = "log" -version = "0.4.29" +version = "0.4.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" [[package]] name = "maplit" @@ -832,15 +791,15 @@ dependencies = [ [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" [[package]] name = "mimalloc" -version = "0.1.50" +version = "0.1.52" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b3627c4272df786b9260cabaa46aec1d59c93ede723d4c3ef646c503816b0640" +checksum = "2d4139bb28d14ad1facf21d5eb8825051b326e172d216b39f6d31df53cc97862" dependencies = [ "libmimalloc-sys", ] @@ -857,9 +816,9 @@ dependencies = [ [[package]] name = "mio" -version = "1.2.0" +version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", "wasi", @@ -902,9 +861,9 @@ dependencies = [ [[package]] name = "num-conv" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" [[package]] name = "num-integer" @@ -926,9 +885,9 @@ dependencies = [ [[package]] name = "numpy" -version = "0.28.0" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "778da78c64ddc928ebf5ad9df5edf0789410ff3bdbf3619aed51cd789a6af1e2" +checksum = "6a5b15d63a5ff39e378daed0e1340d3a5964703ea9712eb09a0dc66fade996f4" dependencies = [ "libc", "ndarray", @@ -1050,16 +1009,6 @@ dependencies = [ "zerocopy", ] -[[package]] -name = "prettyplease" -version = "0.2.37" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" -dependencies = [ - "proc-macro2", - "syn", -] - [[package]] name = "proc-macro2" version = "1.0.106" @@ -1086,9 +1035,9 @@ dependencies = [ [[package]] name = "pyo3" -version = "0.28.3" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91fd8e38a3b50ed1167fb981cd6fd60147e091784c427b8f7183a7ee32c31c12" +checksum = "cd274650b21d4bfc26a0a47587962c1edb425f69287324355cd040c3ea66071c" dependencies = [ "libc", "once_cell", @@ -1100,9 +1049,9 @@ dependencies = [ [[package]] name = "pyo3-async-runtimes" -version = "0.28.0" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e7364a95bf00e8377bbf9b0f09d7ff9715a29d8fcf93b47d1a967363b973178" +checksum = "b3ef68daa7316a3fac65e5e18b2203f010346de1c1c53456811a2624673ab046" dependencies = [ "futures-channel", "futures-util", @@ -1114,18 +1063,18 @@ dependencies = [ [[package]] name = "pyo3-build-config" -version = "0.28.3" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e368e7ddfdeb98c9bca7f8383be1648fd84ab466bf2bc015e94008db6d35611e" +checksum = "c5e2a7d2f0d013342f295c048ad19237add5154a55b1c5a254c0ec93d4109078" dependencies = [ "target-lexicon", ] [[package]] name = "pyo3-ffi" -version = "0.28.3" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f29e10af80b1f7ccaf7f69eace800a03ecd13e883acfacc1e5d0988605f651e" +checksum = "ca85c467da1bbc8d866eea5deff9cf29ea5f7785054a17da36e65bda9c05845b" dependencies = [ "libc", "pyo3-build-config", @@ -1133,9 +1082,9 @@ dependencies = [ [[package]] name = "pyo3-log" -version = "0.13.3" +version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26c2ec80932c5c3b2d4fbc578c9b56b2d4502098587edb8bef5b6bfcad43682e" +checksum = "f64083bd3a16a353d9d62335808e8e13d0552d2a2b83fdb084496192dcfa9fcd" dependencies = [ "arc-swap", "log", @@ -1144,9 +1093,9 @@ dependencies = [ [[package]] name = "pyo3-macros" -version = "0.28.3" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df6e520eff47c45997d2fc7dd8214b25dd1310918bbb2642156ef66a67f29813" +checksum = "9ac53762fd065daa3194dd09337a38bd793a188100fd1a9304c4ab312d901771" dependencies = [ "proc-macro2", "pyo3-macros-backend", @@ -1156,22 +1105,21 @@ dependencies = [ [[package]] name = "pyo3-macros-backend" -version = "0.28.3" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4cdc218d835738f81c2338f822078af45b4afdf8b2e33cbb5916f108b813acb" +checksum = "4ca3a1557399783172dc5bf39cfca835157732532cba56b71d2292161e53b362" dependencies = [ "heck", "proc-macro2", - "pyo3-build-config", "quote", "syn", ] [[package]] name = "pyo3-stub-gen" -version = "0.22.0" +version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de78bf1239d5284bf9be8efc69636bb3639d1270f074bf090c6cc735fd85d4bc" +checksum = "a267ea4da9a831f534def7f1d8e14f581d5640d3e5af6527072724f216f4dbbf" dependencies = [ "anyhow", "chrono", @@ -1195,9 +1143,9 @@ dependencies = [ [[package]] name = "pyo3-stub-gen-derive" -version = "0.22.2" +version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2f901834d55c74f36be3353994062e3c3b3d47fda25572f5535e99434612ac8" +checksum = "6573423a5e8cc43ec7565eccccbc2dd35e1fc9032d9ca404afab875429eb8cf0" dependencies = [ "heck", "indexmap", @@ -1246,7 +1194,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" dependencies = [ "chacha20", - "getrandom 0.4.2", + "getrandom 0.4.3", "rand_core 0.10.1", ] @@ -1283,9 +1231,9 @@ checksum = "60a357793950651c4ed0f3f52338f53b2f809f32d83a07f72909fa13e4c6c1e3" [[package]] name = "regex" -version = "1.12.3" +version = "1.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276" +checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba" dependencies = [ "aho-corasick", "memchr", @@ -1306,9 +1254,9 @@ dependencies = [ [[package]] name = "regex-syntax" -version = "0.8.10" +version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] name = "rhexdump" @@ -1459,12 +1407,6 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" -[[package]] -name = "semver" -version = "1.0.28" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" - [[package]] name = "serde" version = "1.0.228" @@ -1497,9 +1439,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.149" +version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ "itoa", "memchr", @@ -1545,9 +1487,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "simd-adler32" @@ -1569,9 +1511,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "socket2" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", "windows-sys 0.61.2", @@ -1591,9 +1533,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "syn" -version = "2.0.117" +version = "2.0.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" dependencies = [ "proc-macro2", "quote", @@ -1628,9 +1570,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.47" +version = "0.3.49" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" +checksum = "711a53c2d47bbd818258c498c8dbfe186a2526c631495cfe7e078567f86b8469" dependencies = [ "deranged", "num-conv", @@ -1641,9 +1583,9 @@ dependencies = [ [[package]] name = "time-core" -version = "0.1.8" +version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" [[package]] name = "tiny-keccak" @@ -1719,9 +1661,9 @@ checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db" [[package]] name = "typenum" -version = "1.20.0" +version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" [[package]] name = "unic-char-property" @@ -1787,12 +1729,6 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" -[[package]] -name = "unicode-xid" -version = "0.2.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" - [[package]] name = "unicode_names2" version = "1.3.0" @@ -1841,27 +1777,18 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" [[package]] name = "wasip2" -version = "1.0.3+wasi-0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" -dependencies = [ - "wit-bindgen 0.57.1", -] - -[[package]] -name = "wasip3" -version = "0.4.0+wasi-0.3.0-rc-2026-01-06" +version = "1.0.4+wasi-0.2.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" dependencies = [ - "wit-bindgen 0.51.0", + "wit-bindgen", ] [[package]] name = "wasm-bindgen" -version = "0.2.121" +version = "0.2.125" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49ace1d07c165b0864824eee619580c4689389afa9dc9ed3a4c75040d82e6790" +checksum = "8ddb3f79143bced6de84270411622a2699cee572fc0875aeaf1e7867cf9fca1a" dependencies = [ "cfg-if", "once_cell", @@ -1872,9 +1799,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.121" +version = "0.2.125" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e68e6f4afd367a562002c05637acb8578ff2dea1943df76afb9e83d177c8578" +checksum = "4e21a184b13fb19e157296e2c46056aec9092264fab83e4ba59e68c61b323c3d" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -1882,9 +1809,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.121" +version = "0.2.125" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d95a9ec35c64b2a7cb35d3fead40c4238d0940c86d107136999567a4703259f2" +checksum = "fecefd9c35bd935a20fc3fc344b5f29138961e4f47fb03297d88f2587afb5ebd" dependencies = [ "bumpalo", "proc-macro2", @@ -1895,52 +1822,18 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.121" +version = "0.2.125" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4e0100b01e9f0d03189a92b96772a1fb998639d981193d7dbab487302513441" +checksum = "23939e44bb9a5d7576fa2b563dc2e136628f1224e88a8deed09e04858b77871f" dependencies = [ "unicode-ident", ] -[[package]] -name = "wasm-encoder" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319" -dependencies = [ - "leb128fmt", - "wasmparser", -] - -[[package]] -name = "wasm-metadata" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" -dependencies = [ - "anyhow", - "indexmap", - "wasm-encoder", - "wasmparser", -] - -[[package]] -name = "wasmparser" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" -dependencies = [ - "bitflags", - "hashbrown 0.15.5", - "indexmap", - "semver", -] - [[package]] name = "webpki-roots" -version = "1.0.7" +version = "1.0.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52f5ee44c96cf55f1b349600768e3ece3a8f26010c05265ab73f945bb1a2eb9d" +checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf" dependencies = [ "rustls-pki-types", ] @@ -2092,114 +1985,26 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0592e1c9d151f854e6fd382574c3a0855250e1d9b2f99d9281c6e6391af352f1" -[[package]] -name = "wit-bindgen" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" -dependencies = [ - "wit-bindgen-rust-macro", -] - [[package]] name = "wit-bindgen" version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" -[[package]] -name = "wit-bindgen-core" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" -dependencies = [ - "anyhow", - "heck", - "wit-parser", -] - -[[package]] -name = "wit-bindgen-rust" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" -dependencies = [ - "anyhow", - "heck", - "indexmap", - "prettyplease", - "syn", - "wasm-metadata", - "wit-bindgen-core", - "wit-component", -] - -[[package]] -name = "wit-bindgen-rust-macro" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a" -dependencies = [ - "anyhow", - "prettyplease", - "proc-macro2", - "quote", - "syn", - "wit-bindgen-core", - "wit-bindgen-rust", -] - -[[package]] -name = "wit-component" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" -dependencies = [ - "anyhow", - "bitflags", - "indexmap", - "log", - "serde", - "serde_derive", - "serde_json", - "wasm-encoder", - "wasm-metadata", - "wasmparser", - "wit-parser", -] - -[[package]] -name = "wit-parser" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" -dependencies = [ - "anyhow", - "id-arena", - "indexmap", - "log", - "semver", - "serde", - "serde_derive", - "serde_json", - "unicode-xid", - "wasmparser", -] - [[package]] name = "zerocopy" -version = "0.8.48" +version = "0.8.52" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +checksum = "ce1022995ff5ff5d841ad7d994facc23098cd40152f2c1d11cd607c6f530653f" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.48" +version = "0.8.52" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930" dependencies = [ "proc-macro2", "quote", @@ -2208,9 +2013,9 @@ dependencies = [ [[package]] name = "zeroize" -version = "1.8.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" [[package]] name = "zmij" diff --git a/Cargo.toml b/Cargo.toml index 299bcd2..7950d54 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,13 +9,13 @@ name = "_aerospike_async_native" crate-type = ["cdylib", "rlib"] [dependencies] -pyo3 = { version = "0.28", default-features = false } -pyo3-async-runtimes = { version = "0.28", features = ["tokio-runtime"] } +pyo3 = { version = "0.29", default-features = false } +pyo3-async-runtimes = { version = "0.29", features = ["tokio-runtime"] } tokio = "1.40" aerospike-core = { git = "https://github.com/aerospike/aerospike-client-rust", branch = "v3", package = "aerospike-core", features = ["rt-tokio"] } hex = "0.4" ordered-float = { version = "3", default-features = false } -pyo3-stub-gen = "=0.22.0" +pyo3-stub-gen = "0.23" log = "0.4" pyo3-log = "0.13" futures = "0.3" diff --git a/README.md b/README.md index 0182f78..94ccefa 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,11 @@ # Aerospike Python Async Client -Async I/O Python bindings for the Aerospike Rust client core. Built with -[PyO3](https://pyo3.rs/); ships pre-built wheels for Linux (x86_64, aarch64), -macOS (x86_64, arm64), and Windows (x86_64) on Python 3.10–3.14, including -free-threaded builds (`cp313t` / `cp314t`). +Ultra-high performance Python bindings for the Aerospike Rust client core +— async and blocking surfaces in one client, with first-class free-threaded +Python support for parallel-thread throughput well past what GIL-bound +clients can sustain. Built with [PyO3](https://pyo3.rs/); ships pre-built +wheels for Linux (x86_64, aarch64), macOS (x86_64, arm64), and Windows +(x86_64) on Python 3.10–3.14, including the free-threaded build (`cp314t`). > **Status:** Public preview (alpha). Not yet production-ready; feedback welcome > via [GitHub Issues](https://github.com/aerospike/aerospike-client-python-async/issues). @@ -37,7 +39,7 @@ pip install aerospike-async==0.6.0a1 # latest on PyPI as of this writing Pre-built wheels are published for every supported platform/Python combination on regular CPython (3.10 – 3.14, ABI tags `cp310`–`cp314`) **and on the -free-threaded builds** (`cp313t` / `cp314t`), so no Rust toolchain is required +free-threaded build** (`cp314t`), so no Rust toolchain is required for ordinary use. If pip resolves to an sdist on your platform, see [Building from source](#building-from-source) below. @@ -80,6 +82,44 @@ async def main(): asyncio.run(main()) ``` +## Blocking (sync) surface + +Every async method has a `*_blocking` counterpart that blocks the calling +thread instead of returning an awaitable. Use these when you need sync +semantics without an event loop — particularly useful on free-threaded +Python, where multiple OS threads can drive blocking calls in parallel +without the GIL serializing them. + +```python +from aerospike_async import ClientPolicy, Key, new_client_blocking + +client = new_client_blocking(ClientPolicy(), "localhost:3000") + +key = Key("test", "demo", "user1") + +# Write a record +client.put_blocking(key, {"name": "Alice", "age": 28}) + +# Read it back +record = client.get_blocking(key) +print(record.bins) # {'name': 'Alice', 'age': 28} + +# Delete the record +client.delete_blocking(key) + +client.close_blocking() +``` + +Naming follows `_blocking` — `get_blocking`, `put_blocking`, +`operate_blocking`, `batch_read_blocking`, `query_execute_blocking`, etc. + +Clients created with `new_client_blocking()` are blocking-only; async +methods on them raise. If you want a single client that exposes both +sync and async ergonomics with a friendlier high-level API (sessions, +behaviors, AEL filters), use the +[Aerospike Python SDK](https://pypi.org/project/aerospike-sdk/) instead +— its `SyncClient` is built directly on top of this blocking surface. + ## TLS configuration The client supports TLS for secure connections and PKI (certificate-based) diff --git a/benchmarks/benchmark.py b/benchmarks/benchmark.py index 91b9fa3..8a3c735 100644 --- a/benchmarks/benchmark.py +++ b/benchmarks/benchmark.py @@ -242,10 +242,10 @@ async def _worker( t0 = time.perf_counter() try: if is_read: - await client.batch_read(None, None, keys, None) + await client.batch_read(keys, None) else: bins_list = [_random_bins_dict(fields) for _ in range(bsz)] - await client.batch_write(None, None, keys, bins_list) + await client.batch_write(keys, bins_list) except Exception: dt = (time.perf_counter() - t0) * 1000.0 # Count one op per batch call to match PSDK's accounting @@ -262,10 +262,10 @@ async def _worker( t0 = time.perf_counter() try: if is_read: - await client.get(rp, key, None) + await client.get(key, None, policy=rp) else: ops = _random_bins(fields) - await client.operate(wp, key, ops) + await client.operate(key, ops, policy=wp) except Exception: dt = (time.perf_counter() - t0) * 1000.0 stats.record(is_read, dt, True) diff --git a/python/aerospike_async/__init__.py b/python/aerospike_async/__init__.py index d452d56..599f998 100644 --- a/python/aerospike_async/__init__.py +++ b/python/aerospike_async/__init__.py @@ -13,25 +13,48 @@ # License for the specific language governing permissions and limitations under # the License. -# Install uvloop as the default event loop policy when available. -# uvloop is a drop-in replacement for asyncio's event loop that provides -# significantly better throughput and latency for async I/O workloads. -# Set AEROSPIKE_NO_UVLOOP=1 to disable. +# Module-level names leak into the package namespace (this is the package +# root), so anything imported here is `_`-prefixed and `del`-ed after use. import os as _os +# Install uvloop as the default event loop policy. +# uvloop is a drop-in replacement for asyncio's event loop that provides +# significantly better throughput and latency for async I/O workloads. +# +# Free-threaded Python (3.14t) note: uvloop 0.22.x has a documented +# libuv race on `loop._ready_len` (MagicStack/uvloop issues #720, #721) +# that triggers when many threads concurrently call +# `loop.call_soon_threadsafe()`. PAC's drainer thread funnels ALL +# wake-ups through ONE persistent thread, eliminating the multi- +# threaded access pattern the race needs. Empirically stable across +# 20+ minutes of stress (z=128 single-loop + AsyncPool 8×64, 241M ops, +# zero stalls). uvloop PR #721 is the proper upstream fix; once that +# releases this comment can be dropped. +# uvloop is required everywhere except Windows (see pyproject.toml's +# `sys_platform != 'win32'` marker on the dependency). On Windows we +# fall back to asyncio's default selector loop. +# +# uvloop is installed by default. Set AEROSPIKE_NO_UVLOOP=1 to opt out +# and keep asyncio's default selector loop — a safety valve for the +# rare environment that hits a uvloop bug, with no need to uninstall +# the dependency. +# +# uvloop/warnings are imported lazily inside the guard on purpose: skip +# loading the uvloop C extension entirely when opted out, and tolerate +# its absence on Windows (the import is genuinely conditional). if not _os.environ.get("AEROSPIKE_NO_UVLOOP"): try: import uvloop as _uvloop import warnings as _warnings - # uvloop.install() uses set_event_loop_policy which is deprecated - # in Python 3.14+. Suppress the warning until uvloop provides a + # uvloop.install() uses set_event_loop_policy which is deprecated in + # Python 3.14+. Suppress the warning until uvloop provides a # non-deprecated alternative. with _warnings.catch_warnings(): _warnings.filterwarnings( "ignore", category=DeprecationWarning, module=r"uvloop" ) _uvloop.install() - del _warnings + del _warnings, _uvloop except ImportError: pass diff --git a/python/aerospike_async/_aerospike_async_native.pyi b/python/aerospike_async/_aerospike_async_native.pyi index eca7aca..eef40fa 100644 --- a/python/aerospike_async/_aerospike_async_native.pyi +++ b/python/aerospike_async/_aerospike_async_native.pyi @@ -1449,9 +1449,6 @@ class ClientPolicy: @property def fail_if_not_connected(self) -> builtins.bool: r""" - Size of the thread pool used in scan and query commands. These commands are often sent to - multiple server nodes in parallel threads. A thread pool improves performance because - threads do not have to be created/destroyed for each command. Throw exception if host connection fails during addHost(). """ @fail_if_not_connected.setter diff --git a/src/cdt.rs b/src/cdt.rs index 6b2b9e6..d78936c 100644 --- a/src/cdt.rs +++ b/src/cdt.rs @@ -988,7 +988,7 @@ use crate::record::PythonValue; #[staticmethod] pub fn from_base64(b64: &str) -> PyResult> { let core_ctxs = aerospike_core::operations::cdt_context::ctx_from_base64(b64) - .map_err(|e| crate::errors::RustClientError(e))?; + .map_err(crate::errors::RustClientError)?; Ok(core_ctxs.into_iter().map(|c| CTX { ctx: c }).collect()) } @@ -999,7 +999,7 @@ use crate::record::PythonValue; #[staticmethod] pub fn from_bytes(bytes: Vec) -> PyResult> { let core_ctxs = aerospike_core::operations::cdt_context::ctx_from_bytes(&bytes) - .map_err(|e| crate::errors::RustClientError(e))?; + .map_err(crate::errors::RustClientError)?; Ok(core_ctxs.into_iter().map(|c| CTX { ctx: c }).collect()) } } @@ -1223,7 +1223,7 @@ use crate::record::PythonValue; ) -> PyResult { let f = match &write_flags { None => 0u8, - Some(obj) => bit_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => bit_policy_flags_from_py(obj.bind(py))?, }; Ok(BitPolicy { _as: aerospike_core::operations::bitwise::BitPolicy::new(f), @@ -1297,7 +1297,7 @@ use crate::record::PythonValue; let order = order.unwrap_or(ListOrderType::Unordered); let f = match &write_flags { None => 0u8, - Some(obj) => list_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => list_policy_flags_from_py(obj.bind(py))?, }; Ok(ListPolicy { _as: aerospike_core::operations::lists::ListPolicy { @@ -1489,7 +1489,7 @@ use crate::record::PythonValue; ) -> PyResult { let f: i64 = match &write_flags { None => 0, - Some(obj) => hll_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => hll_policy_flags_from_py(obj.bind(py))?, }; Ok(HLLPolicy { _as: aerospike_core::operations::hll::HLLPolicy { flags: f }, @@ -1639,7 +1639,7 @@ use crate::record::PythonValue; let core_order: aerospike_core::operations::maps::MapOrder = (&order).into(); let f = match &flags { None => 0u8, - Some(obj) => map_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => map_policy_flags_from_py(obj.bind(py))?, }; let _as = if persist_index == Some(true) { aerospike_core::operations::maps::MapPolicy::new_with_flags_and_persisted_index( diff --git a/src/cluster.rs b/src/cluster.rs index fbc9438..a01b9d4 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -268,7 +268,7 @@ use crate::policies::AdminPolicy; ) -> PyResult> { let node = std::sync::Arc::clone(&self._as); let admin_policy = - policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + policy.map(|p| p._as).unwrap_or_default(); pyo3_asyncio::future_into_py(py, async move { let response = node diff --git a/src/completion.rs b/src/completion.rs index 213ce77..b47a181 100644 --- a/src/completion.rs +++ b/src/completion.rs @@ -32,7 +32,7 @@ use std::future::Future; use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use futures::FutureExt; use pyo3::prelude::*; @@ -41,6 +41,7 @@ use pyo3::IntoPyObjectExt; use pyo3_async_runtimes::tokio as pyo3_asyncio; use crate::client_runtime::ClientRuntime; +use crate::waker; type Converter = Box) -> PyResult> + Send>; @@ -50,18 +51,23 @@ struct PendingResult { } /// State shared between a bridge and its drainer. -struct CompletionInner { +pub(crate) struct CompletionInner { queue: Mutex>, drain_scheduled: AtomicBool, /// Latched true on first `call_soon_threadsafe` failure (loop closed). /// Once set, subsequent enqueues fast-path-fail the pending Python future /// rather than queuing it — prevents indefinite hangs on shutdown races. - closed: AtomicBool, + pub(crate) closed: AtomicBool, /// Event loop captured at bridge construction. Every operation routed /// through this bridge must run on this loop; mismatches fail at the /// `batched_future_into_py` entry rather than silently corrupting state /// (`set_result` on loop B's future from loop A's thread). - owning_loop: Py, + pub(crate) owning_loop: Py, + /// Drainer pyobject for this bridge. Populated by `CompletionBridge::new` + /// after the Arc exists (the Drainer holds an `Arc` + /// back-reference, so it can't be constructed before Inner). Reads on the + /// hot path go through the waker thread. + pub(crate) drainer: OnceLock>, /// Per-Client runtime: when Some, spawn each future on this dedicated /// Tokio runtime instead of the shared global multi-thread one. Set /// from `ClientPolicy.per_client_runtime_workers`. Held by Arc so the @@ -69,24 +75,12 @@ struct CompletionInner { client_rt: Option>, } -/// Owned by each `Client`. -/// -/// `Py` requires Python-thread attachment to bump its refcount, so `Clone` -/// goes through `Python::attach`. The hot path (`batched_future_into_py`) -/// already holds a `Python<'_>` token and uses `clone_ref(py)` instead to skip -/// the redundant attach check. +/// Owned by each `Client`. Thin wrapper around `Arc` so the +/// hot path's per-op clone is a pure `Arc::clone` — no GIL acquisition, no +/// `Py<>` refcount bump. +#[derive(Clone)] pub(crate) struct CompletionBridge { inner: Arc, - drainer: Py, -} - -impl Clone for CompletionBridge { - fn clone(&self) -> Self { - Python::attach(|py| CompletionBridge { - inner: self.inner.clone(), - drainer: self.drainer.clone_ref(py), - }) - } } /// Python-callable invoked on the event loop thread via `call_soon_threadsafe`. @@ -104,70 +98,45 @@ impl CompletionDrainer { } } -impl CompletionBridge { - pub(crate) fn new( - py: Python<'_>, - owning_loop: Py, - client_rt: Option>, - ) -> PyResult { - let inner = Arc::new(CompletionInner { - queue: Mutex::new(Vec::new()), - drain_scheduled: AtomicBool::new(false), - closed: AtomicBool::new(false), - owning_loop, - client_rt, - }); - let drainer = Py::new(py, CompletionDrainer { inner: inner.clone() })?; - Ok(CompletionBridge { inner, drainer }) - } - - pub(crate) fn clone_ref(&self, py: Python<'_>) -> Self { - CompletionBridge { - inner: self.inner.clone(), - drainer: self.drainer.clone_ref(py), - } - } - - fn enqueue(&self, pr: PendingResult) { +impl CompletionInner { + /// Tokio-worker-side entry: queue a result for batch delivery via the + /// waker thread. Pure Rust — no `Python::attach`, no GIL acquisition. + fn enqueue(inner: &Arc, pr: PendingResult) { // Fast-path: bridge already known dead. Fail the future immediately // so the caller doesn't hang. - if self.inner.closed.load(Ordering::Acquire) { + if inner.closed.load(Ordering::Acquire) { fail_pr(pr, "event loop is closed"); return; } - self.inner.queue.lock().unwrap().push(pr); - if !self.inner.drain_scheduled.swap(true, Ordering::AcqRel) { - // First completion in this batch — schedule a drain callback. - // This is the only GIL acquisition from the Tokio side per batch. - let scheduled = Python::attach(|py| { - let el = self.inner.owning_loop.bind(py); - el.call_method1( - pyo3::intern!(py, "call_soon_threadsafe"), - (self.drainer.bind(py),), - ) - .map(|_| ()) - }); - if scheduled.is_err() { - // Loop closed during/after enqueue. Latch — only this thread - // ever takes this branch per batch (atomic swap above), so - // there is no fail-storm even under heavy concurrent enqueue. - self.inner.closed.store(true, Ordering::Release); - self.fail_all_pending(); - return; + inner.queue.lock().unwrap().push(pr); + if !inner.drain_scheduled.swap(true, Ordering::AcqRel) { + // First completion in this batch — hand a wake token to the + // single persistent waker thread. No GIL needed here; the waker + // thread does the `call_soon_threadsafe(drainer)` call with its + // own persistent PyThreadState. + if waker::send_wake(inner.clone()).is_err() { + // Channel overflowed (or waker not initialized). Reset the + // gate so the NEXT enqueue can re-arm — otherwise the queue + // stalls with items pushed but no drain ever fired. + // + // Race-benign: a duplicate wake from the resetting enqueue + // is harmless (drain only fires when the gate is won, and + // drain itself idempotently resets the gate). + inner.drain_scheduled.store(false, Ordering::Release); } } - // Defensive: another worker's call_soon_threadsafe may have failed - // and latched `closed` after our push but before our swap-check. - // fail_all_pending is idempotent (Mutex+take — second caller drains - // empty), so the rare double-call here is harmless. - if self.inner.closed.load(Ordering::Acquire) { - self.fail_all_pending(); + // Defensive: the waker thread may have failed call_soon_threadsafe + // (loop closed) and latched `closed` after our push but before our + // swap-check. fail_all_pending is idempotent (Mutex+take — second + // caller drains empty), so the rare double-call here is harmless. + if inner.closed.load(Ordering::Acquire) { + inner.fail_all_pending(); } } - fn fail_all_pending(&self) { + pub(crate) fn fail_all_pending(&self) { let pending: Vec = - std::mem::take(&mut *self.inner.queue.lock().unwrap()); + std::mem::take(&mut *self.queue.lock().unwrap()); if pending.is_empty() { return; } @@ -183,6 +152,29 @@ impl CompletionBridge { } } +impl CompletionBridge { + pub(crate) fn new( + py: Python<'_>, + owning_loop: Py, + client_rt: Option>, + ) -> PyResult { + // Ensure the waker thread exists before any completion can fire. + waker::ensure_waker(py)?; + + let inner = Arc::new(CompletionInner { + queue: Mutex::new(Vec::new()), + drain_scheduled: AtomicBool::new(false), + closed: AtomicBool::new(false), + owning_loop, + drainer: OnceLock::new(), + client_rt, + }); + let drainer = Py::new(py, CompletionDrainer { inner: inner.clone() })?; + let _ = inner.drainer.set(drainer); + Ok(CompletionBridge { inner }) + } +} + fn fail_pr(pr: PendingResult, msg: &'static str) { Python::attach(|py| { let future = pr.future.bind(py); @@ -236,39 +228,49 @@ where F: Future> + Send + 'static, T: for<'a> IntoPyObject<'a> + Send + 'static, { - let locals = pyo3_asyncio::get_current_locals(py)?; - let event_loop = locals.event_loop(py); - if !event_loop.is(bridge.inner.owning_loop.bind(py)) { - return Err(pyo3::exceptions::PyRuntimeError::new_err( - "Aerospike Client used from a different event loop than the one that created it. \ - One Client must be paired with one event loop — construct a separate Client per \ - loop (AsyncPool does this automatically).", - )); + // Cross-loop misuse fails downstream at set_result time with an + // InvalidStateError anyway; the up-front check is a diagnostic, not a + // safety invariant, so reserve it for debug builds. Phase 1 of the + // CompletionBridge optimization plan (~0.5-1 µs per op saved). + #[cfg(debug_assertions)] + { + let locals = pyo3_asyncio::get_current_locals(py)?; + let event_loop = locals.event_loop(py); + if !event_loop.is(bridge.inner.owning_loop.bind(py)) { + return Err(pyo3::exceptions::PyRuntimeError::new_err( + "Aerospike Client used from a different event loop than the one that created it. \ + One Client must be paired with one event loop — construct a separate Client per \ + loop (AsyncPool does this automatically).", + )); + } } + let event_loop = bridge.inner.owning_loop.bind(py); let py_fut = event_loop.call_method0(pyo3::intern!(py, "create_future"))?; let future_ref: Py = py_fut.clone().unbind(); - let bridge = bridge.clone_ref(py); - // Capture the per-Client runtime handle before moving `bridge` into - // the async closure. Arc clone is cheap and keeps the runtime alive - // for the lifetime of the spawn. - let client_rt = bridge.inner.client_rt.clone(); + // Pure Arc::clone — no GIL acquisition, no Py<> refcount bump. The + // spawned Tokio task only needs `Arc` to enqueue; + // wake-up is delegated to the persistent waker thread. + let inner = bridge.inner.clone(); + // Capture the per-Client runtime handle before moving `inner` into the + // async closure. Arc clone is cheap and keeps the runtime alive for the + // lifetime of the spawn. + let client_rt = inner.client_rt.clone(); let spawn_fut = async move { let outcome = AssertUnwindSafe(fut).catch_unwind().await; // Backstop for the shutdown-race panic: when a user-code exception // escapes `asyncio.run()` while Tokio tasks are still airborne, the - // interpreter begins finalizing. The `Err(panic)` arm below - // constructs a `PyRuntimeError` to deliver, and `bridge.enqueue` - // calls `call_soon_threadsafe` — both go through `Python::attach`, - // which asserts on `Py_IsInitialized` and panics during - // finalization. Catching here keeps the panic from leaving a dead - // Tokio worker; the asyncio future is unreachable in that scenario - // anyway, so silently dropping `future_ref` and `bridge` is fine. - // The stderr noise from the panic hook firing before this catches - // is suppressed separately by the `take_hook`/`set_hook` filter - // installed in `lib.rs` module init. + // interpreter begins finalizing. Even with the waker thread doing + // the GIL work, `fail_pr` (cold path) still goes through + // `Python::attach`, which asserts on `Py_IsInitialized` and panics + // during finalization. Catching here keeps the panic from leaving + // a dead Tokio worker; the asyncio future is unreachable in that + // scenario anyway, so silently dropping `future_ref` and `inner` is + // fine. The stderr noise from the panic hook firing before this + // catches is suppressed separately by the `take_hook`/`set_hook` + // filter installed in `lib.rs` module init. let _: std::result::Result<(), _> = catch_unwind(AssertUnwindSafe(move || { let pr = match outcome { Ok(Ok(val)) => PendingResult { @@ -293,7 +295,7 @@ where } } }; - bridge.enqueue(pr); + CompletionInner::enqueue(&inner, pr); })); }; diff --git a/src/expressions.rs b/src/expressions.rs index 97e711c..7a18cc1 100644 --- a/src/expressions.rs +++ b/src/expressions.rs @@ -691,7 +691,7 @@ use crate::record::PythonValue; FilterExpression { _as: aerospike_core::expressions::list_val( val.into_iter() - .map(|v| aerospike_core::Value::from(v)) + .map(aerospike_core::Value::from) .collect(), ), } diff --git a/src/filter.rs b/src/filter.rs index 245964f..1c0283f 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -102,7 +102,7 @@ use crate::record::{Key, PythonValue, Record}; #[getter] pub fn get_digest(&self) -> Option { - self._as.digest.map(|d| hex::encode(d)) + self._as.digest.map(hex::encode) } #[setter] @@ -139,12 +139,12 @@ use crate::record::{Key, PythonValue, Record}; "id" => Ok(self.get_id().into_pyobject(py).unwrap().into_any().into()), "bval" => match self.get_bval() { Some(v) => Ok(v.into_pyobject(py).unwrap().into_any().into()), - None => Ok(py.None().into()), + None => Ok(py.None()), }, "retry" => Ok(PyBool::new(py, self.get_retry()).into_bound_py_any(py).unwrap().into()), "digest" => match self.get_digest() { Some(v) => Ok(v.into_pyobject(py).unwrap().into_any().into()), - None => Ok(py.None().into()), + None => Ok(py.None()), }, _ => Err(PyKeyError::new_err(format!("Unknown key: '{}'. Valid keys: 'id', 'bval', 'retry', 'digest'", key_str))), } @@ -264,7 +264,7 @@ use crate::record::{Key, PythonValue, Record}; #[getter] pub fn get_digest(&self) -> Option { - self._as.digest.map(|d| hex::encode(d)) + self._as.digest.map(hex::encode) } #[setter] diff --git a/src/lib.rs b/src/lib.rs index c61a5b8..95ddfdb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,6 +14,9 @@ // the License. #![deny(warnings)] +#![allow(clippy::new_without_default)] +#![allow(clippy::should_implement_trait)] +#![allow(clippy::too_many_arguments)] extern crate pyo3; #[global_allocator] @@ -41,6 +44,7 @@ use aerospike_core::errors::Error; mod blocking; mod completion; +mod waker; mod enums; mod errors; mod runtime; @@ -1377,7 +1381,7 @@ use crate::operations::{ ) -> PyResult { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let lang: aerospike_core::UDFLang = language.into(); let raw = run_blocking(py, async move { client.register_udf(&admin_policy, &udf_body, &server_path, lang).await @@ -1398,7 +1402,7 @@ use crate::operations::{ ) -> PyResult { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let lang: aerospike_core::UDFLang = language.into(); let raw = run_blocking(py, async move { client.register_udf_from_file(&admin_policy, &client_path, &server_path, lang).await @@ -1417,7 +1421,7 @@ use crate::operations::{ ) -> PyResult { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let raw = run_blocking(py, async move { client.remove_udf(&admin_policy, &server_path).await .map_err(|e| PyErr::from(RustClientError(e))) @@ -1437,7 +1441,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let before_nanos = before_nanos.unwrap_or_default(); run_blocking(py, async move { client.truncate(&admin_policy, &namespace, &set_name, before_nanos).await @@ -1461,7 +1465,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let cit = (&cit.unwrap_or(CollectionIndexType::Default)).into(); let index_type = (&index_type).into(); let ctx_core = ctx.map(|c| ctx_to_vec(&c)); @@ -1487,7 +1491,7 @@ use crate::operations::{ ) -> PyResult { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let raw = run_blocking(py, async move { client.drop_index(&admin_policy, &namespace, &set_name, &index_name).await .map_err(|e| PyErr::from(RustClientError(e))) @@ -1553,7 +1557,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); client.create_user(&admin_policy, &user, &password, &roles).await @@ -1572,7 +1576,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); client.create_pki_user(&admin_policy, &user, &roles).await @@ -1590,7 +1594,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { client.drop_user(&admin_policy, &user).await .map_err(|e| PyErr::from(RustClientError(e))) @@ -1608,7 +1612,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { client.change_password(&admin_policy, &user, &password).await .map_err(|e| PyErr::from(RustClientError(e))) @@ -1626,7 +1630,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); client.grant_roles(&admin_policy, &user, &roles).await @@ -1645,7 +1649,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); client.revoke_roles(&admin_policy, &user, &roles).await @@ -1663,7 +1667,7 @@ use crate::operations::{ ) -> PyResult> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let raw = run_blocking(py, async move { let user_ref = user.as_deref(); client.query_users(&admin_policy, user_ref).await @@ -1682,7 +1686,7 @@ use crate::operations::{ ) -> PyResult> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let raw = run_blocking(py, async move { let role_ref = role.as_deref(); client.query_roles(&admin_policy, role_ref).await @@ -1705,7 +1709,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let core_privileges: Vec = privileges.iter().map(|r| r._as.clone()).collect(); run_blocking(py, async move { @@ -1728,7 +1732,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { client.drop_role(&admin_policy, &role_name).await .map_err(|e| PyErr::from(RustClientError(e))) @@ -1746,7 +1750,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let core_privileges: Vec = privileges.iter().map(|p| p._as.clone()).collect(); run_blocking(py, async move { @@ -1766,7 +1770,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let core_privileges: Vec = privileges.iter().map(|p| p._as.clone()).collect(); run_blocking(py, async move { @@ -1786,7 +1790,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { let allowlist: Vec<&str> = allowlist.iter().map(|al| &**al).collect(); client.set_allowlist(&admin_policy, &role_name, &allowlist).await @@ -1806,7 +1810,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); run_blocking(py, async move { client.set_quotas(&admin_policy, &role_name, read_quota, write_quota).await .map_err(|e| PyErr::from(RustClientError(e))) @@ -1828,7 +1832,7 @@ use crate::operations::{ ) -> PyResult { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let expr = expression._as.clone(); let cit = (&cit.unwrap_or(CollectionIndexType::Default)).into(); let index_type = (&index_type).into(); @@ -1854,7 +1858,7 @@ use crate::operations::{ ) -> PyResult<()> { let client = self._as.clone(); let admin_policy = policy.map(|p| p._as) - .unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + .unwrap_or_default(); let expr = filter_expression.clone(); run_blocking(py, async move { client.set_xdr_filter( @@ -1956,7 +1960,7 @@ use crate::operations::{ for (key, bins) in rust_keys.into_iter().zip(bins_vecs.iter()) { let ops: Vec = bins .iter() - .map(|bin| operations::put(bin)) + .map(operations::put) .collect(); batch_ops.push(BatchOperation::write(&write_policy, key, ops)); } @@ -2650,7 +2654,7 @@ use crate::operations::{ for (key, bins) in rust_keys.into_iter().zip(bins_vecs.iter()) { let ops: Vec = bins .iter() - .map(|bin| operations::put(bin)) + .map(operations::put) .collect(); batch_ops.push(BatchOperation::write(&write_policy, key, ops)); } @@ -3021,7 +3025,7 @@ use crate::operations::{ // route each per-yield future through the same loop + per-Client // runtime as the Client itself. let bridge = self.require_bridge()?; - let stream_bridge = bridge.clone_ref(py); + let stream_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let batch_ops = build_batch_operations(&extracted)?; @@ -3070,7 +3074,7 @@ use crate::operations::{ }); completion::batched_future_into_py(self.require_bridge()?, py, async move { - let rust_args_ref = rust_args.as_ref().map(|a| a.as_slice()); + let rust_args_ref = rust_args.as_deref(); let result = client .execute_udf(&policy, &key, &server_path, &function_name, rust_args_ref) .await @@ -3113,12 +3117,12 @@ use crate::operations::{ let (core_ops, _) = convert_scalar_ops_to_core(&rust_ops).map_err(|e| { PyValueError::new_err(format!( "query_operate supports scalar and expression operations (put, add, delete, touch, append, prepend, ExpOperation.write). List/map/bit/HLL operations are not supported for background query. {}", - e.to_string() + e )) })?; let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let task = client .query_operate(&policy, core_statement, &core_ops) @@ -3159,7 +3163,7 @@ use crate::operations::{ core_statement.set_aggregate_function(&package_name, &function_name, rust_args.as_deref()); let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let args_ref = rust_args.as_deref(); let task = client @@ -3191,11 +3195,11 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); let lang: aerospike_core::UDFLang = language.into(); let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let task = client .register_udf(&admin_policy, &udf_body, &server_path, lang) @@ -3226,11 +3230,11 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); let lang: aerospike_core::UDFLang = language.into(); let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let task = client .register_udf_from_file(&admin_policy, &client_path, &server_path, lang) @@ -3257,10 +3261,10 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let task = client .remove_udf(&admin_policy, &server_path) @@ -3341,7 +3345,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); let before_nanos = before_nanos.unwrap_or_default(); @@ -3372,7 +3376,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); let cit = (&cit.unwrap_or(CollectionIndexType::Default)).into(); let index_type = (&index_type).into(); @@ -3409,10 +3413,10 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let task = client .drop_index(&admin_policy, &namespace, &set_name, &index_name) @@ -3438,13 +3442,13 @@ use crate::operations::{ ) -> PyResult> { let client = self._as.clone(); let admin_policy = - policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + policy.map(|p| p._as).unwrap_or_default(); let expr = expression._as.clone(); let cit = (&cit.unwrap_or(CollectionIndexType::Default)).into(); let index_type = (&index_type).into(); let bridge = self.require_bridge()?; - let task_bridge = bridge.clone_ref(py); + let task_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let task = client .create_index_using_expression( @@ -3478,7 +3482,7 @@ use crate::operations::{ let client = self._as.clone(); let stmt = statement.clone()._as; let bridge = self.require_bridge()?; - let recordset_bridge = bridge.clone_ref(py); + let recordset_bridge = bridge.clone(); completion::batched_future_into_py(bridge, py, async move { let res = client @@ -3507,7 +3511,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); @@ -3532,7 +3536,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); @@ -3550,7 +3554,7 @@ use crate::operations::{ #[pyo3(signature = (user, *, policy = None))] pub fn drop_user<'a>(&self, user: String, policy: Option, py: Python<'a>) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { client @@ -3573,7 +3577,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { client @@ -3596,7 +3600,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); @@ -3620,7 +3624,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let roles: Vec<&str> = roles.iter().map(|r| &**r).collect(); @@ -3644,7 +3648,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let user = user.as_deref(); @@ -3669,7 +3673,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let role: Option<&str> = role.as_deref(); @@ -3699,7 +3703,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let allowlist: Vec<&str> = allowlist.iter().map(|al| &**al).collect(); @@ -3724,7 +3728,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { client @@ -3747,7 +3751,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let privileges: Vec = @@ -3772,7 +3776,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let privileges: Vec = @@ -3798,7 +3802,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { let allowlist: Vec<&str> = allowlist.iter().map(|al| &**al).collect(); @@ -3826,7 +3830,7 @@ use crate::operations::{ py: Python<'a>, ) -> PyResult> { let client = self._as.clone(); - let admin_policy = policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + let admin_policy = policy.map(|p| p._as).unwrap_or_default(); completion::batched_future_into_py(self.require_bridge()?, py, async move { client @@ -4033,7 +4037,7 @@ use crate::operations::{ ) -> PyResult> { let client = self._as.clone(); let admin_policy = - policy.map(|p| p._as).unwrap_or_else(|| aerospike_core::AdminPolicy::default()); + policy.map(|p| p._as).unwrap_or_default(); let expr = filter_expression.clone(); completion::batched_future_into_py(self.require_bridge()?, py, async move { diff --git a/src/operations.rs b/src/operations.rs index a14b309..b118e7d 100644 --- a/src/operations.rs +++ b/src/operations.rs @@ -1325,7 +1325,7 @@ use crate::record::PythonValue; ) -> PyResult { let f = match &flags { None => 0i64, - Some(obj) => crate::cdt::hll_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => crate::cdt::hll_policy_flags_from_py(obj.bind(py))?, }; Ok(HllOperation { op: OperationType::HllInit(bin_name, index_bit_count, min_hash_bit_count, f), @@ -1351,7 +1351,7 @@ use crate::record::PythonValue; ) -> PyResult { let f = match &flags { None => 0i64, - Some(obj) => crate::cdt::hll_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => crate::cdt::hll_policy_flags_from_py(obj.bind(py))?, }; Ok(HllOperation { op: OperationType::HllAdd(bin_name, values, index_bit_count, min_hash_bit_count, f), @@ -1452,7 +1452,7 @@ use crate::record::PythonValue; ) -> PyResult { let f = match &flags { None => 0i64, - Some(obj) => crate::cdt::hll_policy_flags_from_py(&obj.bind(py))?, + Some(obj) => crate::cdt::hll_policy_flags_from_py(obj.bind(py))?, }; Ok(HllOperation { op: OperationType::HllSetUnion(bin_name, hll_list, f), @@ -1503,7 +1503,7 @@ use crate::record::PythonValue; ) -> PyResult { let f = match &flags { None => 0i64, - Some(obj) => crate::expressions::exp_read_flags_from_py(&obj.bind(py))?, + Some(obj) => crate::expressions::exp_read_flags_from_py(obj.bind(py))?, }; Ok(ExpOperation { op: OperationType::ExpRead(name, exp, f), @@ -1533,7 +1533,7 @@ use crate::record::PythonValue; ) -> PyResult { let f = match &flags { None => 0i64, - Some(obj) => crate::expressions::exp_write_flags_from_py(&obj.bind(py))?, + Some(obj) => crate::expressions::exp_write_flags_from_py(obj.bind(py))?, }; Ok(ExpOperation { op: OperationType::ExpWrite(bin_name, exp, f), diff --git a/src/policies.rs b/src/policies.rs index 6a77447..4bb7af4 100644 --- a/src/policies.rs +++ b/src/policies.rs @@ -1071,7 +1071,7 @@ use crate::TlsConfig; #[getter] pub fn get_result_code(&self) -> Option { - self._as.result_code.map(|rc| ResultCode(rc)) + self._as.result_code.map(ResultCode) } #[getter] @@ -1751,7 +1751,6 @@ use crate::TlsConfig; impl ClientPolicy { #[new] fn new() -> PyResult { - let mut cp = aerospike_core::ClientPolicy::default(); // Tuned for the primary async use case: a single Tokio runtime // (or per-Client runtime in AsyncPool) serializes pool access // through one or two workers, so contention is naturally low @@ -1759,9 +1758,11 @@ use crate::TlsConfig; // from many caller threads (e.g. PSDK's SyncClient) should // override this on the policy before construction; 8 is a good // value for ~32-thread sync workloads. - cp.conn_pools_per_node = 4; let res = ClientPolicy { - _as: cp, + _as: aerospike_core::ClientPolicy { + conn_pools_per_node: 4, + ..Default::default() + }, per_client_runtime_workers: None, }; @@ -1891,13 +1892,13 @@ use crate::TlsConfig; self._as.auth_mode = aerospike_core::AuthMode::None; } AuthMode::Internal => { - let user = user.unwrap_or_else(|| "".to_string()); - let password = password.unwrap_or_else(|| "".to_string()); + let user = user.unwrap_or_default(); + let password = password.unwrap_or_default(); self._as.auth_mode = aerospike_core::AuthMode::Internal(user, password); } AuthMode::External => { - let user = user.unwrap_or_else(|| "".to_string()); - let password = password.unwrap_or_else(|| "".to_string()); + let user = user.unwrap_or_default(); + let password = password.unwrap_or_default(); self._as.auth_mode = aerospike_core::AuthMode::External(user, password); } AuthMode::PKI => { @@ -2017,9 +2018,9 @@ use crate::TlsConfig; self._as.rack_ids = value.map(|v| v.into_iter().collect()); } - /// Size of the thread pool used in scan and query commands. These commands are often sent to - /// multiple server nodes in parallel threads. A thread pool improves performance because - /// threads do not have to be created/destroyed for each command. + // Size of the thread pool used in scan and query commands. These commands are often sent to + // multiple server nodes in parallel threads. A thread pool improves performance because + // threads do not have to be created/destroyed for each command. // thread_pool_size field doesn't exist in TLS branch // #[getter] // pub fn get_thread_pool_size(&self) -> usize { diff --git a/src/waker.rs b/src/waker.rs new file mode 100644 index 0000000..c90dc05 --- /dev/null +++ b/src/waker.rs @@ -0,0 +1,170 @@ +// Copyright 2025-2026 Aerospike, Inc. +// +// Portions may be licensed to Aerospike, Inc. under one or more contributor +// license agreements WHICH ARE COMPATIBLE WITH THE APACHE LICENSE, VERSION 2.0. +// +// Licensed 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. + +//! Single persistent waker thread for `call_soon_threadsafe` dispatch. +//! +//! Tokio workers must not call `Python::attach` per completion: each +//! `PyGILState_Release` runs `PyThreadState_Clear`, which mimalloc's +//! abandoned-segment collector picks up at ~9.66% process CPU under +//! sustained async load. Pinning each Tokio worker's thread-state +//! (Branch D) hangs at `z=32` even with the uvloop FT gate active +//! (zero-byte stdout wedge; confirmed 2026-06-18) — the failure mode +//! is intrinsic to keeping many concurrent persistent thread-states +//! on Tokio workers, not a uvloop side effect. +//! +//! Instead: Tokio workers push pure-Rust wake tokens (`Arc`) +//! over an mpsc channel. ONE persistent daemon thread, spawned via Python's +//! `threading.Thread` so CPython owns its thread-state lifecycle, consumes +//! tokens and issues `call_soon_threadsafe(drainer)` to the correct asyncio +//! loop. Because exactly one extra thread is persistently attached (not +//! many Tokio workers), the failure condition Branch D hit does not apply. +//! +//! Per-batch semantics are unchanged: `CompletionInner::drain_scheduled` +//! still gates wake tokens to one per batch, and `drain()` still runs on the +//! loop thread. + +use std::sync::atomic::Ordering; +use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; +use std::sync::{Arc, Mutex, OnceLock}; + +use pyo3::prelude::*; +use pyo3::types::PyDict; + +use crate::completion::CompletionInner; + +/// Bounded so a runaway producer can't grow the channel without limit. Each +/// token gates to one-per-batch, so even at 1M ops/sec the channel sees at +/// most a few thousand tokens/sec — 65k is comfortable headroom. +const WAKER_CHANNEL_CAPACITY: usize = 65_536; + +static SENDER: OnceLock>> = OnceLock::new(); +// The Receiver is moved into the spawned Python thread on first start. +// Stashed in a global Mutex> so the Python-side recv loop can +// `take()` it; `OnceLock` alone wouldn't allow the take(). +static PENDING_RX: OnceLock>>>> = OnceLock::new(); + +/// Tokio-worker-side: push a wake token. Pure Rust, no GIL. +/// +/// Returns `Err` on channel overflow OR if the waker hasn't been initialized +/// yet (extremely unlikely — `ensure_waker` runs from `CompletionBridge::new`). +/// Callers MUST reset `inner.drain_scheduled` to false on `Err`, otherwise +/// the queue stalls — the gate stays armed but no wake will fire, and the +/// next op's enqueue won't try to re-arm. +#[inline] +#[must_use = "channel overflow requires resetting drain_scheduled to avoid a stall"] +pub(crate) fn send_wake(inner: Arc) -> Result<(), ()> { + match SENDER.get() { + Some(tx) => tx.try_send(inner).map_err(|_| ()), + None => Err(()), + } +} + +/// Lazy-initialize the single waker thread. Idempotent across all +/// `CompletionBridge::new` calls (AsyncPool may construct many). +/// +/// Must be called under GIL (uses Python to spawn a `threading.Thread`). +pub(crate) fn ensure_waker(py: Python<'_>) -> PyResult<()> { + if SENDER.get().is_some() { + return Ok(()); // already initialized + } + let (tx, rx) = sync_channel(WAKER_CHANNEL_CAPACITY); + // Race: another thread may have raced us to set SENDER. set() returns + // Err if already set; in that case our channel is dropped and we bail. + if SENDER.set(tx).is_err() { + return Ok(()); + } + // Store the receiver so the spawned Python thread can take() it. + let _ = PENDING_RX.set(Mutex::new(Some(rx))); + + // Spawn the daemon Python thread. Using Python's `threading.Thread` + // (Variant 1 of the plan) keeps the persistent PyThreadState under + // CPython's lifecycle management — no unsafe `PyGILState_Ensure` + // without a matching Release as Branch D required. + let threading = py.import(pyo3::intern!(py, "threading"))?; + let kwargs = PyDict::new(py); + let target = wrap_pyfunction!(waker_loop_py, py)?; + kwargs.set_item(pyo3::intern!(py, "target"), target)?; + kwargs.set_item(pyo3::intern!(py, "name"), "aerospike-waker")?; + kwargs.set_item(pyo3::intern!(py, "daemon"), true)?; + let thread = threading.call_method(pyo3::intern!(py, "Thread"), (), Some(&kwargs))?; + thread.call_method0(pyo3::intern!(py, "start"))?; + Ok(()) +} + +/// The waker thread's recv loop. Runs as a Python daemon thread; CPython +/// owns this thread's `PyThreadState` for the interpreter's lifetime. +#[pyfunction] +fn waker_loop_py(py: Python<'_>) { + // Take the receiver from the global; if already taken (started twice), + // bail. This is defensive — `ensure_waker` is idempotent and only + // starts the thread once via `SENDER.set` racing semantics. + let rx = match PENDING_RX.get().and_then(|m| m.lock().ok()).and_then(|mut g| g.take()) { + Some(rx) => rx, + None => return, + }; + + loop { + // Release the GIL while blocked on the channel. Required for GIL-on + // builds (otherwise we'd block other Python threads waiting to run); + // benign for FT (no GIL exists to release). + // + // `Python::detach` won't compile here because `&Receiver` isn't + // `Send` (Receiver is !Sync) — go straight to the FFI. Same + // mechanism PyO3's `detach` uses internally; we just sidestep its + // closure-Send bound. + let token = unsafe { + let saved = pyo3::ffi::PyEval_SaveThread(); + let result = rx.recv(); + pyo3::ffi::PyEval_RestoreThread(saved); + result + }; + let inner = match token { + Ok(i) => i, + // All senders dropped — interpreter shutdown. Exit cleanly. + Err(_) => return, + }; + + // Finalization guard: if the interpreter is being torn down, + // Python::attach itself would assert. + if unsafe { pyo3::ffi::Py_IsInitialized() } == 0 { + return; + } + + // We're back attached after detach(); call_soon_threadsafe runs + // here on our persistent PyThreadState — no churn per token. + let result: PyResult<()> = (|| { + let el = inner.owning_loop.bind(py); + let drainer = inner + .drainer + .get() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err( + "drainer not set on CompletionInner", + ))?; + el.call_method1( + pyo3::intern!(py, "call_soon_threadsafe"), + (drainer.bind(py),), + )?; + Ok(()) + })(); + + if result.is_err() { + // Loop closed during/after enqueue (call_soon_threadsafe rejects + // submissions on a closed loop). Latch + drain pending so + // callers don't hang. + inner.closed.store(true, Ordering::Release); + inner.fail_all_pending(); + } + } +} From ab8f48efbdb351b7046bd2860838f3c10030af8f Mon Sep 17 00:00:00 2001 From: Paul Brackin Date: Fri, 19 Jun 2026 11:39:01 -0700 Subject: [PATCH 2/3] CI update to remove 3.13t target --- .github/workflows/build-test.yml | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml index 497d0dc..1b9620c 100644 --- a/.github/workflows/build-test.yml +++ b/.github/workflows/build-test.yml @@ -537,7 +537,7 @@ jobs: runs-on: ubuntu-22.04 strategy: matrix: - python: ['3.13t', '3.14t'] + python: ['3.14t'] fail-fast: false steps: - uses: actions/checkout@v4 @@ -561,9 +561,10 @@ jobs: sccache: 'true' # Build against the host Python installed by setup-python above. # The default manylinux Docker container does not include - # free-threaded CPython builds (3.13t / 3.14t), so spinning one - # up causes maturin to fail with "Couldn't find any python - # interpreters". + # free-threaded CPython builds (3.14t), so spinning one up causes + # maturin to fail with "Couldn't find any python interpreters". + # (3.13t was dropped — PyO3 0.29 removed support; FT now starts + # at 3.14t.) manylinux: off env: AWS_LC_SYS_CMAKE_BUILDER: "0" From 66ba49585774e1865e76322a254630f6da73ddd1 Mon Sep 17 00:00:00 2001 From: Paul Brackin Date: Fri, 19 Jun 2026 11:52:51 -0700 Subject: [PATCH 3/3] fix test --- .../integration/multi_loop_completion_test.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/python/tests/integration/multi_loop_completion_test.py b/python/tests/integration/multi_loop_completion_test.py index f0b173b..a2797ac 100644 --- a/python/tests/integration/multi_loop_completion_test.py +++ b/python/tests/integration/multi_loop_completion_test.py @@ -138,11 +138,19 @@ async def make_client(): try: async def use_client_on_wrong_loop(): - # The Client was bound to loop_a; calling .get() while running - # on loop_b must fail at batched_future_into_py entry. + # The Client was bound to loop_a; calling .get() while + # running on loop_b must fail. In debug builds the failure + # is the explicit upfront check in `batched_future_into_py` + # ("...different event loop than the one that created it"). + # In release builds that diagnostic is gated out + # (`#[cfg(debug_assertions)]`) and the misuse surfaces + # downstream when asyncio tries to resolve a future + # attached to the wrong loop ("...attached to a different + # loop"). Either is acceptable — both are loud, + # raise-immediately failures. return await client.get(Key("test", "test", "x"), policy=ReadPolicy()) - with pytest.raises(RuntimeError, match="different event loop"): + with pytest.raises(RuntimeError, match=r"different .*loop"): _submit(loop_b, use_client_on_wrong_loop()) finally: try: