From 866152d7203c8f11a686d073a2c3f617f9131675 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Tue, 1 Oct 2019 21:19:18 -0400 Subject: [PATCH 01/16] Fix a minor timing issue, load nodes at start with an old time --- src/datastore.rs | 3 ++- src/main.rs | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/datastore.rs b/src/datastore.rs index 2f0e9a8..0e6dea3 100644 --- a/src/datastore.rs +++ b/src/datastore.rs @@ -234,6 +234,7 @@ impl Store { } let nodes_future = File::open(store.clone() + "/nodes").and_then(|f| { + let start_time = Instant::now() - Duration::from_secs(60 * 60 * 24); let mut res = nodes_uninitd!(); let l = BufReader::new(f).lines(); for line_res in l { @@ -272,7 +273,7 @@ impl Store { } } } - res.state_next_scan[node.state.to_num() as usize].push((Instant::now(), sockaddr)); + res.state_next_scan[node.state.to_num() as usize].push((start_time, sockaddr)); res.nodes_to_state.insert(sockaddr, node); } future::ok(res) diff --git a/src/main.rs b/src/main.rs index d25020e..f9de3d4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -283,7 +283,7 @@ fn scan_net() { let mut scan_nodes = store.get_next_scan_nodes(); printer.add_line(format!("Got {} addresses to scan", scan_nodes.len()), false); if !scan_nodes.is_empty() { - let per_iter_time = Duration::from_millis(datastore::SECS_PER_SCAN_RESULTS / scan_nodes.len() as u64); + let per_iter_time = Duration::from_millis(datastore::SECS_PER_SCAN_RESULTS * 1000 / scan_nodes.len() as u64); let mut iter_time = start_time; for node in scan_nodes.drain(..) { -- 2.30.2 From 909fc5eb1b1a59ac06d7189c0cce9ea78630269b Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Tue, 1 Oct 2019 21:34:17 -0400 Subject: [PATCH 02/16] Include debug symbols in release mode, so backtraces still work --- Cargo.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/Cargo.toml b/Cargo.toml index fc9a4d3..075f296 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ num_cpus = "1" panic = "abort" [profile.release] +debug = true lto = true codegen-units = 1 panic = "abort" -- 2.30.2 From 7aaa26debb3bb875c5e49fbd39dec642cd535f88 Mon Sep 17 00:00:00 2001 From: Carl Dong Date: Fri, 23 Aug 2019 15:07:34 -0400 Subject: [PATCH 03/16] We don't need BTreeMaps --- src/bgp_client.rs | 39 +++++++++++++++++++-------------------- 1 file changed, 19 insertions(+), 20 deletions(-) diff --git a/src/bgp_client.rs b/src/bgp_client.rs index 4d393c3..04be136 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -1,8 +1,7 @@ use std::sync::{Arc, Mutex}; use std::sync::atomic::{AtomicBool, Ordering}; use std::cmp; -use std::ops::Bound::Included; -use std::collections::BTreeMap; +use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::time::{Duration, Instant}; @@ -30,23 +29,22 @@ struct Route { } struct RoutingTable { - v4_table: BTreeMap<(Ipv4Addr, u8, u32), Arc>, - v6_table: BTreeMap<(Ipv6Addr, u8, u32), Arc>, + v4_table: HashMap<(Ipv4Addr, u8), HashMap>>, + v6_table: HashMap<(Ipv6Addr, u8), HashMap>>, } impl RoutingTable { fn new() -> Self { Self { - v4_table: BTreeMap::new(), - v6_table: BTreeMap::new(), + v4_table: HashMap::new(), + v6_table: HashMap::new(), } } fn get_route_attrs(&self, ip: IpAddr) -> Vec> { macro_rules! lookup_res { ($addrty: ty, $addr: expr, $table: expr, $addr_bits: expr) => { { - let mut res = Vec::new(); - //TODO: Optimize this! + //TODO: Optimize this (probably means making the tables btrees)! for i in (0..$addr_bits).rev() { let mut lookup = $addr.octets(); for b in 0..(i / 8) { @@ -54,12 +52,13 @@ impl RoutingTable { } lookup[lookup.len() - (i/8) - 1] &= !(((1u16 << (i % 8)) - 1) as u8); let lookup_addr = <$addrty>::from(lookup); - for attrs in $table.range((Included((lookup_addr, $addr_bits - i as u8, 0)), Included((lookup_addr, $addr_bits - i as u8, std::u32::MAX)))) { - res.push(Arc::clone(&attrs.1)); + if let Some(routes) = $table.get(&(lookup_addr, $addr_bits - i as u8)).map(|hm| hm.values()) { + if routes.len() > 0 { + return routes.map(|x| Arc::clone(&x)).collect(); + } } - if !res.is_empty() { break; } } - res + vec![] } } } match ip { @@ -73,15 +72,15 @@ impl RoutingTable { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.remove(&(v4a, len, 0)), - IpAddr::V6(v6a) => self.v6_table.remove(&(v6a, len, 0)), + IpAddr::V4(v4a) => self.v4_table.get_mut(&(v4a, len)).and_then(|hm| hm.remove(&0)), + IpAddr::V6(v6a) => self.v6_table.get_mut(&(v6a, len)).and_then(|hm| hm.remove(&0)), } }, NLRIEncoding::IP_WITH_PATH_ID((p, id)) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.remove(&(v4a, len, id)), - IpAddr::V6(v6a) => self.v6_table.remove(&(v6a, len, id)), + IpAddr::V4(v4a) => self.v4_table.get_mut(&(v4a, len)).and_then(|hm| hm.remove(&id)), + IpAddr::V6(v6a) => self.v6_table.get_mut(&(v6a, len)).and_then(|hm| hm.remove(&id)), } }, NLRIEncoding::IP_MPLS(_) => None, @@ -93,15 +92,15 @@ impl RoutingTable { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.insert((v4a, len, 0), route), - IpAddr::V6(v6a) => self.v6_table.insert((v6a, len, 0), route), + IpAddr::V4(v4a) => self.v4_table.entry((v4a, len)).or_insert(HashMap::new()).insert(0, route), + IpAddr::V6(v6a) => self.v6_table.entry((v6a, len)).or_insert(HashMap::new()).insert(0, route), } }, NLRIEncoding::IP_WITH_PATH_ID((p, id)) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.insert((v4a, len, id), route), - IpAddr::V6(v6a) => self.v6_table.insert((v6a, len, id), route), + IpAddr::V4(v4a) => self.v4_table.entry((v4a, len)).or_insert(HashMap::new()).insert(id, route), + IpAddr::V6(v6a) => self.v6_table.entry((v6a, len)).or_insert(HashMap::new()).insert(id, route), } }, NLRIEncoding::IP_MPLS(_) => None, -- 2.30.2 From c579e91f083d369055230497b1c7bf4191a50880 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Sun, 1 Dec 2019 17:31:33 -0500 Subject: [PATCH 04/16] Drop v6 tclass and scope to cut down on mem usage a bit --- src/datastore.rs | 81 +++++++++++++++++++++++++++++++++++++----------- 1 file changed, 63 insertions(+), 18 deletions(-) diff --git a/src/datastore.rs b/src/datastore.rs index 0e6dea3..354a37d 100644 --- a/src/datastore.rs +++ b/src/datastore.rs @@ -1,7 +1,7 @@ use std::{cmp, mem}; use std::collections::{HashSet, HashMap, hash_map}; use std::sync::{Arc, RwLock}; -use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}; use std::time::{Duration, Instant}; use std::io::{BufRead, BufReader}; @@ -126,17 +126,60 @@ struct Node { state: AddressState, } +/// Essentially SocketAddr but without a traffic class or scope +#[derive(Clone, PartialEq, Eq, Hash)] +enum SockAddr { + V4(SocketAddrV4), + V6((Ipv6Addr, u16)), +} +impl From for SockAddr { + fn from(addr: SocketAddr) -> SockAddr { + match addr { + SocketAddr::V4(sa) => SockAddr::V4(sa), + SocketAddr::V6(sa) => SockAddr::V6((sa.ip().clone(), sa.port())), + } + } +} +impl Into for &SockAddr { + fn into(self) -> SocketAddr { + match self { + &SockAddr::V4(sa) => SocketAddr::V4(sa), + &SockAddr::V6(sa) => SocketAddr::V6(SocketAddrV6::new(sa.0, sa.1, 0, 0)) + } + } +} +impl ToString for SockAddr { + fn to_string(&self) -> String { + let sa: SocketAddr = self.into(); + sa.to_string() + } +} +impl SockAddr { + pub fn port(&self) -> u16 { + match *self { + SockAddr::V4(sa) => sa.port(), + SockAddr::V6((_, port)) => port, + } + } + pub fn ip(&self) -> IpAddr { + match *self { + SockAddr::V4(sa) => IpAddr::V4(sa.ip().clone()), + SockAddr::V6((ip, _)) => IpAddr::V6(ip), + } + } +} + struct Nodes { - good_node_services: Vec>, - nodes_to_state: HashMap, - state_next_scan: Vec>, + good_node_services: Vec>, + nodes_to_state: HashMap, + state_next_scan: Vec>, } struct NodesMutRef<'a> { - good_node_services: &'a mut Vec>, - nodes_to_state: &'a mut HashMap, - state_next_scan: &'a mut Vec>, - + good_node_services: &'a mut Vec>, + nodes_to_state: &'a mut HashMap, + state_next_scan: &'a mut Vec>, } + impl Nodes { fn borrow_mut<'a>(&'a mut self) -> NodesMutRef<'a> { NodesMutRef { @@ -269,12 +312,12 @@ impl Store { if node.state == AddressState::Good { for i in 0..64 { if node.last_services & (1 << i) != 0 { - res.good_node_services[i].insert(sockaddr); + res.good_node_services[i].insert(sockaddr.into()); } } } - res.state_next_scan[node.state.to_num() as usize].push((start_time, sockaddr)); - res.nodes_to_state.insert(sockaddr, node); + res.state_next_scan[node.state.to_num() as usize].push((start_time, sockaddr.into())); + res.nodes_to_state.insert(sockaddr.into(), node); } future::ok(res) }).or_else(|_| -> future::FutureResult { @@ -315,7 +358,7 @@ impl Store { let mut nodes = self.nodes.write().unwrap(); let cur_time = Instant::now(); for addr in addresses { - match nodes.nodes_to_state.entry(addr.clone()) { + match nodes.nodes_to_state.entry(addr.into()) { hash_map::Entry::Vacant(e) => { e.insert(Node { state: AddressState::Untested, @@ -323,7 +366,7 @@ impl Store { last_update: cur_time, last_good: cur_time, }); - nodes.state_next_scan[AddressState::Untested.to_num() as usize].push((cur_time, addr)); + nodes.state_next_scan[AddressState::Untested.to_num() as usize].push((cur_time, addr.into())); res += 1; }, hash_map::Entry::Occupied(_) => {}, @@ -341,12 +384,14 @@ impl Store { })); } - pub fn set_node_state(&self, addr: SocketAddr, state: AddressState, services: u64) -> AddressState { + pub fn set_node_state(&self, sockaddr: SocketAddr, state: AddressState, services: u64) -> AddressState { + let addr: SockAddr = sockaddr.into(); + let now = Instant::now(); + let mut nodes_lock = self.nodes.write().unwrap(); let nodes = nodes_lock.borrow_mut(); - let now = Instant::now(); - let state_ref = nodes.nodes_to_state.entry(addr).or_insert(Node { + let state_ref = nodes.nodes_to_state.entry(addr.clone()).or_insert(Node { state: AddressState::Untested, last_services: 0, last_update: now, @@ -369,7 +414,7 @@ impl Store { if state == AddressState::Good { for i in 0..64 { if services & (1 << i) != 0 && state_ref.last_services & (1 << i) == 0 { - nodes.good_node_services[i].insert(addr); + nodes.good_node_services[i].insert(addr.clone()); } else if services & (1 << i) == 0 && state_ref.last_services & (1 << i) != 0 { nodes.good_node_services[i].remove(&addr); } @@ -541,7 +586,7 @@ impl Store { let mut new_nodes = state_nodes.split_off(split_point as usize); mem::swap(&mut new_nodes, state_nodes); for (_, node) in new_nodes.drain(..) { - res.push(node); + res.push((&node).into()); } } } -- 2.30.2 From 0c6846d425dea8331a586420c74f367cef879687 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Sun, 1 Dec 2019 18:08:59 -0500 Subject: [PATCH 05/16] Drop prepends in BGP paths since it doesn't change our results --- src/bgp_client.rs | 35 +++++++++++++++++++++++++++-------- 1 file changed, 27 insertions(+), 8 deletions(-) diff --git a/src/bgp_client.rs b/src/bgp_client.rs index 04be136..10880a6 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -22,8 +22,10 @@ use futures::sync::mpsc; use crate::printer::{Printer, Stat}; use crate::timeout_stream::TimeoutStream; +const PATH_SUFFIX_LEN: usize = 3; struct Route { - path: Vec, + path_suffix: [u32; PATH_SUFFIX_LEN], + path_len: u32, pref: u32, med: u32, } @@ -180,20 +182,27 @@ impl BGPClient { path_vecs.sort_unstable_by(|path_a, path_b| { path_a.pref.cmp(&path_b.pref) - .then(path_b.path.len().cmp(&path_a.path.len())) + .then(path_b.path_len.cmp(&path_a.path_len)) .then(path_b.med.cmp(&path_a.med)) }); let primary_route = path_vecs.pop().unwrap(); - 'asn_candidates: for asn in primary_route.path.iter().rev() { + '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.contains(asn) { + if !secondary_route.path_suffix.contains(asn) { continue 'asn_candidates; } } return *asn; } - *primary_route.path.last().unwrap_or(&0) + + for asn in primary_route.path_suffix.iter().rev() { + if *asn != 0 { + return *asn; + } + } + 0 } pub fn disconnect(&self) { @@ -215,15 +224,25 @@ impl BGPClient { } } if let Some(mut aspath) = as4_path.or(as_path) { - let mut path = Vec::new(); + let mut pathvec = Vec::new(); for seg in aspath.segments.drain(..) { match seg { - Segment::AS_SEQUENCE(mut asn) => path.append(&mut asn), + Segment::AS_SEQUENCE(mut asn) => pathvec.append(&mut asn), Segment::AS_SET(_) => {}, // Ignore sets for now, they're not that common anyway } } + let path_len = pathvec.len() as u32; + pathvec.dedup_by(|a, b| (*a).eq(b)); // Drop prepends, cause we don't care in this case + + let mut path_suffix = [0; PATH_SUFFIX_LEN]; + for (idx, asn) in pathvec.iter().rev().enumerate() { + path_suffix[PATH_SUFFIX_LEN - idx - 1] = *asn; + if idx == PATH_SUFFIX_LEN - 1 { break; } + } + return Some(Route { - path: path.clone(), + path_suffix, + path_len, pref, med, }) -- 2.30.2 From 9a19c37ba4266971a4fd9503ac4fded63061a41b Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Sun, 1 Dec 2019 23:40:22 -0500 Subject: [PATCH 06/16] Drop Arcs around paths, and add more debug info for looked-up ASNs --- src/bgp_client.rs | 36 ++++++++++++++++++++++++++---------- src/reader.rs | 2 +- 2 files changed, 27 insertions(+), 11 deletions(-) diff --git a/src/bgp_client.rs b/src/bgp_client.rs index 10880a6..e38cffb 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -23,7 +23,8 @@ use crate::printer::{Printer, Stat}; use crate::timeout_stream::TimeoutStream; const PATH_SUFFIX_LEN: usize = 3; -struct Route { +#[derive(Clone)] +struct Route { // 32 bytes path_suffix: [u32; PATH_SUFFIX_LEN], path_len: u32, pref: u32, @@ -31,8 +32,8 @@ struct Route { } struct RoutingTable { - v4_table: HashMap<(Ipv4Addr, u8), HashMap>>, - v6_table: HashMap<(Ipv6Addr, u8), HashMap>>, + v4_table: HashMap<(Ipv4Addr, u8), HashMap>, + v6_table: HashMap<(Ipv6Addr, u8), HashMap>, } impl RoutingTable { @@ -43,7 +44,7 @@ impl RoutingTable { } } - fn get_route_attrs(&self, ip: IpAddr) -> Vec> { + fn get_route_attrs(&self, ip: IpAddr) -> (u8, Vec<&Route>) { macro_rules! lookup_res { ($addrty: ty, $addr: expr, $table: expr, $addr_bits: expr) => { { //TODO: Optimize this (probably means making the tables btrees)! @@ -56,11 +57,11 @@ impl RoutingTable { let lookup_addr = <$addrty>::from(lookup); if let Some(routes) = $table.get(&(lookup_addr, $addr_bits - i as u8)).map(|hm| hm.values()) { if routes.len() > 0 { - return routes.map(|x| Arc::clone(&x)).collect(); + return ($addr_bits - i as u8, routes.collect()); } } } - vec![] + (0, vec![]) } } } match ip { @@ -89,7 +90,7 @@ impl RoutingTable { }; } - fn announce(&mut self, prefix: NLRIEncoding, route: Arc) { + fn announce(&mut self, prefix: NLRIEncoding, route: Route) { match prefix { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); @@ -177,7 +178,8 @@ pub struct BGPClient { } impl BGPClient { pub fn get_asn(&self, addr: IpAddr) -> u32 { - let mut path_vecs = self.routes.lock().unwrap().get_route_attrs(addr).clone(); + let lock = self.routes.lock().unwrap(); + let mut path_vecs = lock.get_route_attrs(addr).1; if path_vecs.is_empty() { return 0; } path_vecs.sort_unstable_by(|path_a, path_b| { @@ -205,6 +207,21 @@ impl BGPClient { 0 } + pub fn get_path(&self, addr: IpAddr) -> (u8, [u32; PATH_SUFFIX_LEN]) { + let lock = self.routes.lock().unwrap(); + let (prefixlen, mut path_vecs) = lock.get_route_attrs(addr); + if path_vecs.is_empty() { return (0, [0; PATH_SUFFIX_LEN]); } + + path_vecs.sort_unstable_by(|path_a, path_b| { + path_a.pref.cmp(&path_b.pref) + .then(path_b.path_len.cmp(&path_a.path_len)) + .then(path_b.med.cmp(&path_a.med)) + }); + + let primary_route = path_vecs.pop().unwrap(); + (prefixlen, primary_route.path_suffix) + } + pub fn disconnect(&self) { self.shutdown.store(true, Ordering::Relaxed); } @@ -302,9 +319,8 @@ impl BGPClient { route_table.withdraw(r); } if let Some(path) = Self::map_attrs(upd.attributes) { - let path_arc = Arc::new(path); for r in upd.announced_routes { - route_table.announce(r, Arc::clone(&path_arc)); + route_table.announce(r, path.clone()); } } printer.set_stat(Stat::V4RoutingTableSize(route_table.v4_table.len())); diff --git a/src/reader.rs b/src/reader.rs index 967d717..fc4df12 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -122,7 +122,7 @@ pub fn read(store: &'static Store, printer: &'static Printer, bgp_client: Arc { let ip = try_parse_next_chunk!(IpAddr); - printer.add_line(format!("ASN for {} is {}", ip, bgp_client.get_asn(ip)), false); + printer.add_line(format!("ASN for {} is {} ({:?})", ip, bgp_client.get_asn(ip), bgp_client.get_path(ip)), false); }, "r" => { match AddressState::from_num(try_parse_next_chunk!(u8)) { -- 2.30.2 From ec39152171e7e0a2b5c69dc13b12d417f360ad5b Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 2 Dec 2019 13:17:11 -0500 Subject: [PATCH 07/16] Fix BGP table lookup --- src/bgp_client.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/bgp_client.rs b/src/bgp_client.rs index e38cffb..c20323d 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -48,18 +48,15 @@ impl RoutingTable { macro_rules! lookup_res { ($addrty: ty, $addr: expr, $table: expr, $addr_bits: expr) => { { //TODO: Optimize this (probably means making the tables btrees)! - for i in (0..$addr_bits).rev() { - let mut lookup = $addr.octets(); - for b in 0..(i / 8) { - lookup[lookup.len() - b - 1] = 0; - } - lookup[lookup.len() - (i/8) - 1] &= !(((1u16 << (i % 8)) - 1) as u8); + let mut lookup = $addr.octets(); + for i in 0..$addr_bits { let lookup_addr = <$addrty>::from(lookup); if let Some(routes) = $table.get(&(lookup_addr, $addr_bits - i as u8)).map(|hm| hm.values()) { if routes.len() > 0 { return ($addr_bits - i as u8, routes.collect()); } } + lookup[lookup.len() - (i/8) - 1] &= !(1u8 << (i % 8)); } (0, vec![]) } } -- 2.30.2 From 85cbeb34b69c8f3dc283b3963c98fd509c067fbf Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 2 Dec 2019 14:11:17 -0500 Subject: [PATCH 08/16] Clean up print format --- src/reader.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/reader.rs b/src/reader.rs index fc4df12..32bb812 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -122,7 +122,7 @@ pub fn read(store: &'static Store, printer: &'static Printer, bgp_client: Arc { let ip = try_parse_next_chunk!(IpAddr); - printer.add_line(format!("ASN for {} is {} ({:?})", ip, bgp_client.get_asn(ip), bgp_client.get_path(ip)), false); + printer.add_line(format!("ASN for {} is {} (prefixlen, path: {:?})", ip, bgp_client.get_asn(ip), bgp_client.get_path(ip)), false); }, "r" => { match AddressState::from_num(try_parse_next_chunk!(u8)) { -- 2.30.2 From 0e8e7d68a60e43ceaf3aa7b35bdb2317b51f3b07 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 2 Dec 2019 14:37:00 -0500 Subject: [PATCH 09/16] Drop a needless Vec --- src/datastore.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/datastore.rs b/src/datastore.rs index 354a37d..840863a 100644 --- a/src/datastore.rs +++ b/src/datastore.rs @@ -170,12 +170,12 @@ impl SockAddr { } struct Nodes { - good_node_services: Vec>, + good_node_services: [HashSet; 64], nodes_to_state: HashMap, state_next_scan: Vec>, } struct NodesMutRef<'a> { - good_node_services: &'a mut Vec>, + good_node_services: &'a mut [HashSet; 64], nodes_to_state: &'a mut HashMap, state_next_scan: &'a mut Vec>, } @@ -264,10 +264,7 @@ impl Store { for _ in 0..AddressState::get_count() { state_vecs.push(Vec::new()); } - let mut good_node_services = Vec::with_capacity(64); - for _ in 0..64 { - good_node_services.push(HashSet::new()); - } + let good_node_services = [HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new(), HashSet::new()]; Nodes { good_node_services, nodes_to_state: HashMap::new(), -- 2.30.2 From 2590cec11b1c75ecd959267f07596219b40c2daa Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 2 Dec 2019 14:55:06 -0500 Subject: [PATCH 10/16] Add some basic COMPACT_FILTERS support --- src/datastore.rs | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/src/datastore.rs b/src/datastore.rs index 840863a..829fe0e 100644 --- a/src/datastore.rs +++ b/src/datastore.rs @@ -486,7 +486,28 @@ impl Store { let mut dns_buff = String::new(); { let mut rng = thread_rng(); - for i in &[1u64, 4, 5, 8, 9, 12, 13, 1024, 1025, 1028, 1029, 1032, 1033, 1036, 1037] { + for i in &[ 0b00000000001u64, + 0b00000000100, + 0b00000000101, + 0b00000001000, + 0b00000001001, + 0b00000001100, + 0b00000001101, + 0b00001001001, + 0b10000000000, + 0b10000000001, + 0b10000000100, + 0b10000000101, + 0b10000001000, + 0b10000001001, + 0b10000001100, + 0b10000001101, + 0b10001001000] { + // ^ NODE_NETWORK_LIIMTED + //COMPACT_FILTERS ^ ^ NODE_BLOOM + // NODE_WITNESS ^ ^ NODE_NETWORK + // We support all combos of NETWORK, NETWORK_LIMITED, BLOOM, and WITNESS + // We support COMPACT_FILTERS with WITNESS and NETWORK or NETWORK_LIIMTED. let mut tor_set: Vec = Vec::new(); let mut v6_set: Vec = Vec::new(); let mut v4_set: Vec = Vec::new(); -- 2.30.2 From 515fa38dfacb27ca8ca53e51f2bcc6270ceca617 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 2 Dec 2019 15:09:36 -0500 Subject: [PATCH 11/16] Make sure we dont ever queue things twice --- src/datastore.rs | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/src/datastore.rs b/src/datastore.rs index 829fe0e..5f9874f 100644 --- a/src/datastore.rs +++ b/src/datastore.rs @@ -124,6 +124,7 @@ struct Node { last_good: Instant, // Ignored unless state is Good or WasGood last_services: u64, state: AddressState, + queued: bool, } /// Essentially SocketAddr but without a traffic class or scope @@ -305,6 +306,7 @@ impl Store { last_services, last_update: Instant::now(), last_good: Instant::now(), + queued: true, }; if node.state == AddressState::Good { for i in 0..64 { @@ -362,6 +364,7 @@ impl Store { last_services: 0, last_update: cur_time, last_good: cur_time, + queued: true, }); nodes.state_next_scan[AddressState::Untested.to_num() as usize].push((cur_time, addr.into())); res += 1; @@ -393,6 +396,7 @@ impl Store { last_services: 0, last_update: now, last_good: now, + queued: false, }); let ret = state_ref.state; if (state_ref.state == AddressState::Good || state_ref.state == AddressState::WasGood) @@ -405,7 +409,10 @@ impl Store { } } state_ref.last_services = 0; - nodes.state_next_scan[AddressState::WasGood.to_num() as usize].push((now, addr)); + if !state_ref.queued { + nodes.state_next_scan[AddressState::WasGood.to_num() as usize].push((now, addr)); + state_ref.queued = true; + } } else { state_ref.state = state; if state == AddressState::Good { @@ -419,7 +426,10 @@ impl Store { state_ref.last_services = services; state_ref.last_good = now; } - nodes.state_next_scan[state.to_num() as usize].push((now, addr)); + if !state_ref.queued { + nodes.state_next_scan[state.to_num() as usize].push((now, addr)); + state_ref.queued = true; + } } state_ref.last_update = now; ret @@ -595,7 +605,8 @@ impl Store { let cur_time = Instant::now(); { - let mut nodes = self.nodes.write().unwrap(); + let mut nodes_lock = self.nodes.write().unwrap(); + let nodes = nodes_lock.borrow_mut(); for (idx, state_nodes) in nodes.state_next_scan.iter_mut().enumerate() { let rescan_interval = cmp::max(self.get_u64(U64Setting::RescanInterval(AddressState::from_num(idx as u8).unwrap())), 1); let cmp_time = cur_time - Duration::from_secs(rescan_interval); @@ -604,6 +615,7 @@ impl Store { let mut new_nodes = state_nodes.split_off(split_point as usize); mem::swap(&mut new_nodes, state_nodes); for (_, node) in new_nodes.drain(..) { + nodes.nodes_to_state.get_mut(&node).unwrap().queued = false; res.push((&node).into()); } } -- 2.30.2 From 62770d56227d5458fd7502fedf065a80b6b16373 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 2 Dec 2019 15:09:47 -0500 Subject: [PATCH 12/16] Drop processing threads a bit --- src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index f9de3d4..ba326f7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -424,7 +424,7 @@ fn main() { unsafe { REQUEST_BLOCK = Some(Box::new(Mutex::new(Arc::new((0, genesis_block(Network::Bitcoin).bitcoin_hash(), genesis_block(Network::Bitcoin)))))) }; let trt = tokio::runtime::Builder::new() - .blocking_threads(2).core_threads(num_cpus::get().max(1) * 3) + .blocking_threads(2).core_threads(num_cpus::get().max(1) * 2) .build().unwrap(); let _ = trt.block_on_all(future::lazy(|| { -- 2.30.2 From b38bd902d7e10b3e944f613823a3d4d8000030be Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Tue, 3 Dec 2019 01:46:35 -0500 Subject: [PATCH 13/16] Impose a max conns/sec --- src/datastore.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/datastore.rs b/src/datastore.rs index 5f9874f..1ae9dd9 100644 --- a/src/datastore.rs +++ b/src/datastore.rs @@ -19,6 +19,7 @@ use regex::Regex; use crate::bgp_client::BGPClient; pub const SECS_PER_SCAN_RESULTS: u64 = 15; +const MAX_CONNS_PER_SEC_PER_STATUS: u64 = 30; #[derive(Clone, Copy, Hash, PartialEq, Eq)] pub enum AddressState { @@ -610,7 +611,8 @@ impl Store { for (idx, state_nodes) in nodes.state_next_scan.iter_mut().enumerate() { let rescan_interval = cmp::max(self.get_u64(U64Setting::RescanInterval(AddressState::from_num(idx as u8).unwrap())), 1); let cmp_time = cur_time - Duration::from_secs(rescan_interval); - let split_point = cmp::min(SECS_PER_SCAN_RESULTS * state_nodes.len() as u64 / rescan_interval, + let split_point = cmp::min(cmp::min(SECS_PER_SCAN_RESULTS * state_nodes.len() as u64 / rescan_interval, + SECS_PER_SCAN_RESULTS * MAX_CONNS_PER_SEC_PER_STATUS), state_nodes.binary_search_by(|a| a.0.cmp(&cmp_time)).unwrap_or_else(|idx| idx) as u64); let mut new_nodes = state_nodes.split_off(split_point as usize); mem::swap(&mut new_nodes, state_nodes); -- 2.30.2 From 8e27de4c043eb7967697c8c38b2e0b65bf546b8c Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Wed, 4 Dec 2019 19:29:22 -0500 Subject: [PATCH 14/16] Align routing table entries to single bytes --- src/bgp_client.rs | 67 +++++++++++++++++++++++++++++++++++------------ 1 file changed, 50 insertions(+), 17 deletions(-) diff --git a/src/bgp_client.rs b/src/bgp_client.rs index c20323d..9038005 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -31,9 +31,42 @@ struct Route { // 32 bytes med: u32, } +// 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. + +#[repr(packed)] +#[derive(PartialEq, Eq, Hash)] +struct V4Addr { + addr: [u8; 4], + pfxlen: u8, +} +impl From<(Ipv4Addr, u8)> for V4Addr { + fn from(p: (Ipv4Addr, u8)) -> Self { + Self { + addr: p.0.octets(), + pfxlen: p.1, + } + } +} + +#[repr(packed)] +#[derive(PartialEq, Eq, Hash)] +struct V6Addr { + addr: [u8; 16], + pfxlen: u8, +} +impl From<(Ipv6Addr, u8)> for V6Addr { + fn from(p: (Ipv6Addr, u8)) -> Self { + Self { + addr: p.0.octets(), + pfxlen: p.1, + } + } +} + struct RoutingTable { - v4_table: HashMap<(Ipv4Addr, u8), HashMap>, - v6_table: HashMap<(Ipv6Addr, u8), HashMap>, + v4_table: HashMap>, + v6_table: HashMap>, } impl RoutingTable { @@ -48,22 +81,22 @@ impl RoutingTable { macro_rules! lookup_res { ($addrty: ty, $addr: expr, $table: expr, $addr_bits: expr) => { { //TODO: Optimize this (probably means making the tables btrees)! - let mut lookup = $addr.octets(); + let mut lookup = <$addrty>::from(($addr, $addr_bits)); for i in 0..$addr_bits { - let lookup_addr = <$addrty>::from(lookup); - if let Some(routes) = $table.get(&(lookup_addr, $addr_bits - i as u8)).map(|hm| hm.values()) { + if let Some(routes) = $table.get(&lookup).map(|hm| hm.values()) { if routes.len() > 0 { - return ($addr_bits - i as u8, routes.collect()); + return (lookup.pfxlen, routes.collect()); } } - lookup[lookup.len() - (i/8) - 1] &= !(1u8 << (i % 8)); + lookup.addr[lookup.addr.len() - (i/8) - 1] &= !(1u8 << (i % 8)); + lookup.pfxlen -= 1; } (0, vec![]) } } } match ip { - IpAddr::V4(v4a) => lookup_res!(Ipv4Addr, v4a, self.v4_table, 32), - IpAddr::V6(v6a) => lookup_res!(Ipv6Addr, v6a, self.v6_table, 128) + IpAddr::V4(v4a) => lookup_res!(V4Addr, v4a, self.v4_table, 32), + IpAddr::V6(v6a) => lookup_res!(V6Addr, v6a, self.v6_table, 128) } } @@ -72,15 +105,15 @@ impl RoutingTable { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.get_mut(&(v4a, len)).and_then(|hm| hm.remove(&0)), - IpAddr::V6(v6a) => self.v6_table.get_mut(&(v6a, len)).and_then(|hm| hm.remove(&0)), + 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)), } }, 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)).and_then(|hm| hm.remove(&id)), - IpAddr::V6(v6a) => self.v6_table.get_mut(&(v6a, len)).and_then(|hm| hm.remove(&id)), + 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)), } }, NLRIEncoding::IP_MPLS(_) => None, @@ -92,15 +125,15 @@ impl RoutingTable { NLRIEncoding::IP(p) => { let (ip, len) = <(IpAddr, u8)>::from(&p); match ip { - IpAddr::V4(v4a) => self.v4_table.entry((v4a, len)).or_insert(HashMap::new()).insert(0, route), - IpAddr::V6(v6a) => self.v6_table.entry((v6a, len)).or_insert(HashMap::new()).insert(0, route), + 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), } }, 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)).or_insert(HashMap::new()).insert(id, route), - IpAddr::V6(v6a) => self.v6_table.entry((v6a, len)).or_insert(HashMap::new()).insert(id, route), + 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), } }, NLRIEncoding::IP_MPLS(_) => None, -- 2.30.2 From 7f3e64fc449bc64c569eb45d5e52472f56e1f3ed Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Wed, 4 Dec 2019 19:48:28 -0500 Subject: [PATCH 15/16] Further reduce memory in bgp_client --- src/bgp_client.rs | 66 +++++++++++++++++++++++++++++++++++------------ 1 file changed, 49 insertions(+), 17 deletions(-) diff --git a/src/bgp_client.rs b/src/bgp_client.rs index 9038005..ba3a164 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}; @@ -22,14 +22,16 @@ use futures::sync::mpsc; use crate::printer::{Printer, Stat}; use crate::timeout_stream::TimeoutStream; -const PATH_SUFFIX_LEN: usize = 3; +const PATH_SUFFIX_LEN: usize = 2; #[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 = 32 - 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,10 +69,16 @@ 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>, } impl RoutingTable { @@ -83,9 +95,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 +113,62 @@ 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) => { + 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(_) => (), }; } fn announce(&mut self, prefix: NLRIEncoding, route: Route) { + macro_rules! insert { + ($rt: expr, $v: expr, $id: expr) => { { + let entry = $rt.entry($v.into()).or_insert(Vec::new()); + entry.retain(|e| e.0 != $id); + entry.push(($id, route)); + } } + } 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(_) => (), }; } } -- 2.30.2 From 7ef74563891ca09202cd0a59f96f516fa44318f7 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Sun, 31 May 2020 13:30:07 -0400 Subject: [PATCH 16/16] Update to upstream bgp-rs --- Cargo.toml | 2 +- src/bgp_client.rs | 31 +++++++++++++++++++------------ 2 files changed, 20 insertions(+), 13 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 075f296..a628f3f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ edition = "2018" [dependencies] bitcoin = "0.19" bitcoin_hashes = "0.7" -bgp-rs = { git = "https://github.com/TheBlueMatt/bgp-rs", rev = "98f91c4e9e386bfdac0f8e2e9886fb15b15c861f" } +bgp-rs = { git = "https://github.com/TheBlueMatt/bgp-rs", rev = "96bd15e67789b91f679f835fc62ff3beb16685bf" } tokio = "0.1" bytes = "0.4" futures = "0.1" diff --git a/src/bgp_client.rs b/src/bgp_client.rs index ba3a164..dc5c673 100644 --- a/src/bgp_client.rs +++ b/src/bgp_client.rs @@ -142,6 +142,10 @@ impl RoutingTable { } }, NLRIEncoding::IP_MPLS(_) => (), + NLRIEncoding::IP_MPLS_WITH_PATH_ID(_) => (), + NLRIEncoding::IP_VPN_MPLS(_) => (), + NLRIEncoding::L2VPN(_) => (), + NLRIEncoding::FLOWSPEC(_) => (), }; } @@ -169,6 +173,10 @@ impl RoutingTable { } }, NLRIEncoding::IP_MPLS(_) => (), + NLRIEncoding::IP_MPLS_WITH_PATH_ID(_) => (), + NLRIEncoding::IP_VPN_MPLS(_) => (), + NLRIEncoding::L2VPN(_) => (), + NLRIEncoding::FLOWSPEC(_) => (), }; } } @@ -196,8 +204,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; @@ -206,13 +214,12 @@ 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); Ok(Some(msg)) @@ -224,12 +231,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(()) } } @@ -340,7 +347,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(|_| { @@ -359,7 +366,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) { -- 2.30.2