X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fbgp_client.rs;h=3e5419002bbf83e8e47370f13d0bb210007a4867;hb=479a29348b70bdfbf07015e88b13f3b98bd1e3b4;hp=9038005a3d412aadc075bf26422c38ee32ed494b;hpb=8e27de4c043eb7967697c8c38b2e0b65bf546b8c;p=dnsseed-rust diff --git a/src/bgp_client.rs b/src/bgp_client.rs index 9038005..3e54190 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -1,7 +1,7 @@ use std::sync::{Arc, Mutex}; use std::sync::atomic::{AtomicBool, Ordering}; use std::cmp; -use std::collections::HashMap; +use std::collections::{HashMap, hash_map}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::time::{Duration, Instant}; @@ -24,12 +24,14 @@ use crate::timeout_stream::TimeoutStream; const PATH_SUFFIX_LEN: usize = 3; #[derive(Clone)] -struct Route { // 32 bytes +struct Route { // 32 bytes with a path id u32 path_suffix: [u32; PATH_SUFFIX_LEN], path_len: u32, pref: u32, med: u32, } +#[allow(dead_code)] +const ROUTE_LEN: usize = 36 - std::mem::size_of::<(u32, Route)>(); // To keep memory tight (and since we dont' need such close alignment), newtype the v4/v6 routing // table entries to make sure they are aligned to single bytes. @@ -48,6 +50,10 @@ impl From<(Ipv4Addr, u8)> for V4Addr { } } } +#[allow(dead_code)] +const V4_ALIGN: usize = 1 - std::mem::align_of::(); +#[allow(dead_code)] +const V4_SIZE: usize = 5 - std::mem::size_of::(); #[repr(packed)] #[derive(PartialEq, Eq, Hash)] @@ -63,17 +69,27 @@ impl From<(Ipv6Addr, u8)> for V6Addr { } } } +#[allow(dead_code)] +const V6_ALIGN: usize = 1 - std::mem::align_of::(); +#[allow(dead_code)] +const V6_SIZE: usize = 17 - std::mem::size_of::(); struct RoutingTable { - v4_table: HashMap>, - v6_table: HashMap>, + // We really want a HashMap for the values here, but they'll only ever contain a few entries, + // and Vecs are way more memory-effecient in that case. + v4_table: HashMap>, + v6_table: HashMap>, + max_paths: usize, + routes_with_max: usize, } impl RoutingTable { fn new() -> Self { Self { - v4_table: HashMap::new(), - v6_table: HashMap::new(), + v4_table: HashMap::with_capacity(900_000), + v6_table: HashMap::with_capacity(100_000), + max_paths: 0, + routes_with_max: 0, } } @@ -83,9 +99,9 @@ impl RoutingTable { //TODO: Optimize this (probably means making the tables btrees)! let mut lookup = <$addrty>::from(($addr, $addr_bits)); for i in 0..$addr_bits { - if let Some(routes) = $table.get(&lookup).map(|hm| hm.values()) { + if let Some(routes) = $table.get(&lookup) { if routes.len() > 0 { - return (lookup.pfxlen, routes.collect()); + return (lookup.pfxlen, routes.iter().map(|v| &v.1).collect()); } } lookup.addr[lookup.addr.len() - (i/8) - 1] &= !(1u8 << (i % 8)); @@ -101,42 +117,87 @@ impl RoutingTable { } fn withdraw(&mut self, route: NLRIEncoding) { + macro_rules! remove { + ($rt: expr, $v: expr, $id: expr) => { { + match $rt.entry($v.into()) { + hash_map::Entry::Occupied(mut entry) => { + if entry.get().len() == self.max_paths { + self.routes_with_max -= 1; + if self.routes_with_max == 0 { + self.max_paths = 0; + } + } + entry.get_mut().retain(|e| e.0 != $id); + if entry.get_mut().is_empty() { + entry.remove(); + } + }, + _ => {}, + } + } } + } match route { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.get_mut(&(v4a, len).into()).and_then(|hm| hm.remove(&0)), - IpAddr::V6(v6a) => self.v6_table.get_mut(&(v6a, len).into()).and_then(|hm| hm.remove(&0)), + IpAddr::V4(v4a) => remove!(self.v4_table, (v4a, len), 0), + IpAddr::V6(v6a) => remove!(self.v6_table, (v6a, len), 0), } }, NLRIEncoding::IP_WITH_PATH_ID((p, id)) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.get_mut(&(v4a, len).into()).and_then(|hm| hm.remove(&id)), - IpAddr::V6(v6a) => self.v6_table.get_mut(&(v6a, len).into()).and_then(|hm| hm.remove(&id)), + IpAddr::V4(v4a) => remove!(self.v4_table, (v4a, len), id), + IpAddr::V6(v6a) => remove!(self.v6_table, (v6a, len), id), } }, - NLRIEncoding::IP_MPLS(_) => None, + NLRIEncoding::IP_MPLS(_) => (), + NLRIEncoding::IP_MPLS_WITH_PATH_ID(_) => (), + NLRIEncoding::IP_VPN_MPLS(_) => (), + NLRIEncoding::L2VPN(_) => (), }; } fn announce(&mut self, prefix: NLRIEncoding, route: Route) { + macro_rules! insert { + ($rt: expr, $v: expr, $id: expr) => { { + let old_max_paths = self.max_paths; + let entry = $rt.entry($v.into()).or_insert_with(|| Vec::with_capacity(old_max_paths)); + let entry_had_max = entry.len() == self.max_paths; + entry.retain(|e| e.0 != $id); + if entry_had_max { + entry.reserve_exact(1); + } else { + entry.reserve_exact(cmp::max(self.max_paths, entry.len() + 1) - entry.len()); + } + entry.push(($id, route)); + if entry.len() > self.max_paths { + self.max_paths = entry.len(); + self.routes_with_max = 1; + } else if entry.len() == self.max_paths { + if !entry_had_max { self.routes_with_max += 1; } + } + } } + } match prefix { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.entry((v4a, len).into()).or_insert(HashMap::new()).insert(0, route), - IpAddr::V6(v6a) => self.v6_table.entry((v6a, len).into()).or_insert(HashMap::new()).insert(0, route), + IpAddr::V4(v4a) => insert!(self.v4_table, (v4a, len), 0), + IpAddr::V6(v6a) => insert!(self.v6_table, (v6a, len), 0), } }, NLRIEncoding::IP_WITH_PATH_ID((p, id)) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.entry((v4a, len).into()).or_insert(HashMap::new()).insert(id, route), - IpAddr::V6(v6a) => self.v6_table.entry((v6a, len).into()).or_insert(HashMap::new()).insert(id, route), + IpAddr::V4(v4a) => insert!(self.v4_table, (v4a, len), id), + IpAddr::V6(v6a) => insert!(self.v6_table, (v6a, len), id), } }, - NLRIEncoding::IP_MPLS(_) => None, + NLRIEncoding::IP_MPLS(_) => (), + NLRIEncoding::IP_MPLS_WITH_PATH_ID(_) => (), + NLRIEncoding::IP_VPN_MPLS(_) => (), + NLRIEncoding::L2VPN(_) => (), }; } } @@ -164,8 +225,8 @@ impl<'a> std::io::Read for BytesDecoder<'a> { } } -struct MsgCoder<'a>(&'a Printer); -impl<'a> codec::Decoder for MsgCoder<'a> { +struct MsgCoder(Option); +impl codec::Decoder for MsgCoder { type Item = Message; type Error = std::io::Error; @@ -174,15 +235,17 @@ impl<'a> codec::Decoder for MsgCoder<'a> { buf: bytes, pos: 0 }; - match (Reader { + let def_cap = Default::default(); + let mut reader = Reader { stream: &mut decoder, - capabilities: Capabilities { - FOUR_OCTET_ASN_SUPPORT: true, - EXTENDED_PATH_NLRI_SUPPORT: true, - } - }).read() { + capabilities: if let Some(cap) = &self.0 { cap } else { &def_cap }, + }; + match reader.read() { Ok((_header, msg)) => { decoder.buf.advance(decoder.pos); + if let Message::Open(ref o) = &msg { + self.0 = Some(Capabilities::from_parameters(o.parameters.clone())); + } Ok(Some(msg)) }, Err(e) => match e.kind() { @@ -192,12 +255,12 @@ impl<'a> codec::Decoder for MsgCoder<'a> { } } } -impl<'a> codec::Encoder for MsgCoder<'a> { +impl codec::Encoder for MsgCoder { type Item = Message; type Error = std::io::Error; fn encode(&mut self, msg: Message, res: &mut bytes::BytesMut) -> Result<(), std::io::Error> { - msg.write(&mut BytesCoder(res))?; + msg.encode(&mut BytesCoder(res))?; Ok(()) } } @@ -219,14 +282,35 @@ impl BGPClient { }); let primary_route = path_vecs.pop().unwrap(); - 'asn_candidates: for asn in primary_route.path_suffix.iter().rev() { - if *asn == 0 { continue 'asn_candidates; } - for secondary_route in path_vecs.iter() { - if !secondary_route.path_suffix.contains(asn) { - continue 'asn_candidates; + if path_vecs.len() > 3 { + // If we have at least 3 paths, try to find the last unique ASN which doesn't show up in other paths + // If we hit a T1 that is reasonably assumed to care about net neutrality, return the + // previous ASN. + let mut prev_asn = 0; + 'asn_candidates: for asn in primary_route.path_suffix.iter().rev() { + if *asn == 0 { continue 'asn_candidates; } + match *asn { + // Included: CenturyLink (L3), Cogent, Telia, NTT, GTT, Level3, + // GBLX (L3), Zayo, TI Sparkle Seabone, HE, Telefonica + // Left out from Caida top-20: TATA, PCCW, Vodafone, RETN, Orange, Telstra, + // Singtel, Rostelecom, DTAG + 209|174|1299|2914|3257|3356|3549|6461|6762|6939|12956 if prev_asn != 0 => return prev_asn, + _ => if path_vecs.iter().any(|route| !route.path_suffix.contains(asn)) { + if prev_asn != 0 { return prev_asn } else { + // Multi-origin prefix, just give up and take the last AS in the + // default path + break 'asn_candidates; + } + } else { + // We only ever possibly return an ASN if it appears in all paths + prev_asn = *asn; + }, } } - return *asn; + // All paths were the same, if the first ASN is non-0, return it. + if prev_asn != 0 { + return prev_asn; + } } for asn in primary_route.path_suffix.iter().rev() { @@ -308,7 +392,7 @@ impl BGPClient { future::err(()) }) }).and_then(move |stream| { - let (write, read) = Framed::new(stream.0, MsgCoder(printer)).split(); + let (write, read) = Framed::new(stream.0, MsgCoder(None)).split(); let (mut sender, receiver) = mpsc::channel(10); // We never really should send more than 10 messages unless they're dumb tokio::spawn(write.sink_map_err(|_| { () }).send_all(receiver) .then(|_| { @@ -327,7 +411,7 @@ impl BGPClient { OpenCapability::AddPath(vec![ (AFI::IPV4, SAFI::Unicast, AddPathDirection::ReceivePaths), (AFI::IPV6, SAFI::Unicast, AddPathDirection::ReceivePaths)]), - ])] + ])], })); TimeoutStream::new_persistent(read, timeout).for_each(move |bgp_msg| { if client.shutdown.load(Ordering::Relaxed) { @@ -355,6 +439,7 @@ impl BGPClient { } printer.set_stat(Stat::V4RoutingTableSize(route_table.v4_table.len())); printer.set_stat(Stat::V6RoutingTableSize(route_table.v6_table.len())); + printer.set_stat(Stat::RoutingTablePaths(route_table.max_paths)); }, _ => {} }