diff --git a/Cargo.toml b/Cargo.toml index 12caead..d6cad3e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,11 @@ path = "src/lib.rs" # openssl backend — the crate keeps all `unsafe` internal, so the daemon stays # `#![forbid(unsafe_code)]`). A pure-Rust `rustls` backend can slot in beside it. openssl = "0.10" +# epoll/kqueue reactor for the client socket engine — the minimal readiness layer +# Tokio itself is built on. Lets one thread drive tens of thousands of connections +# instead of 2 OS threads per client. Its `unsafe` stays internal (like openssl), +# so the daemon is still `#![forbid(unsafe_code)]`; no async runtime is pulled in. +mio = { version = "1", features = ["os-poll", "net"] } [profile.release] opt-level = 3 diff --git a/scripts/native-rust-guard.sh b/scripts/native-rust-guard.sh index e56286b..10b60e9 100755 --- a/scripts/native-rust-guard.sh +++ b/scripts/native-rust-guard.sh @@ -30,11 +30,11 @@ ffi=$(grep -rnE 'extern[[:space:]]+"C"|\blibc::|std::ffi|#\[no_mangle\]' src/ 2> # 4. dependency-light — only openssl is allowed as an external crate deps=$(awk '/^\[dependencies\]/{f=1;next} /^\[/{f=0} f && NF {print}' Cargo.toml 2>/dev/null \ - | grep -vE '^[[:space:]]*#' | sed -E 's/[[:space:]=].*//' | grep -vE '^(openssl)?$') -[ -n "$deps" ] && flag "unexpected dependency (only openssl allowed):" "$deps" + | grep -vE '^[[:space:]]*#' | sed -E 's/[[:space:]=].*//' | grep -vE '^(openssl|mio)?$') +[ -n "$deps" ] && flag "unexpected dependency (only openssl + mio allowed):" "$deps" if [ "$fail" -eq 0 ]; then - echo "native-rust-guard: OK — original Rust, no-unsafe, no C/FFI, openssl-only." + echo "native-rust-guard: OK — original Rust, no-unsafe, no C/FFI, openssl+mio only." exit 0 fi echo "native-rust-guard: FAILED — see violations above." >&2 diff --git a/src/ircd.rs b/src/ircd.rs index 5c01dca..5f97f8a 100644 --- a/src/ircd.rs +++ b/src/ircd.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use std::net::{SocketAddr, TcpStream}; -use std::sync::mpsc::{Receiver, Sender}; +use std::sync::mpsc::Receiver; use crate::command::Command; use crate::config::Config; @@ -13,6 +13,7 @@ use crate::message; use crate::module::{Hook, ModResult, Module}; use crate::numeric::{ERR_NEEDMOREPARAMS, ERR_NOTREGISTERED, ERR_UNKNOWNCOMMAND}; use crate::server::Server; +use crate::socketengine::OutSink; use crate::Uid; /// What the I/O threads hand to the core. @@ -20,8 +21,8 @@ pub enum Event { Connect { uid: Uid, addr: SocketAddr, - out: Sender, - sock: TcpStream, + out: OutSink, + sock: Option, secure: bool, link: bool, // a server-to-server connection, not a client outbound: bool, // (link) we dialed them diff --git a/src/link.rs b/src/link.rs index bd3855c..8761c20 100644 --- a/src/link.rs +++ b/src/link.rs @@ -20,20 +20,20 @@ //! CAPAB/FJOIN/metadata wire format. Also: SASL relays here once a services links in. use std::net::{SocketAddr, TcpStream}; -use std::sync::mpsc::Sender; use std::collections::HashSet; use crate::channels::{Ban, Channel, Member, Topic}; use crate::message::Message; use crate::server::{now, Server}; +use crate::socketengine::OutSink; use crate::users::User; use crate::Uid; /// A local server-link connection (one hop away). Distinct from a client `User`. pub struct Link { pub uid: Uid, - pub out: Sender, + pub out: OutSink, pub outbound: bool, // we dialed them (so we introduce ourselves first) pub registered: bool, // handshake complete pub sent_server: bool, // we've sent our own SERVER line @@ -101,8 +101,8 @@ impl Server { &mut self, uid: Uid, addr: SocketAddr, - out: Sender, - _sock: TcpStream, // held by the reader/writer threads; closed gracefully + out: OutSink, + _sock: Option, // held by the reader/writer threads; closed gracefully outbound: bool, ) { let mut sent_server = false; @@ -113,7 +113,7 @@ impl Server { .find(|b| b.ip == addr.ip().to_string()) .map(|b| b.password.clone()); if let Some(pass) = pass { - let _ = out.send(format!( + out.send(format!( "SERVER {} {} {} :{}", self.name, pass, self.sid, self.server_desc )); @@ -137,7 +137,7 @@ impl Server { fn link_out(&self, uid: Uid, line: String) { if let Some(l) = self.links.get(&uid) { - let _ = l.out.send(line); + l.out.send(line); } } diff --git a/src/main.rs b/src/main.rs index ad23b3f..27d5409 100644 --- a/src/main.rs +++ b/src/main.rs @@ -20,7 +20,16 @@ fn main() { .unwrap_or_else(|| "echoircd.conf".to_string()); let cfg = Config::load(&path); - let listener = match TcpListener::bind(&cfg.bind) { + // Client plaintext connections run on the mio reactor, so bind a mio listener + // (fail fast if the main port is taken). + let bind_addr: std::net::SocketAddr = match cfg.bind.parse() { + Ok(a) => a, + Err(e) => { + eprintln!("echoircd: bad bind address {}: {e}", cfg.bind); + std::process::exit(1); + } + }; + let client_listener = match mio::net::TcpListener::bind(bind_addr) { Ok(l) => l, Err(e) => { eprintln!("echoircd: cannot bind {}: {e}", cfg.bind); @@ -101,6 +110,7 @@ fn main() { }); } - socketengine::accept_loop(listener, tx, None, counter, false); + // client plaintext connections: one mio reactor thread drives them all + thread::spawn(move || socketengine::run_reactor(client_listener, tx, counter)); let _ = core.join(); } diff --git a/src/server.rs b/src/server.rs index 3eb542e..7fa53c6 100644 --- a/src/server.rs +++ b/src/server.rs @@ -7,7 +7,6 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::net::{SocketAddr, TcpStream}; -use std::sync::mpsc::Sender; use std::time::{SystemTime, UNIX_EPOCH}; use crate::channels::Channel; @@ -15,6 +14,7 @@ use crate::config::{Config, LinkBlock}; use crate::extensible::Extensible; use crate::link::{Link, RemoteServer, RemoteUser}; use crate::module::Hook; +use crate::socketengine::OutSink; use crate::users::{Caps, User, UserFlags}; use crate::xline::XLine; use crate::Uid; @@ -160,8 +160,8 @@ impl Server { &mut self, uid: Uid, addr: SocketAddr, - out: Sender, - sock: TcpStream, + out: OutSink, + sock: Option, secure: bool, ) { let uuid = self.next_uuid(); @@ -197,7 +197,7 @@ impl Server { ping_sent: false, ext: Extensible::default(), out, - sock: Some(sock), + sock, }, ); } @@ -268,7 +268,7 @@ impl Server { } else { line }; - let _ = u.out.send(line); + u.out.send(line); } } @@ -346,7 +346,7 @@ impl Server { } else { format!("@{} {body}", tags.join(";")) }; - let _ = u.out.send(line); + u.out.send(line); } } @@ -498,7 +498,7 @@ mod tests { last_active: 0, ping_sent: false, ext: Extensible::default(), - out: tx, + out: OutSink::Thread(tx), sock: None, }, ); diff --git a/src/socketengine.rs b/src/socketengine.rs index 9b44171..290eb66 100644 --- a/src/socketengine.rs +++ b/src/socketengine.rs @@ -1,11 +1,19 @@ -//! The socket engine: the I/O edge. Accept connections and, per socket, ferry -//! the wire to/from the core. Plaintext sockets get a blocking reader thread + -//! writer thread; TLS sockets get one thread that owns the session and polls -//! (a single TLS object can't be split across two threads). The core never -//! touches a socket except to shut it down. (InspIRCd has a `socketengines/` -//! dir of epoll/kqueue/select backends; ours is threads.) +//! The socket engine: the I/O edge. Two coexisting models feed the one core: +//! +//! - **Client plaintext** connections run on a single **mio epoll reactor** +//! ([`run_reactor`]) — one thread drives tens of thousands of sockets, so the +//! daemon scales to ~50k users without a thread per connection. This is the +//! same readiness layer Tokio is built on; the core stays single-threaded and +//! there is no async runtime. +//! - **TLS** and **server links** keep a thread per connection (few of them, and +//! a TLS session can't be split across reader+writer threads). +//! +//! Both hand the core the same [`OutSink`] output handle, so the core never +//! knows or cares which model a connection uses. (InspIRCd has a `socketengines/` +//! dir of epoll/kqueue/select backends; this is ours, written from scratch.) -use std::io::{self, BufRead, BufReader, Write}; +use std::collections::{HashMap, HashSet}; +use std::io::{self, BufRead, BufReader, Read, Write}; use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc::{self, Receiver, Sender, TryRecvError}; @@ -13,18 +21,333 @@ use std::sync::Arc; use std::thread; use std::time::Duration; +use mio::net::{TcpListener as MioListener, TcpStream as MioStream}; +use mio::{Events, Interest, Poll, Token, Waker}; + use crate::ircd::Event; use crate::tls::TlsBackend; use crate::Uid; /// Longest single line we'll buffer before dropping it (crude flood guard). const MAX_LINE: usize = 16 * 1024; +/// Most bytes we'll queue to a slow client before dropping them (backpressure). +const MAX_WBUF: usize = 1 << 20; // 1 MiB /// How long a TLS thread blocks on a read before draining its write queue. const TLS_POLL: Duration = Duration::from_millis(100); -/// Accept forever, wiring each connection to the core. `tls` = the backend to -/// wrap sockets in (None for a plaintext listener). `counter` is shared across -/// every listener so uids stay unique. +/// A queued output action the core hands the reactor: a line to write to a +/// connection, or a request to flush-then-close it (sent when the core drops the +/// [`OutSink`], e.g. on quit). +pub enum Out { + Line(usize, String), + Close(usize), +} + +/// The core's handle to one connection's output. Thread-model connections (TLS, +/// server links) get a plain channel to their writer thread; reactor connections +/// (plaintext clients) get a token plus the shared reactor channel and its waker. +/// Either way the core just calls [`OutSink::send`]. +pub enum OutSink { + Thread(Sender), + Reactor { + token: usize, + tx: Sender, + waker: Arc, + }, +} + +impl OutSink { + /// Queue one line for delivery (the writer appends CRLF). + pub fn send(&self, line: String) { + match self { + OutSink::Thread(s) => { + let _ = s.send(line); + } + OutSink::Reactor { token, tx, waker } => { + if tx.send(Out::Line(*token, line)).is_ok() { + let _ = waker.wake(); // wakes coalesce: many sends → one epoll wakeup + } + } + } + } +} + +impl Drop for OutSink { + fn drop(&mut self) { + // The core dropping this handle means "this connection is done". For the + // thread model, dropping the Sender ends the writer loop (which flushes + // first). For the reactor, ask it to flush any queued lines then close. + if let OutSink::Reactor { token, tx, waker } = self { + let _ = tx.send(Out::Close(*token)); + let _ = waker.wake(); + } + } +} + +// === mio reactor: all client plaintext connections on one thread ============= + +const LISTENER: Token = Token(0); +const WAKE: Token = Token(1); +const FIRST_CONN: usize = 16; // conn tokens start past the reserved ones + +struct Conn { + stream: MioStream, + uid: Uid, + rbuf: Vec, // bytes read, awaiting a newline + wbuf: Vec, // bytes queued to write + wpos: usize, // how far into wbuf we've written + want_write: bool, + closing: bool, // flush wbuf, then close +} + +impl Conn { + fn pending(&self) -> usize { + self.wbuf.len() - self.wpos + } +} + +/// Run the client plaintext reactor on this thread. `listener` is an already-bound +/// mio listener (bound in `main` so a bind failure is fatal and fails fast). +pub fn run_reactor(mut listener: MioListener, core: Sender, counter: Arc) { + let mut poll = match Poll::new() { + Ok(p) => p, + Err(e) => { + eprintln!("reactor: cannot create poll: {e}"); + return; + } + }; + if poll + .registry() + .register(&mut listener, LISTENER, Interest::READABLE) + .is_err() + { + eprintln!("reactor: cannot register listener"); + return; + } + let waker = match Waker::new(poll.registry(), WAKE) { + Ok(w) => Arc::new(w), + Err(e) => { + eprintln!("reactor: cannot create waker: {e}"); + return; + } + }; + let (out_tx, out_rx) = mpsc::channel::(); + + let mut conns: HashMap = HashMap::new(); + let mut next_token = FIRST_CONN; + let mut events = Events::with_capacity(1024); + + loop { + if poll.poll(&mut events, None).is_err() { + continue; + } + for event in events.iter() { + match event.token() { + LISTENER => loop { + match listener.accept() { + Ok((mut stream, _addr)) => { + let _ = stream.set_nodelay(true); + let token = next_token; + next_token += 1; + if poll + .registry() + .register(&mut stream, Token(token), Interest::READABLE) + .is_err() + { + continue; + } + let uid = counter.fetch_add(1, Ordering::Relaxed); + let addr = stream + .peer_addr() + .unwrap_or_else(|_| "0.0.0.0:0".parse().unwrap()); + conns.insert( + token, + Conn { + stream, + uid, + rbuf: Vec::new(), + wbuf: Vec::new(), + wpos: 0, + want_write: false, + closing: false, + }, + ); + let out = OutSink::Reactor { + token, + tx: out_tx.clone(), + waker: waker.clone(), + }; + if core + .send(Event::Connect { + uid, + addr, + out, + sock: None, + secure: false, + link: false, + outbound: false, + }) + .is_err() + { + return; // core gone + } + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, + Err(_) => break, + } + }, + WAKE => { + // drain everything the core queued, then flush the touched conns + let mut touched: HashSet = HashSet::new(); + while let Ok(msg) = out_rx.try_recv() { + match msg { + Out::Line(t, line) => { + if let Some(c) = conns.get_mut(&t) { + if c.pending() + line.len() + 2 > MAX_WBUF { + // slow client: drop queued data and close + c.wbuf.clear(); + c.wpos = 0; + c.closing = true; + } else { + if c.wpos > 0 { + c.wbuf.drain(..c.wpos); // reclaim written prefix + c.wpos = 0; + } + c.wbuf.extend_from_slice(line.as_bytes()); + c.wbuf.extend_from_slice(b"\r\n"); + } + touched.insert(t); + } + } + Out::Close(t) => { + if let Some(c) = conns.get_mut(&t) { + c.closing = true; + touched.insert(t); + } + } + } + } + for t in touched { + flush_conn(&mut poll, &mut conns, t, &core); + } + } + Token(t) => { + if event.is_readable() { + read_conn(&mut poll, &mut conns, t, &core); + } + if event.is_writable() && conns.contains_key(&t) { + flush_conn(&mut poll, &mut conns, t, &core); + } + } + } + } + } +} + +/// Drain readable bytes from `t` (edge-triggered: read until WouldBlock), frame +/// complete lines and forward them to the core; close on EOF/error. +fn read_conn(poll: &mut Poll, conns: &mut HashMap, t: usize, core: &Sender) { + let mut chunk = [0u8; 8192]; + let mut lines: Vec<(Uid, String)> = Vec::new(); + let mut close = false; + if let Some(c) = conns.get_mut(&t) { + loop { + match c.stream.read(&mut chunk) { + Ok(0) => { + close = true; + break; + } + Ok(n) => { + c.rbuf.extend_from_slice(&chunk[..n]); + while let Some(pos) = c.rbuf.iter().position(|&b| b == b'\n') { + let raw: Vec = c.rbuf.drain(..=pos).collect(); + let text = String::from_utf8_lossy(&raw); + let l = text.trim_end_matches(['\r', '\n']); + if !l.is_empty() { + lines.push((c.uid, l.to_string())); + } + } + if c.rbuf.len() > MAX_LINE { + c.rbuf.clear(); // overlong line with no newline: drop it + } + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(_) => { + close = true; + break; + } + } + } + } + for (uid, line) in lines { + if core.send(Event::Line { uid, line }).is_err() { + return; + } + } + if close { + close_conn(poll, conns, t, core); + } +} + +/// Write as much of `t`'s queued output as the socket accepts, adjust WRITABLE +/// interest, and close once a `closing` connection's buffer is drained. +fn flush_conn(poll: &mut Poll, conns: &mut HashMap, t: usize, core: &Sender) { + let mut close = false; + if let Some(c) = conns.get_mut(&t) { + while c.wpos < c.wbuf.len() { + match c.stream.write(&c.wbuf[c.wpos..]) { + Ok(0) => break, + Ok(n) => c.wpos += n, + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(_) => { + close = true; + break; + } + } + } + if c.wpos == c.wbuf.len() { + c.wbuf.clear(); + c.wpos = 0; + } + // re-arm WRITABLE only while there's a backlog (edge-triggered) + let want = !c.wbuf.is_empty(); + if want != c.want_write { + c.want_write = want; + let interest = if want { + Interest::READABLE | Interest::WRITABLE + } else { + Interest::READABLE + }; + let _ = poll + .registry() + .reregister(&mut c.stream, Token(t), interest); + } + if c.closing && c.wbuf.is_empty() { + close = true; + } + } + if close { + close_conn(poll, conns, t, core); + } +} + +/// Deregister + drop `t`'s socket and tell the core the connection is gone. +fn close_conn(poll: &mut Poll, conns: &mut HashMap, t: usize, core: &Sender) { + if let Some(mut c) = conns.remove(&t) { + let _ = poll.registry().deregister(&mut c.stream); + let uid = c.uid; + drop(c); // closes the socket + let _ = core.send(Event::Disconnect { uid }); + } +} + +// === thread model: TLS + server links ======================================== + +/// Accept forever on a thread-per-connection listener (TLS or S2S). `tls` is the +/// backend to wrap sockets in (None ⇒ plaintext link). `counter` is shared with +/// the reactor so uids stay unique across every listener. pub fn accept_loop( listener: TcpListener, core: Sender, @@ -54,8 +377,8 @@ pub fn accept_loop( .send(Event::Connect { uid, addr, - out: out_tx, - sock: shutdown, + out: OutSink::Thread(out_tx), + sock: Some(shutdown), secure: false, link, outbound: false, @@ -101,8 +424,8 @@ pub fn connect_link(addr: &str, core: Sender, counter: Arc) { .send(Event::Connect { uid, addr: peer, - out: out_tx, - sock: shutdown, + out: OutSink::Thread(out_tx), + sock: Some(shutdown), secure: false, link: true, outbound: true, @@ -114,7 +437,7 @@ pub fn connect_link(addr: &str, core: Sender, counter: Arc) { thread::spawn(move || reader_loop(reader, uid, core)); } -// --- plaintext: two blocking threads ---------------------------------------- +// --- plaintext link: two blocking threads ----------------------------------- fn reader_loop(stream: TcpStream, uid: Uid, core: Sender) { let mut buf = BufReader::new(stream); @@ -183,8 +506,8 @@ fn tls_conn( .send(Event::Connect { uid, addr, - out: out_tx, - sock: shutdown, + out: OutSink::Thread(out_tx), + sock: Some(shutdown), secure: true, link, outbound: false, diff --git a/src/users.rs b/src/users.rs index 3a937c8..ec763d6 100644 --- a/src/users.rs +++ b/src/users.rs @@ -4,12 +4,12 @@ use std::collections::HashSet; use std::net::{SocketAddr, TcpStream}; -use std::sync::mpsc::Sender; use crate::extensible::Extensible; use crate::module::Hook; use crate::numeric::*; use crate::server::{Server, VERSION}; +use crate::socketengine::OutSink; use crate::Uid; /// User modes and session flags. Kept in one `Default` bag so adding a mode @@ -219,7 +219,7 @@ pub struct User { pub last_active: u64, // unix secs of the last line we received pub ping_sent: bool, // a server PING is outstanding pub ext: Extensible, // typed, module-owned per-user metadata - pub out: Sender, + pub out: OutSink, pub sock: Option, // core-side fd handle; dropped on quit so the // writer thread flushes then closes (None in tests) }