diff --git a/src/net-node.ts b/src/net-node.ts index ee29795..0b04ecc 100644 --- a/src/net-node.ts +++ b/src/net-node.ts @@ -34,6 +34,9 @@ function parseIp(s: string | undefined): Uint8Array { const ipString = (a: Uint8Array) => (E.isWildcard(a) ? "0.0.0.0" : `${a[0]}.${a[1]}.${a[2]}.${a[3]}`); const peerString = (a: Uint8Array) => (E.isWildcard(a) ? "127.0.0.1" : ipString(a)); +// whether a is an IPv4 address of one of this host's interfaces +const hostHas = (a: Uint8Array) => + Object.values(os.networkInterfaces()).some((list) => list?.some((i) => i.family === "IPv4" && i.address === ipString(a))); function sysfs(name: string, file: string): number | null { try { @@ -54,6 +57,22 @@ interface TcpImpl { srv?: net.Server; werr?: number; resume?: () => void; + // listener: connections the host accepted beyond the backlog, and the clients in this VM connecting to it by local port + overflow?: { c: Sock; pairing?: Pairing }[]; + clients?: Map; + // client: its connect to a listener of this VM + pairing?: Pairing; +} + +// A connect from this VM to a listener of this VM finishes once the listener's accept queue +// admits the connection, as a kernel that drops the SYN of a full queue makes it. +interface Pairing { + client: Sock; + listener: Sock; + port: number; + connected: boolean; + admitted: boolean; + finish?: () => void; } interface UdpImpl { @@ -192,14 +211,21 @@ export class NodeNet implements NetBackend { c.peerInVm = client !== undefined; } this.attach(c, ns); - s.backlog.push(c); + const im = s.impl as TcpImpl; + const pairing = im.clients?.get(ns.remotePort ?? -1); + // the queue holds maxBacklog + 1, as Linux's does + if (s.backlog.length > s.maxBacklog) im.overflow!.push({ c, pairing }); + else { + s.backlog.push(c); + if (pairing) this.admit(pairing); + } s.notify(); }); srv.listen(options, () => { srv.removeAllListeners("error"); srv.on("error", () => {}); bound(srv); - s.impl = { srv } satisfies TcpImpl; + s.impl = { srv, overflow: [], clients: new Map() } satisfies TcpImpl; this.live.add(s); resolve(0); }); @@ -233,11 +259,38 @@ export class NodeNet implements NetBackend { }); } - listen(s: Sock) { + listen(s: Sock, backlog: number) { s.listening = true; + s.maxBacklog = E.backlogLimit(backlog); + this.accepted(s); return 0; } + // The accept queue lost its head: connections waiting beyond the backlog move in, and + // the clients of this VM among them finish connecting. + accepted(l: Sock) { + const q = (l.impl as TcpImpl | null)?.overflow; + while (q && q.length > 0 && l.backlog.length <= l.maxBacklog) { + const { c, pairing } = q.shift()!; + l.backlog.push(c); + l.notify(); + if (pairing) this.admit(pairing); + } + } + + private admit(p: Pairing) { + p.admitted = true; + this.finishPairing(p); + } + + private finishPairing(p: Pairing) { + if (!p.admitted || !p.connected) return; + (p.listener.impl as TcpImpl | null)?.clients?.delete(p.port); + const im = p.client.impl as TcpImpl | null; + if (im) im.pairing = undefined; + p.finish!(); + } + // Wires a connected host socket to s. private attach(s: Sock, ns: net.Socket) { const im: TcpImpl = { ns, resume: () => ns.resume() }; @@ -285,22 +338,56 @@ export class NodeNet implements NetBackend { }, (ns) => { s.local = { addr: parseIp(ns.localAddress), port: ns.localPort ?? 0 }; s.remote = { addr: ep.addr.slice(), port: ep.port }; - }); + }, undefined, this.listenerAt(ep)); } - private connectTo(s: Sock, options: net.NetConnectOpts, connected: (ns: net.Socket) => void, failed?: () => void): number { + private listenerAt(ep: Endpoint): Sock | undefined { + const to = E.isWildcard(ep.addr) ? Uint8Array.of(127, 0, 0, 1) : ep.addr; + let found: Sock | undefined; + for (const o of this.live) { + if (!o.listening || o.unix || o.local?.port !== ep.port || !E.overlap(o.local.addr, to)) continue; + // listeners sharing the port (SO_REUSEPORT): the host picks one, so the connect cannot be held + if (found) return undefined; + found = o; + } + // a wildcard listener matches any address: the connect must go to this host to reach it + if (found && E.isWildcard(found.local!.addr) && to[0] !== 127 && !hostHas(to)) return undefined; + return found; + } + + // listener is a listening socket of this VM that the connect goes to + private connectTo( + s: Sock, options: net.NetConnectOpts, connected: (ns: net.Socket) => void, failed?: () => void, listener?: Sock, + ): number { s.connecting = true; const start = () => { const ns = net.connect({ ...options, allowHalfOpen: true }); this.attach(s, ns); + const im = s.impl as TcpImpl; + let pairing: Pairing | undefined; + // Registered on the next tick: net.connect issues the connect in a tick queued before this one, + // so the local port is known, and ticks run before any I/O callback, so before the listener's + // connection event. That event can still come before this socket's own connect event. + if (listener) process.nextTick(() => { + const clients = (listener.impl as TcpImpl | null)?.clients; + if (!clients || ns.localPort === undefined || s.impl !== im) return; + pairing = im.pairing = { client: s, listener, port: ns.localPort, connected: false, admitted: false }; + clients.set(ns.localPort, pairing); + }); let made = false; if (failed) ns.once("close", () => { if (!made) failed(); }); ns.once("connect", () => { made = true; - s.connecting = false; - s.connected = true; - connected(ns); - s.notify(); + const finish = () => { + s.connecting = false; + s.connected = true; + connected(ns); + s.notify(); + }; + if (!pairing) return finish(); + pairing.finish = finish; + pairing.connected = true; + this.finishPairing(pairing); }); }; // a socket bound before connecting holds its port in a listening server @@ -442,16 +529,35 @@ export class NodeNet implements NetBackend { close(s: Sock) { this.live.delete(s); // a listener resets the connections nobody accepted - for (const b of s.backlog) { - this.live.delete(b); - const ns = (b.impl as TcpImpl | null)?.ns; - b.impl = null; - if (ns) reset(ns); - } + for (const b of s.backlog) this.dropQueued(b); s.backlog = []; const im = s.impl as TcpImpl & UdpImpl | null; s.impl = null; if (!im) return; + // a connect that timed out or was abandoned leaves the listener's queues + const p = im.pairing; + if (p) { + const l = p.listener.impl as TcpImpl | null; + l?.clients?.delete(p.port); + const at = l?.overflow?.findIndex((e) => e.pairing === p) ?? -1; + if (at >= 0) this.dropQueued(l!.overflow!.splice(at, 1)[0].c); + } + // the listener resets what nobody accepted; like a kernel that dropped the SYN, a held connect then gets refused + for (const { c, pairing } of im.overflow ?? []) { + this.dropQueued(c); + const client = pairing?.client; + const ci = client?.impl as TcpImpl | undefined; + if (!client || !ci) continue; + ci.pairing = undefined; + ci.ns?.removeAllListeners(); + ci.ns?.on("error", () => {}); + ci.ns?.destroy(); + client.impl = null; + client.connecting = false; + client.fail(E.ECONNREFUSED); + } + im.overflow = []; + im.clients?.clear(); if (im.srv) { // libuv removes a closed server's socket file, a kernel leaves it const path = im.srv.address(); @@ -469,6 +575,13 @@ export class NodeNet implements NetBackend { else ns.destroy(); } + private dropQueued(c: Sock) { + this.live.delete(c); + const ns = (c.impl as TcpImpl | null)?.ns; + c.impl = null; + if (ns) reset(ns); + } + closeAll() { for (const s of [...this.live]) this.close(s); } diff --git a/src/sockets.ts b/src/sockets.ts index 7e20493..df8c26a 100644 --- a/src/sockets.ts +++ b/src/sockets.ts @@ -303,7 +303,8 @@ export const isWildcard = (a: Uint8Array) => a[0] === 0 && a[1] === 0 && a[2] == export const sameAddr = (a: Uint8Array, b: Uint8Array) => a[0] === b[0] && a[1] === b[1] && a[2] === b[2] && a[3] === b[3]; const WILDCARD = new Uint8Array(4); const SOMAXCONN = 4096; -const overlap = (a: Uint8Array, b: Uint8Array) => isWildcard(a) || isWildcard(b) || sameAddr(a, b); +export const backlogLimit = (backlog: number) => (backlog < 0 ? SOMAXCONN : Math.min(backlog, SOMAXCONN)); +export const overlap = (a: Uint8Array, b: Uint8Array) => isWildcard(a) || isWildcard(b) || sameAddr(a, b); export function ipv4Literal(s: string | undefined): Uint8Array | null { const m = s ? /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s) : null; @@ -390,7 +391,7 @@ export class LoopbackNet implements NetBackend { listen(s: Sock, backlog: number) { s.listening = true; - s.maxBacklog = backlog < 0 ? SOMAXCONN : Math.min(backlog, SOMAXCONN); + s.maxBacklog = backlogLimit(backlog); return 0; } diff --git a/test/java/ListenBacklog.java b/test/java/ListenBacklog.java new file mode 100644 index 0000000..89e98e4 --- /dev/null +++ b/test/java/ListenBacklog.java @@ -0,0 +1,113 @@ +import java.io.IOException; +import java.net.*; +import java.nio.channels.*; +import java.util.*; +import java.util.concurrent.*; + +public class ListenBacklog { + static final InetAddress LO = InetAddress.getLoopbackAddress(); + + public static void main(String[] args) throws Exception { + fullQueueTimesOut(); + nonBlockingBurst(); + acceptMakesRoom(); + listenerCloseRefuses(); + } + + static String connect(int port, int timeoutMs, List keep) { + Socket s = new Socket(); + try { + s.connect(new InetSocketAddress(LO, port), timeoutMs); + keep.add(s); + return "ok"; + } catch (SocketTimeoutException e) { + close(s); + return "timeout"; + } catch (IOException e) { + close(s); + return e.getClass().getSimpleName(); + } + } + + static void close(Socket s) { + try { s.close(); } catch (IOException e) { } + } + + // a listener that never accepts holds backlog + 1 connections, the rest time out + static void fullQueueTimesOut() throws Exception { + List clients = new ArrayList<>(); + try (ServerSocket ss = new ServerSocket(0, 2, LO)) { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < 8; i++) sb.append(connect(ss.getLocalPort(), 300, clients)).append(' '); + System.out.println("backlog 2: " + sb.toString().trim()); + } + clients.forEach(ListenBacklog::close); + } + + // non-blocking connects made at once: backlog + 1 finish, the rest stay pending + static void nonBlockingBurst() throws Exception { + try (ServerSocket ss = new ServerSocket(0, 10, LO); Selector sel = Selector.open()) { + List channels = new ArrayList<>(); + for (int i = 0; i < 40; i++) { + SocketChannel sc = SocketChannel.open(); + sc.configureBlocking(false); + if (!sc.connect(new InetSocketAddress(LO, ss.getLocalPort()))) sc.register(sel, SelectionKey.OP_CONNECT); + channels.add(sc); + } + int finished = 0; + long end = System.currentTimeMillis() + 1500; + while (System.currentTimeMillis() < end) { + sel.select(100); + for (SelectionKey k : sel.selectedKeys()) { + if (((SocketChannel) k.channel()).finishConnect()) { + finished++; + k.cancel(); + } + } + sel.selectedKeys().clear(); + } + System.out.println("burst finished " + finished); + for (SocketChannel sc : channels) sc.close(); + } + } + + // a blocked connect finishes once accept removes a connection from the queue + static void acceptMakesRoom() throws Exception { + List clients = new ArrayList<>(); + try (ServerSocket ss = new ServerSocket(0, 1, LO)) { + int port = ss.getLocalPort(); + int ok = 0; + for (int i = 0; i < 2; i++) if (connect(port, 1000, clients).equals("ok")) ok++; + System.out.println("queued " + ok); + System.out.println("full " + connect(port, 300, clients)); + ExecutorService pool = Executors.newSingleThreadExecutor(); + Future waiting = pool.submit(() -> connect(port, 10000, clients)); + Thread.sleep(300); + System.out.println("still waiting " + !waiting.isDone()); + try (Socket a = ss.accept()) { + System.out.println("accepted " + a.getInetAddress().isLoopbackAddress()); + } + System.out.println("waiting connect " + waiting.get(10, TimeUnit.SECONDS)); + try (Socket a = ss.accept()) { + System.out.println("accepted again " + a.getInetAddress().isLoopbackAddress()); + } + pool.shutdown(); + } + clients.forEach(ListenBacklog::close); + } + + // a listener that closes while a connect waits refuses it + static void listenerCloseRefuses() throws Exception { + List clients = new ArrayList<>(); + ServerSocket ss = new ServerSocket(0, 1, LO); + int port = ss.getLocalPort(); + for (int i = 0; i < 2; i++) connect(port, 1000, clients); + ExecutorService pool = Executors.newSingleThreadExecutor(); + Future waiting = pool.submit(() -> connect(port, 10000, clients)); + Thread.sleep(300); + ss.close(); + System.out.println("after close " + waiting.get(10, TimeUnit.SECONDS)); + pool.shutdown(); + clients.forEach(ListenBacklog::close); + } +} diff --git a/test/jtreg/passing.txt b/test/jtreg/passing.txt index cef9a3a..28bf054 100644 --- a/test/jtreg/passing.txt +++ b/test/jtreg/passing.txt @@ -578,6 +578,7 @@ java/net/httpclient/HeadersTest.java java/net/httpclient/HttpClientAuthRetryLimitTest.java java/net/httpclient/HttpClientBuilderTest.java java/net/httpclient/HttpClientSNITest.java +java/net/httpclient/HttpGetInCancelledFuture.java java/net/httpclient/HttpSlowServerTest.java java/net/httpclient/HttpsTunnelAuthTest.java java/net/httpclient/HttpsTunnelTest.java diff --git a/test/jtreg/times.tsv b/test/jtreg/times.tsv index 6a785fd..a6e465c 100644 --- a/test/jtreg/times.tsv +++ b/test/jtreg/times.tsv @@ -433,6 +433,7 @@ 16965 java/util/Date/DateGregorianCalendarTest.java 16815 javax/crypto/spec/PBEKeySpec/PBEKeySpecRacing.java 16776 java/net/DatagramSocket/DatagramSocketMulticasting.java +16774 java/net/httpclient/HttpGetInCancelledFuture.java 16760 java/lang/module/ModuleDescriptorHashCodeTest.java 16687 java/io/Serializable/GetField/ReadFieldsCNF.java 16658 java/text/Format/common/Bug6215962.java