Performance Improvements

Cargo.toml

@@ -28,11 +28,12 @@ serde={version="1.0", features=["derive"], optional = true}
 itertools = {version = "0.11.0", optional = true}
 ipnet = {version="2.7", optional = true}
 bitflags = {version="2.3.3", features = ["serde"], optional = true}
+thiserror = {version = "1.0.44", optional = true}
+smallvec = {version = "1.11.0", features = ["union"], optional = true}
 
 #######################
 # Parser dependencies #
 #######################
-bytes = {version = "1.4.0", optional = true}
 hex= {version = "0.4.3", optional = true} # bmp/openbmp parsing
 log= {version = "0.4", optional = true }
 oneio = {version= "0.11.0", features=["lib"], optional = true }
@@ -54,9 +55,10 @@ models = [
     "ipnet",
     "itertools",
     "bitflags",
+    "thiserror",
+    "smallvec",
 ]
 parser = [
-    "bytes",
     "chrono",
     "env_logger",
     "log",
@@ -80,16 +82,25 @@ rislive = [
 serde = [
     "dep:serde",
     "ipnet/serde",
+    "smallvec/serde",
 ]
 
 [[bench]]
 name = "internals"
 harness = false
 
+[[bench]]
+name = "mrt_type"
+harness = false
+
 [[bench]]
 name = "bench_main"
 harness = false
 
+# Enable debug symbols for benchmarks for easier profiling
+[profile.bench]
+debug = true
+
 [dev-dependencies]
 anyhow = "1"
 bgpkit-broker = "0.6.2"

benches/mrt_type.rs

@@ -0,0 +1,114 @@
+use bgpkit_parser::models::EntryType;
+use bgpkit_parser::mrt_record::parse_common_header;
+use bgpkit_parser::BgpkitParser;
+use bzip2::bufread::BzDecoder;
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use flate2::bufread::GzDecoder;
+use std::fs::File;
+use std::io::{BufReader, Read, Seek, SeekFrom};
+
+mod data_source;
+
+const RECORDS_PER_TYPE: usize = 100;
+
+/// Choose a mix of records with a given MRT type and subtype. The records are chosen to get a
+/// uniform distribution of different length records.
+fn select_mrt_records<R: Read>(mut input_reader: R, mrt_type: EntryType, subtype: u16) -> Vec<u8> {
+    let mut included = Vec::new();
+    let mut buffer = Vec::with_capacity(4096);
+
+    while let Ok(header) = parse_common_header(&mut input_reader) {
+        buffer.clear();
+        header
+            .write_header(&mut buffer)
+            .expect("able to write header to vec");
+        (&mut input_reader)
+            .take(header.length as u64)
+            .read_to_end(&mut buffer)
+            .expect("able to read message body");
+
+        if header.entry_type == mrt_type && header.entry_subtype == subtype {
+            included.push(std::mem::replace(&mut buffer, Vec::with_capacity(4096)));
+        }
+    }
+
+    included.sort_by_key(Vec::len);
+
+    if included.is_empty() {
+        println!("No records found for MRT {:?} {:?}", mrt_type, subtype);
+        return Vec::new();
+    }
+
+    let mut record_output = Vec::new();
+    for n in 0..RECORDS_PER_TYPE {
+        let index = (n * included.len()) / RECORDS_PER_TYPE;
+        record_output.extend_from_slice(&included[index][..]);
+    }
+
+    record_output
+}
+
+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 mut rib_reader = BufReader::new(File::open(rib_data).unwrap());
+
+    println!("Decompressing input data and loading into memory...");
+    let bgp4mp_updates = select_mrt_records(GzDecoder::new(updates_reader), EntryType::BGP4MP, 4);
+    let rib_ipv4_unicast =
+        select_mrt_records(BzDecoder::new(&mut rib_reader), EntryType::TABLE_DUMP_V2, 2);
+    rib_reader.seek(SeekFrom::Start(0)).unwrap();
+    let rib_ipv6_unicast =
+        select_mrt_records(BzDecoder::new(&mut rib_reader), EntryType::TABLE_DUMP_V2, 4);
+
+    c.bench_function("BGP4MP Update", |b| {
+        b.iter_with_large_drop(|| {
+            let mut reader = black_box(&bgp4mp_updates[..]);
+            let mut holder: [Option<_>; RECORDS_PER_TYPE] = std::array::from_fn(|_| None);
+
+            BgpkitParser::from_reader(&mut reader)
+                .into_record_iter()
+                .enumerate()
+                .for_each(|(index, x)| holder[index] = Some(x));
+
+            holder
+        })
+    });
+
+    c.bench_function("TABLE_DUMP_V2 IPv4 Unicast", |b| {
+        b.iter_with_large_drop(|| {
+            let mut reader = black_box(&rib_ipv4_unicast[..]);
+            let mut holder: [Option<_>; RECORDS_PER_TYPE] = std::array::from_fn(|_| None);
+
+            BgpkitParser::from_reader(&mut reader)
+                .into_record_iter()
+                .enumerate()
+                .for_each(|(index, x)| holder[index] = Some(x));
+
+            holder
+        })
+    });
+
+    c.bench_function("TABLE_DUMP_V2 IPv6 Unicast", |b| {
+        b.iter_with_large_drop(|| {
+            let mut reader = black_box(&rib_ipv6_unicast[..]);
+            let mut holder: [Option<_>; RECORDS_PER_TYPE] = std::array::from_fn(|_| None);
+
+            BgpkitParser::from_reader(&mut reader)
+                .into_record_iter()
+                .enumerate()
+                .for_each(|(index, x)| holder[index] = Some(x));
+
+            holder
+        })
+    });
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default();
+    targets = criterion_benchmark
+}
+criterion_main!(benches);

examples/cache_reading.rs

@@ -15,7 +15,10 @@ fn main() {
         let parser =
             BgpkitParser::new_cached(item.url.as_str(), "/tmp/bgpkit-cache-example/").unwrap();
         // iterating through the parser. the iterator returns `BgpElem` one at a time.
-        let elems = parser.into_elem_iter().collect::<Vec<BgpElem>>();
+        let elems = parser
+            .into_elem_iter()
+            .filter_map(Result::ok)
+            .collect::<Vec<BgpElem>>();
         log::info!("{} {} {}", item.collector_id, item.url, elems.len());
     }
 }

examples/deprecated_attributes.rs

@@ -9,6 +9,7 @@ fn main() {
     )
     .unwrap()
     {
+        let elem = elem.unwrap();
         if elem.deprecated.is_some() {
             println!(
                 "{}",

examples/display_elems.rs

@@ -4,6 +4,7 @@ fn main() {
     let url = "http://archive.routeviews.org/bgpdata/\
         2021.10/UPDATES/updates.20211001.0000.bz2";
     for elem in BgpkitParser::new(url).unwrap() {
+        let elem = elem.unwrap();
         println!(
             "{:?}|{:?}|{:?}|{:?}|{:?}",
             elem.elem_type, elem.timestamp, elem.prefix, elem.as_path, elem.next_hop,

examples/extended_communities.rs

@@ -16,6 +16,7 @@ fn main() {
     log::info!("parsing updates file");
     // iterating through the parser. the iterator returns `BgpElem` one at a time.
     for elem in parser {
+        let elem = elem.unwrap();
         if let Some(cs) = &elem.communities {
             for c in cs {
                 match c {

examples/filters.rs

@@ -1,3 +1,4 @@
+use bgpkit_parser::filter::Filter;
 use bgpkit_parser::BgpkitParser;
 
 /// This example shows how to parse a MRT file and filter by prefix.
@@ -27,13 +28,12 @@ fn main() {
         "http://archive.routeviews.org/bgpdata/2021.10/UPDATES/updates.20211001.0000.bz2",
     )
     .unwrap()
-    .add_filter("prefix", "211.98.251.0/24")
-    .unwrap();
+    .add_filter(Filter::prefix("211.98.251.0/24").unwrap());
 
     log::info!("parsing updates file");
     // iterating through the parser. the iterator returns `BgpElem` one at a time.
     for elem in parser {
-        log::info!("{}", &elem);
+        log::info!("{}", elem.unwrap());
     }
     log::info!("done");
 }

examples/find_as_set_messages.rs

@@ -19,6 +19,7 @@ fn main() {
         let collector = item.collector_id.clone();
         let mut origins: HashSet<Asn> = HashSet::new();
         for elem in parser {
+            let elem = elem.unwrap();
             if !elem.elem_type.is_announce() {
                 continue;
             }

examples/only-to-customer.rs

@@ -6,6 +6,7 @@ fn main() {
     )
     .unwrap()
     {
+        let elem = elem.unwrap();
         if let Some(otc) = elem.only_to_customer {
             println!(
                 "OTC found: {} for path {}\n{}\n",

examples/parse-files-from-broker.rs

@@ -19,6 +19,7 @@ fn main() {
         // iterating through the parser. the iterator returns `BgpElem` one at a time.
         let elems = parser
             .into_elem_iter()
+            .map(Result::unwrap)
             .filter_map(|elem| {
                 if let Some(origins) = &elem.origin_asns {
                     if origins.contains(&13335.into()) {

examples/parse-single-file.rs

@@ -15,7 +15,7 @@ fn main() {
     log::info!("parsing updates file");
     // iterating through the parser. the iterator returns `BgpElem` one at a time.
     for elem in parser {
-        log::info!("{}", &elem);
+        log::info!("{}", elem.unwrap());
     }
     log::info!("done");
 }

examples/peer_index_table.rs

@@ -8,6 +8,6 @@ fn main() {
     let url = "https://data.ris.ripe.net/rrc03/2021.11/bview.20211128.1600.gz";
     let parser = bgpkit_parser::BgpkitParser::new(url).unwrap();
     for record in parser.into_record_iter().take(1) {
-        println!("{}", to_string_pretty(&json!(record)).unwrap());
+        println!("{}", to_string_pretty(&json!(record.unwrap())).unwrap());
     }
 }

examples/real-time-routeviews-kafka-openbmp.rs

@@ -3,7 +3,6 @@ extern crate core;
 use bgpkit_parser::parser::bmp::messages::MessageBody;
 use bgpkit_parser::Elementor;
 pub use bgpkit_parser::{parse_bmp_msg, parse_openbmp_header};
-use bytes::Bytes;
 use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage};
 use kafka::error::Error as KafkaError;
 use log::{error, info};
@@ -30,7 +29,7 @@ fn consume_and_print(group: String, topic: String, brokers: Vec<String>) -> Resu
 
         for ms in mss.iter() {
             for m in ms.messages() {
-                let mut bytes = Bytes::from(m.value.to_vec());
+                let mut bytes = m.value;
                 let header = parse_openbmp_header(&mut bytes).unwrap();
                 let bmp_msg = parse_bmp_msg(&mut bytes);
                 match bmp_msg {

examples/records_iter.rs

@@ -6,7 +6,7 @@ fn main() {
     let url = "http://archive.routeviews.org/route-views.amsix/bgpdata/2023.02/UPDATES/updates.20230222.0430.bz2";
     let parser = BgpkitParser::new(url).unwrap();
     for record in parser.into_record_iter() {
-        match record.message {
+        match record.unwrap().message {
             MrtMessage::TableDumpMessage(_) => {}
             MrtMessage::TableDumpV2Message(_) => {}
             MrtMessage::Bgp4Mp(msg) => match msg {