commit 2b0127c47610fc854a8eae52e584ffcacae4d351 Author: Karim Abdul-Samad Date: Sat Mar 7 10:45:02 2026 -0500 Init. Tested against `cargo bench`, the single-threaded consumer is confirmed working. diff --git a/.envrc b/.envrc new file mode 100644 index 0000000..3550a30 --- /dev/null +++ b/.envrc @@ -0,0 +1 @@ +use flake diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3dbe527 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +target/ +.direnv/ +.claude/ +CLAUDE.md diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..51d5713 --- /dev/null +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..b22d2bb --- /dev/null +++ b/Cargo.toml @@ -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 } diff --git a/benches/epoll_bench.rs b/benches/epoll_bench.rs new file mode 100644 index 0000000..cbda973 --- /dev/null +++ b/benches/epoll_bench.rs @@ -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); diff --git a/flake.lock b/flake.lock new file mode 100644 index 0000000..5f9c783 --- /dev/null +++ b/flake.lock @@ -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 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..27b128f --- /dev/null +++ b/flake.nix @@ -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; }; + } + ); +} diff --git a/package.nix b/package.nix new file mode 100644 index 0000000..59e3503 --- /dev/null +++ b/package.nix @@ -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; + }; +} diff --git a/rust-toolchain.toml b/rust-toolchain.toml new file mode 100644 index 0000000..b7be73c --- /dev/null +++ b/rust-toolchain.toml @@ -0,0 +1,4 @@ +[toolchain] +channel = "stable" +components = ["rustfmt", "rustc", "clippy", "rust-src", "rust-analyzer"] +profile = "default" diff --git a/src/buffer.rs b/src/buffer.rs new file mode 100644 index 0000000..7608659 --- /dev/null +++ b/src/buffer.rs @@ -0,0 +1,63 @@ +use std::{cell::RefCell, os::fd::RawFd}; + +use nix::{errno::Errno, libc::read}; + +thread_local! { + static READ_BUF: RefCell> = RefCell::new( + Box::new([0u8; 1024]) + ); +} + +pub fn with_read_buf(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)) +} diff --git a/src/client/config.rs b/src/client/config.rs new file mode 100644 index 0000000..cb77ff7 --- /dev/null +++ b/src/client/config.rs @@ -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, +} diff --git a/src/client/mod.rs b/src/client/mod.rs new file mode 100644 index 0000000..1638d8a --- /dev/null +++ b/src/client/mod.rs @@ -0,0 +1,5 @@ +mod config; +mod spawner; + +pub use config::ClientConfig; +pub use spawner::ClientPool; diff --git a/src/client/spawner.rs b/src/client/spawner.rs new file mode 100644 index 0000000..dc70238 --- /dev/null +++ b/src/client/spawner.rs @@ -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>, + state: Arc<(Mutex, Condvar)>, + shutdown: Arc, + 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(); + } + } +} diff --git a/src/epoll/event_loop.rs b/src/epoll/event_loop.rs new file mode 100644 index 0000000..e8f91e2 --- /dev/null +++ b/src/epoll/event_loop.rs @@ -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()); + } + } +} diff --git a/src/epoll/instance.rs b/src/epoll/instance.rs new file mode 100644 index 0000000..c855194 --- /dev/null +++ b/src/epoll/instance.rs @@ -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() + } +} diff --git a/src/epoll/mod.rs b/src/epoll/mod.rs new file mode 100644 index 0000000..bd2ed8c --- /dev/null +++ b/src/epoll/mod.rs @@ -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; diff --git a/src/epoll/wait_buf.rs b/src/epoll/wait_buf.rs new file mode 100644 index 0000000..65b0126 --- /dev/null +++ b/src/epoll/wait_buf.rs @@ -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] + } +} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..55885b0 --- /dev/null +++ b/src/lib.rs @@ -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}; diff --git a/src/metrics/collector.rs b/src/metrics/collector.rs new file mode 100644 index 0000000..e69de29 diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs new file mode 100644 index 0000000..e024786 --- /dev/null +++ b/src/metrics/mod.rs @@ -0,0 +1 @@ +pub mod collector; diff --git a/src/server/handle.rs b/src/server/handle.rs new file mode 100644 index 0000000..d526783 --- /dev/null +++ b/src/server/handle.rs @@ -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, + shutdown: Arc, + threads: Vec>, +} + +impl ServerHandle { + pub(crate) fn new( + addr: SocketAddr, + processed_bytes: Arc, + shutdown: Arc, + threads: Vec>, + ) -> 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(); + } + } +} diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 0000000..e700b98 --- /dev/null +++ b/src/server/mod.rs @@ -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), + } + } +} diff --git a/src/server/per_worker.rs b/src/server/per_worker.rs new file mode 100644 index 0000000..d9c6e6f --- /dev/null +++ b/src/server/per_worker.rs @@ -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 { + 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::() 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::() 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::(sockaddr) }, + std::mem::size_of::() 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) +} diff --git a/src/server/shared_epoll.rs b/src/server/shared_epoll.rs new file mode 100644 index 0000000..8e774ac --- /dev/null +++ b/src/server/shared_epoll.rs @@ -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) +} diff --git a/src/server/single_thread.rs b/src/server/single_thread.rs new file mode 100644 index 0000000..a90ba0a --- /dev/null +++ b/src/server/single_thread.rs @@ -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]) +} diff --git a/src/worker/mod.rs b/src/worker/mod.rs new file mode 100644 index 0000000..337edbc --- /dev/null +++ b/src/worker/mod.rs @@ -0,0 +1,2 @@ +pub mod pool; +pub mod task; diff --git a/src/worker/pool.rs b/src/worker/pool.rs new file mode 100644 index 0000000..e69de29 diff --git a/src/worker/task.rs b/src/worker/task.rs new file mode 100644 index 0000000..e69de29 diff --git a/treefmt.toml b/treefmt.toml new file mode 100644 index 0000000..178d549 --- /dev/null +++ b/treefmt.toml @@ -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 +# 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