From d2d172721276133de01cfc606cc2e623feb5629e Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Mon, 17 Aug 2026 21:11:50 -0700 Subject: [PATCH 1/3] Extract UDP native endpoint route slots Colocate native UDP endpoint readiness, errors, epoll identity, and lifecycle state behind route-aware slots while preserving the existing external-only behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b5a1a347-37a8-4246-8bbc-306590921475 --- .../src/socket.rs | 191 ++--- .../src/socket/tests/udp.rs | 66 ++ .../src/socket/udp.rs | 717 +++++++++++++----- 3 files changed, 687 insertions(+), 287 deletions(-) diff --git a/litebox_broker_platform_linux_userland/src/socket.rs b/litebox_broker_platform_linux_userland/src/socket.rs index 29058da21..dc8e15850 100644 --- a/litebox_broker_platform_linux_userland/src/socket.rs +++ b/litebox_broker_platform_linux_userland/src/socket.rs @@ -50,8 +50,8 @@ use tcp::{ PendingGuestConnectionMatch, PendingGuestTcpConnection, }; use udp::{ - ReactorUdpBinding, ReactorUdpPeer, ReactorUdpState, UDP_EVENT_TOKEN_FLAG, UdpNativeErrorState, - UdpReceiveOrigin, UdpSocketState, is_local_ipv4_address, + ReactorUdpBinding, ReactorUdpPeer, ReactorUdpState, UDP_EVENT_TOKEN_FLAG, UdpEndpointSlot, + UdpNativeRoute, UdpReceiveOrigin, UdpSocketState, is_local_ipv4_address, }; /// Epoll token reserved for the eventfd that wakes the reactor for commands. @@ -617,6 +617,16 @@ impl ReactorClient { receive.recv().unwrap() } + #[cfg(test)] + fn udp_native_event_token_count(&self) -> usize { + let (response, receive) = sync_channel(1); + self.commands + .send(ReactorCommand::UdpNativeEventTokenCount { response }) + .unwrap(); + self.signal().unwrap(); + receive.recv().unwrap() + } + #[cfg(test)] fn udp_native_receive_buffer_size(&self, guest_port: u16) -> Option { let (response, receive) = sync_channel(1); @@ -856,6 +866,10 @@ enum ReactorCommand { response: SyncSender, }, #[cfg(test)] + UdpNativeEventTokenCount { + response: SyncSender, + }, + #[cfg(test)] UdpNativeReceiveBufferSize { guest_port: u16, response: SyncSender>>, @@ -1098,6 +1112,8 @@ impl Reactor { } }; let mut reused_host_address = None; + let native_route = UdpNativeRoute::External; + let endpoint_slot = UdpEndpointSlot::Active(native_route); let mut staged_endpoint = match peer { ReactorUdpPeer::Guest { .. } => None, ReactorUdpPeer::External(address) => { @@ -1107,10 +1123,10 @@ impl Reactor { .ok_or(PlatformConnectError::PeerUnchanged(BrokerError::Internal))? .udp_state() .map_err(PlatformConnectError::PeerUnchanged)? - .external_endpoint + .native_endpoint(native_route) .is_some() { - match self.connect_existing_udp_endpoint(id, address)? { + match self.connect_existing_udp_endpoint(id, native_route, address)? { SocketOutcome::Completed(host_address) => { reused_host_address = Some(host_address); None @@ -1121,7 +1137,13 @@ impl Reactor { } } else { match self - .stage_udp_endpoint(id, address, Some(address)) + .stage_udp_endpoint( + id, + native_route, + endpoint_slot, + address, + Some(address), + ) .map_err(PlatformConnectError::PeerUnchanged)? { SocketOutcome::Completed(endpoint) => Some(endpoint), @@ -1170,7 +1192,11 @@ impl Reactor { } }; if reused_host_address.is_none() { - self.replace_udp_endpoint(id, staged_endpoint.take()); + self.replace_all_udp_endpoints(id, staged_endpoint.take()) + .map_err(PlatformConnectError::PeerIndeterminate)?; + } else { + self.retain_udp_endpoint(id, native_route) + .map_err(PlatformConnectError::PeerIndeterminate)?; } self.clear_udp_external_peers(id) .map_err(PlatformConnectError::PeerIndeterminate)?; @@ -1242,6 +1268,8 @@ impl Reactor { self.enqueue_guest_datagram(id, socket_generation, data) } ReactorUdpPeer::External(address) => { + let native_route = UdpNativeRoute::External; + let endpoint_slot = UdpEndpointSlot::Active(native_route); let external_peer_added = if authorize_external_reply { self.reserve_udp_external_peer(id, address)? } else { @@ -1252,10 +1280,16 @@ impl Reactor { .get(&id) .ok_or(BrokerError::Internal)? .udp_state()? - .external_endpoint + .native_endpoint(native_route) .is_none() { - let endpoint = match self.stage_udp_endpoint(id, address, None) { + let endpoint = match self.stage_udp_endpoint( + id, + native_route, + endpoint_slot, + address, + None, + ) { Ok(SocketOutcome::Completed(endpoint)) => endpoint, Ok(SocketOutcome::Failed(error)) => { if external_peer_added { @@ -1270,9 +1304,9 @@ impl Reactor { return Err(error); } }; - self.replace_udp_endpoint(id, Some(endpoint)); + self.replace_udp_endpoint(id, native_route, Some(endpoint))?; } - let outcome = self.send_external_udp(id, data, address); + let outcome = self.send_external_udp(id, endpoint_slot, data, address); if !matches!(outcome, Ok(SocketOutcome::Completed(_))) && external_peer_added { self.remove_udp_external_peer(id, address); } @@ -1315,10 +1349,10 @@ impl Reactor { socket.write_shutdown |= shut_write; if shut_read { let udp = socket.udp_state_mut()?; - if udp.peeked_origin == Some(UdpReceiveOrigin::Native) { + if udp.peeked_origin == Some(UdpReceiveOrigin::External) { udp.peeked_origin = None; } - if let Some(endpoint) = udp.external_endpoint.as_mut() { + if let Some(endpoint) = udp.native_endpoint_mut(UdpNativeRoute::External) { endpoint.readable = false; } } @@ -1338,71 +1372,14 @@ impl Reactor { .get(&socket_id) .map(SocketEntry::kind) .ok_or(BrokerError::Internal)?; - let response = status_socket( + if kind == SocketKind::Udp { + return self.status_udp_socket(socket_id); + } + status_socket( self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)?, - ); - let response = match response { - Ok(response) => response, - Err(error) if kind == SocketKind::Udp => { - self.rearm_udp_endpoint_if_needed(socket_id)?; - return Err(error); - } - Err(error) => return Err(error), - }; - if kind == SocketKind::Udp { - let readiness = match self.udp_readiness(socket_id) { - Ok(readiness) => readiness, - Err(error) => { - self.rearm_udp_endpoint_if_needed(socket_id)?; - return Err(error); - } - }; - let publication = update_snapshot( - self.sockets - .get_mut(&socket_id) - .ok_or(BrokerError::Internal)?, - None, - readiness, - ); - let rearm = self.rearm_udp_endpoint_if_needed(socket_id); - // The synchronous response carries the consumed UDP error and the - // cached snapshot is already authoritative. Do not discard that - // error if rearming the endpoint fails after consumption. - if let Err(error) = rearm { - if let Some(pending_error) = response.pending_error { - let socket = self - .sockets - .get_mut(&socket_id) - .expect("UDP status socket disappeared after rearm failure"); - let next_pending_error = { - let mut snapshot = socket - .snapshot - .lock() - .expect("Linux socket snapshot mutex poisoned"); - let next_pending_error = snapshot.pending_error.replace(pending_error); - snapshot.readiness = snapshot.readiness | ReadinessFlags::ERROR; - next_pending_error - }; - if let Some(next_pending_error) = next_pending_error { - socket - .udp_state_mut() - .expect("UDP status socket changed kind after rearm failure") - .native_error = UdpNativeErrorState::Consumed(next_pending_error); - } - let readiness = socket - .snapshot - .lock() - .expect("Linux socket snapshot mutex poisoned") - .readiness; - let _ = socket.readiness.publish(readiness); - } - return Err(error); - } - let _ = publication; - } - Ok(response) + ) } fn receive_from_socket( @@ -1434,13 +1411,15 @@ impl Reactor { .next_receive_origin; let first = pinned.unwrap_or(next); let second = match first { - UdpReceiveOrigin::Guest => UdpReceiveOrigin::Native, - UdpReceiveOrigin::Native => UdpReceiveOrigin::Guest, + UdpReceiveOrigin::Guest => UdpReceiveOrigin::External, + UdpReceiveOrigin::External => UdpReceiveOrigin::Guest, }; for origin in [first, second] { let outcome = match origin { UdpReceiveOrigin::Guest => self.receive_guest_udp(id, length, flags)?, - UdpReceiveOrigin::Native => self.receive_native_udp(id, length, flags)?, + UdpReceiveOrigin::External => { + self.receive_native_udp(id, UdpNativeRoute::External, length, flags)? + } }; if let Some(outcome) = outcome { return Ok(outcome); @@ -1463,7 +1442,7 @@ impl Reactor { }; if kind == SocketKind::Udp { let _ = self.clear_udp_receive_queue(id); - self.replace_udp_endpoint(id, None); + let _ = self.replace_all_udp_endpoints(id, None); if let Some(binding) = self.udp.binding_for_socket(id) { self.udp .remove_binding(binding.guest_binding.requested().port(), id); @@ -1745,7 +1724,7 @@ impl Reactor { .get(guest_port) .and_then(|binding| self.sockets.get(&binding.socket_id)) .and_then(|socket| socket.udp_state().ok()) - .and_then(|udp| udp.external_endpoint.as_ref()) + .and_then(|udp| udp.native_endpoint(UdpNativeRoute::External)) .map(|endpoint| endpoint.host_address) }); let _ = response.send(host_address); @@ -1767,6 +1746,10 @@ impl Reactor { let _ = response.send(self.udp.native_endpoints.len()); } #[cfg(test)] + ReactorCommand::UdpNativeEventTokenCount { response } => { + let _ = response.send(self.udp.event_tokens.len()); + } + #[cfg(test)] ReactorCommand::UdpNativeReceiveBufferSize { guest_port, response, @@ -1777,7 +1760,7 @@ impl Reactor { .get(guest_port) .and_then(|binding| self.sockets.get(&binding.socket_id)) .and_then(|socket| socket.udp_state().ok()) - .and_then(|udp| udp.external_endpoint.as_ref()) + .and_then(|udp| udp.native_endpoint(UdpNativeRoute::External)) .map(|endpoint| { sockopt::socket_recv_buffer_size(&endpoint.socket) .map_err(broker_error_from_errno) @@ -1824,12 +1807,16 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(UdpNativeRoute::External) .ok_or(BrokerError::Internal)? .readable = true; if self - .receive_native_udp(socket_id, 1, ReceiveFromFlags::NONE)? + .receive_native_udp( + socket_id, + UdpNativeRoute::External, + 1, + ReceiveFromFlags::NONE, + )? .is_some() { return Err(BrokerError::Internal); @@ -1839,8 +1826,7 @@ impl Reactor { .get(&socket_id) .ok_or(BrokerError::Internal)? .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(UdpNativeRoute::External) .map(|endpoint| endpoint.readable) .ok_or(BrokerError::Internal)?; let head_datagram_bytes = self.udp_native_head_datagram_bytes(socket_id)?; @@ -1998,17 +1984,7 @@ fn zeroed_vec(length: usize) -> BrokerResult> { } fn take_socket_error(socket: &SocketEntry) -> BrokerResult> { - let native_socket = match socket.kind() { - SocketKind::Tcp => Some(&socket.tcp_state()?.socket), - SocketKind::Udp => socket - .udp_state()? - .external_endpoint - .as_ref() - .map(|endpoint| &endpoint.socket), - }; - let Some(native_socket) = native_socket else { - return Ok(None); - }; + let native_socket = &socket.tcp_state()?.socket; match sockopt::socket_error(native_socket) { Ok(Ok(())) => Ok(None), Ok(Err(error)) | Err(error) => socket_operation_error_from_errno(error).map(Some), @@ -2028,30 +2004,11 @@ fn status_socket(socket: &mut SocketEntry) -> BrokerResult .snapshot .lock() .expect("Linux socket snapshot mutex poisoned"); - (socket.connection_status == SocketConnectionStatus::Connected - || socket.kind() == SocketKind::Udp) + socket.connection_status == SocketConnectionStatus::Connected && snapshot.readiness.contains(ReadinessFlags::ERROR) }; let socket_error = if query_socket_error { - let native_error = if socket.kind() == SocketKind::Udp { - socket.udp_state()?.native_error - } else { - UdpNativeErrorState::None - }; - let error = match native_error { - UdpNativeErrorState::Consumed(error) => Some(error), - UdpNativeErrorState::None | UdpNativeErrorState::PendingKernel => { - take_socket_error(socket)? - } - }; - if socket.kind() == SocketKind::Udp { - let udp = socket.udp_state_mut()?; - udp.native_error = UdpNativeErrorState::None; - if let Some(endpoint) = udp.external_endpoint.as_mut() { - endpoint.readable = false; - } - } - error + take_socket_error(socket)? } else { None }; diff --git a/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs b/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs index 215eb59fd..8a5249890 100644 --- a/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs +++ b/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs @@ -1254,6 +1254,71 @@ fn udp_exact_bindings_coexist_and_wildcard_covers_loopback() { assert_eq!(&reply, b"pong"); } +#[test] +fn udp_external_route_slot_reuses_and_retires_its_native_endpoint() { + let provider = Arc::new(LinuxSocketProvider::new(2, 2).unwrap()); + let broker = BrokerCore::new_with_limits( + PolicyEngine::with_unauthenticated_rights(ObjectRights::all()) + .with_socket_policy(SocketPolicy::Ipv4Loopback), + BrokerCoreLimits::new_with_all_limits(4, 0, 2, 2), + provider.clone(), + ) + .unwrap(); + let session = broker + .create_session(CallerCredential::Unauthenticated) + .unwrap(); + let (published, _publications) = channel(); + let (retired, retirements) = channel(); + let readiness = Arc::new(TestReadinessSink { published, retired }); + let socket = create_udp_socket(&session, readiness); + let first = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let second = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + first.set_read_timeout(Some(TEST_TIMEOUT)).unwrap(); + second.set_read_timeout(Some(TEST_TIMEOUT)).unwrap(); + let first_address = socket_address_v4(first.local_addr().unwrap()); + let second_address = socket_address_v4(second.local_addr().unwrap()); + + assert_eq!(provider.reactor.udp_native_endpoint_count(), 0); + assert_eq!(provider.reactor.udp_native_event_token_count(), 0); + assert_eq!( + send_datagram( + &session, + socket, + b"first", + SendFlags::NONE, + Some(first_address), + ), + Ok(SocketOutcome::Completed(5)) + ); + let mut first_payload = [0; 5]; + let (_, first_source) = first.recv_from(&mut first_payload).unwrap(); + assert_eq!(&first_payload, b"first"); + assert_eq!(provider.reactor.udp_native_endpoint_count(), 1); + assert_eq!(provider.reactor.udp_native_event_token_count(), 1); + + assert_eq!( + send_datagram( + &session, + socket, + b"second", + SendFlags::NONE, + Some(second_address), + ), + Ok(SocketOutcome::Completed(6)) + ); + let mut second_payload = [0; 6]; + let (_, second_source) = second.recv_from(&mut second_payload).unwrap(); + assert_eq!(&second_payload, b"second"); + assert_eq!(second_source, first_source); + assert_eq!(provider.reactor.udp_native_endpoint_count(), 1); + assert_eq!(provider.reactor.udp_native_event_token_count(), 1); + + session.close_object_reference(socket).unwrap(); + assert_eq!(retirements.recv_timeout(TEST_TIMEOUT).unwrap(), socket); + assert_eq!(provider.reactor.udp_native_endpoint_count(), 0); + assert_eq!(provider.reactor.udp_native_event_token_count(), 0); +} + #[test] fn udp_endpoint_staging_error_rolls_back_external_peer_reservation() { let provider = Arc::new(LinuxSocketProvider::new(2, 2).unwrap()); @@ -1287,6 +1352,7 @@ fn udp_endpoint_staging_error_rolls_back_external_peer_reservation() { ); assert_eq!(provider.reactor.udp_external_peer_count(), 0); assert_eq!(provider.reactor.udp_native_endpoint_count(), 0); + assert_eq!(provider.reactor.udp_native_event_token_count(), 0); session.close_object_reference(socket).unwrap(); } diff --git a/litebox_broker_platform_linux_userland/src/socket/udp.rs b/litebox_broker_platform_linux_userland/src/socket/udp.rs index 73280d370..6b2d60a54 100644 --- a/litebox_broker_platform_linux_userland/src/socket/udp.rs +++ b/litebox_broker_platform_linux_userland/src/socket/udp.rs @@ -29,8 +29,9 @@ use rustix::net::{ }; use super::{ - Reactor, ReactorReceiveFromOutcome, SocketKind, WAKE_TOKEN, broker_error_from_errno, - local_socket_address, socket_operation_error_from_errno, update_snapshot, zeroed_vec, + PlatformSocketStatus, Reactor, ReactorReceiveFromOutcome, SocketKind, WAKE_TOKEN, + broker_error_from_errno, local_socket_address, socket_operation_error_from_errno, + update_snapshot, zeroed_vec, }; pub(super) const MAX_REJECTED_UDP_DATAGRAMS_PER_COMMAND: usize = 64; @@ -56,6 +57,7 @@ pub(super) struct ReactorUdpState { pub(super) event_tokens: HashMap, pub(super) next_event_token: u64, pub(super) next_endpoint_generation: u64, + next_error_sequence: u64, } #[derive(Default)] @@ -96,6 +98,7 @@ impl Default for ReactorUdpState { event_tokens: HashMap::new(), next_event_token: 1, next_endpoint_generation: 1, + next_error_sequence: 1, } } } @@ -127,10 +130,8 @@ pub(super) struct UdpSocketState { queued_by_source: HashMap, pub(super) peeked_origin: Option, pub(super) next_receive_origin: UdpReceiveOrigin, - pub(super) external_endpoint: Option, + external_endpoint: Option, pub(super) external_peers: HashSet, - native_write_blocked: bool, - pub(super) native_error: UdpNativeErrorState, } impl Default for UdpSocketState { @@ -145,8 +146,6 @@ impl Default for UdpSocketState { next_receive_origin: UdpReceiveOrigin::Guest, external_endpoint: None, external_peers: HashSet::new(), - native_write_blocked: false, - native_error: UdpNativeErrorState::None, } } } @@ -155,8 +154,13 @@ impl Default for UdpSocketState { pub(super) enum UdpNativeErrorState { #[default] None, - PendingKernel, - Consumed(SocketError), + PendingKernel { + sequence: u64, + }, + Consumed { + sequence: u64, + error: SocketError, + }, } impl UdpNativeErrorState { @@ -164,14 +168,32 @@ impl UdpNativeErrorState { !matches!(self, Self::None) } - fn record_kernel(&mut self) { + const fn sequence(self) -> Option { + match self { + Self::None => None, + Self::PendingKernel { sequence } | Self::Consumed { sequence, .. } => Some(sequence), + } + } + + fn record_kernel(&mut self, sequence: u64) { if matches!(self, Self::None) { - *self = Self::PendingKernel; + *self = Self::PendingKernel { sequence }; } } - fn record_consumed(&mut self, error: SocketError) { - *self = Self::Consumed(error); + fn record_consumed(&mut self, sequence: u64, error: SocketError) { + match *self { + Self::None => *self = Self::Consumed { sequence, error }, + Self::PendingKernel { + sequence: pending_sequence, + } => { + *self = Self::Consumed { + sequence: pending_sequence, + error, + }; + } + Self::Consumed { .. } => {} + } } } @@ -182,16 +204,23 @@ struct GuestDatagram { source_session_id: SessionId, } -pub(super) struct ExternalUdpEndpoint { +pub(super) struct UdpNativeEndpoint { pub(super) socket: OwnedFd, + socket_id: u64, + route: UdpNativeRoute, + slot: UdpEndpointSlot, generation: u64, event_token: u64, pub(super) host_address: SocketAddrV4, pub(super) readable: bool, + write_blocked: bool, + error: UdpNativeErrorState, } -#[derive(Clone, Copy)] +#[derive(Clone, Copy, PartialEq, Eq)] pub(super) struct UdpNativeEndpointIdentity { + socket_id: u64, + slot: UdpEndpointSlot, endpoint_generation: u64, } @@ -199,6 +228,7 @@ pub(super) struct UdpNativeEndpointIdentity { pub(super) struct UdpEventTarget { socket_id: u64, endpoint_generation: u64, + slot: UdpEndpointSlot, } #[derive(Clone, Copy, Default)] @@ -210,7 +240,62 @@ pub(super) struct UdpQueueAccounting { #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum UdpReceiveOrigin { Guest, - Native, + External, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum UdpNativeRoute { + External, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum UdpEndpointSlot { + Active(UdpNativeRoute), +} + +const UDP_ACTIVE_ENDPOINT_SLOTS: [UdpEndpointSlot; 1] = + [UdpEndpointSlot::Active(UdpNativeRoute::External)]; + +#[derive(Clone, Copy)] +enum UdpErrorSource { + Endpoint(UdpEndpointSlot), +} + +impl UdpNativeRoute { + const fn receive_origin(self) -> UdpReceiveOrigin { + match self { + Self::External => UdpReceiveOrigin::External, + } + } +} + +impl UdpSocketState { + pub(super) fn native_endpoint(&self, route: UdpNativeRoute) -> Option<&UdpNativeEndpoint> { + match route { + UdpNativeRoute::External => self.external_endpoint.as_ref(), + } + } + + pub(super) fn native_endpoint_mut( + &mut self, + route: UdpNativeRoute, + ) -> Option<&mut UdpNativeEndpoint> { + match route { + UdpNativeRoute::External => self.external_endpoint.as_mut(), + } + } + + fn endpoint(&self, slot: UdpEndpointSlot) -> Option<&UdpNativeEndpoint> { + match slot { + UdpEndpointSlot::Active(route) => self.native_endpoint(route), + } + } + + fn endpoint_mut(&mut self, slot: UdpEndpointSlot) -> Option<&mut UdpNativeEndpoint> { + match slot { + UdpEndpointSlot::Active(route) => self.native_endpoint_mut(route), + } + } } #[derive(Clone, Copy)] @@ -470,19 +555,25 @@ impl Reactor { .pending_error .is_some(); let mut readiness = ReadinessFlags::default(); - if socket.read_shutdown - || !udp.guest_receive_queue.is_empty() - || udp - .external_endpoint - .as_ref() - .is_some_and(|endpoint| endpoint.readable) - { + let native_readable = UDP_ACTIVE_ENDPOINT_SLOTS + .iter() + .filter_map(|slot| udp.endpoint(*slot)) + .any(|endpoint| endpoint.readable); + if socket.read_shutdown || !udp.guest_receive_queue.is_empty() || native_readable { readiness = readiness | ReadinessFlags::READ; } - if !socket.write_shutdown && !udp.native_write_blocked { + let native_write_blocked = UDP_ACTIVE_ENDPOINT_SLOTS + .iter() + .filter_map(|slot| udp.endpoint(*slot)) + .any(|endpoint| endpoint.write_blocked); + if !socket.write_shutdown && !native_write_blocked { readiness = readiness | ReadinessFlags::WRITE; } - if udp.native_error.is_pending() || cached_error { + let native_error = UDP_ACTIVE_ENDPOINT_SLOTS + .iter() + .filter_map(|slot| udp.endpoint(*slot)) + .any(|endpoint| endpoint.error.is_pending()); + if native_error || cached_error { readiness = readiness | ReadinessFlags::ERROR; } Ok(readiness) @@ -497,6 +588,131 @@ impl Reactor { update_snapshot(socket, None, readiness) } + fn oldest_udp_error_source(&self, socket_id: u64) -> BrokerResult> { + let udp = self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()?; + let mut oldest = None; + for slot in UDP_ACTIVE_ENDPOINT_SLOTS { + let Some(sequence) = udp + .endpoint(slot) + .and_then(|endpoint| endpoint.error.sequence()) + else { + continue; + }; + if oldest.is_none_or(|(_, oldest_sequence)| sequence < oldest_sequence) { + oldest = Some((UdpErrorSource::Endpoint(slot), sequence)); + } + } + Ok(oldest.map(|(source, _)| source)) + } + + fn take_udp_error( + &mut self, + socket_id: u64, + source: UdpErrorSource, + ) -> BrokerResult> { + match source { + UdpErrorSource::Endpoint(slot) => { + let (state, was_readable) = { + let endpoint = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state_mut()? + .endpoint_mut(slot) + .ok_or(BrokerError::Internal)?; + let was_readable = endpoint.readable; + endpoint.readable = false; + (std::mem::take(&mut endpoint.error), was_readable) + }; + match state { + UdpNativeErrorState::None => Err(BrokerError::Internal), + UdpNativeErrorState::Consumed { error, .. } => Ok(Some(error)), + UdpNativeErrorState::PendingKernel { .. } => { + let endpoint = self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()? + .endpoint(slot) + .ok_or(BrokerError::Internal)?; + let socket_error = take_udp_socket_error(&endpoint.socket); + if socket_error.is_err() { + let endpoint = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state_mut()? + .endpoint_mut(slot) + .ok_or(BrokerError::Internal)?; + endpoint.error = state; + endpoint.readable = was_readable; + } + socket_error + } + } + } + } + } + + pub(super) fn status_udp_socket( + &mut self, + socket_id: u64, + ) -> BrokerResult { + let (status, local_address, cached_error) = { + let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; + let mut snapshot = socket + .snapshot + .lock() + .expect("Linux socket snapshot mutex poisoned"); + ( + socket.connection_status, + snapshot.local_address, + snapshot.pending_error.take(), + ) + }; + let mut pending_error = cached_error; + while pending_error.is_none() { + let Some(source) = self.oldest_udp_error_source(socket_id)? else { + break; + }; + pending_error = self.take_udp_error(socket_id, source)?; + } + let readiness = self.udp_readiness(socket_id)?; + let socket = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)?; + let _ = update_snapshot(socket, None, readiness); + if let Err(error) = self.rearm_udp_endpoint_if_needed(socket_id) { + if let Some(pending_error) = pending_error { + let socket = self + .sockets + .get_mut(&socket_id) + .expect("UDP status socket disappeared after rearm failure"); + let readiness = { + let mut snapshot = socket + .snapshot + .lock() + .expect("Linux socket snapshot mutex poisoned"); + snapshot.pending_error = Some(pending_error); + snapshot.readiness = snapshot.readiness | ReadinessFlags::ERROR; + snapshot.readiness + }; + let _ = socket.readiness.publish(readiness); + } + return Err(error); + } + Ok(PlatformSocketStatus { + status, + local_address, + pending_error, + }) + } + fn udp_queue_would_drop( &self, receiver_id: u64, @@ -793,6 +1009,14 @@ impl Reactor { Ok((generation, UDP_EVENT_TOKEN_FLAG | token_id)) } + fn next_udp_error_sequence(&mut self) -> BrokerResult { + let sequence = self.udp.next_error_sequence; + self.udp.next_error_sequence = sequence + .checked_add(1) + .ok_or(BrokerError::ResourceExhausted)?; + Ok(sequence) + } + fn udp_port_conflicts( &self, socket_id: u64, @@ -821,9 +1045,14 @@ impl Reactor { pub(super) fn stage_udp_endpoint( &mut self, socket_id: u64, + route: UdpNativeRoute, + slot: UdpEndpointSlot, current_destination: SocketAddrV4, connected_peer: Option, - ) -> BrokerResult> { + ) -> BrokerResult> { + if !matches!(slot, UdpEndpointSlot::Active(active_route) if active_route == route) { + return Err(BrokerError::Internal); + } let read_enabled = !self .sockets .get(&socket_id) @@ -847,7 +1076,12 @@ impl Reactor { ) .map_err(broker_error_from_errno)?; configure_udp_native_receive_buffer(&socket)?; - let wildcard = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0); + let wildcard = SocketAddrV4::new( + match route { + UdpNativeRoute::External => Ipv4Addr::UNSPECIFIED, + }, + 0, + ); loop { match bind(&socket, &wildcard) { Ok(()) => break, @@ -888,6 +1122,8 @@ impl Reactor { .insert( host_address.port(), UdpNativeEndpointIdentity { + socket_id, + slot, endpoint_generation: generation, }, ) @@ -903,6 +1139,7 @@ impl Reactor { UdpEventTarget { socket_id, endpoint_generation: generation, + slot, }, ) .is_some() @@ -920,26 +1157,31 @@ impl Reactor { self.udp.native_endpoints.remove(&host_address.port()); return Err(broker_error_from_errno(error)); } - return Ok(SocketOutcome::Completed(ExternalUdpEndpoint { + return Ok(SocketOutcome::Completed(UdpNativeEndpoint { socket, + socket_id, + route, + slot, generation, event_token, host_address, readable: false, + write_blocked: false, + error: UdpNativeErrorState::None, })); } Ok(SocketOutcome::Failed(SocketError::AddressInUse)) } - pub(super) fn unregister_udp_endpoint(&mut self, endpoint: ExternalUdpEndpoint) { + pub(super) fn unregister_udp_endpoint(&mut self, endpoint: UdpNativeEndpoint) { let _ = epoll::delete(&self.epoll, &endpoint.socket); self.udp.event_tokens.remove(&endpoint.event_token); - if self - .udp - .native_endpoints - .get(&endpoint.host_address.port()) - .is_some_and(|identity| identity.endpoint_generation == endpoint.generation) - { + let identity = UdpNativeEndpointIdentity { + socket_id: endpoint.socket_id, + slot: endpoint.slot, + endpoint_generation: endpoint.generation, + }; + if self.udp.native_endpoints.get(&endpoint.host_address.port()) == Some(&identity) { self.udp .native_endpoints .remove(&endpoint.host_address.port()); @@ -949,6 +1191,7 @@ impl Reactor { pub(super) fn connect_existing_udp_endpoint( &mut self, socket_id: u64, + route: UdpNativeRoute, peer: SocketAddrV4, ) -> core::result::Result, PlatformConnectError> { { @@ -961,8 +1204,7 @@ impl Reactor { let endpoint = socket .udp_state() .map_err(PlatformConnectError::PeerIndeterminate)? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(PlatformConnectError::PeerIndeterminate( BrokerError::Internal, ))?; @@ -992,8 +1234,7 @@ impl Reactor { let endpoint = socket .udp_state() .map_err(PlatformConnectError::PeerIndeterminate)? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(PlatformConnectError::PeerIndeterminate( BrokerError::Internal, ))?; @@ -1012,21 +1253,21 @@ impl Reactor { let udp = socket .udp_state_mut() .map_err(PlatformConnectError::PeerIndeterminate)?; - udp.native_write_blocked = false; - udp.native_error = UdpNativeErrorState::None; - if udp.peeked_origin == Some(UdpReceiveOrigin::Native) { + if udp.peeked_origin == Some(route.receive_origin()) { udp.peeked_origin = None; } - let endpoint = udp.external_endpoint.as_mut().ok_or( + let endpoint = udp.native_endpoint_mut(route).ok_or( PlatformConnectError::PeerIndeterminate(BrokerError::Internal), )?; + endpoint.write_blocked = false; + endpoint.error = UdpNativeErrorState::None; endpoint.readable = false; let host_address = local_socket_address(&endpoint.socket) .map_err(PlatformConnectError::PeerIndeterminate)?; endpoint.host_address = host_address; host_address }; - self.rearm_udp_endpoint(socket_id) + self.rearm_udp_endpoint(socket_id, UdpEndpointSlot::Active(route)) .map_err(PlatformConnectError::PeerIndeterminate)?; return Ok(SocketOutcome::Completed(host_address)); } @@ -1041,25 +1282,88 @@ impl Reactor { pub(super) fn replace_udp_endpoint( &mut self, socket_id: u64, - endpoint: Option, - ) { + route: UdpNativeRoute, + mut endpoint: Option, + ) -> BrokerResult<()> { + let endpoint_matches = endpoint.as_ref().is_none_or(|endpoint| { + endpoint.socket_id == socket_id + && endpoint.route == route + && endpoint.slot == UdpEndpointSlot::Active(route) + }); + let socket_is_udp = self + .sockets + .get(&socket_id) + .is_some_and(|socket| socket.kind() == SocketKind::Udp); + if !endpoint_matches || !socket_is_udp { + if let Some(endpoint) = endpoint.take() { + self.unregister_udp_endpoint(endpoint); + } + return Err(BrokerError::Internal); + } + debug_assert!(endpoint.as_ref().is_none_or(|endpoint| { + endpoint.socket_id == socket_id + && endpoint.route == route + && endpoint.slot == UdpEndpointSlot::Active(route) + })); let old = { let udp = self .sockets .get_mut(&socket_id) - .expect("UDP socket disappeared during endpoint replacement") - .udp_state_mut() - .expect("socket changed kind during UDP endpoint replacement"); - udp.native_write_blocked = false; - udp.native_error = UdpNativeErrorState::None; - if udp.peeked_origin == Some(UdpReceiveOrigin::Native) { + .ok_or(BrokerError::Internal)? + .udp_state_mut()?; + if udp.peeked_origin == Some(route.receive_origin()) { udp.peeked_origin = None; } - std::mem::replace(&mut udp.external_endpoint, endpoint) + match route { + UdpNativeRoute::External => std::mem::replace(&mut udp.external_endpoint, endpoint), + } }; if let Some(old) = old { self.unregister_udp_endpoint(old); } + Ok(()) + } + + pub(super) fn replace_all_udp_endpoints( + &mut self, + socket_id: u64, + replacement: Option, + ) -> BrokerResult<()> { + let replacement_route = replacement.as_ref().map(|endpoint| endpoint.route); + for slot in UDP_ACTIVE_ENDPOINT_SLOTS { + let UdpEndpointSlot::Active(route) = slot; + if Some(route) != replacement_route { + self.replace_udp_endpoint(socket_id, route, None)?; + } + } + if let Some(endpoint) = replacement { + self.replace_udp_endpoint(socket_id, endpoint.route, Some(endpoint))?; + } + Ok(()) + } + + pub(super) fn retain_udp_endpoint( + &mut self, + socket_id: u64, + retained_route: UdpNativeRoute, + ) -> BrokerResult<()> { + if self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()? + .native_endpoint(retained_route) + .is_none() + { + return Err(BrokerError::Internal); + } + for slot in UDP_ACTIVE_ENDPOINT_SLOTS { + let UdpEndpointSlot::Active(route) = slot; + if route != retained_route { + self.replace_udp_endpoint(socket_id, route, None)?; + } + } + Ok(()) } pub(super) fn handle_udp_endpoint_event( @@ -1072,29 +1376,43 @@ impl Reactor { }; let readable_event = events.contains(epoll::EventFlags::IN); let error_event = events.intersects(epoll::EventFlags::ERR | epoll::EventFlags::HUP); + let endpoint_valid = self + .sockets + .get(&target.socket_id) + .and_then(|socket| socket.udp_state().ok()) + .and_then(|udp| udp.endpoint(target.slot)) + .is_some_and(|endpoint| { + endpoint.event_token == event_token + && endpoint.slot == target.slot + && endpoint.generation == target.endpoint_generation + }); + if !endpoint_valid { + return Ok(()); + } + let error_sequence = error_event + .then(|| self.next_udp_error_sequence()) + .transpose()?; { let Some(socket) = self.sockets.get_mut(&target.socket_id) else { return Ok(()); }; let udp = socket.udp_state_mut()?; - let Some(endpoint) = udp.external_endpoint.as_mut() else { + let Some(endpoint) = udp.endpoint_mut(target.slot) else { return Ok(()); }; - if endpoint.generation != target.endpoint_generation { - return Ok(()); - } if events.contains(epoll::EventFlags::OUT) { - udp.native_write_blocked = false; + endpoint.write_blocked = false; } - if error_event { + if let Some(sequence) = error_sequence { endpoint.readable = true; - udp.native_error.record_kernel(); + endpoint.error.record_kernel(sequence); } } if readable_event && !error_event { - self.drain_invalid_udp_ingress(target.socket_id)?; + let UdpEndpointSlot::Active(route) = target.slot; + self.drain_invalid_udp_ingress(target.socket_id, route)?; } - self.rearm_udp_endpoint_if_needed(target.socket_id)?; + self.rearm_udp_endpoint(target.socket_id, target.slot)?; // The cached snapshot remains authoritative if this association cannot // accept a notification. One session's sink must not fail the shared // reactor and every other session using it. @@ -1102,13 +1420,13 @@ impl Reactor { Ok(()) } - fn rearm_udp_endpoint(&mut self, socket_id: u64) -> BrokerResult<()> { + fn rearm_udp_endpoint(&mut self, socket_id: u64, slot: UdpEndpointSlot) -> BrokerResult<()> { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let udp = socket.udp_state()?; - let Some(endpoint) = udp.external_endpoint.as_ref() else { + let Some(endpoint) = udp.endpoint(slot) else { return Ok(()); }; - if udp.native_error.is_pending() { + if endpoint.error.is_pending() { return Ok(()); } epoll::modify( @@ -1118,29 +1436,31 @@ impl Reactor { udp_rearm_events( socket.read_shutdown, endpoint.readable, - udp.native_write_blocked, + endpoint.write_blocked, ), ) .map_err(broker_error_from_errno) } pub(super) fn rearm_udp_endpoint_if_needed(&mut self, socket_id: u64) -> BrokerResult<()> { - let should_rearm = { - let udp = self - .sockets - .get(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state()?; - udp.external_endpoint.as_ref().is_some_and(|endpoint| { - udp_endpoint_needs_rearm( - endpoint.readable, - udp.native_write_blocked, - udp.native_error.is_pending(), - ) - }) - }; - if should_rearm { - self.rearm_udp_endpoint(socket_id)?; + for slot in UDP_ACTIVE_ENDPOINT_SLOTS { + let should_rearm = { + let udp = self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()?; + udp.endpoint(slot).is_some_and(|endpoint| { + udp_endpoint_needs_rearm( + endpoint.readable, + endpoint.write_blocked, + endpoint.error.is_pending(), + ) + }) + }; + if should_rearm { + self.rearm_udp_endpoint(socket_id, slot)?; + } } Ok(()) } @@ -1148,28 +1468,85 @@ impl Reactor { fn udp_native_source_authorized( &self, socket_id: u64, + route: UdpNativeRoute, source_address: SocketAddrV4, ) -> BrokerResult { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; - if socket.read_shutdown || self.is_private_udp_host_endpoint(source_address) { + if socket.read_shutdown { return Ok(false); } let udp = socket.udp_state()?; - Ok(match udp.peer { - Some(ReactorUdpPeer::Guest { .. }) => false, - Some(ReactorUdpPeer::External(peer)) => peer == source_address, - None => udp.external_peers.contains(&source_address), + Ok(match route { + UdpNativeRoute::External => { + if self.is_private_udp_host_endpoint(source_address) { + false + } else { + match udp.peer { + Some(ReactorUdpPeer::Guest { .. }) => false, + Some(ReactorUdpPeer::External(peer)) => peer == source_address, + None => udp.external_peers.contains(&source_address), + } + } + } }) } - fn drain_invalid_udp_ingress(&mut self, socket_id: u64) -> BrokerResult<()> { + fn record_udp_endpoint_error( + &mut self, + socket_id: u64, + slot: UdpEndpointSlot, + error: SocketError, + ) -> BrokerResult<()> { + let sequence = self.next_udp_error_sequence()?; + let endpoint = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state_mut()? + .endpoint_mut(slot) + .ok_or(BrokerError::Internal)?; + endpoint.error.record_consumed(sequence, error); + Ok(()) + } + + fn finish_udp_receive_error( + &mut self, + socket_id: u64, + route: UdpNativeRoute, + error: SocketError, + ) -> BrokerResult> { + let slot = UdpEndpointSlot::Active(route); + { + let endpoint = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state_mut()? + .native_endpoint_mut(route) + .ok_or(BrokerError::Internal)?; + endpoint.readable = false; + endpoint.error = UdpNativeErrorState::None; + } + if let Err(rearm_error) = self.rearm_udp_endpoint(socket_id, slot) { + self.record_udp_endpoint_error(socket_id, slot, error)?; + return Err(rearm_error); + } + let _ = self.publish_udp_readiness(socket_id); + Ok(Some(ReactorReceiveFromOutcome::Failed(error))) + } + + fn drain_invalid_udp_ingress( + &mut self, + socket_id: u64, + route: UdpNativeRoute, + ) -> BrokerResult<()> { + let slot = UdpEndpointSlot::Active(route); if let Some(endpoint) = self .sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) { endpoint.readable = false; } @@ -1178,23 +1555,21 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::PEEK) }; let source_address = match peek { Ok(ReactorReceiveFromOutcome::Received { source_address, .. }) => source_address, Ok(ReactorReceiveFromOutcome::Failed(error)) => { - let udp = self - .sockets + self.record_udp_endpoint_error(socket_id, slot, error)?; + self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? - .udp_state_mut()?; - udp.native_error.record_consumed(error); - if let Some(endpoint) = udp.external_endpoint.as_mut() { - endpoint.readable = true; - } + .udp_state_mut()? + .endpoint_mut(slot) + .ok_or(BrokerError::Internal)? + .readable = true; return Ok(()); } Err(BrokerError::WouldBlock) => { @@ -1203,8 +1578,7 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) { endpoint.readable = false; } @@ -1212,13 +1586,12 @@ impl Reactor { } Err(error) => return Err(error), }; - if self.udp_native_source_authorized(socket_id, source_address)? { + if self.udp_native_source_authorized(socket_id, route, source_address)? { self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)? .readable = true; return Ok(()); @@ -1227,23 +1600,21 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::NONE) }; match consumed { Ok(ReactorReceiveFromOutcome::Received { .. }) => {} Ok(ReactorReceiveFromOutcome::Failed(error)) => { - let udp = self - .sockets + self.record_udp_endpoint_error(socket_id, slot, error)?; + self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? - .udp_state_mut()?; - udp.native_error.record_consumed(error); - if let Some(endpoint) = udp.external_endpoint.as_mut() { - endpoint.readable = true; - } + .udp_state_mut()? + .endpoint_mut(slot) + .ok_or(BrokerError::Internal)? + .readable = true; return Ok(()); } Err(BrokerError::WouldBlock) => return Ok(()), @@ -1254,8 +1625,7 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)? .readable = false; Ok(()) @@ -1264,6 +1634,7 @@ impl Reactor { pub(super) fn send_external_udp( &mut self, socket_id: u64, + slot: UdpEndpointSlot, data: &[u8], address: SocketAddrV4, ) -> BrokerResult> { @@ -1272,8 +1643,7 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .endpoint(slot) .ok_or(BrokerError::Internal)?; match sendto( &endpoint.socket, @@ -1302,12 +1672,9 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)?; let udp = socket.udp_state_mut()?; - let endpoint = udp - .external_endpoint - .as_mut() - .ok_or(BrokerError::Internal)?; - udp.native_write_blocked = true; - if !udp.native_error.is_pending() { + let endpoint = udp.endpoint_mut(slot).ok_or(BrokerError::Internal)?; + endpoint.write_blocked = true; + if !endpoint.error.is_pending() { epoll::modify( &self.epoll, &endpoint.socket, @@ -1369,7 +1736,7 @@ impl Reactor { .ok_or(BrokerError::Internal)? .udp_state_mut()?; udp.peeked_origin = None; - udp.next_receive_origin = UdpReceiveOrigin::Native; + udp.next_receive_origin = UdpReceiveOrigin::External; // The dequeue is committed and the cached snapshot is authoritative // even if this association cannot accept the notification. let _ = self.publish_udp_readiness(socket_id); @@ -1384,16 +1751,17 @@ impl Reactor { pub(super) fn receive_native_udp( &mut self, socket_id: u64, + route: UdpNativeRoute, length: usize, flags: ReceiveFromFlags, ) -> BrokerResult> { + let slot = UdpEndpointSlot::Active(route); let can_receive = self .sockets .get(&socket_id) .ok_or(BrokerError::Internal)? .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(route) .is_some_and(|endpoint| endpoint.readable); if !can_receive { return Ok(None); @@ -1404,8 +1772,7 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, length, ReceiveFromFlags::PEEK) }; @@ -1416,7 +1783,7 @@ impl Reactor { source_address, }) => (data, datagram_length, source_address), Ok(ReactorReceiveFromOutcome::Failed(error)) => { - return Ok(Some(ReactorReceiveFromOutcome::Failed(error))); + return self.finish_udp_receive_error(socket_id, route, error); } Err(BrokerError::WouldBlock) => { let endpoint = self @@ -1424,23 +1791,22 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)?; endpoint.readable = false; - self.rearm_udp_endpoint(socket_id)?; + self.rearm_udp_endpoint(socket_id, slot)?; return Ok(None); } Err(error) => return Err(error), }; - let authorized = self.udp_native_source_authorized(socket_id, source_address)?; + let authorized = self.udp_native_source_authorized(socket_id, route, source_address)?; if authorized { if flags.contains(ReceiveFromFlags::PEEK) { self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .peeked_origin = Some(UdpReceiveOrigin::Native); + .peeked_origin = Some(route.receive_origin()); return Ok(Some(ReactorReceiveFromOutcome::Received { data, datagram_length, @@ -1454,23 +1820,22 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()?; - if udp.native_error.is_pending() { + let endpoint = udp + .native_endpoint_mut(route) + .ok_or(BrokerError::Internal)?; + if endpoint.error.is_pending() { false } else { - udp.external_endpoint - .as_mut() - .ok_or(BrokerError::Internal)? - .readable = false; + endpoint.readable = false; true } }; - if should_rearm && let Err(error) = self.rearm_udp_endpoint(socket_id) { + if should_rearm && let Err(error) = self.rearm_udp_endpoint(socket_id, slot) { self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)? .readable = true; return Err(error); @@ -1479,23 +1844,14 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::NONE) }; match consumed { Ok(ReactorReceiveFromOutcome::Received { .. }) => {} Ok(ReactorReceiveFromOutcome::Failed(error)) => { - self.sockets - .get_mut(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state_mut()? - .external_endpoint - .as_mut() - .ok_or(BrokerError::Internal)? - .readable = true; - return Ok(Some(ReactorReceiveFromOutcome::Failed(error))); + return self.finish_udp_receive_error(socket_id, route, error); } Err(BrokerError::WouldBlock) => { let _ = self.publish_udp_readiness(socket_id); @@ -1506,8 +1862,7 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)? .readable = true; return Err(error); @@ -1524,7 +1879,7 @@ impl Reactor { // the dequeue. Refresh the cached head state so another // authorized datagram keeps READ asserted without requiring // an asynchronous low-to-high publication. - let _ = self.drain_invalid_udp_ingress(socket_id); + let _ = self.drain_invalid_udp_ingress(socket_id, route); // The cached snapshot remains authoritative if notification // delivery fails after irreversible consumption. let _ = self.publish_udp_readiness(socket_id); @@ -1537,13 +1892,12 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(route) .ok_or(BrokerError::Internal)?; match receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::NONE) { Ok(ReactorReceiveFromOutcome::Received { .. }) => {} Ok(ReactorReceiveFromOutcome::Failed(error)) => { - return Ok(Some(ReactorReceiveFromOutcome::Failed(error))); + return self.finish_udp_receive_error(socket_id, route, error); } Err(BrokerError::WouldBlock) => { let endpoint = self @@ -1551,11 +1905,10 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)?; endpoint.readable = false; - self.rearm_udp_endpoint(socket_id)?; + self.rearm_udp_endpoint(socket_id, slot)?; return Ok(None); } Err(error) => return Err(error), @@ -1565,11 +1918,10 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .external_endpoint - .as_mut() + .native_endpoint_mut(route) .ok_or(BrokerError::Internal)? .readable = false; - self.rearm_udp_endpoint(socket_id)?; + self.rearm_udp_endpoint(socket_id, slot)?; Ok(None) } @@ -1578,8 +1930,7 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .external_endpoint - .as_ref() + .native_endpoint(UdpNativeRoute::External) .ok_or(BrokerError::Internal)?; usize::try_from(ioctl_fionread(&endpoint.socket).map_err(broker_error_from_errno)?) .map_err(|_| BrokerError::Internal) @@ -1624,6 +1975,13 @@ fn receive_datagram_fd( } } +fn take_udp_socket_error(socket: &OwnedFd) -> BrokerResult> { + match sockopt::socket_error(socket) { + Ok(Ok(())) => Ok(None), + Ok(Err(error)) | Err(error) => socket_operation_error_from_errno(error).map(Some), + } +} + fn udp_epoll_events(read_enabled: bool, waiting_for_write: bool) -> epoll::EventFlags { let mut events = epoll::EventFlags::ONESHOT; if read_enabled { @@ -1717,15 +2075,34 @@ mod tests { #[test] fn consumed_udp_native_error_preserves_the_exact_error() { - let mut state = UdpNativeErrorState::PendingKernel; - state.record_consumed(SocketError::ConnectionRefused); + let mut state = UdpNativeErrorState::PendingKernel { sequence: 7 }; + state.record_consumed(8, SocketError::ConnectionRefused); assert_eq!( std::mem::take(&mut state), - UdpNativeErrorState::Consumed(SocketError::ConnectionRefused) + UdpNativeErrorState::Consumed { + sequence: 7, + error: SocketError::ConnectionRefused, + } ); assert_eq!(state, UdpNativeErrorState::None); } + #[test] + fn udp_native_error_coalescing_preserves_the_oldest_observation() { + let mut state = UdpNativeErrorState::Consumed { + sequence: 7, + error: SocketError::ConnectionRefused, + }; + state.record_consumed(8, SocketError::Other); + assert_eq!( + state, + UdpNativeErrorState::Consumed { + sequence: 7, + error: SocketError::ConnectionRefused, + } + ); + } + #[test] fn private_udp_host_ports_reject_only_local_addresses() { let port = 49152; From 5dfd2229417dfb6a0eb0db12535fae8c3f6e81b2 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Tue, 18 Aug 2026 15:09:15 -0700 Subject: [PATCH 2/3] Simplify UDP native endpoint ownership Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b5a1a347-37a8-4246-8bbc-306590921475 --- .../src/socket.rs | 82 +-- .../src/socket/tests/udp.rs | 4 +- .../src/socket/udp.rs | 618 ++++++------------ 3 files changed, 228 insertions(+), 476 deletions(-) diff --git a/litebox_broker_platform_linux_userland/src/socket.rs b/litebox_broker_platform_linux_userland/src/socket.rs index dc8e15850..0f0fc0819 100644 --- a/litebox_broker_platform_linux_userland/src/socket.rs +++ b/litebox_broker_platform_linux_userland/src/socket.rs @@ -50,8 +50,8 @@ use tcp::{ PendingGuestConnectionMatch, PendingGuestTcpConnection, }; use udp::{ - ReactorUdpBinding, ReactorUdpPeer, ReactorUdpState, UDP_EVENT_TOKEN_FLAG, UdpEndpointSlot, - UdpNativeRoute, UdpReceiveOrigin, UdpSocketState, is_local_ipv4_address, + ReactorUdpBinding, ReactorUdpPeer, ReactorUdpState, UDP_EVENT_TOKEN_FLAG, UdpReceiveOrigin, + UdpSocketState, is_local_ipv4_address, }; /// Epoll token reserved for the eventfd that wakes the reactor for commands. @@ -664,10 +664,10 @@ impl ReactorClient { } #[cfg(test)] - fn exhaust_udp_endpoint_generation(&self) { + fn exhaust_udp_event_tokens(&self) { let (response, receive) = sync_channel(1); self.commands - .send(ReactorCommand::ExhaustUdpEndpointGeneration { response }) + .send(ReactorCommand::ExhaustUdpEventTokens { response }) .unwrap(); self.signal().unwrap(); receive.recv().unwrap(); @@ -891,7 +891,7 @@ enum ReactorCommand { response: SyncSender>, }, #[cfg(test)] - ExhaustUdpEndpointGeneration { + ExhaustUdpEventTokens { response: SyncSender<()>, }, #[cfg(test)] @@ -1112,8 +1112,6 @@ impl Reactor { } }; let mut reused_host_address = None; - let native_route = UdpNativeRoute::External; - let endpoint_slot = UdpEndpointSlot::Active(native_route); let mut staged_endpoint = match peer { ReactorUdpPeer::Guest { .. } => None, ReactorUdpPeer::External(address) => { @@ -1123,10 +1121,11 @@ impl Reactor { .ok_or(PlatformConnectError::PeerUnchanged(BrokerError::Internal))? .udp_state() .map_err(PlatformConnectError::PeerUnchanged)? - .native_endpoint(native_route) + .native_endpoint + .as_ref() .is_some() { - match self.connect_existing_udp_endpoint(id, native_route, address)? { + match self.connect_existing_udp_endpoint(id, address)? { SocketOutcome::Completed(host_address) => { reused_host_address = Some(host_address); None @@ -1137,13 +1136,7 @@ impl Reactor { } } else { match self - .stage_udp_endpoint( - id, - native_route, - endpoint_slot, - address, - Some(address), - ) + .stage_udp_endpoint(id, address, Some(address)) .map_err(PlatformConnectError::PeerUnchanged)? { SocketOutcome::Completed(endpoint) => Some(endpoint), @@ -1192,10 +1185,7 @@ impl Reactor { } }; if reused_host_address.is_none() { - self.replace_all_udp_endpoints(id, staged_endpoint.take()) - .map_err(PlatformConnectError::PeerIndeterminate)?; - } else { - self.retain_udp_endpoint(id, native_route) + self.replace_udp_endpoint(id, staged_endpoint.take()) .map_err(PlatformConnectError::PeerIndeterminate)?; } self.clear_udp_external_peers(id) @@ -1268,8 +1258,6 @@ impl Reactor { self.enqueue_guest_datagram(id, socket_generation, data) } ReactorUdpPeer::External(address) => { - let native_route = UdpNativeRoute::External; - let endpoint_slot = UdpEndpointSlot::Active(native_route); let external_peer_added = if authorize_external_reply { self.reserve_udp_external_peer(id, address)? } else { @@ -1280,16 +1268,11 @@ impl Reactor { .get(&id) .ok_or(BrokerError::Internal)? .udp_state()? - .native_endpoint(native_route) + .native_endpoint + .as_ref() .is_none() { - let endpoint = match self.stage_udp_endpoint( - id, - native_route, - endpoint_slot, - address, - None, - ) { + let endpoint = match self.stage_udp_endpoint(id, address, None) { Ok(SocketOutcome::Completed(endpoint)) => endpoint, Ok(SocketOutcome::Failed(error)) => { if external_peer_added { @@ -1304,9 +1287,9 @@ impl Reactor { return Err(error); } }; - self.replace_udp_endpoint(id, native_route, Some(endpoint))?; + self.replace_udp_endpoint(id, Some(endpoint))?; } - let outcome = self.send_external_udp(id, endpoint_slot, data, address); + let outcome = self.send_external_udp(id, data, address); if !matches!(outcome, Ok(SocketOutcome::Completed(_))) && external_peer_added { self.remove_udp_external_peer(id, address); } @@ -1349,10 +1332,10 @@ impl Reactor { socket.write_shutdown |= shut_write; if shut_read { let udp = socket.udp_state_mut()?; - if udp.peeked_origin == Some(UdpReceiveOrigin::External) { + if udp.peeked_origin == Some(UdpReceiveOrigin::Native) { udp.peeked_origin = None; } - if let Some(endpoint) = udp.native_endpoint_mut(UdpNativeRoute::External) { + if let Some(endpoint) = udp.native_endpoint.as_mut() { endpoint.readable = false; } } @@ -1411,15 +1394,13 @@ impl Reactor { .next_receive_origin; let first = pinned.unwrap_or(next); let second = match first { - UdpReceiveOrigin::Guest => UdpReceiveOrigin::External, - UdpReceiveOrigin::External => UdpReceiveOrigin::Guest, + UdpReceiveOrigin::Guest => UdpReceiveOrigin::Native, + UdpReceiveOrigin::Native => UdpReceiveOrigin::Guest, }; for origin in [first, second] { let outcome = match origin { UdpReceiveOrigin::Guest => self.receive_guest_udp(id, length, flags)?, - UdpReceiveOrigin::External => { - self.receive_native_udp(id, UdpNativeRoute::External, length, flags)? - } + UdpReceiveOrigin::Native => self.receive_native_udp(id, length, flags)?, }; if let Some(outcome) = outcome { return Ok(outcome); @@ -1442,7 +1423,7 @@ impl Reactor { }; if kind == SocketKind::Udp { let _ = self.clear_udp_receive_queue(id); - let _ = self.replace_all_udp_endpoints(id, None); + let _ = self.replace_udp_endpoint(id, None); if let Some(binding) = self.udp.binding_for_socket(id) { self.udp .remove_binding(binding.guest_binding.requested().port(), id); @@ -1724,7 +1705,7 @@ impl Reactor { .get(guest_port) .and_then(|binding| self.sockets.get(&binding.socket_id)) .and_then(|socket| socket.udp_state().ok()) - .and_then(|udp| udp.native_endpoint(UdpNativeRoute::External)) + .and_then(|udp| udp.native_endpoint.as_ref()) .map(|endpoint| endpoint.host_address) }); let _ = response.send(host_address); @@ -1760,7 +1741,7 @@ impl Reactor { .get(guest_port) .and_then(|binding| self.sockets.get(&binding.socket_id)) .and_then(|socket| socket.udp_state().ok()) - .and_then(|udp| udp.native_endpoint(UdpNativeRoute::External)) + .and_then(|udp| udp.native_endpoint.as_ref()) .map(|endpoint| { sockopt::socket_recv_buffer_size(&endpoint.socket) .map_err(broker_error_from_errno) @@ -1807,16 +1788,12 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(UdpNativeRoute::External) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = true; if self - .receive_native_udp( - socket_id, - UdpNativeRoute::External, - 1, - ReceiveFromFlags::NONE, - )? + .receive_native_udp(socket_id, 1, ReceiveFromFlags::NONE)? .is_some() { return Err(BrokerError::Internal); @@ -1826,7 +1803,8 @@ impl Reactor { .get(&socket_id) .ok_or(BrokerError::Internal)? .udp_state()? - .native_endpoint(UdpNativeRoute::External) + .native_endpoint + .as_ref() .map(|endpoint| endpoint.readable) .ok_or(BrokerError::Internal)?; let head_datagram_bytes = self.udp_native_head_datagram_bytes(socket_id)?; @@ -1835,8 +1813,8 @@ impl Reactor { let _ = response.send(outcome); } #[cfg(test)] - ReactorCommand::ExhaustUdpEndpointGeneration { response } => { - self.udp.next_endpoint_generation = u64::MAX; + ReactorCommand::ExhaustUdpEventTokens { response } => { + self.udp.next_event_token = UDP_EVENT_TOKEN_FLAG - 1; let _ = response.send(()); } #[cfg(test)] diff --git a/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs b/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs index 8a5249890..a02e3382f 100644 --- a/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs +++ b/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs @@ -1255,7 +1255,7 @@ fn udp_exact_bindings_coexist_and_wildcard_covers_loopback() { } #[test] -fn udp_external_route_slot_reuses_and_retires_its_native_endpoint() { +fn udp_native_endpoint_is_reused_and_retired() { let provider = Arc::new(LinuxSocketProvider::new(2, 2).unwrap()); let broker = BrokerCore::new_with_limits( PolicyEngine::with_unauthenticated_rights(ObjectRights::all()) @@ -1339,7 +1339,7 @@ fn udp_endpoint_staging_error_rolls_back_external_peer_reservation() { let external = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); let external_address = socket_address_v4(external.local_addr().unwrap()); - provider.reactor.exhaust_udp_endpoint_generation(); + provider.reactor.exhaust_udp_event_tokens(); assert_eq!( send_datagram( &session, diff --git a/litebox_broker_platform_linux_userland/src/socket/udp.rs b/litebox_broker_platform_linux_userland/src/socket/udp.rs index 6b2d60a54..974284ecf 100644 --- a/litebox_broker_platform_linux_userland/src/socket/udp.rs +++ b/litebox_broker_platform_linux_userland/src/socket/udp.rs @@ -49,15 +49,13 @@ pub(super) const UDP_EVENT_TOKEN_FLAG: u64 = 1 << 63; /// Reactor-wide UDP namespace, native endpoint, and queue-accounting state. pub(super) struct ReactorUdpState { pub(super) bindings: ReactorUdpBindings, - pub(super) native_endpoints: HashMap, + pub(super) native_endpoints: HashSet, pub(super) external_peer_count: usize, pub(super) queued_datagrams: usize, pub(super) queued_bytes: usize, pub(super) queued_by_source: HashMap, pub(super) event_tokens: HashMap, pub(super) next_event_token: u64, - pub(super) next_endpoint_generation: u64, - next_error_sequence: u64, } #[derive(Default)] @@ -90,21 +88,19 @@ impl Default for ReactorUdpState { fn default() -> Self { Self { bindings: ReactorUdpBindings::default(), - native_endpoints: HashMap::new(), + native_endpoints: HashSet::new(), external_peer_count: 0, queued_datagrams: 0, queued_bytes: 0, queued_by_source: HashMap::new(), event_tokens: HashMap::new(), next_event_token: 1, - next_endpoint_generation: 1, - next_error_sequence: 1, } } } impl ReactorUdpState { - /// Clears live registrations and accounting without reusing endpoint IDs. + /// Clears live registrations and accounting without reusing event tokens. pub(super) fn clear_live_state(&mut self) { self.bindings.clear(); self.native_endpoints.clear(); @@ -130,7 +126,7 @@ pub(super) struct UdpSocketState { queued_by_source: HashMap, pub(super) peeked_origin: Option, pub(super) next_receive_origin: UdpReceiveOrigin, - external_endpoint: Option, + pub(super) native_endpoint: Option, pub(super) external_peers: HashSet, } @@ -144,7 +140,7 @@ impl Default for UdpSocketState { queued_by_source: HashMap::new(), peeked_origin: None, next_receive_origin: UdpReceiveOrigin::Guest, - external_endpoint: None, + native_endpoint: None, external_peers: HashSet::new(), } } @@ -154,13 +150,8 @@ impl Default for UdpSocketState { pub(super) enum UdpNativeErrorState { #[default] None, - PendingKernel { - sequence: u64, - }, - Consumed { - sequence: u64, - error: SocketError, - }, + PendingKernel, + Consumed(SocketError), } impl UdpNativeErrorState { @@ -168,31 +159,16 @@ impl UdpNativeErrorState { !matches!(self, Self::None) } - const fn sequence(self) -> Option { - match self { - Self::None => None, - Self::PendingKernel { sequence } | Self::Consumed { sequence, .. } => Some(sequence), - } - } - - fn record_kernel(&mut self, sequence: u64) { + fn record_kernel(&mut self) { if matches!(self, Self::None) { - *self = Self::PendingKernel { sequence }; + *self = Self::PendingKernel; } } - fn record_consumed(&mut self, sequence: u64, error: SocketError) { + fn record_consumed(&mut self, error: SocketError) { match *self { - Self::None => *self = Self::Consumed { sequence, error }, - Self::PendingKernel { - sequence: pending_sequence, - } => { - *self = Self::Consumed { - sequence: pending_sequence, - error, - }; - } - Self::Consumed { .. } => {} + Self::None | Self::PendingKernel => *self = Self::Consumed(error), + Self::Consumed(_) => {} } } } @@ -207,9 +183,6 @@ struct GuestDatagram { pub(super) struct UdpNativeEndpoint { pub(super) socket: OwnedFd, socket_id: u64, - route: UdpNativeRoute, - slot: UdpEndpointSlot, - generation: u64, event_token: u64, pub(super) host_address: SocketAddrV4, pub(super) readable: bool, @@ -217,18 +190,9 @@ pub(super) struct UdpNativeEndpoint { error: UdpNativeErrorState, } -#[derive(Clone, Copy, PartialEq, Eq)] -pub(super) struct UdpNativeEndpointIdentity { - socket_id: u64, - slot: UdpEndpointSlot, - endpoint_generation: u64, -} - #[derive(Clone, Copy)] pub(super) struct UdpEventTarget { socket_id: u64, - endpoint_generation: u64, - slot: UdpEndpointSlot, } #[derive(Clone, Copy, Default)] @@ -240,62 +204,7 @@ pub(super) struct UdpQueueAccounting { #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum UdpReceiveOrigin { Guest, - External, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(super) enum UdpNativeRoute { - External, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(super) enum UdpEndpointSlot { - Active(UdpNativeRoute), -} - -const UDP_ACTIVE_ENDPOINT_SLOTS: [UdpEndpointSlot; 1] = - [UdpEndpointSlot::Active(UdpNativeRoute::External)]; - -#[derive(Clone, Copy)] -enum UdpErrorSource { - Endpoint(UdpEndpointSlot), -} - -impl UdpNativeRoute { - const fn receive_origin(self) -> UdpReceiveOrigin { - match self { - Self::External => UdpReceiveOrigin::External, - } - } -} - -impl UdpSocketState { - pub(super) fn native_endpoint(&self, route: UdpNativeRoute) -> Option<&UdpNativeEndpoint> { - match route { - UdpNativeRoute::External => self.external_endpoint.as_ref(), - } - } - - pub(super) fn native_endpoint_mut( - &mut self, - route: UdpNativeRoute, - ) -> Option<&mut UdpNativeEndpoint> { - match route { - UdpNativeRoute::External => self.external_endpoint.as_mut(), - } - } - - fn endpoint(&self, slot: UdpEndpointSlot) -> Option<&UdpNativeEndpoint> { - match slot { - UdpEndpointSlot::Active(route) => self.native_endpoint(route), - } - } - - fn endpoint_mut(&mut self, slot: UdpEndpointSlot) -> Option<&mut UdpNativeEndpoint> { - match slot { - UdpEndpointSlot::Active(route) => self.native_endpoint_mut(route), - } - } + Native, } #[derive(Clone, Copy)] @@ -394,7 +303,7 @@ impl ReactorUdpState { } pub(super) fn is_private_host_port(&self, port: u16) -> bool { - self.native_endpoints.contains_key(&port) + self.native_endpoints.contains(&port) } } @@ -555,24 +464,24 @@ impl Reactor { .pending_error .is_some(); let mut readiness = ReadinessFlags::default(); - let native_readable = UDP_ACTIVE_ENDPOINT_SLOTS - .iter() - .filter_map(|slot| udp.endpoint(*slot)) - .any(|endpoint| endpoint.readable); + let native_readable = udp + .native_endpoint + .as_ref() + .is_some_and(|endpoint| endpoint.readable); if socket.read_shutdown || !udp.guest_receive_queue.is_empty() || native_readable { readiness = readiness | ReadinessFlags::READ; } - let native_write_blocked = UDP_ACTIVE_ENDPOINT_SLOTS - .iter() - .filter_map(|slot| udp.endpoint(*slot)) - .any(|endpoint| endpoint.write_blocked); + let native_write_blocked = udp + .native_endpoint + .as_ref() + .is_some_and(|endpoint| endpoint.write_blocked); if !socket.write_shutdown && !native_write_blocked { readiness = readiness | ReadinessFlags::WRITE; } - let native_error = UDP_ACTIVE_ENDPOINT_SLOTS - .iter() - .filter_map(|slot| udp.endpoint(*slot)) - .any(|endpoint| endpoint.error.is_pending()); + let native_error = udp + .native_endpoint + .as_ref() + .is_some_and(|endpoint| endpoint.error.is_pending()); if native_error || cached_error { readiness = readiness | ReadinessFlags::ERROR; } @@ -588,72 +497,46 @@ impl Reactor { update_snapshot(socket, None, readiness) } - fn oldest_udp_error_source(&self, socket_id: u64) -> BrokerResult> { - let udp = self - .sockets - .get(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state()?; - let mut oldest = None; - for slot in UDP_ACTIVE_ENDPOINT_SLOTS { - let Some(sequence) = udp - .endpoint(slot) - .and_then(|endpoint| endpoint.error.sequence()) - else { - continue; - }; - if oldest.is_none_or(|(_, oldest_sequence)| sequence < oldest_sequence) { - oldest = Some((UdpErrorSource::Endpoint(slot), sequence)); - } - } - Ok(oldest.map(|(source, _)| source)) - } - - fn take_udp_error( - &mut self, - socket_id: u64, - source: UdpErrorSource, - ) -> BrokerResult> { - match source { - UdpErrorSource::Endpoint(slot) => { - let (state, was_readable) = { + fn take_udp_error(&mut self, socket_id: u64) -> BrokerResult> { + let (state, was_readable) = { + let endpoint = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state_mut()? + .native_endpoint + .as_mut() + .ok_or(BrokerError::Internal)?; + let was_readable = endpoint.readable; + endpoint.readable = false; + (std::mem::take(&mut endpoint.error), was_readable) + }; + match state { + UdpNativeErrorState::None => Err(BrokerError::Internal), + UdpNativeErrorState::Consumed(error) => Ok(Some(error)), + UdpNativeErrorState::PendingKernel => { + let endpoint = self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()? + .native_endpoint + .as_ref() + .ok_or(BrokerError::Internal)?; + let socket_error = take_udp_socket_error(&endpoint.socket); + if socket_error.is_err() { let endpoint = self .sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .endpoint_mut(slot) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)?; - let was_readable = endpoint.readable; - endpoint.readable = false; - (std::mem::take(&mut endpoint.error), was_readable) - }; - match state { - UdpNativeErrorState::None => Err(BrokerError::Internal), - UdpNativeErrorState::Consumed { error, .. } => Ok(Some(error)), - UdpNativeErrorState::PendingKernel { .. } => { - let endpoint = self - .sockets - .get(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state()? - .endpoint(slot) - .ok_or(BrokerError::Internal)?; - let socket_error = take_udp_socket_error(&endpoint.socket); - if socket_error.is_err() { - let endpoint = self - .sockets - .get_mut(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state_mut()? - .endpoint_mut(slot) - .ok_or(BrokerError::Internal)?; - endpoint.error = state; - endpoint.readable = was_readable; - } - socket_error - } + endpoint.error = state; + endpoint.readable = was_readable; } + socket_error } } } @@ -675,11 +558,16 @@ impl Reactor { ) }; let mut pending_error = cached_error; - while pending_error.is_none() { - let Some(source) = self.oldest_udp_error_source(socket_id)? else { - break; - }; - pending_error = self.take_udp_error(socket_id, source)?; + let native_error_pending = self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()? + .native_endpoint + .as_ref() + .is_some_and(|endpoint| endpoint.error.is_pending()); + if pending_error.is_none() && native_error_pending { + pending_error = self.take_udp_error(socket_id)?; } let readiness = self.udp_readiness(socket_id)?; let socket = self @@ -996,25 +884,13 @@ impl Reactor { Ok(()) } - fn next_udp_endpoint_ids(&mut self) -> BrokerResult<(u64, u64)> { - let generation = self.udp.next_endpoint_generation; - self.udp.next_endpoint_generation = generation - .checked_add(1) - .ok_or(BrokerError::ResourceExhausted)?; + fn next_udp_event_token(&mut self) -> BrokerResult { let token_id = self.udp.next_event_token; self.udp.next_event_token = token_id .checked_add(1) .filter(|token| *token < UDP_EVENT_TOKEN_FLAG) .ok_or(BrokerError::ResourceExhausted)?; - Ok((generation, UDP_EVENT_TOKEN_FLAG | token_id)) - } - - fn next_udp_error_sequence(&mut self) -> BrokerResult { - let sequence = self.udp.next_error_sequence; - self.udp.next_error_sequence = sequence - .checked_add(1) - .ok_or(BrokerError::ResourceExhausted)?; - Ok(sequence) + Ok(UDP_EVENT_TOKEN_FLAG | token_id) } fn udp_port_conflicts( @@ -1027,7 +903,7 @@ impl Reactor { return Err(BrokerError::Internal); } Ok(udp_address_matches_host_port(current_destination, port) - || self.udp.native_endpoints.contains_key(&port) + || self.udp.native_endpoints.contains(&port) || self.sockets.values().any(|socket| { socket.udp_state().is_ok_and(|udp| { udp.external_peers @@ -1045,14 +921,9 @@ impl Reactor { pub(super) fn stage_udp_endpoint( &mut self, socket_id: u64, - route: UdpNativeRoute, - slot: UdpEndpointSlot, current_destination: SocketAddrV4, connected_peer: Option, ) -> BrokerResult> { - if !matches!(slot, UdpEndpointSlot::Active(active_route) if active_route == route) { - return Err(BrokerError::Internal); - } let read_enabled = !self .sockets .get(&socket_id) @@ -1076,12 +947,7 @@ impl Reactor { ) .map_err(broker_error_from_errno)?; configure_udp_native_receive_buffer(&socket)?; - let wildcard = SocketAddrV4::new( - match route { - UdpNativeRoute::External => Ipv4Addr::UNSPECIFIED, - }, - 0, - ); + let wildcard = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0); loop { match bind(&socket, &wildcard) { Ok(()) => break, @@ -1115,33 +981,14 @@ impl Reactor { } else { host_address }; - let (generation, event_token) = self.next_udp_endpoint_ids()?; - if self - .udp - .native_endpoints - .insert( - host_address.port(), - UdpNativeEndpointIdentity { - socket_id, - slot, - endpoint_generation: generation, - }, - ) - .is_some() - { + let event_token = self.next_udp_event_token()?; + if !self.udp.native_endpoints.insert(host_address.port()) { return Err(BrokerError::Internal); } if self .udp .event_tokens - .insert( - event_token, - UdpEventTarget { - socket_id, - endpoint_generation: generation, - slot, - }, - ) + .insert(event_token, UdpEventTarget { socket_id }) .is_some() { self.udp.native_endpoints.remove(&host_address.port()); @@ -1160,9 +1007,6 @@ impl Reactor { return Ok(SocketOutcome::Completed(UdpNativeEndpoint { socket, socket_id, - route, - slot, - generation, event_token, host_address, readable: false, @@ -1176,22 +1020,14 @@ impl Reactor { pub(super) fn unregister_udp_endpoint(&mut self, endpoint: UdpNativeEndpoint) { let _ = epoll::delete(&self.epoll, &endpoint.socket); self.udp.event_tokens.remove(&endpoint.event_token); - let identity = UdpNativeEndpointIdentity { - socket_id: endpoint.socket_id, - slot: endpoint.slot, - endpoint_generation: endpoint.generation, - }; - if self.udp.native_endpoints.get(&endpoint.host_address.port()) == Some(&identity) { - self.udp - .native_endpoints - .remove(&endpoint.host_address.port()); - } + self.udp + .native_endpoints + .remove(&endpoint.host_address.port()); } pub(super) fn connect_existing_udp_endpoint( &mut self, socket_id: u64, - route: UdpNativeRoute, peer: SocketAddrV4, ) -> core::result::Result, PlatformConnectError> { { @@ -1204,7 +1040,8 @@ impl Reactor { let endpoint = socket .udp_state() .map_err(PlatformConnectError::PeerIndeterminate)? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(PlatformConnectError::PeerIndeterminate( BrokerError::Internal, ))?; @@ -1234,7 +1071,8 @@ impl Reactor { let endpoint = socket .udp_state() .map_err(PlatformConnectError::PeerIndeterminate)? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(PlatformConnectError::PeerIndeterminate( BrokerError::Internal, ))?; @@ -1253,10 +1091,10 @@ impl Reactor { let udp = socket .udp_state_mut() .map_err(PlatformConnectError::PeerIndeterminate)?; - if udp.peeked_origin == Some(route.receive_origin()) { + if udp.peeked_origin == Some(UdpReceiveOrigin::Native) { udp.peeked_origin = None; } - let endpoint = udp.native_endpoint_mut(route).ok_or( + let endpoint = udp.native_endpoint.as_mut().ok_or( PlatformConnectError::PeerIndeterminate(BrokerError::Internal), )?; endpoint.write_blocked = false; @@ -1267,7 +1105,7 @@ impl Reactor { endpoint.host_address = host_address; host_address }; - self.rearm_udp_endpoint(socket_id, UdpEndpointSlot::Active(route)) + self.rearm_udp_endpoint(socket_id) .map_err(PlatformConnectError::PeerIndeterminate)?; return Ok(SocketOutcome::Completed(host_address)); } @@ -1282,14 +1120,11 @@ impl Reactor { pub(super) fn replace_udp_endpoint( &mut self, socket_id: u64, - route: UdpNativeRoute, mut endpoint: Option, ) -> BrokerResult<()> { - let endpoint_matches = endpoint.as_ref().is_none_or(|endpoint| { - endpoint.socket_id == socket_id - && endpoint.route == route - && endpoint.slot == UdpEndpointSlot::Active(route) - }); + let endpoint_matches = endpoint + .as_ref() + .is_none_or(|endpoint| endpoint.socket_id == socket_id); let socket_is_udp = self .sockets .get(&socket_id) @@ -1300,23 +1135,21 @@ impl Reactor { } return Err(BrokerError::Internal); } - debug_assert!(endpoint.as_ref().is_none_or(|endpoint| { - endpoint.socket_id == socket_id - && endpoint.route == route - && endpoint.slot == UdpEndpointSlot::Active(route) - })); + debug_assert!( + endpoint + .as_ref() + .is_none_or(|endpoint| endpoint.socket_id == socket_id) + ); let old = { let udp = self .sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()?; - if udp.peeked_origin == Some(route.receive_origin()) { + if udp.peeked_origin == Some(UdpReceiveOrigin::Native) { udp.peeked_origin = None; } - match route { - UdpNativeRoute::External => std::mem::replace(&mut udp.external_endpoint, endpoint), - } + std::mem::replace(&mut udp.native_endpoint, endpoint) }; if let Some(old) = old { self.unregister_udp_endpoint(old); @@ -1324,48 +1157,6 @@ impl Reactor { Ok(()) } - pub(super) fn replace_all_udp_endpoints( - &mut self, - socket_id: u64, - replacement: Option, - ) -> BrokerResult<()> { - let replacement_route = replacement.as_ref().map(|endpoint| endpoint.route); - for slot in UDP_ACTIVE_ENDPOINT_SLOTS { - let UdpEndpointSlot::Active(route) = slot; - if Some(route) != replacement_route { - self.replace_udp_endpoint(socket_id, route, None)?; - } - } - if let Some(endpoint) = replacement { - self.replace_udp_endpoint(socket_id, endpoint.route, Some(endpoint))?; - } - Ok(()) - } - - pub(super) fn retain_udp_endpoint( - &mut self, - socket_id: u64, - retained_route: UdpNativeRoute, - ) -> BrokerResult<()> { - if self - .sockets - .get(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state()? - .native_endpoint(retained_route) - .is_none() - { - return Err(BrokerError::Internal); - } - for slot in UDP_ACTIVE_ENDPOINT_SLOTS { - let UdpEndpointSlot::Active(route) = slot; - if route != retained_route { - self.replace_udp_endpoint(socket_id, route, None)?; - } - } - Ok(()) - } - pub(super) fn handle_udp_endpoint_event( &mut self, event_token: u64, @@ -1380,39 +1171,31 @@ impl Reactor { .sockets .get(&target.socket_id) .and_then(|socket| socket.udp_state().ok()) - .and_then(|udp| udp.endpoint(target.slot)) - .is_some_and(|endpoint| { - endpoint.event_token == event_token - && endpoint.slot == target.slot - && endpoint.generation == target.endpoint_generation - }); + .and_then(|udp| udp.native_endpoint.as_ref()) + .is_some_and(|endpoint| endpoint.event_token == event_token); if !endpoint_valid { return Ok(()); } - let error_sequence = error_event - .then(|| self.next_udp_error_sequence()) - .transpose()?; { let Some(socket) = self.sockets.get_mut(&target.socket_id) else { return Ok(()); }; let udp = socket.udp_state_mut()?; - let Some(endpoint) = udp.endpoint_mut(target.slot) else { + let Some(endpoint) = udp.native_endpoint.as_mut() else { return Ok(()); }; if events.contains(epoll::EventFlags::OUT) { endpoint.write_blocked = false; } - if let Some(sequence) = error_sequence { + if error_event { endpoint.readable = true; - endpoint.error.record_kernel(sequence); + endpoint.error.record_kernel(); } } if readable_event && !error_event { - let UdpEndpointSlot::Active(route) = target.slot; - self.drain_invalid_udp_ingress(target.socket_id, route)?; + self.drain_invalid_udp_ingress(target.socket_id)?; } - self.rearm_udp_endpoint(target.socket_id, target.slot)?; + self.rearm_udp_endpoint(target.socket_id)?; // The cached snapshot remains authoritative if this association cannot // accept a notification. One session's sink must not fail the shared // reactor and every other session using it. @@ -1420,10 +1203,10 @@ impl Reactor { Ok(()) } - fn rearm_udp_endpoint(&mut self, socket_id: u64, slot: UdpEndpointSlot) -> BrokerResult<()> { + fn rearm_udp_endpoint(&mut self, socket_id: u64) -> BrokerResult<()> { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let udp = socket.udp_state()?; - let Some(endpoint) = udp.endpoint(slot) else { + let Some(endpoint) = udp.native_endpoint.as_ref() else { return Ok(()); }; if endpoint.error.is_pending() { @@ -1443,24 +1226,22 @@ impl Reactor { } pub(super) fn rearm_udp_endpoint_if_needed(&mut self, socket_id: u64) -> BrokerResult<()> { - for slot in UDP_ACTIVE_ENDPOINT_SLOTS { - let should_rearm = { - let udp = self - .sockets - .get(&socket_id) - .ok_or(BrokerError::Internal)? - .udp_state()?; - udp.endpoint(slot).is_some_and(|endpoint| { - udp_endpoint_needs_rearm( - endpoint.readable, - endpoint.write_blocked, - endpoint.error.is_pending(), - ) - }) - }; - if should_rearm { - self.rearm_udp_endpoint(socket_id, slot)?; - } + let should_rearm = { + let udp = self + .sockets + .get(&socket_id) + .ok_or(BrokerError::Internal)? + .udp_state()?; + udp.native_endpoint.as_ref().is_some_and(|endpoint| { + udp_endpoint_needs_rearm( + endpoint.readable, + endpoint.write_blocked, + endpoint.error.is_pending(), + ) + }) + }; + if should_rearm { + self.rearm_udp_endpoint(socket_id)?; } Ok(()) } @@ -1468,7 +1249,6 @@ impl Reactor { fn udp_native_source_authorized( &self, socket_id: u64, - route: UdpNativeRoute, source_address: SocketAddrV4, ) -> BrokerResult { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; @@ -1476,17 +1256,13 @@ impl Reactor { return Ok(false); } let udp = socket.udp_state()?; - Ok(match route { - UdpNativeRoute::External => { - if self.is_private_udp_host_endpoint(source_address) { - false - } else { - match udp.peer { - Some(ReactorUdpPeer::Guest { .. }) => false, - Some(ReactorUdpPeer::External(peer)) => peer == source_address, - None => udp.external_peers.contains(&source_address), - } - } + Ok(if self.is_private_udp_host_endpoint(source_address) { + false + } else { + match udp.peer { + Some(ReactorUdpPeer::Guest { .. }) => false, + Some(ReactorUdpPeer::External(peer)) => peer == source_address, + None => udp.external_peers.contains(&source_address), } }) } @@ -1494,59 +1270,53 @@ impl Reactor { fn record_udp_endpoint_error( &mut self, socket_id: u64, - slot: UdpEndpointSlot, error: SocketError, ) -> BrokerResult<()> { - let sequence = self.next_udp_error_sequence()?; let endpoint = self .sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .endpoint_mut(slot) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)?; - endpoint.error.record_consumed(sequence, error); + endpoint.error.record_consumed(error); Ok(()) } fn finish_udp_receive_error( &mut self, socket_id: u64, - route: UdpNativeRoute, error: SocketError, ) -> BrokerResult> { - let slot = UdpEndpointSlot::Active(route); { let endpoint = self .sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)?; endpoint.readable = false; endpoint.error = UdpNativeErrorState::None; } - if let Err(rearm_error) = self.rearm_udp_endpoint(socket_id, slot) { - self.record_udp_endpoint_error(socket_id, slot, error)?; + if let Err(rearm_error) = self.rearm_udp_endpoint(socket_id) { + self.record_udp_endpoint_error(socket_id, error)?; return Err(rearm_error); } let _ = self.publish_udp_readiness(socket_id); Ok(Some(ReactorReceiveFromOutcome::Failed(error))) } - fn drain_invalid_udp_ingress( - &mut self, - socket_id: u64, - route: UdpNativeRoute, - ) -> BrokerResult<()> { - let slot = UdpEndpointSlot::Active(route); + fn drain_invalid_udp_ingress(&mut self, socket_id: u64) -> BrokerResult<()> { if let Some(endpoint) = self .sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() { endpoint.readable = false; } @@ -1555,19 +1325,21 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::PEEK) }; let source_address = match peek { Ok(ReactorReceiveFromOutcome::Received { source_address, .. }) => source_address, Ok(ReactorReceiveFromOutcome::Failed(error)) => { - self.record_udp_endpoint_error(socket_id, slot, error)?; + self.record_udp_endpoint_error(socket_id, error)?; self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .endpoint_mut(slot) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = true; return Ok(()); @@ -1578,7 +1350,8 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() { endpoint.readable = false; } @@ -1586,12 +1359,13 @@ impl Reactor { } Err(error) => return Err(error), }; - if self.udp_native_source_authorized(socket_id, route, source_address)? { + if self.udp_native_source_authorized(socket_id, source_address)? { self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = true; return Ok(()); @@ -1600,19 +1374,21 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::NONE) }; match consumed { Ok(ReactorReceiveFromOutcome::Received { .. }) => {} Ok(ReactorReceiveFromOutcome::Failed(error)) => { - self.record_udp_endpoint_error(socket_id, slot, error)?; + self.record_udp_endpoint_error(socket_id, error)?; self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .endpoint_mut(slot) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = true; return Ok(()); @@ -1625,7 +1401,8 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = false; Ok(()) @@ -1634,7 +1411,6 @@ impl Reactor { pub(super) fn send_external_udp( &mut self, socket_id: u64, - slot: UdpEndpointSlot, data: &[u8], address: SocketAddrV4, ) -> BrokerResult> { @@ -1643,7 +1419,8 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .endpoint(slot) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; match sendto( &endpoint.socket, @@ -1672,7 +1449,7 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)?; let udp = socket.udp_state_mut()?; - let endpoint = udp.endpoint_mut(slot).ok_or(BrokerError::Internal)?; + let endpoint = udp.native_endpoint.as_mut().ok_or(BrokerError::Internal)?; endpoint.write_blocked = true; if !endpoint.error.is_pending() { epoll::modify( @@ -1736,7 +1513,7 @@ impl Reactor { .ok_or(BrokerError::Internal)? .udp_state_mut()?; udp.peeked_origin = None; - udp.next_receive_origin = UdpReceiveOrigin::External; + udp.next_receive_origin = UdpReceiveOrigin::Native; // The dequeue is committed and the cached snapshot is authoritative // even if this association cannot accept the notification. let _ = self.publish_udp_readiness(socket_id); @@ -1751,17 +1528,16 @@ impl Reactor { pub(super) fn receive_native_udp( &mut self, socket_id: u64, - route: UdpNativeRoute, length: usize, flags: ReceiveFromFlags, ) -> BrokerResult> { - let slot = UdpEndpointSlot::Active(route); let can_receive = self .sockets .get(&socket_id) .ok_or(BrokerError::Internal)? .udp_state()? - .native_endpoint(route) + .native_endpoint + .as_ref() .is_some_and(|endpoint| endpoint.readable); if !can_receive { return Ok(None); @@ -1772,7 +1548,8 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, length, ReceiveFromFlags::PEEK) }; @@ -1783,7 +1560,7 @@ impl Reactor { source_address, }) => (data, datagram_length, source_address), Ok(ReactorReceiveFromOutcome::Failed(error)) => { - return self.finish_udp_receive_error(socket_id, route, error); + return self.finish_udp_receive_error(socket_id, error); } Err(BrokerError::WouldBlock) => { let endpoint = self @@ -1791,22 +1568,23 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)?; endpoint.readable = false; - self.rearm_udp_endpoint(socket_id, slot)?; + self.rearm_udp_endpoint(socket_id)?; return Ok(None); } Err(error) => return Err(error), }; - let authorized = self.udp_native_source_authorized(socket_id, route, source_address)?; + let authorized = self.udp_native_source_authorized(socket_id, source_address)?; if authorized { if flags.contains(ReceiveFromFlags::PEEK) { self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .peeked_origin = Some(route.receive_origin()); + .peeked_origin = Some(UdpReceiveOrigin::Native); return Ok(Some(ReactorReceiveFromOutcome::Received { data, datagram_length, @@ -1820,9 +1598,7 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()?; - let endpoint = udp - .native_endpoint_mut(route) - .ok_or(BrokerError::Internal)?; + let endpoint = udp.native_endpoint.as_mut().ok_or(BrokerError::Internal)?; if endpoint.error.is_pending() { false } else { @@ -1830,12 +1606,13 @@ impl Reactor { true } }; - if should_rearm && let Err(error) = self.rearm_udp_endpoint(socket_id, slot) { + if should_rearm && let Err(error) = self.rearm_udp_endpoint(socket_id) { self.sockets .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = true; return Err(error); @@ -1844,14 +1621,15 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::NONE) }; match consumed { Ok(ReactorReceiveFromOutcome::Received { .. }) => {} Ok(ReactorReceiveFromOutcome::Failed(error)) => { - return self.finish_udp_receive_error(socket_id, route, error); + return self.finish_udp_receive_error(socket_id, error); } Err(BrokerError::WouldBlock) => { let _ = self.publish_udp_readiness(socket_id); @@ -1862,7 +1640,8 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = true; return Err(error); @@ -1879,7 +1658,7 @@ impl Reactor { // the dequeue. Refresh the cached head state so another // authorized datagram keeps READ asserted without requiring // an asynchronous low-to-high publication. - let _ = self.drain_invalid_udp_ingress(socket_id, route); + let _ = self.drain_invalid_udp_ingress(socket_id); // The cached snapshot remains authoritative if notification // delivery fails after irreversible consumption. let _ = self.publish_udp_readiness(socket_id); @@ -1892,12 +1671,13 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .native_endpoint(route) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; match receive_datagram_fd(&endpoint.socket, 0, ReceiveFromFlags::NONE) { Ok(ReactorReceiveFromOutcome::Received { .. }) => {} Ok(ReactorReceiveFromOutcome::Failed(error)) => { - return self.finish_udp_receive_error(socket_id, route, error); + return self.finish_udp_receive_error(socket_id, error); } Err(BrokerError::WouldBlock) => { let endpoint = self @@ -1905,10 +1685,11 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)?; endpoint.readable = false; - self.rearm_udp_endpoint(socket_id, slot)?; + self.rearm_udp_endpoint(socket_id)?; return Ok(None); } Err(error) => return Err(error), @@ -1918,10 +1699,11 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)? .udp_state_mut()? - .native_endpoint_mut(route) + .native_endpoint + .as_mut() .ok_or(BrokerError::Internal)? .readable = false; - self.rearm_udp_endpoint(socket_id, slot)?; + self.rearm_udp_endpoint(socket_id)?; Ok(None) } @@ -1930,7 +1712,8 @@ impl Reactor { let socket = self.sockets.get(&socket_id).ok_or(BrokerError::Internal)?; let endpoint = socket .udp_state()? - .native_endpoint(UdpNativeRoute::External) + .native_endpoint + .as_ref() .ok_or(BrokerError::Internal)?; usize::try_from(ioctl_fionread(&endpoint.socket).map_err(broker_error_from_errno)?) .map_err(|_| BrokerError::Internal) @@ -2075,31 +1858,22 @@ mod tests { #[test] fn consumed_udp_native_error_preserves_the_exact_error() { - let mut state = UdpNativeErrorState::PendingKernel { sequence: 7 }; - state.record_consumed(8, SocketError::ConnectionRefused); + let mut state = UdpNativeErrorState::PendingKernel; + state.record_consumed(SocketError::ConnectionRefused); assert_eq!( std::mem::take(&mut state), - UdpNativeErrorState::Consumed { - sequence: 7, - error: SocketError::ConnectionRefused, - } + UdpNativeErrorState::Consumed(SocketError::ConnectionRefused) ); assert_eq!(state, UdpNativeErrorState::None); } #[test] fn udp_native_error_coalescing_preserves_the_oldest_observation() { - let mut state = UdpNativeErrorState::Consumed { - sequence: 7, - error: SocketError::ConnectionRefused, - }; - state.record_consumed(8, SocketError::Other); + let mut state = UdpNativeErrorState::Consumed(SocketError::ConnectionRefused); + state.record_consumed(SocketError::Other); assert_eq!( state, - UdpNativeErrorState::Consumed { - sequence: 7, - error: SocketError::ConnectionRefused, - } + UdpNativeErrorState::Consumed(SocketError::ConnectionRefused) ); } From 06bf0e3643b717351892d482bc5277a402ca6b9b Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Tue, 18 Aug 2026 16:33:37 -0700 Subject: [PATCH 3/3] Republish readiness for queued UDP errors Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b5a1a347-37a8-4246-8bbc-306590921475 --- .../src/socket.rs | 38 ++++++++++++++ .../src/socket/tests/udp.rs | 52 +++++++++++++++++++ .../src/socket/udp.rs | 41 +++++++++++++++ 3 files changed, 131 insertions(+) diff --git a/litebox_broker_platform_linux_userland/src/socket.rs b/litebox_broker_platform_linux_userland/src/socket.rs index 0f0fc0819..ee4af0400 100644 --- a/litebox_broker_platform_linux_userland/src/socket.rs +++ b/litebox_broker_platform_linux_userland/src/socket.rs @@ -627,6 +627,26 @@ impl ReactorClient { receive.recv().unwrap() } + #[cfg(test)] + fn inject_udp_status_errors( + &self, + guest_port: u16, + cached_error: SocketError, + native_error: SocketError, + ) { + let (response, receive) = sync_channel(1); + self.commands + .send(ReactorCommand::InjectUdpStatusErrors { + guest_port, + cached_error, + native_error, + response, + }) + .unwrap(); + self.signal().unwrap(); + receive.recv().unwrap().unwrap(); + } + #[cfg(test)] fn udp_native_receive_buffer_size(&self, guest_port: u16) -> Option { let (response, receive) = sync_channel(1); @@ -870,6 +890,13 @@ enum ReactorCommand { response: SyncSender, }, #[cfg(test)] + InjectUdpStatusErrors { + guest_port: u16, + cached_error: SocketError, + native_error: SocketError, + response: SyncSender>, + }, + #[cfg(test)] UdpNativeReceiveBufferSize { guest_port: u16, response: SyncSender>>, @@ -1731,6 +1758,17 @@ impl Reactor { let _ = response.send(self.udp.event_tokens.len()); } #[cfg(test)] + ReactorCommand::InjectUdpStatusErrors { + guest_port, + cached_error, + native_error, + response, + } => { + let outcome = + self.inject_udp_status_errors(guest_port, cached_error, native_error); + let _ = response.send(outcome); + } + #[cfg(test)] ReactorCommand::UdpNativeReceiveBufferSize { guest_port, response, diff --git a/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs b/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs index a02e3382f..35ee81ac8 100644 --- a/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs +++ b/litebox_broker_platform_linux_userland/src/socket/tests/udp.rs @@ -285,6 +285,58 @@ fn udp_status_publication_failure_still_rearms_native_endpoint() { readiness.wait_for_publish_failure_consumed(); } +#[test] +fn udp_status_republishes_when_another_error_remains_pending() { + let provider = Arc::new(LinuxSocketProvider::new(1, 1).unwrap()); + let broker = BrokerCore::new_with_limits( + PolicyEngine::with_unauthenticated_rights(ObjectRights::all()) + .with_socket_policy(SocketPolicy::Ipv4Loopback), + BrokerCoreLimits::new_with_all_limits(2, 0, 1, 1), + provider.clone(), + ) + .unwrap(); + let session = broker + .create_session(CallerCredential::Unauthenticated) + .unwrap(); + let (published, publications) = channel(); + let (retired, _retirements) = channel(); + let readiness = Arc::new(TestReadinessSink { published, retired }); + let socket = create_udp_socket(&session, readiness); + let peer = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let peer_address = socket_address_v4(peer.local_addr().unwrap()); + assert_eq!( + litebox_broker_core::socket::connect(&session, socket, peer_address), + Ok(SocketOutcome::Completed(SocketConnectionStatus::Connected)) + ); + let local_address = litebox_broker_core::socket::status(&session, socket) + .unwrap() + .local_address + .unwrap(); + while publications.try_recv().is_ok() {} + + provider.reactor.inject_udp_status_errors( + local_address.port(), + SocketError::ConnectionRefused, + SocketError::Other, + ); + assert_eq!( + litebox_broker_core::socket::status(&session, socket) + .unwrap() + .pending_error, + Some(SocketError::ConnectionRefused) + ); + let (published_socket, published_readiness) = publications.recv_timeout(TEST_TIMEOUT).unwrap(); + assert_eq!(published_socket, socket); + assert!(published_readiness.contains(ReadinessFlags::ERROR)); + + assert_eq!( + litebox_broker_core::socket::status(&session, socket) + .unwrap() + .pending_error, + Some(SocketError::Other) + ); +} + #[test] fn guest_udp_queue_pressure_drops_new_datagrams_successfully() { let provider = Arc::new(LinuxSocketProvider::new(2, 1).unwrap()); diff --git a/litebox_broker_platform_linux_userland/src/socket/udp.rs b/litebox_broker_platform_linux_userland/src/socket/udp.rs index 974284ecf..2577b24fe 100644 --- a/litebox_broker_platform_linux_userland/src/socket/udp.rs +++ b/litebox_broker_platform_linux_userland/src/socket/udp.rs @@ -566,6 +566,7 @@ impl Reactor { .native_endpoint .as_ref() .is_some_and(|endpoint| endpoint.error.is_pending()); + let republish_error = cached_error.is_some() && native_error_pending; if pending_error.is_none() && native_error_pending { pending_error = self.take_udp_error(socket_id)?; } @@ -575,6 +576,11 @@ impl Reactor { .get_mut(&socket_id) .ok_or(BrokerError::Internal)?; let _ = update_snapshot(socket, None, readiness); + if republish_error { + // Returning the cached error leaves another error pending without + // changing readiness, so wake a waiter for the remaining error. + let _ = socket.readiness.republish(readiness); + } if let Err(error) = self.rearm_udp_endpoint_if_needed(socket_id) { if let Some(pending_error) = pending_error { let socket = self @@ -601,6 +607,41 @@ impl Reactor { }) } + #[cfg(test)] + pub(super) fn inject_udp_status_errors( + &mut self, + guest_port: u16, + cached_error: SocketError, + native_error: SocketError, + ) -> BrokerResult<()> { + let socket_id = self + .udp + .bindings + .get(guest_port) + .ok_or(BrokerError::Internal)? + .socket_id; + let socket = self + .sockets + .get_mut(&socket_id) + .ok_or(BrokerError::Internal)?; + { + let mut snapshot = socket + .snapshot + .lock() + .expect("Linux socket snapshot mutex poisoned"); + snapshot.pending_error = Some(cached_error); + snapshot.readiness = snapshot.readiness | ReadinessFlags::ERROR; + } + socket + .udp_state_mut()? + .native_endpoint + .as_mut() + .ok_or(BrokerError::Internal)? + .error + .record_consumed(native_error); + Ok(()) + } + fn udp_queue_would_drop( &self, receiver_id: u64,