diff --git a/cranelift/runtime/src/dns.rs b/cranelift/runtime/src/dns.rs new file mode 100644 index 00000000..5563581b --- /dev/null +++ b/cranelift/runtime/src/dns.rs @@ -0,0 +1,362 @@ +//! hostname resolution that does not park a green worker. +//! +//! `getaddrinfo` is synchronous and there is no portable way to poll it, so a +//! lookup made from a green task holds the *worker* OS thread for its whole +//! duration — every other task pinned to that worker stalls with it. sockets do +//! not have this problem (a would-block yields to the epoll reactor in +//! `netpoll`), which leaves dns as the one blocking call on the common client +//! path: every grpc, http, and database dial starts with one. +//! +//! the fix is the usual one for a syscall that cannot be polled: keep the +//! blocking call, but make some other thread do it. a small pool of ordinary OS +//! threads owns `to_socket_addrs`; the green task hands over a job, parks, and a +//! pool thread wakes it when the answer is ready. the worker is free the whole +//! time. +//! +//! the park/wake handshake is the same one `netpoll` uses for socket readiness, +//! for the same reason: a wake that lands before the task finishes suspending +//! must not be lost. green records it as pending and the yield arm re-enqueues, +//! so the block site only has to re-check its condition and re-park. +//! +//! under the os-thread backend none of this applies — blocking is what a thread +//! is for, and the pool would only add a handoff — so `resolve` calls +//! `to_socket_addrs` inline there, exactly as the callers used to. + +use crate::concurrency::green; +use crate::concurrency::scheduler::{backend, Backend}; +use std::collections::VecDeque; +use std::net::{SocketAddr, ToSocketAddrs}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard, OnceLock}; + +/// the most OS threads the resolver pool will ever run. a thread per lookup +/// would let a burst of dials (or a hostile one) spawn threads without bound, +/// so lookups past this many in flight queue up instead. +const MAX_THREADS: usize = 4; + +/// the addresses one lookup produced, in resolution order. +/// +/// an empty list means "no address", and covers both a resolver error and a +/// lookup that succeeded with nothing in it. collapsing the two loses nothing: +/// every caller in this runtime already treats them the same (a failed connect, +/// a null resolve result), and it keeps the value that crosses between threads a +/// plain `Vec`. +type Answer = Vec; + +/// resolve `host:port` to its addresses, returning an empty list on failure. +/// +/// this is the only entry point; it picks the offloaded or the inline path +/// itself so callers do not have to care which backend they are on. +pub(crate) fn resolve(host: &str, port: i64) -> Answer { + let target = format!("{}:{}", host, port); + // a numeric address never reaches a resolver: `to_socket_addrs` parses it + // in place, with no syscall to block in, so offloading would buy nothing + // and cost a thread handoff. dialing a literal ip is common enough — tests, + // benchmarks, a sidecar on 127.0.0.1 — to check for first. + if let Ok(addr) = target.parse::() { + return vec![addr]; + } + match offload_target() { + Some(task) => resolve_offloaded(target, task), + None => resolve_inline(&target), + } +} + +/// the green task that should park for this lookup, or `None` when the caller +/// should just block: the os-thread backend, or a green program calling from a +/// thread that is not running a task (main, or an os-thread spawn). +fn offload_target() -> Option { + if backend() != Backend::Green { + return None; + } + green::current_task() +} + +/// the blocking lookup itself, byte-for-byte what the call sites used to do +/// inline. runs on the calling thread under the os-thread backend and on a pool +/// thread under green. +fn resolve_inline(target: &str) -> Answer { + match target.to_socket_addrs() { + Ok(addrs) => addrs.collect(), + Err(_) => Vec::new(), + } +} + +/// hand the lookup to the pool and park the green task until the answer lands. +fn resolve_offloaded(target: String, task: usize) -> Answer { + let slot = Arc::new(Slot { + task, + answer: Mutex::new(None), + }); + let job = Job { + target, + slot: Arc::clone(&slot), + }; + if let Err(job) = pool().submit(job) { + // no pool thread could be started at all. blocking the worker is worse + // than not blocking it, but much better than failing a lookup the + // system can answer. + return resolve_inline(&job.target); + } + + // park until a pool thread fills the slot. a wake that lands before we + // finish suspending is not lost — green flags it as pending and the yield + // arm re-enqueues us instead of parking — and a spurious wake just sends us + // round the loop to park again, the tolerance every green block site has. + loop { + if let Some(answer) = slot.take() { + return answer; + } + green::park_current(); + } +} + +/// where a pool thread leaves the answer for the task that asked for it. +struct Slot { + /// green slab id of the parked task, passed straight to `green::wake`. + task: usize, + /// `None` until the lookup finishes, `Some` (possibly empty) after. + answer: Mutex>, +} + +impl Slot { + /// publish the answer and wake the waiting task. the store happens first so + /// the task cannot resume and find the slot still empty. + fn complete(&self, answer: Answer) { + *lock(&self.answer) = Some(answer); + green::wake(self.task); + } + + /// take the answer if it has arrived. + fn take(&self) -> Option { + lock(&self.answer).take() + } +} + +/// one queued lookup: the `host:port` string to resolve and where to put the +/// answer. +struct Job { + target: String, + slot: Arc, +} + +/// the queue the pool threads share, plus the bookkeeping `submit` needs to +/// decide whether to start another thread. +struct Queue { + jobs: VecDeque, + /// threads started so far. pool threads never retire, so this only grows, + /// and never past `MAX_THREADS`. + threads: usize, + /// threads currently waiting on `ready` for work. + idle: usize, +} + +struct Pool { + queue: Mutex, + ready: Condvar, +} + +/// built on the first offloaded lookup, so a program that never resolves a +/// hostname under green never starts a thread. +static POOL: OnceLock = OnceLock::new(); + +fn pool() -> &'static Pool { + POOL.get_or_init(|| Pool { + queue: Mutex::new(Queue { + jobs: VecDeque::new(), + threads: 0, + idle: 0, + }), + ready: Condvar::new(), + }) +} + +impl Pool { + /// queue a lookup, growing the pool if nothing is free to pick it up. + /// + /// hands the job back only when the pool has no thread at all and the OS + /// refused to give us one — the caller then resolves inline rather than + /// leaving a task parked on an answer that will never come. + fn submit(&'static self, job: Job) -> Result<(), Job> { + let mut queue = lock(&self.queue); + + // grow lazily: one thread is enough for a program that resolves one + // name at a time, and a burst gets up to `MAX_THREADS` of them. + if queue.idle == 0 && queue.threads < MAX_THREADS { + match std::thread::Builder::new() + .name("pith-dns".to_string()) + .spawn(|| self.worker()) + { + Ok(_) => queue.threads += 1, + Err(_) => { + if queue.threads == 0 { + return Err(job); + } + // some other thread is already running; it will get to this + // job when it finishes the one it has. + } + } + } + + queue.jobs.push_back(job); + drop(queue); + self.ready.notify_one(); + Ok(()) + } + + /// a pool thread: resolve queued names forever, waking each waiting task. + /// + /// threads are never retired. the cap is four, an idle one costs a parked + /// futex wait and its stack, and a program that resolved once is very likely + /// to resolve again — a client dials more than one host. + fn worker(&self) { + loop { + let job = self.next_job(); + // a panic here would strand the parked task forever, and unwinding + // out of a pool thread would cross no FFI boundary but would still + // silently shrink the pool. contain it and report the lookup as a + // failure, which is how a resolver error already reads. + let answer = + std::panic::catch_unwind(|| resolve_inline(&job.target)).unwrap_or_default(); + job.slot.complete(answer); + } + } + + /// take the next queued lookup, waiting for one if the queue is empty. + fn next_job(&self) -> Job { + let mut queue = lock(&self.queue); + loop { + if let Some(job) = queue.jobs.pop_front() { + return job; + } + queue.idle += 1; + queue = self + .ready + .wait(queue) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + queue.idle -= 1; + } + } +} + +/// lock through a poisoned mutex: a panicking pool thread must not wedge every +/// later lookup, and none of this state has an invariant a panic could break. +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::{Duration, Instant}; + + /// a hostname whose lookup fails without touching the network: the port is + /// out of range, so `to_socket_addrs` rejects the target before it would + /// query a resolver. keeps the failure tests fast and offline. + const UNPARSEABLE_PORT: i64 = 99999; + + /// run one job through the pool from a plain test thread and wait for the + /// answer. the slot names a task id no green task will ever have; `wake` + /// against an unknown id is a documented no-op, so this exercises the queue, + /// the threads, and the slot without needing a live green task. + fn resolve_via_pool(target: &str) -> Answer { + let slot = Arc::new(Slot { + task: usize::MAX, + answer: Mutex::new(None), + }); + assert!(pool() + .submit(Job { + target: target.to_string(), + slot: Arc::clone(&slot), + }) + .is_ok()); + await_slot(&slot) + } + + /// spin until a slot fills, with a generous ceiling so a hung lookup fails + /// the test instead of hanging the suite. + fn await_slot(slot: &Slot) -> Answer { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Some(answer) = slot.take() { + return answer; + } + assert!(Instant::now() < deadline, "pool never answered"); + std::thread::sleep(Duration::from_millis(1)); + } + } + + #[test] + fn inline_resolve_finds_loopback() { + let addrs = resolve_inline("localhost:80"); + assert!(!addrs.is_empty()); + assert!(addrs.iter().all(|a| a.ip().is_loopback() && a.port() == 80)); + } + + #[test] + fn inline_resolve_reports_failure_as_no_addresses() { + assert!(resolve_inline(&format!("localhost:{}", UNPARSEABLE_PORT)).is_empty()); + } + + #[test] + fn pool_answers_a_lookup() { + let addrs = resolve_via_pool("localhost:8080"); + assert!(!addrs.is_empty()); + assert!(addrs.iter().all(|a| a.ip().is_loopback())); + } + + #[test] + fn pool_reports_failure_as_no_addresses() { + assert!(resolve_via_pool(&format!("localhost:{}", UNPARSEABLE_PORT)).is_empty()); + } + + #[test] + fn pool_never_exceeds_its_thread_cap() { + // more jobs at once than the cap, so the growth check is exercised past + // the point where it must stop starting threads. + let slots: Vec> = (0..MAX_THREADS * 4) + .map(|_| { + let slot = Arc::new(Slot { + task: usize::MAX, + answer: Mutex::new(None), + }); + assert!(pool() + .submit(Job { + target: "localhost:80".to_string(), + slot: Arc::clone(&slot), + }) + .is_ok()); + slot + }) + .collect(); + + for slot in &slots { + assert!(!await_slot(slot).is_empty()); + } + assert!(lock(&pool().queue).threads <= MAX_THREADS); + } + + #[test] + fn resolve_outside_a_green_task_stays_inline() { + // nothing here runs on a green worker, so `resolve` must take the + // blocking path and still produce the same answer. + assert!(offload_target().is_none()); + assert!(!resolve("localhost", 443).is_empty()); + } + + #[test] + fn literal_addresses_skip_the_resolver() { + // both families, and the answer must match what `to_socket_addrs` + // produces for the same string — the fast path is a shortcut, not a + // second implementation. + for (host, port) in [("127.0.0.1", 8080), ("[::1]", 443), ("::1", 443)] { + let target = format!("{}:{}", host, port); + assert_eq!(resolve(host, port), resolve_inline(&target), "{}", target); + } + } + + #[test] + fn out_of_range_ports_still_fail() { + // the literal fast path must not accept what `to_socket_addrs` rejects. + assert!(resolve("127.0.0.1", UNPARSEABLE_PORT).is_empty()); + } +} diff --git a/cranelift/runtime/src/lib.rs b/cranelift/runtime/src/lib.rs index f09b3369..d32842e8 100644 --- a/cranelift/runtime/src/lib.rs +++ b/cranelift/runtime/src/lib.rs @@ -15,6 +15,7 @@ pub mod bytes; pub mod collections; pub mod concurrency; pub mod crypto; +pub mod dns; pub mod encoding; pub mod ffi_util; pub mod handle_registry; diff --git a/cranelift/runtime/src/network.rs b/cranelift/runtime/src/network.rs index 8bee3693..d9f85536 100644 --- a/cranelift/runtime/src/network.rs +++ b/cranelift/runtime/src/network.rs @@ -204,20 +204,15 @@ fn fill_sockaddr( /// then check `SO_ERROR` to distinguish a completed connection from a refusal. /// returns the fd on success or `0` on failure, matching the blocking path. /// -/// dns resolution stays blocking here (getaddrinfo, a documented gap) — only the -/// TCP handshake yields. we try every resolved address in order and keep the -/// first that connects, matching `TcpStream::connect`: a name like `localhost` -/// often resolves to `::1` before `127.0.0.1`, and a server bound only to IPv4 -/// refuses the first before the second succeeds. +/// the name lookup runs on the resolver pool (see `crate::dns`) so getaddrinfo +/// blocks one of its threads rather than this worker. we try every resolved +/// address in order and keep the first that connects, matching +/// `TcpStream::connect`: a name like `localhost` often resolves to `::1` before +/// `127.0.0.1`, and a server bound only to IPv4 refuses the first before the +/// second succeeds. no addresses (a resolver failure or an empty answer) falls +/// through to `0`, as it always has. fn green_connect(host: &str, port: i64) -> i64 { - use std::net::ToSocketAddrs; - - let target = format!("{}:{}", host, port); - let addrs = match target.to_socket_addrs() { - Ok(it) => it, - Err(_) => return 0, - }; - for addr in addrs { + for addr in crate::dns::resolve(host, port) { let fd = green_connect_addr(&addr); if fd != 0 { return fd; @@ -637,23 +632,19 @@ pub extern "C" fn pith_tcp_close(fd: i64) { } /// DNS resolve — resolve hostname to IP address string +/// +/// under the green backend the lookup runs on the resolver pool and this task +/// parks, so the worker stays free; under os threads it blocks here as before. +/// either way a failed lookup and one that returns no address both come back +/// null, which is what the caller already treats as an error. #[no_mangle] pub unsafe extern "C" fn pith_dns_resolve(hostname: *const i8) -> *mut i8 { - use std::net::ToSocketAddrs; - let Some(host) = cstr_str(hostname) else { return std::ptr::null_mut(); }; - let addr_str = format!("{}:0", host); - match addr_str.to_socket_addrs() { - Ok(mut addrs) => { - if let Some(addr) = addrs.next() { - crate::pith_strdup_string(&addr.ip().to_string()) - } else { - std::ptr::null_mut() - } - } - Err(_) => std::ptr::null_mut(), + match crate::dns::resolve(host, 0).first() { + Some(addr) => crate::pith_strdup_string(&addr.ip().to_string()), + None => std::ptr::null_mut(), } } diff --git a/docs/concurrency.md b/docs/concurrency.md index 5af6fb1f..042a5ee0 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -194,6 +194,13 @@ an epoll reactor: a read or write that would block parks the task on the reactor and frees the worker to run something else, so the whole net stack — raw tcp, tls, http/2, grpc — yields the same way without any code of its own. +name resolution gets there by a different route. `getaddrinfo` is synchronous +and there is nothing to poll, so instead of yielding to the reactor the lookup +goes to a small pool of ordinary blocking threads (four at most, started on +demand) and the task parks until one of them answers. the worker stays free for +the whole lookup, so a dial no longer has a blocking step sitting in front of +its non-blocking one. + nothing in your program changes. the same `spawn`, `Channel`, `Mutex`, and `await` run on either backend, and a correct program prints the same thing both ways. the two green examples in this repo run identically with the flag @@ -244,11 +251,11 @@ answer is why the os-thread backend is still here, because it is not just waiting to be deleted. a blocking call on a green worker stalls the worker, not only the task making -it. the epoll reactor covers sockets, so the whole net stack yields, but -nothing else does — dns at dial, file reads and writes, and any slow native -call hold the thread they run on, and every task pinned to that worker waits -behind them. on os threads that same call costs one task, because the task is -a thread and the kernel just runs someone else. +it. the epoll reactor covers sockets and the resolver pool covers dns, so a +client dial yields the whole way through, but nothing else does: file reads and +writes and any slow native call hold the thread they run on, and every task +pinned to that worker waits behind them. on os threads that same call costs one +task, because the task is a thread and the kernel just runs someone else. the reactor is also linux-only. it is epoll and eventfd; macos and the bsds compile a stand-in with no reactor at all, so a green task waiting on a socket @@ -266,8 +273,6 @@ is worth keeping around past the point where one of them is faster. it is off by default and still experimental. the known rough edges: -- dns resolution at dial still blocks the worker (`getaddrinfo` is - synchronous); only the tcp handshake and the bytes after it yield - a compute-bound task that loops without ever touching a channel, await, or socket can now be preempted, but you opt in at build time. compile with `PITH_GREEN_PREEMPT=1` and the backend puts a safe-point at every loop @@ -278,8 +283,8 @@ it is off by default and still experimental. the known rough edges: tight arithmetic loop), and the default build is almost always run os-thread, so it plants no check and pays nothing. code that uses channels and sockets already yields on its own and never needs this. one gap in this first version: - a task sitting inside a long native runtime call (a slow `getaddrinfo`, say) - is not preempted until it returns to pith code and hits the next back-edge + a task sitting inside a long native runtime call (a large file read, say) is + not preempted until it returns to pith code and hits the next back-edge - fewer workers means more locality: a task pins to the first worker that runs it and every later wake goes back to that one worker, never the whole pool, so a coordinated pipeline stays put and its handoffs stay in userspace. at diff --git a/docs/limitations.md b/docs/limitations.md index 510527d0..72803228 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -139,13 +139,11 @@ the scheduler; it is the list below. os threads or the reactor would need a kqueue sibling. - **blocking calls stall a worker, not just their task** — file i/o and native calls have no yield point, so they hold the worker and everything pinned to - it. on os threads a blocking call costs only the task making it. this is the - structural difference that keeps the os-thread backend worth having - regardless of the benchmark numbers. -- **dns still blocks a worker** — `green_connect` resolves with a synchronous - `getaddrinfo` before its non-blocking connect, so a dial that waits on dns - parks the whole worker. the tcp handshake and everything after it already - yield. offloading resolution to a small blocking pool is the fix. + it. dns was the worst case here and is not any more: `getaddrinfo` runs on a + small pool of blocking threads while the calling task parks. on os threads a + blocking call costs only the task making it. this is the structural difference + that keeps the os-thread backend worth having regardless of the benchmark + numbers. - **placement is left to luck** — a task pins to the first worker that runs it, so whether two tasks that talk to each other land together is chance. the fan-out benchmark is bimodal because of it: ~60 ms when the pipeline happens diff --git a/tests/cases/test_dns_resolve_and_dial.pith b/tests/cases/test_dns_resolve_and_dial.pith new file mode 100644 index 00000000..2a877123 --- /dev/null +++ b/tests/cases/test_dns_resolve_and_dial.pith @@ -0,0 +1,54 @@ +# name resolution followed by a dial, on whichever backend is running. +# +# under the green backend the lookup is handed to the runtime's resolver pool +# and the calling task parks until an answer comes back, rather than holding its +# worker inside getaddrinfo. that is a scheduling change and nothing else, so the +# answers below have to be identical to the os-thread run — which is exactly what +# comparing this case against its expected output on both backends checks. +# +# everything here is loopback and hosts-file only, so it is deterministic and +# needs no network. + +import std.net.dns as dns +import std.net.tcp as tcp + +# find a free loopback port so repeated runs never collide with TIME_WAIT. +fn choose_port(start: Int) -> Int!: + mut port := start + while port < start + 200: + probe := tcp.listen("127.0.0.1", port) + if probe.is_ok: + tcp.close(probe.ok) + return port + port = port + 1 + fail "no free tcp port" + +fn client(port: Int) -> Int!: + conn := tcp.connect("localhost", port)! + n := tcp.write(conn, "ping")! + tcp.close(conn) + return n + +fn main() -> Int!: + local := dns.resolve("localhost")! + print("localhost-resolved:{local.len() > 0}") + + # nothing listens on port 1, so every address `localhost` resolves to + # refuses. the lookup still has to succeed for the dial to get that far. + refused := tcp.connect("localhost", 1) + print("refused-dial-fails:{refused.is_err}") + + port := choose_port(19420)! + server := tcp.listen("127.0.0.1", port)! + # bind before spawning the dialer, so the client cannot beat the listener. + dialer := spawn client(port) + + conn := tcp.accept(server)! + data := tcp.read(conn, 16)! + sent := await dialer + print("server-received:{data}") + print("client-sent:{sent.unwrap_or(-1)}") + + tcp.close(conn) + tcp.close(server) + return 0 diff --git a/tests/expected/test_dns_resolve_and_dial.txt b/tests/expected/test_dns_resolve_and_dial.txt new file mode 100644 index 00000000..791830fb --- /dev/null +++ b/tests/expected/test_dns_resolve_and_dial.txt @@ -0,0 +1,4 @@ +localhost-resolved:true +refused-dial-fails:true +server-received:ping +client-sent:4