From f278f8178340fbca4f363a29bf72f4b8f73e30a0 Mon Sep 17 00:00:00 2001 From: longjin Date: Tue, 22 Sep 2026 17:05:42 +0000 Subject: [PATCH 1/2] fix(net): preserve TCP port ownership through TIME_WAIT Separate descriptor bindings from protocol ownership under a namespace-wide port authority. Inherit accepted-child identities independently, enforce both-sided reuse rules, and retain explicit bindings across refused connections. Transfer unreachable TIME_WAIT sockets to compact smoltcp storage without dropping tuple protection. Maintain expiry in direct and NAPI polling and refresh the protocol clock under its locks so idle safe reuse does not depend on unrelated traffic. Pin smoltcp 98c706ee7d5e62c77f74278531402637482a70bc and add 31 mandatory lifecycle regression tests. Validate with make fmt, make kernel, 168 existing guest network cases, 671 dependency tests, dual-stack nginx restart cycles, real 60-second expiry, and 40000 connections in each active-close direction. Signed-off-by: longjin --- kernel/Cargo.lock | 2 +- kernel/Cargo.toml | 2 +- kernel/src/driver/net/iface_common.rs | 6 +- kernel/src/net/socket/inet/common/port.rs | 741 +++++++++++++++++- kernel/src/net/socket/inet/stream/inner.rs | 241 +++++- .../src/net/socket/inet/stream/lifecycle.rs | 16 +- kernel/src/net/socket/inet/stream/option.rs | 8 +- .../src/net/socket/inet/stream/stream_core.rs | 27 +- kernel/src/net/tcp_close_defer.rs | 32 +- kernel/src/process/namespace/net_namespace.rs | 8 +- user/apps/tests/dunitest/no_skip.txt | 1 + .../suites/normal/tcp_port_lifetime.cc | 476 +++++++++++ user/apps/tests/dunitest/whitelist.txt | 1 + 13 files changed, 1446 insertions(+), 115 deletions(-) create mode 100644 user/apps/tests/dunitest/suites/normal/tcp_port_lifetime.cc diff --git a/kernel/Cargo.lock b/kernel/Cargo.lock index 647e8d0be..177345d6a 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=98c706ee7d5e62c77f74278531402637482a70bc#98c706ee7d5e62c77f74278531402637482a70bc" dependencies = [ "bitflags 1.3.2", "byteorder", diff --git a/kernel/Cargo.toml b/kernel/Cargo.toml index 21bf0ea11..46f66cae1 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 = "98c706ee7d5e62c77f74278531402637482a70bc", 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 From 3aedf29b89134bff6951cf63143a0494ffafe3eb Mon Sep 17 00:00:00 2001 From: longjin Date: Tue, 22 Sep 2026 17:10:54 +0000 Subject: [PATCH 2/2] chore(net): pin merged smoltcp TIME_WAIT revision Update the manifest and lockfile to the merge commit of DragonOS-Community/smoltcp#33 on dragonos/v0.12.0. The merged tree is identical to the previously validated dependency revision. Verify the updated dependency with make kernel. Signed-off-by: longjin --- kernel/Cargo.lock | 2 +- kernel/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/kernel/Cargo.lock b/kernel/Cargo.lock index 177345d6a..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=98c706ee7d5e62c77f74278531402637482a70bc#98c706ee7d5e62c77f74278531402637482a70bc" +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 46f66cae1..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 = "98c706ee7d5e62c77f74278531402637482a70bc", 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",