diff --git a/.gitignore b/.gitignore index aca14940..3665f378 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,7 @@ benchmarks docs/ *.md !DIAGNOSTIC_ITERATOR_DESIGN.md +!benches/*.md !examples/**/*.md !src/wasm/**/*.md !tests/fixtures/ripe/**/*.md diff --git a/benches/README.md b/benches/README.md index e8786cec..2c3413cb 100644 --- a/benches/README.md +++ b/benches/README.md @@ -1,20 +1,53 @@ # Benchmarking -We implement benchmark here with comparisons against some popular alternative parsers. +Two Criterion benchmark targets: -## Setting Up +- `internals` measures the parser itself on an in-memory byte buffer, so decompression and I/O are + excluded. It always covers the small update file and the small RIB fixture, and can stream real + archives when pointed at them. +- `bench_main` compares the `bgpkit-parser` CLI against other parsers installed on the machine + (`bgpdump`, `libparsebgp`, `bgpscanner`, `bgpgrep`, `mrt-parser`). Missing programs are skipped. -Run `bash download.sh` to download test data files to `/tmp` directory. +## Running -## Run benchmarks +```sh +cargo bench --bench internals # ~8 min, all fixtures +cargo bench --bench internals -- rib # filter by benchmark id substring +cargo bench --bench bench_main # needs `cargo build --release --features cli` first +``` -Run `cargo bench` anywhere in the project will run all benchmarks. +Test data is downloaded on the first run into `/test_data` (the target directory +comes from `CARGO_TARGET_TMPDIR`, so `CARGO_TARGET_DIR` and `build.target-dir` are respected) and is +verified against the md5 sums in `data_source.rs`. -## Implemented Benchmarks +## Real archives -- [X] BGPKIT Parser -- [X] [bgpdump](https://github.com/RIPE-NCC/bgpdump) -- [ ] [libparsebgp](https://github.com/CAIDA/libparsebgp) -- [ ] [bgpscanner](https://gitlab.com/Isolario/bgpscanner) -- [ ] [micro bgp suite](https://git.doublefourteen.io/bgp/ubgpsuite) -- [ ] [mrt-parser](https://github.com/sdstrowes/mrt-parser) \ No newline at end of file +Point the benchmark at local MRT files to include decompression and a realistic corpus size. These +run only when the variables are set, and nothing is downloaded. + +```sh +BGPKIT_BENCH_RIB=/data/bview.gz \ +BGPKIT_BENCH_UPDATES=/data/updates.gz \ +BGPKIT_BENCH_STREAM_ELEMS=1000000 \ + cargo bench --bench internals -- "real" +``` + +## Reading the numbers + +- Record-oriented benchmarks report records/s, elem-oriented benchmarks report elems/s. The same + corpus yields more elems than records (about 2x for a RIB dump), so compare a benchmark only with + itself, never `into_record_iter` against `into_elem_iter`. +- The fixtures are small (4290 update records, 235306 RIB records). They are enough to compare + revisions of the parser, not to predict wall time for a full archive. + +## Comparing revisions + +Criterion keeps named baselines under the target directory: + +```sh +git switch main && cargo bench --bench internals -- --save-baseline before +git switch my-branch && cargo bench --bench internals -- --baseline before +``` + +The same command shape works for a before/after record on one machine; results from different +machines are not comparable. diff --git a/benches/RESULTS.md b/benches/RESULTS.md new file mode 100644 index 00000000..e788ae20 --- /dev/null +++ b/benches/RESULTS.md @@ -0,0 +1,230 @@ +# Parsing performance: benchmark revision, measured gains, BGPStream comparison + +Recorded on branch `perf/parser-throughput`, November 2026. Every number below is reproducible with +the commands at the end. + +## Summary + +Parse-only instruction counts per full parse of the fixture (decompression excluded), measured at +`5a5afbc`; the two post-review changes recorded below add a further 0.5% end to end: + +| workload | before | after | delta | +| --- | --- | --- | --- | +| RIB fixture, framing only (235306 records) | 0.252 G | 0.221 G | -12.1% | +| RIB fixture, record iteration | 2.815 G | 2.338 G | -16.9% | +| RIB fixture, elem iteration (470642 elems) | 3.718 G | 3.251 G | -12.6% | +| Updates fixture, record iteration (4290 records) | 35.15 M | 30.68 M | -12.7% | +| Updates fixture, elem iteration (8160 elems) | 50.02 M | 44.83 M | -10.4% | + +End to end through the CLI on compressed archives (decompression included): + +| workload | mode | before | after | delta | +| --- | --- | --- | --- | --- | +| rrc00 bview 2026-10-14, 420 MB gz | elems, 55.8 M | 463.1 G / 54.3 s | 392.4 G / 42.6 s | -15.3% instr, -21.5% wall | +| rrc00 bview 2026-10-14 | records, 1.43 M | 361.4 G / 42.3 s | 288.6 G / 29.8 s | -20.2% instr, -29.4% wall | +| RIS updates 2026-10-14 00:10, 4.5 MB gz | elems, 417788 | 5.04 G / 0.51 s | 4.89 G / 0.47 s | -3.0% instr, -7.8% wall | +| RIS updates 2026-10-14 00:10 | records, 150360 | 4.32 G / 0.41 s | 4.17 G / 0.40 s | -3.3% instr, -2.4% wall | + +Allocations per item (counter wraps the global allocator): + +| workload / mode | before | after | +| --- | --- | --- | +| RIB, record iteration | 14.00 allocs, 7218 B per record | 7.24 allocs, 3490 B per record | +| RIB, elem iteration | 9.00 allocs, 3667 B per elem | 5.61 allocs, 1866 B per elem | +| Updates, record iteration | 10.23 allocs, 5328 B per record | 5.55 allocs, 1887 B per record | +| Updates, elem iteration | 7.34 allocs, 2947 B per elem | 4.88 allocs, 1138 B per elem | + +## What these numbers do and do not establish + +- Instruction counts come from `perf stat` on a virtualized host whose PMU reports them; they are + stable to 0.01% across runs of one binary and are used for A/B only, not as physical counts. +- Host wall times move by up to 18% between runs of the *same* binary because an unrelated archive + backfill holds 10+ cores. Wall figures above are means of repeated alternating runs. +- Parse-only numbers exclude decompression by construction; the CLI rows include it. They are not + interchangeable: the small updates fixture improves 10% parse-only but only 3% end to end. +- `bgpkit-parser -r` / `-e` are counting paths: they parse and drop, without formatting or writing + output. These rows say nothing about export throughput. +- The fast framing path covers the default iterators. The fallible iterators still go through the + raw-record chunking path (`src/parser/iters/mod.rs`) and the route iterator still calls + `chunk_mrt_record` (`src/parser/iters/route.rs`), so they do not inherit the whole win. +- Fixture gains do not transfer one-to-one to a full RIB; the rrc00 row is the number to quote. + +## Harness repairs + +`cargo bench` could not run on a clean checkout, and where it ran it measured the wrong thing: + +- the test-data download opened the destination with a decompressing reader and then truncated it, + so every download failed with `decompression not finished but EOF reached`; the published md5 was + also compared against decompressed bytes; +- the target directory was found by walking up for a literal `target/`, which panics whenever + `CARGO_TARGET_DIR` or `build.target-dir` moves artifacts out of the tree (now + `CARGO_TARGET_TMPDIR`); +- `bench_main` panicked when `target/release/bgpkit-parser` was missing, though `cargo bench` never + builds that binary (`required-features = ["cli"]`); +- record modes were capped at 100k records while elem modes were capped at 100k elems, so the modes + measured different amounts of work; +- the record-to-elem benchmark produced no elems for RIB records because its `Elementor` had no + peer index table; +- `into_update_iter` divided by the record count instead of the update count, preload truncated a + too-large corpus silently, and two benchmarks divided by a fabricated 1 when they produced + nothing. + +Every mode now parses the whole in-memory fixture, except the record-to-elem benchmark, which keeps +100k parsed records in memory between iterations and says so. Throughput is reported next to each +time, and the real-archive path streams a bounded number of elems +(`BGPKIT_BENCH_STREAM_ELEMS`, default 1M) from `BGPKIT_BENCH_RIB` / `BGPKIT_BENCH_UPDATES`. + +## Changes + +1. `perf(parser): cut attribute-path allocations` (993f61a) + + AS_PATH segments stay inline and ASNs decode straight into the segment `SmallVec` instead of + growing a `Vec` and copying it; the attribute vector is sized from the section length with a + realistic per-attribute floor instead of one slot per three bytes; RIB entries reserve one slot + per entry instead of two. + +2. `perf(mrt): keep record headers on the stack` (f742b18) + + Framing allocated a `Vec` for the captured header and a `Bytes` for the header on every record, + only for the raw-record API and error reporting. The header now lives in a 16-byte stack buffer. + +3. `perf(parser): decode AS_PATH from the record body` (7cf3b9e) + + AS_PATH is in essentially every announcement and was the last attribute still split out of the + record body into its own `Bytes` before parsing. + +4. `fix(mrt): keep partial extended-timestamp bytes on header I/O error` (be3ff0a) + + Found by an external review: the stack-header rewrite advanced the captured length only after a + successful read, so an I/O error inside the 4-byte microsecond field dropped the bytes already + read from the error report. Covered now at 13, 14 and 15 header bytes, for an I/O error and for + a clean EOF. + +5. `perf(parser): reserve eight attribute slots instead of four` (a8a5924) + + A floor of four made 71.4% of RIB entries grow their attribute vector (measured over all 470,642 + fixture entries); six leaves 3.7% short, so eight was chosen. Four, six and eight were measured + against the same binary and stayed within 0.5% of each other on the fixtures; the end-to-end + numbers quoted below were all measured with eight. + +## Post-review changes + +Two of the review's low-risk items were implemented and kept, one was deferred. + +| commit | change | measured effect | +| --- | --- | --- | +| `d4dd8af` | no attribute slots for an empty section | real RIS updates elems -0.25%, records -0.36%; RIB fixture +0.07% | +| `42d0020` | ORIGIN, NEXT_HOP, MED, LOCAL_PREF parse from the section slice | RIB records -1.1%, RIB elems -0.9%, updates records -0.7%, real updates records -0.2% | + +Criterion medians for the tree after both changes, against the `pre-opt` baseline recorded before any +of this work (same host, same harness, host load unchanged in kind): + +| benchmark | before | after | delta | +| --- | --- | --- | --- | +| updates/into_raw_record_iter | 513.48 us | 445.18 us | -13.3% | +| updates/into_record_iter | 3.9379 ms | 3.2590 ms | -17.2% | +| updates/into_elem_iter | 5.6239 ms | 5.1181 ms | -9.0% | +| updates/into_route_iter | 2.5819 ms | 2.4096 ms | -6.7% | +| rib/into_raw_record_iter | 25.330 ms | 23.972 ms | -5.4% | +| rib/into_record_iter | 296.01 ms | 240.11 ms | -18.9% | +| rib/into_elem_iter | 385.12 ms | 341.07 ms | -11.4% | +| rib/into_route_iter | 174.23 ms | 159.36 ms | -8.5% | + +End to end, rrc00 bview 2026-10-14, four alternating runs, `-e` counting path: + +``` +before (5a5afbc) 392.451 G / 44.11 s 392.376 G / 43.45 s +after (42d0020) 390.588 G / 42.90 s 390.569 G / 41.45 s +``` + +-0.47% instructions, wall within host noise. The fixture-level CLI deltas above are diluted by +decompression, which the parse-only rows exclude. + +Deferred: dense peer lookup for RIB elems. `Elementor.peer_table` is a public field of a public +struct, so an id-indexed cache cannot live there without a breaking change, and a cache held beside +it has a staleness window when a caller replaces the table mid-iteration. The hash lookup of a +32-bit key against roughly a thousand peers is a small share of the remaining cost; it belongs with +the structural rework, not before it. + +## Comparison with CAIDA BGPStream + +Same files, both sides reading the compressed input, walking every record and (in `elems` mode) +every element, then discarding the result. The BGPStream probe and container are in +`benches/bgpstream-compare/`. + +| workload | mode | BGPStream | bgpkit-parser | ratio | +| --- | --- | --- | --- | --- | +| rrc00 bview, 420 MB gz | records | 85.8 G / 5.28 s | 288.6 G / 29.8 s | 3.4x instr, 5.7x wall | +| rrc00 bview | elems (55.8 M) | 116.2 G / 9.06 s | 392.4 G / 42.6 s | 3.4x instr, 4.7x wall | +| RIS updates, 4.5 MB gz | records | 0.76 G / 0.09 s | 4.17 G / 0.40 s | 5.5x instr, 4.3x wall | +| RIS updates | elems (417788) | 0.93 G / 0.11 s | 4.89 G / 0.47 s | 5.3x instr, 4.2x wall | +| RIB fixture | records | 3.64 G / 0.61 s | 5.53 G / 0.71 s | 1.5x instr, 1.2x wall | +| RIB fixture | elems | 3.90 G / 0.58 s | 6.49 G / 0.85 s | 1.7x instr, 1.5x wall | + +BGPStream is faster, and the gap is structural rather than a few instructions per record: + +- BGPStream hands out elems as views into the record buffer; `bgpkit-parser` returns owned values + (owned `AsPath`, owned community vector, owned origin list), so every elem allocates. That is an + API contract difference and is most of the elem-mode gap. +- The record-mode gap on a real RIB (3.4x) is larger than on the small fixture (1.5x) because + bgpkit parses and stores every attribute type eagerly, while BGPStream decodes on demand. + +Caveats: element counts on a RIB match exactly (55,786,629), record counts differ by one (BGPStream +does not surface the PEER_INDEX_TABLE record), the updates fixture yields 8,183 elems for BGPStream +against 8,160 for bgpkit, and the probe never reads a field, so any lazily parsed BGPStream field is +not charged to it. All bgpkit numbers here are the counting paths (`-r` / `-e`): the default output +path also formats and writes every element, which roughly doubles the cost (12.35 G instead of +6.49 G on the RIB fixture, 12.36 G instead of 4.89 G on the updates file) and is a different +workload. + +## What did not work + +Routing all thirteen frequently-seen attribute parsers through one +`Option>` slice dispatcher measured **+6.8% on RIB elems and +7.9% on RIB +records**. Removing the per-attribute `Bytes` split did not pay for the lost per-arm +specialization; the change was reverted. Only AS_PATH, whose parse stays inside the dispatch arm, +kept the slice treatment and gained. + +## Directions not taken yet + +Ranked by expected payoff per unit of risk, from an external review of this branch: + +1. ~~Skip the attribute-vector allocation for empty attribute sections.~~ Done: `d4dd8af`. +2. ~~Borrowed-slice parsing for ORIGIN, NEXT_HOP, MED and LOCAL_PREF, one inlined arm each.~~ Done: + `42d0020`; the per-arm result confirms the thirteen-way dispatcher was the problem, not slices. +3. Dense peer lookup for RIB elems: `next_rib_elem` hashes a dense `u16` id per elem. Deferred: + `Elementor.peer_table` is a public field, so the cache either breaks the public struct or goes + stale when a caller swaps the table mid-iteration. +4. Structural: emit elems directly from the record body instead of building a full + `Vec`/`Attributes` set and then dismantling it. This is the direction that could close + part of the BGPStream gap; it needs an internal streaming builder plus a batch/view API for + shared attributes, not a change to `BgpElem`. +5. Reuse the per-record body buffer. Blocked on ownership: unknown, deprecated and failed attributes + keep `Bytes` into the body, so the buffer cannot be recycled until those handles are bounded. +6. Measure decompression separately (decompress-only versus pre-decompressed parse) before + optimizing it; there is no current phase breakdown for the final binary. + +## Reproducing + +```sh +# fixed-workload instruction counts (committed example) +cargo build --release --example perf_workload +perf stat -e instructions target/release/examples/perf_workload \ + /path/to/rib-example-small.bz2 elems 4 # also: records, raw; use 10 for the slope + +# criterion suites +cargo build --release --features cli # bench_main needs the CLI binary +cargo bench --bench internals # ~8 min, whole fixtures + +# end-to-end on a real archive +perf stat -e instructions target/release/bgpkit-parser -e /path/to/bview.gz + +# CAIDA BGPStream side of the comparison +cd benches/bgpstream-compare && docker build -t bgpstream-caida . +docker run --rm --privileged -v /path/to/bview.gz:/data/input.gz bgpstream-caida \ + bash -lc "perf stat -e instructions /usr/local/bin/bgps_probe rib /data/input.gz elems" +``` + +Parse-only cost is separated from fixed startup by running two iteration counts: +`P = (I(10) - I(4)) / 6`. Per-commit deltas quoted in the commit messages are totals at one fixed +iteration count (startup included) and are not compoundable into the table above. diff --git a/benches/bench_main.rs b/benches/bench_main.rs index 195d167a..1f4f02e6 100644 --- a/benches/bench_main.rs +++ b/benches/bench_main.rs @@ -98,10 +98,21 @@ fn benchmark(c: &mut Criterion) { } = program; let path_result = match executable { - // Ensure we always use the latest build of the project. Optimized binary targets are - // compiled before benchmarks to allow for integration tests so it will always be found - // at this path. - "bgpkit-parser" => Ok(data_source::locate_target_dir().join("release/bgpkit-parser")), + // The CLI binary carries `required-features = ["cli"]`, which `cargo bench` does not + // enable, so it only exists after an explicit release build. + "bgpkit-parser" => { + let path = data_source::locate_target_dir().join("release/bgpkit-parser"); + if path.is_file() { + Ok(path) + } else { + println!( + "skipping bgpkit-parser: {} is missing; run \ + `cargo build --release --features cli` first", + path.display() + ); + continue; + } + } // For remaining programs use which crate to locate their executables on the system name => which::which(name), }; diff --git a/benches/bgpstream-compare/Dockerfile b/benches/bgpstream-compare/Dockerfile new file mode 100644 index 00000000..a0d0cedf --- /dev/null +++ b/benches/bgpstream-compare/Dockerfile @@ -0,0 +1,14 @@ +FROM debian:bookworm-slim + +# CAIDA's packages are built for bookworm; newer Debian releases are not served. +RUN apt-get update && apt-get install -y --no-install-recommends \ + curl ca-certificates gnupg sudo lsb-release procps linux-perf gcc libc6-dev make + +RUN curl -s https://pkg.caida.org/os/debian/bootstrap.sh | bash + +RUN apt-get update && apt-get install -y --no-install-recommends bgpstream libbgpstream2-dev + +COPY bgps_probe.c /work/bgps_probe.c +RUN gcc -O2 -o /usr/local/bin/bgps_probe /work/bgps_probe.c -lbgpstream + +WORKDIR /work diff --git a/benches/bgpstream-compare/bgps_probe.c b/benches/bgpstream-compare/bgps_probe.c new file mode 100644 index 00000000..45849350 --- /dev/null +++ b/benches/bgpstream-compare/bgps_probe.c @@ -0,0 +1,57 @@ +// Parse a single MRT file with CAIDA BGPStream and count what was produced, without formatting. +// Two modes: +// records : iterate bgpstream records only +// elems : iterate records and drain each record's elems (this is what forces attribute parsing) +#include +#include +#include + +int main(int argc, char **argv) { + if (argc < 4) { + fprintf(stderr, "usage: %s \n", argv[0]); + return 2; + } + const char *kind = argv[1]; + const char *path = argv[2]; + const char *mode = argv[3]; + + bgpstream_t *bs = bgpstream_create(); + if (bs == NULL) { + fprintf(stderr, "bgpstream_create failed\n"); + return 1; + } + bgpstream_set_data_interface(bs, BGPSTREAM_DATA_INTERFACE_SINGLEFILE); + bgpstream_data_interface_id_t di = BGPSTREAM_DATA_INTERFACE_SINGLEFILE; + const char *opt_name = strcmp(kind, "rib") == 0 ? "rib-file" : "upd-file"; + bgpstream_data_interface_option_t *opt = + bgpstream_get_data_interface_option_by_name(bs, di, opt_name); + if (opt == NULL) { + fprintf(stderr, "option %s not found\n", opt_name); + return 1; + } + bgpstream_set_data_interface_option(bs, opt, path); + if (bgpstream_start(bs) < 0) { + fprintf(stderr, "bgpstream_start failed\n"); + return 1; + } + + bgpstream_record_t *record = NULL; + bgpstream_elem_t *elem = NULL; + long long records = 0; + long long elems = 0; + int elems_mode = strcmp(mode, "elems") == 0; + + int rv; + while ((rv = bgpstream_get_next_record(bs, &record)) > 0) { + records++; + if (elems_mode) { + while (bgpstream_record_get_next_elem(record, &elem) > 0) { + elems++; + } + } + } + + printf("records=%lld elems=%lld\n", records, elems); + bgpstream_destroy(bs); + return 0; +} diff --git a/benches/data_source.rs b/benches/data_source.rs index 272bbb0f..77c5502e 100644 --- a/benches/data_source.rs +++ b/benches/data_source.rs @@ -1,7 +1,6 @@ use md5::Digest; -use oneio::get_reader; use std::fs::{create_dir_all, File}; -use std::io::{self, BufWriter, Write}; +use std::io; use std::path::{Path, PathBuf}; use std::sync::Once; @@ -32,6 +31,25 @@ pub fn test_data_file(name: &str) -> PathBuf { test_data_dir().join(name) } +/// Returns the path named by an environment variable pointing at a local corpus, or `None`. +/// +/// Used for opt-in benchmarks against real archives, which are neither checked in nor downloaded +/// by the benchmark itself. +#[allow(dead_code)] // only the internals benchmark reads opt-in corpora; bench_main includes this module too +pub fn env_corpus(var: &str) -> Option { + let value = std::env::var_os(var)?; + if value.is_empty() { + return None; + } + let path = PathBuf::from(value); + if path.is_file() { + Some(path) + } else { + eprintln!("ignoring {var}: {} is not a file", path.display()); + None + } +} + /// Downloads any missing test data files listed in [DATA_SOURCES]. If an error is encountered /// during this process the program panics with a (hopefully) helpful error. pub fn download_test_data() { @@ -99,45 +117,46 @@ fn md5_checksum_for_file>(path: P) -> io::Result { Ok(context.finalize()) } -struct HashingWriter { - writer: W, - hasher: md5::Context, -} - -impl Write for HashingWriter { - fn write(&mut self, buf: &[u8]) -> io::Result { - let length = self.writer.write(buf)?; - self.hasher.write_all(&buf[..length])?; - Ok(length) - } - - fn flush(&mut self) -> io::Result<()> { - self.writer.flush() - } -} - +/// Download `url` to `target`, returning the md5 of the raw (still compressed) bytes. +/// +/// The download goes to a sibling `.part` file that is renamed only after the checksum is +/// computed, so an interrupted run leaves nothing a later run would accept as complete. The +/// checksum is computed over the bytes as downloaded: decompressing first would both mismatch the +/// published hashes and truncate the file being read. fn download_file>(url: &str, target: P) -> io::Result { - let target_path = target.as_ref().to_str().unwrap().to_string(); - - oneio::download(url, target_path.as_str()).map_err(io::Error::other)?; - - let mut reader = get_reader(target_path.as_str()).unwrap(); - let mut writer = HashingWriter { - writer: BufWriter::new(File::create(target)?), - hasher: md5::Context::new(), - }; - - io::copy(&mut reader, &mut writer)?; - writer.flush()?; - - Ok(writer.hasher.finalize()) + let target = target.as_ref(); + let mut part = target.as_os_str().to_owned(); + part.push(".part"); + let part = PathBuf::from(part); + + let part_str = part + .to_str() + .ok_or_else(|| io::Error::other("test data path is not valid UTF-8"))?; + oneio::download(url, part_str).map_err(io::Error::other)?; + + let digest = md5_checksum_for_file(&part)?; + std::fs::rename(&part, target)?; + Ok(digest) } pub fn test_data_dir() -> PathBuf { locate_target_dir().join("test_data") } +/// The cargo target directory this benchmark was built into. +/// +/// Cargo exports `CARGO_TARGET_TMPDIR` (`/tmp`) for benchmark targets, which keeps the +/// lookup correct when `build.target-dir` or `CARGO_TARGET_DIR` moves artifacts out of the source +/// tree. Falling back to the historical "walk up until a `target/` directory appears" keeps older +/// toolchains working. pub fn locate_target_dir() -> PathBuf { + if let Some(tmp_dir) = option_env!("CARGO_TARGET_TMPDIR") { + let tmp_dir = PathBuf::from(tmp_dir); + if let Some(target_dir) = tmp_dir.parent() { + return target_dir.to_path_buf(); + } + } + // CARGO_MANIFEST_DIR is an absolute path that is always present at compile time when building // with cargo. We know the target directory is most likely in the same directory as the manifest // or in a parent directory in the case of a normal workspace. @@ -156,7 +175,10 @@ pub fn locate_target_dir() -> PathBuf { } else { // If we still can't find it, then it is likely defined by a workspace in some strange // location. At this point, we just give up as it will be near impossible to find. - panic!("Unable to find cargo target directory") + panic!( + "Unable to find cargo target directory (set CARGO_TARGET_DIR, or use a toolchain \ + that exports CARGO_TARGET_TMPDIR for benchmark targets)" + ) } } } diff --git a/benches/internals.rs b/benches/internals.rs index 7f2a0c41..97f28731 100644 --- a/benches/internals.rs +++ b/benches/internals.rs @@ -1,209 +1,294 @@ +//! In-process parsing benchmarks. +//! +//! Every benchmark parses an in-memory MRT byte buffer, so decompression and I/O are excluded by +//! design and the numbers describe the parser itself. Two corpora are always measured: the small +//! update file and the small RIB fixture. Real archives can be added without touching the code: +//! +//! ```sh +//! BGPKIT_BENCH_RIB=/path/to/bview.gz BGPKIT_BENCH_UPDATES=/path/to/updates.gz \ +//! cargo bench --bench internals -- --save-baseline main +//! ``` +//! +//! Those opt-in corpora are streamed from the file (so decompression is included) and are read +//! with `BGPKIT_BENCH_STREAM_ELEMS` elements per iteration (default 1,000,000). +//! +//! Each benchmark reports throughput next to the wall time. Record-oriented benchmarks report +//! records/s and elem-oriented benchmarks report elems/s; the two are not interchangeable, so +//! compare a benchmark only with itself. + +use bgpkit_parser::models::{MrtMessage, TableDumpV2Message}; use bgpkit_parser::{BgpkitParser, Elementor}; use bzip2::bufread::BzDecoder; -use criterion::{criterion_group, criterion_main, BatchSize, Criterion}; +use criterion::{criterion_group, criterion_main, BatchSize, Criterion, Throughput}; use flate2::bufread::GzDecoder; use std::fs::File; use std::hint::black_box; use std::io::{BufReader, Read}; +use std::path::Path; use std::time::Duration; mod data_source; -fn preload_min_for_n_entries(input_reader: R, n: usize) -> Vec { - struct CopyReader<'a, R> { - input: R, - copy: &'a mut Vec, - } - - impl Read for CopyReader<'_, R> { - fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - let bytes_read = self.input.read(buf)?; - self.copy.extend_from_slice(&buf[..bytes_read]); - Ok(bytes_read) - } - } +/// Upper bound on the decompressed bytes held in memory for a corpus. The fixtures are far +/// smaller; the bound only protects against pointing an opt-in variable at a full RIB archive. +const MAX_PRELOAD_BYTES: usize = 64 * 1024 * 1024; - let mut buffer = Vec::new(); - let parser = BgpkitParser::from_reader(CopyReader { - input: input_reader, - copy: &mut buffer, - }); +/// How many records the record-to-elem benchmark keeps in memory between iterations. +const ELEMENTOR_RECORDS: usize = 100_000; - parser.into_record_iter().take(n).for_each(drop); +/// Default number of elems streamed per iteration for the opt-in file corpora. +const STREAM_ELEMS: usize = 1_000_000; +fn preload(input_reader: R) -> Vec { + let mut buffer = Vec::new(); + input_reader + .take(MAX_PRELOAD_BYTES as u64 + 1) + .read_to_end(&mut buffer) + .expect("read corpus"); + assert!( + buffer.len() <= MAX_PRELOAD_BYTES, + "corpus is larger than the {} MiB in-memory budget; a truncated buffer would cut a record \ + in half and the iterator would hide it", + MAX_PRELOAD_BYTES / 1_048_576 + ); buffer } -/// Since we uncompress the input data first, put a hard limit on the number of records so we don't -/// use up too much memory. -const RECORD_LIMIT: usize = 100_000; - -pub fn criterion_benchmark(c: &mut Criterion) { - let update_data = data_source::test_data_file("update-example.gz"); - let rib_data = data_source::test_data_file("rib-example-small.bz2"); - - let updates_reader = BufReader::new(File::open(update_data).unwrap()); - let rib_reader = BufReader::new(File::open(rib_data).unwrap()); - - println!("Decompressing input data and loading into memory..."); - let updates = preload_min_for_n_entries(GzDecoder::new(updates_reader), RECORD_LIMIT); - let rib_dump = preload_min_for_n_entries(BzDecoder::new(rib_reader), RECORD_LIMIT); +/// Items each iteration of a parse mode yields for `data`. +struct Counts { + records: u64, + elems: u64, + updates: u64, + routes: u64, +} - println!("Required {} bytes to store updates", updates.len()); - println!("Required {} bytes to store rib table dump", rib_dump.len()); +fn count_items(data: &[u8]) -> Counts { + let records = BgpkitParser::from_reader(black_box(data)) + .into_record_iter() + .count() as u64; + let elems = BgpkitParser::from_reader(black_box(data)) + .into_elem_iter() + .count() as u64; + let updates = BgpkitParser::from_reader(black_box(data)) + .into_update_iter() + .count() as u64; + let routes = BgpkitParser::from_reader(black_box(data)) + .into_route_iter() + .count() as u64; + Counts { + records, + elems, + updates, + routes, + } +} - c.bench_function("updates into_record_iter", |b| { +fn bench_corpus(c: &mut Criterion, label: &str, data: &[u8]) { + let counts = count_items(data); + println!( + "{label}: {} bytes in memory, {} records, {} elems, {} updates, {} routes", + data.len(), + counts.records, + counts.elems, + counts.updates, + counts.routes + ); + + let mut group = c.benchmark_group(label); + group.measurement_time(Duration::from_secs(30)); + + group.throughput(Throughput::Elements(counts.records)); + group.bench_function("into_raw_record_iter", |b| { b.iter(|| { - let mut reader = black_box(&updates[..]); - + let mut reader = black_box(data); BgpkitParser::from_reader(&mut reader) - .into_record_iter() - .take(RECORD_LIMIT) + .into_raw_record_iter() .for_each(|x| { black_box(x); }); }) }); - - c.bench_function("updates into_elem_iter", |b| { + group.bench_function("into_record_iter", |b| { b.iter(|| { - let mut reader = black_box(&updates[..]); - + let mut reader = black_box(data); BgpkitParser::from_reader(&mut reader) - .into_elem_iter() - .take(RECORD_LIMIT) + .into_record_iter() .for_each(|x| { black_box(x); }); }) }); - - // Record-to-elem conversion on its own: the records are parsed during setup, so only the - // Elementor is timed. - let update_records: Vec<_> = BgpkitParser::from_reader(&updates[..]) - .into_record_iter() - .take(RECORD_LIMIT) - .collect(); - c.bench_function("updates elementor record_to_elems_iter", |b| { - b.iter_batched( - || update_records.clone(), - |records| { - let elementor = Elementor::new(); - for record in records { - if let Ok(elems) = elementor.record_to_elems_iter(record) { - elems.for_each(|x| { - black_box(x); - }); - } - } - }, - BatchSize::LargeInput, - ) - }); - - c.bench_function("updates into_update_iter", |b| { + // The update iterator yields one item per UPDATE record, not one per parsed record. + group.throughput(Throughput::Elements(counts.updates)); + group.bench_function("into_update_iter", |b| { b.iter(|| { - let mut reader = black_box(&updates[..]); - + let mut reader = black_box(data); BgpkitParser::from_reader(&mut reader) .into_update_iter() - .take(RECORD_LIMIT) .for_each(|x| { black_box(x); }); }) }); - c.bench_function("updates into_route_iter", |b| { + group.throughput(Throughput::Elements(counts.elems)); + group.bench_function("into_elem_iter", |b| { b.iter(|| { - let mut reader = black_box(&updates[..]); - + let mut reader = black_box(data); BgpkitParser::from_reader(&mut reader) - .into_route_iter() - .take(RECORD_LIMIT) + .into_elem_iter() .for_each(|x| { black_box(x); }); }) }); - c.bench_function("updates into_raw_record_iter", |b| { + group.throughput(Throughput::Elements(counts.routes)); + group.bench_function("into_route_iter", |b| { b.iter(|| { - let mut reader = black_box(&updates[..]); - + let mut reader = black_box(data); BgpkitParser::from_reader(&mut reader) - .into_raw_record_iter() - .take(RECORD_LIMIT) + .into_route_iter() .for_each(|x| { black_box(x); }); }) }); - c.bench_function("rib into_record_iter", |b| { - b.iter(|| { - let mut reader = black_box(&rib_dump[..]); - - BgpkitParser::from_reader(&mut reader) - .into_record_iter() - .take(RECORD_LIMIT) - .for_each(|x| { - black_box(x); - }); - }) + // Record-to-elem conversion on its own: the records are parsed during setup, so only the + // Elementor is timed. + // + // TableDumpV2 RIB records can only be converted when the Elementor holds the peer index + // table, so the table record is lifted out of the parsed prefix and given to one Elementor. + let mut records: Vec<_> = BgpkitParser::from_reader(black_box(data)) + .into_record_iter() + .take(ELEMENTOR_RECORDS) + .collect(); + let peer_table = records.iter().position(|record| { + matches!( + &record.message, + MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(_)) + ) }); - - c.bench_function("rib into_elem_iter", |b| { - b.iter(|| { - let mut reader = black_box(&rib_dump[..]); - - BgpkitParser::from_reader(&mut reader) - .into_elem_iter() - .take(RECORD_LIMIT) - .for_each(|x| { - black_box(x); - }); + let elementor = match peer_table { + Some(index) => match records.remove(index).message { + MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(table)) => { + Elementor::with_peer_table(table) + } + _ => Elementor::new(), + }, + None => Elementor::new(), + }; + let elem_count: u64 = records + .iter() + .map(|record| { + elementor + .record_to_elems_iter(record.clone()) + .map(|iter| iter.count() as u64) + .unwrap_or(0) }) + .sum(); + println!("elementor: {} records, {} elems", records.len(), elem_count); + assert!( + elem_count > 0, + "elementor produced no elems (missing peer index table?); refusing a made-up throughput" + ); + group.throughput(Throughput::Elements(elem_count)); + group.bench_function("elementor record_to_elems_iter", |b| { + b.iter_batched( + || records.clone(), + |records| { + for record in records { + if let Ok(elems) = elementor.record_to_elems_iter(record) { + elems.for_each(|x| { + black_box(x); + }); + } + } + }, + BatchSize::LargeInput, + ) }); - c.bench_function("rib into_update_iter", |b| { - b.iter(|| { - let mut reader = black_box(&rib_dump[..]); - - BgpkitParser::from_reader(&mut reader) - .into_update_iter() - .take(RECORD_LIMIT) - .for_each(|x| { - black_box(x); - }); - }) - }); + group.finish(); +} - c.bench_function("rib into_route_iter", |b| { +/// End-to-end benchmark against a real archive: open, decompress, parse, but stop after a fixed +/// number of elems so the iteration time stays bounded. +fn bench_file(c: &mut Criterion, label: &str, path: &Path, elems: usize) { + let path_str = path + .to_str() + .expect("corpus path is valid UTF-8") + .to_string(); + let available = BgpkitParser::new(path_str.as_str()) + .expect("open corpus") + .into_elem_iter() + .take(elems) + .count() as u64; + println!( + "{label}: {} ({:.1} MiB on disk), streaming {} elems per iteration", + path.display(), + std::fs::metadata(path) + .map(|m| m.len() as f64 / 1_048_576.0) + .unwrap_or(0.0), + available + ); + + assert!( + available > 0, + "no elems streamed from {}; check the corpus and BGPKIT_BENCH_STREAM_ELEMS", + path.display() + ); + let mut group = c.benchmark_group(label); + group.sample_size(10); + group.measurement_time(Duration::from_secs(30)); + group.throughput(Throughput::Elements(available)); + group.bench_function("into_elem_iter (file)", |b| { b.iter(|| { - let mut reader = black_box(&rib_dump[..]); - - BgpkitParser::from_reader(&mut reader) - .into_route_iter() - .take(RECORD_LIMIT) - .for_each(|x| { - black_box(x); - }); + let parser = BgpkitParser::new(black_box(path_str.as_str())).expect("open corpus"); + parser.into_elem_iter().take(elems).for_each(|x| { + black_box(x); + }); }) }); + group.finish(); +} - c.bench_function("rib into_raw_record_iter", |b| { - b.iter(|| { - let mut reader = black_box(&rib_dump[..]); +pub fn criterion_benchmark(c: &mut Criterion) { + let update_data = data_source::test_data_file("update-example.gz"); + let rib_data = data_source::test_data_file("rib-example-small.bz2"); - BgpkitParser::from_reader(&mut reader) - .into_raw_record_iter() - .take(RECORD_LIMIT) - .for_each(|x| { - black_box(x); - }); - }) - }); + println!("Decompressing input data and loading into memory..."); + let updates = preload(GzDecoder::new(BufReader::new( + File::open(&update_data).expect("open update fixture"), + ))); + let rib_dump = preload(BzDecoder::new(BufReader::new( + File::open(&rib_data).expect("open rib fixture"), + ))); + println!( + "Update file: {} ({} bytes decompressed)", + update_data.display(), + updates.len() + ); + println!( + "RIB file: {} ({} bytes decompressed)", + rib_data.display(), + rib_dump.len() + ); + + bench_corpus(c, "updates", &updates); + bench_corpus(c, "rib", &rib_dump); + + let stream_elems = std::env::var("BGPKIT_BENCH_STREAM_ELEMS") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(STREAM_ELEMS); + if let Some(path) = data_source::env_corpus("BGPKIT_BENCH_RIB") { + bench_file(c, "real rib", &path, stream_elems); + } + if let Some(path) = data_source::env_corpus("BGPKIT_BENCH_UPDATES") { + bench_file(c, "real updates", &path, stream_elems); + } } criterion_group! { diff --git a/examples/perf_workload.rs b/examples/perf_workload.rs new file mode 100644 index 00000000..adb5d738 --- /dev/null +++ b/examples/perf_workload.rs @@ -0,0 +1,63 @@ +//! Fixed-workload parse loop used to measure parser changes. +//! +//! Decompresses the corpus once, then parses the in-memory buffer `iters` times, so the reported +//! cost covers parsing only. Run it under `perf stat -e instructions` (or `/usr/bin/time`) and +//! compare revisions on the same machine: +//! +//! ```sh +//! cargo build --release --example perf_workload +//! perf stat -e instructions target/release/examples/perf_workload \ +//! /path/to/rib-example-small.bz2 elems 10 # or records, raw +//! ``` +//! +//! Fixed startup (open, decompress) is included once; run two iteration counts and use +//! `P = (I(10) - I(2)) / 8` to get the per-parse cost without it. + +use bgpkit_parser::BgpkitParser; +use std::hint::black_box; +use std::io::Read; + +fn main() { + let mut args = std::env::args().skip(1); + let path = args + .next() + .expect("usage: perf_workload "); + let mode = args.next().expect("records or elems"); + let iters: usize = args.next().unwrap_or_else(|| "1".into()).parse().unwrap(); + + let mut data = Vec::new(); + oneio::get_reader(&path) + .expect("open corpus") + .read_to_end(&mut data) + .expect("read corpus"); + + let mut total = 0usize; + for _ in 0..iters { + match mode.as_str() { + "raw" => { + let mut reader = black_box(data.as_slice()); + total += BgpkitParser::from_reader(&mut reader) + .into_raw_record_iter() + .count(); + } + "records" => { + let mut reader = black_box(data.as_slice()); + total += BgpkitParser::from_reader(&mut reader) + .into_record_iter() + .count(); + } + "elems" => { + let mut reader = black_box(data.as_slice()); + total += BgpkitParser::from_reader(&mut reader) + .into_elem_iter() + .map(|elem| { + black_box(elem); + 1 + }) + .sum::(); + } + other => panic!("unknown mode {other}"), + } + } + black_box(total); +} diff --git a/src/parser/bgp/attributes/attr_01_origin.rs b/src/parser/bgp/attributes/attr_01_origin.rs index 94f9cbca..34d38fcc 100644 --- a/src/parser/bgp/attributes/attr_01_origin.rs +++ b/src/parser/bgp/attributes/attr_01_origin.rs @@ -4,7 +4,7 @@ use crate::ParserError; use bytes::Bytes; use std::convert::TryFrom; -pub fn parse_origin(mut input: Bytes) -> Result { +pub(crate) fn parse_origin(mut input: &[u8]) -> Result { match Origin::try_from(input.read_u8()?) { Ok(v) => Ok(AttributeValue::Origin(v)), Err(_) => Err(ParserError::ParseError( @@ -44,18 +44,18 @@ mod tests { fn test_parse_origin() { assert_eq!( AttributeValue::Origin(Origin::IGP), - parse_origin(Bytes::from_static(&[0u8])).unwrap() + parse_origin(&[0u8]).unwrap() ); assert_eq!( AttributeValue::Origin(Origin::EGP), - parse_origin(Bytes::from_static(&[1u8])).unwrap() + parse_origin(&[1u8]).unwrap() ); assert_eq!( AttributeValue::Origin(Origin::INCOMPLETE), - parse_origin(Bytes::from_static(&[2u8])).unwrap() + parse_origin(&[2u8]).unwrap() ); assert!(matches!( - parse_origin(Bytes::from_static(&[3u8])).unwrap_err(), + parse_origin(&[3u8]).unwrap_err(), ParserError::ParseError(_) )); } @@ -65,19 +65,19 @@ mod tests { let value = Origin::IGP; assert_eq!( AttributeValue::Origin(value), - parse_origin(encode_origin(&value)).unwrap() + parse_origin(&encode_origin(&value)).unwrap() ); let value = Origin::EGP; assert_eq!( AttributeValue::Origin(value), - parse_origin(encode_origin(&value)).unwrap() + parse_origin(&encode_origin(&value)).unwrap() ); let value = Origin::INCOMPLETE; assert_eq!( AttributeValue::Origin(value), - parse_origin(encode_origin(&value)).unwrap() + parse_origin(&encode_origin(&value)).unwrap() ); } } diff --git a/src/parser/bgp/attributes/attr_02_17_as_path.rs b/src/parser/bgp/attributes/attr_02_17_as_path.rs index 9cfa47d7..8475d7b7 100644 --- a/src/parser/bgp/attributes/attr_02_17_as_path.rs +++ b/src/parser/bgp/attributes/attr_02_17_as_path.rs @@ -11,9 +11,22 @@ const AS_PATH_AS_SEQUENCE: u8 = 2; const AS_PATH_CONFED_SEQUENCE: u8 = 3; const AS_PATH_CONFED_SET: u8 = 4; -pub fn parse_as_path(mut input: Bytes, asn_len: &AsnLength) -> Result { +pub fn parse_as_path(input: Bytes, asn_len: &AsnLength) -> Result { + parse_as_path_slice(input.as_ref(), asn_len) +} + +/// [`parse_as_path`] over a borrowed slice. +/// +/// AS_PATH is the attribute that appears in essentially every record, so it decodes straight from +/// the record body instead of taking an owned `Bytes` per attribute. +pub(crate) fn parse_as_path_slice( + mut input: &[u8], + asn_len: &AsnLength, +) -> Result { let mut output = AsPath { - segments: SmallVec::with_capacity(5), + // Segments are inline up to the SmallVec capacity; pre-sizing to the common + // multi-segment case would allocate for the single-segment paths that dominate. + segments: SmallVec::new(), }; while input.remaining() > 0 { let segment = parse_as_path_segment(&mut input, asn_len)?; @@ -23,8 +36,34 @@ pub fn parse_as_path(mut input: Bytes, asn_len: &AsnLength) -> Result Result, ParserError> { + let mut asns: SmallVec<[Asn; 6]> = SmallVec::with_capacity(count); + match asn_len { + AsnLength::Bits16 => { + for _ in 0..count { + asns.push(Asn::new_16bit(input.read_u16()?)); + } + } + AsnLength::Bits32 => { + for _ in 0..count { + asns.push(Asn::new_32bit(input.read_u32()?)); + } + } + } + Ok(asns) +} + fn parse_as_path_segment( - input: &mut Bytes, + input: &mut &[u8], asn_len: &AsnLength, ) -> Result { let segment_type = input.read_u8()?; @@ -43,7 +82,7 @@ fn parse_as_path_segment( ))); } - let path: SmallVec<[Asn; 6]> = input.read_asns(asn_len, count)?.into(); + let path: SmallVec<[Asn; 6]> = read_asns_into(input, asn_len, count)?; match segment_type { AS_PATH_AS_SET => Ok(AsPathSegment::AsSet(path)), AS_PATH_AS_SEQUENCE => Ok(AsPathSegment::AsSequence(path)), @@ -154,62 +193,62 @@ mod tests { ////////////////////// // 16 bits sequence // ////////////////////// - let mut data = Bytes::from(vec![ + let data = Bytes::from(vec![ 2, // sequence 3, // 3 ASes in path 0, 1, // AS1 0, 2, // AS2 0, 3, // AS3 ]); - let res = parse_as_path_segment(&mut data, &AsnLength::Bits16).unwrap(); + let res = parse_as_path_segment(&mut data.as_ref(), &AsnLength::Bits16).unwrap(); assert_eq!(res, AsPathSegment::sequence([1, 2, 3])); ////////////////////// // 32 bits sequence // ////////////////////// - let mut data = Bytes::from(vec![ + let data = Bytes::from(vec![ 2, // sequence 3, // 3 ASes in path 0, 0, 0, 1, // AS1 0, 0, 0, 2, // AS2 0, 0, 0, 3, // AS3 ]); - let res = parse_as_path_segment(&mut data, &AsnLength::Bits32).unwrap(); + let res = parse_as_path_segment(&mut data.as_ref(), &AsnLength::Bits32).unwrap(); assert_eq!(res, AsPathSegment::sequence([1, 2, 3])); ///////////////// // other types // ///////////////// - let mut data = Bytes::from(vec![ + let data = Bytes::from(vec![ 1, // AS Set 1, // 1 AS in path 0, 1, ]); - let res = parse_as_path_segment(&mut data, &AsnLength::Bits16).unwrap(); + let res = parse_as_path_segment(&mut data.as_ref(), &AsnLength::Bits16).unwrap(); assert_eq!(res, AsPathSegment::set([1])); - let mut data = Bytes::from(vec![ + let data = Bytes::from(vec![ 3, // Confed Sequence 1, // 1 AS in path 0, 1, ]); - let res = parse_as_path_segment(&mut data, &AsnLength::Bits16).unwrap(); + let res = parse_as_path_segment(&mut data.as_ref(), &AsnLength::Bits16).unwrap(); assert!(matches!(res, AsPathSegment::ConfedSequence(_))); - let mut data = Bytes::from(vec![ + let data = Bytes::from(vec![ 4, // Confed Set 1, // 1 AS in path 0, 1, ]); - let res = parse_as_path_segment(&mut data, &AsnLength::Bits16).unwrap(); + let res = parse_as_path_segment(&mut data.as_ref(), &AsnLength::Bits16).unwrap(); assert!(matches!(res, AsPathSegment::ConfedSet(_))); - let mut data = Bytes::from(vec![ + let data = Bytes::from(vec![ 5, // ERROR 1, // 1 AS in path 0, 1, ]); - let res = parse_as_path_segment(&mut data, &AsnLength::Bits16).unwrap_err(); + let res = parse_as_path_segment(&mut data.as_ref(), &AsnLength::Bits16).unwrap_err(); assert!(matches!(res, ParserError::ParseError(_))); } diff --git a/src/parser/bgp/attributes/attr_03_next_hop.rs b/src/parser/bgp/attributes/attr_03_next_hop.rs index b33d3114..597f237d 100644 --- a/src/parser/bgp/attributes/attr_03_next_hop.rs +++ b/src/parser/bgp/attributes/attr_03_next_hop.rs @@ -4,7 +4,10 @@ use crate::ParserError; use bytes::{Bytes, BytesMut}; use std::net::IpAddr; -pub fn parse_next_hop(mut input: Bytes, afi: &Option) -> Result { +pub(crate) fn parse_next_hop( + mut input: &[u8], + afi: &Option, +) -> Result { if let Some(afi) = afi { Ok(input.read_address(afi).map(AttributeValue::NextHop)?) } else { @@ -112,12 +115,12 @@ mod tests { let ipv4_bytes = Bytes::from(ipv4.octets().to_vec()); let ipv6_bytes = Bytes::from(ipv6.octets().to_vec()); - let res = parse_next_hop(ipv4_bytes, &None).unwrap(); + let res = parse_next_hop(&ipv4_bytes, &None).unwrap(); if let AttributeValue::NextHop(n) = res { assert_eq!(n.to_string(), "10.0.0.1".to_string()) } - let res = parse_next_hop(ipv6_bytes, &Some(Afi::Ipv6)).unwrap(); + let res = parse_next_hop(&ipv6_bytes, &Some(Afi::Ipv6)).unwrap(); if let AttributeValue::NextHop(n) = res { assert_eq!(n.to_string().to_ascii_uppercase(), "FC00::1".to_string()) } @@ -130,9 +133,9 @@ mod tests { let ipv4_bytes = Bytes::from(ipv4.octets().to_vec()); let ipv6_bytes = Bytes::from(ipv6.octets().to_vec()); - let _res = parse_next_hop(ipv4_bytes.clone(), &None).unwrap(); + let _res = parse_next_hop(&ipv4_bytes, &None).unwrap(); assert_eq!(ipv4_bytes, encode_next_hop(&ipv4.into())); - let _res = parse_next_hop(ipv6_bytes.clone(), &None).unwrap(); + let _res = parse_next_hop(&ipv6_bytes, &None).unwrap(); assert_eq!(ipv6_bytes, encode_next_hop(&ipv6.into())); } diff --git a/src/parser/bgp/attributes/attr_04_med.rs b/src/parser/bgp/attributes/attr_04_med.rs index a29bfb16..dba559d1 100644 --- a/src/parser/bgp/attributes/attr_04_med.rs +++ b/src/parser/bgp/attributes/attr_04_med.rs @@ -3,7 +3,7 @@ use crate::parser::ReadUtils; use crate::ParserError; use bytes::Bytes; -pub fn parse_med(mut input: Bytes) -> Result { +pub(crate) fn parse_med(mut input: &[u8]) -> Result { Ok(AttributeValue::MultiExitDiscriminator(input.read_u32()?)) } @@ -18,7 +18,7 @@ mod tests { #[test] fn test_parse_med() { assert_eq!( - parse_med(Bytes::from(vec![0, 0, 0, 123])).unwrap(), + parse_med(&[0, 0, 0, 123]).unwrap(), AttributeValue::MultiExitDiscriminator(123) ); } diff --git a/src/parser/bgp/attributes/attr_05_local_pref.rs b/src/parser/bgp/attributes/attr_05_local_pref.rs index 7618bdec..c5f46e56 100644 --- a/src/parser/bgp/attributes/attr_05_local_pref.rs +++ b/src/parser/bgp/attributes/attr_05_local_pref.rs @@ -3,7 +3,7 @@ use crate::parser::ReadUtils; use crate::ParserError; use bytes::Bytes; -pub fn parse_local_pref(mut input: Bytes) -> Result { +pub(crate) fn parse_local_pref(mut input: &[u8]) -> Result { Ok(AttributeValue::LocalPreference(input.read_u32()?)) } @@ -18,7 +18,7 @@ mod tests { #[test] fn test_parse_local_pref() { assert_eq!( - parse_local_pref(Bytes::from(vec![0, 0, 0, 123])).unwrap(), + parse_local_pref(&[0, 0, 0, 123]).unwrap(), AttributeValue::LocalPreference(123) ); } diff --git a/src/parser/bgp/attributes/mod.rs b/src/parser/bgp/attributes/mod.rs index 6c740219..ca5c1e21 100644 --- a/src/parser/bgp/attributes/mod.rs +++ b/src/parser/bgp/attributes/mod.rs @@ -31,7 +31,9 @@ use crate::encoder::sink::{with_u16_len, with_u8_len}; use crate::error::{BgpValidationWarning, EncodingError, ParserError}; use crate::parser::bgp::attributes::attr_01_origin::{encode_origin, parse_origin}; use crate::parser::bgp::attributes::attr_02_17_as_path::encode_as_path; -pub(crate) use crate::parser::bgp::attributes::attr_02_17_as_path::parse_as_path; +pub(crate) use crate::parser::bgp::attributes::attr_02_17_as_path::{ + parse_as_path, parse_as_path_slice, +}; use crate::parser::bgp::attributes::attr_03_next_hop::{encode_next_hop, parse_next_hop}; use crate::parser::bgp::attributes::attr_04_med::{encode_med, parse_med}; use crate::parser::bgp::attributes::attr_05_local_pref::{encode_local_pref, parse_local_pref}; @@ -445,10 +447,17 @@ pub fn parse_attributes( safi: Option, prefixes: Option<&[NetworkPrefix]>, ) -> Result { - // Estimate capacity from data size: each attribute is at least 3 bytes - // (flag + type + length). Cap at 256 to avoid over-allocation for corrupted data. - let estimated_attrs = (data.remaining() / 3).min(256); - let mut attributes: Vec = Vec::with_capacity(estimated_attrs.max(8)); + // Estimate capacity from data size. A wire attribute is at least three bytes (flags, type, + // length) and real attributes average far larger, so `len / 3` claimed one slot per three + // bytes and over-allocated by an order of magnitude on short records. The floor keeps the + // common five-attribute record from growing; the cap keeps a corrupt section from reserving + // hundreds of slots. An empty section (withdrawals, End-of-RIB) reserves nothing. + let estimated_attrs = if data.is_empty() { + 0 + } else { + (data.remaining() / 16).clamp(8, 32) + }; + let mut attributes: Vec = Vec::with_capacity(estimated_attrs); let mut validation = AttributeValidationState::new(); // A handle on the whole attribute section, so an attribute that fails to parse can be kept // raw by re-slicing it, without cloning every attribute's bytes up front. @@ -489,95 +498,122 @@ pub fn parse_attributes( // we know data has enough bytes to read, so we can split the bytes into a new Bytes object data.has_n_remaining(attr_length)?; let value_start = section.len() - data.len(); - let mut attr_data = data.split_to(attr_length); let raw_code = u8::from(attr_type); - if let Some(t) = get_deprecated_attr_type(raw_code) { - debug!("deprecated attribute type: {} - {}", raw_code, t); - attributes.push(Attribute { - value: AttributeValue::Deprecated(AttrRaw { - code: raw_code, - bytes: attr_data, - }), - flag, - }); - continue; - } - - if matches!(attr_type, AttrType::Unknown(_)) { - debug!("unknown attribute type: {}", raw_code); - attributes.push(Attribute { - value: AttributeValue::Unknown(AttrRaw { - code: raw_code, - bytes: attr_data, - }), - flag, - }); - continue; - } - - if is_raw_retained_attr(attr_type) { - debug!("raw-retained attribute type: {}", raw_code); - attributes.push(Attribute { - value: AttributeValue::Raw(AttrRaw { - code: raw_code, - bytes: attr_data, - }), - flag, - }); - continue; - } + // The attributes that make up nearly every record decode from a borrowed slice, which + // skips the per-attribute `Bytes` split and its refcount traffic. Each stays in its own + // branch: routing them through one dispatcher costs more than the split it saves. + let attr = if matches!(attr_type, AttrType::AS_PATH | AttrType::AS4_PATH) { + let value = §ion[value_start..value_start + attr_length]; + let path_len = match attr_type { + AttrType::AS4_PATH => &AsnLength::Bits32, + _ => asn_len, + }; + let parsed = parse_as_path_slice(value, path_len); + data.advance(attr_length); + parsed.map(|path| match attr_type { + AttrType::AS4_PATH => AttributeValue::As4Path(path), + _ => AttributeValue::AsPath(path), + }) + } else if attr_type == AttrType::ORIGIN { + let value = §ion[value_start..value_start + attr_length]; + data.advance(attr_length); + parse_origin(value) + } else if attr_type == AttrType::NEXT_HOP { + let value = §ion[value_start..value_start + attr_length]; + data.advance(attr_length); + parse_next_hop(value, &afi) + } else if attr_type == AttrType::MULTI_EXIT_DISCRIMINATOR { + let value = §ion[value_start..value_start + attr_length]; + data.advance(attr_length); + parse_med(value) + } else if attr_type == AttrType::LOCAL_PREFERENCE { + let value = §ion[value_start..value_start + attr_length]; + data.advance(attr_length); + parse_local_pref(value) + } else { + let mut attr_data = data.split_to(attr_length); - let attr = match attr_type { - AttrType::ORIGIN => parse_origin(attr_data), - AttrType::AS_PATH => parse_as_path(attr_data, asn_len).map(AttributeValue::AsPath), - AttrType::NEXT_HOP => parse_next_hop(attr_data, &afi), - AttrType::MULTI_EXIT_DISCRIMINATOR => parse_med(attr_data), - AttrType::LOCAL_PREFERENCE => parse_local_pref(attr_data), - AttrType::ATOMIC_AGGREGATE => Ok(AttributeValue::AtomicAggregate), - AttrType::AGGREGATOR => parse_aggregator(attr_data, asn_len) - .map(|(asn, id)| AttributeValue::Aggregator { asn, id }), - AttrType::ORIGINATOR_ID => parse_originator_id(attr_data), - AttrType::CLUSTER_LIST => parse_clusters(attr_data), - AttrType::MP_REACHABLE_NLRI => { - parse_nlri(attr_data, &afi, &safi, &prefixes, true, add_path) - } - AttrType::MP_UNREACHABLE_NLRI => { - parse_nlri(attr_data, &afi, &safi, &prefixes, false, add_path) + if let Some(t) = get_deprecated_attr_type(raw_code) { + debug!("deprecated attribute type: {} - {}", raw_code, t); + attributes.push(Attribute { + value: AttributeValue::Deprecated(AttrRaw { + code: raw_code, + bytes: attr_data, + }), + flag, + }); + continue; } - AttrType::AS4_PATH => { - parse_as_path(attr_data, &AsnLength::Bits32).map(AttributeValue::As4Path) + + if matches!(attr_type, AttrType::Unknown(_)) { + debug!("unknown attribute type: {}", raw_code); + attributes.push(Attribute { + value: AttributeValue::Unknown(AttrRaw { + code: raw_code, + bytes: attr_data, + }), + flag, + }); + continue; } - AttrType::AS4_AGGREGATOR => parse_aggregator(attr_data, &AsnLength::Bits32) - .map(|(asn, id)| AttributeValue::As4Aggregator { asn, id }), - - // communities - AttrType::COMMUNITIES => parse_regular_communities(attr_data), - AttrType::LARGE_COMMUNITIES => parse_large_communities(attr_data), - AttrType::EXTENDED_COMMUNITIES => parse_extended_community(attr_data), - AttrType::IPV6_ADDRESS_SPECIFIC_EXTENDED_COMMUNITIES => { - parse_ipv6_extended_community(attr_data) + + if is_raw_retained_attr(attr_type) { + debug!("raw-retained attribute type: {}", raw_code); + attributes.push(Attribute { + value: AttributeValue::Raw(AttrRaw { + code: raw_code, + bytes: attr_data, + }), + flag, + }); + continue; } - AttrType::DEVELOPMENT => { - let mut value = vec![]; - for _i in 0..attr_length { - value.push(attr_data.read_u8()?); + + match attr_type { + // ORIGIN, NEXT_HOP, MED and LOCAL_PREF are handled above from a borrowed slice. + AttrType::ATOMIC_AGGREGATE => Ok(AttributeValue::AtomicAggregate), + AttrType::AGGREGATOR => parse_aggregator(attr_data, asn_len) + .map(|(asn, id)| AttributeValue::Aggregator { asn, id }), + AttrType::ORIGINATOR_ID => parse_originator_id(attr_data), + AttrType::CLUSTER_LIST => parse_clusters(attr_data), + AttrType::MP_REACHABLE_NLRI => { + parse_nlri(attr_data, &afi, &safi, &prefixes, true, add_path) + } + AttrType::MP_UNREACHABLE_NLRI => { + parse_nlri(attr_data, &afi, &safi, &prefixes, false, add_path) + } + AttrType::AS4_AGGREGATOR => parse_aggregator(attr_data, &AsnLength::Bits32) + .map(|(asn, id)| AttributeValue::As4Aggregator { asn, id }), + + // communities + AttrType::COMMUNITIES => parse_regular_communities(attr_data), + AttrType::LARGE_COMMUNITIES => parse_large_communities(attr_data), + AttrType::EXTENDED_COMMUNITIES => parse_extended_community(attr_data), + AttrType::IPV6_ADDRESS_SPECIFIC_EXTENDED_COMMUNITIES => { + parse_ipv6_extended_community(attr_data) + } + AttrType::DEVELOPMENT => { + let mut value = vec![]; + for _i in 0..attr_length { + value.push(attr_data.read_u8()?); + } + Ok(AttributeValue::Development(value)) } - Ok(AttributeValue::Development(value)) + AttrType::ONLY_TO_CUSTOMER => parse_only_to_customer(attr_data), + AttrType::AIGP => parse_aigp(attr_data), + AttrType::TUNNEL_ENCAPSULATION => parse_tunnel_encapsulation_attribute(attr_data), + AttrType::TRAFFIC_ENGINEERING => parse_traffic_engineering(attr_data), + AttrType::BGP_LS_ATTRIBUTE => parse_link_state_attribute(attr_data), + AttrType::BGP_DOMAIN_PATH => parse_domain_path(attr_data), + AttrType::SFP_ATTRIBUTE => parse_sfp(attr_data), + AttrType::BFD_DISCRIMINATOR => parse_bfd_discriminator(attr_data), + AttrType::BGP_PREFIX_SID => parse_bgp_prefix_sid(attr_data), + AttrType::BIER => parse_bier(attr_data), + _ => Err(ParserError::Unsupported(format!( + "unsupported attribute type: {attr_type:?}" + ))), } - AttrType::ONLY_TO_CUSTOMER => parse_only_to_customer(attr_data), - AttrType::AIGP => parse_aigp(attr_data), - AttrType::TUNNEL_ENCAPSULATION => parse_tunnel_encapsulation_attribute(attr_data), - AttrType::TRAFFIC_ENGINEERING => parse_traffic_engineering(attr_data), - AttrType::BGP_LS_ATTRIBUTE => parse_link_state_attribute(attr_data), - AttrType::BGP_DOMAIN_PATH => parse_domain_path(attr_data), - AttrType::SFP_ATTRIBUTE => parse_sfp(attr_data), - AttrType::BFD_DISCRIMINATOR => parse_bfd_discriminator(attr_data), - AttrType::BGP_PREFIX_SID => parse_bgp_prefix_sid(attr_data), - AttrType::BIER => parse_bier(attr_data), - _ => Err(ParserError::Unsupported(format!( - "unsupported attribute type: {attr_type:?}" - ))), }; match attr { diff --git a/src/parser/mrt/messages/table_dump_v2/rib_afi_entries.rs b/src/parser/mrt/messages/table_dump_v2/rib_afi_entries.rs index 369408b5..0fd3e2e5 100644 --- a/src/parser/mrt/messages/table_dump_v2/rib_afi_entries.rs +++ b/src/parser/mrt/messages/table_dump_v2/rib_afi_entries.rs @@ -70,10 +70,11 @@ pub fn parse_rib_afi_entries( let prefix = data.read_nlri_prefix(&afi, false)?; let entry_count = data.read_u16()?; - // Pre-allocate cautiously to avoid overflow/OOM with malformed inputs + // Pre-allocate cautiously to avoid overflow/OOM with malformed inputs. One slot per entry: + // the RIB wire format never carries more entries than the remaining bytes can express. let min_entry_size = rib_entry_min_len(is_add_path); let max_possible = data.remaining() / min_entry_size; - let reserve = (entry_count as usize).min(max_possible).saturating_mul(2); + let reserve = (entry_count as usize).min(max_possible); let mut rib_entries = Vec::with_capacity(reserve); // get the u8 slice of the rest of the data diff --git a/src/parser/mrt/mrt_header.rs b/src/parser/mrt/mrt_header.rs index 2522e3da..f0195a29 100644 --- a/src/parser/mrt/mrt_header.rs +++ b/src/parser/mrt/mrt_header.rs @@ -132,18 +132,57 @@ pub(crate) struct FramedHeader { pub(crate) header: Result, /// The header bytes as read from the wire. pub(crate) raw_bytes: Bytes, +} + +/// A common header read from the wire, with its raw bytes kept in a fixed buffer. +/// +/// The default parse path only needs the raw header bytes when something goes wrong, so capturing +/// them on the stack avoids one small heap allocation per record. +pub(crate) struct HeaderBuf { + bytes: [u8; 16], + len: usize, +} + +impl HeaderBuf { + pub(crate) fn new() -> Self { + HeaderBuf { + bytes: [0u8; 16], + len: 0, + } + } + + pub(crate) fn as_slice(&self) -> &[u8] { + &self.bytes[..self.len] + } +} + +/// A common header as read from the wire, borrowing its raw bytes from the caller's buffer. +pub(crate) struct FramedHeaderRef<'a> { + /// The parsed header, or why its contents were rejected. + pub(crate) header: Result, + /// The header bytes as read from the wire. + pub(crate) raw_bytes: &'a [u8], /// Length of the record body following `raw_bytes`. pub(crate) body_length: u32, } -/// Read a common header, keeping its body length even when the contents are +/// Read a common header into `header`, keeping its body length even when the contents are /// rejected. Only I/O failures are returned as `Err`. -pub(crate) fn frame_common_header(input: &mut T) -> std::io::Result { - let mut base_bytes = [0u8; 12]; - input.read_exact(&mut base_bytes)?; +pub(crate) fn frame_common_header_captured<'a, T: Read>( + input: &mut T, + header: &'a mut HeaderBuf, +) -> std::io::Result> { + header.len = 0; + read_header_upto(input, header, 12)?; + if header.len < 12 { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "failed to fill whole buffer", + )); + } // Single bounds check via zerocopy instead of four sequential cursor reads. - let raw = RawMrtCommonHeader::ref_from_bytes(&base_bytes) + let raw = RawMrtCommonHeader::ref_from_bytes(&header.bytes[..12]) .expect("base_bytes is exactly 12 bytes with no alignment requirement"); let timestamp = raw.timestamp.get(); @@ -153,41 +192,49 @@ pub(crate) fn frame_common_header(input: &mut T) -> std::io::Result entry_type, - Err(error) => return Ok(rejected(error.into())), + Err(error) => { + return Ok(FramedHeaderRef { + header: Err(error.into()), + raw_bytes: header.as_slice(), + body_length: length, + }) + } }; - let (microsecond_timestamp, raw_bytes, body_length) = match &entry_type { + let (microsecond_timestamp, body_length) = match &entry_type { EntryType::BGP4MP_ET => { // For ET records, the on-wire length includes the extra 4-byte microsecond timestamp // that lives in the header. Internally we store `length` as the message length only, // so subtract 4 after validating to avoid underflow. if length < 4 { - return Ok(rejected(ParserError::ParseError( - "invalid MRT header length for ET record: length < 4".into(), - ))); + return Ok(FramedHeaderRef { + header: Err(ParserError::ParseError( + "invalid MRT header length for ET record: length < 4".into(), + )), + raw_bytes: header.as_slice(), + body_length: length, + }); + } + read_header_upto(input, header, 16)?; + if header.len < 16 { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "failed to fill whole buffer", + )); } - let mut combined = [0u8; 16]; - combined[..12].copy_from_slice(&base_bytes); - input.read_exact(&mut combined[12..])?; - let microseconds = u32::from_be_bytes(combined[12..16].try_into().unwrap()); ( - Some(microseconds), - Bytes::copy_from_slice(&combined), + Some(u32::from_be_bytes( + header.bytes[12..16].try_into().expect("four bytes"), + )), length - 4, ) } - _ => (None, Bytes::copy_from_slice(&base_bytes), length), + _ => (None, length), }; - Ok(FramedHeader { + Ok(FramedHeaderRef { header: Ok(CommonHeader { timestamp, microsecond_timestamp, @@ -195,11 +242,42 @@ pub(crate) fn frame_common_header(input: &mut T) -> std::io::Result std::io::Result<()> { + while header.len < end { + match input.read(&mut header.bytes[header.len..end]) { + Ok(0) => break, + Ok(read) => header.len += read, + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + } + Ok(()) +} + +/// Read a common header, keeping its body length even when the contents are +/// rejected. Only I/O failures are returned as `Err`. +pub(crate) fn frame_common_header(input: &mut T) -> std::io::Result { + let mut header = HeaderBuf::new(); + let framed = frame_common_header_captured(input, &mut header)?; + Ok(FramedHeader { + header: framed.header, + raw_bytes: Bytes::copy_from_slice(framed.raw_bytes), + }) +} + impl CommonHeader { pub fn encode(&self) -> Bytes { let raw = RawMrtHeader::from(self); diff --git a/src/parser/mrt/mrt_record.rs b/src/parser/mrt/mrt_record.rs index 9ea51ee7..b3c48fb0 100644 --- a/src/parser/mrt/mrt_record.rs +++ b/src/parser/mrt/mrt_record.rs @@ -1,4 +1,4 @@ -use super::mrt_header::frame_common_header; +use super::mrt_header::{frame_common_header_captured, HeaderBuf}; use crate::bmp::messages::{BmpMessage, BmpMessageBody}; use crate::error::{check_max, EncodingError, ParserError}; use crate::models::*; @@ -134,19 +134,41 @@ pub fn chunk_mrt_record(input: &mut impl Read) -> Result, +} + +impl ChunkedRecord { + fn header_bytes(&self) -> &[u8] { + self.header.as_slice() + } +} + /// Read one raw MRT record while retaining parsed header context on framing errors. pub(crate) fn chunk_mrt_record_with_context( input: &mut impl Read, ) -> Result { - // read the common header and capture raw bytes - let mut consumed_header = Vec::with_capacity(16); - let framed = match frame_common_header(&mut CapturingReader { - inner: input, - captured: &mut consumed_header, - }) { + let chunk = read_chunk(input)?; + Ok(RawMrtRecord { + common_header: chunk.common_header, + header_bytes: Bytes::copy_from_slice(chunk.header_bytes()), + message_bytes: Bytes::from(chunk.body), + }) +} + +/// Read one MRT record, keeping the header bytes in a stack buffer. +fn read_chunk(input: &mut impl Read) -> Result { + let mut header = HeaderBuf::new(); + let framed = match frame_common_header_captured(input, &mut header) { Ok(framed) => framed, Err(error) => { - if error.kind() == std::io::ErrorKind::UnexpectedEof && consumed_header.is_empty() { + if error.kind() == std::io::ErrorKind::UnexpectedEof && header.as_slice().is_empty() { return Err(RawMrtRecordError { error: ParserError::EofExpected, common_header: None, @@ -157,7 +179,7 @@ pub(crate) fn chunk_mrt_record_with_context( return Err(RawMrtRecordError { error: error.into(), common_header: None, - bytes: Some(consumed_header), + bytes: Some(header.as_slice().to_vec()), record_consumed: false, }); } @@ -184,12 +206,7 @@ pub(crate) fn chunk_mrt_record_with_context( .take(framed.body_length as u64) .read_to_end(&mut buffer) { - return Err(record_io_error( - error, - parsed_header, - &header_bytes, - &buffer, - )); + return Err(record_io_error(error, parsed_header, header_bytes, &buffer)); } if buffer.len() != framed.body_length as usize { return Err(record_io_error( @@ -202,7 +219,7 @@ pub(crate) fn chunk_mrt_record_with_context( ), ), parsed_header, - &header_bytes, + header_bytes, &buffer, )); } @@ -214,7 +231,7 @@ pub(crate) fn chunk_mrt_record_with_context( return Err(RawMrtRecordError { error, common_header: None, - bytes: Some([&header_bytes[..], &buffer].concat()), + bytes: Some([header_bytes, &buffer].concat()), record_consumed: true, }) } @@ -232,7 +249,7 @@ pub(crate) fn chunk_mrt_record_with_context( return Err(record_io_error( error, Some(common_header), - &header_bytes, + header_bytes, &buffer, )); } @@ -244,7 +261,7 @@ pub(crate) fn chunk_mrt_record_with_context( "truncated historical TABLE_DUMP length correction", ), Some(common_header), - &header_bytes, + header_bytes, &buffer, )); } @@ -256,26 +273,13 @@ pub(crate) fn chunk_mrt_record_with_context( ); } - Ok(RawMrtRecord { + Ok(ChunkedRecord { common_header, - header_bytes, - message_bytes: Bytes::from(buffer), + header, + body: buffer, }) } -struct CapturingReader<'a, R: ?Sized> { - inner: &'a mut R, - captured: &'a mut Vec, -} - -impl Read for CapturingReader<'_, R> { - fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { - let read = self.inner.read(buffer)?; - self.captured.extend_from_slice(&buffer[..read]); - Ok(read) - } -} - fn record_io_error( error: std::io::Error, common_header: Option, @@ -310,8 +314,51 @@ pub(crate) fn raw_record_uses_zebra_compat(raw_record: &RawMrtRecord) -> bool { pub(crate) fn parse_mrt_record_with_zebra_compat( input: &mut impl Read, ) -> Result<(MrtRecord, bool), ParserErrorWithBytes> { - let raw_record = chunk_mrt_record(input)?; - parse_chunked_record_with_zebra_compat(raw_record) + let chunk = read_chunk(input).map_err(|error| ParserErrorWithBytes { + error: error.error, + bytes: error.bytes, + })?; + + let common_header = chunk.common_header; + // Only the error path needs the wire bytes; cloning the `Bytes` handle below is a refcount + // bump, while keeping the header on the stack avoids a per-record allocation on success. + let mut header = [0u8; 16]; + let header_bytes = chunk.header_bytes(); + let header_len = header_bytes.len(); + header[..header_len].copy_from_slice(header_bytes); + let body = Bytes::from(chunk.body); + let wire_bytes = body.clone(); + + let used_zebra_compat = matches!( + common_header.entry_type, + EntryType::BGP4MP | EntryType::BGP4MP_ET + ) && crate::parser::mrt::messages::bgp4mp::uses_zebra_compat( + common_header.entry_subtype, + &body, + ); + + match parse_mrt_body( + common_header.entry_type as u16, + common_header.entry_subtype, + body, + ) { + Ok(message) => Ok(( + MrtRecord { + common_header, + message, + }, + used_zebra_compat, + )), + Err(error) => { + let mut bytes = Vec::with_capacity(header_len + wire_bytes.len()); + bytes.extend_from_slice(&header[..header_len]); + bytes.extend_from_slice(&wire_bytes); + Err(ParserErrorWithBytes { + error, + bytes: Some(bytes), + }) + } + } } /// Parse an already-chunked raw record instead of reading the next one. @@ -837,6 +884,94 @@ mod tests { assert!(matches!(error, ParserError::TruncatedMsg(_))); } + /// A reader that returns `data` up to `fail_after` bytes and then raises an I/O error. + struct FailingReader { + data: Vec, + pos: usize, + fail_after: usize, + } + + impl Read for FailingReader { + fn read(&mut self, out: &mut [u8]) -> std::io::Result { + if self.pos >= self.fail_after { + return Err(std::io::Error::other("reader exploded")); + } + let n = (self.fail_after - self.pos) + .min(out.len()) + .min(self.data.len() - self.pos); + out[..n].copy_from_slice(&self.data[self.pos..self.pos + n]); + self.pos += n; + Ok(n) + } + } + + /// Bytes read before a mid-header I/O error must stay in the reported error bytes, including + /// the microseconds of an ET header. + #[test] + fn chunk_errors_keep_partial_extended_timestamp_on_io_error() { + let mut wire = BytesMut::new(); + wire.put_u32(1); // timestamp + wire.put_u16(EntryType::BGP4MP_ET as u16); + wire.put_u16(4); // subtype + wire.put_u32(9); // length, including the 4-byte microsecond field + wire.put_u32(230_000); // microseconds + let wire = wire.freeze(); + + for consumed in [13usize, 14, 15] { + let mut reader = FailingReader { + data: wire.to_vec(), + pos: 0, + fail_after: consumed, + }; + let error = chunk_mrt_record(&mut reader).unwrap_err(); + assert_eq!( + error.bytes.as_deref(), + Some(&wire[..consumed]), + "an I/O error after {consumed} header bytes must report those bytes" + ); + } + } + + /// A reader that yields `data` and then reports a clean end of stream. + struct ShortReader { + data: Vec, + pos: usize, + } + + impl Read for ShortReader { + fn read(&mut self, out: &mut [u8]) -> std::io::Result { + let n = out.len().min(self.data.len() - self.pos); + out[..n].copy_from_slice(&self.data[self.pos..self.pos + n]); + self.pos += n; + Ok(n) + } + } + + /// The same partial-header accounting applies when the stream simply ends. + #[test] + fn chunk_errors_keep_partial_extended_timestamp_on_clean_eof() { + let mut wire = BytesMut::new(); + wire.put_u32(1); + wire.put_u16(EntryType::BGP4MP_ET as u16); + wire.put_u16(4); + wire.put_u32(9); + wire.put_u32(230_000); + let wire = wire.freeze(); + + for consumed in [13usize, 14, 15] { + let mut reader = ShortReader { + data: wire[..consumed].to_vec(), + pos: 0, + }; + let error = chunk_mrt_record(&mut reader).unwrap_err(); + assert_eq!( + error.bytes.as_deref(), + Some(&wire[..consumed]), + "an end of stream after {consumed} header bytes must report those bytes" + ); + } + } + #[test] fn chunk_errors_include_invalid_header_and_partial_body_bytes() { let invalid_header = Bytes::from_static(&[0, 0, 0, 1, 0xff, 0xff, 0, 0, 0, 0, 0, 0]); diff --git a/src/parser/utils.rs b/src/parser/utils.rs index 11cbfc52..ae335de2 100644 --- a/src/parser/utils.rs +++ b/src/parser/utils.rs @@ -18,6 +18,9 @@ use std::net::IpAddr; impl ReadUtils for Bytes {} +// Attribute values that decode straight from a borrowed record body. +impl ReadUtils for &[u8] {} + // Allow reading IPs from Reads pub trait ReadUtils: Buf { #[inline]