From e9d6854e7a2e96a8c63569b56fdc1a5755acf1b1 Mon Sep 17 00:00:00 2001 From: Serkyo Date: Tue, 14 Jul 2026 01:38:50 +0200 Subject: [PATCH] feat(net): add threaded tokio runtime bridge for connections --- crates/net/src/lib.rs | 6 + crates/net/src/runtime.rs | 260 +++++++++++++++++++++++++++++++++++++- 2 files changed, 264 insertions(+), 2 deletions(-) diff --git a/crates/net/src/lib.rs b/crates/net/src/lib.rs index 594e421..7bfe863 100644 --- a/crates/net/src/lib.rs +++ b/crates/net/src/lib.rs @@ -12,6 +12,12 @@ pub mod error; pub mod handshake; pub mod runtime; +pub use runtime::{ConnectOutcome, NetworkServer, ServerEvent, connect_in_background}; + +/// Default UDP port the server binds and the client connects to when none is configured. +// TODO: make the bind address and port configurable through server/client configuration. +pub const DEFAULT_PORT: u16 = 25565; + #[cfg(test)] #[path = "tests/handshake.rs"] mod handshake_tests; diff --git a/crates/net/src/runtime.rs b/crates/net/src/runtime.rs index 8f7328b..b5b021a 100644 --- a/crates/net/src/runtime.rs +++ b/crates/net/src/runtime.rs @@ -1,5 +1,261 @@ // SPDX-License-Identifier: AGPL-3.0-only //! Threaded `tokio` runtime bridge between the async network and the synchronous simulation. -//! -//! Owns the runtime thread and the channels that carry inbound and outbound messages across the async-sync boundary, keeping the async runtime confined to this crate. Populated in a later concept; currently a placeholder. + +use std::net::SocketAddr; +use std::thread; + +use shared::protocol::{ClientHello, HandshakeAck}; +use tracing::{info, warn}; + +use crate::endpoint::{client_endpoint, server_endpoint}; +use crate::error::NetError; +use crate::handshake::{ServerConnection, accept_connection, connect}; + +/// Channel receiver delivering the outcome of a background client connect: the negotiated [`HandshakeAck`] on success, or a human-readable error string on failure. +pub type ConnectOutcome = crossbeam_channel::Receiver>; + +/// An event surfaced by the network thread to the synchronous server loop. +#[derive(Debug)] +pub enum ServerEvent { + /// A client completed the Synvael handshake. Carries the stable per-session id and the `ClientHello` it presented. + ClientConnected { + /// Stable identifier assigned to this session for the lifetime of the connection. + id: u64, + /// The identity and build parameters the client advertised. + hello: ClientHello, + }, + /// A previously connected client's session ended. + ClientDisconnected { + /// Identifier of the session that ended, matching the earlier [`ServerEvent::ClientConnected`]. + id: u64, + /// Human-readable description of why the connection closed. + reason: String, + }, +} + +/// Handle to the background networking thread and its owned `tokio` runtime. +#[derive(Debug)] +pub struct NetworkServer { + /// Events produced by the accept loop, drained by the synchronous simulation thread. + events: crossbeam_channel::Receiver, + /// Shutdown signal. Dropping this sender resolves the accept loop's receiver and breaks the loop. + shutdown: Option>, + /// Join handle for the network thread, awaited on drop for an orderly shutdown. + thread: Option>, +} + +impl NetworkServer { + /// Spawns a dedicated network thread, builds a current-thread `tokio` runtime on it, binds a QUIC server endpoint on `bind`, and runs the accept loop. + /// + /// # Errors + /// + /// Returns [`NetError::Io`] if the runtime cannot be built, the network thread cannot be spawned, or the thread exits before reporting a bound address, and any error from [`server_endpoint`] if the endpoint cannot be constructed or bound. + pub fn spawn( + bind: SocketAddr, + server_build: String, + tick_rate_hint: u16, + ) -> Result<(Self, SocketAddr), NetError> { + let (events_tx, events_rx) = crossbeam_channel::unbounded(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); + // Reports the bound address (or the error that prevented binding) back to this thread, so `spawn` can surface it synchronously. + let (ready_tx, ready_rx) = crossbeam_channel::bounded::>(1); + + let thread = thread::Builder::new() + .name("net-server".to_owned()) + .spawn(move || { + let runtime = match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(runtime) => runtime, + Err(error) => { + let _ = ready_tx.send(Err(NetError::Io(error))); + return; + } + }; + + runtime.block_on(async move { + // The endpoint is constructed inside the runtime context because `quinn` binds its driver task onto the current runtime. + let endpoint = match server_endpoint(bind) { + Ok(endpoint) => endpoint, + Err(error) => { + let _ = ready_tx.send(Err(error)); + return; + } + }; + let local_addr = match endpoint.local_addr() { + Ok(addr) => addr, + Err(error) => { + let _ = ready_tx.send(Err(NetError::Io(error))); + return; + } + }; + // If the caller has already given up, there is nothing to serve. + if ready_tx.send(Ok(local_addr)).is_err() { + return; + } + + accept_loop( + endpoint, + events_tx, + shutdown_rx, + server_build, + tick_rate_hint, + ) + .await; + }); + }) + .map_err(NetError::Io)?; + + let local_addr = match ready_rx.recv() { + Ok(Ok(addr)) => addr, + Ok(Err(error)) => return Err(error), + Err(_) => { + return Err(NetError::Io(std::io::Error::other( + "network thread exited before reporting a bound address", + ))); + } + }; + + Ok(( + Self { + events: events_rx, + shutdown: Some(shutdown_tx), + thread: Some(thread), + }, + local_addr, + )) + } + + /// Drains every [`ServerEvent`] currently queued, without blocking. + pub fn poll_events(&self) -> impl Iterator + '_ { + self.events.try_iter() + } +} + +impl Drop for NetworkServer { + fn drop(&mut self) { + // Dropping the sender resolves the accept loop's shutdown receiver, breaking the loop and letting `block_on` return so the thread unwinds and the runtime is dropped. + self.shutdown.take(); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +/// Accepts incoming QUIC connections until the endpoint stops yielding them or a shutdown is signalled, spawning one handshake task per connection. +async fn accept_loop( + endpoint: quinn::Endpoint, + events: crossbeam_channel::Sender, + mut shutdown: tokio::sync::oneshot::Receiver<()>, + server_build: String, + tick_rate_hint: u16, +) { + // Session ids are handed out sequentially; the accept loop is the sole assigner, so a plain counter suffices. + let mut next_id: u64 = 0; + loop { + tokio::select! { + incoming = endpoint.accept() => { + let Some(incoming) = incoming else { break }; + let id = next_id; + next_id += 1; + tokio::spawn(handle_connection( + incoming, + id, + events.clone(), + server_build.clone(), + tick_rate_hint, + )); + } + // Resolves when the `NetworkServer` handle is dropped (sender gone) or an explicit signal is sent. + _ = &mut shutdown => break, + } + } + info!("network accept loop shutting down"); +} + +/// Performs the handshake for one incoming connection and, on success, reports connect and disconnect events for its session. +async fn handle_connection( + incoming: quinn::Incoming, + id: u64, + events: crossbeam_channel::Sender, + server_build: String, + tick_rate_hint: u16, +) { + match accept_connection(incoming, server_build, tick_rate_hint).await { + Ok(ServerConnection { + connection, hello, .. + }) => { + // If the receiver is gone the server is shutting down; drop the connection silently. + if events + .send(ServerEvent::ClientConnected { id, hello }) + .is_err() + { + return; + } + let reason = connection.closed().await; + let _ = events.send(ServerEvent::ClientDisconnected { + id, + reason: reason.to_string(), + }); + } + Err(error) => { + warn!(%error, id, "connection handshake failed"); + } + } +} + +/// Runs a one-shot connect and Synvael handshake against `server_addr` on a background `tokio` thread, reporting the outcome to the returned receiver. +#[must_use] +pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> ConnectOutcome { + let (outcome_tx, outcome_rx) = crossbeam_channel::bounded(1); + // Retained so a failure to spawn the thread can still be reported to the caller. + let spawn_err_tx = outcome_tx.clone(); + + let spawned = thread::Builder::new() + .name("net-client".to_owned()) + .spawn(move || { + let runtime = match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(runtime) => runtime, + Err(error) => { + let _ = outcome_tx.send(Err(format!("failed to build tokio runtime: {error}"))); + return; + } + }; + + runtime.block_on(async move { + let endpoint = match client_endpoint() { + Ok(endpoint) => endpoint, + Err(error) => { + let _ = outcome_tx.send(Err(error.to_string())); + return; + } + }; + + // The server name matches the self-signed certificate's subject; the current verifier accepts any certificate regardless. + match connect(&endpoint, server_addr, "localhost", hello).await { + Ok(connected) => { + if outcome_tx.send(Ok(connected.ack.clone())).is_err() { + return; + } + // Keep the connection alive on the network thread until the server closes it. A full client session pump is a later concept. + let reason = connected.connection.closed().await; + info!(%reason, "server connection closed"); + } + Err(error) => { + let _ = outcome_tx.send(Err(error.to_string())); + } + } + }); + }); + + if let Err(error) = spawned { + let _ = spawn_err_tx.send(Err(format!("failed to spawn network thread: {error}"))); + } + + outcome_rx +}