diff --git a/kernel/Cargo.lock b/kernel/Cargo.lock index 24babd205..cfbe7af70 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=225c271ef40fa483c60a8f69c8d000d2af58a483#225c271ef40fa483c60a8f69c8d000d2af58a483" +source = "git+https://github.com/DragonOS-Community/smoltcp?rev=9805dae47858c7247303d9d6d19ca504632c1e4a#9805dae47858c7247303d9d6d19ca504632c1e4a" dependencies = [ "bitflags 1.3.2", "byteorder", diff --git a/kernel/Cargo.toml b/kernel/Cargo.toml index ce74f50fc..19848e32d 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 = "225c271ef40fa483c60a8f69c8d000d2af58a483", default-features = false, features = [ +smoltcp = { version = "=0.12.0", git = "https://github.com/DragonOS-Community/smoltcp", rev = "9805dae47858c7247303d9d6d19ca504632c1e4a", 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 c29b8b0e4..71d52f0f9 100644 --- a/kernel/src/driver/net/iface_common.rs +++ b/kernel/src/driver/net/iface_common.rs @@ -145,8 +145,9 @@ impl IfaceCommon { id: u64, domain: crate::net::socket::inet::common::port::TcpBindDomain, port: u16, + device: u32, ) { - self.tcp_listeners.register(id, domain, port); + self.tcp_listeners.register(id, domain, port, device); } /// Unregister an active TCP listener port on this iface. diff --git a/kernel/src/driver/net/local_output.rs b/kernel/src/driver/net/local_output.rs index cc6d73a66..8e87df476 100644 --- a/kernel/src/driver/net/local_output.rs +++ b/kernel/src/driver/net/local_output.rs @@ -82,6 +82,24 @@ pub(super) struct RoutedTxDevice<'a, D: SmolDevice + ?Sized> { pub(super) backend_policy: OutputBackendPolicy<'a>, } +/// Metadata describes the physical ingress, not the TCP SocketSet owner. +pub(super) struct RoutedRxToken { + inner: T, + ifindex: u32, +} + +impl RxToken for RoutedRxToken { + fn consume R>(self, f: F) -> R { + self.inner.consume(f) + } + + fn meta(&self) -> PacketMeta { + let mut meta = self.inner.meta(); + meta.id = self.ifindex; + meta + } +} + /// A physical transmit token with a lazily admitted namespace-routed fallback. /// /// The physical token remains the common path. `LocalInputTxToken` is used @@ -524,7 +542,7 @@ impl SmolDevice for LocalInputDevice<'_, D> { impl SmolDevice for RoutedTxDevice<'_, D> { type RxToken<'a> - = D::RxToken<'a> + = RoutedRxToken> where Self: 'a; type TxToken<'a> @@ -539,7 +557,10 @@ impl SmolDevice for RoutedTxDevice<'_, D> { let capabilities = self.device.capabilities(); let (rx_token, physical) = self.device.receive(timestamp)?; Some(( - rx_token, + RoutedRxToken { + inner: rx_token, + ifindex: self.backend_policy.owner_ifindex, + }, RoutedTxToken { physical: Some(physical), routed: None, diff --git a/kernel/src/net/route/source.rs b/kernel/src/net/route/source.rs index d4d376dce..6ecbce9ae 100644 --- a/kernel/src/net/route/source.rs +++ b/kernel/src/net/route/source.rs @@ -43,6 +43,17 @@ pub(crate) fn resolve_ipv4_route( let devices = netns.device_list(); let router = netns.router(); let fib = router.fib.read(); + // Linux's default route_localnet=0 rejects loopback sources on an + // explicitly selected non-loopback device, including TCP self-connect. + if let (Some(IpAddress::Ipv4(source)), Some(oif)) = (fixed_source, required_oif) { + if source.is_loopback() + && devices + .get(&(oif as usize)) + .is_some_and(|iface| !iface.flags().contains(InterfaceFlags::LOOPBACK)) + { + return Err(SystemError::EINVAL); + } + } // Linux derives an output device from a fixed local source only for // multicast and limited broadcast when no caller supplied an OIF. This is // the ip_route_output_key_hash_rcu() compatibility path that lets an diff --git a/kernel/src/net/socket/inet/common/device_binding.rs b/kernel/src/net/socket/inet/common/device_binding.rs index 3c2d4d2d9..6dcfb6290 100644 --- a/kernel/src/net/socket/inet/common/device_binding.rs +++ b/kernel/src/net/socket/inet/common/device_binding.rs @@ -30,6 +30,15 @@ impl Default for SocketDeviceBinding { } impl SocketDeviceBinding { + /// Construct independent state for a connection inheriting a listener's + /// SYN-time device constraint. Subsequent parent/child updates are separate. + pub(crate) fn from_ifindex(ifindex: usize) -> Self { + Self { + ifindex: AtomicUsize::new(ifindex), + update_lock: Mutex::new(()), + } + } + #[inline] pub fn ifindex(&self) -> usize { self.ifindex.load(Ordering::Acquire) diff --git a/kernel/src/net/socket/inet/common/mod.rs b/kernel/src/net/socket/inet/common/mod.rs index bfa4d526c..2d99f2fc6 100644 --- a/kernel/src/net/socket/inet/common/mod.rs +++ b/kernel/src/net/socket/inet/common/mod.rs @@ -162,7 +162,19 @@ impl BoundInner { where T: smoltcp::socket::AnySocket<'static>, { - let target = match get_ephemeral_bind_target(&remote, netns.clone()) { + Self::bind_ephemeral_recoverable_on_device(socket, remote, netns, None) + } + + pub(crate) fn bind_ephemeral_recoverable_on_device( + socket: T, + remote: smoltcp::wire::IpAddress, + netns: Arc, + device: Option>, + ) -> Result<(Self, smoltcp::wire::IpAddress), (T, SystemError)> + where + T: smoltcp::socket::AnySocket<'static>, + { + let target = match tcp_connect_target(&remote, &netns, device) { Ok(result) => result, Err(err) => return Err((socket, err)), }; @@ -193,6 +205,34 @@ impl BoundInner { &self.iface } + /// Place an inactive TCP endpoint before its first SYN is published. + /// Notification ownership is updated by the caller after releasing inner. + pub(crate) fn move_closed_tcp_to_iface( + &mut self, + iface: Arc, + ) -> Result<(), SystemError> { + if Arc::ptr_eq(&self.iface, &iface) { + return Ok(()); + } + let socket = { + let mut sockets = self.iface.sockets().lock(); + if sockets + .get::(self.handle) + .state() + != smoltcp::socket::tcp::State::Closed + { + return Err(SystemError::EINVAL); + } + sockets.remove(self.handle) + }; + let smoltcp::socket::Socket::Tcp(socket) = socket else { + unreachable!("validated TCP socket"); + }; + self.handle = iface.sockets().lock().add(socket); + self.iface = iface; + Ok(()) + } + pub fn move_udp_to_iface(&mut self, iface: Arc) -> Result<(), SystemError> { self.move_udp_to_iface_with(iface, || {}) } @@ -385,6 +425,17 @@ fn loopback_iface_contains_v6(iface: &Arc, v6_addr: smoltcp::wire::Ip /// Get a suitable iface to deal with sendto/connect request if the socket is not bound to an iface. /// Linux-like behavior: for implicit bind on connect/sendto, the stack must be able to select a /// valid local source address for the given remote destination. +pub(crate) fn tcp_connect_target( + remote: &smoltcp::wire::IpAddress, + netns: &Arc, + device: Option>, +) -> Result { + match device { + Some(iface) => ephemeral_bind_target_on_iface(netns, iface, remote), + None => get_ephemeral_bind_target(remote, netns.clone()), + } +} + fn get_ephemeral_bind_target( remote_ip_addr: &smoltcp::wire::IpAddress, netns: Arc, diff --git a/kernel/src/net/socket/inet/common/port.rs b/kernel/src/net/socket/inet/common/port.rs index aeb35f41d..815dc6962 100644 --- a/kernel/src/net/socket/inet/common/port.rs +++ b/kernel/src/net/socket/inet/common/port.rs @@ -4,6 +4,7 @@ use hashbrown::HashMap; 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}; @@ -53,6 +54,7 @@ impl TcpBindDomain { struct Binding { id: u64, domain: TcpBindDomain, + device: Arc, } /// Network-namespace TCP reservations, independently of the interface hosting @@ -92,6 +94,21 @@ impl Drop for TcpPortReservation { } } +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)) + { + binding.domain = domain; + } + self.domain = domain; + } +} + impl PortManager { pub fn local_port_range() -> (u16, u16) { ProcessManager::current_netns().local_port_range() @@ -105,6 +122,7 @@ impl PortManager { netns: Arc, domain: TcpBindDomain, port: u16, + device: Arc, ) -> Result { let manager = netns.tcp_ports(); let (min, max) = netns.local_port_range(); @@ -120,10 +138,14 @@ impl PortManager { }; for _ in 0..if port == 0 { count } else { 1 } { let bucket = bindings.entry(candidate).or_default(); - if !bucket.iter().any(|binding| domain.overlaps(binding.domain)) { + if !bucket.iter().any(|binding| { + let a = device.ifindex(); + let b = binding.device.ifindex(); + (a == 0 || b == 0 || a == b) && domain.overlaps(binding.domain) + }) { bucket.try_reserve(1).map_err(|_| SystemError::ENOMEM)?; let id = manager.next_id.fetch_add(1, Ordering::Relaxed); - bucket.push(Binding { id, domain }); + bucket.push(Binding { id, domain, device }); if port == 0 { manager.next_ephemeral.store( if candidate == max { min } else { candidate + 1 }, diff --git a/kernel/src/net/socket/inet/stream/events.rs b/kernel/src/net/socket/inet/stream/events.rs index 87c14e2c2..a28beac47 100644 --- a/kernel/src/net/socket/inet/stream/events.rs +++ b/kernel/src/net/socket/inet/stream/events.rs @@ -59,21 +59,6 @@ impl TcpSocket { Some(inner::Inner::Established(established)) => { established.update_io_events(&self.pollee); - // If SHUT_WR was requested while there were pending TX bytes, send FIN once - // the TX queue drains to preserve Linux-like semantics. - if self.is_send_shutdown() - && self - .send_fin_deferred - .load(core::sync::atomic::Ordering::Relaxed) - { - let pending = established.with(|socket| socket.send_queue()); - if pending == 0 { - established.with_mut(|socket| socket.close()); - self.send_fin_deferred - .store(false, core::sync::atomic::Ordering::Relaxed); - } - } - // If SHUT_WR, set EPOLLOUT so send() wakes up and returns EPIPE. if self.is_send_shutdown() { self.pollee.fetch_or( @@ -81,36 +66,19 @@ impl TcpSocket { core::sync::atomic::Ordering::Relaxed, ); } - // If SHUT_RD, set EPOLLIN so recv() wakes up and returns 0 (EOF). + // Linux tcp_poll combines local shutdown with transport state: + // receive shutdown is a half-close even before a peer FIN. if self.is_recv_shutdown() { self.pollee.fetch_or( - (EP::EPOLLIN | EP::EPOLLRDNORM).bits() as usize, - core::sync::atomic::Ordering::Relaxed, - ); - } - - // Note: EPOLLHUP/EPOLLRDHUP/EPOLLERR are now handled in - // Established::update_io_events() based on socket state. - false - } - Some(inner::Inner::SelfConnected(sc)) => { - // Self-connect is modeled by an internal receive queue. Readable becomes true - // when the queue has data OR after SHUT_WR (EOF). Writable depends on queue - // free space unless SHUT_WR (then send() returns EPIPE). - sc.update_io_events(&self.pollee, self.is_recv_shutdown()); - - // Match established behavior for shutdown bits. - if self.is_send_shutdown() { - self.pollee.fetch_or( - (EP::EPOLLOUT | EP::EPOLLWRNORM).bits() as usize, - core::sync::atomic::Ordering::Relaxed, - ); - } - if self.is_recv_shutdown() { - self.pollee.fetch_or( - (EP::EPOLLIN | EP::EPOLLRDNORM).bits() as usize, + (EP::EPOLLIN | EP::EPOLLRDNORM | EP::EPOLLRDHUP).bits() as usize, core::sync::atomic::Ordering::Relaxed, ); + if self.is_send_shutdown() { + self.pollee.fetch_or( + EP::EPOLLHUP.bits() as usize, + core::sync::atomic::Ordering::Relaxed, + ); + } } false } diff --git a/kernel/src/net/socket/inet/stream/inner.rs b/kernel/src/net/socket/inet/stream/inner.rs index 7fbc2314d..82ec35cab 100644 --- a/kernel/src/net/socket/inet/stream/inner.rs +++ b/kernel/src/net/socket/inet/stream/inner.rs @@ -1,16 +1,15 @@ -use alloc::collections::VecDeque; +use crate::net::socket::inet::common::SocketDeviceBinding; use alloc::sync::{Arc, Weak}; +use core::num::NonZeroU32; use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use crate::filesystem::epoll::EPollEventType; -use crate::libs::mutex::Mutex; use crate::libs::rwsem::RwSem; use crate::net::socket::{ self, inet::common::port::{PortManager, TcpBindDomain, TcpPortReservation}, }; use crate::process::namespace::net_namespace::NetNamespace; -use crate::syscall::user_buffer::UserBuffer; use alloc::boxed::Box; use alloc::vec::Vec; use smoltcp; @@ -59,12 +58,14 @@ fn new_smoltcp_socket() -> smoltcp::socket::tcp::Socket<'static> { fn new_listen_smoltcp_socket( local_endpoint: T, ip_version: Option, + device: Option, ) -> Result, SystemError> where T: Into, { let mut socket = new_smoltcp_socket(); socket.set_listen_ip_version(ip_version); + socket.set_listen_bound_device(device); socket.listen(local_endpoint).map_err(|e| match e { tcp::ListenError::InvalidState => SystemError::EINVAL, // TODO: Check is right impl tcp::ListenError::Unaddressable => SystemError::EADDRINUSE, @@ -110,6 +111,7 @@ impl Init { new_sock.set_timeout(socket.timeout()); new_sock.set_hop_limit(socket.hop_limit()); new_sock.set_listen_ip_version(socket.listen_ip_version()); + new_sock.set_bound_device(socket.bound_device()); **socket = new_sock; Ok(()) @@ -129,9 +131,11 @@ impl Init { local_endpoint: smoltcp::wire::IpEndpoint, netns: Arc, v6_only: bool, + device_binding: Arc, ) -> Result { match self { - Init::Unbound((socket, ver)) => { + Init::Unbound((mut socket, ver)) => { + socket.set_bound_device(NonZeroU32::new(device_binding.ifindex() as u32)); let bound = match socket::inet::BoundInner::bind_recoverable( *socket, &local_endpoint.addr, @@ -147,6 +151,7 @@ impl Init { netns, TcpBindDomain::new(local_endpoint.addr, v6_only), local_endpoint.port, + device_binding, ) { Ok(reservation) => reservation, Err(err) => { @@ -173,6 +178,7 @@ impl Init { self, remote_endpoint: smoltcp::wire::IpEndpoint, netns: Arc, + device_binding: Arc, ) -> Result< ( socket::inet::BoundInner, @@ -182,27 +188,38 @@ impl Init { (Self, SystemError), > { match self { - Init::Unbound((socket, ver)) => { - let (bound, address) = match socket::inet::BoundInner::bind_ephemeral_recoverable( - *socket, - remote_endpoint.addr, - netns.clone(), - ) { - Ok(result) => result, - Err((socket, err)) => { - return Err((Self::Unbound((Box::new(socket), ver)), err)) - } + Init::Unbound((mut socket, ver)) => { + let device = match device_binding.resolve_iface(&netns) { + Ok(device) => device, + Err(err) => return Err((Self::Unbound((socket, ver)), err)), }; - let reservation = - match PortManager::reserve(netns, TcpBindDomain::new(address, false), 0) { - Ok(reservation) => reservation, - Err(err) => { - let smoltcp::socket::Socket::Tcp(socket) = bound.into_socket() else { - unreachable!("TCP BoundInner should contain a TCP socket"); - }; - return Err((Self::Unbound((Box::new(socket), ver)), err)); + socket.set_bound_device(NonZeroU32::new(device_binding.ifindex() as u32)); + let (bound, address) = + match socket::inet::BoundInner::bind_ephemeral_recoverable_on_device( + *socket, + remote_endpoint.addr, + netns.clone(), + device, + ) { + Ok(result) => result, + Err((socket, err)) => { + return Err((Self::Unbound((Box::new(socket), ver)), err)) } }; + let reservation = match PortManager::reserve( + netns, + TcpBindDomain::new(address, false), + 0, + device_binding, + ) { + Ok(reservation) => reservation, + Err(err) => { + let smoltcp::socket::Socket::Tcp(socket) = bound.into_socket() else { + unreachable!("TCP BoundInner should contain a TCP socket"); + }; + return Err((Self::Unbound((Box::new(socket), ver)), err)); + } + }; let endpoint = smoltcp::wire::IpEndpoint::new(address, reservation.port); Ok((bound, endpoint, reservation)) } @@ -216,9 +233,12 @@ impl Init { netns: Arc, wrapper: Weak, ver: smoltcp::wire::IpVersion, + device_binding: Arc, ) -> Result { - let (inner, local, reservation) = match self { - Init::Unbound(_) => self.bind_to_ephemeral(remote_endpoint, netns)?, + let (mut inner, mut local, mut reservation) = match self { + Init::Unbound(_) => { + self.bind_to_ephemeral(remote_endpoint, netns.clone(), device_binding.clone())? + } Init::Bound(inner) => inner, }; if let Err(err) = socket::inet::common::ensure_bound_dual_stack_remote_compatible( @@ -227,11 +247,20 @@ impl Init { ) { return Err((Init::Bound((inner, local, reservation)), err)); } + let original_local = local; + let original_iface = inner.iface().clone(); if local.addr.is_unspecified() { - return Err(( - Init::Bound((inner, local, reservation)), - SystemError::EINVAL, - )); + let target = device_binding.resolve_iface(&netns).and_then(|device| { + socket::inet::common::tcp_connect_target(&remote_endpoint.addr, &netns, device) + }); + let target = match target { + Ok(target) => target, + Err(err) => return Err((Init::Bound((inner, local, reservation)), err)), + }; + if let Err(err) = inner.move_closed_tcp_to_iface(target.stack_owner) { + return Err((Init::Bound((inner, local, reservation)), err)); + } + local.addr = target.local_addr; } // Publish before taking the SocketSet lock and making the first SYN // visible to poll. The RAII reservation survives until Connecting is @@ -239,9 +268,15 @@ impl Init { // socket lifetime. let registration = match ConnectingRegistration::try_new(inner.iface().clone(), wrapper) { Ok(registration) => registration, - Err(err) => return Err((Init::Bound((inner, local, reservation)), err)), + Err(err) => { + inner + .move_closed_tcp_to_iface(original_iface) + .expect("inactive TCP placement rollback"); + return Err((Init::Bound((inner, original_local, reservation)), err)); + } }; let result = inner.with_mut::(|socket| { + socket.set_bound_device(NonZeroU32::new(device_binding.ifindex() as u32)); socket .connect( inner.iface().smol_iface().lock().context(), @@ -251,15 +286,27 @@ impl Init { .map_err(|_| SystemError::ECONNREFUSED) }); match result { - Ok(_) => Ok(Connecting::new( - inner, - registration, - local, - remote_endpoint, - reservation, - ver, - )), - Err(err) => Err((Init::Bound((inner, local, reservation)), err)), + Ok(_) => { + // Narrow only after the last fallible step. Widening on rollback + // could overlap a bind admitted while the domain was narrower. + if original_local.addr.is_unspecified() { + reservation.update_domain(TcpBindDomain::new(local.addr, false)); + } + Ok(Connecting::new( + inner, + registration, + local, + remote_endpoint, + reservation, + ver, + )) + } + Err(err) => { + inner + .move_closed_tcp_to_iface(original_iface) + .expect("failed connect leaves TCP closed"); + Err((Init::Bound((inner, original_local, reservation)), err)) + } } } @@ -273,6 +320,7 @@ impl Init { backlog: usize, netns: Arc, v6_only: bool, + device_binding: Arc, ) -> Result { // If unbound, auto-bind to INADDR_ANY:ephemeral (Linux compat). let bound_self = if matches!(self, Init::Unbound(_)) { @@ -289,7 +337,7 @@ impl Init { } }; let auto_bind_ep = smoltcp::wire::IpEndpoint::new(unspec_addr, 0); - match self.bind(auto_bind_ep, netns.clone(), v6_only) { + match self.bind(auto_bind_ep, netns.clone(), v6_only, device_binding.clone()) { Ok(bound) => bound, Err((init, err)) => return Err((init, err)), } @@ -337,7 +385,11 @@ impl Init { continue; // primary inner already covers this iface } let new_listen = socket::inet::BoundInner::bind_on_iface( - new_listen_smoltcp_socket(listen_addr, domain.ip_version)?, + new_listen_smoltcp_socket( + listen_addr, + domain.ip_version, + NonZeroU32::new(device_binding.ifindex() as u32), + )?, iface.clone(), inner.netns(), )?; @@ -354,7 +406,9 @@ impl Init { let primary_index = inners.len(); inners.push(inner); - if let Err(err) = Listening::grow_slots(&mut inners, backlog, listen_addr, domain) { + if let Err(err) = + Listening::grow_slots(&mut inners, backlog, listen_addr, domain, &device_binding) + { let inner = inners.remove(primary_index); for bound in inners { bound.release(); @@ -365,6 +419,7 @@ impl Init { if let Err(err) = 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.listen(listen_addr).map_err(|err| match err { tcp::ListenError::InvalidState => SystemError::EINVAL, tcp::ListenError::Unaddressable => SystemError::EINVAL, @@ -386,6 +441,7 @@ impl Init { local, domain, reservation: Some(reservation), + device_binding, }); } @@ -765,9 +821,39 @@ pub struct Listening { local: smoltcp::wire::IpEndpoint, pub domain: TcpBindDomain, pub reservation: Option, + device_binding: Arc, } impl Listening { + /// Update idle slots and the overflow lookup under the same SocketSet lock. + /// Handshake/accepted snapshots remain owned by each TCP socket. + pub(super) fn set_bound_device(&mut self, device: Option) { + for (index, bound) in self.inners.iter().enumerate() { + if self.inners[..index] + .iter() + .any(|other| Arc::ptr_eq(other.iface(), bound.iface())) + { + continue; + } + let mut sockets = bound.iface().sockets().lock(); + for slot in &self.inners { + if Arc::ptr_eq(slot.iface(), bound.iface()) { + sockets + .get_mut::(slot.handle()) + .set_listen_bound_device(device); + } + } + if let Some(reservation) = &self.reservation { + bound.iface().common().register_tcp_listener( + reservation.id, + self.domain, + self.local.port, + device.map_or(0, NonZeroU32::get), + ); + } + } + } + /// Ordinary passive opens become acceptable only after the final ACK. /// A peer may already have sent FIN, so CLOSE_WAIT remains acceptable. /// Snapshot both endpoints under the caller's SocketSet lock: a later RST @@ -838,6 +924,9 @@ impl Listening { for bound in &self.inners { bound.with_mut::(|socket| { if socket.state() == tcp::State::Closed { + socket.set_listen_bound_device(NonZeroU32::new( + self.device_binding.ifindex() as u32 + )); socket .listen(self.listen_addr) .expect("valid listener endpoint"); @@ -851,6 +940,7 @@ impl Listening { target: usize, listen_addr: smoltcp::wire::IpListenEndpoint, domain: TcpBindDomain, + device_binding: &SocketDeviceBinding, ) -> Result<(), SystemError> { // Prepare sockets before publishing any new handles; keep existing // connections and the previous capacity on a recoverable failure. @@ -868,7 +958,11 @@ impl Listening { .count(); for _ in count..target { sockets.push(( - new_listen_smoltcp_socket(listen_addr, domain.ip_version)?, + new_listen_smoltcp_socket( + listen_addr, + domain.ip_version, + NonZeroU32::new(device_binding.ifindex() as u32), + )?, bound.iface().clone(), bound.netns(), )); @@ -892,7 +986,13 @@ impl Listening { pub(super) fn set_backlog(&mut self, backlog: usize) -> Result<(), SystemError> { let target = Self::slot_capacity(backlog); - Self::grow_slots(&mut self.inners, target, self.listen_addr, self.domain)?; + Self::grow_slots( + &mut self.inners, + target, + self.listen_addr, + self.domain, + &self.device_binding, + )?; self.slots_per_iface = target; self.trim_excess(); Ok(()) @@ -930,13 +1030,21 @@ impl Listening { // where each interface has its own listen socket in the smoltcp SocketSet. let mut new_listen = if self.listen_addr.addr.is_none() { socket::inet::BoundInner::bind_on_iface( - new_listen_smoltcp_socket(self.listen_addr, self.domain.ip_version)?, + new_listen_smoltcp_socket( + self.listen_addr, + self.domain.ip_version, + NonZeroU32::new(self.device_binding.ifindex() as u32), + )?, connected.iface().clone(), connected.netns(), )? } else { socket::inet::BoundInner::bind( - new_listen_smoltcp_socket(self.listen_addr, self.domain.ip_version)?, + new_listen_smoltcp_socket( + self.listen_addr, + self.domain.ip_version, + NonZeroU32::new(self.device_binding.ifindex() as u32), + )?, self.listen_addr .addr .as_ref() @@ -1221,278 +1329,12 @@ impl Established { } } -/// Linux-compatible TCP "self-connect" (connect to the same local addr:port on the same socket). -/// -/// Linux allows this with a single socket FD, and bytes written to the socket are readable -/// back from the same socket. smoltcp's TCP socket cannot model this with a single instance, -/// because a TCP endpoint should not receive its own outbound segments. -/// -/// We implement the user-visible semantics by internally queueing sent bytes into a local -/// receive queue, and driving readiness/EOF based on shutdown state. -#[derive(Debug)] -pub struct SelfConnected { - inner: socket::inet::BoundInner, - reservation: Option, - local: smoltcp::wire::IpEndpoint, - state: Mutex, -} - -#[derive(Debug)] -struct SelfConnectedState { - /// Effective receive capacity for the loopback queue (bytes). - rx_cap: usize, - buf: VecDeque, - /// Ordered EOF marker for the self-connected byte stream. - /// - /// This must be protected by the same lock as `buf`: a concurrent - /// `send()` and `shutdown(SHUT_WR)` must linearize as either - /// data-before-FIN or FIN-before-send, never as an independently visible EOF. - send_shutdown: bool, -} - -impl SelfConnected { - pub fn new( - inner: socket::inet::BoundInner, - local: smoltcp::wire::IpEndpoint, - rx_cap: usize, - reservation: TcpPortReservation, - ) -> Self { - Self { - inner, - local, - reservation: Some(reservation), - state: Mutex::new(SelfConnectedState { - rx_cap, - buf: VecDeque::new(), - send_shutdown: false, - }), - } - } - - #[allow(dead_code)] - pub fn iface(&self) -> &Arc { - self.inner.iface() - } - - #[allow(dead_code)] - pub fn handle(&self) -> smoltcp::iface::SocketHandle { - self.inner.handle() - } - - #[inline] - pub fn get_name(&self) -> smoltcp::wire::IpEndpoint { - self.local - } - - #[inline] - pub fn get_peer_name(&self) -> smoltcp::wire::IpEndpoint { - self.local - } - - #[inline] - pub fn recv_queue(&self) -> usize { - self.state.lock().buf.len() - } - - pub fn discard_all(&self) { - self.state.lock().buf.clear(); - } - - pub fn set_recv_buffer_size(&self, rx_size: usize) { - self.state.lock().rx_cap = rx_size; - } - - pub fn recv_capacity(&self) -> usize { - self.state.lock().rx_cap - } - - pub fn send_capacity(&self) -> usize { - // For self-connect, use the same capacity for "send" as the local receive queue. - self.state.lock().rx_cap - } - - pub fn send_slice(&self, data: &[u8], send_shutdown: bool) -> Result { - if send_shutdown { - return Err(SystemError::EPIPE); - } - if data.is_empty() { - return Ok(0); - } - - let mut state = self.state.lock(); - if state.send_shutdown { - return Err(SystemError::EPIPE); - } - - let free = state.rx_cap.saturating_sub(state.buf.len()); - if free == 0 { - return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); - } - let n = core::cmp::min(free, data.len()); - state.buf.extend(data[..n].iter().copied()); - Ok(n) - } - - /// Set the send_shutdown flag (called when SHUT_WR is performed). - pub fn set_send_shutdown(&self) { - self.state.lock().send_shutdown = true; - } - - /// Check if send_shutdown flag is set. - #[inline] - #[allow(dead_code)] - pub fn is_send_shutdown(&self) -> bool { - self.state.lock().send_shutdown - } - - pub fn recv_into(&self, out: &mut [u8], peek: bool, trunc: bool) -> Result { - if out.is_empty() { - return Ok(0); - } - let mut state = self.state.lock(); - if state.buf.is_empty() { - // EOF after SHUT_WR once all queued data is drained. - if state.send_shutdown { - return Ok(0); - } - return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); - } - - let n = core::cmp::min(out.len(), state.buf.len()); - if !trunc { - for (i, b) in state.buf.iter().take(n).enumerate() { - out[i] = *b; - } - } - - if !peek { - for _ in 0..n { - let _ = state.buf.pop_front(); - } - } - Ok(n) - } - - pub fn recv_to_user( - &self, - out: &mut UserBuffer<'_>, - offset: usize, - max_len: usize, - ) -> Result { - if offset > out.len() { - return Err(SystemError::EINVAL); - } - let available = core::cmp::min(out.len() - offset, max_len); - if available == 0 { - return Ok(0); - } - - let mut state = self.state.lock(); - if state.buf.is_empty() { - if state.send_shutdown { - return Ok(0); - } - return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); - } - - let n = core::cmp::min(available, state.buf.len()); - let mut tmp = Vec::with_capacity(n); - tmp.extend(state.buf.iter().take(n).copied()); - - match out.write_to_user(offset, &tmp) { - Ok(_) => { - for _ in 0..n { - let _ = state.buf.pop_front(); - } - Ok(n) - } - Err(SystemError::EFAULT) => Err(SystemError::EFAULT), - Err(e) => Err(e), - } - } - - /// 重算 self-connect 套接字的就绪掩码。 - /// - /// `recv_shutdown` 是外层 `TcpSocket` 记录的 `SHUT_RD` 状态（内层只跟踪发送侧）。 - /// - /// Linux 6.6 `tcp_poll()` 的挂断语义： - /// ```c - /// if (shutdown == SHUTDOWN_MASK || state == TCP_CLOSE) - /// mask |= EPOLLHUP; - /// if (shutdown & RCV_SHUTDOWN) - /// mask |= EPOLLIN | EPOLLRDNORM | EPOLLRDHUP; - /// ``` - /// self-connect 发出的 FIN 会回到自己（`tcp_fin()` 收到 FIN 时置 `RCV_SHUTDOWN`）， - /// 因此 `SHUT_WR` 之后本端的 `sk_shutdown` 已经是 `SHUTDOWN_MASK`：读侧同时进入 EOF， - /// `poll()` 应报 `EPOLLRDHUP | EPOLLHUP`（Linux 实测为 `IN|OUT|HUP|RDHUP`）。 - /// 只有 `SHUT_RD` 时是半关闭，只报 `EPOLLRDHUP`，不得报 `EPOLLHUP`。 - /// 连接仍完全建立时必须清掉历史残留：套接字在 `bind()` 后即被 iface 通知， - /// `Init` 分支会打上 `EPOLLHUP`，不清理会让 `poll`/`epoll` 永久误报挂断。 - /// `EPOLLERR` 在 self-connect 上没有产生路径，只清不置。 - pub fn update_io_events(&self, pollee: &AtomicUsize, recv_shutdown: bool) { - let state = self.state.lock(); - let send_shutdown = state.send_shutdown; - let writable = !send_shutdown && state.buf.len() < state.rx_cap; - let readable = !state.buf.is_empty() || send_shutdown; - drop(state); - - let read_shutdown = send_shutdown || recv_shutdown; - - let hangup_bits = - (EPollEventType::EPOLLHUP | EPollEventType::EPOLLRDHUP | EPollEventType::EPOLLERR) - .bits() as usize; - let mut rebuilt = 0usize; - if read_shutdown { - rebuilt |= EPollEventType::EPOLLRDHUP.bits() as usize; - } - if send_shutdown { - // 自身 FIN 回环到读侧，`SHUT_WR` 之后 sk_shutdown 已是 SHUTDOWN_MASK。 - rebuilt |= EPollEventType::EPOLLHUP.bits() as usize; - } - - // 一次性清除并重建，避免清除与置位之间被并发观察到一个不存在的中间态。 - let _ = pollee.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |bits| { - Some((bits & !hangup_bits) | rebuilt) - }); - - if writable { - pollee.fetch_or( - (EPollEventType::EPOLLOUT | EPollEventType::EPOLLWRNORM).bits() as usize, - Ordering::Relaxed, - ); - } else { - pollee.fetch_and( - !(EPollEventType::EPOLLOUT | EPollEventType::EPOLLWRNORM).bits() as usize, - Ordering::Relaxed, - ); - } - - if readable { - pollee.fetch_or( - (EPollEventType::EPOLLIN | EPollEventType::EPOLLRDNORM).bits() as usize, - Ordering::Relaxed, - ); - } else { - pollee.fetch_and( - !(EPollEventType::EPOLLIN | EPollEventType::EPOLLRDNORM).bits() as usize, - Ordering::Relaxed, - ); - } - } - - pub fn release(&mut self) { - self.inner.release(); - self.reservation.take(); - } -} - #[derive(Debug)] pub enum Inner { Init(Init), Connecting(Connecting), Listening(Listening), Established(Established), - SelfConnected(SelfConnected), Closed(Closed), } @@ -1509,11 +1351,6 @@ impl Inner { Inner::Connecting(conn) => conn.with(f), Inner::Listening(listen) => listen.inners[0].with(f), Inner::Established(est) => est.with(f), - Inner::SelfConnected(_) => { - // SelfConnected keeps a BoundInner for resource management, but does not - // model its data path via smoltcp. Avoid touching the underlying socket. - panic!("Inner::with_socket called on SelfConnected socket") - } Inner::Closed(_) => { // Closed 状态不应再触达任何 smoltcp socket。 // 调用者应当在更上层对 Closed 做分支处理。 @@ -1538,7 +1375,6 @@ impl Inner { } } Inner::Established(est) => est.with_mut(f), - Inner::SelfConnected(_) => {} Inner::Closed(_) => {} } } @@ -1546,7 +1382,6 @@ impl Inner { pub fn send_buffer_size(&self) -> usize { match self { Inner::Closed(_) => 0, - Inner::SelfConnected(sc) => sc.send_capacity(), _ => self.with_socket(|socket| socket.send_capacity()), } } @@ -1554,7 +1389,6 @@ impl Inner { pub fn recv_buffer_size(&self) -> usize { match self { Inner::Closed(_) => 0, - Inner::SelfConnected(sc) => sc.recv_capacity(), _ => self.with_socket(|socket| socket.recv_capacity()), } } @@ -1566,7 +1400,6 @@ impl Inner { Inner::Connecting(conn) => Some(conn.inner.iface()), Inner::Listening(listen) => Some(listen.inners[0].iface()), Inner::Established(est) => Some(est.inner.iface()), - Inner::SelfConnected(sc) => Some(sc.inner.iface()), Inner::Closed(_) => None, } } @@ -1589,7 +1422,6 @@ impl Inner { Inner::Connecting(conn) => conn.get_name(), Inner::Listening(listen) => listen.get_name(), Inner::Established(est) => est.get_name(), - Inner::SelfConnected(sc) => sc.get_name(), Inner::Closed(closed) => match closed.ver { smoltcp::wire::IpVersion::Ipv4 => smoltcp::wire::IpEndpoint::new( smoltcp::wire::IpAddress::Ipv4(smoltcp::wire::Ipv4Address::UNSPECIFIED), @@ -1609,7 +1441,6 @@ impl Inner { Inner::Listening(_) => None, Inner::Connecting(conn) => Some(conn.get_peer_name()), Inner::Established(est) => Some(est.get_peer_name()), - Inner::SelfConnected(sc) => Some(sc.get_peer_name()), Inner::Closed(_) => None, } } diff --git a/kernel/src/net/socket/inet/stream/io.rs b/kernel/src/net/socket/inet/stream/io.rs index 7f459cc73..73d74dc15 100644 --- a/kernel/src/net/socket/inet/stream/io.rs +++ b/kernel/src/net/socket/inet/stream/io.rs @@ -79,29 +79,19 @@ impl TcpSocket { } } + // Called with the empty cork buffer locked. Shutdown and enqueue use the + // same cork -> inner order, so no data can be published behind this FIN. fn maybe_complete_shutdown_wr_fin(&self) { if !self.is_send_shutdown() { return; } - if !self - .send_fin_deferred - .swap(false, core::sync::atomic::Ordering::Relaxed) - { - return; - } - let mut writer = self.inner.write(); - let inner = match writer.take() { - Some(inner) => inner, - None => return, - }; - match inner { - inner::Inner::Established(established) => { + if let Some(inner::Inner::Established(established)) = writer.as_mut() { + if self + .send_fin_deferred + .swap(false, core::sync::atomic::Ordering::Relaxed) + { established.with_mut(|socket| socket.close()); - writer.replace(inner::Inner::Established(established)); - } - other => { - writer.replace(other); } } } @@ -378,8 +368,6 @@ impl TcpSocket { let mut total_read = 0; loop { - // SelfConnected does not rely on protocol-stack progress; avoid calling iface.poll() - // here to prevent hangs when running the whole syscall test suite. if let Some(iface) = self.stack_poll_iface_snapshot() { if let Some(netns) = iface.common().net_namespace() { netns.wakeup_poll_thread(); @@ -396,40 +384,6 @@ impl TcpSocket { inner::Inner::Established(established) => established.with_mut(|socket| { self.recv_established(socket, &mut buf[total_read..], flags) }), - inner::Inner::SelfConnected(sc) => { - // Self-connect: data path is a local queue. - let mut current_buf = &mut buf[total_read..]; - let mut recv_exhausted = false; - if self.is_recv_shutdown() { - let remaining = self.recv_shutdown.remaining_limit(); - if remaining == 0 { - sc.discard_all(); - recv_exhausted = true; - } else { - let cap = core::cmp::min(current_buf.len(), remaining); - current_buf = &mut current_buf[..cap]; - } - } - - if recv_exhausted { - Ok(0) - } else { - let peek = flags.contains(PMSG::PEEK); - let trunc = flags.contains(PMSG::TRUNC); - match sc.recv_into(current_buf, peek, trunc) { - Ok(n) => { - if self.is_recv_shutdown() - && !peek - && self.recv_shutdown.record_read(n) - { - sc.discard_all(); - } - Ok(n) - } - Err(error) => Err(error), - } - } - } inner::Inner::Connecting(connecting) => { if let Some(err) = connecting.failure_reason() { connecting.consume_error(); @@ -510,33 +464,6 @@ impl TcpSocket { inner::Inner::Established(established) => { self.recv_established_to_user(established, user_buffer, total_read) } - inner::Inner::SelfConnected(sc) => { - let mut limit = user_buffer.len().saturating_sub(total_read); - let mut recv_exhausted = false; - if self.is_recv_shutdown() { - let remaining = self.recv_shutdown.remaining_limit(); - if remaining == 0 { - sc.discard_all(); - recv_exhausted = true; - } else { - limit = core::cmp::min(limit, remaining); - } - } - - if recv_exhausted { - Ok(0) - } else { - match sc.recv_to_user(user_buffer, total_read, limit) { - Ok(n) => { - if self.is_recv_shutdown() && self.recv_shutdown.record_read(n) { - sc.discard_all(); - } - Ok(n) - } - Err(error) => Err(error), - } - } - } inner::Inner::Connecting(connecting) => { if let Some(err) = connecting.failure_reason() { connecting.consume_error(); @@ -668,6 +595,9 @@ impl TcpSocket { // keep ordering by enqueueing new bytes behind them, and opportunistically flush. { let mut cork_buf = self.cork_buf.lock(); + if self.is_send_shutdown() { + return Err(SystemError::EPIPE); + } if !cork_buf.is_empty() { let cap = self .send_buf_size() @@ -710,6 +640,9 @@ impl TcpSocket { .send_buf_size() .load(core::sync::atomic::Ordering::Relaxed); let mut cork_buf = self.cork_buf.lock(); + if self.is_send_shutdown() { + return Err(SystemError::EPIPE); + } let free = cap.saturating_sub(cork_buf.len()); if free == 0 { return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); @@ -770,22 +703,6 @@ impl TcpSocket { } fn try_send_direct(&self, buf: &[u8]) -> Result { - // Self-connect fast path: avoid smoltcp and iface polling. - let self_connected_result = { - let inner_guard = self.inner.read(); - if let Some(inner::Inner::SelfConnected(sc)) = inner_guard.as_ref() { - Some(sc.send_slice(buf, self.is_send_shutdown())) - } else { - None - } - }; - if let Some(result) = self_connected_result { - let n = result?; - // Wake reader (same fd in another thread) and refresh events. `inner_guard` - // must be gone because notify() refreshes events by reading `inner` again. - self.notify(); - return Ok(n); - } // TODO: add nonblock check of connecting socket // // IMPORTANT: to avoid "all sleepers, no pollers" stalls on loopback (gVisor BlockingLargeSend), diff --git a/kernel/src/net/socket/inet/stream/lifecycle.rs b/kernel/src/net/socket/inet/stream/lifecycle.rs index e3d8a2e87..219c19235 100644 --- a/kernel/src/net/socket/inet/stream/lifecycle.rs +++ b/kernel/src/net/socket/inet/stream/lifecycle.rs @@ -179,7 +179,12 @@ impl TcpSocket { .ipv6_only .load(core::sync::atomic::Ordering::Relaxed); match writer.take().expect("Tcp inner::Inner is None") { - inner::Inner::Init(inner) => match inner.bind(local_endpoint, self.netns(), v6_only) { + inner::Inner::Init(inner) => match inner.bind( + local_endpoint, + self.netns(), + v6_only, + self.device_binding.clone(), + ) { Ok(bound) => { // Linux inet6_bind() makes a concrete native IPv6 binding v6-only. if local_endpoint.addr.version() == smoltcp::wire::IpVersion::Ipv6 @@ -222,6 +227,7 @@ impl TcpSocket { self.options .ipv6_only .load(core::sync::atomic::Ordering::Relaxed), + self.device_binding.clone(), ); match listen_result { Ok(listening) => { @@ -241,6 +247,7 @@ impl TcpSocket { listening.reservation.as_ref().unwrap().id, listening.domain, port, + self.device_binding.ifindex() as u32, ); b.iface().common().bind_socket(me.clone()); registered_ifaces.push(nic_id); @@ -339,48 +346,41 @@ impl TcpSocket { crate::net::socket::inet::common::normalize_unspecified_endpoint_to_loopback( remote_endpoint, ); + // Explicit-source and self-connect paths need the same device/route + // validation as implicit binding, before changing the socket state. + if let Some(inner::Inner::Init(inner::Init::Bound((_, local, _)))) = writer.as_ref() { + if local.addr.version() == smoltcp::wire::IpVersion::Ipv4 + && !local.addr.is_unspecified() + { + let device = self.device_binding.resolve_iface(&self.netns)?; + crate::net::route::resolve_ipv4_route( + &self.netns, + remote_endpoint.addr, + device.map(|iface| iface.nic_id() as u32), + Some(local.addr), + )?; + } + } let inner = writer.take().expect("Tcp inner::Inner is None"); + let old_iface = inner.iface().cloned(); let (init, result) = match inner { inner::Inner::Init(init) => { - // Linux-compatible self-connect: connect() to our own bound addr:port on the - // same socket is allowed and results in a socket that can send/recv to itself. - // smoltcp cannot model this with a single TCP socket instance, so we special-case - // it into `Inner::SelfConnected`. - match init { - inner::Init::Bound((bound, local, reservation)) if local == remote_endpoint => { - // Capture an effective queue capacity from the underlying socket's recv buffer. - let rx_cap = bound - .with::(|s| s.recv_capacity()) - .clamp(1 << 20, super::constants::MAX_SOCKET_BUFFER); - ( - inner::Inner::SelfConnected(inner::SelfConnected::new( - bound, - local, - rx_cap, - reservation, - )), - Ok(()), - ) - } - other => { - let conn_result = other.connect( - remote_endpoint, - self.netns(), - self.self_ref.clone(), - self.ip_version, - ); - match conn_result { - Ok(connecting) => ( - inner::Inner::Connecting(connecting), - if !self.is_nonblock() { - Ok(()) - } else { - Err(SystemError::EINPROGRESS) - }, - ), - Err((init, err)) => (inner::Inner::Init(init), Err(err)), - } - } + match init.connect( + remote_endpoint, + self.netns(), + self.self_ref.clone(), + self.ip_version, + self.device_binding.clone(), + ) { + Ok(connecting) => ( + inner::Inner::Connecting(connecting), + if self.is_nonblock() { + Err(SystemError::EINPROGRESS) + } else { + Ok(()) + }, + ), + Err((init, err)) => (inner::Inner::Init(init), Err(err)), } } inner::Inner::Connecting(connecting) => { @@ -409,17 +409,12 @@ impl TcpSocket { inner::Inner::Established(inner) => { (inner::Inner::Established(inner), Err(SystemError::EISCONN)) } - inner::Inner::SelfConnected(inner) => ( - inner::Inner::SelfConnected(inner), - Err(SystemError::EISCONN), - ), inner::Inner::Closed(_) => (inner, Err(SystemError::ENOTCONN)), }; - // 先落状态再做 iface 侧绑定，避免与 poll 路径形成锁顺序反转死锁： - // - poll: bounds.read -> socket.notify -> socket.inner.read/write - // - connect: socket.inner.write -> bounds.write (会与上面互锁) - // SelfConnected 不依赖协议栈推进，不应触发 iface.poll() + // Publish the state before releasing inner. Connection registration + // retains its own cancellation-aware handoff; polling runs only after + // dropping inner. let need_poll_progress = matches!(init, inner::Inner::Connecting(_)); let registration_publisher = match &init { inner::Inner::Connecting(connecting) => Some(connecting.registration_publisher()), @@ -427,6 +422,20 @@ impl TcpSocket { }; let maybe_iface = init.iface().cloned(); writer.replace(init); + + // Current iface notifications release the bounds snapshot lock before + // calling into a socket. Keep cleanup serialized with a subsequent + // connect, which could otherwise publish a new registration here. + if let Some(old_iface) = old_iface { + if !maybe_iface + .as_ref() + .is_some_and(|iface| Arc::ptr_eq(iface, &old_iface)) + { + old_iface + .common() + .unbind_socket(self.self_ref.upgrade().unwrap()); + } + } drop(writer); // 关键语义：connect(2) 进入 Connecting 状态后，socket 必须能被网络轮询推进。 @@ -464,7 +473,6 @@ impl TcpSocket { let (replace, result) = match inner { inner::Inner::Connecting(conn) => conn.into_result(), inner::Inner::Established(es) => (inner::Inner::Established(es), Ok(())), // TODO check established - inner::Inner::SelfConnected(sc) => (inner::Inner::SelfConnected(sc), Ok(())), _ => { log::warn!("TODO: connecting socket error options"); (inner, Err(SystemError::EINVAL)) @@ -482,11 +490,7 @@ impl TcpSocket { if how.contains(ShutdownBit::SHUT_WR) { if let Err(e) = self.flush_cork_buffer() { - if e == SystemError::EAGAIN_OR_EWOULDBLOCK { - // Defer FIN until cork-buffered bytes are flushed into the TCP stack. - self.send_fin_deferred - .store(true, core::sync::atomic::Ordering::Relaxed); - } else { + if e != SystemError::EAGAIN_OR_EWOULDBLOCK { return Err(e); } } @@ -498,6 +502,10 @@ impl TcpSocket { // Linux/gVisor 语义：TIME_WAIT/Closed 的 stream socket 上 shutdown 应返回 ENOTCONN。 // 但 Listening 和 Connecting 状态下的 shutdown 是允许的。 + // Serialize pending cork data, FIN and the shutdown bit with send's + // enqueue path. A concurrent flusher may have returned early above. + // Keep the same cork -> inner order as cork-flush completion. + let cork_buf = self.cork_buf.lock(); let mut writer = self.inner.write(); let inner = writer.take().expect("Tcp inner::Inner is None"); @@ -518,17 +526,11 @@ impl TcpSocket { } if how.contains(ShutdownBit::SHUT_WR) { - let pending = established.with(|socket| socket.send_queue()); - if pending > 0 { - // Defer FIN until all queued data has been sent. - self.send_fin_deferred - .store(true, core::sync::atomic::Ordering::Relaxed); - } else if self - .send_fin_deferred - .load(core::sync::atomic::Ordering::Relaxed) - { - // FIN will be sent once deferred bytes are fully flushed. - } else { + self.send_fin_deferred + .store(!cork_buf.is_empty(), core::sync::atomic::Ordering::Relaxed); + if cork_buf.is_empty() { + // smoltcp orders FIN after its queued data; waiting for + // ACKs here would unnecessarily delay shutdown. established.with_mut(|socket| socket.close()); } post_poll_rounds = @@ -609,16 +611,9 @@ impl TcpSocket { } if how.contains(ShutdownBit::SHUT_WR) { - let pending = established.with(|socket| socket.send_queue()); - if pending > 0 { - self.send_fin_deferred - .store(true, core::sync::atomic::Ordering::Relaxed); - } else if self - .send_fin_deferred - .load(core::sync::atomic::Ordering::Relaxed) - { - // FIN will be sent once deferred bytes are fully flushed. - } else { + self.send_fin_deferred + .store(!cork_buf.is_empty(), core::sync::atomic::Ordering::Relaxed); + if cork_buf.is_empty() { established.with_mut(|socket| socket.close()); } post_poll_rounds = @@ -647,20 +642,6 @@ impl TcpSocket { ) } } - inner::Inner::SelfConnected(sc) => { - // SelfConnected: shutdown affects only the user-visible data path. - // - SHUT_WR: subsequent send() returns EPIPE; recv() returns EOF once queue drains. - // - SHUT_RD: subsequent recv() returns 0. - // No smoltcp close/abort is needed. - if how.contains(ShutdownBit::SHUT_WR) { - sc.set_send_shutdown(); - } - if how.contains(ShutdownBit::SHUT_RD) { - let queued = sc.recv_queue(); - self.recv_shutdown.init(queued); - } - (inner::Inner::SelfConnected(sc), how) - } other => { writer.replace(other); return Err(SystemError::ENOTCONN); @@ -676,6 +657,7 @@ impl TcpSocket { writer.replace(replace); drop(writer); + drop(cork_buf); // 唤醒等待者（含 poll/epoll），让状态变化可见。 if let Some(iface) = post_poll_iface { @@ -790,15 +772,6 @@ impl TcpSocket { post_poll_iface = Some(iface); writer.replace(inner::Inner::Established(es)); } - inner::Inner::SelfConnected(mut sc) => { - // Release the bound handle and switch to explicit Closed to avoid stale handle access. - let ver = match sc.get_name().addr { - smoltcp::wire::IpAddress::Ipv6(_) => smoltcp::wire::IpVersion::Ipv6, - _ => smoltcp::wire::IpVersion::Ipv4, - }; - sc.release(); - writer.replace(inner::Inner::Closed(inner::Closed::new(ver))); - } inner::Inner::Listening(mut ls) => { // Each backlog slot can already be in a state that emits FIN // when closed. Publish every owning stack before removing the diff --git a/kernel/src/net/socket/inet/stream/option.rs b/kernel/src/net/socket/inet/stream/option.rs index 0ea4b199b..7cb728416 100644 --- a/kernel/src/net/socket/inet/stream/option.rs +++ b/kernel/src/net/socket/inet/stream/option.rs @@ -213,17 +213,17 @@ impl super::TcpSocket { Self::write_i32_opt(value, v) } - /// Helper to get a socket property with a default for closed/self-connected/none states. + /// Helper to get a socket property with a default for closed/uninitialized states. /// /// This pattern is repeated when getting socket options that need to handle - /// cases where the socket is closed, self-connected, or not yet initialized. + /// cases where the socket is closed or not yet initialized. #[inline] fn with_socket_property(&self, default: R, f: F) -> R where F: FnOnce(&inner::Inner) -> R, { match self.inner.read().as_ref() { - Some(inner::Inner::Closed(_)) | Some(inner::Inner::SelfConnected(_)) | None => default, + Some(inner::Inner::Closed(_)) | None => default, Some(inner) => f(inner), } } @@ -243,6 +243,19 @@ impl super::TcpSocket { /// Sets a SOL_SOCKET option. pub(super) fn set_socket_option(&self, opt: PSO, val: &[u8]) -> Result<(), SystemError> { match opt { + PSO::BINDTODEVICE => { + let mut inner = self.inner.write(); + let mut update = self.device_binding.prepare_update(&self.netns, val)?; + let device = update + .target_iface() + .and_then(|iface| core::num::NonZeroU32::new(iface.nic_id() as u32)); + match inner.as_mut().expect("TCP state is present") { + inner::Inner::Listening(listener) => listener.set_bound_device(device), + state => state.for_each_socket_mut(|socket| socket.set_bound_device(device)), + } + update.commit(); + Ok(()) + } PSO::SNDTIMEO_OLD | PSO::SNDTIMEO_NEW => { let d = parse_timeval_opt(val)?; let us = d.map(|v| v.total_micros()).unwrap_or(u64::MAX); @@ -544,6 +557,10 @@ impl super::TcpSocket { value: &mut [u8], ) -> Result { match opt { + PSO::BINDTODEVICE => { + let _state = self.inner.read(); + self.device_binding.get(&self.netns, value) + } PSO::ACCEPTCONN => { let shutdown = self.shutdown.load(core::sync::atomic::Ordering::Acquire); let is_listening = self.is_listening(); diff --git a/kernel/src/net/socket/inet/stream/stream_core.rs b/kernel/src/net/socket/inet/stream/stream_core.rs index d4880b92b..94e1dc78c 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::SocketDeviceBinding; use crate::net::socket::Socket; use crate::process::namespace::net_namespace::NetNamespace; use crate::process::ProcessManager; @@ -150,6 +151,7 @@ pub struct TcpSocket { pub(crate) self_ref: Weak, pub(crate) pollee: AtomicUsize, pub(crate) netns: Arc, + pub(crate) device_binding: Arc, pub(crate) epoll_items: EPollItems, pub(crate) fasync_items: FAsyncItems, pub(crate) options: TcpSocketOptions, @@ -172,6 +174,14 @@ 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, + }; Self { inner: RwSem::new(Some(inner)), shutdown: AtomicUsize::new(0), @@ -184,6 +194,7 @@ impl TcpSocket { self_ref: me.clone(), pollee: AtomicUsize::new(pollee_bits), netns, + device_binding: Arc::new(SocketDeviceBinding::from_ifindex(inherited_device)), epoll_items: EPollItems::default(), fasync_items: FAsyncItems::default(), options: TcpSocketOptions::new(), @@ -248,7 +259,7 @@ impl TcpSocket { pub(super) fn stack_poll_iface_snapshot(&self) -> Option> { let inner = self.inner.read(); match inner.as_ref() { - Some(inner::Inner::SelfConnected(_)) | None => None, + None => None, Some(inner) => inner.iface().cloned(), } } @@ -266,7 +277,6 @@ impl TcpSocket { .as_ref() .map(|inner| match inner { inner::Inner::Closed(_) => 0, - inner::Inner::SelfConnected(sc) => sc.recv_queue(), _ => inner.with_socket(|s| s.recv_queue()), }) .unwrap_or(0) @@ -455,10 +465,6 @@ impl TcpSocket { }); established.update_io_events(&self.pollee); } - Some(inner::Inner::SelfConnected(sc)) => { - sc.set_recv_buffer_size(rx_size); - sc.update_io_events(&self.pollee, self.is_recv_shutdown()); - } Some(inner::Inner::Connecting(connecting)) => { connecting.with_mut(|socket| { socket.set_send_buffer_size(tx_size); diff --git a/kernel/src/net/tcp_listener.rs b/kernel/src/net/tcp_listener.rs index e87964b24..0aef09200 100644 --- a/kernel/src/net/tcp_listener.rs +++ b/kernel/src/net/tcp_listener.rs @@ -15,6 +15,7 @@ struct Listener { id: u64, domain: TcpBindDomain, port: u16, + device: u32, } /// Per-interface listener facts, with no owning socket or interface references. @@ -30,9 +31,14 @@ impl TcpListenerRegistry { } } - pub fn register(&self, id: u64, domain: TcpBindDomain, port: u16) { + pub fn register(&self, id: u64, domain: TcpBindDomain, port: u16, device: u32) { let mut listeners = self.listeners.write(); - let entry = Listener { id, domain, port }; + let entry = Listener { + id, + domain, + port, + device, + }; if let Some(existing) = listeners.iter_mut().find(|listener| listener.id == id) { *existing = entry; } else { @@ -49,10 +55,12 @@ impl TcpListenerRegistry { } impl smoltcp::iface::TcpListenRegistry for TcpListenerRegistry { - fn is_listening(&self, local_endpoint: IpEndpoint) -> bool { + fn is_listening(&self, local_endpoint: IpEndpoint, meta: smoltcp::phy::PacketMeta) -> bool { // Called with SocketSet locked: never acquire socket/iface locks here. self.listeners.read().iter().any(|listener| { - listener.port == local_endpoint.port && listener.domain.matches(local_endpoint.addr) + listener.port == local_endpoint.port + && listener.domain.matches(local_endpoint.addr) + && (listener.device == 0 || listener.device == meta.id) }) } } diff --git a/user/apps/tests/dunitest/no_skip.txt b/user/apps/tests/dunitest/no_skip.txt index cbd870fbf..e73dd9e3f 100644 --- a/user/apps/tests/dunitest/no_skip.txt +++ b/user/apps/tests/dunitest/no_skip.txt @@ -25,6 +25,7 @@ normal/tcp_dual_stack_semantics normal/tcp_relisten normal/tcp_listener_overflow normal/tcp_accept_handshake +normal/tcp_device_binding normal/socket_nonblocking normal/poll_timeout_semantics diff --git a/user/apps/tests/dunitest/suites/normal/tcp_accept_handshake.cc b/user/apps/tests/dunitest/suites/normal/tcp_accept_handshake.cc index bcd5e0dcd..fb42b7130 100644 --- a/user/apps/tests/dunitest/suites/normal/tcp_accept_handshake.cc +++ b/user/apps/tests/dunitest/suites/normal/tcp_accept_handshake.cc @@ -57,6 +57,7 @@ class TcpAcceptHandshake : public testing::Test { int interface_ = 0; uint16_t port_ = 0; uint32_t server_sequence_ = 0; + uint16_t peer_port_ = kPeerPort; std::array source_mac_{}, destination_mac_{}; void SetUp() override { @@ -126,7 +127,7 @@ class TcpAcceptHandshake : public testing::Test { Put32(ip + 12, kPeer); Put32(ip + 16, kServer); Put16(ip + 10, Checksum(ip, 20)); auto* tcp = ip + 20; - Put16(tcp, kPeerPort); Put16(tcp + 2, port_); + Put16(tcp, peer_port_); Put16(tcp + 2, port_); Put32(tcp + 4, sequence); Put32(tcp + 8, acknowledgement); tcp[12] = 0x50; tcp[13] = flags; Put16(tcp + 14, 65535); const uint32_t pseudo = (kPeer >> 16) + (kPeer & 0xffff) + @@ -155,7 +156,7 @@ class TcpAcceptHandshake : public testing::Test { Get32(ip + 16) != kPeer) continue; auto* tcp = ip + header; if ((uint16_t(tcp[0]) << 8 | tcp[1]) != port_ || - (uint16_t(tcp[2]) << 8 | tcp[3]) != kPeerPort || + (uint16_t(tcp[2]) << 8 | tcp[3]) != peer_port_ || (tcp[13] & required_flags) != required_flags || (acknowledgement != 0 && Get32(tcp + 8) != acknowledgement)) continue; server_sequence_ = Get32(tcp + 4); @@ -167,6 +168,22 @@ class TcpAcceptHandshake : public testing::Test { ASSERT_NO_FATAL_FAILURE(Segment(0x02, kInitialSequence, 0)); ASSERT_NO_FATAL_FAILURE(WaitReply(0x12)); } + void ExpectDevice(int fd, const char* expected) { + char device[IFNAMSIZ]{}; + socklen_t length = sizeof(device); + ASSERT_EQ(getsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE, device, &length), 0); + EXPECT_EQ(length, strlen(expected) + 1); + EXPECT_STREQ(device, expected); + } + void AcceptAndCheckDevice(const char* device) { + ASSERT_NO_FATAL_FAILURE(Segment(0x10, kInitialSequence + 1, server_sequence_ + 1)); + pollfd event{listener_.get(), POLLIN, 0}; + ASSERT_EQ(poll(&event, 1, 3000), 1); + Fd accepted(accept4(listener_.get(), nullptr, nullptr, SOCK_NONBLOCK)); + ASSERT_GE(accepted.get(), 0) << strerror(errno); + ASSERT_NO_FATAL_FAILURE(ExpectDevice(accepted.get(), device)); + ASSERT_NO_FATAL_FAILURE(Segment(0x14, kInitialSequence + 1, server_sequence_ + 1)); + } }; TEST_F(TcpAcceptHandshake, SynReceivedIsNotReadableOrAcceptable) { @@ -205,6 +222,27 @@ TEST_F(TcpAcceptHandshake, FinalAckWithFinRemainsAcceptable) { EXPECT_EQ(recv(accepted.get(), &byte, 1, 0), 0); ASSERT_NO_FATAL_FAILURE(Segment(0x14, kInitialSequence + 2, server_sequence_ + 1)); } + +TEST_F(TcpAcceptHandshake, AcceptedAndReplenishedSlotsKeepDeviceBinding) { + // With one listener slot, the second connection must use a replenished slot. + ASSERT_EQ(listen(listener_.get(), 1), 0); + for (int i = 0; i < 3; ++i) { + ASSERT_NO_FATAL_FAILURE(StartHandshake()); + ASSERT_NO_FATAL_FAILURE(AcceptAndCheckDevice("veth2")); + ++peer_port_; + } +} + +TEST_F(TcpAcceptHandshake, PendingChildKeepsDeviceWhenListenerChanges) { + ASSERT_NO_FATAL_FAILURE(StartHandshake()); + constexpr char new_device[] = "veth1"; + ASSERT_EQ(setsockopt(listener_.get(), SOL_SOCKET, SO_BINDTODEVICE, + new_device, sizeof(new_device)), 0); + ASSERT_NO_FATAL_FAILURE(ExpectDevice(listener_.get(), "veth1")); + // The SYN arrived on veth2 before the change. Its ACK and the accepted + // child must retain that snapshot even though future SYNs use veth1. + ASSERT_NO_FATAL_FAILURE(AcceptAndCheckDevice("veth2")); +} } // namespace int main(int argc, char** argv) { diff --git a/user/apps/tests/dunitest/suites/normal/tcp_device_binding.cc b/user/apps/tests/dunitest/suites/normal/tcp_device_binding.cc new file mode 100644 index 000000000..637d7e9c3 --- /dev/null +++ b/user/apps/tests/dunitest/suites/normal/tcp_device_binding.cc @@ -0,0 +1,293 @@ +// IPv4 SO_BINDTODEVICE semantics, also runnable against the Linux fixture. +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace { +class Fd { + public: + explicit Fd(int type = SOCK_STREAM) : value_(socket(AF_INET, type, 0)) {} + ~Fd() { if (value_ >= 0) close(value_); } + Fd(const Fd&) = delete; + Fd& operator=(const Fd&) = delete; + int get() const { return value_; } + private: + int value_; +}; + +void SetDevice(int fd, const char* device) { + ASSERT_EQ(setsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE, + device, strlen(device) + 1), 0) << strerror(errno); +} + +void CheckDevice(int fd, const char* expected) { + char device[IFNAMSIZ]{}; + socklen_t length = sizeof(device); + ASSERT_EQ(getsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE, device, &length), 0); + EXPECT_EQ(length, expected[0] ? strlen(expected) + 1 : 0u); + EXPECT_STREQ(device, expected); +} + +void FinishConnect(int fd, const sockaddr_in& peer) { + const int result = connect(fd, reinterpret_cast(&peer), sizeof(peer)); + if (result != 0) { + ASSERT_EQ(errno, EINPROGRESS) << strerror(errno); + pollfd event{fd, POLLOUT, 0}; + ASSERT_EQ(poll(&event, 1, 3000), 1); + } + int error = -1; + socklen_t length = sizeof(error); + ASSERT_EQ(getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &length), 0); + ASSERT_EQ(error, 0) << strerror(error); + sockaddr_in remote{}; + length = sizeof(remote); + ASSERT_EQ(getpeername(fd, reinterpret_cast(&remote), &length), 0); + EXPECT_EQ(remote.sin_addr.s_addr, peer.sin_addr.s_addr); + EXPECT_EQ(remote.sin_port, peer.sin_port); +} + +TEST(TcpDeviceBinding, SetGetUnknownClearAndShortBuffer) { + Fd socket; + ASSERT_GE(socket.get(), 0); + ASSERT_NE(if_nametoindex("lo"), 0u); + ASSERT_NO_FATAL_FAILURE(CheckDevice(socket.get(), "")); + ASSERT_NO_FATAL_FAILURE(SetDevice(socket.get(), "lo")); + ASSERT_NO_FATAL_FAILURE(CheckDevice(socket.get(), "lo")); + + // Linux requires IFNAMSIZ, even when the actual name fits a shorter buffer. + char short_buffer[IFNAMSIZ - 1]{}; + socklen_t length = sizeof(short_buffer); + EXPECT_EQ(getsockopt(socket.get(), SOL_SOCKET, SO_BINDTODEVICE, + short_buffer, &length), -1); + EXPECT_EQ(errno, EINVAL); + constexpr char missing[] = "ngc009_missing"; + ASSERT_EQ(if_nametoindex(missing), 0u); + EXPECT_EQ(setsockopt(socket.get(), SOL_SOCKET, SO_BINDTODEVICE, + missing, sizeof(missing)), -1); + EXPECT_EQ(errno, ENODEV); + ASSERT_NO_FATAL_FAILURE(CheckDevice(socket.get(), "lo")); + + ASSERT_NO_FATAL_FAILURE(SetDevice(socket.get(), "")); + ASSERT_NO_FATAL_FAILURE(CheckDevice(socket.get(), "")); + length = 0; + ASSERT_EQ(getsockopt(socket.get(), SOL_SOCKET, SO_BINDTODEVICE, + short_buffer, &length), 0); + EXPECT_EQ(length, 0u); + ASSERT_NO_FATAL_FAILURE(SetDevice(socket.get(), "lo")); + ASSERT_EQ(setsockopt(socket.get(), SOL_SOCKET, SO_BINDTODEVICE, nullptr, 0), 0); + ASSERT_NO_FATAL_FAILURE(CheckDevice(socket.get(), "")); +} + +class TcpDeviceBindingPorts : public testing::Test { + protected: + void SetUp() override { + ASSERT_NE(if_nametoindex("veth1"), 0u) << "requires the standard veth fixture"; + ASSERT_NE(if_nametoindex("veth2"), 0u); + } + void BindEphemeral(int fd, sockaddr_in* local) { + local->sin_family = AF_INET; + local->sin_addr.s_addr = htonl(INADDR_ANY); + ASSERT_EQ(bind(fd, reinterpret_cast(local), sizeof(*local)), 0); + socklen_t length = sizeof(*local); + ASSERT_EQ(getsockname(fd, reinterpret_cast(local), &length), 0); + ASSERT_NE(local->sin_port, 0); + } +}; + +TEST_F(TcpDeviceBindingPorts, DifferentDevicesAllowSameAddressAndPort) { + Fd first, second; + ASSERT_GE(first.get(), 0); + ASSERT_GE(second.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(first.get(), "veth1")); + ASSERT_NO_FATAL_FAILURE(SetDevice(second.get(), "veth2")); + sockaddr_in local{}; + ASSERT_NO_FATAL_FAILURE(BindEphemeral(first.get(), &local)); + ASSERT_EQ(bind(second.get(), reinterpret_cast(&local), sizeof(local)), 0) + << strerror(errno); + ASSERT_EQ(listen(first.get(), 1), 0); + ASSERT_EQ(listen(second.get(), 1), 0); +} + +TEST_F(TcpDeviceBindingPorts, SameDeviceOrUnboundDeviceConflicts) { + for (const char* initial_device : {"veth1", ""}) { + Fd first; + ASSERT_GE(first.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(first.get(), initial_device)); + sockaddr_in local{}; + ASSERT_NO_FATAL_FAILURE(BindEphemeral(first.get(), &local)); + for (const char* other_device : {"veth1", ""}) { + Fd other; + ASSERT_GE(other.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(other.get(), other_device)); + EXPECT_EQ(bind(other.get(), reinterpret_cast(&local), sizeof(local)), -1); + EXPECT_EQ(errno, EADDRINUSE); + } + } +} + +TEST_F(TcpDeviceBindingPorts, ChangingDeviceUpdatesFutureBindConflicts) { + Fd first, old_device, new_device; + ASSERT_GE(first.get(), 0); + ASSERT_GE(old_device.get(), 0); + ASSERT_GE(new_device.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(first.get(), "veth1")); + sockaddr_in local{}; + ASSERT_NO_FATAL_FAILURE(BindEphemeral(first.get(), &local)); + ASSERT_NO_FATAL_FAILURE(SetDevice(first.get(), "veth2")); + ASSERT_NO_FATAL_FAILURE(SetDevice(old_device.get(), "veth1")); + ASSERT_EQ(bind(old_device.get(), reinterpret_cast(&local), sizeof(local)), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(new_device.get(), "veth2")); + EXPECT_EQ(bind(new_device.get(), reinterpret_cast(&local), sizeof(local)), -1); + EXPECT_EQ(errno, EADDRINUSE); + // Linux permits changing the device after bind even if that creates an + // overlap; setsockopt must not perform a second bind conflict check. + ASSERT_NO_FATAL_FAILURE(SetDevice(first.get(), "veth1")); + ASSERT_NO_FATAL_FAILURE(CheckDevice(first.get(), "veth1")); +} + +TEST_F(TcpDeviceBindingPorts, SelfConnectHonorsDeviceConstraint) { + for (const char* device : {"lo", "veth1"}) { + SCOPED_TRACE(device); + Fd client(SOCK_STREAM | SOCK_NONBLOCK); + ASSERT_GE(client.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), device)); + sockaddr_in local{}; + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ASSERT_EQ(bind(client.get(), reinterpret_cast(&local), sizeof(local)), 0); + socklen_t length = sizeof(local); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&local), &length), 0); + if (strcmp(device, "lo") == 0) { + ASSERT_NO_FATAL_FAILURE(FinishConnect(client.get(), local)); + } else { + // Linux rejects a loopback source constrained to a non-loopback + // output device, including the otherwise valid self-connect case. + ASSERT_EQ(connect(client.get(), reinterpret_cast(&local), sizeof(local)), -1); + EXPECT_EQ(errno, EINVAL); + } + } +} + +TEST_F(TcpDeviceBindingPorts, ExplicitLoopbackSourceRejectsNonLoopbackOutput) { + Fd client(SOCK_STREAM | SOCK_NONBLOCK); + ASSERT_GE(client.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), "veth1")); + sockaddr_in local{}; + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ASSERT_EQ(bind(client.get(), reinterpret_cast(&local), sizeof(local)), 0); + socklen_t length = sizeof(local); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&local), &length), 0); + sockaddr_in peer = local; + peer.sin_addr.s_addr = htonl(0x6f6f0b02); // Standard fixture veth2 address. + ASSERT_EQ(connect(client.get(), reinterpret_cast(&peer), sizeof(peer)), -1); + EXPECT_EQ(errno, EINVAL); + sockaddr_in after{}; + length = sizeof(after); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&after), &length), 0); + EXPECT_EQ(after.sin_addr.s_addr, local.sin_addr.s_addr); + EXPECT_EQ(after.sin_port, local.sin_port); + ASSERT_NO_FATAL_FAILURE(CheckDevice(client.get(), "veth1")); +} + +TEST_F(TcpDeviceBindingPorts, CrossDeviceSelfConnectRequiresNetworkHandshake) { + Fd client(SOCK_STREAM | SOCK_NONBLOCK); + ASSERT_GE(client.get(), 0); + sockaddr_in local{}; + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(0x6f6f0b01); // Standard fixture veth1 address. + ASSERT_EQ(bind(client.get(), reinterpret_cast(&local), sizeof(local)), 0); + socklen_t length = sizeof(local); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&local), &length), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), "veth2")); + // Equal endpoints alone do not imply a local route: the device constraint + // selects the veth2 output path. Linux starts a real asynchronous handshake. + // Do not wait for timeout, whose duration depends on TCP retransmission. + ASSERT_EQ(connect(client.get(), reinterpret_cast(&local), sizeof(local)), -1); + EXPECT_EQ(errno, EINPROGRESS); +} + +TEST_F(TcpDeviceBindingPorts, EstablishedSelfConnectHonorsDeviceChanges) { + Fd client(SOCK_STREAM | SOCK_NONBLOCK); + ASSERT_GE(client.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), "lo")); + sockaddr_in local{}; + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ASSERT_EQ(bind(client.get(), reinterpret_cast(&local), sizeof(local)), 0); + socklen_t length = sizeof(local); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&local), &length), 0); + ASSERT_NO_FATAL_FAILURE(FinishConnect(client.get(), local)); + + // First leave an already received byte queued across the device change. + const char old_byte = 'a'; + ASSERT_EQ(send(client.get(), &old_byte, 1, 0), 1); + pollfd event{client.get(), POLLIN, 0}; + ASSERT_EQ(poll(&event, 1, 3000), 1); + ASSERT_NE(event.revents & POLLIN, 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), "veth1")); + char received = 0; + ASSERT_EQ(recv(client.get(), &received, 1, 0), 1); + EXPECT_EQ(received, old_byte); + + // send() may queue data, but a non-loopback device must prevent delivery + // to the loopback endpoint even for a connection with equal endpoints. + const char new_byte = 'b'; + ASSERT_EQ(send(client.get(), &new_byte, 1, 0), 1); + event.revents = 0; + EXPECT_EQ(poll(&event, 1, 150), 0); + ASSERT_EQ(recv(client.get(), &received, 1, 0), -1); + EXPECT_EQ(errno, EAGAIN); + + // Clearing the constraint restores routing. The byte already queued for + // transmission must arrive through TCP retransmission without another send. + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), "")); + event.revents = 0; + ASSERT_EQ(poll(&event, 1, 5000), 1); + ASSERT_NE(event.revents & POLLIN, 0); + ASSERT_EQ(recv(client.get(), &received, 1, 0), 1); + EXPECT_EQ(received, new_byte); +} + +TEST_F(TcpDeviceBindingPorts, WildcardBindConnectPreservesPortAndSelectsLoopbackSource) { + Fd listener(SOCK_STREAM | SOCK_NONBLOCK), client(SOCK_STREAM | SOCK_NONBLOCK); + ASSERT_GE(listener.get(), 0); + ASSERT_GE(client.get(), 0); + ASSERT_NO_FATAL_FAILURE(SetDevice(listener.get(), "lo")); + ASSERT_NO_FATAL_FAILURE(SetDevice(client.get(), "lo")); + 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(), 1), 0); + sockaddr_in before{}; + ASSERT_NO_FATAL_FAILURE(BindEphemeral(client.get(), &before)); + ASSERT_EQ(before.sin_addr.s_addr, htonl(INADDR_ANY)); + ASSERT_NO_FATAL_FAILURE(FinishConnect(client.get(), server)); + sockaddr_in after{}; + length = sizeof(after); + ASSERT_EQ(getsockname(client.get(), reinterpret_cast(&after), &length), 0); + EXPECT_EQ(after.sin_port, before.sin_port); + EXPECT_EQ(after.sin_addr.s_addr, htonl(INADDR_LOOPBACK)); + ASSERT_NO_FATAL_FAILURE(CheckDevice(client.get(), "lo")); + pollfd event{listener.get(), POLLIN, 0}; + ASSERT_EQ(poll(&event, 1, 3000), 1); + const int accepted = accept(listener.get(), nullptr, nullptr); + ASSERT_GE(accepted, 0) << strerror(errno); + EXPECT_EQ(close(accepted), 0); +} +} // namespace + +int main(int argc, char** argv) { + testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/user/apps/tests/dunitest/suites/normal/tcp_self_connect_semantics.cc b/user/apps/tests/dunitest/suites/normal/tcp_self_connect_semantics.cc index 9c93a8f41..aa9743db0 100644 --- a/user/apps/tests/dunitest/suites/normal/tcp_self_connect_semantics.cc +++ b/user/apps/tests/dunitest/suites/normal/tcp_self_connect_semantics.cc @@ -151,6 +151,18 @@ PollOutcome PollNow(int socket_fd, int events) { return outcome; } +void WaitForEvent(int socket_fd, short event) { + // Request only the prerequisite event: POLLOUT or already queued POLLIN + // would return immediately while a FIN is still travelling through TCP. + pollfd descriptor {}; + descriptor.fd = socket_fd; + descriptor.events = event; + ASSERT_EQ(poll(&descriptor, 1, 5000), 1) + << "waiting for event=" << event << ": " << ErrnoString(errno); + ASSERT_NE(descriptor.revents & event, 0) + << "requested event=" << event << ", revents=" << descriptor.revents; +} + SendAllResult SendAll(int socket_fd, const std::uint8_t* data, std::size_t length) { SendAllResult result; while (result.bytes_sent < length) { @@ -188,6 +200,7 @@ TEST_P(TcpSelfConnectSemantics, PartialProgressWinsOverWouldBlock) { ASSERT_EQ(send(socket_fd.Get(), kReadPayload.data(), kReadPayload.size(), 0), static_cast(kReadPayload.size())) << "send before read failed: " << ErrnoString(errno); + ASSERT_NO_FATAL_FAILURE(WaitForEvent(socket_fd.Get(), POLLIN)); ASSERT_EQ(read(socket_fd.Get(), buffer.data(), buffer.size()), static_cast(kReadPayload.size())) << "read must return progress copied before the second probe observed EAGAIN: " @@ -204,6 +217,8 @@ TEST_P(TcpSelfConnectSemantics, PartialProgressWinsOverWouldBlock) { ASSERT_EQ(send(socket_fd.Get(), kRecvPayload.data(), kRecvPayload.size(), 0), static_cast(kRecvPayload.size())) << "send before recv failed: " << ErrnoString(errno); + // A successful send queues bytes; Nagle/delayed ACK may defer delivery. + ASSERT_NO_FATAL_FAILURE(WaitForEvent(socket_fd.Get(), POLLIN)); ASSERT_EQ(recv(socket_fd.Get(), buffer.data(), buffer.size(), MSG_DONTWAIT), static_cast(kRecvPayload.size())) << "recv must return progress copied before the second probe observed EAGAIN: " @@ -275,6 +290,8 @@ TEST_P(TcpSelfConnectSemantics, PollWriteShutdownReportsFullHangup) { ASSERT_EQ(shutdown(socket_fd.Get(), SHUT_WR), 0) << "shutdown(SHUT_WR) failed: " << ErrnoString(errno); + // The receive shutdown becomes visible only after our FIN loops back. + ASSERT_NO_FATAL_FAILURE(WaitForEvent(socket_fd.Get(), POLLRDHUP)); const PollOutcome outcome = PollNow(socket_fd.Get(), kPollEvents); ASSERT_EQ(outcome.ret, 1) << "poll revents=" << outcome.revents; EXPECT_NE(outcome.revents & POLLIN, 0) << "revents=" << outcome.revents; @@ -293,6 +310,25 @@ TEST_P(TcpSelfConnectSemantics, PollWriteShutdownReportsFullHangup) { EXPECT_EQ(read(socket_fd.Get(), buffer.data(), buffer.size()), 0); } +TEST_P(TcpSelfConnectSemantics, WriteShutdownFlushesCorkedDataBeforeEof) { + FdGuard socket_fd = CreateSelfConnectedSocket(GetParam()); + ASSERT_GE(socket_fd.Get(), 0); + + constexpr std::array payload {1, 2, 3, 4, 5, 6}; + ASSERT_EQ(send(socket_fd.Get(), payload.data(), payload.size(), MSG_MORE), + static_cast(payload.size())); + ASSERT_EQ(shutdown(socket_fd.Get(), SHUT_WR), 0) << ErrnoString(errno); + ASSERT_NO_FATAL_FAILURE(WaitForEvent(socket_fd.Get(), POLLRDHUP)); + + std::array buffer {}; + ASSERT_EQ(recv(socket_fd.Get(), buffer.data(), buffer.size(), MSG_DONTWAIT), + static_cast(payload.size())); + EXPECT_TRUE(std::equal(payload.begin(), payload.end(), buffer.begin())); + EXPECT_EQ(recv(socket_fd.Get(), buffer.data(), buffer.size(), MSG_DONTWAIT), 0); + EXPECT_EQ(send(socket_fd.Get(), payload.data(), payload.size(), MSG_NOSIGNAL), -1); + EXPECT_EQ(errno, EPIPE); +} + // SHUT_RD 只是半关闭：Linux 6.6 tcp_poll() 此时只置 RCV_SHUTDOWN， // 报告 POLLRDHUP（以及 POLLIN），不得报告 POLLHUP。 TEST_P(TcpSelfConnectSemantics, PollReadShutdownReportsHalfClose) { diff --git a/user/apps/tests/dunitest/whitelist.txt b/user/apps/tests/dunitest/whitelist.txt index 02cb96eec..29d76b6a1 100644 --- a/user/apps/tests/dunitest/whitelist.txt +++ b/user/apps/tests/dunitest/whitelist.txt @@ -60,6 +60,7 @@ normal/tcp_listen_poll_semantics normal/tcp_relisten normal/tcp_listener_overflow normal/tcp_accept_handshake +normal/tcp_device_binding normal/socket_nonblocking normal/tcp_self_connect_semantics normal/poll_timeout_semantics