Init. Tested against cargo bench, the single-threaded consumer is confirmed working.

This commit is contained in:
2026-03-11 11:54:17 -04:00
commit 2b0127c476
29 changed files with 1919 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
use flake
+4
View File
@@ -0,0 +1,4 @@
target/
.direnv/
.claude/
CLAUDE.md
Generated
+811
View File
@@ -0,0 +1,811 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4
[[package]]
name = "aho-corasick"
version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
dependencies = [
"memchr",
]
[[package]]
name = "anes"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
[[package]]
name = "anstream"
version = "0.6.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a"
dependencies = [
"anstyle",
"anstyle-parse",
"anstyle-query",
"anstyle-wincon",
"colorchoice",
"is_terminal_polyfill",
"utf8parse",
]
[[package]]
name = "anstyle"
version = "1.0.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78"
[[package]]
name = "anstyle-parse"
version = "0.2.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e7644824f0aa2c7b9384579234ef10eb7efb6a0deb83f9630a49594dd9c15c2"
dependencies = [
"utf8parse",
]
[[package]]
name = "anstyle-query"
version = "1.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
dependencies = [
"windows-sys",
]
[[package]]
name = "anstyle-wincon"
version = "3.0.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
dependencies = [
"anstyle",
"once_cell_polyfill",
"windows-sys",
]
[[package]]
name = "autocfg"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
[[package]]
name = "bitflags"
version = "2.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af"
[[package]]
name = "bumpalo"
version = "3.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb"
[[package]]
name = "cast"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
[[package]]
name = "cfg-if"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
[[package]]
name = "cfg_aliases"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "ciborium"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e"
dependencies = [
"ciborium-io",
"ciborium-ll",
"serde",
]
[[package]]
name = "ciborium-io"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757"
[[package]]
name = "ciborium-ll"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9"
dependencies = [
"ciborium-io",
"half",
]
[[package]]
name = "clap"
version = "4.5.60"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2797f34da339ce31042b27d23607e051786132987f595b02ba4f6a6dffb7030a"
dependencies = [
"clap_builder",
"clap_derive",
]
[[package]]
name = "clap_builder"
version = "4.5.60"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24a241312cea5059b13574bb9b3861cabf758b879c15190b37b6d6fd63ab6876"
dependencies = [
"anstream",
"anstyle",
"clap_lex",
"strsim",
]
[[package]]
name = "clap_derive"
version = "4.5.55"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a92793da1a46a5f2a02a6f4c46c6496b28c43638adea8306fcb0caa1634f24e5"
dependencies = [
"heck",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "clap_lex"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a822ea5bc7590f9d40f1ba12c0dc3c2760f3482c6984db1573ad11031420831"
[[package]]
name = "colorchoice"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
[[package]]
name = "criterion"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f"
dependencies = [
"anes",
"cast",
"ciborium",
"clap",
"criterion-plot",
"is-terminal",
"itertools",
"num-traits",
"once_cell",
"oorandom",
"plotters",
"rayon",
"regex",
"serde",
"serde_derive",
"serde_json",
"tinytemplate",
"walkdir",
]
[[package]]
name = "criterion-plot"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1"
dependencies = [
"cast",
"itertools",
]
[[package]]
name = "crossbeam-deque"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51"
dependencies = [
"crossbeam-epoch",
"crossbeam-utils",
]
[[package]]
name = "crossbeam-epoch"
version = "0.9.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-utils"
version = "0.8.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
[[package]]
name = "crunchy"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5"
[[package]]
name = "either"
version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
[[package]]
name = "epoll-perf"
version = "0.1.0"
dependencies = [
"clap",
"criterion",
"io-uring",
"nix",
"serde",
"serde_json",
"serde_toml",
"serde_yaml",
"toml",
]
[[package]]
name = "equivalent"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
[[package]]
name = "half"
version = "2.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b"
dependencies = [
"cfg-if",
"crunchy",
"zerocopy",
]
[[package]]
name = "hashbrown"
version = "0.16.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
[[package]]
name = "heck"
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 = "indexmap"
version = "2.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017"
dependencies = [
"equivalent",
"hashbrown",
]
[[package]]
name = "io-uring"
version = "0.7.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fdd7bddefd0a8833b88a4b68f90dae22c7450d11b354198baee3874fd811b344"
dependencies = [
"bitflags",
"cfg-if",
"libc",
]
[[package]]
name = "is-terminal"
version = "0.4.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
dependencies = [
"hermit-abi",
"libc",
"windows-sys",
]
[[package]]
name = "is_terminal_polyfill"
version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
[[package]]
name = "itertools"
version = "0.10.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473"
dependencies = [
"either",
]
[[package]]
name = "itoa"
version = "1.0.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2"
[[package]]
name = "js-sys"
version = "0.3.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b49715b7073f385ba4bc528e5747d02e66cb39c6146efb66b781f131f0fb399c"
dependencies = [
"once_cell",
"wasm-bindgen",
]
[[package]]
name = "libc"
version = "0.2.182"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112"
[[package]]
name = "memchr"
version = "2.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
[[package]]
name = "nix"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46"
dependencies = [
"bitflags",
"cfg-if",
"cfg_aliases",
"libc",
]
[[package]]
name = "num-traits"
version = "0.2.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841"
dependencies = [
"autocfg",
]
[[package]]
name = "once_cell"
version = "1.21.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
[[package]]
name = "once_cell_polyfill"
version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
[[package]]
name = "oorandom"
version = "11.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
[[package]]
name = "plotters"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747"
dependencies = [
"num-traits",
"plotters-backend",
"plotters-svg",
"wasm-bindgen",
"web-sys",
]
[[package]]
name = "plotters-backend"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a"
[[package]]
name = "plotters-svg"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670"
dependencies = [
"plotters-backend",
]
[[package]]
name = "proc-macro2"
version = "1.0.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934"
dependencies = [
"unicode-ident",
]
[[package]]
name = "quote"
version = "1.0.45"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924"
dependencies = [
"proc-macro2",
]
[[package]]
name = "rayon"
version = "1.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f"
dependencies = [
"either",
"rayon-core",
]
[[package]]
name = "rayon-core"
version = "1.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91"
dependencies = [
"crossbeam-deque",
"crossbeam-utils",
]
[[package]]
name = "regex"
version = "1.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
dependencies = [
"aho-corasick",
"memchr",
"regex-automata",
"regex-syntax",
]
[[package]]
name = "regex-automata"
version = "0.4.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f"
dependencies = [
"aho-corasick",
"memchr",
"regex-syntax",
]
[[package]]
name = "regex-syntax"
version = "0.8.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
[[package]]
name = "rustversion"
version = "1.0.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
[[package]]
name = "ryu"
version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
[[package]]
name = "same-file"
version = "1.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502"
dependencies = [
"winapi-util",
]
[[package]]
name = "serde"
version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
dependencies = [
"serde_core",
"serde_derive",
]
[[package]]
name = "serde_core"
version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "serde_json"
version = "1.0.149"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86"
dependencies = [
"itoa",
"memchr",
"serde",
"serde_core",
"zmij",
]
[[package]]
name = "serde_spanned"
version = "0.6.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3"
dependencies = [
"serde",
]
[[package]]
name = "serde_toml"
version = "0.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e68184169792863bfb6d8b5de8e15456ebf4302f28236f0ff1136591eb66541"
[[package]]
name = "serde_yaml"
version = "0.9.34+deprecated"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
dependencies = [
"indexmap",
"itoa",
"ryu",
"serde",
"unsafe-libyaml",
]
[[package]]
name = "strsim"
version = "0.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
[[package]]
name = "syn"
version = "2.0.117"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "tinytemplate"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc"
dependencies = [
"serde",
"serde_json",
]
[[package]]
name = "toml"
version = "0.8.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
dependencies = [
"serde",
"serde_spanned",
"toml_datetime",
"toml_edit",
]
[[package]]
name = "toml_datetime"
version = "0.6.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c"
dependencies = [
"serde",
]
[[package]]
name = "toml_edit"
version = "0.22.27"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
dependencies = [
"indexmap",
"serde",
"serde_spanned",
"toml_datetime",
"toml_write",
"winnow",
]
[[package]]
name = "toml_write"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
[[package]]
name = "unicode-ident"
version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
[[package]]
name = "unsafe-libyaml"
version = "0.2.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
[[package]]
name = "utf8parse"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "walkdir"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b"
dependencies = [
"same-file",
"winapi-util",
]
[[package]]
name = "wasm-bindgen"
version = "0.2.114"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6532f9a5c1ece3798cb1c2cfdba640b9b3ba884f5db45973a6f442510a87d38e"
dependencies = [
"cfg-if",
"once_cell",
"rustversion",
"wasm-bindgen-macro",
"wasm-bindgen-shared",
]
[[package]]
name = "wasm-bindgen-macro"
version = "0.2.114"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "18a2d50fcf105fb33bb15f00e7a77b772945a2ee45dcf454961fd843e74c18e6"
dependencies = [
"quote",
"wasm-bindgen-macro-support",
]
[[package]]
name = "wasm-bindgen-macro-support"
version = "0.2.114"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "03ce4caeaac547cdf713d280eda22a730824dd11e6b8c3ca9e42247b25c631e3"
dependencies = [
"bumpalo",
"proc-macro2",
"quote",
"syn",
"wasm-bindgen-shared",
]
[[package]]
name = "wasm-bindgen-shared"
version = "0.2.114"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75a326b8c223ee17883a4251907455a2431acc2791c98c26279376490c378c16"
dependencies = [
"unicode-ident",
]
[[package]]
name = "web-sys"
version = "0.3.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "854ba17bb104abfb26ba36da9729addc7ce7f06f5c0f90f3c391f8461cca21f9"
dependencies = [
"js-sys",
"wasm-bindgen",
]
[[package]]
name = "winapi-util"
version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys",
]
[[package]]
name = "windows-link"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
[[package]]
name = "windows-sys"
version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
"windows-link",
]
[[package]]
name = "winnow"
version = "0.7.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945"
dependencies = [
"memchr",
]
[[package]]
name = "zerocopy"
version = "0.8.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a789c6e490b576db9f7e6b6d661bcc9799f7c0ac8352f56ea20193b2681532e5"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
version = "0.8.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f65c489a7071a749c849713807783f70672b28094011623e200cb86dcb835953"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "zmij"
version = "1.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
+23
View File
@@ -0,0 +1,23 @@
[package]
name = "epoll-perf"
version = "0.1.0"
edition = "2024"
[dependencies]
clap = { version = "4.5.28", features = ["derive"] }
nix = { version = "0.29.0", features = ["event", "fs", "process", "signal", "zerocopy", "poll"] }
serde = { version = "1.0.219", features = ["derive"] }
serde_json = "1.0.140"
serde_toml = "0.0.1"
serde_yaml = "0.9.34"
toml = "0.8.20"
[dev-dependencies]
criterion = { version = "0.5", features = ["html_reports"] }
[[bench]]
name = "epoll_bench"
harness = false
[target.'cfg(target_os = "linux")'.dependencies]
io-uring = { version = "0.7.4", optional = true }
+56
View File
@@ -0,0 +1,56 @@
use criterion::{Criterion, criterion_group, criterion_main};
use epoll_perf::{
DrainPolicy, ServerConfig, ServerModel,
client::{ClientConfig, ClientPool},
};
use std::time::{Duration, Instant};
static BASIC_CLIENT_CONFIG: ClientConfig = ClientConfig {
num_clients: 10,
payload_size: 1024,
writes_per_client: 2048,
addr: None,
};
fn bench_single_consumer(c: &mut Criterion) {
let server = ServerConfig {
bind_addr: "127.0.0.1:0".parse().unwrap(),
model: ServerModel::SingleThread,
edge_triggered: false,
drain_policy: DrainPolicy::UntilWouldBlock,
event_buffer_size: 1024,
}
.start();
let client = ClientPool::spawn(&ClientConfig {
addr: Some(server.addr),
..BASIC_CLIENT_CONFIG
});
c.bench_function("single_consumer", |b| {
b.iter_custom(|iters| {
let mut total = Duration::ZERO;
for _i in 0..iters {
// reset server/client/timer
server.reset();
client.reset();
let start = Instant::now();
// wait for completion
client.wait();
while server.processed_bytes() < client.total_bytes() {
std::hint::spin_loop();
}
// tally result
total += start.elapsed();
}
total
});
});
server.shutdown();
}
criterion_group!(epoll_benches, bench_single_consumer);
criterion_main!(epoll_benches);
Generated
+82
View File
@@ -0,0 +1,82 @@
{
"nodes": {
"flake-utils": {
"inputs": {
"systems": "systems"
},
"locked": {
"lastModified": 1731533236,
"narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "flake-utils",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1772773019,
"narHash": "sha256-E1bxHxNKfDoQUuvriG71+f+s/NT0qWkImXsYZNFFfCs=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "aca4d95fce4914b3892661bcb80b8087293536c6",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"root": {
"inputs": {
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs",
"rust-overlay": "rust-overlay"
}
},
"rust-overlay": {
"inputs": {
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1772852295,
"narHash": "sha256-3FB/WzLZSiU2Mc50C9q9VXU1LRUZbsU6UHKmZG1C+hU=",
"owner": "oxalica",
"repo": "rust-overlay",
"rev": "c10801f59c68e14c308aea8fa6b0b3d81d43c61e",
"type": "github"
},
"original": {
"owner": "oxalica",
"repo": "rust-overlay",
"type": "github"
}
},
"systems": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"type": "github"
}
}
},
"root": "root",
"version": 7
}
+38
View File
@@ -0,0 +1,38 @@
{
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
flake-utils.url = "github:numtide/flake-utils";
rust-overlay.url = "github:oxalica/rust-overlay";
rust-overlay.inputs.nixpkgs.follows = "nixpkgs";
};
outputs =
{
self,
nixpkgs,
flake-utils,
rust-overlay,
}:
flake-utils.lib.eachDefaultSystem (
system:
let
overlays = [ (import rust-overlay) ];
pkgs = import nixpkgs {
inherit system overlays;
};
rustToolchain = pkgs.pkgsBuildHost.rust-bin.fromRustupToolchainFile ./rust-toolchain.toml;
rustPlatform = pkgs.makeRustPlatform {
cargo = rustToolchain;
rustc = rustToolchain;
};
in
with pkgs;
{
formatter = nixfmt-rfc-style;
devShells.default = mkShell {
buildInputs = [ rustToolchain ];
};
packages.default = callPackage ./package.nix { inherit rustPlatform; };
}
);
}
+13
View File
@@ -0,0 +1,13 @@
{ rustPlatform, ... }:
let
cargoToml = builtins.fromTOML (builtins.readFile ./Cargo.toml);
version = cargoToml.package.version;
pname = cargoToml.package.name;
in
rustPlatform.buildRustPackage {
inherit pname version;
src = ./.;
cargoLock = {
lockFile = ./Cargo.lock;
};
}
+4
View File
@@ -0,0 +1,4 @@
[toolchain]
channel = "stable"
components = ["rustfmt", "rustc", "clippy", "rust-src", "rust-analyzer"]
profile = "default"
+63
View File
@@ -0,0 +1,63 @@
use std::{cell::RefCell, os::fd::RawFd};
use nix::{errno::Errno, libc::read};
thread_local! {
static READ_BUF: RefCell<Box<[u8]>> = RefCell::new(
Box::new([0u8; 1024])
);
}
pub fn with_read_buf<F, R>(f: F) -> R
where
F: FnOnce(&mut [u8]) -> R,
{
READ_BUF.with(|buf| f(&mut buf.borrow_mut()))
}
#[derive(Copy, Clone)]
pub enum DrainPolicy {
Multi(usize),
UntilWouldBlock,
}
pub fn drain_(fd: RawFd, buf: &mut [u8], policy: DrainPolicy) -> usize {
let mut total_bytes_read: usize = 0;
match policy {
DrainPolicy::Multi(n) => {
for _ in 0..(n as _) {
match Errno::result(unsafe { read(fd, buf.as_mut_ptr().cast(), buf.len()) }) {
Ok(n_read) => {
total_bytes_read += n_read as usize;
if n_read == 0 {
return total_bytes_read;
}
}
Err(errno) => match errno {
Errno::EAGAIN => return total_bytes_read,
_ => todo!(),
},
}
}
total_bytes_read
}
DrainPolicy::UntilWouldBlock => loop {
match Errno::result(unsafe { read(fd, buf.as_mut_ptr().cast(), buf.len()) }) {
Ok(n_read) => {
total_bytes_read += n_read as usize;
if n_read == 0 {
return total_bytes_read;
}
}
Err(errno) => match errno {
Errno::EAGAIN => return total_bytes_read,
_ => todo!(),
},
}
},
}
}
pub fn drain(fd: RawFd, policy: DrainPolicy) -> usize {
with_read_buf(|buf| drain_(fd, buf, policy))
}
+9
View File
@@ -0,0 +1,9 @@
use std::net::SocketAddr;
#[derive(Copy, Clone)]
pub struct ClientConfig {
pub num_clients: usize,
pub payload_size: usize,
pub writes_per_client: usize,
pub addr: Option<SocketAddr>,
}
+5
View File
@@ -0,0 +1,5 @@
mod config;
mod spawner;
pub use config::ClientConfig;
pub use spawner::ClientPool;
+119
View File
@@ -0,0 +1,119 @@
use super::ClientConfig;
use std::{
io::Write,
net::TcpStream,
sync::{
Arc, Condvar, Mutex,
atomic::{AtomicBool, Ordering},
},
thread::{self, JoinHandle},
};
struct GenerationState {
generation: u64,
ready: usize,
done: usize,
}
pub struct ClientPool {
handles: Vec<JoinHandle<()>>,
state: Arc<(Mutex<GenerationState>, Condvar)>,
shutdown: Arc<AtomicBool>,
config: ClientConfig,
}
impl ClientPool {
pub fn spawn(config: &ClientConfig) -> Self {
let state = Arc::new((
Mutex::new(GenerationState {
generation: 0,
ready: 0,
done: 0,
}),
Condvar::new(),
));
let shutdown = Arc::new(AtomicBool::new(false));
let addr = config.addr.unwrap();
let handles = (0..config.num_clients)
.map(|_i| {
let state = Arc::clone(&state);
let shutdown = Arc::clone(&shutdown);
let payload_size = config.payload_size;
let writes = config.writes_per_client;
thread::spawn(move || {
let mut sock = TcpStream::connect(addr).unwrap();
let buf = vec![0u8; payload_size];
loop {
// Signal ready and wait for next generation
let (lock, cvar) = &*state;
let mut guard = lock.lock().unwrap();
guard.ready += 1;
cvar.notify_all(); // Wake start_generation if it's waiting for ready
let current_gen = guard.generation;
while guard.generation == current_gen {
guard = cvar.wait(guard).unwrap();
}
drop(guard);
if shutdown.load(Ordering::Relaxed) {
break;
}
for _ in 0..writes {
sock.write_all(&buf).unwrap();
}
// Signal done
let (lock, cvar) = &*state;
let mut guard = lock.lock().unwrap();
guard.done += 1;
cvar.notify_all();
}
})
})
.collect();
ClientPool {
handles,
state,
shutdown,
config: *config,
}
}
pub fn reset(&self) {
let (lock, cvar) = &*self.state;
let mut guard = lock.lock().unwrap();
// Wait until all clients are ready
while guard.ready < self.config.num_clients {
guard = cvar.wait(guard).unwrap();
}
guard.ready = 0;
guard.generation += 1;
cvar.notify_all();
}
pub fn total_bytes(&self) -> usize {
self.config.num_clients * self.config.writes_per_client * self.config.payload_size
}
pub fn wait(&self) {
let (lock, cvar) = &*self.state;
let mut guard = lock.lock().unwrap();
while guard.done < self.config.num_clients {
guard = cvar.wait(guard).unwrap();
}
guard.done = 0;
}
pub fn shutdown(self) {
self.shutdown.store(true, Ordering::Relaxed);
self.reset(); // unblock threads so they can see the flag
for h in self.handles {
h.join().unwrap();
}
}
}
+16
View File
@@ -0,0 +1,16 @@
use crate::epoll::{instance::EpollInterface, wait_buf::EpollBuffer};
use nix::{poll::PollTimeout, sys::epoll::EpollFlags};
use std::os::fd::RawFd;
pub fn run_event_loop(
epoll: EpollInterface,
events_buffer: &mut EpollBuffer,
mut callback: impl FnMut(RawFd, EpollFlags),
) {
loop {
for event in events_buffer.wait(epoll.epfd, PollTimeout::NONE) {
callback(event.data() as RawFd, event.events());
}
}
}
+52
View File
@@ -0,0 +1,52 @@
use std::os::fd::RawFd;
use nix::{
errno::Errno,
libc::{self},
sys::epoll::{EpollCreateFlags, EpollFlags},
};
#[derive(Debug)]
pub struct EpollInterface {
pub epfd: RawFd,
}
impl EpollInterface {
pub fn new() -> Self {
let flags = EpollCreateFlags::empty();
let res = unsafe { libc::epoll_create1(flags.bits()) };
let epfd = Errno::result(res).unwrap();
Self { epfd }
}
pub fn add(&self, data: RawFd, events: EpollFlags) {
let mut event = libc::epoll_event {
events: events.bits() as u32,
u64: data as u64,
};
Errno::result(unsafe { libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_ADD, data, &mut event) })
.unwrap();
}
pub fn modify(&self, data: RawFd, events: EpollFlags) {
let mut event = libc::epoll_event {
events: events.bits() as u32,
u64: data as u64,
};
Errno::result(unsafe { libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_MOD, data, &mut event) })
.unwrap();
}
pub fn remove(&self, fd: RawFd) {
Errno::result(unsafe {
libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut())
})
.unwrap();
}
}
impl Default for EpollInterface {
fn default() -> Self {
Self::new()
}
}
+7
View File
@@ -0,0 +1,7 @@
mod event_loop;
mod instance;
mod wait_buf;
pub use event_loop::run_event_loop;
pub use instance::EpollInterface;
pub use wait_buf::EpollBuffer;
+31
View File
@@ -0,0 +1,31 @@
use std::os::fd::RawFd;
use nix::errno::Errno;
use nix::libc;
use nix::poll::PollTimeout;
use nix::sys::epoll::EpollEvent;
pub struct EpollBuffer {
buffer: Box<[EpollEvent]>,
}
impl EpollBuffer {
pub fn new(len: usize) -> Self {
Self {
buffer: unsafe { Box::new_uninit_slice(len).assume_init() },
}
}
pub fn wait(&mut self, epfd: RawFd, timeout: PollTimeout) -> &[EpollEvent] {
let n = Errno::result(unsafe {
libc::epoll_wait(
epfd,
self.buffer.as_mut_ptr().cast(),
self.buffer.len() as libc::c_int,
timeout.into(),
)
})
.unwrap() as usize;
&self.buffer[..n]
}
}
+9
View File
@@ -0,0 +1,9 @@
mod buffer;
pub mod client;
pub mod epoll;
pub mod metrics;
pub mod server;
pub mod worker;
pub use buffer::{DrainPolicy, drain};
pub use server::{ServerConfig, ServerHandle, ServerModel};
View File
+1
View File
@@ -0,0 +1 @@
pub mod collector;
+52
View File
@@ -0,0 +1,52 @@
use std::{
net::SocketAddr,
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc,
},
thread::JoinHandle,
};
pub struct ServerHandle {
pub addr: SocketAddr,
processed_bytes: Arc<AtomicUsize>,
shutdown: Arc<AtomicBool>,
threads: Vec<JoinHandle<()>>,
}
impl ServerHandle {
pub(crate) fn new(
addr: SocketAddr,
processed_bytes: Arc<AtomicUsize>,
shutdown: Arc<AtomicBool>,
threads: Vec<JoinHandle<()>>,
) -> Self {
Self {
addr,
processed_bytes,
shutdown,
threads,
}
}
/// Returns the current number of bytes processed.
pub fn processed_bytes(&self) -> usize {
self.processed_bytes.load(Ordering::Relaxed)
}
/// Resets the processed byte counter to zero.
/// Call this between benchmark iterations.
pub fn reset(&self) {
self.processed_bytes.store(0, Ordering::Relaxed);
}
/// Signals all worker threads to stop and waits for them to finish.
pub fn shutdown(self) {
self.shutdown.store(true, Ordering::Relaxed);
for handle in self.threads {
// Workers may be blocked on epoll_wait; they'll exit on next event
// or when the socket is closed. For now, we just wait.
let _ = handle.join();
}
}
}
+40
View File
@@ -0,0 +1,40 @@
mod handle;
mod per_worker;
mod shared_epoll;
mod single_thread;
pub use handle::ServerHandle;
use std::net::SocketAddr;
use crate::DrainPolicy;
pub struct ServerConfig {
pub bind_addr: SocketAddr,
pub model: ServerModel,
pub edge_triggered: bool,
pub drain_policy: DrainPolicy,
pub event_buffer_size: usize,
}
pub enum ServerModel {
SingleThread,
SharedEpoll {
num_workers: usize,
use_oneshot: bool,
use_exclusive: bool,
},
EpollPerWorker {
num_workers: usize,
},
}
impl ServerConfig {
pub fn start(self) -> ServerHandle {
match self.model {
ServerModel::SingleThread => single_thread::start(self),
ServerModel::SharedEpoll { .. } => shared_epoll::start(self),
ServerModel::EpollPerWorker { .. } => per_worker::start(self),
}
}
}
+185
View File
@@ -0,0 +1,185 @@
use std::{
net::{SocketAddr, TcpListener},
os::fd::{AsRawFd, FromRawFd, RawFd},
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc,
},
thread,
};
use nix::{libc, poll::PollTimeout, sys::epoll::EpollFlags};
use crate::{
drain,
epoll::{EpollBuffer, EpollInterface},
};
use super::{ServerConfig, ServerHandle, ServerModel};
/// Creates a TcpListener with SO_REUSEPORT enabled.
fn create_reuseport_listener(addr: SocketAddr) -> std::io::Result<TcpListener> {
let domain = if addr.is_ipv4() {
libc::AF_INET
} else {
libc::AF_INET6
};
let fd = unsafe { libc::socket(domain, libc::SOCK_STREAM | libc::SOCK_NONBLOCK, 0) };
if fd < 0 {
return Err(std::io::Error::last_os_error());
}
// Set SO_REUSEPORT
let optval: libc::c_int = 1;
let ret = unsafe {
libc::setsockopt(
fd,
libc::SOL_SOCKET,
libc::SO_REUSEPORT,
&optval as *const _ as *const libc::c_void,
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
)
};
if ret < 0 {
unsafe { libc::close(fd) };
return Err(std::io::Error::last_os_error());
}
// Set SO_REUSEADDR
let ret = unsafe {
libc::setsockopt(
fd,
libc::SOL_SOCKET,
libc::SO_REUSEADDR,
&optval as *const _ as *const libc::c_void,
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
)
};
if ret < 0 {
unsafe { libc::close(fd) };
return Err(std::io::Error::last_os_error());
}
// Bind
let (sockaddr, socklen) = match addr {
SocketAddr::V4(v4) => {
let sockaddr = libc::sockaddr_in {
sin_family: libc::AF_INET as libc::sa_family_t,
sin_port: v4.port().to_be(),
sin_addr: libc::in_addr {
s_addr: u32::from_ne_bytes(v4.ip().octets()),
},
sin_zero: [0; 8],
};
(
unsafe { std::mem::transmute::<libc::sockaddr_in, libc::sockaddr>(sockaddr) },
std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
)
}
SocketAddr::V6(_) => {
unsafe { libc::close(fd) };
return Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"IPv6 not implemented",
));
}
};
let ret = unsafe { libc::bind(fd, &sockaddr, socklen) };
if ret < 0 {
unsafe { libc::close(fd) };
return Err(std::io::Error::last_os_error());
}
// Listen
let ret = unsafe { libc::listen(fd, 128) };
if ret < 0 {
unsafe { libc::close(fd) };
return Err(std::io::Error::last_os_error());
}
Ok(unsafe { TcpListener::from_raw_fd(fd) })
}
pub fn start(config: ServerConfig) -> ServerHandle {
let num_workers = match config.model {
ServerModel::EpollPerWorker { num_workers } => num_workers,
_ => unreachable!(),
};
// Create first listener to get the actual bound address
let first_listener = create_reuseport_listener(config.bind_addr).expect("failed to bind");
let addr = first_listener.local_addr().expect("failed to get local addr");
let processed_bytes = Arc::new(AtomicUsize::new(0));
let shutdown = Arc::new(AtomicBool::new(false));
let mut handles = Vec::with_capacity(num_workers);
for i in 0..num_workers {
let listener = if i == 0 {
first_listener.try_clone().expect("failed to clone listener")
} else {
create_reuseport_listener(addr).expect("failed to bind reuseport listener")
};
let processed_clone = Arc::clone(&processed_bytes);
let shutdown_clone = Arc::clone(&shutdown);
let drain_policy = config.drain_policy;
let edge_triggered = config.edge_triggered;
let event_buffer_size = config.event_buffer_size;
let handle = thread::spawn(move || {
let epoll = EpollInterface::new();
let mut wait_buf = EpollBuffer::new(event_buffer_size);
let listen_fd = listener.as_raw_fd();
let mut listen_flags = EpollFlags::EPOLLIN;
if edge_triggered {
listen_flags |= EpollFlags::EPOLLET;
}
epoll.add(listen_fd, listen_flags);
let mut streams = Vec::new();
while !shutdown_clone.load(Ordering::Relaxed) {
for event in wait_buf.wait(epoll.epfd, PollTimeout::from(100u16)) {
let fd = event.data() as RawFd;
if fd == listen_fd {
loop {
match listener.accept() {
Ok((stream, _)) => {
stream
.set_nonblocking(true)
.expect("failed to set nonblocking");
let mut flags = EpollFlags::EPOLLIN;
if edge_triggered {
flags |= EpollFlags::EPOLLET;
}
epoll.add(stream.as_raw_fd(), flags);
streams.push(stream);
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
Err(_) => break,
}
}
} else {
let bytes = drain(fd, drain_policy);
processed_clone.fetch_add(bytes, Ordering::Relaxed);
}
}
}
});
handles.push(handle);
}
// Keep the first listener alive by moving it into the first worker above
// Actually, we cloned it, so we need to keep it alive
std::mem::forget(first_listener);
ServerHandle::new(addr, processed_bytes, shutdown, handles)
}
+126
View File
@@ -0,0 +1,126 @@
use std::{
net::TcpListener,
os::fd::{AsRawFd, RawFd},
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc,
},
thread,
};
use nix::{poll::PollTimeout, sys::epoll::EpollFlags};
use crate::{
drain,
epoll::{EpollBuffer, EpollInterface},
};
use super::{ServerConfig, ServerHandle, ServerModel};
pub fn start(config: ServerConfig) -> ServerHandle {
let (num_workers, use_oneshot, use_exclusive) = match config.model {
ServerModel::SharedEpoll {
num_workers,
use_oneshot,
use_exclusive,
} => (num_workers, use_oneshot, use_exclusive),
_ => unreachable!(),
};
let listener = TcpListener::bind(config.bind_addr).expect("failed to bind");
listener
.set_nonblocking(true)
.expect("failed to set nonblocking");
let addr = listener.local_addr().expect("failed to get local addr");
let listen_fd = listener.as_raw_fd();
let processed_bytes = Arc::new(AtomicUsize::new(0));
let shutdown = Arc::new(AtomicBool::new(false));
// Shared epoll instance across all workers
let epoll = Arc::new(EpollInterface::new());
// Register listener on the shared epoll
let mut listen_flags = EpollFlags::EPOLLIN;
if config.edge_triggered {
listen_flags |= EpollFlags::EPOLLET;
}
if use_exclusive {
listen_flags |= EpollFlags::EPOLLEXCLUSIVE;
}
epoll.add(listen_fd, listen_flags);
// We need to keep the listener alive; wrap in Arc
let listener = Arc::new(listener);
// Shared storage for accepted streams to keep them alive
let streams = Arc::new(std::sync::Mutex::new(Vec::new()));
let mut handles = Vec::with_capacity(num_workers);
for _ in 0..num_workers {
let epoll = Arc::clone(&epoll);
let listener = Arc::clone(&listener);
let streams = Arc::clone(&streams);
let processed_clone = Arc::clone(&processed_bytes);
let shutdown_clone = Arc::clone(&shutdown);
let drain_policy = config.drain_policy;
let edge_triggered = config.edge_triggered;
let event_buffer_size = config.event_buffer_size;
let handle = thread::spawn(move || {
let mut wait_buf = EpollBuffer::new(event_buffer_size);
while !shutdown_clone.load(Ordering::Relaxed) {
for event in wait_buf.wait(epoll.epfd, PollTimeout::from(100u16)) {
let fd = event.data() as RawFd;
if fd == listen_fd {
// Accept connections
loop {
match listener.accept() {
Ok((stream, _)) => {
stream
.set_nonblocking(true)
.expect("failed to set nonblocking");
let mut flags = EpollFlags::EPOLLIN;
if edge_triggered {
flags |= EpollFlags::EPOLLET;
}
if use_oneshot {
flags |= EpollFlags::EPOLLONESHOT;
}
if use_exclusive {
flags |= EpollFlags::EPOLLEXCLUSIVE;
}
epoll.add(stream.as_raw_fd(), flags);
streams.lock().unwrap().push(stream);
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
Err(_) => break,
}
}
} else {
let bytes = drain(fd, drain_policy);
processed_clone.fetch_add(bytes, Ordering::Relaxed);
// Re-arm if using EPOLLONESHOT
if use_oneshot {
let mut flags = EpollFlags::EPOLLIN | EpollFlags::EPOLLONESHOT;
if edge_triggered {
flags |= EpollFlags::EPOLLET;
}
epoll.modify(fd, flags);
}
}
}
}
});
handles.push(handle);
}
ServerHandle::new(addr, processed_bytes, shutdown, handles)
}
+80
View File
@@ -0,0 +1,80 @@
use std::{
net::TcpListener,
os::fd::{AsRawFd, RawFd},
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc,
},
thread,
};
use nix::{poll::PollTimeout, sys::epoll::EpollFlags};
use crate::{
drain,
epoll::{EpollBuffer, EpollInterface},
};
use super::{ServerConfig, ServerHandle};
pub fn start(config: ServerConfig) -> ServerHandle {
let listener = TcpListener::bind(config.bind_addr).expect("failed to bind");
listener
.set_nonblocking(true)
.expect("failed to set nonblocking");
let addr = listener.local_addr().expect("failed to get local addr");
let processed_bytes = Arc::new(AtomicUsize::new(0));
let shutdown = Arc::new(AtomicBool::new(false));
let processed_clone = Arc::clone(&processed_bytes);
let shutdown_clone = Arc::clone(&shutdown);
let event_buffer_size = config.event_buffer_size;
let drain_policy = config.drain_policy;
let edge_triggered = config.edge_triggered;
let handle = thread::spawn(move || {
let epoll = EpollInterface::new();
let mut wait_buf = EpollBuffer::new(event_buffer_size);
let listen_fd = listener.as_raw_fd();
let mut listen_flags = EpollFlags::EPOLLIN;
if edge_triggered {
listen_flags |= EpollFlags::EPOLLET;
}
epoll.add(listen_fd, listen_flags);
let mut streams = Vec::new();
while !shutdown_clone.load(Ordering::Relaxed) {
for event in wait_buf.wait(epoll.epfd, PollTimeout::from(100u16)) {
let fd = event.data() as RawFd;
if fd == listen_fd {
// Accept all pending connections (important for edge-triggered)
loop {
match listener.accept() {
Ok((stream, _)) => {
stream
.set_nonblocking(true)
.expect("failed to set nonblocking");
let mut flags = EpollFlags::EPOLLIN;
if edge_triggered {
flags |= EpollFlags::EPOLLET;
}
epoll.add(stream.as_raw_fd(), flags);
streams.push(stream);
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
Err(e) => eprintln!("accept failed: {e}"),
}
}
} else {
let bytes = drain(fd, drain_policy);
processed_clone.fetch_add(bytes, Ordering::Relaxed);
}
}
}
});
ServerHandle::new(addr, processed_bytes, shutdown, vec![handle])
}
+2
View File
@@ -0,0 +1,2 @@
pub mod pool;
pub mod task;
View File
View File
+90
View File
@@ -0,0 +1,90 @@
# One CLI to format the code tree - https://github.com/numtide/treefmt
# Do not exit with error if a configured formatter is missing
# Env $TREEFMT_ALLOW_MISSING_FORMATTER
# allow-missing-formatter = true
# The file into which a cpu profile will be written
# Env $TREEFMT_CPU_PROFILE
# cpu-profile = ./cpu.pprof
# Exclude files or directories matching the specified globs
# Env $TREEFMT_EXCLUDES
# excludes = ["*.md", "*.gif"]
excludes = [
"*.toml",
"*.yaml",
"*.md",
"*.nu",
"*.jpg",
"*.astro",
"*.css",
"*.webmanifest",
"*.svg",
"*.json",
"*.mjs",
"*.cjs",
"*.ts",
"*.txt",
"*.js",
"*.png",
"*.woff",
"*.woff2",
"*.ttf",
"*.kak",
"*.xml",
".gitignore",
"flake.lock",
"examples/*",
".envrc",
"Cargo.lock"
]
# Exit with error if any changes were made during execution
# Useful for CI
# Env $TREEFMT_FAIL_ON_CHANGE
# fail-on-change = true
# A list of formatters to apply
# Defaults to all configured formatters
# Env $TREEFMT_FORMATTERS
# formatters = ["gofmt", "prettier"]
# Log paths that did not match any formatters at the specified log level
# Possible values are <debug|info|warn|error|fatal>
# Env $TREEFMT_ON_UNMATCHED
# on-unmatched = "info"
# The root directory from which treefmt will start walking the filesystem
# Defaults to the directory containing the config file
# Env $TREEFMT_TREE_ROOT
# tree-root = "/tmp/foo"
# File to search for to find the tree root (if tree-root is not set)
# Env $TREEFMT_TREE_ROOT_FILE
# tree-root-file = ".git/config"
# Set the verbosity of logs
# 0 = warn, 1 = info, 2 = debug
# Env $TREEFMT_VERBOSE
# verbose = 2
# The method used to traverse the files within the tree root
# Currently, we support 'auto', 'git' or 'filesystem'
# Env $TREEFMT_WALK
# walk = "filesystem"
[formatter.nix]
command = "nix"
options = ["fmt"]
includes = [ "*.nix" ]
excludes = []
priority = 0
[formatter.rust]
command = "rustfmt"
options = [ "--config", "skip_children=true", "--edition", "2021" ]
includes = [ "*.rs" ]
excludes = []
priority = 0