scale the socket engine: mio epoll reactor for client connections (tens of thousands on a few threads)

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<String>,
-        sock: TcpStream,
+        out: OutSink,
+        sock: Option<TcpStream>,
         secure: bool,
         link: bool,     // a server-to-server connection, not a client
         outbound: bool, // (link) we dialed them

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<String>,
+    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<String>,
-        _sock: TcpStream, // held by the reader/writer threads; closed gracefully
+        out: OutSink,
+        _sock: Option<TcpStream>, // 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);
         }
     }
 

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();
 }

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<String>,
-        sock: TcpStream,
+        out: OutSink,
+        sock: Option<TcpStream>,
         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,
             },
         );

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<String>),
+    Reactor {
+        token: usize,
+        tx: Sender<Out>,
+        waker: Arc<Waker>,
+    },
+}
+
+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<u8>, // bytes read, awaiting a newline
+    wbuf: Vec<u8>, // 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<Event>, counter: Arc<AtomicU64>) {
+    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::<Out>();
+
+    let mut conns: HashMap<usize, Conn> = 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<usize> = 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<usize, Conn>, t: usize, core: &Sender<Event>) {
+    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<u8> = 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<usize, Conn>, t: usize, core: &Sender<Event>) {
+    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<usize, Conn>, t: usize, core: &Sender<Event>) {
+    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<Event>,
@@ -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<Event>, counter: Arc<AtomicU64>) {
         .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<Event>, counter: Arc<AtomicU64>) {
     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<Event>) {
     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,

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<String>,
+    pub out: OutSink,
     pub sock: Option<TcpStream>, // core-side fd handle; dropped on quit so the
                                  // writer thread flushes then closes (None in tests)
 }