diff --git a/Cargo.lock b/Cargo.lock index 3049be1..06324c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -585,7 +585,7 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" [[package]] name = "flux" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bitcode", "core_affinity", @@ -609,7 +609,7 @@ dependencies = [ [[package]] name = "flux-communication" -version = "0.2.0" +version = "0.2.1" dependencies = [ "directories", "flux-timing", @@ -624,7 +624,7 @@ dependencies = [ [[package]] name = "flux-ctl" -version = "0.2.0" +version = "0.2.1" dependencies = [ "clap", "crossterm", @@ -643,7 +643,7 @@ dependencies = [ [[package]] name = "flux-disk" -version = "0.2.0" +version = "0.2.1" dependencies = [ "flux-timing", "flux-utils", @@ -655,7 +655,7 @@ dependencies = [ [[package]] name = "flux-network" -version = "0.2.0" +version = "0.2.1" dependencies = [ "flux", "flux-communication", @@ -674,7 +674,7 @@ dependencies = [ [[package]] name = "flux-profiler" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bytesize", "clap", @@ -692,7 +692,7 @@ dependencies = [ [[package]] name = "flux-profiler-macros" -version = "0.2.0" +version = "0.2.1" dependencies = [ "quote", "syn 2.0.117", @@ -700,7 +700,7 @@ dependencies = [ [[package]] name = "flux-timekeeper" -version = "0.2.0" +version = "0.2.1" dependencies = [ "auto_impl", "bitflags 2.10.0", @@ -715,7 +715,7 @@ dependencies = [ [[package]] name = "flux-timing" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bitcode", "chrono", @@ -733,7 +733,7 @@ dependencies = [ [[package]] name = "flux-utils" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bytes", "core_affinity", @@ -755,7 +755,7 @@ dependencies = [ [[package]] name = "flux-versioned-types" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bincode", "flux", @@ -765,7 +765,7 @@ dependencies = [ [[package]] name = "flux-versioned-types-macros" -version = "0.2.0" +version = "0.2.1" dependencies = [ "proc-macro2", "quote", @@ -1791,7 +1791,7 @@ dependencies = [ [[package]] name = "spine-derive" -version = "0.2.0" +version = "0.2.1" dependencies = [ "proc-macro2", "quote", @@ -2034,11 +2034,11 @@ dependencies = [ [[package]] name = "type-hash" -version = "0.2.0" +version = "0.2.1" [[package]] name = "type-hash-derive" -version = "0.2.0" +version = "0.2.1" dependencies = [ "proc-macro-crate", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index 1b30149..9d31d54 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ license = "Apache-2.0 AND MIT" publish = false repository = "https://github.com/gattaca-com/flux" rust-version = "1.91.0" -version = "0.2.0" +version = "0.2.1" [workspace.dependencies] diff --git a/crates/flux-network/src/tcp/mod.rs b/crates/flux-network/src/tcp/mod.rs index 3d5599b..cad2054 100644 --- a/crates/flux-network/src/tcp/mod.rs +++ b/crates/flux-network/src/tcp/mod.rs @@ -3,6 +3,9 @@ mod network; mod stream; pub use connector::{PollEvent, SendBehavior, TcpConnector}; -pub use network::{Framing, PayloadBuf, TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork}; +pub use network::{ + Framing, PayloadBuf, TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork, TcpNetworkCore, + TcpNetworkWithExternalPoll, +}; pub(crate) use stream::set_socket_buf_size; pub use stream::{ConnState, TcpStream, TcpTelemetry}; diff --git a/crates/flux-network/src/tcp/network.rs b/crates/flux-network/src/tcp/network.rs index c5dcb20..cb9175f 100644 --- a/crates/flux-network/src/tcp/network.rs +++ b/crates/flux-network/src/tcp/network.rs @@ -1,7 +1,7 @@ use std::{ io::{self, IoSlice, Read, Write}, net::{Shutdown, SocketAddr}, - ops::{Deref, DerefMut}, + ops::{Deref, DerefMut, Range}, }; use flux_communication::Timer; @@ -332,11 +332,14 @@ struct PendingDisconnect { } struct NetworkState { - poll: Poll, + registry: Registry, groups: Vec, listeners: Vec, connections: Vec, pending_disconnects: Vec, + /// Range used to assign tokens to new listeners and connections. + /// Assigned tokens are not reused. + token_range: Range, next_token: usize, /// Frames staged for the next socket write, each as a contiguous /// `[header][payload]` for length-prefixed groups or bare bytes for raw @@ -344,24 +347,29 @@ struct NetworkState { send_buffer: Vec, } -impl Default for NetworkState { - fn default() -> Self { +impl NetworkState { + fn new(registry: Registry, tokens: Range) -> Self { Self { - poll: Poll::new().expect("couldn't set up a poll for tcp network"), + registry, groups: Vec::with_capacity(INITIAL_GROUP_CAPACITY), listeners: Vec::with_capacity(INITIAL_LISTENER_CAPACITY), connections: Vec::with_capacity(INITIAL_CONNECTION_CAPACITY), pending_disconnects: Vec::with_capacity(INITIAL_CONNECTION_CAPACITY), - next_token: 0, + next_token: tokens.start, + token_range: tokens, send_buffer: Vec::with_capacity(INITIAL_SEND_BUFFER_SIZE), } } -} -impl NetworkState { fn next_token(&mut self) -> Token { + // Reject allocations beyond the caller-provided range. + assert!( + self.next_token < self.token_range.end, + "tcp token range {:?} exhausted", + self.token_range + ); let token = Token(self.next_token); - self.next_token = self.next_token.checked_add(1).expect("tcp token space exhausted"); + self.next_token += 1; token } @@ -375,7 +383,7 @@ impl NetworkState { } let mut socket = TcpListener::bind(addr)?; let token = self.next_token(); - self.poll.registry().register(&mut socket, token, Interest::READABLE)?; + self.registry.register(&mut socket, token, Interest::READABLE)?; self.listeners.push(Listener { token, group, socket }); Ok(()) } @@ -420,7 +428,7 @@ impl NetworkState { if let Some(size) = socket_buf_size { set_socket_buf_size(&socket, size); } - if let Err(err) = self.poll.registry().register(&mut socket, token, Interest::WRITABLE) { + if let Err(err) = self.registry.register(&mut socket, token, Interest::WRITABLE) { warn!(?err, %peer_addr, "couldn't register connecting tcp stream"); let _ = socket.shutdown(Shutdown::Both); return; @@ -498,7 +506,7 @@ impl NetworkState { let Err(err) = socket.set_nodelay(true) { warn!(?err, %peer_addr, "couldn't set nodelay on tcp stream"); - let _ = self.poll.registry().deregister(&mut socket); + let _ = self.registry.deregister(&mut socket); let _ = socket.shutdown(Shutdown::Both); return false; } @@ -506,12 +514,12 @@ impl NetworkState { let Err(err) = set_keepalive(&socket) { warn!(?err, %peer_addr, "couldn't set keepalive on tcp stream"); - let _ = self.poll.registry().deregister(&mut socket); + let _ = self.registry.deregister(&mut socket); let _ = socket.shutdown(Shutdown::Both); return false; } set_user_timeout(&socket, config.user_timeout_ms); - if let Err(err) = self.poll.registry().reregister(&mut socket, token, Interest::READABLE) { + if let Err(err) = self.registry.reregister(&mut socket, token, Interest::READABLE) { warn!(?err, %peer_addr, "couldn't register connected tcp stream"); let _ = socket.shutdown(Shutdown::Both); return false; @@ -524,15 +532,10 @@ impl NetworkState { write_frame_header(&mut header, message.len(), Nanos::now()); header }); - if stream.write_frame( - self.poll.registry(), - header.as_ref(), - message, - config, - &mut timers, - ) == StreamState::Disconnected + if stream.write_frame(&self.registry, header.as_ref(), message, config, &mut timers) == + StreamState::Disconnected { - stream.close(self.poll.registry()); + stream.close(&self.registry); self.connections[index].timers = timers; return false; } @@ -581,9 +584,7 @@ impl NetworkState { continue; } set_user_timeout(&socket, config.user_timeout_ms); - if let Err(err) = - self.poll.registry().register(&mut socket, token, Interest::READABLE) - { + if let Err(err) = self.registry.register(&mut socket, token, Interest::READABLE) { warn!(?err, %peer_addr, "couldn't register accepted tcp stream"); let _ = socket.shutdown(Shutdown::Both); continue; @@ -604,14 +605,14 @@ impl NetworkState { header }); if stream.write_frame( - self.poll.registry(), + &self.registry, header.as_ref(), message, config, &mut timers, ) == StreamState::Disconnected { - stream.close(self.poll.registry()); + stream.close(&self.registry); continue; } } @@ -663,7 +664,7 @@ impl NetworkState { let connection = &mut self.connections[index]; let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; let state = stream.poll_with( - self.poll.registry(), + &self.registry, event, config, &mut connection.timers, @@ -687,12 +688,12 @@ impl NetworkState { match old_state { ConnectionState::Disconnected => false, ConnectionState::Connecting(mut socket) => { - let _ = self.poll.registry().deregister(&mut socket); + let _ = self.registry.deregister(&mut socket); let _ = socket.shutdown(Shutdown::Both); false } ConnectionState::Connected(mut stream) => { - stream.close(self.poll.registry()); + stream.close(&self.registry); true } } @@ -799,7 +800,7 @@ impl NetworkState { let connection = &mut self.connections[index]; let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; let state = stream.write_frame( - self.poll.registry(), + &self.registry, None, &self.send_buffer, config, @@ -952,24 +953,96 @@ impl NetworkState { } } +impl Drop for NetworkState { + fn drop(&mut self) { + // Deregister before closing. On Linux, a duplicated descriptor can + // keep an epoll registration alive after the original descriptor closes. + for index in 0..self.connections.len() { + self.close_connection_socket(index); + } + for listener in &mut self.listeners { + let _ = self.registry.deregister(&mut listener.socket); + } + } +} + /// A grouped collection of TCP listeners and persistent outbound endpoints -/// driven by one nonblocking poll. +/// with an internal nonblocking poll. /// -/// Unlike [`super::TcpConnector`], queued bytes are never retained across a -/// disconnected socket. Use `TcpConnector` when reconnect backlog replay is -/// required. +/// Shared network operations are provided by [`TcpNetworkCore`] through +/// `Deref`. Use [`TcpNetworkWithExternalPoll`] to register the sockets with a +/// caller-owned poll instead. pub struct TcpNetwork { events: Events, - state: NetworkState, + core: TcpNetworkCore, + poll: Poll, } impl Default for TcpNetwork { fn default() -> Self { - Self { events: Events::with_capacity(EVENTS_CAPACITY), state: NetworkState::default() } + let poll = Poll::new().expect("failed to create poll"); + let registry = poll.registry().try_clone().expect("failed to clone poll registry"); + // This network owns the poll, so its token range starts at zero. + Self { + events: Events::with_capacity(EVENTS_CAPACITY), + core: TcpNetworkCore::new(registry, 0..usize::MAX), + poll, + } + } +} + +impl Deref for TcpNetwork { + type Target = TcpNetworkCore; + + fn deref(&self) -> &Self::Target { + &self.core + } +} + +impl DerefMut for TcpNetwork { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.core } } impl TcpNetwork { + pub fn poll_with(&mut self, mut handler: F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.core.state.drain_pending_disconnects(&mut handler); + self.core.state.maybe_reconnect(); + if let Err(err) = self.poll.poll(&mut self.events, Some(std::time::Duration::ZERO)) { + if err.kind() != io::ErrorKind::Interrupted { + flux_utils::safe_panic!("couldn't poll tcp network: {err}"); + } + return; + } + for event in &self.events { + self.core.state.handle_event(event, &mut handler); + } + self.core.state.drain_pending_disconnects(&mut handler); + } +} + +/// Shared state and network operations used by [`TcpNetwork`] and +/// [`TcpNetworkWithExternalPoll`]. +/// +/// This type stores protocol groups, listeners, and connections, but does not +/// own or poll a [`Poll`]. +/// +/// Unlike [`super::TcpConnector`], queued bytes are never retained across a +/// disconnected socket. Use `TcpConnector` when reconnect backlog replay is +/// required. +pub struct TcpNetworkCore { + state: NetworkState, +} + +impl TcpNetworkCore { + fn new(registry: Registry, tokens: Range) -> Self { + Self { state: NetworkState::new(registry, tokens) } + } + /// Adds a protocol group and returns its handle. #[must_use = "the group handle identifies listeners and outbound endpoints"] pub fn add_group(&mut self, config: TcpGroupConfig) -> TcpGroup { @@ -1011,24 +1084,6 @@ impl TcpNetwork { self.state.connect(group, peer_addr) } - pub fn poll_with(&mut self, mut handler: F) - where - F: for<'a> FnMut(TcpEvent<'a>), - { - self.state.drain_pending_disconnects(&mut handler); - self.state.maybe_reconnect(); - if let Err(err) = self.state.poll.poll(&mut self.events, Some(std::time::Duration::ZERO)) { - if err.kind() != io::ErrorKind::Interrupted { - flux_utils::safe_panic!("couldn't poll tcp network: {err}"); - } - return; - } - for event in &self.events { - self.state.handle_event(event, &mut handler); - } - self.state.drain_pending_disconnects(&mut handler); - } - /// Serializes and sends one payload to a connected token. Length-prefixed /// groups add a frame header; raw groups send the payload unchanged. The /// closure is not called when the token is unknown or currently @@ -1068,7 +1123,7 @@ impl TcpNetwork { /// Serializes multiple payloads once and sends the batch to every /// connected member of `group`. Framing, size limits, and skipping of - /// invalid payloads follow [`TcpNetwork::send_many_with`]; each member + /// invalid payloads follow [`Self::send_many_with`]; each member /// receives the batch in one socket write when it has no backlog. The /// closure is not called when the group has no connected member. Returns /// the number of recipients attempted. @@ -1102,6 +1157,118 @@ impl TcpNetwork { } } +/// A TCP network whose sockets are registered with a poll owned by the caller. +/// +/// A polling pass consists of [`Self::pre_poll`], polling, calling +/// [`Self::handle_event`] for each event in this network's token range, and +/// [`Self::post_poll`]. The caller owns the poll, event buffer, and timeout. +/// Dropping this value attempts to deregister its sockets before closing them. +/// +/// # Example +/// +/// ```no_run +/// use std::time::Duration; +/// +/// use flux_network::tcp::{TcpEvent, TcpGroupConfig, TcpNetworkWithExternalPoll}; +/// use mio::{Events, Poll}; +/// +/// // Every source sharing the poll must use non-overlapping token range. +/// const NETWORK_TOKENS: std::ops::Range = (1 << 48)..(2 << 48); +/// +/// let mut poll = Poll::new().unwrap(); +/// let mut events = Events::with_capacity(128); +/// let mut network = TcpNetworkWithExternalPoll::new( +/// poll.registry().try_clone().unwrap(), +/// NETWORK_TOKENS, +/// ); +/// let group = network.add_group(TcpGroupConfig::default()); +/// network.listen(group, "127.0.0.1:9099".parse().unwrap()).unwrap(); +/// +/// let mut handle_tcp_event = |_event: TcpEvent<'_>| { +/// // Process the event here. Copy payload bytes if they must outlive the +/// // callback. +/// }; +/// +/// loop { +/// network.pre_poll(&mut handle_tcp_event); +/// poll.poll(&mut events, Some(Duration::from_millis(1))).unwrap(); +/// for event in &events { +/// if NETWORK_TOKENS.contains(&event.token().0) { +/// network.handle_event(event, &mut handle_tcp_event); +/// } else { +/// // Route the event to another source registered with this poll. +/// } +/// } +/// network.post_poll(&mut handle_tcp_event); +/// } +/// ``` +pub struct TcpNetworkWithExternalPoll { + core: TcpNetworkCore, +} + +impl Deref for TcpNetworkWithExternalPoll { + type Target = TcpNetworkCore; + + fn deref(&self) -> &Self::Target { + &self.core + } +} + +impl DerefMut for TcpNetworkWithExternalPoll { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.core + } +} + +impl TcpNetworkWithExternalPoll { + /// Creates a network that assigns listener and connection tokens from + /// `tokens` in ascending order without reuse. + /// + /// An operation that needs a new token panics if the range is exhausted. + /// The caller is responsible for avoiding collisions with other tokens + /// registered with the same poll. + pub fn new(registry: Registry, tokens: Range) -> Self { + Self { core: TcpNetworkCore::new(registry, tokens) } + } + + /// Delivers pending disconnect notifications and attempts due reconnects. + pub fn pre_poll(&mut self, handler: &mut F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.core.state.drain_pending_disconnects(handler); + self.core.state.maybe_reconnect(); + } + + /// Delivers pending disconnect notifications. + /// + /// If this step is skipped, notifications remain queued for a later call + /// to this method or [`Self::pre_poll`]. + pub fn post_poll(&mut self, handler: &mut F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.core.state.drain_pending_disconnects(handler); + } + + /// Processes one readiness event routed to this network. + /// + /// The event token must be in the range supplied to [`Self::new`]. This is + /// checked by an assertion in debug builds. + pub fn handle_event(&mut self, event: &Event, handler: &mut F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + debug_assert!( + self.core.state.token_range.contains(&event.token().0), + "event token {:?} lies outside this network's token range {:?}", + event.token(), + self.core.state.token_range + ); + self.core.state.handle_event(event, handler); + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum StreamState { Alive, @@ -1660,4 +1827,47 @@ mod tests { let mut payload = PayloadBuf::new(&mut bytes); payload.resize(usize::MAX, 0); } + + #[cfg(target_os = "linux")] + #[test] + fn drop_removes_listener_and_endpoint_registrations() { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + + use flux_timing::{Duration, Repeater}; + + use super::{GroupState, NetworkState, TcpGroup}; + + fn registered_fds(epoll_fd: i32) -> usize { + std::fs::read_to_string(format!("/proc/self/fdinfo/{epoll_fd}")) + .unwrap() + .lines() + .filter(|line| line.starts_with("tfd:")) + .count() + } + + let poll = Poll::new().unwrap(); + let epoll_fd = poll.as_raw_fd(); + let registry = poll.registry().try_clone().unwrap(); + let mut state = NetworkState::new(registry, 100..200); + state.groups.push(GroupState { + config: TcpGroupConfig::default(), + reconnector: Repeater::every(Duration::from_secs(2)), + }); + let group = TcpGroup(0); + state.listen(group, (Ipv4Addr::LOCALHOST, 0).into()).unwrap(); + let addr = state.listeners[0].socket.local_addr().unwrap(); + let _endpoint = state.connect(group, addr); + assert_eq!(registered_fds(epoll_fd), 2); + + // Keep the listener's file description alive after its mio socket drops. + // SAFETY: the listener owns this descriptor for the duration of the call. + let dup_fd = unsafe { libc::dup(state.listeners[0].socket.as_raw_fd()) }; + assert!(dup_fd >= 0); + // SAFETY: dup_fd is valid after the check above, and ownership is + // transferred exactly once. + let _dup = unsafe { OwnedFd::from_raw_fd(dup_fd) }; + + drop(state); + assert_eq!(registered_fds(epoll_fd), 0); + } } diff --git a/crates/flux-network/tests/tcp_network.rs b/crates/flux-network/tests/tcp_network.rs index 67470ee..9f5b509 100644 --- a/crates/flux-network/tests/tcp_network.rs +++ b/crates/flux-network/tests/tcp_network.rs @@ -728,3 +728,14 @@ fn tcp_network_client_is_wire_compatible_with_tcp_connector_server() { } assert!(contains(&network_messages, RESPONSE)); } + +#[test] +fn owned_poll_assigns_tokens_from_zero() { + let mut network = TcpNetwork::default(); + let group = network.add_group(TcpGroupConfig::default()); + let first = network.connect(group, unused_addr()); + assert_eq!(first, mio::Token(0)); + network.listen(group, unused_addr()).unwrap(); + let second = network.connect(group, unused_addr()); + assert_eq!(second, mio::Token(2), "the listener consumed token 1 from the same counter"); +} diff --git a/crates/flux-network/tests/tcp_network_external_poll.rs b/crates/flux-network/tests/tcp_network_external_poll.rs new file mode 100644 index 0000000..a94b899 --- /dev/null +++ b/crates/flux-network/tests/tcp_network_external_poll.rs @@ -0,0 +1,275 @@ +use std::{ + cell::Cell, + net::{Ipv4Addr, SocketAddr}, + ops::Range, + time::{Duration, Instant}, +}; + +use flux_network::tcp::{TcpEvent, TcpGroupConfig, TcpNetworkWithExternalPoll}; +use mio::{Events, Poll, Token}; + +const SERVER_TOKENS: Range = 100..200; +const CLIENT_TOKENS: Range = 200..300; + +const SERVER_HELLO: &[u8] = b"server-hello"; +const CLIENT_HELLO: &[u8] = b"client-hello"; +const REQUEST: &[u8] = b"request-payload"; +const RESPONSE: &[u8] = b"response-payload"; +const AFTER_RECONNECT: &[u8] = b"after-reconnect"; + +/// The external-poll phase that invoked a handler. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Phase { + Pre, + Event, + Post, +} + +#[derive(Debug)] +enum Ev { + Accepted(Token), + Connected(Token), + Message(Token, Vec), + Disconnected(Token), +} + +#[derive(Debug)] +struct Record { + phase: Phase, + event: Ev, +} + +fn unused_addr() -> SocketAddr { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + addr +} + +fn accepts(log: &[Record]) -> Vec { + log.iter() + .filter_map(|r| if let Ev::Accepted(token) = r.event { Some(token) } else { None }) + .collect() +} + +fn connects(log: &[Record]) -> usize { + log.iter().filter(|r| matches!(r.event, Ev::Connected(_))).count() +} + +fn has_message(log: &[Record], payload: &[u8]) -> bool { + log.iter().any(|r| matches!(&r.event, Ev::Message(_, p) if p == payload)) +} + +fn message_token(log: &[Record], payload: &[u8]) -> Option { + log.iter().find_map(|r| match &r.event { + Ev::Message(token, p) if p == payload => Some(*token), + _ => None, + }) +} + +fn disconnect_phase(log: &[Record], token: Token) -> Option { + log.iter().find_map(|r| match r.event { + Ev::Disconnected(t) if t == token => Some(r.phase), + _ => None, + }) +} + +#[test] +#[allow(clippy::too_many_lines)] +fn one_poll_drives_two_networks_with_disjoint_ranges() { + let addr = unused_addr(); + let mut poll = Poll::new().unwrap(); + let mut events = Events::with_capacity(128); + + let mut server = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), SERVER_TOKENS); + let mut client = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), CLIENT_TOKENS); + + let server_group = server.add_group(TcpGroupConfig { + name: "server", + on_connect_msg: Some(SERVER_HELLO.to_vec()), + ..TcpGroupConfig::default() + }); + let client_group = client.add_group(TcpGroupConfig { + name: "client", + on_connect_msg: Some(CLIENT_HELLO.to_vec()), + reconnect_interval: flux_timing::Duration::from_millis(1), + ..TcpGroupConfig::default() + }); + + server.listen(server_group, addr).unwrap(); + let client_token = client.connect(client_group, addr); + assert!(CLIENT_TOKENS.contains(&client_token.0)); + + let phase = Cell::new(Phase::Pre); + let mut server_log: Vec = Vec::new(); + let mut client_log: Vec = Vec::new(); + + let mut request_sent = false; + let mut response_sent = false; + let mut disconnect_issued = false; + let mut resent = false; + let mut done = false; + + let deadline = Instant::now() + Duration::from_secs(10); + while !done && Instant::now() < deadline { + { + let mut on_server = |event: TcpEvent<'_>| { + let ev = match event { + TcpEvent::Accepted { group, token, .. } => { + assert_eq!(group, server_group); + Ev::Accepted(token) + } + TcpEvent::Message { group, token, payload, .. } => { + assert_eq!(group, server_group); + Ev::Message(token, payload.to_vec()) + } + TcpEvent::Disconnected { group, token, .. } => { + assert_eq!(group, server_group); + Ev::Disconnected(token) + } + TcpEvent::Connected { .. } => { + panic!("server network has no outbound endpoints") + } + }; + let token = match &ev { + Ev::Accepted(t) | + Ev::Message(t, _) | + Ev::Disconnected(t) | + Ev::Connected(t) => *t, + }; + assert!( + SERVER_TOKENS.contains(&token.0), + "server event token {token:?} out of range" + ); + server_log.push(Record { phase: phase.get(), event: ev }); + }; + let mut on_client = |event: TcpEvent<'_>| { + let ev = match event { + TcpEvent::Connected { group, token, .. } => { + assert_eq!(group, client_group); + assert_eq!(token, client_token, "reconnect must keep the endpoint token"); + Ev::Connected(token) + } + TcpEvent::Message { group, token, payload, .. } => { + assert_eq!(group, client_group); + assert_eq!(token, client_token); + Ev::Message(token, payload.to_vec()) + } + TcpEvent::Disconnected { group, token, .. } => { + assert_eq!(group, client_group); + assert_eq!(token, client_token); + Ev::Disconnected(token) + } + TcpEvent::Accepted { .. } => panic!("client network has no listeners"), + }; + client_log.push(Record { phase: phase.get(), event: ev }); + }; + + phase.set(Phase::Pre); + server.pre_poll(&mut on_server); + client.pre_poll(&mut on_client); + + poll.poll(&mut events, Some(Duration::from_millis(1))).unwrap(); + phase.set(Phase::Event); + for event in &events { + let token = event.token().0; + if SERVER_TOKENS.contains(&token) { + server.handle_event(event, &mut on_server); + } else if CLIENT_TOKENS.contains(&token) { + client.handle_event(event, &mut on_client); + } else { + panic!("event token {token} lies outside every network's range"); + } + } + + phase.set(Phase::Post); + server.post_poll(&mut on_server); + client.post_poll(&mut on_client); + } + + if !request_sent && connects(&client_log) >= 1 && has_message(&client_log, SERVER_HELLO) { + assert!(client.send_with(client_token, |buf| buf.extend_from_slice(REQUEST))); + request_sent = true; + } + if !response_sent && has_message(&server_log, REQUEST) { + let token = accepts(&server_log)[0]; + assert!(server.send_with(token, |buf| buf.extend_from_slice(RESPONSE))); + response_sent = true; + } + if !disconnect_issued && has_message(&client_log, RESPONSE) { + assert!(server.disconnect(accepts(&server_log)[0])); + disconnect_issued = true; + } + if disconnect_issued && + !resent && + accepts(&server_log).len() >= 2 && + connects(&client_log) >= 2 + { + assert!(client.send_with(client_token, |buf| buf.extend_from_slice(AFTER_RECONNECT))); + resent = true; + } + if resent && has_message(&server_log, AFTER_RECONNECT) { + done = true; + } + } + assert!(done, "the two-network exchange did not complete before the deadline"); + + // Verify the initial bidirectional exchange. + assert!(has_message(&server_log, CLIENT_HELLO)); + assert!(has_message(&client_log, SERVER_HELLO)); + assert!(has_message(&server_log, REQUEST)); + assert!(has_message(&client_log, RESPONSE)); + + // The reconnect uses the next server-side token and preserves the client + // token, which is checked in the handler. + let accepted = accepts(&server_log); + assert_eq!(accepted.len(), 2); + assert_eq!(accepted[1].0, accepted[0].0 + 1); + assert_eq!(message_token(&server_log, AFTER_RECONNECT), Some(accepted[1])); + + // A requested disconnect is reported by the next pre_poll; the peer close + // is reported by handle_event. + assert_eq!(disconnect_phase(&server_log, accepted[0]), Some(Phase::Pre)); + assert_eq!(disconnect_phase(&client_log, client_token), Some(Phase::Event)); +} + +#[test] +#[should_panic(expected = "tcp token range 100..102 exhausted")] +fn exhausted_token_range_panics_naming_the_range() { + let poll = Poll::new().unwrap(); + let mut network = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), 100..102); + let group = network.add_group(TcpGroupConfig::default()); + + network.listen(group, unused_addr()).unwrap(); + let second = network.connect(group, unused_addr()); + assert_eq!(second, Token(101), "tokens are assigned in ascending order from the range start"); + + let _ = network.connect(group, unused_addr()); +} + +#[test] +#[cfg_attr(debug_assertions, should_panic(expected = "lies outside this network's token range"))] +fn foreign_token_trips_the_containment_assert() { + let mut poll = Poll::new().unwrap(); + let mut network = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), 100..200); + + // Register a readiness source outside the network's token range. + let mut listener = mio::net::TcpListener::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + poll.registry().register(&mut listener, Token(7), mio::Interest::READABLE).unwrap(); + let _client = std::net::TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + + let mut events = Events::with_capacity(4); + let deadline = Instant::now() + Duration::from_secs(5); + while events.is_empty() && Instant::now() < deadline { + poll.poll(&mut events, Some(Duration::from_millis(10))).unwrap(); + } + let event = events.iter().next().expect("listener readiness did not arrive"); + assert_eq!(event.token(), Token(7)); + + // Debug builds reject the token; release builds emit no TcpEvent for it. + network.handle_event(event, &mut |_| panic!("no TcpEvent expected for a foreign token")); +}