socketengine: reap TLS conns that stall mid-handshake (tls_handshake_timeout, default 15s) — a connection that opens the TLS port but never negotiates no longer leaks a slot

This commit is contained in:
Jean Chevronnet 2026-08-12 15:51:45 +00:00
parent e74763619b
commit 59e18b0205
3 changed files with 54 additions and 6 deletions

View file

@ -144,6 +144,7 @@ amu_target = both
# threads that frame lines / run TLS crypto off the single state core. One acceptor # threads that frame lines / run TLS crypto off the single state core. One acceptor
# round-robins connections across the pool; the core stays single-threaded and lock-free. # round-robins connections across the pool; the core stays single-threaded and lock-free.
# io_threads = 0 # reactor workers; 0 = auto (one per core, capped at 4) # io_threads = 0 # reactor workers; 0 = auto (one per core, capped at 4)
# tls_handshake_timeout = 15 # drop a TLS conn that stalls mid-handshake (secs; 0 = off)
# --- ident (RFC1413): off by default; a connection class can also enable it --- # --- ident (RFC1413): off by default; a connection class can also enable it ---
# useident = yes # look up every client's ident (adds connect latency) # useident = yes # look up every client's ident (adds connect latency)

View file

@ -58,8 +58,12 @@ fn main() {
}; };
let max_line = raw_num("max_line", socketengine::DEFAULT_MAX_LINE); let max_line = raw_num("max_line", socketengine::DEFAULT_MAX_LINE);
let max_sendq = raw_num("max_sendq", socketengine::DEFAULT_MAX_SENDQ); let max_sendq = raw_num("max_sendq", socketengine::DEFAULT_MAX_SENDQ);
// plaintext reactor-pool size (0 = auto: one worker per core, capped) // reactor-pool size (0 = auto: one worker per core, capped)
let io_threads = raw_num("io_threads", 0); let io_threads = raw_num("io_threads", 0);
// reap a TLS handshake that stalls this long (0 = never); guards the TLS port
// against connections that open but never negotiate
let hs = raw_num("tls_handshake_timeout", 15);
let handshake_timeout = (hs > 0).then(|| Duration::from_secs(hs as u64));
// trusted PROXY-protocol source globs (reactor rewrites the client IP from them) // trusted PROXY-protocol source globs (reactor rewrites the client IP from them)
let proxy_trust: Vec<String> = cfg.raw.get("proxy").cloned().unwrap_or_default(); let proxy_trust: Vec<String> = cfg.raw.get("proxy").cloned().unwrap_or_default();
@ -104,7 +108,8 @@ fn main() {
// reactor worker pool: shared by the plaintext acceptor and the direct-TLS // reactor worker pool: shared by the plaintext acceptor and the direct-TLS
// acceptor, so client I/O (framing + TLS crypto) spreads across cores. // acceptor, so client I/O (framing + TLS crypto) spreads across cores.
let reactors = socketengine::spawn_reactors(tx.clone(), max_line, max_sendq, io_threads); let reactors =
socketengine::spawn_reactors(tx.clone(), max_line, max_sendq, io_threads, handshake_timeout);
// optional TLS listener (bind_tls + tls_cert + tls_key). A cert/bind problem // optional TLS listener (bind_tls + tls_cert + tls_key). A cert/bind problem
// disables TLS but never takes the plaintext listener down. // disables TLS but never takes the plaintext listener down.

View file

@ -22,7 +22,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{self, Receiver, Sender, TryRecvError}; use std::sync::mpsc::{self, Receiver, Sender, TryRecvError};
use std::sync::Arc; use std::sync::Arc;
use std::thread; use std::thread;
use std::time::Duration; use std::time::{Duration, Instant};
use mio::net::{TcpListener as MioListener, TcpStream as MioStream}; use mio::net::{TcpListener as MioListener, TcpStream as MioStream};
use mio::{Events, Interest, Poll, Token, Waker}; use mio::{Events, Interest, Poll, Token, Waker};
@ -250,11 +250,12 @@ pub fn spawn_reactors(
max_line: usize, max_line: usize,
max_sendq: usize, max_sendq: usize,
io_threads: usize, io_threads: usize,
handshake_timeout: Option<Duration>,
) -> Vec<ReactorHandle> { ) -> Vec<ReactorHandle> {
let workers = resolve_io_threads(io_threads); let workers = resolve_io_threads(io_threads);
let mut reactors = Vec::with_capacity(workers); let mut reactors = Vec::with_capacity(workers);
for _ in 0..workers { for _ in 0..workers {
match spawn_reactor(core.clone(), max_line, max_sendq) { match spawn_reactor(core.clone(), max_line, max_sendq, handshake_timeout) {
Ok(h) => reactors.push(h), Ok(h) => reactors.push(h),
Err(e) => eprintln!("reactor: cannot start a worker: {e}"), Err(e) => eprintln!("reactor: cannot start a worker: {e}"),
} }
@ -354,6 +355,7 @@ fn spawn_reactor(
core: Sender<Event>, core: Sender<Event>,
max_line: usize, max_line: usize,
max_sendq: usize, max_sendq: usize,
handshake_timeout: Option<Duration>,
) -> io::Result<ReactorHandle> { ) -> io::Result<ReactorHandle> {
let poll = Poll::new()?; let poll = Poll::new()?;
let waker = Arc::new(Waker::new(poll.registry(), WAKE)?); let waker = Arc::new(Waker::new(poll.registry(), WAKE)?);
@ -364,7 +366,17 @@ fn spawn_reactor(
waker: waker.clone(), waker: waker.clone(),
}; };
thread::spawn(move || { thread::spawn(move || {
reactor_loop(poll, waker, handoff_rx, out_tx, out_rx, core, max_line, max_sendq) reactor_loop(
poll,
waker,
handoff_rx,
out_tx,
out_rx,
core,
max_line,
max_sendq,
handshake_timeout,
)
}); });
Ok(handle) Ok(handle)
} }
@ -380,15 +392,40 @@ fn reactor_loop(
core: Sender<Event>, core: Sender<Event>,
max_line: usize, max_line: usize,
max_sendq: usize, max_sendq: usize,
handshake_timeout: Option<Duration>,
) { ) {
let mut conns: HashMap<usize, Conn> = HashMap::new(); let mut conns: HashMap<usize, Conn> = HashMap::new();
let mut next_token = FIRST_CONN; let mut next_token = FIRST_CONN;
let mut events = Events::with_capacity(1024); let mut events = Events::with_capacity(1024);
// TLS conns still negotiating, with the deadline by which they must finish; a
// stalled handshake holds no uid so nothing else would ever reap it.
let mut pending_hs: Vec<(usize, Instant)> = Vec::new();
loop { loop {
if poll.poll(&mut events, None).is_err() { // block indefinitely when idle; while handshakes are pending, wake ~1s to reap
// any that blew their deadline (slow-loris on the TLS port).
let timeout = (!pending_hs.is_empty()).then(|| Duration::from_millis(1000));
if poll.poll(&mut events, timeout).is_err() {
continue; continue;
} }
if !pending_hs.is_empty() {
let now = Instant::now();
let mut expired = Vec::new();
pending_hs.retain(|&(tok, dl)| match conns.get(&tok) {
Some(c) if c.handshaking => {
if now >= dl {
expired.push(tok);
false
} else {
true
}
}
_ => false, // handshake finished, or the conn is already gone
});
for tok in expired {
close_conn(&mut poll, &mut conns, tok, &core);
}
}
for event in events.iter() { for event in events.iter() {
match event.token() { match event.token() {
WAKE => { WAKE => {
@ -445,6 +482,11 @@ fn reactor_loop(
pending_out: Some(out), pending_out: Some(out),
}, },
); );
if handshaking {
if let Some(d) = handshake_timeout {
pending_hs.push((token, Instant::now() + d));
}
}
// announce now only if nothing defers it: a TLS conn waits for // announce now only if nothing defers it: a TLS conn waits for
// its handshake, a proxy conn for its header. // its handshake, a proxy conn for its header.
if !a.via_proxy && !handshaking { if !a.via_proxy && !handshaking {