Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-12-25 15:29:32 -05:00
Code Issues Packages Projects Releases Wiki Activity

dont run routing table ticks that require the network until it has started up

Browse Source
This commit is contained in:
Christien Rioux 2024-05-19 10:49:37 -04:00 committed by k8wu
parent b37e2cc3c9
commit e82cdbbfce
8 changed files with 283 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
veilid-core/src/attachment_manager.rs
View File

@ -237,7 +237,7 @@ impl AttachmentManager {
}
// see if we need to restart the network
if netman.needs_restart() {
if netman.network_needs_restart() {
info!("Restarting network");
restart = true;
break;

23
veilid-core/src/network_manager/mod.rs
View File

@ -117,8 +117,9 @@ pub(crate) enum NodeContactMethod {
/// Must use outbound relay to reach the node
OutboundRelay(NodeRef),
}
#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd, Hash)]
#[derive(Clone, Debug, PartialEq, Eq, Ord, PartialOrd, Hash)]
struct NodeContactMethodCacheKey {
node_ids: TypedKeyGroup,
own_node_info_ts: Timestamp,
target_node_info_ts: Timestamp,
target_node_ref_filter: Option<NodeRefFilter>,
@ -305,6 +306,13 @@ impl NetworkManager {
.net
.clone()
}
fn opt_net(&self) -> Option<Network> {
self.unlocked_inner
.components
.read()
.as_ref()
.map(|x| x.net.clone())
}
fn receipt_manager(&self) -> ReceiptManager {
self.unlocked_inner
.components
@ -512,9 +520,16 @@ impl NetworkManager {
}
}
pub fn needs_restart(&self) -> bool {
let net = self.net();
net.needs_restart()
pub fn network_needs_restart(&self) -> bool {
self.opt_net()
.map(|net| net.needs_restart())
.unwrap_or(false)
}
pub fn network_is_started(&self) -> bool {
self.opt_net()
.and_then(|net| net.is_started())
.unwrap_or(false)
}
pub fn generate_node_status(&self, _routing_domain: RoutingDomain) -> NodeStatus {

355
veilid-core/src/network_manager/native/mod.rs
View File

@ -72,8 +72,8 @@ pub const MAX_CAPABILITIES: usize = 64;
/////////////////////////////////////////////////////////////////
struct NetworkInner {
/// true if the low-level network is running
network_started: bool,
/// Some(true) if the low-level network is running, Some(false) if it is not, None if it is in transit
network_started: Option<bool>,
/// set if the network needs to be restarted due to a low level configuration change
/// such as dhcp release or change of address or interfaces being added or removed
network_needs_restart: bool,
@ -137,7 +137,7 @@ pub(in crate::network_manager) struct Network {
impl Network {
fn new_inner() -> NetworkInner {
NetworkInner {
network_started: false,
network_started: Some(false),
network_needs_restart: false,
needs_public_dial_info_check: false,
network_already_cleared: false,
@ -675,196 +675,209 @@ impl Network {
#[instrument(level = "debug", err, skip_all)]
pub async fn startup(&self) -> EyreResult<()> {
// initialize interfaces
self.unlocked_inner.interfaces.refresh().await?;
self.inner.lock().network_started = None;
let startup_func = async {
// initialize interfaces
self.unlocked_inner.interfaces.refresh().await?;
// build the set of networks we should consider for the 'LocalNetwork' routing domain
let mut local_networks: HashSet<(IpAddr, IpAddr)> = HashSet::new();
self.unlocked_inner
.interfaces
.with_interfaces(|interfaces| {
log_net!(debug "interfaces: {:#?}", interfaces);
// build the set of networks we should consider for the 'LocalNetwork' routing domain
let mut local_networks: HashSet<(IpAddr, IpAddr)> = HashSet::new();
self.unlocked_inner
.interfaces
.with_interfaces(|interfaces| {
log_net!(debug "interfaces: {:#?}", interfaces);
for intf in interfaces.values() {
// Skip networks that we should never encounter
if intf.is_loopback() || !intf.is_running() {
continue;
for intf in interfaces.values() {
// Skip networks that we should never encounter
if intf.is_loopback() || !intf.is_running() {
continue;
}
// Add network to local networks table
for addr in &intf.addrs {
let netmask = addr.if_addr().netmask();
let network_ip = ipaddr_apply_netmask(addr.if_addr().ip(), netmask);
local_networks.insert((network_ip, netmask));
}
}
// Add network to local networks table
for addr in &intf.addrs {
let netmask = addr.if_addr().netmask();
let network_ip = ipaddr_apply_netmask(addr.if_addr().ip(), netmask);
local_networks.insert((network_ip, netmask));
}
}
});
let local_networks: Vec<(IpAddr, IpAddr)> = local_networks.into_iter().collect();
self.unlocked_inner
.routing_table
.configure_local_network_routing_domain(local_networks);
});
let local_networks: Vec<(IpAddr, IpAddr)> = local_networks.into_iter().collect();
self.unlocked_inner
.routing_table
.configure_local_network_routing_domain(local_networks);
// determine if we have ipv4/ipv6 addresses
{
let mut inner = self.inner.lock();
inner.enable_ipv4 = false;
for addr in self.get_stable_interface_addresses() {
if addr.is_ipv4() {
log_net!(debug "enable address {:?} as ipv4", addr);
inner.enable_ipv4 = true;
} else if addr.is_ipv6() {
let address = Address::from_ip_addr(addr);
if address.is_global() {
log_net!(debug "enable address {:?} as ipv6 global", address);
inner.enable_ipv6_global = true;
} else if address.is_local() {
log_net!(debug "enable address {:?} as ipv6 local", address);
inner.enable_ipv6_local = true;
// determine if we have ipv4/ipv6 addresses
{
let mut inner = self.inner.lock();
inner.enable_ipv4 = false;
for addr in self.get_stable_interface_addresses() {
if addr.is_ipv4() {
log_net!(debug "enable address {:?} as ipv4", addr);
inner.enable_ipv4 = true;
} else if addr.is_ipv6() {
let address = Address::from_ip_addr(addr);
if address.is_global() {
log_net!(debug "enable address {:?} as ipv6 global", address);
inner.enable_ipv6_global = true;
} else if address.is_local() {
log_net!(debug "enable address {:?} as ipv6 local", address);
inner.enable_ipv6_local = true;
}
}
}
}
}
// Build our protocol config to share it with other nodes
let protocol_config = {
let mut inner = self.inner.lock();
// Create stop source
inner.stop_source = Some(StopSource::new());
// get protocol config
// Build our protocol config to share it with other nodes
let protocol_config = {
let c = self.config.get();
let mut inbound = ProtocolTypeSet::new();
let mut inner = self.inner.lock();
if c.network.protocol.udp.enabled {
inbound.insert(ProtocolType::UDP);
}
if c.network.protocol.tcp.listen {
inbound.insert(ProtocolType::TCP);
}
if c.network.protocol.ws.listen {
inbound.insert(ProtocolType::WS);
}
if c.network.protocol.wss.listen {
inbound.insert(ProtocolType::WSS);
}
// Create stop source
inner.stop_source = Some(StopSource::new());
let mut outbound = ProtocolTypeSet::new();
if c.network.protocol.udp.enabled {
outbound.insert(ProtocolType::UDP);
}
if c.network.protocol.tcp.connect {
outbound.insert(ProtocolType::TCP);
}
if c.network.protocol.ws.connect {
outbound.insert(ProtocolType::WS);
}
if c.network.protocol.wss.connect {
outbound.insert(ProtocolType::WSS);
}
// get protocol config
let protocol_config = {
let c = self.config.get();
let mut inbound = ProtocolTypeSet::new();
let mut family_global = AddressTypeSet::new();
let mut family_local = AddressTypeSet::new();
if inner.enable_ipv4 {
family_global.insert(AddressType::IPV4);
family_local.insert(AddressType::IPV4);
}
if inner.enable_ipv6_global {
family_global.insert(AddressType::IPV6);
}
if inner.enable_ipv6_local {
family_local.insert(AddressType::IPV6);
}
if c.network.protocol.udp.enabled {
inbound.insert(ProtocolType::UDP);
}
if c.network.protocol.tcp.listen {
inbound.insert(ProtocolType::TCP);
}
if c.network.protocol.ws.listen {
inbound.insert(ProtocolType::WS);
}
if c.network.protocol.wss.listen {
inbound.insert(ProtocolType::WSS);
}
// set up the routing table's network config
// if we have static public dialinfo, upgrade our network class
let public_internet_capabilities = {
PUBLIC_INTERNET_CAPABILITIES
.iter()
.copied()
.filter(|cap| !c.capabilities.disable.contains(cap))
.collect::<Vec<Capability>>()
};
let local_network_capabilities = {
LOCAL_NETWORK_CAPABILITIES
.iter()
.copied()
.filter(|cap| !c.capabilities.disable.contains(cap))
.collect::<Vec<Capability>>()
let mut outbound = ProtocolTypeSet::new();
if c.network.protocol.udp.enabled {
outbound.insert(ProtocolType::UDP);
}
if c.network.protocol.tcp.connect {
outbound.insert(ProtocolType::TCP);
}
if c.network.protocol.ws.connect {
outbound.insert(ProtocolType::WS);
}
if c.network.protocol.wss.connect {
outbound.insert(ProtocolType::WSS);
}
let mut family_global = AddressTypeSet::new();
let mut family_local = AddressTypeSet::new();
if inner.enable_ipv4 {
family_global.insert(AddressType::IPV4);
family_local.insert(AddressType::IPV4);
}
if inner.enable_ipv6_global {
family_global.insert(AddressType::IPV6);
}
if inner.enable_ipv6_local {
family_local.insert(AddressType::IPV6);
}
// set up the routing table's network config
// if we have static public dialinfo, upgrade our network class
let public_internet_capabilities = {
PUBLIC_INTERNET_CAPABILITIES
.iter()
.copied()
.filter(|cap| !c.capabilities.disable.contains(cap))
.collect::<Vec<Capability>>()
};
let local_network_capabilities = {
LOCAL_NETWORK_CAPABILITIES
.iter()
.copied()
.filter(|cap| !c.capabilities.disable.contains(cap))
.collect::<Vec<Capability>>()
};
ProtocolConfig {
outbound,
inbound,
family_global,
family_local,
public_internet_capabilities,
local_network_capabilities,
}
};
inner.protocol_config = protocol_config.clone();
ProtocolConfig {
outbound,
inbound,
family_global,
family_local,
public_internet_capabilities,
local_network_capabilities,
}
protocol_config
};
inner.protocol_config = protocol_config.clone();
protocol_config
};
// Start editing routing table
let mut editor_public_internet = self
.unlocked_inner
.routing_table
.edit_routing_domain(RoutingDomain::PublicInternet);
let mut editor_local_network = self
.unlocked_inner
.routing_table
.edit_routing_domain(RoutingDomain::LocalNetwork);
// Start editing routing table
let mut editor_public_internet = self
.unlocked_inner
.routing_table
.edit_routing_domain(RoutingDomain::PublicInternet);
let mut editor_local_network = self
.unlocked_inner
.routing_table
.edit_routing_domain(RoutingDomain::LocalNetwork);
// start listeners
if protocol_config.inbound.contains(ProtocolType::UDP) {
self.bind_udp_protocol_handlers(&mut editor_public_internet, &mut editor_local_network)
// start listeners
if protocol_config.inbound.contains(ProtocolType::UDP) {
self.bind_udp_protocol_handlers(
&mut editor_public_internet,
&mut editor_local_network,
)
.await?;
}
if protocol_config.inbound.contains(ProtocolType::WS) {
self.start_ws_listeners(&mut editor_public_internet, &mut editor_local_network)
.await?;
}
if protocol_config.inbound.contains(ProtocolType::WSS) {
self.start_wss_listeners(&mut editor_public_internet, &mut editor_local_network)
.await?;
}
if protocol_config.inbound.contains(ProtocolType::TCP) {
self.start_tcp_listeners(&mut editor_public_internet, &mut editor_local_network)
.await?;
}
editor_public_internet.setup_network(
protocol_config.outbound,
protocol_config.inbound,
protocol_config.family_global,
protocol_config.public_internet_capabilities,
);
editor_local_network.setup_network(
protocol_config.outbound,
protocol_config.inbound,
protocol_config.family_local,
protocol_config.local_network_capabilities,
);
let detect_address_changes = {
let c = self.config.get();
c.network.detect_address_changes
};
if !detect_address_changes {
let inner = self.inner.lock();
if !inner.static_public_dialinfo.is_empty() {
editor_public_internet.set_network_class(Some(NetworkClass::InboundCapable));
}
}
if protocol_config.inbound.contains(ProtocolType::WS) {
self.start_ws_listeners(&mut editor_public_internet, &mut editor_local_network)
.await?;
}
if protocol_config.inbound.contains(ProtocolType::WSS) {
self.start_wss_listeners(&mut editor_public_internet, &mut editor_local_network)
.await?;
}
if protocol_config.inbound.contains(ProtocolType::TCP) {
self.start_tcp_listeners(&mut editor_public_internet, &mut editor_local_network)
.await?;
}
// commit routing table edits
editor_public_internet.commit(true).await;
editor_local_network.commit(true).await;
editor_public_internet.setup_network(
protocol_config.outbound,
protocol_config.inbound,
protocol_config.family_global,
protocol_config.public_internet_capabilities,
);
editor_local_network.setup_network(
protocol_config.outbound,
protocol_config.inbound,
protocol_config.family_local,
protocol_config.local_network_capabilities,
);
let detect_address_changes = {
let c = self.config.get();
c.network.detect_address_changes
};
if !detect_address_changes {
let inner = self.inner.lock();
if !inner.static_public_dialinfo.is_empty() {
editor_public_internet.set_network_class(Some(NetworkClass::InboundCapable));
}
}
// commit routing table edits
editor_public_internet.commit(true).await;
editor_local_network.commit(true).await;
Ok(())
};
let res = startup_func.await;
if res.is_err() {
info!("network failed to start");
self.inner.lock().network_started = Some(false);
return res;
}
info!("network started");
self.inner.lock().network_started = true;
self.inner.lock().network_started = Some(true);
Ok(())
}
@ -872,7 +885,7 @@ impl Network {
self.inner.lock().network_needs_restart
}
pub fn is_started(&self) -> bool {
pub fn is_started(&self) -> Option<bool> {
self.inner.lock().network_started
}

1
veilid-core/src/network_manager/send_data.rs
View File

@ -394,6 +394,7 @@ impl NetworkManager {
// Get cache key
let ncm_key = NodeContactMethodCacheKey {
node_ids: target_node_ref.node_ids(),
own_node_info_ts: routing_table.get_own_node_info_ts(routing_domain),
target_node_info_ts: target_node_ref.node_info_ts(routing_domain),
target_node_ref_filter: target_node_ref.filter_ref().cloned(),

10
veilid-core/src/network_manager/stats.rs
View File

@ -76,15 +76,7 @@ impl NetworkManager {
}
pub fn get_veilid_state(&self) -> Box<VeilidStateNetwork> {
let has_state = self
.unlocked_inner
.components
.read()
.as_ref()
.map(|c| c.net.is_started())
.unwrap_or(false);
if !has_state {
if !self.network_is_started() {
return Box::new(VeilidStateNetwork {
started: false,
bps_down: 0.into(),

127
veilid-core/src/network_manager/wasm/mod.rs
View File

@ -52,7 +52,7 @@ pub const MAX_CAPABILITIES: usize = 64;
/////////////////////////////////////////////////////////////////
struct NetworkInner {
network_started: bool,
network_started: Option<bool>,
network_needs_restart: bool,
protocol_config: ProtocolConfig,
}
@ -74,7 +74,7 @@ pub(in crate::network_manager) struct Network {
impl Network {
fn new_inner() -> NetworkInner {
NetworkInner {
network_started: false,
network_started: Some(false),
network_needs_restart: false,
protocol_config: Default::default(),
}
@ -334,70 +334,81 @@ impl Network {
/////////////////////////////////////////////////////////////////
pub async fn startup(&self) -> EyreResult<()> {
log_net!(debug "starting network");
// get protocol config
let protocol_config = {
let c = self.config.get();
let inbound = ProtocolTypeSet::new();
let mut outbound = ProtocolTypeSet::new();
self.inner.lock().network_started = None;
let startup_func = async {
log_net!(debug "starting network");
// get protocol config
let protocol_config = {
let c = self.config.get();
let inbound = ProtocolTypeSet::new();
let mut outbound = ProtocolTypeSet::new();
if c.network.protocol.ws.connect {
outbound.insert(ProtocolType::WS);
}
if c.network.protocol.wss.connect {
outbound.insert(ProtocolType::WSS);
}
if c.network.protocol.ws.connect {
outbound.insert(ProtocolType::WS);
}
if c.network.protocol.wss.connect {
outbound.insert(ProtocolType::WSS);
}
let supported_address_types: AddressTypeSet = if is_ipv6_supported() {
AddressType::IPV4 | AddressType::IPV6
} else {
AddressType::IPV4.into()
let supported_address_types: AddressTypeSet = if is_ipv6_supported() {
AddressType::IPV4 | AddressType::IPV6
} else {
AddressType::IPV4.into()
};
let family_global = supported_address_types;
let family_local = supported_address_types;
let public_internet_capabilities = {
PUBLIC_INTERNET_CAPABILITIES
.iter()
.copied()
.filter(|cap| !c.capabilities.disable.contains(cap))
.collect::<Vec<Capability>>()
};
ProtocolConfig {
outbound,
inbound,
family_global,
family_local,
local_network_capabilities: vec![],
public_internet_capabilities,
}
};
self.inner.lock().protocol_config = protocol_config.clone();
let family_global = supported_address_types;
let family_local = supported_address_types;
// Start editing routing table
let mut editor_public_internet = self
.unlocked_inner
.routing_table
.edit_routing_domain(RoutingDomain::PublicInternet);
let public_internet_capabilities = {
PUBLIC_INTERNET_CAPABILITIES
.iter()
.copied()
.filter(|cap| !c.capabilities.disable.contains(cap))
.collect::<Vec<Capability>>()
};
// set up the routing table's network config
// if we have static public dialinfo, upgrade our network class
ProtocolConfig {
outbound,
inbound,
family_global,
family_local,
local_network_capabilities: vec![],
public_internet_capabilities,
}
editor_public_internet.setup_network(
protocol_config.outbound,
protocol_config.inbound,
protocol_config.family_global,
protocol_config.public_internet_capabilities.clone(),
);
editor_public_internet.set_network_class(Some(NetworkClass::WebApp));
// commit routing table edits
editor_public_internet.commit(true).await;
Ok(())
};
self.inner.lock().protocol_config = protocol_config.clone();
// Start editing routing table
let mut editor_public_internet = self
.unlocked_inner
.routing_table
.edit_routing_domain(RoutingDomain::PublicInternet);
let res = startup_func.await;
if res.is_err() {
info!("network failed to start");
self.inner.lock().network_started = Some(false);
return res;
}
// set up the routing table's network config
// if we have static public dialinfo, upgrade our network class
editor_public_internet.setup_network(
protocol_config.outbound,
protocol_config.inbound,
protocol_config.family_global,
protocol_config.public_internet_capabilities.clone(),
);
editor_public_internet.set_network_class(Some(NetworkClass::WebApp));
// commit routing table edits
editor_public_internet.commit(true).await;
self.inner.lock().network_started = true;
log_net!(debug "network started");
info!("network started");
self.inner.lock().network_started = Some(true);
Ok(())
}
@ -405,7 +416,7 @@ impl Network {
self.inner.lock().network_needs_restart
}
pub fn is_started(&self) -> bool {
pub fn is_started(&self) -> Option<bool> {
self.inner.lock().network_started
}

3
veilid-core/src/routing_table/tasks/bootstrap.rs
View File

@ -287,6 +287,9 @@ impl RoutingTable {
Ok(NodeContactMethod::Direct(v)) => v,
Ok(v) => {
log_rtab!(warn "invalid contact method for bootstrap, ignoring peer: {:?}", v);
let _ = routing_table
.network_manager()
.get_node_contact_method(nr.clone());
return;
}
Err(e) => {

5
veilid-core/src/routing_table/tasks/mod.rs
View File

@ -149,6 +149,11 @@ impl RoutingTable {
inner.refresh_cached_entry_counts()
};
// Only do the rest if the network has started
if !self.network_manager().network_is_started() {
return Ok(());
}
let min_peer_count = self.with_config(|c| c.network.dht.min_peer_count as usize);
// Figure out which tables need bootstrap or peer minimum refresh