From 5efbf4853a04fff6da13e14da82eb190ab322dfb Mon Sep 17 00:00:00 2001 From: "sijie.sun" Date: Sat, 13 Jun 2026 13:44:58 +0800 Subject: [PATCH] feat: apply shared nic member ifcfg Route shared nic IP and route configuration through member claims. Repeated addresses or routes now touch the OS device only when the first owner appears, and are removed only after the last owner leaves. Forward NicBackend shared operations to the member claim layer. Dynamic proxy and public IPv6 route updates now go through NicBackend, so they use the same ownership merge behavior. Clean member claims when shared member registration is dropped. Keep raw ifcfg access for non-Linux platform cleanup only, and document that this raw path does not carry the netns guard. --- easytier/src/instance/shared_virtual_nic.rs | 259 +++++++++++++++++++- easytier/src/instance/virtual_nic.rs | 134 +++++----- 2 files changed, 322 insertions(+), 71 deletions(-) diff --git a/easytier/src/instance/shared_virtual_nic.rs b/easytier/src/instance/shared_virtual_nic.rs index f04c128d..aaa0009c 100644 --- a/easytier/src/instance/shared_virtual_nic.rs +++ b/easytier/src/instance/shared_virtual_nic.rs @@ -17,6 +17,9 @@ use crate::{ use super::virtual_nic::{VirtualNic, VirtualNicConfig}; +#[cfg(not(target_os = "linux"))] +use crate::common::ifcfg::IfConfiger; + mod dispatcher; use dispatcher::{SharedVirtualNicDispatcher, SharedVirtualNicMemberTunnelTable}; @@ -239,6 +242,13 @@ impl SharedIfConfig { effective_mtu: self.effective_mtu(), } } + + fn claims_of(&self, member_id: SharedVirtualNicMemberId) -> SharedIfConfigClaims { + self.member_claims + .get(&member_id) + .cloned() + .unwrap_or_default() + } } pub struct SharedVirtualNic { @@ -280,6 +290,97 @@ impl SharedVirtualNic { self.nic.clone() } + #[cfg(not(target_os = "linux"))] + async fn ifcfg_and_ifname(&self) -> Result<(IfConfiger, String), Error> { + self.ensure_valid()?; + let nic = self.nic.lock().await; + Ok((nic.get_ifcfg(), nic.ifname().to_owned())) + } + + async fn link_up(&self) -> Result<(), Error> { + self.ensure_valid()?; + self.nic.lock().await.link_up().await + } + + async fn apply_member_claims( + &mut self, + member_id: SharedVirtualNicMemberId, + claims: SharedIfConfigClaims, + ) -> Result<(), Error> { + self.ensure_valid()?; + + let mut next_ifcfg = self.ifcfg.clone(); + let delta = next_ifcfg.apply_member_claims(member_id, claims); + self.apply_ifcfg_delta(&delta).await?; + self.ifcfg = next_ifcfg; + + Ok(()) + } + + async fn remove_member_claims( + &mut self, + member_id: SharedVirtualNicMemberId, + ) -> Result<(), Error> { + self.ensure_valid()?; + + let mut next_ifcfg = self.ifcfg.clone(); + let Some(delta) = next_ifcfg.remove_member(member_id) else { + return Ok(()); + }; + self.apply_ifcfg_delta(&delta).await?; + self.ifcfg = next_ifcfg; + + Ok(()) + } + + async fn apply_ifcfg_delta(&self, delta: &SharedIfConfigDelta) -> Result<(), Error> { + let nic = self.nic.lock().await; + + for route in &delta.ipv4_routes.removed { + nic.remove_route(route.address, route.prefix).await?; + } + for route in &delta.ipv6_routes.removed { + nic.remove_ipv6_route(route.address, route.prefix).await?; + } + for ip in &delta.ipv4_addresses.removed { + nic.remove_ip(Some(*ip)).await?; + } + for ip in &delta.ipv6_addresses.removed { + nic.remove_ipv6(Some(*ip)).await?; + } + + for ip in &delta.ipv4_addresses.added { + nic.add_ip(ip.address(), ip.network_length() as i32).await?; + } + for ip in &delta.ipv6_addresses.added { + nic.add_ipv6(ip.address(), ip.network_length() as i32) + .await?; + } + for route in &delta.ipv4_routes.added { + nic.add_route_with_cost(route.address, route.prefix, route.cost) + .await?; + } + for route in &delta.ipv6_routes.added { + nic.add_ipv6_route_with_cost(route.address, route.prefix, route.cost) + .await?; + } + + if let Some(mtu) = &delta.mtu { + nic.set_mtu(mtu.new.unwrap_or_else(|| nic.configured_mtu())) + .await?; + } + + Ok(()) + } + + fn ensure_valid(&self) -> Result<(), Error> { + if self.is_valid() { + return Ok(()); + } + + Err(anyhow::anyhow!("shared virtual nic is invalid").into()) + } + fn member_tunnel_table(&self) -> SharedVirtualNicMemberTunnelTable { self.member_tunnel_table.clone() } @@ -289,9 +390,7 @@ impl SharedVirtualNic { } async fn ensure_dispatcher(&mut self) -> Result<(), Error> { - if !self.is_valid() { - return Err(anyhow::anyhow!("shared virtual nic is invalid").into()); - } + self.ensure_valid()?; if self.dispatcher.is_some() { return Ok(()); @@ -309,6 +408,7 @@ impl SharedVirtualNic { struct SharedVirtualNicMemberRegistration { member_id: SharedVirtualNicMemberId, + shared_nic: Arc>, member_tunnel_table: SharedVirtualNicMemberTunnelTable, } @@ -326,6 +426,27 @@ impl SharedVirtualNicMemberRegistration { impl Drop for SharedVirtualNicMemberRegistration { fn drop(&mut self) { self.member_tunnel_table.unregister(self.member_id); + let shared_nic = self.shared_nic.clone(); + let member_id = self.member_id; + + let Ok(handle) = tokio::runtime::Handle::try_current() else { + tracing::warn!( + ?member_id, + "skip shared virtual nic member claim cleanup without tokio runtime" + ); + return; + }; + + handle.spawn(async move { + let mut shared_nic = shared_nic.lock().await; + if let Err(err) = shared_nic.remove_member_claims(member_id).await { + tracing::warn!( + ?member_id, + ?err, + "failed to clean shared virtual nic member claims" + ); + } + }); } } @@ -346,10 +467,11 @@ impl SharedVirtualNicMember { ) -> Self { Self { member_id, - shared_nic, + shared_nic: shared_nic.clone(), close_notifier, registration: Arc::new(SharedVirtualNicMemberRegistration { member_id, + shared_nic: shared_nic.clone(), member_tunnel_table, }), } @@ -377,6 +499,119 @@ impl SharedVirtualNicMember { .register_tunnel(shared_tunnel, self.close_notifier.clone())?; Ok(member_tunnel) } + + #[cfg(not(target_os = "linux"))] + pub async fn ifcfg_and_ifname(&self) -> Result<(IfConfiger, String), Error> { + self.shared_nic.lock().await.ifcfg_and_ifname().await + } + + pub async fn link_up(&self) -> Result<(), Error> { + self.shared_nic.lock().await.link_up().await + } + + pub async fn add_ip(&self, ip: Ipv4Addr, cidr: i32) -> Result<(), Error> { + let ip = ipv4_inet(ip, cidr)?; + self.update_claims(|claims| { + claims.ipv4_addresses.insert(ip); + }) + .await + } + + pub async fn remove_ip(&self, ip: Option) -> Result<(), Error> { + self.update_claims(|claims| match ip { + Some(ip) => { + claims.ipv4_addresses.remove(&ip); + } + None => { + claims.ipv4_addresses.clear(); + } + }) + .await + } + + pub async fn add_ipv6(&self, ip: Ipv6Addr, cidr: i32) -> Result<(), Error> { + let ip = ipv6_inet(ip, cidr)?; + self.update_claims(|claims| { + claims.ipv6_addresses.insert(ip); + }) + .await + } + + pub async fn remove_ipv6(&self, ip: Option) -> Result<(), Error> { + self.update_claims(|claims| match ip { + Some(ip) => { + claims.ipv6_addresses.remove(&ip); + } + None => { + claims.ipv6_addresses.clear(); + } + }) + .await + } + + pub async fn add_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + self.add_route_with_cost(address, cidr, None).await + } + + pub async fn add_route_with_cost( + &self, + address: Ipv4Addr, + cidr: u8, + cost: Option, + ) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv4_routes + .insert(SharedIpv4Route::new(address, cidr, cost)); + }) + .await + } + + pub async fn remove_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv4_routes + .retain(|route| route.address != address || route.prefix != cidr); + }) + .await + } + + pub async fn add_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { + self.add_ipv6_route_with_cost(address, cidr, None).await + } + + pub async fn add_ipv6_route_with_cost( + &self, + address: Ipv6Addr, + cidr: u8, + cost: Option, + ) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv6_routes + .insert(SharedIpv6Route::new(address, cidr, cost)); + }) + .await + } + + pub async fn remove_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv6_routes + .retain(|route| route.address != address || route.prefix != cidr); + }) + .await + } + + async fn update_claims(&self, update: F) -> Result<(), Error> + where + F: FnOnce(&mut SharedIfConfigClaims) + Send, + { + let mut shared_nic = self.shared_nic.lock().await; + let mut claims = shared_nic.ifcfg.claims_of(self.member_id); + update(&mut claims); + shared_nic.apply_member_claims(self.member_id, claims).await + } } #[derive(Default)] @@ -577,6 +812,22 @@ where owners.get(item).cloned().unwrap_or_default() } +fn ipv4_inet(address: Ipv4Addr, prefix: i32) -> Result { + let prefix = u8::try_from(prefix) + .map_err(|_| anyhow::anyhow!("invalid IPv4 prefix length {}", prefix))?; + Ipv4Inet::new(address, prefix).map_err(|err| { + anyhow::anyhow!("invalid IPv4 address {}/{}: {:?}", address, prefix, err).into() + }) +} + +fn ipv6_inet(address: Ipv6Addr, prefix: i32) -> Result { + let prefix = u8::try_from(prefix) + .map_err(|_| anyhow::anyhow!("invalid IPv6 prefix length {}", prefix))?; + Ipv6Inet::new(address, prefix).map_err(|err| { + anyhow::anyhow!("invalid IPv6 address {}/{}: {:?}", address, prefix, err).into() + }) +} + #[cfg(test)] mod tests { use std::str::FromStr as _; diff --git a/easytier/src/instance/virtual_nic.rs b/easytier/src/instance/virtual_nic.rs index e01b8cff..b0cc00ca 100644 --- a/easytier/src/instance/virtual_nic.rs +++ b/easytier/src/instance/virtual_nic.rs @@ -742,9 +742,26 @@ impl VirtualNic { } pub async fn add_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + self.add_route_with_cost(address, cidr, None).await + } + + pub async fn add_route_with_cost( + &self, + address: Ipv4Addr, + cidr: u8, + cost: Option, + ) -> Result<(), Error> { let _g = self.config.net_ns.guard(); self.ifcfg - .add_ipv4_route(self.ifname(), address, cidr, None) + .add_ipv4_route(self.ifname(), address, cidr, cost) + .await?; + Ok(()) + } + + pub async fn remove_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + let _g = self.config.net_ns.guard(); + self.ifcfg + .remove_ipv4_route(self.ifname(), address, cidr) .await?; Ok(()) } @@ -802,6 +819,16 @@ impl VirtualNic { Ok(()) } + pub async fn set_mtu(&self, mtu: u32) -> Result<(), Error> { + let _g = self.config.net_ns.guard(); + self.ifcfg.set_mtu(self.ifname(), mtu).await?; + Ok(()) + } + + pub fn configured_mtu(&self) -> u32 { + self.config.mtu + } + pub fn get_ifcfg(&self) -> IfConfiger { IfConfiger {} } @@ -814,10 +841,6 @@ pub enum NicBackend { } impl NicBackend { - fn shared_not_implemented(operation: &str) -> Error { - anyhow::anyhow!("shared virtual nic {} is not implemented", operation).into() - } - pub fn dedicated(nic: Arc>) -> Self { Self::Dedicated(nic) } @@ -867,34 +890,46 @@ impl NicBackend { } } + #[cfg(not(target_os = "linux"))] + /// Returns a raw ifcfg handle and interface name for platform cleanup. + /// + /// This does not carry `VirtualNic`'s netns guard. Use the typed + /// `NicBackend` methods for normal IP and route configuration. pub async fn ifcfg_and_ifname(&self) -> Result<(IfConfiger, String), Error> { match self { Self::Dedicated(nic) => { let nic = nic.lock().await; Ok((nic.get_ifcfg(), nic.ifname().to_owned())) } - Self::Shared(_) => Err(Self::shared_not_implemented("ifcfg")), + Self::Shared(member) => member.ifcfg_and_ifname().await, } } pub async fn link_up(&self) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.link_up().await, - Self::Shared(_) => Err(Self::shared_not_implemented("link_up")), + Self::Shared(member) => member.link_up().await, } } pub async fn add_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_route(address, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_route")), + Self::Shared(member) => member.add_route(address, cidr).await, + } + } + + pub async fn remove_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + match self { + Self::Dedicated(nic) => nic.lock().await.remove_route(address, cidr).await, + Self::Shared(member) => member.remove_route(address, cidr).await, } } pub async fn add_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_ipv6_route(address, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_ipv6_route")), + Self::Shared(member) => member.add_ipv6_route(address, cidr).await, } } @@ -911,42 +946,42 @@ impl NicBackend { .add_ipv6_route_with_cost(address, cidr, cost) .await } - Self::Shared(_) => Err(Self::shared_not_implemented("add_ipv6_route_with_cost")), + Self::Shared(member) => member.add_ipv6_route_with_cost(address, cidr, cost).await, } } pub async fn remove_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.remove_ipv6_route(address, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("remove_ipv6_route")), + Self::Shared(member) => member.remove_ipv6_route(address, cidr).await, } } pub async fn remove_ip(&self, ip: Option) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.remove_ip(ip).await, - Self::Shared(_) => Err(Self::shared_not_implemented("remove_ip")), + Self::Shared(member) => member.remove_ip(ip).await, } } pub async fn remove_ipv6(&self, ip: Option) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.remove_ipv6(ip).await, - Self::Shared(_) => Err(Self::shared_not_implemented("remove_ipv6")), + Self::Shared(member) => member.remove_ipv6(ip).await, } } pub async fn add_ip(&self, ip: Ipv4Addr, cidr: i32) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_ip(ip, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_ip")), + Self::Shared(member) => member.add_ip(ip, cidr).await, } } pub async fn add_ipv6(&self, ip: Ipv6Addr, cidr: i32) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_ipv6(ip, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_ipv6")), + Self::Shared(member) => member.add_ipv6(ip, cidr).await, } } } @@ -1280,9 +1315,7 @@ impl NicCtx { } async fn apply_route_changes( - ifcfg: &impl IfConfiguerTrait, - ifname: &str, - net_ns: &crate::common::netns::NetNS, + backend: &NicBackend, cur_proxy_cidrs: &mut BTreeSet, added: Vec, removed: Vec, @@ -1294,9 +1327,8 @@ impl NicCtx { if !cur_proxy_cidrs.contains(&cidr) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .remove_ipv4_route(ifname, cidr.first_address(), cidr.network_length()) + let ret = backend + .remove_route(cidr.first_address(), cidr.network_length()) .await; if ret.is_err() { @@ -1314,9 +1346,8 @@ impl NicCtx { if cur_proxy_cidrs.contains(&cidr) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .add_ipv4_route(ifname, cidr.first_address(), cidr.network_length(), None) + let ret = backend + .add_route(cidr.first_address(), cidr.network_length()) .await; if ret.is_err() { @@ -1331,9 +1362,7 @@ impl NicCtx { } async fn apply_public_ipv6_route_changes( - ifcfg: &impl IfConfiguerTrait, - ifname: &str, - net_ns: &crate::common::netns::NetNS, + backend: &NicBackend, cur_routes: &mut BTreeSet, added: Vec, removed: Vec, @@ -1342,9 +1371,8 @@ impl NicCtx { if !cur_routes.contains(&route) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .remove_ipv6_route(ifname, route.address(), route.network_length()) + let ret = backend + .remove_ipv6_route(route.address(), route.network_length()) .await; if ret.is_err() { tracing::trace!(route = ?route, err = ?ret, "remove public ipv6 route failed"); @@ -1356,9 +1384,8 @@ impl NicCtx { if cur_routes.contains(&route) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .add_ipv6_route(ifname, route.address(), route.network_length(), None) + let ret = backend + .add_ipv6_route(route.address(), route.network_length()) .await; if ret.is_err() { tracing::trace!(route = ?route, err = ?ret, "add public ipv6 route failed"); @@ -1373,8 +1400,7 @@ impl NicCtx { return Err(anyhow::anyhow!("peer manager not available").into()); }; let global_ctx = self.global_ctx.clone(); - let net_ns = self.global_ctx.net_ns.clone(); - let (ifcfg, ifname) = self.backend.ifcfg_and_ifname().await?; + let backend = self.backend.clone(); let mut event_receiver = global_ctx.subscribe(); self.tasks.spawn(async move { @@ -1387,15 +1413,7 @@ impl NicCtx { &cur_proxy_cidrs, ) .await; - Self::apply_route_changes( - &ifcfg, - &ifname, - &net_ns, - &mut cur_proxy_cidrs, - added, - removed, - ) - .await; + Self::apply_route_changes(&backend, &mut cur_proxy_cidrs, added, removed).await; loop { let event = match event_receiver.recv().await { @@ -1426,15 +1444,7 @@ impl NicCtx { _ => continue, }; - Self::apply_route_changes( - &ifcfg, - &ifname, - &net_ns, - &mut cur_proxy_cidrs, - added, - removed, - ) - .await; + Self::apply_route_changes(&backend, &mut cur_proxy_cidrs, added, removed).await; } }); @@ -1446,8 +1456,7 @@ impl NicCtx { return Err(anyhow::anyhow!("peer manager not available").into()); }; let global_ctx = self.global_ctx.clone(); - let net_ns = self.global_ctx.net_ns.clone(); - let (ifcfg, ifname) = self.backend.ifcfg_and_ifname().await?; + let backend = self.backend.clone(); let mut event_receiver = global_ctx.subscribe(); self.tasks.spawn(async move { @@ -1455,9 +1464,7 @@ impl NicCtx { let initial_routes = peer_mgr.list_public_ipv6_routes().await; let initial_added = initial_routes.iter().copied().collect::>(); Self::apply_public_ipv6_route_changes( - &ifcfg, - &ifname, - &net_ns, + &backend, &mut cur_routes, initial_added, Vec::new(), @@ -1482,15 +1489,8 @@ impl NicCtx { _ => continue, }; - Self::apply_public_ipv6_route_changes( - &ifcfg, - &ifname, - &net_ns, - &mut cur_routes, - added, - removed, - ) - .await; + Self::apply_public_ipv6_route_changes(&backend, &mut cur_routes, added, removed) + .await; } });