diff --git a/scripts/benchmarks/gateway_split/.gitattributes b/scripts/benchmarks/gateway_split/.gitattributes new file mode 100644 index 00000000000..26bae010e4b --- /dev/null +++ b/scripts/benchmarks/gateway_split/.gitattributes @@ -0,0 +1 @@ +bifrost.patch -whitespace diff --git a/scripts/benchmarks/gateway_split/.gitignore b/scripts/benchmarks/gateway_split/.gitignore new file mode 100644 index 00000000000..117bde4d82c --- /dev/null +++ b/scripts/benchmarks/gateway_split/.gitignore @@ -0,0 +1,4 @@ +target/ +__pycache__/ +runtime/ +results/ diff --git a/scripts/benchmarks/gateway_split/AGENTS.md b/scripts/benchmarks/gateway_split/AGENTS.md new file mode 100644 index 00000000000..3fd37774af7 --- /dev/null +++ b/scripts/benchmarks/gateway_split/AGENTS.md @@ -0,0 +1,7 @@ +# Benchmark tooling + +Run the contract tests with `python -m unittest discover -s scripts/benchmarks/gateway_split -p 'test_*.py'` + +Use `run.py` for a short live infrastructure smoke check. Full performance sweeps run through `cloud.py` on separate otherwise idle machines. Never benchmark several topologies concurrently on the same host + +All performance measurements use the pinned Bifrost suite and saved patch. Generated trials, logs and local tools stay under ignored results/ and runtime/ directories. Do not commit laptop performance claims as cloud results diff --git a/scripts/benchmarks/gateway_split/Cargo.lock b/scripts/benchmarks/gateway_split/Cargo.lock new file mode 100644 index 00000000000..6d4f5df329f --- /dev/null +++ b/scripts/benchmarks/gateway_split/Cargo.lock @@ -0,0 +1,849 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +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 = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "bytes" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" + +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "clap" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa8876b300ab35ba921adea3dfd70157a46249b33f95c9084ae5709785478946" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0797fb7aeb1406c84efac526901f7ec3ead2124f946b494e72879d4b54704d" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9c751b79415d4e559e3d1fcf128e09e720eb673a06d26cf6f392d37d75b66e0" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "clap_lex" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-macro", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "gateway-split-bench" +version = "0.1.0" +dependencies = [ + "async-stream", + "axum", + "bytes", + "clap", + "futures-util", + "http-body-util", + "hyper-util", + "serde_json", + "sha2", + "tokio", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "hyper", + "libc", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.190" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + +[[package]] +name = "mio" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[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_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.16.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[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.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[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 = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/scripts/benchmarks/gateway_split/Cargo.toml b/scripts/benchmarks/gateway_split/Cargo.toml new file mode 100644 index 00000000000..22b153a02ed --- /dev/null +++ b/scripts/benchmarks/gateway_split/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "gateway-split-bench" +version = "0.1.0" +edition = "2024" +publish = false + +[workspace] + +[dependencies] +async-stream = "=0.3.6" +axum = "=0.8.9" +bytes = "=1.11.1" +clap = { version = "=4.6.7", features = ["derive"] } +futures-util = "=0.3.34" +http-body-util = "=0.1.5" +hyper-util = { version = "=0.1.20", features = ["client-legacy", "http1", "tokio"] } +serde_json = "=1.0.149" +sha2 = "=0.10.9" +tokio = { version = "=1.53.1", features = ["full"] } diff --git a/scripts/benchmarks/gateway_split/README.md b/scripts/benchmarks/gateway_split/README.md new file mode 100644 index 00000000000..acd009371dd --- /dev/null +++ b/scripts/benchmarks/gateway_split/README.md @@ -0,0 +1,102 @@ +# Axum and FastAPI cloud benchmark + +Shared tooling for a four-path forwarding comparison. This changes no production routing and includes no completed cloud performance claims + +| Row | Request path | +| --- | --- | +| Direct Axum | Client -> Axum -> fixture | +| Direct FastAPI | Client -> FastAPI -> fixture | +| Axum ingress with FastAPI fallback | Client -> Axum -> FastAPI -> fixture | +| FastAPI ingress with Axum fallback | Client -> FastAPI -> Axum -> fixture | + +Fallback rows delegate every request before provider execution. The path and a benchmark header select the destination. Authentication, model resolution and production lifecycle behavior are excluded. All paths use the same deterministic native Messages fixture + +## Matrix + +There is one profile, `litellm-current`. FastAPI uses the repository's `AsyncHTTPHandler`, including its aiohttp transport behind the HTTPX interface. Uvicorn uses uvloop and its automatic HTTP parser selection, matching the proxy CLI. The resolved parser and installed versions are recorded. Python collects request bodies, while both stacks stream responses. Axum forwards request bodies through Hyper without collecting them + +The SDK's pool defaults come from repository code, currently 1000 total connections, 500 per host and 15-second upstream keepalive. Rust retains up to 500 idle connections with the same keepalive. Both sides have a one-second connect timeout and 30-second request deadline for the experiment. Each gateway uses one event-loop thread or worker + +| Dimension | Values | +| --- | --- | +| Input text | 256 KiB, 1 MiB, 4 MiB, plus a small JSON envelope | +| Response | Complete JSON or streamed SSE | +| Concurrent in-flight requests | 1, 1000, 2000 | +| Measurement | Three 60-second repetitions | +| Warmup | Five seconds per topology | + +`matrix.json` defines 18 jobs and 216 measured trials. Each job runs all four topologies sequentially on one host. Order uses a recorded seed. A machine-wide lock prevents concurrent benchmark jobs from different checkouts + +Use identical, otherwise idle VM types across workers. Parallelize jobs across machines. At 2000 concurrent 4 MiB requests, each fully buffered copy can occupy roughly 8 GiB before responses and runtime overhead, so provision enough RAM. The common fixture and generator can limit throughput. These tests measure forwarding under controlled load rather than full LiteLLM production capacity + +## Setup + +Use the repository's Rust toolchain, a C linker, uv, Git, curl and shasum. From the repository root + +```bash +uv sync --frozen --extra proxy --group e2e-dev +bash scripts/benchmarks/gateway_split/bootstrap.sh +ulimit -n 32768 +``` + +The existing e2e dependency group supplies the SDK probe client and process sampler. The repository lockfile defines Python dependencies. Bootstrap verifies a pinned Go 1.27.2 archive, checks out Bifrost at `3cb72a95cacaee6f73f9db5636a1a651c7dcb1e9`, applies `bifrost.patch` and builds the binaries. It installs no system services. CGO is disabled to avoid an upstream Apple CPU detector startup crash. Go uses `-trimpath`, and Cargo.lock pins Rust dependencies + +## Cloud commands + +Regenerate the shared manifest after changing the definition + +```bash +.venv/bin/python scripts/benchmarks/gateway_split/cloud.py plan +``` + +Run one job + +```bash +bench=scripts/benchmarks/gateway_split +.venv/bin/python "$bench/cloud.py" run \ + --job litellm-current-4194304-buffered-c1000 \ + --machine-class YOUR_IDENTICAL_VM_CLASS \ + --output "$bench/results/cloud" +``` + +Or partition across six workers, assigning each a different index from 0 through 5 + +```bash +bench=scripts/benchmarks/gateway_split +.venv/bin/python "$bench/cloud.py" run \ + --shard-index 0 --shard-count 6 \ + --machine-class YOUR_IDENTICAL_VM_CLASS \ + --output "$bench/results/cloud" +``` + +Use the same manifest and commit everywhere. Each worker runs its assigned jobs sequentially. Collect the job directories into one location, then merge + +```bash +bench=scripts/benchmarks/gateway_split +.venv/bin/python "$bench/cloud.py" merge \ + --inputs "$bench/results/cloud" --output "$bench/results/merged" +``` + +Completed matching jobs are validated and skipped on rerun. Partial output is rejected, so preserve it for diagnosis and retry in a fresh directory. The merger requires every planned topology/repetition exactly once, passing probes, matching parameters and compatible hardware, dependency lock, versions, source hashes and suite binary + +`summary.json` contains four rows per workload with median and range for throughput, completion p50/p95/p99, first-event latency, CPU and sampled RSS. `trials.jsonl` retains individual trials. A median of trial percentiles is not a pooled percentile + +A short infrastructure smoke check + +```bash +.venv/bin/python scripts/benchmarks/gateway_split/run.py \ + --sizes 4194304 --concurrencies 1 --response buffered \ + --seconds 1 --repetitions 1 --warmup 1 +``` + +## Measurement boundaries + +The requested [Bifrost suite](https://github.com/maximhq/bifrost-benchmarking/tree/3cb72a95cacaee6f73f9db5636a1a651c7dcb1e9) drives every trial. Its original concurrency mode timed response headers, closed unread bodies and did not wait for requests launched by workers to finish. The saved patch adds native Messages payload/header files, full-body draining, in-flight completion, latency percentiles, elapsed-time throughput and first complete SSE frame timing + +Before loading, live probes verify SDK Messages and SSE reconstruction, request digests, headers, upload forwarding, 429/500 pass-through without retry and upstream cancellation. Every measured request must reach the fixture exactly once and return HTTP 200. SSE fixture completion counts must match client counts, with no unfinished or canceled streams. Request digests are checked during load. Bifrost drains responses, while SDK probes validate returned content before loading + +The SSE fixture sends metadata immediately, then eight text deltas with requested five-millisecond sleeps. First-event timing ends at a complete metadata frame and is not time to first text token. Completion timing includes the full response and in-flight drain. Local HTTP/1.1 loopback excludes TLS, provider networking, databases and caches. The fixture parses and hashes every request, which can dominate large-body measurements + +Resource samples include both gateways in fallback rows and exclude the fixture and client. RSS is sampled every 100 ms and may double-count shared pages. Source hashes, dependency versions, parser choice and load-generator CPU are recorded + +The FastAPI prototype uses the actual SDK transport but omits the proxy's JSON parsing, authentication and logging. Its default transport is aiohttp, while HTTP/2 selects HTTPX in production. The experiment explicitly uses HTTP/1.1. Closed-loop concurrency does not establish externally offered-rate capacity. Bifrost's [run guide](https://docs.getbifrost.ai/benchmarking/run-your-own-benchmarks) distinguishes those two load modes diff --git a/scripts/benchmarks/gateway_split/analyze_bifrost.py b/scripts/benchmarks/gateway_split/analyze_bifrost.py new file mode 100644 index 00000000000..981cab2a7d0 --- /dev/null +++ b/scripts/benchmarks/gateway_split/analyze_bifrost.py @@ -0,0 +1,124 @@ +import statistics +from collections.abc import Iterable, Iterator +from pathlib import Path +from typing import Final, Literal + +from bifrost_bench import FrozenModel, Stats + + +class Record(FrozenModel): + generator: Literal["bifrost"] + variant: str + prompt_bytes: int + body_bytes: int + stream: bool + concurrency: int + repetition: int + python_body: str + python_client: str + rust_client: str + event_loop: str + profile: str + http_parser: str + gateway_cpu_s: float + gateway_cpu_cores: float + gateway_peak_rss_bytes: int + loadgen_cpu_cores: float + stats: Stats + backend_completed: int + backend_requests: int + backend_cancelled: int + + +class Range(FrozenModel): + median: float + minimum: float + maximum: float + + +class Summary(FrozenModel): + variant: str + prompt_bytes: int + body_bytes: int + stream: bool + concurrency: int + python_body: str + python_client: str + rust_client: str + event_loop: str + profile: str + http_parser: str + repetitions: int + requests: int + rps: Range + p50_ms: Range + p95_ms: Range + p99_ms: Range + first_event_p50_ms: Range + first_event_p99_ms: Range + gateway_cpu_cores: Range + gateway_cpu_us_per_request: Range + gateway_peak_rss_mib: Range + loadgen_cpu_cores: Range + + +def distribution(values: Iterable[float]) -> Range: + samples: Final = tuple(values) + return Range(median=statistics.median(samples), minimum=min(samples), maximum=max(samples)) + + +def records(paths: tuple[Path, ...]) -> Iterator[Record]: + for path in paths: + for line in path.read_text().splitlines(): + yield validate_record(line) + + +def validate_record(line: str) -> Record: + row: Final = Record.model_validate_json(line) + assert row.stats.status_code_counts == {"200": row.stats.requests} + assert row.backend_requests == row.stats.requests + assert row.backend_cancelled == 0 + assert not row.stats.drop_reasons + if row.stream: + assert row.backend_completed == row.stats.requests + return row + + +def key(row: Record) -> tuple[str, int, bool, int, str]: + return ( + row.variant, + row.prompt_bytes, + row.stream, + row.concurrency, + row.profile, + ) + + +def summarize(rows: tuple[Record, ...], repetitions: int) -> Summary: + assert tuple(sorted(row.repetition for row in rows)) == tuple(range(repetitions)) + first: Final = rows[0] + return Summary( + variant=first.variant, + prompt_bytes=first.prompt_bytes, + body_bytes=first.body_bytes, + stream=first.stream, + concurrency=first.concurrency, + python_body=first.python_body, + python_client=first.python_client, + rust_client=first.rust_client, + http_parser=first.http_parser, + event_loop=first.event_loop, + profile=first.profile, + repetitions=len(rows), + requests=sum(row.stats.requests for row in rows), + rps=distribution(row.stats.throughput_rps for row in rows), + p50_ms=distribution(row.stats.p50_latency_ms for row in rows), + p95_ms=distribution(row.stats.p95_latency_ms for row in rows), + p99_ms=distribution(row.stats.p99_latency_ms for row in rows), + first_event_p50_ms=distribution(row.stats.first_event_p50_ms for row in rows), + first_event_p99_ms=distribution(row.stats.first_event_p99_ms for row in rows), + gateway_cpu_cores=distribution(row.gateway_cpu_cores for row in rows), + gateway_cpu_us_per_request=distribution(row.gateway_cpu_s * 1_000_000 / row.stats.requests for row in rows), + gateway_peak_rss_mib=distribution(row.gateway_peak_rss_bytes / 1024**2 for row in rows), + loadgen_cpu_cores=distribution(row.loadgen_cpu_cores for row in rows), + ) diff --git a/scripts/benchmarks/gateway_split/bifrost.patch b/scripts/benchmarks/gateway_split/bifrost.patch new file mode 100644 index 00000000000..4bea227f312 --- /dev/null +++ b/scripts/benchmarks/gateway_split/bifrost.patch @@ -0,0 +1,324 @@ +diff --git a/benchmark.go b/benchmark.go +index de26d8f..1dd03bd 100644 +--- a/benchmark.go ++++ b/benchmark.go +@@ -10,6 +10,7 @@ import ( + "log" + "net/http" + "os" ++ "sort" + "strconv" + "strings" + "sync" +@@ -29,6 +30,7 @@ import ( + // Provider represents an API provider to be benchmarked + // It holds the necessary information to target the provider's API. + type Provider struct { ++ ExtraHeaders http.Header + Name string // Name of the provider (e.g., "bifrost", "litellm") + Endpoint string // API endpoint path (e.g., "v1/chat/completions") + Port string // Port number the provider's server is listening on +@@ -41,6 +43,7 @@ type Provider struct { + + // BenchmarkResult holds the aggregated metrics from a single benchmark run for a provider. + type BenchmarkResult struct { ++ FirstEvents []time.Duration + ProviderName string // Name of the provider benchmarked + Metrics *vegeta.Metrics // Vegeta metrics (latency, success rate, etc.) + CPUUsage float64 // (Currently unused) Placeholder for CPU usage metrics +@@ -89,6 +92,8 @@ func main() { + rampUp := flag.Bool("ramp-up", false, "Enable gradual ramp-up of users (only with --users, requires --ramp-up-duration)") + rampUpDuration := flag.Int("ramp-up-duration", 0, "Duration in seconds to ramp up to target users (only with --users and --ramp-up)") + debug := flag.Bool("debug", false, "Enable debug mode with detailed logging and periodic status updates") ++ requestFile := flag.String("request-file", "", "Exact request JSON file") ++ headersFile := flag.String("headers-file", "", "Additional request headers as JSON") + + // Parse the command line flags. + flag.Parse() +@@ -136,6 +141,32 @@ func main() { + + // Initialize providers + providers := initializeProviders(*bigPayload, *model, *maxOutputTokens, *suffix, *path, *requestType, filePrompt, *host) ++ if *requestFile != "" { ++ payload, err := os.ReadFile(*requestFile) ++ if err != nil { ++ log.Fatal(err) ++ } ++ for i := range providers { ++ providers[i].Payload = payload ++ providers[i].PayloadTemplate = string(payload) ++ } ++ } ++ if *headersFile != "" { ++ data, err := os.ReadFile(*headersFile) ++ if err != nil { ++ log.Fatal(err) ++ } ++ var headers map[string]string ++ if err := sonic.Unmarshal(data, &headers); err != nil { ++ log.Fatal(err) ++ } ++ for i := range providers { ++ providers[i].ExtraHeaders = make(http.Header) ++ for key, value := range headers { ++ providers[i].ExtraHeaders.Set(key, value) ++ } ++ } ++ } + + // Filter providers if specific provider is requested + if *provider != "" { +@@ -392,6 +423,7 @@ func runBenchmarks(providers []Provider, rate int, users int, duration int, time + + // Run the benchmark based on mode + var metrics vegeta.Metrics ++ firstEvents := make([]time.Duration, 0) + + if users > 0 { + // Users mode: use concurrent package to maintain N concurrent requests +@@ -419,6 +451,19 @@ func runBenchmarks(providers []Provider, rate int, users int, duration int, time + metrics.Latencies.Mean = meanLatency + metrics.Latencies.Min = concurrentMetrics.MinLatency + metrics.Latencies.Max = concurrentMetrics.MaxLatency ++ latencies := make([]time.Duration, len(concurrentMetrics.Results)) ++ for i, result := range concurrentMetrics.Results { ++ latencies[i] = result.Latency ++ if result.Success && result.FirstEvent > 0 { ++ firstEvents = append(firstEvents, result.FirstEvent) ++ } ++ } ++ sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) ++ if len(latencies) > 0 { ++ metrics.Latencies.P50 = latencies[int(float64(len(latencies)-1)*0.50)] ++ metrics.Latencies.P95 = latencies[int(float64(len(latencies)-1)*0.95)] ++ metrics.Latencies.P99 = latencies[int(float64(len(latencies)-1)*0.99)] ++ } + + // Count status codes and failures + statusCodes := make(map[string]int) +@@ -435,8 +480,9 @@ func runBenchmarks(providers []Provider, rate int, users int, duration int, time + metrics.StatusCodes = statusCodes + + // Calculate request rate and throughput +- metrics.Rate = float64(concurrentMetrics.TotalRequests) / float64(duration) +- metrics.Throughput = metrics.Rate // Approximate as same as request rate ++ metrics.Duration = concurrentMetrics.Elapsed ++ metrics.Rate = float64(concurrentMetrics.TotalRequests) / concurrentMetrics.Elapsed.Seconds() ++ metrics.Throughput = float64(concurrentMetrics.SuccessCount) / concurrentMetrics.Elapsed.Seconds() + } else { + // Rate mode: use Vegeta with fixed RPS + attacker := vegeta.NewAttacker(vegeta.Client(httpClient)) +@@ -482,6 +528,7 @@ func runBenchmarks(providers []Provider, rate int, users int, duration int, time + // Add results + results = append(results, BenchmarkResult{ + ProviderName: provider.Name, ++ FirstEvents: firstEvents, + Metrics: &metrics, + ServerMemoryStats: serverMemStatsCopy, + DropReasons: dropReasons, +@@ -635,6 +682,10 @@ func createTargeter(provider Provider) vegeta.Targeter { + // "x-bf-vk": []string{"f452b625-a65e-4dfd-b48d-0ee3ba0e8d46"}, + } + ++ for key, values := range provider.ExtraHeaders { ++ tgt.Header[key] = values ++ } ++ + // Add Authorization header for OpenAI + if provider.Name == "OpenAI" { + openaiApiKey := os.Getenv("OPENAI_API_KEY") +@@ -678,6 +729,10 @@ func createConcurrentTargeter(provider Provider) func() (concurrent.Request, err + "Content-Type": []string{"application/json"}, + } + ++ for key, values := range provider.ExtraHeaders { ++ headers[key] = values ++ } ++ + // Add Authorization header for OpenAI + if provider.Name == "OpenAI" { + openaiApiKey := os.Getenv("OPENAI_API_KEY") +@@ -712,6 +767,11 @@ func createConcurrentTargeter(provider Provider) func() (concurrent.Request, err + // and memory values to megabytes for the output. + func saveResults(results []BenchmarkResult, outputFile string) { + type SerializableResult struct { ++ FirstEventP50Ms float64 `json:"first_event_p50_ms"` ++ FirstEventP95Ms float64 `json:"first_event_p95_ms"` ++ FirstEventP99Ms float64 `json:"first_event_p99_ms"` ++ P95LatencyMs float64 `json:"p95_latency_ms"` ++ ElapsedSeconds float64 `json:"elapsed_seconds"` + Requests uint64 `json:"requests"` + Rate float64 `json:"rate"` + SuccessRate float64 `json:"success_rate"` +@@ -768,13 +828,19 @@ func saveResults(results []BenchmarkResult, outputFile string) { + avgMem = float64(totalMem) / float64(len(res.ServerMemoryStats)) / (1024 * 1024) + } + ++ sort.Slice(res.FirstEvents, func(i, j int) bool { return res.FirstEvents[i] < res.FirstEvents[j] }) + resultsMap[strings.ToLower(res.ProviderName)] = SerializableResult{ + Requests: res.Metrics.Requests, ++ FirstEventP50Ms: float64(quantile(res.FirstEvents, 0.5)) / float64(time.Millisecond), ++ FirstEventP95Ms: float64(quantile(res.FirstEvents, 0.95)) / float64(time.Millisecond), ++ FirstEventP99Ms: float64(quantile(res.FirstEvents, 0.99)) / float64(time.Millisecond), + Rate: res.Metrics.Rate, + SuccessRate: 100.0 * res.Metrics.Success, + MeanLatencyMs: float64(res.Metrics.Latencies.Mean) / float64(time.Millisecond), + P50LatencyMs: float64(res.Metrics.Latencies.P50) / float64(time.Millisecond), + P99LatencyMs: float64(res.Metrics.Latencies.P99) / float64(time.Millisecond), ++ P95LatencyMs: float64(res.Metrics.Latencies.P95) / float64(time.Millisecond), ++ ElapsedSeconds: (res.Metrics.Duration + res.Metrics.Wait).Seconds(), + MaxLatencyMs: float64(res.Metrics.Latencies.Max) / float64(time.Millisecond), + ThroughputRPS: res.Metrics.Throughput, + Timestamp: time.Now().Format(time.RFC3339), +@@ -801,3 +867,10 @@ func saveResults(results []BenchmarkResult, outputFile string) { + + fmt.Printf("Results saved to %s\n", outputFile) + } ++ ++func quantile(values []time.Duration, fraction float64) time.Duration { ++ if len(values) == 0 { ++ return 0 ++ } ++ return values[int(float64(len(values)-1)*fraction)] ++} +diff --git a/pkg/concurrent/concurrent.go b/pkg/concurrent/concurrent.go +index 9bd5042..7c44548 100644 +--- a/pkg/concurrent/concurrent.go ++++ b/pkg/concurrent/concurrent.go +@@ -3,11 +3,13 @@ + package concurrent + + import ( ++ "bufio" + "bytes" + "context" + "fmt" + "io" + "net/http" ++ "strings" + "sync" + "time" + ) +@@ -22,6 +24,7 @@ type Request struct { + + // Result represents the outcome of a single request. + type Result struct { ++ FirstEvent time.Duration + StatusCode int + Latency time.Duration + Error string +@@ -30,15 +33,16 @@ type Result struct { + + // Metrics holds aggregated metrics from a concurrent benchmark run. + type Metrics struct { +- TotalRequests int +- SuccessCount int +- FailureCount int +- SuccessRate float64 +- Results []Result +- TotalLatency time.Duration +- MinLatency time.Duration +- MaxLatency time.Duration +- mu sync.Mutex ++ Elapsed time.Duration ++ TotalRequests int ++ SuccessCount int ++ FailureCount int ++ SuccessRate float64 ++ Results []Result ++ TotalLatency time.Duration ++ MinLatency time.Duration ++ MaxLatency time.Duration ++ mu sync.Mutex + } + + // Runner executes requests concurrently while maintaining a fixed number of in-flight requests. +@@ -66,7 +70,7 @@ func NewRunner(client *http.Client, numUsers int, duration time.Duration, reques + Results: make([]Result, 0), + }, + semaphore: make(chan struct{}, numUsers), +- debug: debug, ++ debug: debug, + } + } + +@@ -79,6 +83,7 @@ func (r *Runner) WithRampUp(rampUpDuration time.Duration) *Runner { + + // Run executes the concurrent request benchmark and returns metrics. + func (r *Runner) Run(ctx context.Context) *Metrics { ++ started := time.Now() + ctx, cancel := context.WithTimeout(ctx, r.duration) + defer cancel() + +@@ -100,6 +105,7 @@ func (r *Runner) Run(ctx context.Context) *Metrics { + + // Wait for all workers to complete + r.wg.Wait() ++ r.metrics.Elapsed = time.Since(started) + + // Calculate success rate + if r.metrics.TotalRequests > 0 { +@@ -218,7 +224,7 @@ func (r *Runner) worker(ctx context.Context) { + select { + case r.semaphore <- struct{}{}: + // Slot acquired, make request in background +- go r.makeRequest() ++ r.makeRequest() + case <-ctx.Done(): + return + } +@@ -263,23 +269,29 @@ func (r *Runner) makeRequest() { + // Make request and measure latency + start := time.Now() + resp, err := r.client.Do(httpReq) +- latency := time.Since(start) + + // Handle request error + if err != nil { + r.recordResult(Result{ + Success: false, + Error: fmt.Sprintf("request failed: %v", err), +- Latency: latency, ++ Latency: time.Since(start), + }) + return + } + defer resp.Body.Close() ++ firstEvent, readErr := readResponse(resp, start) ++ if readErr != nil { ++ r.recordResult(Result{Success: false, Error: fmt.Sprintf("response read failed: %v", readErr), Latency: time.Since(start)}) ++ return ++ } ++ latency := time.Since(start) + + // Record result + success := resp.StatusCode >= 200 && resp.StatusCode < 300 + r.recordResult(Result{ + StatusCode: resp.StatusCode, ++ FirstEvent: firstEvent, + Latency: latency, + Success: success, + }) +@@ -308,3 +320,22 @@ func (r *Runner) recordResult(result Result) { + + r.metrics.Results = append(r.metrics.Results, result) + } ++ ++func readResponse(response *http.Response, started time.Time) (time.Duration, error) { ++ if !strings.HasPrefix(response.Header.Get("Content-Type"), "text/event-stream") { ++ _, err := io.Copy(io.Discard, response.Body) ++ return 0, err ++ } ++ reader := bufio.NewReader(response.Body) ++ for { ++ line, err := reader.ReadString('\n') ++ if err != nil { ++ return 0, err ++ } ++ if line == "\n" || line == "\r\n" { ++ first := time.Since(started) ++ _, err := io.Copy(io.Discard, reader) ++ return first, err ++ } ++ } ++} diff --git a/scripts/benchmarks/gateway_split/bifrost_bench.py b/scripts/benchmarks/gateway_split/bifrost_bench.py new file mode 100644 index 00000000000..d5a137ebfa7 --- /dev/null +++ b/scripts/benchmarks/gateway_split/bifrost_bench.py @@ -0,0 +1,287 @@ +from __future__ import annotations + +import hashlib +import itertools +import json +import logging +import os +import random +import resource +import subprocess +import time +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Final, TextIO + +import httpx +import psutil +from pydantic import BaseModel, ConfigDict + +if TYPE_CHECKING: + from run import Variant + +LOGGER: Final = logging.getLogger(__name__) + + +class FrozenModel(BaseModel): + model_config = ConfigDict(frozen=True) + + +class MemoryStats(FrozenModel): + rss: int + + +class Stats(FrozenModel): + requests: int + rate: float + success_rate: float + mean_latency_ms: float + p50_latency_ms: float + p95_latency_ms: float + p99_latency_ms: float + max_latency_ms: float + elapsed_seconds: float + throughput_rps: float + first_event_p50_ms: float + first_event_p95_ms: float + first_event_p99_ms: float + status_code_counts: Mapping[str, int] + drop_reasons: Mapping[str, int] + + +class Result(FrozenModel): + litellm: Stats + + +class Counters(FrozenModel): + requests: int + active: int + completed: int + cancelled: int + + +class Settings(FrozenModel): + binary: Path + suite_root: Path + output: Path + seconds: int + repetitions: int + python_body: str + rust_client: str + python_client: str + http_parser: str + sizes: tuple[int, ...] + concurrencies: tuple[int, ...] + responses: tuple[bool, ...] = (False, True) + event_loop: str = "uvloop" + profile: str = "litellm-current" + seed: int = 20261019 + warmup_seconds: int = 2 + + +def counters(client: httpx.Client) -> Counters: + return Counters.model_validate_json(client.get("http://127.0.0.1:19400/metrics").content) + + +def processes_for(variant: Variant, roots: Mapping[str, psutil.Process]) -> tuple[psutil.Process, ...]: + selected: Final = tuple(roots[name] for name in variant.process_names) + return selected + tuple(itertools.chain.from_iterable(process.children(recursive=True) for process in selected)) + + +def cpu_seconds(processes: Sequence[psutil.Process]) -> float: + return sum(process.cpu_times().user + process.cpu_times().system for process in processes) + + +def execute( + args: tuple[str, ...], group: tuple[psutil.Process, ...], settings: Settings, port: int +) -> tuple[float, int, float, float]: + before: Final = cpu_seconds(group) + children_before: Final = resource.getrusage(resource.RUSAGE_CHILDREN) + started: Final = time.perf_counter() + with ( + (settings.output.parent / "runtime-bifrost.log").open("w") as log, + subprocess.Popen( + args, + cwd=settings.suite_root, + stdout=log, + stderr=log, + env=dict(os.environ) | {"LITELLM_PORT": str(port), "NO_PROXY": "127.0.0.1,localhost"}, + ) as process, + ): + samples: Final = tuple(sample_rss(process, group)) + status: Final = process.wait(timeout=40) + assert status == 0, (settings.output.parent / "runtime-bifrost.log").read_text() + wall: Final = time.perf_counter() - started + children_after: Final = resource.getrusage(resource.RUSAGE_CHILDREN) + loadgen_cpu: Final = ( + children_after.ru_utime + children_after.ru_stime - children_before.ru_utime - children_before.ru_stime + ) + return cpu_seconds(group) - before, max(samples, default=0), wall, loadgen_cpu + + +def sample_rss(process: subprocess.Popen[bytes], group: tuple[psutil.Process, ...]) -> Iterator[int]: + while process.poll() is None: + yield sum(MemoryStats.model_validate(item.memory_info(), from_attributes=True).rss for item in group) + time.sleep(0.1) + + +@dataclass(frozen=True, slots=True) +class Cell: + size: int + streaming: bool + concurrency: int + variant: Variant + + +@dataclass(frozen=True, slots=True) +class Trial: + cell: Cell + index: int + total: int + settings: Settings + group: tuple[psutil.Process, ...] + args: tuple[str, ...] + stats_path: Path + body_bytes: int + + +def run_bifrost(roots: Mapping[str, psutil.Process], variants: Sequence[Variant], settings: Settings) -> None: + runtime: Final = settings.output.resolve().parent / "bifrost-runtime" + runtime.mkdir(exist_ok=True) + cells: Final = tuple(itertools.product(settings.sizes, settings.responses, settings.concurrencies, variants)) + ordered: Final = random.Random(settings.seed).sample(cells, len(cells)) + with settings.output.open("w") as output, httpx.Client(trust_env=False, timeout=5) as client: + for cell_index, (size, streaming, concurrency, variant) in enumerate(ordered): + run_cell( + roots, + settings, + Cell(size, streaming, concurrency, variant), + runtime, + client, + output, + cell_index, + len(ordered), + ) + + +def run_cell( + roots: Mapping[str, psutil.Process], + settings: Settings, + cell: Cell, + runtime: Path, + client: httpx.Client, + output: TextIO, + cell_index: int, + total: int, +) -> None: + size, streaming, concurrency, variant = cell.size, cell.streaming, cell.concurrency, cell.variant + payload: Final = { + "model": "fixture-model", + "max_tokens": 32, + "stream": streaming, + "messages": [{"role": "user", "content": "q" * size}], + } + data: Final = json.dumps(payload, separators=(",", ":")).encode() + headers: Final = { + "x-bench-expected-sha256": hashlib.sha256(data).hexdigest(), + "x-bench-delay-ms": "5", + } | ({"x-bench-python": "1"} if variant.delegate else {}) + request_path: Final = runtime / "request.json" + header_path: Final = runtime / "headers.json" + stats_path: Final = runtime / "result.json" + request_path.write_bytes(data) + header_path.write_text(json.dumps(headers)) + group: Final = processes_for(variant, roots) + base_args: Final = ( + str(settings.binary), + "-provider", + "litellm", + "-host", + "127.0.0.1", + "-path", + "messages", + "-model", + "fixture-model", + "-max-output-tokens", + "32", + "-users", + str(concurrency), + "-request-file", + str(request_path), + "-headers-file", + str(header_path), + "-timeout", + str(settings.seconds + 30), + "-cooldown", + "0", + "-output", + str(stats_path), + ) + trial: Final = Trial(cell, cell_index, total, settings, group, base_args, stats_path, len(data)) + for repetition in range(-1, settings.repetitions): + run_trial(trial, repetition, client, output) + + +def run_trial(trial: Trial, repetition: int, client: httpx.Client, output: TextIO) -> None: + settings, group, base_args, stats_path = trial.settings, trial.group, trial.args, trial.stats_path + size, streaming, concurrency, variant = ( + trial.cell.size, + trial.cell.streaming, + trial.cell.concurrency, + trial.cell.variant, + ) + cell_index, total = trial.index, trial.total + before: Final = counters(client) + seconds: Final = settings.warmup_seconds if repetition < 0 else settings.seconds + cpu, rss, wall, loadgen_cpu = execute(base_args + ("-duration", str(seconds)), group, settings, variant.port) + stats: Final = Result.model_validate_json(stats_path.read_bytes()).litellm + after: Final = counters(client) + assert stats.requests > 0 + assert stats.requests * stats.mean_latency_ms / 1000 <= concurrency * stats.elapsed_seconds + 1e-6 + assert stats.status_code_counts.get("200", 0) == stats.requests + assert after.requests - before.requests == stats.requests + assert after.cancelled == before.cancelled + assert after.active == before.active + assert 0 < stats.p50_latency_ms <= stats.p95_latency_ms <= stats.p99_latency_ms <= stats.max_latency_ms + if streaming: + assert after.completed - before.completed == stats.requests + assert 0 < stats.first_event_p50_ms <= stats.p50_latency_ms + if repetition < 0: + return + row: Final = { + "generator": "bifrost", + "variant": variant.name, + "prompt_bytes": size, + "body_bytes": trial.body_bytes, + "stream": streaming, + "concurrency": concurrency, + "repetition": repetition, + "python_body": settings.python_body, + "python_client": settings.python_client, + "rust_client": settings.rust_client, + "http_parser": settings.http_parser, + "event_loop": settings.event_loop, + "profile": settings.profile, + "gateway_cpu_s": cpu, + "gateway_cpu_cores": cpu / wall, + "gateway_peak_rss_bytes": rss, + "loadgen_cpu_cores": loadgen_cpu / wall, + "stats": stats.model_dump(), + "backend_completed": after.completed - before.completed, + "backend_requests": after.requests - before.requests, + "backend_cancelled": after.cancelled - before.cancelled, + } + output.write(json.dumps(row) + "\n") + output.flush() + LOGGER.info( + "%s/%s %s c=%s size=%s stream=%s rep=%s %.0f rps", + cell_index + 1, + total, + variant.name, + concurrency, + size, + streaming, + repetition, + stats.throughput_rps, + ) diff --git a/scripts/benchmarks/gateway_split/bootstrap.sh b/scripts/benchmarks/gateway_split/bootstrap.sh new file mode 100755 index 00000000000..fdf8b62f10e --- /dev/null +++ b/scripts/benchmarks/gateway_split/bootstrap.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +set -euo pipefail + +bench_root="$(cd "$(dirname "$0")" && pwd)" +tools_dir="${BENCH_TOOLS_DIR:-$bench_root/runtime/tools}" +mkdir -p "$tools_dir" + +download() { + local url="$1" destination="$2" checksum="$3" + curl --fail --location "$url" --output "$destination" + printf '%s %s\n' "$checksum" "$destination" | shasum -a 256 --check - +} + +case "$(uname -s)-$(uname -m)" in + Darwin-arm64) go_target=darwin-arm64; go_sha=76812b213b1b2302c978d28fa52fa92d541704b9e7d9d5db8002c50e4018c4c5 ;; + Darwin-x86_64) go_target=darwin-amd64; go_sha=587b59182488b23aa6e5fc25110405a3e0e5b38ed2f5b2f46ed13c32aee356fe ;; + Linux-aarch64) go_target=linux-arm64; go_sha=94f3e30b8e374bc285e7dadc11e0865726b9bc6e85b841ccceaabc0214c6b7c8 ;; + Linux-x86_64) go_target=linux-amd64; go_sha=ecbadb99091a3f46e31f5f934b068b1864eafa7995211b39eaddf76996045fe5 ;; + *) printf 'Unsupported Go platform\n' >&2; exit 1 ;; +esac +download "https://go.dev/dl/go1.27.2.$go_target.tar.gz" "$tools_dir/go.tar.gz" "$go_sha" +tar -xzf "$tools_dir/go.tar.gz" -C "$tools_dir" +suite_commit=3cb72a95cacaee6f73f9db5636a1a651c7dcb1e9 +if [ ! -d "$tools_dir/bifrost-benchmarking/.git" ]; then + git clone https://github.com/maximhq/bifrost-benchmarking.git "$tools_dir/bifrost-benchmarking" +fi +git -C "$tools_dir/bifrost-benchmarking" checkout --detach "$suite_commit" +if git -C "$tools_dir/bifrost-benchmarking" apply --check "$bench_root/bifrost.patch"; then + git -C "$tools_dir/bifrost-benchmarking" apply "$bench_root/bifrost.patch" +else + git -C "$tools_dir/bifrost-benchmarking" apply --reverse --check "$bench_root/bifrost.patch" +fi +touch "$tools_dir/bifrost-benchmarking/.env" +( + cd "$tools_dir/bifrost-benchmarking" + CGO_ENABLED=0 "$tools_dir/go/bin/go" build -trimpath -mod=readonly -o "$tools_dir/suite" benchmark.go +) +( + cd "$bench_root" + cargo build --release --locked +) diff --git a/scripts/benchmarks/gateway_split/cloud.py b/scripts/benchmarks/gateway_split/cloud.py new file mode 100644 index 00000000000..b5a8f1e9198 --- /dev/null +++ b/scripts/benchmarks/gateway_split/cloud.py @@ -0,0 +1,222 @@ +import argparse +import hashlib +import itertools +import json +import logging +import signal +from collections.abc import Mapping +from pathlib import Path +from typing import Final, Literal + +from analyze_bifrost import Record, records, summarize +from bifrost_bench import FrozenModel +from matrix import TOPOLOGIES, Job, jobs, shard +from pydantic import TypeAdapter +from run import ROOT, Options, run, stop + +LOGGER: Final = logging.getLogger(__name__) + + +class CLI(FrozenModel): + command: Literal["plan", "run", "merge"] + manifest: Path + output: Path + inputs: Path + bifrost: Path + suite_root: Path + job: str | None + shard_index: int + shard_count: int + machine_class: str | None + + +class Manifest(FrozenModel): + schema_version: Literal[1] = 1 + jobs: tuple[Job, ...] + + +class Metadata(FrozenModel): + platform: str + cpu: str + logical_cpus: int + memory_bytes: int + machine_class: str + python: str + rust: str + git_sha: str + versions: Mapping[str, str] + harness_sha256: Mapping[str, str] + bifrost_binary_sha256: str + bifrost_patch_sha256: str + resolved_http_parser: str + dependency_lock_sha256: str + parameters: Options + + +def check_job(job: Job, path: Path) -> tuple[Record, ...]: + rows: Final = tuple(records((path,))) + expected: Final = frozenset(itertools.product(TOPOLOGIES, range(job.repetitions))) + actual: Final = frozenset((row.variant, row.repetition) for row in rows) + if len(rows) != job.expected_rows or actual != expected: + raise ValueError(f"Incomplete or duplicate trials for {job.id}") + if not all( + row.profile == job.profile + and row.prompt_bytes == job.prompt_bytes + and row.stream == (job.response == "sse") + and row.concurrency == job.concurrency + for row in rows + ): + raise ValueError(f"Workload differs from manifest for {job.id}") + metadata: Final = Metadata.model_validate_json(path.with_suffix(".metadata.json").read_bytes()) + options: Final = metadata.parameters + if (options.profile, options.sizes, options.responses, options.concurrencies) != ( + job.profile, + (job.prompt_bytes,), + (job.response == "sse",), + (job.concurrency,), + ): + raise ValueError(f"Parameters differ from manifest for {job.id}") + if (options.seconds, options.repetitions, options.warmup, options.seed) != ( + job.seconds, + job.repetitions, + job.warmup_seconds, + job.seed, + ): + raise ValueError(f"Timing differs from manifest for {job.id}") + probes: Final = TypeAdapter(tuple[dict[str, object], ...]).validate_json( + path.with_suffix(".probes.json").read_bytes() + ) + if {probe.get("variant") for probe in probes} != set(TOPOLOGIES): + raise ValueError(f"Missing topology probes for {job.id}") + for probe in probes: + if not all( + probe.get(key) is True + for key in ( + "body_digest", + "anthropic_sdk_messages", + "slow_upload_verified", + "headers", + "status_passthrough", + "cancelled_upstream", + ) + ): + raise ValueError(f"Failed correctness probe for {job.id}") + return rows + + +def compatibility(metadata: Metadata) -> str: + return json.dumps(metadata.model_dump(exclude={"parameters"}), sort_keys=True) + + +def merge(manifest: Manifest, source: Path, output: Path) -> None: + paths: Final = tuple(source / job.id / "trials.jsonl" for job in manifest.jobs) + rows: Final = tuple(itertools.chain.from_iterable(check_job(job, path) for job, path in zip(manifest.jobs, paths))) + metadata: Final = tuple( + Metadata.model_validate_json(path.with_suffix(".metadata.json").read_bytes()) for path in paths + ) + if len(frozenset(compatibility(item) for item in metadata)) != 1: + raise ValueError("Cannot combine different hardware classes, revisions, dependencies or benchmark binaries") + summaries: Final = tuple( + summarize( + tuple( + row + for row in rows + if row.variant == variant + and row.profile == job.profile + and row.prompt_bytes == job.prompt_bytes + and row.stream == (job.response == "sse") + and row.concurrency == job.concurrency + ), + job.repetitions, + ) + for job, variant in itertools.product(manifest.jobs, TOPOLOGIES) + ) + output.mkdir(parents=True, exist_ok=True) + (output / "summary.json").write_text(json.dumps(tuple(item.model_dump() for item in summaries), indent=2) + "\n") + (output / "trials.jsonl").write_text("".join(row.model_dump_json() + "\n" for row in rows)) + (output / "manifest.json").write_text(manifest.model_dump_json(indent=2) + "\n") + LOGGER.info("Validated %s jobs, %s trials, four topologies per workload", len(manifest.jobs), len(rows)) + + +def execute(job: Job, options: Options) -> None: + if options.output.exists(): + check_job(job, options.output) + metadata: Final = Metadata.model_validate_json(options.output.with_suffix(".metadata.json").read_bytes()) + current: Final = {ROOT / name: digest for name, digest in metadata.harness_sha256.items()} + if not all(hashlib.sha256(path.read_bytes()).hexdigest() == digest for path, digest in current.items()): + raise ValueError("Existing output belongs to different benchmark sources") + if metadata.bifrost_binary_sha256 != hashlib.sha256(options.bifrost.read_bytes()).hexdigest(): + raise ValueError("Existing output belongs to a different Bifrost binary") + if metadata.parameters != options: + raise ValueError("Existing output belongs to different run parameters") + LOGGER.info("Already complete %s", job.id) + return + run(options) + check_job(job, options.output) + + +def main() -> None: + parser: Final = argparse.ArgumentParser() + parser.add_argument("command", choices=("plan", "run", "merge")) + parser.add_argument("--manifest", type=Path, default=ROOT / "matrix.json") + parser.add_argument("--output", type=Path, default=ROOT / "results/cloud") + parser.add_argument("--inputs", type=Path, default=ROOT / "results/cloud") + parser.add_argument("--bifrost", type=Path, default=ROOT / "runtime/tools/suite") + parser.add_argument("--suite-root", type=Path, default=ROOT / "runtime/tools/bifrost-benchmarking") + parser.add_argument("--job") + parser.add_argument("--shard-index", type=int, default=0) + parser.add_argument("--shard-count", type=int, default=1) + parser.add_argument("--machine-class") + raw: Final = CLI.model_validate(vars(parser.parse_args())) + command: Final = TypeAdapter(str).validate_python(raw.command) + manifest_path: Final = TypeAdapter(Path).validate_python(raw.manifest) + output: Final = TypeAdapter(Path).validate_python(raw.output).resolve() + if command == "plan": + manifest_path.write_text(Manifest(jobs=jobs()).model_dump_json(indent=2) + "\n") + LOGGER.info("Wrote 18 jobs, 216 measured trials to %s", manifest_path) + return + manifest: Final = Manifest.model_validate_json(manifest_path.read_bytes()) + if len(frozenset(job.id for job in manifest.jobs)) != len(manifest.jobs): + raise ValueError("Manifest contains duplicate job IDs") + if command == "merge": + merge(manifest, TypeAdapter(Path).validate_python(raw.inputs), output) + return + machine_class: Final = TypeAdapter(str).validate_python(raw.machine_class) + job_id: Final = raw.job + selected: Final = ( + tuple(job for job in manifest.jobs if job.id == job_id) + if job_id + else shard( + manifest.jobs, + TypeAdapter(int).validate_python(raw.shard_index), + TypeAdapter(int).validate_python(raw.shard_count), + ) + ) + if not selected: + raise ValueError("No jobs selected") + for job in selected: + LOGGER.info("Running %s", job.id) + execute( + job, + Options( + bifrost=TypeAdapter(Path).validate_python(raw.bifrost).resolve(), + suite_root=TypeAdapter(Path).validate_python(raw.suite_root).resolve(), + output=output / job.id / "trials.jsonl", + profile=job.profile, + sizes=(job.prompt_bytes,), + concurrencies=(job.concurrency,), + responses=(job.response == "sse",), + seconds=job.seconds, + repetitions=job.repetitions, + warmup=job.warmup_seconds, + seed=job.seed, + machine_class=machine_class, + ), + ) + + +if __name__ == "__main__": + signal.signal(signal.SIGTERM, stop) + logging.basicConfig(level=logging.INFO, format="%(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + main() diff --git a/scripts/benchmarks/gateway_split/headers.py b/scripts/benchmarks/gateway_split/headers.py new file mode 100644 index 00000000000..4f267836fc5 --- /dev/null +++ b/scripts/benchmarks/gateway_split/headers.py @@ -0,0 +1,21 @@ +from collections.abc import Mapping +from typing import Final + +HOP_HEADERS: Final = frozenset( + ( + "host", + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "transfer-encoding", + "upgrade", + ) +) + + +def clean_headers(headers: Mapping[str, str]) -> Mapping[str, str]: + connection: Final = frozenset(token.strip().lower() for token in headers.get("connection", "").split(",")) + return {key: value for key, value in headers.items() if key.lower() not in HOP_HEADERS | connection} diff --git a/scripts/benchmarks/gateway_split/matrix.json b/scripts/benchmarks/gateway_split/matrix.json new file mode 100644 index 00000000000..851acd6fbf7 --- /dev/null +++ b/scripts/benchmarks/gateway_split/matrix.json @@ -0,0 +1,203 @@ +{ + "schema_version": 1, + "jobs": [ + { + "profile": "litellm-current", + "prompt_bytes": 262144, + "response": "buffered", + "concurrency": 1, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-262144-buffered-c1" + }, + { + "profile": "litellm-current", + "prompt_bytes": 262144, + "response": "buffered", + "concurrency": 1000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-262144-buffered-c1000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 262144, + "response": "buffered", + "concurrency": 2000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-262144-buffered-c2000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 262144, + "response": "sse", + "concurrency": 1, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-262144-sse-c1" + }, + { + "profile": "litellm-current", + "prompt_bytes": 262144, + "response": "sse", + "concurrency": 1000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-262144-sse-c1000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 262144, + "response": "sse", + "concurrency": 2000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-262144-sse-c2000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 1048576, + "response": "buffered", + "concurrency": 1, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-1048576-buffered-c1" + }, + { + "profile": "litellm-current", + "prompt_bytes": 1048576, + "response": "buffered", + "concurrency": 1000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-1048576-buffered-c1000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 1048576, + "response": "buffered", + "concurrency": 2000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-1048576-buffered-c2000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 1048576, + "response": "sse", + "concurrency": 1, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-1048576-sse-c1" + }, + { + "profile": "litellm-current", + "prompt_bytes": 1048576, + "response": "sse", + "concurrency": 1000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-1048576-sse-c1000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 1048576, + "response": "sse", + "concurrency": 2000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-1048576-sse-c2000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 4194304, + "response": "buffered", + "concurrency": 1, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-4194304-buffered-c1" + }, + { + "profile": "litellm-current", + "prompt_bytes": 4194304, + "response": "buffered", + "concurrency": 1000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-4194304-buffered-c1000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 4194304, + "response": "buffered", + "concurrency": 2000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-4194304-buffered-c2000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 4194304, + "response": "sse", + "concurrency": 1, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-4194304-sse-c1" + }, + { + "profile": "litellm-current", + "prompt_bytes": 4194304, + "response": "sse", + "concurrency": 1000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-4194304-sse-c1000" + }, + { + "profile": "litellm-current", + "prompt_bytes": 4194304, + "response": "sse", + "concurrency": 2000, + "seconds": 60, + "repetitions": 3, + "warmup_seconds": 5, + "seed": 20261019, + "id": "litellm-current-4194304-sse-c2000" + } + ] +} diff --git a/scripts/benchmarks/gateway_split/matrix.py b/scripts/benchmarks/gateway_split/matrix.py new file mode 100644 index 00000000000..fc360b5fe39 --- /dev/null +++ b/scripts/benchmarks/gateway_split/matrix.py @@ -0,0 +1,56 @@ +import itertools +from typing import Final, Literal + +from bifrost_bench import FrozenModel +from pydantic import Field, computed_field + +TOPOLOGIES: Final = ("direct_axum", "direct_fastapi", "axum_python", "fastapi_rust") + + +PROFILE_NAMES: Final = ("litellm-current",) + + +class Job(FrozenModel): + profile: Literal["litellm-current"] + prompt_bytes: Literal[262144, 1048576, 4194304] + response: Literal["buffered", "sse"] + concurrency: Literal[1, 1000, 2000] + seconds: int = Field(default=60, ge=1) + repetitions: int = Field(default=3, ge=1) + warmup_seconds: int = Field(default=5, ge=1) + seed: int = 20261019 + + @computed_field + @property + def id(self) -> str: + return f"{self.profile}-{self.prompt_bytes}-{self.response}-c{self.concurrency}" + + @property + def expected_rows(self) -> int: + return len(TOPOLOGIES) * self.repetitions + + +def jobs(seconds: int = 60, repetitions: int = 3, warmup_seconds: int = 5) -> tuple[Job, ...]: + combinations: Final = itertools.product( + PROFILE_NAMES, (262144, 1048576, 4194304), ("buffered", "sse"), (1, 1000, 2000) + ) + return tuple( + Job.model_validate( + { + "profile": profile, + "prompt_bytes": size, + "response": response, + "concurrency": concurrency, + "seconds": seconds, + "repetitions": repetitions, + "warmup_seconds": warmup_seconds, + } + ) + for profile, size, response, concurrency in combinations + ) + + +def shard(selected: tuple[Job, ...], index: int, count: int) -> tuple[Job, ...]: + if count < 1 or not 0 <= index < count: + raise ValueError("Shard index must be within the positive shard count") + return tuple(job for position, job in enumerate(selected) if position % count == index) diff --git a/scripts/benchmarks/gateway_split/messages_probe.py b/scripts/benchmarks/gateway_split/messages_probe.py new file mode 100644 index 00000000000..d61c690dd6f --- /dev/null +++ b/scripts/benchmarks/gateway_split/messages_probe.py @@ -0,0 +1,40 @@ +from typing import Final + +from anthropic import Anthropic + + +def verify_sdk(port: int, delegate: bool) -> None: + with Anthropic( + base_url=f"http://127.0.0.1:{port}", + api_key="fixture-key", + max_retries=0, + timeout=5, + default_headers={"x-bench-python": "1"} if delegate else {}, + ) as client: + response: Final = client.messages.create( + model="fixture-model", + max_tokens=32, + system=[{"type": "text", "text": "Use the provided tool result", "cache_control": {"type": "ephemeral"}}], + tools=[{"name": "lookup", "description": "Look up a value", "input_schema": {"type": "object"}}], + messages=[ + {"role": "user", "content": "Find the answer"}, + { + "role": "assistant", + "content": [{"type": "tool_use", "id": "tool_fixture", "name": "lookup", "input": {}}], + }, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "tool_fixture", "content": "42"}]}, + ], + ) + assert response.type == "message" + assert response.stop_reason == "end_turn" + assert len(response.content) == 1 + assert response.content[0].type == "text" + assert response.content[0].text == "x" * 1024 + with client.messages.stream( + model="fixture-model", max_tokens=32, messages=[{"role": "user", "content": "hello"}] + ) as stream: + text: Final = "".join(stream.text_stream) + final: Final = stream.get_final_message() + assert text == response.content[0].text + assert final.stop_reason == response.stop_reason + assert final.usage == response.usage diff --git a/scripts/benchmarks/gateway_split/python_app.py b/scripts/benchmarks/gateway_split/python_app.py new file mode 100644 index 00000000000..5a63201f02d --- /dev/null +++ b/scripts/benchmarks/gateway_split/python_app.py @@ -0,0 +1,70 @@ +import os +from collections.abc import AsyncGenerator, AsyncIterator +from contextlib import asynccontextmanager +from typing import Final + +import httpx +from fastapi import FastAPI, Request +from headers import clean_headers +from starlette.responses import Response, StreamingResponse + +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler + +BACKEND: Final = os.environ.get("BENCH_BACKEND", "http://127.0.0.1:19400") +RUST: Final = os.environ.get("BENCH_RUST", "http://127.0.0.1:19401") +DISPATCH: Final = os.environ.get("BENCH_DISPATCH") == "1" +STREAM_REQUESTS: Final = os.environ.get("BENCH_STREAM_REQUESTS") == "1" + + +def create_app() -> FastAPI: + handler: Final = AsyncHTTPHandler(timeout=httpx.Timeout(30.0, connect=1.0), follow_redirects=False) + client: Final = handler.client + + @asynccontextmanager + async def lifespan(app: FastAPI) -> AsyncGenerator[None]: + yield + await client.aclose() + + app: Final = FastAPI(lifespan=lifespan) + + async def forward(request: Request, path: str) -> Response: + data: Final = request.stream() if STREAM_REQUESTS else await request.body() + if isinstance(data, bytes) and len(data) > 8 * 1024 * 1024: + return Response(status_code=413) + if int(request.headers.get("content-length", "0")) > 8 * 1024 * 1024: + return Response(status_code=413) + target: Final = ( + RUST if DISPATCH and path == "v1/messages" and "x-bench-python" not in request.headers else BACKEND + ) + url: Final = f"{target}/{path}" + (f"?{request.url.query}" if request.url.query else "") + upstream_request: Final = client.build_request( + request.method, + url, + headers=clean_headers(httpx.Headers(request.headers)), + content=data, + ) + try: + upstream: Final = await client.send(upstream_request, stream=True) + except httpx.HTTPError: + return Response(status_code=502) + + async def chunks() -> AsyncIterator[bytes]: + try: + async for chunk in upstream.aiter_raw(): + yield chunk + finally: + await upstream.aclose() + + return StreamingResponse( + chunks(), + status_code=upstream.status_code, + headers={ + key: value for key, value in clean_headers(upstream.headers).items() if key not in ("date", "server") + }, + ) + + app.add_api_route("/{path:path}", forward, methods=["GET", "POST"]) + return app + + +app: Final = create_app() diff --git a/scripts/benchmarks/gateway_split/run.py b/scripts/benchmarks/gateway_split/run.py new file mode 100644 index 00000000000..d99d07cf20d --- /dev/null +++ b/scripts/benchmarks/gateway_split/run.py @@ -0,0 +1,336 @@ +import argparse +import fcntl +import hashlib +import importlib.metadata +import importlib.util +import json +import logging +import os +import platform +import signal +import socket +import subprocess +import sys +import tempfile +import time +from collections.abc import Generator, Iterator, Mapping, Sequence +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass +from pathlib import Path +from types import FrameType +from typing import Final, Literal + +import httpx +import psutil +from bifrost_bench import Counters, FrozenModel, Settings, run_bifrost +from matrix import PROFILE_NAMES +from messages_probe import verify_sdk +from pydantic import Field + +ROOT: Final = Path(__file__).resolve().parent +BINARY: Final = ROOT / "target/release/gateway-split-bench" +LOGGER: Final = logging.getLogger(__name__) + + +class Metrics(Counters): + upload_bytes: int = 0 + + +class Options(FrozenModel): + bifrost: Path = ROOT / "runtime/tools/suite" + suite_root: Path = ROOT / "runtime/tools/bifrost-benchmarking" + output: Path = ROOT / "results/smoke/trials.jsonl" + profile: str = "litellm-current" + sizes: tuple[int, ...] = (262144, 1048576, 4194304) + concurrencies: tuple[int, ...] = (1, 1000, 2000) + responses: tuple[bool, ...] = (False, True) + seconds: int = Field(default=60, ge=1) + repetitions: int = Field(default=3, ge=1) + warmup: int = Field(default=5, ge=1) + seed: int = 20261019 + machine_class: str = "local" + + +@dataclass(frozen=True, slots=True) +class Variant: + name: str + port: int + delegate: bool + process_names: tuple[str, ...] + + +VARIANTS: Final = ( + Variant("direct_axum", 19401, False, ("axum",)), + Variant("direct_fastapi", 19402, False, ("fastapi",)), + Variant("axum_python", 19403, True, ("axum_dispatch", "fastapi")), + Variant("fastapi_rust", 19404, False, ("python_dispatch", "axum")), +) + + +@contextmanager +def launch(name: str, args: Sequence[str], env: Mapping[str, str]) -> Generator[psutil.Process]: + with (ROOT / "runtime" / f"{name}.log").open("w") as log: + process: Final = subprocess.Popen(args, cwd=ROOT, env=env, stdout=log, stderr=log, start_new_session=True) + try: + yield psutil.Process(process.pid) + finally: + if process.poll() is None: + os.killpg(process.pid, signal.SIGTERM) + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=5) + + +def probe_ready(port: int) -> bool: + try: + with httpx.Client(trust_env=False, timeout=1) as client: + return client.get(f"http://127.0.0.1:{port}/metrics").is_success + except httpx.HTTPError: + return False + + +def ready(port: int) -> None: + for _ in range(100): + if probe_ready(port): + return + time.sleep(0.1) + sys.exit(f"Listener {port} failed, inspect runtime logs") + + +def require_free_ports() -> None: + for port in range(19400, 19405): + with socket.socket() as listener: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + listener.bind(("127.0.0.1", port)) + except OSError: + sys.exit(f"Port {port} is occupied, stop the conflicting listener before running") + + +def metrics(client: httpx.Client) -> Metrics: + return Metrics.model_validate_json(client.get("http://127.0.0.1:19400/metrics").content) + + +def verify(variant: Variant, client: httpx.Client, python_body: str) -> Mapping[str, str | int | float | bool]: + verify_sdk(variant.port, variant.delegate) + headers: Final = {"x-bench-id": "probe", "content-type": "application/json"} | ( + {"x-bench-python": "1"} if variant.delegate else {} + ) + payload: Final = {"model": "fixture-model", "max_tokens": 32, "messages": [{"role": "user", "content": "hello"}]} + data: Final = json.dumps(payload).encode() + url: Final = f"http://127.0.0.1:{variant.port}/v1/messages?beta=true" + response: Final = client.post(url, content=data, headers=headers) + assert response.status_code == 200 + assert response.headers["x-body-sha256"] == hashlib.sha256(data).hexdigest() + assert response.headers["x-bench-id"] == "probe" + before_upload: Final = metrics(client) + + def upload() -> Iterator[bytes]: + yield data[:1] + time.sleep(0.1) + observed: Final = metrics(client).upload_bytes - before_upload.upload_bytes + assert (observed > 0) == (variant.name == "direct_axum" or python_body == "stream") + yield data[1:] + + upload_response: Final = client.post( + url, content=upload(), headers=headers | {"x-bench-upload-probe": "1", "content-length": str(len(data))} + ) + assert upload_response.status_code == 200 + assert upload_response.headers["x-body-sha256"] == hashlib.sha256(data).hexdigest() + before_errors: Final = metrics(client) + for status in (429, 500): + verify_error(client, url, data, headers, status) + after_errors: Final = metrics(client) + assert after_errors.requests - before_errors.requests == 2 + before_stream: Final = metrics(client) + start: Final = time.perf_counter() + with client.stream( + "POST", + url, + json=payload | {"stream": True}, + headers=headers | {"x-bench-events": "10000", "x-bench-delay-ms": "10"}, + ) as stream: + assert stream.status_code == 200 + first: Final = next(line for line in stream.iter_lines() if line.startswith("data:")) + assert "message_start" in first + time.sleep(0.25) + after_stream: Final = metrics(client) + assert after_stream.active == before_stream.active + assert after_stream.cancelled - before_stream.cancelled == 1 + return { + "variant": variant.name, + "body_digest": True, + "anthropic_sdk_messages": True, + "slow_upload_verified": True, + "headers": True, + "status_passthrough": True, + "error_backend_attempts": after_errors.requests - before_errors.requests, + "cancelled_upstream": True, + "cancel_observation_ms": (time.perf_counter() - start) * 1000, + } + + +def verify_error(client: httpx.Client, url: str, data: bytes, headers: Mapping[str, str], status: int) -> None: + failed: Final = client.post(url, content=data, headers=dict(headers) | {"x-bench-status": str(status)}) + assert failed.status_code == status + assert failed.text == "injected failure" + + +class HostMemory(FrozenModel): + total: int + + +class CLI(Options): + response: Literal["both", "buffered", "sse"] + + +def run(options: Options) -> None: + (ROOT / "runtime").mkdir(exist_ok=True) + options.output.parent.mkdir(parents=True, exist_ok=True) + with (Path(tempfile.gettempdir()) / "litellm-gateway-benchmark.lock").open("a+") as lock, ExitStack() as stack: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + require_free_ports() + env: Final = dict(os.environ) | { + "PYTHONPATH": os.pathsep.join((str(ROOT), str(ROOT.parents[2]))), + "LITELLM_LOCAL_MODEL_COST_MAP": "True", + "LITELLM_HTTP2": "False", + "DISABLE_AIOHTTP_TRANSPORT": "False", + "NO_PROXY": "127.0.0.1,localhost", + "BENCH_STREAM_REQUESTS": "0", + } + + def start(name: str, args: Sequence[str], dispatch: bool = False) -> psutil.Process: + return stack.enter_context(launch(name, args, env | {"BENCH_DISPATCH": "1" if dispatch else "0"})) + + def python_server(name: str, port: int, dispatch: bool = False) -> psutil.Process: + return start( + name, + ( + sys.executable, + "-m", + "uvicorn", + "python_app:app", + "--host", + "127.0.0.1", + "--port", + str(port), + "--loop", + "uvloop", + "--http", + "auto", + "--workers", + "1", + "--no-access-log", + "--timeout-keep-alive", + "60", + ), + dispatch, + ) + + processes: Final = { + "backend": start("backend", (str(BINARY), "serve", "--port", "19400")), + "axum": start("axum", (str(BINARY), "serve", "--role", "stream", "--port", "19401")), + "fastapi": python_server("fastapi", 19402), + "axum_dispatch": start("axum_dispatch", (str(BINARY), "serve", "--role", "stream", "--port", "19403")), + "python_dispatch": python_server("python_dispatch", 19404, True), + } + for port in range(19400, 19405): + ready(port) + with httpx.Client(trust_env=False, timeout=5) as client: + probes: Final = tuple(verify(variant, client, "buffered") for variant in VARIANTS) + options.output.with_suffix(".probes.json").write_text(json.dumps(probes, indent=2) + "\n") + metadata: Final = { + "platform": platform.platform(), + "cpu": platform.processor(), + "logical_cpus": psutil.cpu_count(), + "memory_bytes": HostMemory.model_validate(psutil.virtual_memory(), from_attributes=True).total, + "machine_class": options.machine_class, + "python": sys.version, + "git_sha": subprocess.check_output(("git", "rev-parse", "HEAD"), text=True).strip(), + "versions": { + name: importlib.metadata.version(name) + for name in ( + "fastapi", + "starlette", + "uvicorn", + "uvloop", + "httpx", + "aiohttp", + "psutil", + "pydantic", + ) + }, + "rust": subprocess.check_output(("rustc", "--version"), text=True).strip(), + "parameters": options.model_dump(mode="json"), + "resolved_http_parser": "httptools" if importlib.util.find_spec("httptools") else "h11", + "dependency_lock_sha256": hashlib.sha256((ROOT.parents[2] / "uv.lock").read_bytes()).hexdigest(), + "harness_sha256": { + str(path.relative_to(ROOT)): hashlib.sha256(path.read_bytes()).hexdigest() + for path in tuple(ROOT.glob("*.py")) + + tuple((ROOT / "src").glob("*.rs")) + + (ROOT / "Cargo.lock", ROOT / "Cargo.toml") + }, + "bifrost_binary_sha256": hashlib.sha256(options.bifrost.read_bytes()).hexdigest(), + "bifrost_patch_sha256": hashlib.sha256((ROOT / "bifrost.patch").read_bytes()).hexdigest(), + } + options.output.with_suffix(".metadata.json").write_text(json.dumps(metadata, indent=2) + "\n") + run_bifrost( + processes, + VARIANTS, + Settings( + binary=options.bifrost.resolve(), + suite_root=options.suite_root.resolve(), + output=options.output, + seconds=options.seconds, + repetitions=options.repetitions, + warmup_seconds=options.warmup, + python_body="buffered", + rust_client="hyper-stream", + python_client="litellm-aiohttp", + http_parser="httptools" if importlib.util.find_spec("httptools") else "h11", + event_loop="uvloop", + profile=options.profile, + seed=options.seed, + sizes=options.sizes, + concurrencies=options.concurrencies, + responses=options.responses, + ), + ) + + +def stop(signum: int, frame: FrameType | None) -> None: + sys.exit(128 + signum) + + +def main() -> None: + parser: Final = argparse.ArgumentParser() + parser.add_argument("--bifrost", type=Path, default=ROOT / "runtime/tools/suite") + parser.add_argument("--suite-root", type=Path, default=ROOT / "runtime/tools/bifrost-benchmarking") + parser.add_argument("--output", type=Path, default=ROOT / "results/smoke/trials.jsonl") + parser.add_argument("--profile", choices=PROFILE_NAMES, default="litellm-current") + parser.add_argument( + "--sizes", type=int, nargs="+", choices=(262144, 1048576, 4194304), default=(262144, 1048576, 4194304) + ) + parser.add_argument("--concurrencies", type=int, nargs="+", choices=(1, 1000, 2000), default=(1, 1000, 2000)) + parser.add_argument("--response", choices=("both", "buffered", "sse"), default="both") + parser.add_argument("--seconds", type=int, default=60) + parser.add_argument("--repetitions", type=int, default=3) + parser.add_argument("--warmup", type=int, default=5) + parser.add_argument("--seed", type=int, default=20261019) + parser.add_argument("--machine-class", default="local") + raw: Final = CLI.model_validate(vars(parser.parse_args())) + response: Final = raw.response + options: Final = Options.model_validate( + raw.model_dump(exclude={"response"}) + | {"responses": (False, True) if response == "both" else (response == "sse",)} + ) + signal.signal(signal.SIGTERM, stop) + run(options) + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + main() diff --git a/scripts/benchmarks/gateway_split/src/main.rs b/scripts/benchmarks/gateway_split/src/main.rs new file mode 100644 index 00000000000..0ae9aeb77e9 --- /dev/null +++ b/scripts/benchmarks/gateway_split/src/main.rs @@ -0,0 +1,236 @@ +mod streaming; + +use std::{ + convert::Infallible, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + time::Duration, +}; + +use axum::{ + Router, + body::{Body, to_bytes}, + extract::{Request, State}, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::any, +}; +use bytes::Bytes; +use clap::{Parser, Subcommand}; +use futures_util::TryStreamExt; +use serde_json::json; +use sha2::{Digest, Sha256}; + +#[derive(Parser)] +struct Args { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + Serve { + #[arg(long)] + port: u16, + #[arg(long, default_value = "backend")] + role: String, + #[arg(long, default_value = "http://127.0.0.1:19400")] + backend: String, + #[arg(long, default_value = "http://127.0.0.1:19402")] + python: String, + }, +} + +#[derive(Default)] +struct Counters { + requests: AtomicU64, + upload_bytes: AtomicU64, + active: AtomicU64, + completed: AtomicU64, + cancelled: AtomicU64, +} + +struct StreamGuard { + counters: Arc, + completed: bool, +} + +impl Drop for StreamGuard { + fn drop(&mut self) { + self.counters.active.fetch_sub(1, Ordering::Relaxed); + if self.completed { + self.counters.completed.fetch_add(1, Ordering::Relaxed); + } else { + self.counters.cancelled.fetch_add(1, Ordering::Relaxed); + } + } +} + +#[derive(Clone)] +struct App { + role: String, + backend: String, + python: String, + raw_client: streaming::Client, + counters: Arc, +} + +fn clean_headers(headers: &HeaderMap) -> HeaderMap { + let connection_tokens = headers + .get("connection") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + headers + .iter() + .filter(|(name, _)| { + !matches!( + name.as_str(), + "host" + | "connection" + | "keep-alive" + | "proxy-authenticate" + | "proxy-authorization" + | "te" + | "trailer" + | "transfer-encoding" + | "upgrade" + ) && !connection_tokens + .split(',') + .any(|token| token.trim().eq_ignore_ascii_case(name.as_str())) + }) + .map(|(k, v)| (k.clone(), v.clone())) + .collect() +} + +fn header_number(headers: &HeaderMap, key: &str, fallback: u64) -> u64 { + headers + .get(key) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse().ok()) + .unwrap_or(fallback) +} + +async fn backend(app: App, headers: HeaderMap, data: Bytes) -> Response { + app.counters.requests.fetch_add(1, Ordering::Relaxed); + let value = match serde_json::from_slice::(&data) { + Ok(value) => value, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + let status = StatusCode::from_u16(header_number(&headers, "x-bench-status", 200) as u16) + .unwrap_or(StatusCode::BAD_REQUEST); + let digest = format!("{:x}", Sha256::digest(&data)); + if let Some(expected) = headers + .get("x-bench-expected-sha256") + .and_then(|value| value.to_str().ok()) + && expected != digest + { + return StatusCode::BAD_REQUEST.into_response(); + } + let common = [ + ("x-body-sha256", digest), + ( + "x-bench-id", + headers + .get("x-bench-id") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_owned(), + ), + ]; + if status != StatusCode::OK { + return (status, common, "injected failure").into_response(); + } + let message = json!({"id":"msg_fixture", "type":"message", "role":"assistant", "model":"fixture-model", "content":[{"type":"text","text":"x".repeat(1024)}], "stop_reason":"end_turn", "stop_sequence":null, "usage":{"input_tokens":10,"output_tokens":8}}); + if value.get("stream").and_then(|v| v.as_bool()) != Some(true) { + return (common, axum::Json(message)).into_response(); + } + let delay = header_number(&headers, "x-bench-delay-ms", 5); + let count = header_number(&headers, "x-bench-events", 8); + app.counters.active.fetch_add(1, Ordering::Relaxed); + let guard = StreamGuard { + counters: app.counters.clone(), + completed: false, + }; + let stream = async_stream::stream! { + let mut guard = guard; + let start = json!({"type":"message_start","message":{"id":"msg_fixture","type":"message","role":"assistant","model":"fixture-model","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}); + yield Ok::(Bytes::from(format!("event: message_start\ndata: {start}\n\n"))); + yield Ok(Bytes::from_static(b"event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n")); + for _ in 0..count { + tokio::time::sleep(Duration::from_millis(delay)).await; + let event = json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"x".repeat(128)}}); + yield Ok(Bytes::from(format!("event: content_block_delta\ndata: {event}\n\n"))); + } + yield Ok(Bytes::from_static(b"event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\nevent: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"output_tokens\":8}}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")); + guard.completed = true; + }; + let mut response = (common, Body::from_stream(stream)).into_response(); + response + .headers_mut() + .insert("content-type", "text/event-stream".parse().unwrap()); + response +} + +async fn handle(State(app): State, request: Request) -> Response { + if request.uri().path() == "/metrics" { + return axum::Json(json!({ + "requests": app.counters.requests.load(Ordering::Relaxed), + "upload_bytes": app.counters.upload_bytes.load(Ordering::Relaxed), + "active": app.counters.active.load(Ordering::Relaxed), + "completed": app.counters.completed.load(Ordering::Relaxed), + "cancelled": app.counters.cancelled.load(Ordering::Relaxed), + })) + .into_response(); + } + if app.role == "stream" { + return streaming::forward(&app, request).await; + } + let (parts, body) = request.into_parts(); + let body = if app.role == "backend" && parts.headers.contains_key("x-bench-upload-probe") { + let counters = app.counters.clone(); + Body::from_stream(body.into_data_stream().inspect_ok(move |chunk| { + counters + .upload_bytes + .fetch_add(chunk.len() as u64, Ordering::Relaxed); + })) + } else { + body + }; + let data = match to_bytes(body, 8 * 1024 * 1024).await { + Ok(data) => data, + Err(_) => return StatusCode::PAYLOAD_TOO_LARGE.into_response(), + }; + if app.role == "backend" { + return backend(app, parts.headers, data).await; + } + StatusCode::NOT_FOUND.into_response() +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Box> { + match Args::parse().command { + Command::Serve { + port, + role, + backend, + python, + } => { + let app = App { + role, + backend, + python, + raw_client: streaming::client(), + counters: Arc::new(Counters::default()), + }; + let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await?; + axum::serve( + listener, + Router::new().fallback(any(handle)).with_state(app), + ) + .await?; + } + } + Ok(()) +} diff --git a/scripts/benchmarks/gateway_split/src/streaming.rs b/scripts/benchmarks/gateway_split/src/streaming.rs new file mode 100644 index 00000000000..ad6930f0968 --- /dev/null +++ b/scripts/benchmarks/gateway_split/src/streaming.rs @@ -0,0 +1,68 @@ +use std::{io, time::Duration}; + +use axum::{ + body::Body, + extract::Request, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use http_body_util::{BodyExt, Limited, StreamBody}; +use hyper_util::{ + client::legacy::{Client as HyperClient, connect::HttpConnector}, + rt::TokioExecutor, +}; + +use crate::{App, clean_headers, header_number}; + +pub type Client = HyperClient; + +pub fn client() -> Client { + let mut connector = HttpConnector::new(); + connector.set_nodelay(true); + connector.set_connect_timeout(Some(Duration::from_secs(1))); + HyperClient::builder(TokioExecutor::new()) + .pool_max_idle_per_host(500) + .pool_idle_timeout(Duration::from_secs(15)) + .retry_canceled_requests(false) + .build(connector) +} + +pub async fn forward(app: &App, request: Request) -> Response { + if header_number(request.headers(), "content-length", 0) > 8 * 1024 * 1024 { + return StatusCode::PAYLOAD_TOO_LARGE.into_response(); + } + let target = if request.uri().path() != "/v1/messages" + || request.headers().contains_key("x-bench-python") + { + &app.python + } else { + &app.backend + }; + let uri = match format!("{target}{}", request.uri()).parse() { + Ok(uri) => uri, + Err(_) => return StatusCode::BAD_GATEWAY.into_response(), + }; + let (mut parts, body) = request.into_parts(); + parts.uri = uri; + parts.headers = clean_headers(&parts.headers); + let outbound = Request::from_parts(parts, Body::new(Limited::new(body, 8 * 1024 * 1024))); + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + let result = tokio::time::timeout_at(deadline, app.raw_client.request(outbound)).await; + let upstream = match result { + Ok(Ok(response)) => response, + _ => return StatusCode::BAD_GATEWAY.into_response(), + }; + let (mut parts, mut body) = upstream.into_parts(); + parts.headers = clean_headers(&parts.headers); + let frames = async_stream::stream! { + loop { + match tokio::time::timeout_at(deadline, body.frame()).await { + Ok(Some(Ok(frame))) => yield Ok::<_, io::Error>(frame), + Ok(Some(Err(error))) => { yield Err(io::Error::other(error)); break; } + Ok(None) => break, + Err(error) => { yield Err(io::Error::new(io::ErrorKind::TimedOut, error)); break; } + } + } + }; + Response::from_parts(parts, Body::new(StreamBody::new(frames))) +} diff --git a/scripts/benchmarks/gateway_split/test_cloud.py b/scripts/benchmarks/gateway_split/test_cloud.py new file mode 100644 index 00000000000..4ce23db9217 --- /dev/null +++ b/scripts/benchmarks/gateway_split/test_cloud.py @@ -0,0 +1,171 @@ +import itertools +import json +import tempfile +import unittest +from pathlib import Path +from typing import Final + +from analyze_bifrost import Record, Summary +from bifrost_bench import Stats +from cloud import Manifest, Metadata, check_job, merge +from matrix import TOPOLOGIES, Job, jobs, shard +from pydantic import TypeAdapter +from run import Options + + +def trial(job: Job, variant: str, repetition: int) -> Record: + return Record( + generator="bifrost", + variant=variant, + prompt_bytes=job.prompt_bytes, + body_bytes=job.prompt_bytes + 100, + stream=job.response == "sse", + concurrency=job.concurrency, + repetition=repetition, + python_body="stream", + python_client="aiohttp", + rust_client="hyper-stream", + http_parser="httptools", + event_loop="uvloop", + profile=job.profile, + gateway_cpu_s=0.2, + gateway_cpu_cores=0.1, + gateway_peak_rss_bytes=1024, + loadgen_cpu_cores=0.1, + backend_completed=10, + backend_requests=10, + backend_cancelled=0, + stats=Stats( + requests=10, + rate=5, + success_rate=100, + mean_latency_ms=1, + p50_latency_ms=1, + p95_latency_ms=2, + p99_latency_ms=3, + max_latency_ms=4, + elapsed_seconds=2, + throughput_rps=5 + repetition, + first_event_p50_ms=0.5, + first_event_p95_ms=1, + first_event_p99_ms=2, + status_code_counts={"200": 10}, + drop_reasons={}, + ), + ) + + +def fixture(root: Path, job: Job, machine_class: str = "test-machine") -> Path: + path: Final = root / job.id / "trials.jsonl" + path.parent.mkdir(parents=True) + rows: Final = tuple( + trial(job, variant, repetition) for variant, repetition in itertools.product(TOPOLOGIES, range(job.repetitions)) + ) + path.write_text("".join(row.model_dump_json() + "\n" for row in rows)) + metadata: Final = Metadata( + platform="test-platform", + cpu="test-cpu", + logical_cpus=4, + memory_bytes=1024, + machine_class=machine_class, + python="test-python", + rust="test-rust", + git_sha="test-revision", + versions={}, + harness_sha256={}, + bifrost_binary_sha256="test-binary", + bifrost_patch_sha256="test-patch", + resolved_http_parser="h11", + dependency_lock_sha256="test-lock", + parameters=Options( + output=path, + profile=job.profile, + sizes=(job.prompt_bytes,), + concurrencies=(job.concurrency,), + responses=(job.response == "sse",), + seconds=job.seconds, + repetitions=job.repetitions, + warmup=job.warmup_seconds, + seed=job.seed, + machine_class=machine_class, + ), + ) + path.with_suffix(".metadata.json").write_text(metadata.model_dump_json()) + probes: Final = tuple( + { + "variant": variant, + "body_digest": True, + "anthropic_sdk_messages": True, + "slow_upload_verified": True, + "headers": True, + "status_passthrough": True, + "cancelled_upstream": True, + } + for variant in TOPOLOGIES + ) + path.with_suffix(".probes.json").write_text(json.dumps(probes)) + return path + + +class CloudContractTests(unittest.TestCase): + def test_shards_cover_each_workload_exactly_once(self) -> None: + planned: Final = jobs() + for count in (1, 5, 113): + with self.subTest(count=count): + self.assertCountEqual( + tuple( + job.id + for job in itertools.chain.from_iterable(shard(planned, index, count) for index in range(count)) + ), + tuple(job.id for job in planned), + ) + + def test_invalid_shard_assignment_is_rejected(self) -> None: + for index, count in ((0, 0), (-1, 2), (2, 2)): + with self.subTest(index=index, count=count), self.assertRaises(ValueError): + shard(jobs(), index, count) + + def test_merger_includes_every_topology_and_repetition(self) -> None: + selected: Final = jobs(repetitions=3)[:2] + with tempfile.TemporaryDirectory() as temporary: + root: Final = Path(temporary) + for job in selected: + fixture(root, job) + merge(Manifest(jobs=selected), root, root / "merged") + combined: Final = (root / "merged/trials.jsonl").read_text().splitlines() + self.assertEqual(len(combined), sum(job.expected_rows for job in selected)) + summaries: Final = TypeAdapter(tuple[Summary, ...]).validate_json( + (root / "merged/summary.json").read_bytes() + ) + self.assertEqual(len(summaries), len(selected) * len(TOPOLOGIES)) + self.assertTrue(all(row.rps.median == 6 for row in summaries)) + + def test_duplicate_cannot_replace_a_missing_topology(self) -> None: + job: Final = jobs(repetitions=1)[0] + with tempfile.TemporaryDirectory() as temporary: + path: Final = fixture(Path(temporary), job) + rows: Final = path.read_text().splitlines() + path.write_text("\n".join((rows[0], rows[0], *rows[2:])) + "\n") + with self.assertRaisesRegex(ValueError, "Incomplete or duplicate"): + check_job(job, path) + + def test_missing_trial_is_rejected(self) -> None: + job: Final = jobs(repetitions=1)[0] + with tempfile.TemporaryDirectory() as temporary: + path: Final = fixture(Path(temporary), job) + path.write_text("\n".join(path.read_text().splitlines()[:-1]) + "\n") + with self.assertRaisesRegex(ValueError, "Incomplete or duplicate"): + check_job(job, path) + + def test_different_cloud_hardware_cannot_be_merged(self) -> None: + selected: Final = jobs(repetitions=1)[:2] + with tempfile.TemporaryDirectory() as temporary: + root: Final = Path(temporary) + fixture(root, selected[0], "machine-one") + fixture(root, selected[1], "machine-two") + with self.assertRaisesRegex(ValueError, "different hardware"): + merge(Manifest(jobs=selected), root, root / "merged") + + +if __name__ == "__main__": + unittest.main()