diff --git a/kernel/Cargo.lock b/kernel/Cargo.lock index 647e8d0be..5ad486f04 100644 --- a/kernel/Cargo.lock +++ b/kernel/Cargo.lock @@ -1561,7 +1561,7 @@ checksum = "7fcf8323ef1faaee30a44a340193b1ac6814fd9b7b4e88e9d4519a3e4abe1cfd" [[package]] name = "smoltcp" version = "0.12.0" -source = "git+https://github.com/DragonOS-Community/smoltcp?rev=afde7359455a5b9feacd2b8684994232acbf8903#afde7359455a5b9feacd2b8684994232acbf8903" +source = "git+https://github.com/DragonOS-Community/smoltcp?rev=9b9813e25323fb574f80b8aebfbbf2e7809f47ab#9b9813e25323fb574f80b8aebfbbf2e7809f47ab" dependencies = [ "bitflags 1.3.2", "byteorder", diff --git a/kernel/Cargo.toml b/kernel/Cargo.toml index 21bf0ea11..b9958fda0 100644 --- a/kernel/Cargo.toml +++ b/kernel/Cargo.toml @@ -62,7 +62,7 @@ linkme = "=0.3.27" num = { version = "=0.4.0", default-features = false } num-derive = "=0.3" num-traits = { git = "https://git.mirrors.dragonos.org.cn/DragonOS-Community/num-traits.git", rev = "1597c1c", default-features = false } -smoltcp = { version = "=0.12.0", git = "https://github.com/DragonOS-Community/smoltcp", rev = "afde7359455a5b9feacd2b8684994232acbf8903", default-features = false, features = [ +smoltcp = { version = "=0.12.0", git = "https://github.com/DragonOS-Community/smoltcp", rev = "9b9813e25323fb574f80b8aebfbbf2e7809f47ab", default-features = false, features = [ "alloc", "medium-ethernet", "socket-raw", diff --git a/kernel/src/driver/net/iface_common.rs b/kernel/src/driver/net/iface_common.rs index a60469e28..f5ddc0e92 100644 --- a/kernel/src/driver/net/iface_common.rs +++ b/kernel/src/driver/net/iface_common.rs @@ -533,9 +533,9 @@ impl IfaceCommon { let routed_this_round = route_policy.is_some(); let owner_is_up = scope == IfacePollScope::Full; - let timestamp = crate::time::Instant::now().into(); let mut sockets = self.sockets.lock(); let mut interface = self.smol_iface.lock(); + let timestamp = crate::time::Instant::now().into(); if self.poll_scope() != scope { drop(interface); drop(sockets); @@ -720,9 +720,9 @@ impl IfaceCommon { let routed_this_round = route_policy.is_some(); let owner_is_up = scope == IfacePollScope::Full; - let timestamp = crate::time::Instant::now().into(); let mut sockets = self.sockets.lock(); let mut interface = self.smol_iface.lock(); + let timestamp = crate::time::Instant::now().into(); if self.poll_scope() != scope { drop(interface); drop(sockets); @@ -864,6 +864,8 @@ impl IfaceCommon { let _ = interface.poll_egress(timestamp, device, &mut sockets); } + self.tcp_close_defer.reap_closed(timestamp, &mut sockets); + self.release_resolved_routed_outputs( &mut interface, timestamp, diff --git a/kernel/src/net/socket/inet/common/port.rs b/kernel/src/net/socket/inet/common/port.rs index 815dc6962..ced00231c 100644 --- a/kernel/src/net/socket/inet/common/port.rs +++ b/kernel/src/net/socket/inet/common/port.rs @@ -1,11 +1,14 @@ -use alloc::{sync::Arc, vec::Vec}; +use alloc::{ + sync::{Arc, Weak}, + vec::Vec, +}; use core::sync::atomic::{AtomicU16, AtomicU64, Ordering}; -use hashbrown::HashMap; +use hashbrown::{HashMap, HashSet}; +use smoltcp::socket::tcp::{LifecycleObserver, State}; use smoltcp::wire::{IpAddress, IpEndpoint, IpListenEndpoint, IpVersion}; use system_error::SystemError; use super::device_binding::SocketDeviceBinding; -use crate::process::namespace::net_namespace::NetNamespace; use crate::{arch::rand::rand, libs::mutex::Mutex, process::ProcessManager}; /// A normalized TCP receive domain. Mapped IPv6 addresses are normalized to @@ -52,16 +55,58 @@ impl TcpBindDomain { #[derive(Debug)] struct Binding { - id: u64, - domain: TcpBindDomain, + identity: Weak, + observer: Weak, + observer_generation: u64, + port: u16, + domain: Option, device: Arc, + reuse: bool, + fd_bound: bool, + port_locked: bool, + listening: bool, + parent: Option, + protocol: Option, + tuple: Option<(IpEndpoint, IpEndpoint)>, + protocol_device: u32, + time_wait_reuse: Option, +} + +#[derive(Debug, Default)] +struct Bindings { + owners: HashMap, + ports: HashMap>, + tuples: HashMap<(IpEndpoint, IpEndpoint), Vec>, } -/// Network-namespace TCP reservations, independently of the interface hosting -/// the smoltcp socket. Check and insertion are atomic under one table lock. +impl Bindings { + /// Maintain the lookup index under the same lock as the authoritative + /// identity. A tuple bucket normally has one entry; distinct bound devices + /// may legitimately own the same endpoints. + fn set_tuple(&mut self, id: u64, tuple: Option<(IpEndpoint, IpEndpoint)>) { + let old = self.owners[&id].tuple; + if old == tuple { + return; + } + if let Some(old) = old { + let bucket = self.tuples.get_mut(&old).unwrap(); + bucket.retain(|owner| *owner != id); + if bucket.is_empty() { + self.tuples.remove(&old); + } + } + if let Some(tuple) = tuple { + self.tuples.entry(tuple).or_default().push(id); + } + self.owners.get_mut(&id).unwrap().tuple = tuple; + } +} + +/// Namespace-wide binding and tuple identities. This table never takes socket +/// locks and does not own a namespace, interface, or identity handle. #[derive(Debug)] pub struct PortManager { - bindings: Mutex>>, + bindings: Mutex, next_id: AtomicU64, next_ephemeral: AtomicU16, } @@ -69,7 +114,7 @@ pub struct PortManager { impl Default for PortManager { fn default() -> Self { Self { - bindings: Mutex::new(HashMap::new()), + bindings: Mutex::new(Bindings::default()), next_id: AtomicU64::new(1), next_ephemeral: AtomicU16::new(0), } @@ -78,19 +123,113 @@ impl Default for PortManager { pub const DEFAULT_LOCAL_PORT_RANGE: u32 = (32768u32 << 16) | 60999u32; -/// Unique ownership of a reservation. Accepted children deliberately have no -/// reservation; state transitions move this token instead of copying a port. +#[derive(Debug)] +struct Identity { + manager: Arc, + id: u64, +} + +impl Identity { + fn observer(self: &Arc) -> Arc { + let mut table = self.manager.bindings.lock(); + let binding = table.owners.get_mut(&self.id).unwrap(); + if let Some(observer) = binding.observer.upgrade() { + return observer; + } + binding.observer_generation = binding.observer_generation.wrapping_add(1); + let observer = Arc::new(PortObserver { + identity: self.clone(), + generation: binding.observer_generation, + }); + binding.observer = Arc::downgrade(&observer); + observer + } +} + +impl Drop for Identity { + fn drop(&mut self) { + let mut table = self.manager.bindings.lock(); + table.set_tuple(self.id, None); + if let Some(binding) = table.owners.remove(&self.id) { + if let Some(bucket) = table.ports.get_mut(&binding.port) { + bucket.remove(&self.id); + if bucket.is_empty() { + table.ports.remove(&binding.port); + } + } + } + } +} + +/// Socket option owner, including before bind. Cloning this handle does not +/// acquire another FD binding or protocol lifetime. +#[derive(Debug, Clone)] +pub struct TcpPortOwner(Arc); + +impl TcpPortOwner { + pub fn device_binding(&self) -> Arc { + self.0.manager.bindings.lock().owners[&self.0.id] + .device + .clone() + } + pub fn reuse_addr(&self) -> bool { + self.0.manager.bindings.lock().owners[&self.0.id].reuse + } + + pub fn set_reuse_addr(&self, reuse: bool) { + self.0 + .manager + .bindings + .lock() + .owners + .get_mut(&self.0.id) + .unwrap() + .reuse = reuse; + } + + pub fn reserve( + &self, + domain: TcpBindDomain, + port: u16, + range: (u16, u16), + ) -> Result { + self.0 + .manager + .reserve_owner(self, domain, port, range, None) + } + + /// Implicit connect allocates by full tuple, unlike explicit bind. A + /// same-tuple TIME_WAIT candidate still requires the protocol's safe-reuse + /// transaction; a rejected candidate can be released and allocation retried. + pub fn reserve_connect( + &self, + domain: TcpBindDomain, + remote: IpEndpoint, + range: (u16, u16), + ) -> Result { + self.0 + .manager + .reserve_owner(self, domain, 0, range, Some(remote)) + } +} + +/// The unique FD-side binding reference. Protocol observers independently keep +/// the same identity alive after close; TIME_WAIT can outlive the original FD. #[derive(Debug)] pub struct TcpPortReservation { - netns: Arc, + identity: Arc, pub id: u64, pub port: u16, pub domain: TcpBindDomain, + pub(crate) locked_bind_domain: Option, } impl Drop for TcpPortReservation { fn drop(&mut self) { - self.netns.tcp_ports().unbind(self.port, self.id); + let mut table = self.identity.manager.bindings.lock(); + let binding = table.owners.get_mut(&self.id).unwrap(); + binding.fd_bound = false; + binding.listening = false; } } @@ -98,18 +237,316 @@ impl TcpPortReservation { /// A successful implicit source selection narrows a wildcard reservation /// without allocating a new port or repeating bind-time conflict checks. pub(crate) fn update_domain(&mut self, domain: TcpBindDomain) { - let mut bindings = self.netns.tcp_ports().bindings.lock(); - if let Some(binding) = bindings - .get_mut(&self.port) - .and_then(|bucket| bucket.iter_mut().find(|binding| binding.id == self.id)) + let mut bindings = self.identity.manager.bindings.lock(); + bindings.owners.get_mut(&self.id).unwrap().domain = Some(domain); + self.domain = domain; + } + + pub fn owner(&self) -> TcpPortOwner { + TcpPortOwner(self.identity.clone()) + } + + pub fn lifecycle_observer(&self) -> Arc { + self.identity.observer() + } + + pub fn prepare_child( + &self, + device: Arc, + ) -> Result, SystemError> { + let manager = &self.identity.manager; + let owner = manager.new_owner(device); { - binding.domain = domain; + let mut table = manager.bindings.lock(); + table + .ports + .entry(self.port) + .or_default() + .try_reserve(1) + .map_err(|_| SystemError::ENOMEM)?; + let binding = table.owners.get_mut(&owner.0.id).unwrap(); + binding.port = self.port; + binding.parent = Some(self.id); + table.ports.get_mut(&self.port).unwrap().insert(owner.0.id); + } + Ok(owner.0.observer()) + } + + pub fn promote_listener(&self) -> Result { + let mut table = self.identity.manager.bindings.lock(); + let mine = &table.owners[&self.id]; + if let Some(ids) = table.ports.get(&self.port) { + if ids.iter().any(|id| { + *id != self.id + && bind_conflicts( + mine.domain.unwrap(), + mine.device.ifindex() as u32, + mine.reuse, + &table.owners[id], + ) + }) { + return Err(SystemError::EADDRINUSE); + } + } + let previous = table.owners.get_mut(&self.id).unwrap().listening; + table.owners.get_mut(&self.id).unwrap().listening = true; + Ok(ListenPromotion { + identity: self.identity.clone(), + previous, + committed: false, + }) + } +} + +#[derive(Debug)] +pub struct ListenPromotion { + identity: Arc, + previous: bool, + committed: bool, +} + +impl ListenPromotion { + pub fn commit(mut self) { + self.committed = true; + } +} + +impl Drop for ListenPromotion { + fn drop(&mut self) { + if !self.committed { + self.identity + .manager + .bindings + .lock() + .owners + .get_mut(&self.identity.id) + .unwrap() + .listening = self.previous; + } + } +} + +fn devices_overlap(a: u32, b: u32) -> bool { + a == 0 || b == 0 || a == b +} + +fn bind_conflicts(domain: TcpBindDomain, device: u32, reuse: bool, existing: &Binding) -> bool { + if existing.fd_bound + && devices_overlap(device, existing.device.ifindex() as u32) + && domain.overlaps(existing.domain.unwrap()) + && !(reuse && existing.reuse && !existing.listening) + { + return true; + } + if matches!( + existing.protocol, + Some( + State::Established + | State::FinWait1 + | State::FinWait2 + | State::Closing + | State::CloseWait + | State::LastAck + | State::TimeWait + ) + ) { + if let Some((local, _)) = existing.tuple { + let old_reuse = existing.time_wait_reuse.unwrap_or(existing.reuse); + return devices_overlap(device, existing.protocol_device) + && domain.matches(local.addr) + && !(reuse && old_reuse); + } + } + false +} + +#[derive(Debug)] +struct PortObserver { + identity: Arc, + generation: u64, +} + +impl Drop for PortObserver { + fn drop(&mut self) { + let mut table = self.identity.manager.bindings.lock(); + if let Some(binding) = table.owners.get_mut(&self.identity.id) { + if binding.observer_generation != self.generation { + return; + } + binding.protocol = None; + binding.time_wait_reuse = None; + table.set_tuple(self.identity.id, None); + } + } +} + +impl LifecycleObserver for PortObserver { + fn identity(&self) -> u64 { + self.identity.id + } + + fn prepare_open( + &self, + local: IpEndpoint, + remote: IpEndpoint, + device: u32, + replacing: Option, + ) -> bool { + let mut table = self.identity.manager.bindings.lock(); + if replacing == Some(self.identity.id) { + return false; + } + if let Some(id) = replacing { + let Some(old) = table.owners.get(&id) else { + return false; + }; + if old.protocol != Some(State::TimeWait) + || old.tuple != Some((local, remote)) + || !devices_overlap(device, old.protocol_device) + { + return false; + } + } + if table.tuples.get(&(local, remote)).is_some_and(|ids| { + ids.iter().any(|id| { + if *id == self.identity.id || Some(*id) == replacing { + return false; + } + let other = &table.owners[id]; + devices_overlap(device, other.protocol_device) + }) + }) { + return false; + } + let mine = table.owners.get_mut(&self.identity.id).unwrap(); + if mine.port != local.port || mine.tuple.is_some() { + return false; + } + mine.protocol_device = device; + mine.protocol = Some(if mine.parent.is_some() { + State::SynReceived + } else { + State::SynSent + }); + if mine.parent.is_some() { + mine.domain = Some(TcpBindDomain::new(local.addr, true)); + } + mine.time_wait_reuse = None; + table.set_tuple(self.identity.id, Some((local, remote))); + if let Some(old) = replacing { + table.set_tuple(old, None); + } + true + } + + fn on_state_change( + &self, + state: State, + local: Option, + remote: Option, + device: u32, + ) { + let mut table = self.identity.manager.bindings.lock(); + let old = &table.owners[&self.identity.id]; + let previous = old.protocol; + let inherited = if previous == Some(State::SynReceived) + && matches!(state, State::Established | State::CloseWait) + { + old.parent.and_then(|id| { + table + .owners + .get(&id) + .map(|parent| (parent.reuse, parent.port_locked)) + }) + } else { + None + }; + let mine = table.owners.get_mut(&self.identity.id).unwrap(); + if let Some((reuse, port_locked)) = inherited { + mine.reuse = reuse; + mine.port_locked = port_locked; + } + mine.protocol = Some(state); + mine.protocol_device = device; + // Linux tcp_done releases an automatically selected local port even + // while the descriptor survives. Explicit nonzero bind owns it until + // close; protocol/TIME_WAIT ownership remains independent either way. + if !mine.port_locked + && (state == State::TimeWait + || (state == State::Closed + && previous.is_some_and(|previous| { + !matches!(previous, State::Closed | State::Listen) + }))) + { + mine.fd_bound = false; + } + if matches!(state, State::Closed | State::Listen) { + mine.time_wait_reuse = None; + table.set_tuple(self.identity.id, None); + } else { + if state == State::TimeWait && mine.time_wait_reuse.is_none() { + mine.time_wait_reuse = Some(mine.reuse); + } + if let (Some(local), Some(remote)) = (local, remote) { + table.set_tuple(self.identity.id, Some((local, remote))); + } } - self.domain = domain; } } impl PortManager { + pub fn new_owner(self: &Arc, device: Arc) -> TcpPortOwner { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let identity = Arc::new(Identity { + manager: self.clone(), + id, + }); + self.bindings.lock().owners.insert( + id, + Binding { + identity: Arc::downgrade(&identity), + observer: Weak::new(), + observer_generation: 0, + port: 0, + domain: None, + device, + reuse: false, + fd_bound: false, + port_locked: false, + listening: false, + parent: None, + protocol: None, + tuple: None, + protocol_device: 0, + time_wait_reuse: None, + }, + ); + TcpPortOwner(identity) + } + + pub fn claim_fd(&self, id: u64) -> Result { + let mut table = self.bindings.lock(); + let entry = table.owners.get_mut(&id).ok_or(SystemError::EINVAL)?; + if entry.fd_bound + || entry.parent.is_none() + || !matches!(entry.protocol, Some(State::Established | State::CloseWait)) + { + return Err(SystemError::EINVAL); + } + let identity = entry.identity.upgrade().ok_or(SystemError::EINVAL)?; + entry.device = Arc::new(SocketDeviceBinding::from_ifindex( + entry.protocol_device as usize, + )); + entry.fd_bound = true; + Ok(TcpPortReservation { + identity, + id, + port: entry.port, + domain: entry.domain.unwrap(), + locked_bind_domain: entry.port_locked.then_some(entry.domain.unwrap()), + }) + } + pub fn local_port_range() -> (u16, u16) { ProcessManager::current_netns().local_port_range() } @@ -118,16 +555,25 @@ impl PortManager { ProcessManager::current_netns().set_local_port_range(min, max) } - pub fn reserve( - netns: Arc, + fn reserve_owner( + &self, + owner: &TcpPortOwner, domain: TcpBindDomain, port: u16, - device: Arc, + range: (u16, u16), + connect_remote: Option, ) -> Result { - let manager = netns.tcp_ports(); - let (min, max) = netns.local_port_range(); + let manager = self; + let (min, max) = range; let count = u32::from(max) - u32::from(min) + 1; let mut bindings = manager.bindings.lock(); + let mine = &bindings.owners[&owner.0.id]; + if mine.fd_bound || mine.tuple.is_some() { + return Err(SystemError::EINVAL); + } + let reuse = mine.reuse; + let device = mine.device.ifindex() as u32; + let old_port = mine.port; let initial = manager.next_ephemeral.load(Ordering::Relaxed); let mut candidate = if port != 0 { port @@ -137,15 +583,49 @@ impl PortManager { min + (rand() % count as usize) as u16 }; for _ in 0..if port == 0 { count } else { 1 } { - let bucket = bindings.entry(candidate).or_default(); - if !bucket.iter().any(|binding| { - let a = device.ifindex(); - let b = binding.device.ifindex(); - (a == 0 || b == 0 || a == b) && domain.overlaps(binding.domain) + if !bindings.ports.get(&candidate).is_some_and(|ids| { + ids.iter().any(|id| { + if *id == owner.0.id { + return false; + } + let other = &bindings.owners[id]; + if let Some(remote) = connect_remote { + if other.fd_bound + && devices_overlap(device, other.device.ifindex() as u32) + && domain.overlaps(other.domain.unwrap()) + && (other.listening || other.tuple.is_none()) + { + return true; + } + if let Some((local, peer)) = other.tuple { + return devices_overlap(device, other.protocol_device) + && domain.matches(local.addr) + && peer == remote + && other.protocol != Some(State::TimeWait); + } + false + } else { + bind_conflicts(domain, device, reuse, other) + } + }) }) { + let bucket = bindings.ports.entry(candidate).or_default(); bucket.try_reserve(1).map_err(|_| SystemError::ENOMEM)?; - let id = manager.next_id.fetch_add(1, Ordering::Relaxed); - bucket.push(Binding { id, domain, device }); + let id = owner.0.id; + bucket.insert(id); + if old_port != 0 && old_port != candidate { + if let Some(previous) = bindings.ports.get_mut(&old_port) { + previous.remove(&id); + if previous.is_empty() { + bindings.ports.remove(&old_port); + } + } + } + let binding = bindings.owners.get_mut(&id).unwrap(); + binding.port = candidate; + binding.domain = Some(domain); + binding.fd_bound = true; + binding.port_locked = port != 0 && connect_remote.is_none(); if port == 0 { manager.next_ephemeral.store( if candidate == max { min } else { candidate + 1 }, @@ -154,10 +634,11 @@ impl PortManager { } drop(bindings); return Ok(TcpPortReservation { - netns, + identity: owner.0.clone(), id, port: candidate, domain, + locked_bind_domain: (port != 0 && connect_remote.is_none()).then_some(domain), }); } if port != 0 { @@ -165,16 +646,194 @@ impl PortManager { } candidate = if candidate == max { min } else { candidate + 1 }; } - Err(SystemError::EADDRINUSE) + Err(if connect_remote.is_some() { + SystemError::EADDRNOTAVAIL + } else { + SystemError::EADDRINUSE + }) } +} - fn unbind(&self, port: u16, id: u64) { - let mut bindings = self.bindings.lock(); - if let Some(bucket) = bindings.get_mut(&port) { - bucket.retain(|binding| binding.id != id); - if bucket.is_empty() { - bindings.remove(&port); - } - } +#[cfg(test)] +mod tests { + use super::*; + + fn manager() -> Arc { + Arc::new(PortManager::default()) + } + fn owner(manager: &Arc, reuse: bool) -> TcpPortOwner { + let owner = manager.new_owner(Arc::new(SocketDeviceBinding::default())); + owner.set_reuse_addr(reuse); + owner + } + fn local(port: u16) -> IpEndpoint { + IpEndpoint::new(IpAddress::v4(127, 0, 0, 1), port) + } + fn domain() -> TcpBindDomain { + TcpBindDomain::new(local(0).addr, true) + } + fn reserve(owner: &TcpPortOwner) -> Result { + owner.reserve(domain(), 40000, (40000, 40010)) + } + + #[test] + fn reused_bind_has_atomic_single_listener_promotion() { + let manager = manager(); + let a = reserve(&owner(&manager, true)).unwrap(); + let b = reserve(&owner(&manager, true)).unwrap(); + let promotion = a.promote_listener().unwrap(); + assert!(matches!(b.promote_listener(), Err(SystemError::EADDRINUSE))); + drop(promotion); + b.promote_listener().unwrap().commit(); + assert!(matches!(a.promote_listener(), Err(SystemError::EADDRINUSE))); + } + + #[test] + fn child_inherits_at_handshake_not_accept() { + let manager = manager(); + let parent = owner(&manager, false); + let listener = reserve(&parent).unwrap(); + listener.promote_listener().unwrap().commit(); + let child = listener + .prepare_child(Arc::new(SocketDeviceBinding::default())) + .unwrap(); + assert!(child.prepare_open(local(40000), local(41000), 2, None)); + parent.set_reuse_addr(true); + child.on_state_change( + State::Established, + Some(local(40000)), + Some(local(41000)), + 2, + ); + parent.set_reuse_addr(false); + let accepted = manager.claim_fd(child.identity()).unwrap(); + assert!(accepted.owner().reuse_addr()); + assert_eq!(accepted.owner().device_binding().ifindex(), 2); + drop(listener); + assert!(matches!( + reserve(&owner(&manager, false)), + Err(SystemError::EADDRINUSE) + )); + assert!(reserve(&owner(&manager, true)).is_ok()); + } + + #[test] + fn protocol_close_preserves_fd_bind_and_protocol_drop_preserves_options() { + let manager = manager(); + let original = owner(&manager, false); + let fd = reserve(&original).unwrap(); + let protocol = fd.lifecycle_observer(); + assert!(protocol.prepare_open(local(40000), local(41000), 0, None)); + protocol.on_state_change(State::Closed, None, None, 0); + drop(protocol); + assert!(matches!( + reserve(&owner(&manager, false)), + Err(SystemError::EADDRINUSE) + )); + drop(fd); + assert!(reserve(&owner(&manager, false)).is_ok()); + original.set_reuse_addr(true); + assert!(original.reuse_addr()); + } + + #[test] + fn time_wait_reuse_is_frozen_and_replacement_drop_is_by_identity() { + let manager = manager(); + let original = owner(&manager, false); + let fd = reserve(&original).unwrap(); + let old = fd.lifecycle_observer(); + assert!(old.prepare_open(local(40000), local(41000), 0, None)); + old.on_state_change(State::TimeWait, Some(local(40000)), Some(local(41000)), 0); + original.set_reuse_addr(true); + drop(fd); + assert!(matches!( + reserve(&owner(&manager, true)), + Err(SystemError::EADDRINUSE) + )); + let next_owner = owner(&manager, false); + let next = next_owner + .reserve_connect(domain(), local(41000), (40000, 40000)) + .unwrap(); + let next_protocol = next.lifecycle_observer(); + assert!(next_protocol.prepare_open(local(40000), local(41000), 0, Some(old.identity()))); + drop(old); + assert!(matches!( + owner(&manager, false).reserve_connect(domain(), local(41000), (40000, 40000)), + Err(SystemError::EADDRNOTAVAIL) + )); + } + + #[test] + fn cross_interface_tuple_conflict_uses_device_overlap() { + let manager = manager(); + let first = reserve(&owner(&manager, true)).unwrap(); + let second = reserve(&owner(&manager, true)).unwrap(); + let a = first.lifecycle_observer(); + let b = second.lifecycle_observer(); + assert!(a.prepare_open(local(40000), local(41000), 2, None)); + assert!(!b.prepare_open(local(40000), local(41000), 0, None)); + assert!(b.prepare_open(local(40000), local(41000), 3, None)); + } + + #[test] + fn one_identity_has_one_protocol_observer() { + let manager = manager(); + let fd = reserve(&owner(&manager, false)).unwrap(); + let a = fd.lifecycle_observer(); + let b = fd.lifecycle_observer(); + assert!(a.prepare_open(local(40000), local(41000), 0, None)); + drop(a); + assert!(matches!( + owner(&manager, false).reserve_connect(domain(), local(41000), (40000, 40000)), + Err(SystemError::EADDRNOTAVAIL) + )); + drop(b); + drop(fd); + assert!(reserve(&owner(&manager, false)).is_ok()); + } + + #[test] + fn tuple_index_tracks_device_domains_and_observer_release() { + let manager = manager(); + let first = reserve(&owner(&manager, true)).unwrap(); + let second = reserve(&owner(&manager, true)).unwrap(); + let a = first.lifecycle_observer(); + let b = second.lifecycle_observer(); + let tuple = (local(40000), local(41000)); + assert!(a.prepare_open(tuple.0, tuple.1, 2, None)); + assert!(b.prepare_open(tuple.0, tuple.1, 3, None)); + assert_eq!(manager.bindings.lock().tuples[&tuple].len(), 2); + a.on_state_change(State::Closed, None, None, 2); + assert_eq!(manager.bindings.lock().tuples[&tuple], [b.identity()]); + drop(b); + assert!(!manager.bindings.lock().tuples.contains_key(&tuple)); + drop(a); + drop(first); + drop(second); + let table = manager.bindings.lock(); + assert!(table.ports.is_empty()); + assert!(table.owners.is_empty()); + } + + #[test] + fn retired_time_wait_cannot_remove_replacement_tuple_index() { + let manager = manager(); + let first = reserve(&owner(&manager, true)).unwrap(); + let second = reserve(&owner(&manager, true)).unwrap(); + let old = first.lifecycle_observer(); + let new = second.lifecycle_observer(); + let tuple = (local(40000), local(41000)); + assert!(old.prepare_open(tuple.0, tuple.1, 0, None)); + old.on_state_change(State::TimeWait, Some(tuple.0), Some(tuple.1), 0); + assert!(new.prepare_open(tuple.0, tuple.1, 0, Some(old.identity()))); + old.on_state_change(State::Closed, None, None, 0); + drop(old); + drop(first); + assert_eq!(manager.bindings.lock().tuples[&tuple], [new.identity()]); + drop(new); + drop(second); + let table = manager.bindings.lock(); + assert!(table.tuples.is_empty()); + assert!(table.ports.is_empty()); } } diff --git a/kernel/src/net/socket/inet/stream/inner.rs b/kernel/src/net/socket/inet/stream/inner.rs index 82ec35cab..8af7c05f0 100644 --- a/kernel/src/net/socket/inet/stream/inner.rs +++ b/kernel/src/net/socket/inet/stream/inner.rs @@ -7,7 +7,7 @@ use crate::filesystem::epoll::EPollEventType; use crate::libs::rwsem::RwSem; use crate::net::socket::{ self, - inet::common::port::{PortManager, TcpBindDomain, TcpPortReservation}, + inet::common::port::{TcpBindDomain, TcpPortOwner, TcpPortReservation}, }; use crate::process::namespace::net_namespace::NetNamespace; use alloc::boxed::Box; @@ -48,7 +48,13 @@ fn new_smoltcp_socket_with_size( ) -> smoltcp::socket::tcp::Socket<'static> { let rx_buffer = smoltcp::socket::tcp::SocketBuffer::new(vec![0; rx_size]); let tx_buffer = smoltcp::socket::tcp::SocketBuffer::new(vec![0; tx_size]); - smoltcp::socket::tcp::Socket::new(rx_buffer, tx_buffer) + let mut socket = smoltcp::socket::tcp::Socket::new(rx_buffer, tx_buffer); + socket.set_time_wait_duration(smoltcp::time::Duration::from_secs(60)); + socket.set_tsval_generator(Some(|| { + let now: smoltcp::time::Instant = crate::time::Instant::now().into(); + now.total_millis() as u32 + })); + socket } fn new_smoltcp_socket() -> smoltcp::socket::tcp::Socket<'static> { @@ -59,6 +65,7 @@ fn new_listen_smoltcp_socket( local_endpoint: T, ip_version: Option, device: Option, + reservation: &TcpPortReservation, ) -> Result, SystemError> where T: Into, @@ -66,6 +73,9 @@ where let mut socket = new_smoltcp_socket(); socket.set_listen_ip_version(ip_version); socket.set_listen_bound_device(device); + socket.set_lifecycle_observer(Some(reservation.prepare_child(Arc::new( + SocketDeviceBinding::from_ifindex(device.map_or(0, NonZeroU32::get) as usize), + ))?)); socket.listen(local_endpoint).map_err(|e| match e { tcp::ListenError::InvalidState => SystemError::EINVAL, // TODO: Check is right impl tcp::ListenError::Unaddressable => SystemError::EADDRINUSE, @@ -132,6 +142,7 @@ impl Init { netns: Arc, v6_only: bool, device_binding: Arc, + owner: &TcpPortOwner, ) -> Result { match self { Init::Unbound((mut socket, ver)) => { @@ -147,11 +158,10 @@ impl Init { } }; - let reservation = match PortManager::reserve( - netns, + let reservation = match owner.reserve( TcpBindDomain::new(local_endpoint.addr, v6_only), local_endpoint.port, - device_binding, + netns.local_port_range(), ) { Ok(reservation) => reservation, Err(err) => { @@ -179,6 +189,7 @@ impl Init { remote_endpoint: smoltcp::wire::IpEndpoint, netns: Arc, device_binding: Arc, + owner: &TcpPortOwner, ) -> Result< ( socket::inet::BoundInner, @@ -206,11 +217,10 @@ impl Init { return Err((Self::Unbound((Box::new(socket), ver)), err)) } }; - let reservation = match PortManager::reserve( - netns, + let reservation = match owner.reserve_connect( TcpBindDomain::new(address, false), - 0, - device_binding, + remote_endpoint, + netns.local_port_range(), ) { Ok(reservation) => reservation, Err(err) => { @@ -233,12 +243,61 @@ impl Init { netns: Arc, wrapper: Weak, ver: smoltcp::wire::IpVersion, - device_binding: Arc, + owner: &TcpPortOwner, ) -> Result { - let (mut inner, mut local, mut reservation) = match self { - Init::Unbound(_) => { - self.bind_to_ephemeral(remote_endpoint, netns.clone(), device_binding.clone())? + let automatic = matches!(self, Init::Unbound(_)); + let (low, high) = netns.local_port_range(); + let attempts = if automatic { + usize::from(high - low) + 1 + } else { + 1 + }; + let mut pending = self; + for _ in 0..attempts { + match pending.connect_once( + remote_endpoint, + netns.clone(), + wrapper.clone(), + ver, + owner, + automatic, + ) { + Err((Init::Bound((inner, _, reservation)), SystemError::EADDRNOTAVAIL)) + if automatic => + { + // A candidate may collide with a protected TIME_WAIT after + // reservation. Retry the next ephemeral port without + // changing an explicit user binding or bypassing protection. + let smoltcp::socket::Socket::Tcp(mut socket) = inner.into_socket() else { + unreachable!("TCP connection candidate"); + }; + socket.set_lifecycle_observer(None); + drop(reservation); + pending = Init::Unbound((Box::new(socket), ver)); + } + result => return result, } + } + Err((pending, SystemError::EADDRNOTAVAIL)) + } + + fn connect_once( + self, + remote_endpoint: smoltcp::wire::IpEndpoint, + netns: Arc, + wrapper: Weak, + ver: smoltcp::wire::IpVersion, + owner: &TcpPortOwner, + automatic: bool, + ) -> Result { + let device_binding = owner.device_binding(); + let (mut inner, mut local, mut reservation) = match self { + Init::Unbound(_) => self.bind_to_ephemeral( + remote_endpoint, + netns.clone(), + device_binding.clone(), + owner, + )?, Init::Bound(inner) => inner, }; if let Err(err) = socket::inet::common::ensure_bound_dual_stack_remote_compatible( @@ -275,16 +334,40 @@ impl Init { return Err((Init::Bound((inner, original_local, reservation)), err)); } }; - let result = inner.with_mut::(|socket| { + let result = { + let mut sockets = inner.iface().sockets().lock(); + let socket = sockets.get_mut::(inner.handle()); socket.set_bound_device(NonZeroU32::new(device_binding.ifindex() as u32)); - socket - .connect( - inner.iface().smol_iface().lock().context(), - remote_endpoint, - local, - ) - .map_err(|_| SystemError::ECONNREFUSED) - }); + socket.set_lifecycle_observer(Some(reservation.lifecycle_observer())); + let is_loopback = |addr: smoltcp::wire::IpAddress| match addr { + smoltcp::wire::IpAddress::Ipv4(addr) => addr.is_loopback(), + smoltcp::wire::IpAddress::Ipv6(addr) => addr.is_loopback(), + }; + let loopback = is_loopback(local.addr) + || is_loopback(remote_endpoint.addr) + || netns.loopback_iface().is_some_and(|iface| { + use crate::driver::net::Iface; + device_binding.ifindex() == iface.nic_id() + }); + let policy = if automatic { + tcp::TimeWaitReuse::Automatic { loopback } + } else { + tcp::TimeWaitReuse::Explicit + }; + let mut interface = inner.iface().smol_iface().lock(); + let context = interface.context(); + // A rejected reuse attempt emits no packet and need not cause a + // poll. Its safety deadline must use this transaction's time, not + // the last received packet's time. Sample after both locks. + context.now = crate::time::Instant::now().into(); + sockets + .connect_tcp(inner.handle(), context, remote_endpoint, local, policy) + .map_err(|err| match err { + tcp::ConnectError::AddressInUse => SystemError::EADDRNOTAVAIL, + tcp::ConnectError::InvalidState => SystemError::EINVAL, + tcp::ConnectError::Unaddressable => SystemError::ECONNREFUSED, + }) + }; match result { Ok(_) => { // Narrow only after the last fallible step. Widening on rollback @@ -321,6 +404,7 @@ impl Init { netns: Arc, v6_only: bool, device_binding: Arc, + owner: &TcpPortOwner, ) -> Result { // If unbound, auto-bind to INADDR_ANY:ephemeral (Linux compat). let bound_self = if matches!(self, Init::Unbound(_)) { @@ -337,7 +421,13 @@ impl Init { } }; let auto_bind_ep = smoltcp::wire::IpEndpoint::new(unspec_addr, 0); - match self.bind(auto_bind_ep, netns.clone(), v6_only, device_binding.clone()) { + match self.bind( + auto_bind_ep, + netns.clone(), + v6_only, + device_binding.clone(), + owner, + ) { Ok(bound) => bound, Err((init, err)) => return Err((init, err)), } @@ -357,6 +447,10 @@ impl Init { SystemError::EINVAL, )); } + let promotion = match reservation.promote_listener() { + Ok(promotion) => promotion, + Err(err) => return Err((Init::Bound((inner, local, reservation)), err)), + }; // log::debug!("listen at {:?}, backlog {}", listen_addr, backlog); // // Linux semantics: listen(backlog=0) is valid. In practice it still allows @@ -389,6 +483,7 @@ impl Init { listen_addr, domain.ip_version, NonZeroU32::new(device_binding.ifindex() as u32), + &reservation, )?, iface.clone(), inner.netns(), @@ -406,9 +501,14 @@ impl Init { let primary_index = inners.len(); inners.push(inner); - if let Err(err) = - Listening::grow_slots(&mut inners, backlog, listen_addr, domain, &device_binding) - { + if let Err(err) = Listening::grow_slots( + &mut inners, + backlog, + listen_addr, + domain, + &device_binding, + &reservation, + ) { let inner = inners.remove(primary_index); for bound in inners { bound.release(); @@ -420,6 +520,9 @@ impl Init { inners[primary_index].with_mut::(|socket| { socket.set_listen_ip_version(domain.ip_version); socket.set_listen_bound_device(NonZeroU32::new(device_binding.ifindex() as u32)); + socket.set_lifecycle_observer(Some(reservation.prepare_child(Arc::new( + SocketDeviceBinding::from_ifindex(device_binding.ifindex()), + ))?)); socket.listen(listen_addr).map_err(|err| match err { tcp::ListenError::InvalidState => SystemError::EINVAL, tcp::ListenError::Unaddressable => SystemError::EINVAL, @@ -433,6 +536,7 @@ impl Init { return Err((Init::Bound((inner, local, reservation)), err)); } + promotion.commit(); return Ok(Listening { inners, slots_per_iface: backlog, @@ -550,18 +654,28 @@ impl Connecting { | ConnectResult::RefusedConsumed | ConnectResult::ShutdownReset | ConnectResult::ShutdownResetConsumed => { - drop(self.reservation); - let socket = self.inner.into_socket(); - let socket = match socket { - smoltcp::socket::Socket::Tcp(s) => s, - _ => panic!("Connecting socket is not TCP"), - }; let err = match result { ConnectResult::ShutdownReset | ConnectResult::ShutdownResetConsumed => { SystemError::ECONNRESET } _ => SystemError::ECONNREFUSED, }; + if let Some(domain) = self.reservation.locked_bind_domain { + // Like tcp_disconnect(), restore the user's bind address + // after source selection, but retain a nonzero port lock. + self.reservation.update_domain(domain); + let local = smoltcp::wire::IpEndpoint::new(domain.addr, self.reservation.port); + return ( + Inner::Init(Init::Bound((self.inner, local, self.reservation))), + Err(err), + ); + } + drop(self.reservation); + let socket = self.inner.into_socket(); + let socket = match socket { + smoltcp::socket::Socket::Tcp(s) => s, + _ => panic!("Connecting socket is not TCP"), + }; ( Inner::Init(Init::Unbound((Box::new(socket), self.ver))), Err(err), @@ -941,6 +1055,7 @@ impl Listening { listen_addr: smoltcp::wire::IpListenEndpoint, domain: TcpBindDomain, device_binding: &SocketDeviceBinding, + reservation: &TcpPortReservation, ) -> Result<(), SystemError> { // Prepare sockets before publishing any new handles; keep existing // connections and the previous capacity on a recoverable failure. @@ -962,6 +1077,7 @@ impl Listening { listen_addr, domain.ip_version, NonZeroU32::new(device_binding.ifindex() as u32), + reservation, )?, bound.iface().clone(), bound.netns(), @@ -992,6 +1108,9 @@ impl Listening { self.listen_addr, self.domain, &self.device_binding, + self.reservation + .as_ref() + .expect("open listener reservation"), )?; self.slots_per_iface = target; self.trim_excess(); @@ -1001,13 +1120,13 @@ impl Listening { pub fn accept(&mut self) -> Result<(Established, smoltcp::wire::IpEndpoint), SystemError> { // Resizing can invalidate vector indices. Select the current live slot // under the caller's inner write lock instead of caching a poll index. - let (index, local_endpoint, remote_endpoint) = self + let index = self .inners .iter() .enumerate() .find_map(|(index, bound)| { bound.with::(|socket| { - Self::accept_endpoints(socket).map(|(local, peer)| (index, local, peer)) + Self::accept_endpoints(socket).map(|_| index) }) }) .ok_or(SystemError::EAGAIN_OR_EWOULDBLOCK)?; @@ -1016,9 +1135,15 @@ impl Listening { let connected = &mut self.inners[index]; if retire { + let (reservation, local_endpoint, remote_endpoint) = Self::claim_child(connected)?; let connected = self.inners.remove(index); return Ok(( - Established::with_endpoints(connected, None, local_endpoint, remote_endpoint), + Established::with_endpoints( + connected, + Some(reservation), + local_endpoint, + remote_endpoint, + ), remote_endpoint, )); } @@ -1034,6 +1159,9 @@ impl Listening { self.listen_addr, self.domain.ip_version, NonZeroU32::new(self.device_binding.ifindex() as u32), + self.reservation + .as_ref() + .expect("open listener reservation"), )?, connected.iface().clone(), connected.netns(), @@ -1044,6 +1172,9 @@ impl Listening { self.listen_addr, self.domain.ip_version, NonZeroU32::new(self.device_binding.ifindex() as u32), + self.reservation + .as_ref() + .expect("open listener reservation"), )?, self.listen_addr .addr @@ -1055,16 +1186,48 @@ impl Listening { )? }; + let (reservation, local_endpoint, remote_endpoint) = match Self::claim_child(connected) { + Ok(child) => child, + Err(err) => { + new_listen.release(); + return Err(err); + } + }; + // swap the connected socket with the new_listen socket // TODO is smoltcp socket swappable? core::mem::swap(&mut new_listen, connected); return Ok(( - Established::with_endpoints(new_listen, None, local_endpoint, remote_endpoint), + Established::with_endpoints( + new_listen, + Some(reservation), + local_endpoint, + remote_endpoint, + ), remote_endpoint, )); } + fn claim_child( + bound: &socket::inet::BoundInner, + ) -> Result< + ( + TcpPortReservation, + smoltcp::wire::IpEndpoint, + smoltcp::wire::IpEndpoint, + ), + SystemError, + > { + bound.with::(|socket| { + let (local, remote) = + Self::accept_endpoints(socket).ok_or(SystemError::EAGAIN_OR_EWOULDBLOCK)?; + let observer = socket.lifecycle_observer().ok_or(SystemError::EINVAL)?; + let reservation = bound.netns().tcp_ports().claim_fd(observer.identity())?; + Ok((reservation, local, remote)) + }) + } + pub fn update_io_events(&self, pollee: &AtomicUsize) { // A retained pending child can reset after shrink has completed. It // must become a listen slot again, not permanently consume capacity. @@ -1132,6 +1295,12 @@ pub struct Established { } impl Established { + pub fn port_owner(&self) -> TcpPortOwner { + self.reservation + .as_ref() + .expect("established FD owns its TCP binding") + .owner() + } pub fn new(inner: socket::inet::BoundInner, reservation: Option) -> Self { let local = inner .with::(|socket| socket.local_endpoint()) diff --git a/kernel/src/net/socket/inet/stream/lifecycle.rs b/kernel/src/net/socket/inet/stream/lifecycle.rs index eb7becdfa..98aa08b8d 100644 --- a/kernel/src/net/socket/inet/stream/lifecycle.rs +++ b/kernel/src/net/socket/inet/stream/lifecycle.rs @@ -184,6 +184,7 @@ impl TcpSocket { self.netns(), v6_only, self.device_binding.clone(), + &self.port_owner, ) { Ok(bound) => { // Linux inet6_bind() makes a concrete native IPv6 binding v6-only. @@ -228,6 +229,7 @@ impl TcpSocket { .ipv6_only .load(core::sync::atomic::Ordering::Relaxed), self.device_binding.clone(), + &self.port_owner, ); match listen_result { Ok(listening) => { @@ -381,7 +383,7 @@ impl TcpSocket { self.netns(), self.self_ref.clone(), self.ip_version, - self.device_binding.clone(), + &self.port_owner, ) { Ok(connecting) => ( inner::Inner::Connecting(connecting), @@ -725,13 +727,15 @@ impl TcpSocket { // Ensure we have the latest state from smoltcp let _ = conn.update_io_events(&self.pollee); - if conn.failure_reason().is_some() { + if conn.failure_reason().is_some() || conn.is_refused_consumed() { conn.consume_error(); let (new_inner, _) = conn.into_result(); - writer.replace(new_inner); - } else if conn.is_refused_consumed() { - let (new_inner, _) = conn.into_result(); - writer.replace(new_inner); + let inner::Inner::Init(init) = new_inner else { + unreachable!("failed connection returns an initial socket"); + }; + // A failed explicit connect retains its binding for retry, + // but close must still remove the protocol handle. + writer.replace(inner::Inner::Closed(init.close())); } else { let mut conn = unsafe { conn.into_established_after_unbind() }; let handle = conn.handle(); diff --git a/kernel/src/net/socket/inet/stream/option.rs b/kernel/src/net/socket/inet/stream/option.rs index 7cb728416..0fc9bcc78 100644 --- a/kernel/src/net/socket/inet/stream/option.rs +++ b/kernel/src/net/socket/inet/stream/option.rs @@ -348,7 +348,11 @@ impl super::TcpSocket { self.apply_keepalive(interval); Ok(()) }), - PSO::REUSEADDR => Self::set_bool_option(self.so_reuseaddr(), val, |_| Ok(())), + PSO::REUSEADDR => { + self.port_owner + .set_reuse_addr(byte_parser::read_i32(val)? != 0); + Ok(()) + } PSO::BROADCAST => Self::set_bool_option(self.so_broadcast(), val, |_| Ok(())), PSO::PASSCRED => Self::set_bool_option(self.so_passcred(), val, |_| Ok(())), PSO::NO_CHECK => Self::set_bool_option(self.so_no_check(), val, |_| Ok(())), @@ -616,7 +620,7 @@ impl super::TcpSocket { Self::write_i32_opt(value, v) } PSO::KEEPALIVE => Self::write_bool_opt_i32(value, self.so_keepalive_enabled()), - PSO::REUSEADDR => Self::write_bool_opt_i32(value, self.so_reuseaddr()), + PSO::REUSEADDR => Self::write_i32_opt(value, i32::from(self.port_owner.reuse_addr())), PSO::BROADCAST => Self::write_bool_opt_i32(value, self.so_broadcast()), PSO::PASSCRED => Self::write_bool_opt_i32(value, self.so_passcred()), PSO::NO_CHECK => Self::write_bool_opt_i32(value, self.so_no_check()), diff --git a/kernel/src/net/socket/inet/stream/stream_core.rs b/kernel/src/net/socket/inet/stream/stream_core.rs index 94e1dc78c..2066a5ecd 100644 --- a/kernel/src/net/socket/inet/stream/stream_core.rs +++ b/kernel/src/net/socket/inet/stream/stream_core.rs @@ -8,6 +8,7 @@ use crate::libs::mutex::Mutex; use crate::libs::rwsem::RwSem; use crate::libs::wait_queue::WaitQueue; use crate::net::socket::common::EPollItems; +use crate::net::socket::inet::common::port::TcpPortOwner; use crate::net::socket::inet::common::SocketDeviceBinding; use crate::net::socket::Socket; use crate::process::namespace::net_namespace::NetNamespace; @@ -53,8 +54,6 @@ pub struct TcpSocketOptions { /// SO_ATTACH_FILTER: whether a filter is attached. pub(crate) so_filter_attached: AtomicBool, - /// SO_REUSEADDR - pub(crate) so_reuseaddr: AtomicBool, /// SO_BROADCAST pub(crate) so_broadcast: AtomicBool, /// SO_PASSCRED @@ -111,7 +110,6 @@ impl TcpSocketOptions { tcp_window_clamp: AtomicUsize::new(0), tcp_user_timeout: AtomicI32::new(0), so_filter_attached: AtomicBool::new(false), - so_reuseaddr: AtomicBool::new(false), so_broadcast: AtomicBool::new(false), so_passcred: AtomicBool::new(false), so_no_check: AtomicBool::new(false), @@ -152,6 +150,8 @@ pub struct TcpSocket { pub(crate) pollee: AtomicUsize, pub(crate) netns: Arc, pub(crate) device_binding: Arc, + /// SO_REUSEADDR and bind/protocol ownership share one namespace authority. + pub(crate) port_owner: TcpPortOwner, pub(crate) epoll_items: EPollItems, pub(crate) fasync_items: FAsyncItems, pub(crate) options: TcpSocketOptions, @@ -174,14 +174,13 @@ impl TcpSocket { ip_version: smoltcp::wire::IpVersion, me: &Weak, ) -> Self { - let inherited_device = match &inner { - inner::Inner::Established(est) => est.with(|socket| { - socket - .bound_device() - .map_or(0, |device| device.get() as usize) - }), - _ => 0, + let port_owner = match &inner { + inner::Inner::Established(est) => est.port_owner(), + _ => netns + .tcp_ports() + .new_owner(Arc::new(SocketDeviceBinding::from_ifindex(0))), }; + let device_binding = port_owner.device_binding(); Self { inner: RwSem::new(Some(inner)), shutdown: AtomicUsize::new(0), @@ -194,7 +193,8 @@ impl TcpSocket { self_ref: me.clone(), pollee: AtomicUsize::new(pollee_bits), netns, - device_binding: Arc::new(SocketDeviceBinding::from_ifindex(inherited_device)), + device_binding, + port_owner, epoll_items: EPollItems::default(), fasync_items: FAsyncItems::default(), options: TcpSocketOptions::new(), @@ -352,11 +352,6 @@ impl TcpSocket { &self.options.so_filter_attached } - #[inline] - pub(crate) fn so_reuseaddr(&self) -> &AtomicBool { - &self.options.so_reuseaddr - } - #[inline] pub(crate) fn so_broadcast(&self) -> &AtomicBool { &self.options.so_broadcast diff --git a/kernel/src/net/tcp_close_defer.rs b/kernel/src/net/tcp_close_defer.rs index ac879bca4..2291f3eb8 100644 --- a/kernel/src/net/tcp_close_defer.rs +++ b/kernel/src/net/tcp_close_defer.rs @@ -1,4 +1,4 @@ -//! TCP close(2) 语义辅助：延迟回收 smoltcp TCP socket，贴近 Linux 行为且不修改 smoltcp。 +//! TCP close(2) 语义辅助：在途访问结束后回收完整 socket，保留协议生命周期。 //! //! Linux 语义简述： //! - close(fd) 仅释放文件描述符引用，内核仍会让 TCP 状态机继续运行（发送 FIN/重传/进入 TIME_WAIT 等）； @@ -7,11 +7,11 @@ //! DragonOS/smoltcp 适配点： //! - smoltcp 的 `SocketHandle` 必须留在 `SocketSet` 里才能继续推进状态机； //! - 但 close(fd) 后包裹该 handle 的 `TcpSocket` 可能立刻 drop，因此需要一个 -//! “独立于 TcpSocket 生命周期”的回收队列来保存 handle，等状态到 Closed 再 remove。 +//! 独立回收队列来保存 handle；Closed 后移除，TIME_WAIT 则转交轻量协议表。 use alloc::sync::Weak; use alloc::vec::Vec; -use core::sync::atomic::{AtomicUsize, Ordering}; +use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use crate::libs::mutex::Mutex; use crate::net::socket::inet::InetSocket; @@ -83,6 +83,8 @@ struct ClosingTcpSocket { pub struct TcpCloseDefer { closing: Mutex>, pending: AtomicUsize, + /// Compact TIME_WAIT entries still need namespace-routed replies and timers. + time_wait_pending: AtomicBool, reap_cursor: AtomicUsize, stats: Mutex, } @@ -92,6 +94,7 @@ impl TcpCloseDefer { Self { closing: Mutex::new(Vec::new()), pending: AtomicUsize::new(0), + time_wait_pending: AtomicBool::new(false), reap_cursor: AtomicUsize::new(0), stats: Mutex::new(TcpCloseDeferStats::default()), } @@ -129,7 +132,7 @@ impl TcpCloseDefer { #[inline] pub fn has_pending(&self) -> bool { - self.pending.load(Ordering::Acquire) != 0 + self.pending.load(Ordering::Acquire) != 0 || self.time_wait_pending.load(Ordering::Acquire) } #[inline] @@ -150,6 +153,8 @@ impl TcpCloseDefer { now: smoltcp::time::Instant, sockets: &mut smoltcp::iface::SocketSet<'static>, ) { + self.time_wait_pending + .store(sockets.has_tcp_time_wait(), Ordering::Release); let mut closing = self.closing.lock(); if closing.is_empty() { return; @@ -185,7 +190,11 @@ impl TcpCloseDefer { > 0; if should_abort_post_close_data - || (orphan_timed_out && !matches!(state, smoltcp::socket::tcp::State::Closed)) + || (orphan_timed_out + && !matches!( + state, + smoltcp::socket::tcp::State::Closed | smoltcp::socket::tcp::State::TimeWait + )) { sockets .get_mut::(handle) @@ -234,7 +243,18 @@ impl TcpCloseDefer { i += 1; continue; } - sockets.remove(handle); + if state == smoltcp::socket::tcp::State::TimeWait { + // Move protocol state and its port observer together. Dropping + // the full socket must not release the tuple or its timer. + if !sockets.detach_tcp_time_wait(handle) { + // The final ACK may still be waiting for a TX token. + i += 1; + continue; + } + self.time_wait_pending.store(true, Ordering::Release); + } else { + sockets.remove(handle); + } closing.swap_remove(i); self.pending.fetch_sub(1, Ordering::Release); let mut stats = self.stats.lock(); diff --git a/kernel/src/process/namespace/net_namespace.rs b/kernel/src/process/namespace/net_namespace.rs index 80a7da7aa..3446abe44 100644 --- a/kernel/src/process/namespace/net_namespace.rs +++ b/kernel/src/process/namespace/net_namespace.rs @@ -173,7 +173,7 @@ pub struct NetNamespace { neighbor_table: NeighborTable, /// Per-netns UDP port reservation and local-delivery table. udp_bindings: UdpBindingTable, - tcp_ports: crate::net::socket::inet::common::PortManager, + tcp_ports: Arc, /// Lock-free read-side snapshot for AF_PACKET delivery from NAPI context. packet_sockets: RcuArcSlot, /// Serializes all plain/fanout topology updates and owns group IDs. @@ -564,7 +564,7 @@ impl NetNamespace { teardown_work: NetnsTeardownWork::new(), neighbor_table: NeighborTable::new(), udp_bindings: UdpBindingTable::default(), - tcp_ports: crate::net::socket::inet::common::PortManager::default(), + tcp_ports: Arc::new(crate::net::socket::inet::common::PortManager::default()), packet_sockets: RcuArcSlot::new(Arc::new(PacketSocketRegistrySnapshot::default())), packet_sockets_writer: Mutex::new(PacketSocketRegistryWriter::new()), packet_sockets_need_cleanup: AtomicBool::new(false), @@ -606,7 +606,7 @@ impl NetNamespace { teardown_work: NetnsTeardownWork::new(), neighbor_table: NeighborTable::new(), udp_bindings: UdpBindingTable::default(), - tcp_ports: crate::net::socket::inet::common::PortManager::default(), + tcp_ports: Arc::new(crate::net::socket::inet::common::PortManager::default()), packet_sockets: RcuArcSlot::new(Arc::new(PacketSocketRegistrySnapshot::default())), packet_sockets_writer: Mutex::new(PacketSocketRegistryWriter::new()), packet_sockets_need_cleanup: AtomicBool::new(false), @@ -668,7 +668,7 @@ impl NetNamespace { &self.udp_bindings } - pub(crate) fn tcp_ports(&self) -> &crate::net::socket::inet::common::PortManager { + pub(crate) fn tcp_ports(&self) -> &Arc { &self.tcp_ports } diff --git a/user/apps/tests/dunitest/no_skip.txt b/user/apps/tests/dunitest/no_skip.txt index bbb0fcede..81c3cc385 100644 --- a/user/apps/tests/dunitest/no_skip.txt +++ b/user/apps/tests/dunitest/no_skip.txt @@ -34,3 +34,4 @@ normal/epoll_pwait2_semantics normal/internal_shmem normal/proc_task_status +normal/tcp_port_lifetime diff --git a/user/apps/tests/dunitest/suites/normal/tcp_port_lifetime.cc b/user/apps/tests/dunitest/suites/normal/tcp_port_lifetime.cc new file mode 100644 index 000000000..b23a3d5ec --- /dev/null +++ b/user/apps/tests/dunitest/suites/normal/tcp_port_lifetime.cc @@ -0,0 +1,476 @@ +// Exercise observable TCP bind ownership, independently of TCP_INFO on an old +// descriptor: Linux can detach TIME_WAIT while that descriptor reports CLOSE. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace { +class Fd { + public: + ~Fd() { reset(); } + Fd() = default; + Fd(const Fd&) = delete; + Fd& operator=(const Fd&) = delete; + int get() const { return fd_; } + void reset(int fd = -1) { if (fd_ >= 0) close(fd_); fd_ = fd; } + private: + int fd_ = -1; +}; + +struct Address { + sockaddr_storage storage{}; + socklen_t length = sizeof(storage); + sockaddr* ptr() { return reinterpret_cast(&storage); } +}; + +class TcpPortLifetime : public testing::TestWithParam { + protected: + void Socket(Fd& fd, int reuse) { + fd.reset(socket(GetParam(), SOCK_STREAM, 0)); + ASSERT_GE(fd.get(), 0) << strerror(errno); + if (GetParam() == AF_INET6) { + int one = 1; + ASSERT_EQ(setsockopt(fd.get(), IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one)), 0); + } + ASSERT_EQ(setsockopt(fd.get(), SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)), 0); + timeval timeout{3, 0}; + ASSERT_EQ(setsockopt(fd.get(), SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)), 0); + ASSERT_EQ(setsockopt(fd.get(), SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)), 0); + } + Address Loopback() { + Address a; + if (GetParam() == AF_INET) { + auto* p = reinterpret_cast(&a.storage); + p->sin_family = AF_INET; + p->sin_addr.s_addr = htonl(INADDR_LOOPBACK); + a.length = sizeof(*p); + } else { + auto* p = reinterpret_cast(&a.storage); + p->sin6_family = AF_INET6; + p->sin6_addr = in6addr_loopback; + a.length = sizeof(*p); + } + return a; + } + void BoundAddress(Fd& fd, Address& a) { + a.length = sizeof(a.storage); + ASSERT_EQ(getsockname(fd.get(), a.ptr(), &a.length), 0); + } + uint16_t Port(const Address& address) { + if (GetParam() == AF_INET) { + return ntohs(reinterpret_cast(&address.storage)->sin_port); + } + return ntohs(reinterpret_cast(&address.storage)->sin6_port); + } + void Listener(Fd& fd, int reuse, Address& a) { + ASSERT_NO_FATAL_FAILURE(Socket(fd, reuse)); + a = Loopback(); + ASSERT_EQ(bind(fd.get(), a.ptr(), a.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(fd, a)); + ASSERT_EQ(listen(fd.get(), 4), 0); + } + void Accept(Fd& listener, Fd& child) { + pollfd ready{listener.get(), POLLIN, 0}; + ASSERT_EQ(poll(&ready, 1, 3000), 1); + child.reset(accept(listener.get(), nullptr, nullptr)); + ASSERT_GE(child.get(), 0); + timeval timeout{3, 0}; + ASSERT_EQ(setsockopt(child.get(), SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)), 0); + } + void Exchange(Fd& from, Fd& to) { + const char sent = 'x'; + ASSERT_EQ(send(from.get(), &sent, 1, MSG_NOSIGNAL), 1); + char received = 0; + ASSERT_EQ(recv(to.get(), &received, 1, 0), 1); + ASSERT_EQ(received, sent); + } + void Probe(Address a, int reuse, bool allowed) { + Fd fd; + ASSERT_NO_FATAL_FAILURE(Socket(fd, reuse)); + errno = 0; + int result = bind(fd.get(), a.ptr(), a.length); + int error = errno; + EXPECT_EQ(result, allowed ? 0 : -1) << "reuse=" << reuse << " errno=" << error; + if (!allowed) { + EXPECT_EQ(error, EADDRINUSE); + } + } + void ChildSurvives(int old_reuse) { + Fd listener, client, child; + Address address; + ASSERT_NO_FATAL_FAILURE(Listener(listener, old_reuse, address)); + ASSERT_NO_FATAL_FAILURE(Socket(client, 0)); + ASSERT_EQ(connect(client.get(), address.ptr(), address.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child)); + listener.reset(); + ASSERT_NO_FATAL_FAILURE(Exchange(client, child)); + Probe(address, 0, false); + Probe(address, 1, old_reuse != 0); + } + void TimeWait(int old_reuse) { + Fd listener, client, child; + Address server, local; + ASSERT_NO_FATAL_FAILURE(Listener(listener, 0, server)); + ASSERT_NO_FATAL_FAILURE(Socket(client, old_reuse)); + ASSERT_EQ(connect(client.get(), server.ptr(), server.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child)); + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, local)); + ASSERT_NO_FATAL_FAILURE(Exchange(client, child)); + ASSERT_EQ(shutdown(client.get(), SHUT_WR), 0); + char byte; + ASSERT_EQ(recv(child.get(), &byte, 1, 0), 0); + ASSERT_EQ(shutdown(child.get(), SHUT_WR), 0); + ASSERT_EQ(recv(client.get(), &byte, 1, 0), 0); + // Both FINs have been received; no sleep or TCP_INFO state inference. + Probe(local, 0, false); + Probe(local, 1, old_reuse != 0); + client.reset(); + Probe(local, 0, false); + Probe(local, 1, old_reuse != 0); + } + void Inheritance(int inherited) { + Fd listener, client, child; + Address server; + ASSERT_NO_FATAL_FAILURE(Listener(listener, inherited, server)); + ASSERT_NO_FATAL_FAILURE(Socket(client, 0)); + ASSERT_EQ(connect(client.get(), server.ptr(), server.length), 0); + // Readiness proves that the final ACK has produced an accept-ready child + // before changing the parent, not merely that connect returned locally. + pollfd ready{listener.get(), POLLIN, 0}; + ASSERT_EQ(poll(&ready, 1, 3000), 1); + int changed = !inherited; + ASSERT_EQ(setsockopt(listener.get(), SOL_SOCKET, SO_REUSEADDR, &changed, sizeof(changed)), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child)); + int actual = -1; + socklen_t length = sizeof(actual); + ASSERT_EQ(getsockopt(child.get(), SOL_SOCKET, SO_REUSEADDR, &actual, &length), 0); + EXPECT_EQ(actual, inherited); + listener.reset(); + Probe(server, 1, inherited != 0); + } + void PeerReset(int binding) { + Fd listener, client, child, chooser; + Address remote, local = Loopback(); + ASSERT_NO_FATAL_FAILURE(Listener(listener, 0, remote)); + ASSERT_NO_FATAL_FAILURE(Socket(client, binding == 2 ? 1 : 0)); + if (binding == 2) { + ASSERT_NO_FATAL_FAILURE(Socket(chooser, 1)); + ASSERT_EQ(bind(chooser.get(), local.ptr(), local.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(chooser, local)); + } + if (binding != 0) { + ASSERT_EQ(bind(client.get(), local.ptr(), local.length), 0); + } + chooser.reset(); + ASSERT_EQ(connect(client.get(), remote.ptr(), remote.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child)); + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, local)); + ASSERT_NO_FATAL_FAILURE(Exchange(client, child)); + linger reset{1, 0}; + ASSERT_EQ(setsockopt(child.get(), SOL_SOCKET, SO_LINGER, &reset, sizeof(reset)), 0); + child.reset(); + char byte; + ASSERT_EQ(recv(client.get(), &byte, 1, 0), -1); + ASSERT_EQ(errno, ECONNRESET); + // The original descriptor deliberately remains alive. Protocol CLOSE + // releases auto-selected ports, but not a nonzero user bind lock. + Probe(local, 0, binding != 2); + } +}; + +TEST_P(TcpPortLifetime, ChildWithoutReuseRetainsPort) { ChildSurvives(0); } +TEST_P(TcpPortLifetime, ChildWithReuseRequiresBothSides) { ChildSurvives(1); } +TEST_P(TcpPortLifetime, TimeWaitWithoutReuseSurvivesDescriptorClose) { TimeWait(0); } +TEST_P(TcpPortLifetime, TimeWaitWithReuseRequiresBothSides) { TimeWait(1); } +TEST_P(TcpPortLifetime, ChildDoesNotGainParentLaterReuse) { Inheritance(0); } +TEST_P(TcpPortLifetime, ChildDoesNotLoseParentEarlierReuse) { Inheritance(1); } +TEST_P(TcpPortLifetime, PeerResetWithLiveFdReleasesAutomaticallySelectedPort) { + ASSERT_NO_FATAL_FAILURE(PeerReset(0)); + ASSERT_NO_FATAL_FAILURE(PeerReset(1)); +} +TEST_P(TcpPortLifetime, PeerResetWithLiveFdRetainsExplicitNonzeroPort) { PeerReset(2); } + +TEST_P(TcpPortLifetime, SharedBindListenConflictAndRetry) { + Fd first, second; + ASSERT_NO_FATAL_FAILURE(Socket(first, 1)); + ASSERT_NO_FATAL_FAILURE(Socket(second, 1)); + Address local = Loopback(); + ASSERT_EQ(bind(first.get(), local.ptr(), local.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(first, local)); + ASSERT_EQ(bind(second.get(), local.ptr(), local.length), 0); + ASSERT_EQ(listen(first.get(), 4), 0); + ASSERT_EQ(listen(first.get(), 8), 0); + ASSERT_EQ(listen(second.get(), 4), -1); + ASSERT_EQ(errno, EADDRINUSE); + first.reset(); + ASSERT_EQ(listen(second.get(), 4), 0); +} + +TEST_P(TcpPortLifetime, ChildReuseChangeDoesNotModifyParentOrSibling) { + Fd listener, client1, client2, child1, child2; + Address server; + ASSERT_NO_FATAL_FAILURE(Listener(listener, 1, server)); + ASSERT_NO_FATAL_FAILURE(Socket(client1, 0)); + ASSERT_NO_FATAL_FAILURE(Socket(client2, 0)); + ASSERT_EQ(connect(client1.get(), server.ptr(), server.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child1)); + ASSERT_EQ(connect(client2.get(), server.ptr(), server.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child2)); + int disabled = 0; + ASSERT_EQ(setsockopt(child1.get(), SOL_SOCKET, SO_REUSEADDR, &disabled, sizeof(disabled)), 0); + for (int fd : {listener.get(), child2.get()}) { + int reuse = -1; + socklen_t length = sizeof(reuse); + ASSERT_EQ(getsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, &length), 0); + EXPECT_EQ(reuse, 1); + } + listener.reset(); + // The non-reusing child must still block reuse despite its sibling's flag. + Probe(server, 1, false); +} + +TEST_P(TcpPortLifetime, DistinctRemoteAllowedButDuplicateTupleRejected) { + Fd listener1, listener2, client1, client2, duplicate, child1, child2; + Address server1, server2, local; + ASSERT_NO_FATAL_FAILURE(Listener(listener1, 0, server1)); + ASSERT_NO_FATAL_FAILURE(Listener(listener2, 0, server2)); + ASSERT_NO_FATAL_FAILURE(Socket(client1, 1)); + local = Loopback(); + ASSERT_EQ(bind(client1.get(), local.ptr(), local.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(client1, local)); + ASSERT_EQ(connect(client1.get(), server1.ptr(), server1.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener1, child1)); + ASSERT_NO_FATAL_FAILURE(Socket(client2, 1)); + ASSERT_EQ(bind(client2.get(), local.ptr(), local.length), 0); + ASSERT_EQ(connect(client2.get(), server2.ptr(), server2.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener2, child2)); + ASSERT_NO_FATAL_FAILURE(Socket(duplicate, 1)); + ASSERT_EQ(bind(duplicate.get(), local.ptr(), local.length), 0); + ASSERT_EQ(connect(duplicate.get(), server1.ptr(), server1.length), -1); + ASSERT_EQ(errno, EADDRNOTAVAIL); + ASSERT_NO_FATAL_FAILURE(Exchange(client1, child1)); + ASSERT_NO_FATAL_FAILURE(Exchange(client2, child2)); +} + +TEST_P(TcpPortLifetime, ZeroLingerResetReleasesPort) { + Fd listener, client, child; + Address server, local; + ASSERT_NO_FATAL_FAILURE(Listener(listener, 0, server)); + ASSERT_NO_FATAL_FAILURE(Socket(client, 0)); + ASSERT_EQ(connect(client.get(), server.ptr(), server.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(listener, child)); + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, local)); + ASSERT_NO_FATAL_FAILURE(Exchange(client, child)); + linger reset{1, 0}; + ASSERT_EQ(setsockopt(client.get(), SOL_SOCKET, SO_LINGER, &reset, sizeof(reset)), 0); + client.reset(); + char byte; + ASSERT_EQ(recv(child.get(), &byte, 1, 0), -1); + ASSERT_EQ(errno, ECONNRESET); + Probe(local, 0, true); +} + +TEST_P(TcpPortLifetime, RefusedConnectRetainsExplicitNonzeroPort) { + Fd target, chooser, client, child; + ASSERT_NO_FATAL_FAILURE(Socket(target, 0)); + Address remote = Loopback(); + ASSERT_EQ(bind(target.get(), remote.ptr(), remote.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(target, remote)); + // Reserve an unused local port without a close/rebind race. Both sockets + // permit sharing while bound; the chooser disappears before connect. + ASSERT_NO_FATAL_FAILURE(Socket(chooser, 1)); + Address local = Loopback(); + ASSERT_EQ(bind(chooser.get(), local.ptr(), local.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(chooser, local)); + ASSERT_NO_FATAL_FAILURE(Socket(client, 1)); + ASSERT_EQ(bind(client.get(), local.ptr(), local.length), 0); + chooser.reset(); + ASSERT_EQ(connect(client.get(), remote.ptr(), remote.length), -1); + ASSERT_EQ(errno, ECONNREFUSED); + Address after; + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, after)); + EXPECT_EQ(Port(after), Port(local)); + Probe(local, 0, false); + ASSERT_EQ(listen(target.get(), 1), 0); + ASSERT_EQ(connect(client.get(), remote.ptr(), remote.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(target, child)); + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, after)); + EXPECT_EQ(Port(after), Port(local)); + ASSERT_NO_FATAL_FAILURE(Exchange(client, child)); +} + +TEST_P(TcpPortLifetime, RefusedConnectReleasesBindZeroPort) { + Fd target, client, blocker, child; + ASSERT_NO_FATAL_FAILURE(Socket(target, 0)); + Address remote = Loopback(); + ASSERT_EQ(bind(target.get(), remote.ptr(), remote.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(target, remote)); + ASSERT_NO_FATAL_FAILURE(Socket(client, 0)); + Address local = Loopback(); + ASSERT_EQ(bind(client.get(), local.ptr(), local.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, local)); + ASSERT_EQ(connect(client.get(), remote.ptr(), remote.length), -1); + ASSERT_EQ(errno, ECONNREFUSED); + // Linux does not set SOCK_BINDPORT_LOCK for bind(port=0). The reported + // old local port is not proof of ownership after the failed connection. + ASSERT_NO_FATAL_FAILURE(Socket(blocker, 0)); + ASSERT_EQ(bind(blocker.get(), local.ptr(), local.length), 0); + ASSERT_EQ(listen(target.get(), 1), 0); + ASSERT_EQ(connect(client.get(), remote.ptr(), remote.length), 0); + ASSERT_NO_FATAL_FAILURE(Accept(target, child)); + Address after; + ASSERT_NO_FATAL_FAILURE(BoundAddress(client, after)); + EXPECT_NE(Port(after), Port(local)); + ASSERT_NO_FATAL_FAILURE(Exchange(client, child)); +} + +TEST_P(TcpPortLifetime, CloseUnconsumedNonblockingRefusalReleasesExplicitBind) { + Fd target, chooser; + ASSERT_NO_FATAL_FAILURE(Socket(target, 0)); + Address remote = Loopback(); + ASSERT_EQ(bind(target.get(), remote.ptr(), remote.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(target, remote)); + ASSERT_NO_FATAL_FAILURE(Socket(chooser, 1)); + Address local = Loopback(); + ASSERT_EQ(bind(chooser.get(), local.ptr(), local.length), 0); + ASSERT_NO_FATAL_FAILURE(BoundAddress(chooser, local)); + + for (int iteration = 0; iteration < 16; ++iteration) { + SCOPED_TRACE(iteration); + Fd client; + ASSERT_NO_FATAL_FAILURE(Socket(client, 1)); + ASSERT_EQ(bind(client.get(), local.ptr(), local.length), 0); + chooser.reset(); + int flags = fcntl(client.get(), F_GETFL, 0); + ASSERT_GE(flags, 0); + ASSERT_EQ(fcntl(client.get(), F_SETFL, flags | O_NONBLOCK), 0); + ASSERT_EQ(connect(client.get(), remote.ptr(), remote.length), -1); + ASSERT_EQ(errno, EINPROGRESS); + pollfd ready{client.get(), POLLOUT, 0}; + ASSERT_EQ(poll(&ready, 1, 3000), 1); + ASSERT_NE(ready.revents & POLLERR, 0); + // Do not consume SO_ERROR or call connect/recv again: close must + // dispose of the failed Connecting state and its protocol handle. + client.reset(); + Probe(local, 0, true); + } +} + +void CheckAutomaticReuseWithoutOtherTraffic() { + ASSERT_EQ(unshare(CLONE_NEWNET), 0) << strerror(errno); + Fd listener; + listener.reset(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_GE(listener.get(), 0); + ifreq request{}; + strcpy(request.ifr_name, "lo"); + ASSERT_EQ(ioctl(listener.get(), SIOCGIFFLAGS, &request), 0); + request.ifr_flags |= IFF_UP; + ASSERT_EQ(ioctl(listener.get(), SIOCSIFFLAGS, &request), 0); + sockaddr_in server{}; + server.sin_family = AF_INET; + server.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ASSERT_EQ(bind(listener.get(), reinterpret_cast(&server), sizeof(server)), 0); + socklen_t length = sizeof(server); + ASSERT_EQ(getsockname(listener.get(), reinterpret_cast(&server), &length), 0); + ASSERT_EQ(listen(listener.get(), 4), 0); + + constexpr const char* path = "/proc/sys/net/ipv4/ip_local_port_range"; + char saved[64]{}; + Fd range; + range.reset(open(path, O_RDONLY)); + ASSERT_GE(range.get(), 0); + ssize_t count = read(range.get(), saved, sizeof(saved) - 1); + ASSERT_GT(count, 0); + range.reset(); + const int first = ntohs(server.sin_port) == 48000 || ntohs(server.sin_port) == 48001 + ? 48010 : 48000; + char setting[32]; + const int size = snprintf(setting, sizeof(setting), "%d %d\n", first, first + 1); + range.reset(open(path, O_WRONLY)); + ASSERT_GE(range.get(), 0); + ASSERT_EQ(write(range.get(), setting, size), size); + range.reset(); + + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + for (int connection = 0; connection < 10; ++connection) { + SCOPED_TRACE(connection); + Fd client, accepted; + for (;;) { + client.reset(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_GE(client.get(), 0); + timeval timeout{3, 0}; + ASSERT_EQ(setsockopt(client.get(), SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)), 0); + ASSERT_EQ(setsockopt(client.get(), SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)), 0); + if (connect(client.get(), reinterpret_cast(&server), sizeof(server)) == 0) { + break; + } + const int error = errno; + client.reset(); + ASSERT_EQ(error, EADDRNOTAVAIL); + ASSERT_LT(std::chrono::steady_clock::now(), deadline) + << "TIME_WAIT reuse must progress without an unrelated packet refreshing the clock"; + // This only bounds retry frequency. No payload, diagnostic socket, + // or unrelated TCP traffic may advance the interface poll clock. + ASSERT_EQ(poll(nullptr, 0, 25), 0); + } + sockaddr_in source{}; + length = sizeof(source); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&source), &length), 0); + ASSERT_GE(ntohs(source.sin_port), first); + ASSERT_LE(ntohs(source.sin_port), first + 1); + pollfd ready{listener.get(), POLLIN, 0}; + ASSERT_EQ(poll(&ready, 1, 3000), 1); + accepted.reset(accept(listener.get(), nullptr, nullptr)); + ASSERT_GE(accepted.get(), 0); + timeval timeout{3, 0}; + ASSERT_EQ(setsockopt(accepted.get(), SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)), 0); + ASSERT_EQ(shutdown(client.get(), SHUT_WR), 0); + char byte; + ASSERT_EQ(recv(accepted.get(), &byte, 1, 0), 0); + ASSERT_EQ(shutdown(accepted.get(), SHUT_WR), 0); + ASSERT_EQ(recv(client.get(), &byte, 1, 0), 0); + } + range.reset(open(path, O_WRONLY)); + ASSERT_GE(range.get(), 0); + ASSERT_EQ(write(range.get(), saved, count), count); + // Early assertion failures also exit the child and discard its private + // namespace; this test never modifies the runner's ephemeral port range. +} + +TEST(TcpPortLifetimeNamespace, AutomaticReuseAdvancesWithoutOtherTraffic) { + const pid_t child = fork(); + ASSERT_GE(child, 0); + if (child == 0) { + CheckAutomaticReuseWithoutOtherTraffic(); + const bool failed = testing::Test::HasFailure(); + fflush(nullptr); + _exit(failed ? 1 : 0); + } + int status = 0; + ASSERT_EQ(waitpid(child, &status, 0), child); + ASSERT_TRUE(WIFEXITED(status)); + ASSERT_EQ(WEXITSTATUS(status), 0); +} + +INSTANTIATE_TEST_SUITE_P(Loopback, TcpPortLifetime, testing::Values(AF_INET, AF_INET6), + [](const testing::TestParamInfo& info) { + return info.param == AF_INET ? "IPv4" : "IPv6Only"; + }); +} // namespace + +int main(int argc, char** argv) { + testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/user/apps/tests/dunitest/whitelist.txt b/user/apps/tests/dunitest/whitelist.txt index 4c35c6926..0db5461aa 100644 --- a/user/apps/tests/dunitest/whitelist.txt +++ b/user/apps/tests/dunitest/whitelist.txt @@ -155,3 +155,4 @@ normal/procfs_task_semantics normal/timeval_abi normal/posix_timer_relative normal/alarm_zero +normal/tcp_port_lifetime