From b0baa3ea98130cb4ca5cbd2c30ca64554755828a Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 08:03:51 -1000 Subject: [PATCH 1/8] harden UDP parsing and add caps Co-authored-by: Cursor --- src/netty.nim | 229 ++++++++++++++++++---------- tests/fuzz.nim | 155 +++++++++++++++++++ tests/test.nim | 27 ++-- tests/test_hardening.nim | 316 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 638 insertions(+), 89 deletions(-) create mode 100644 tests/fuzz.nim create mode 100644 tests/test_hardening.nim diff --git a/src/netty.nim b/src/netty.nim index 5c9fe74..0f1c769 100644 --- a/src/netty.nim +++ b/src/netty.nim @@ -1,20 +1,25 @@ -import flatty/binny, hashes, nativesockets, net, netty/timeseries, random, - sequtils, std/monotimes, strformat, times, os +import + flatty/binny, hashes, nativesockets, net, netty/timeseries, random, + sequtils, std/monotimes, strformat, times, os export Port, timeseries const - partMagic = 0xFFDDFF33.uint32 - ackMagic = 0xFF33FF11.uint32 - disconnectMagic = 0xFF77FF99.uint32 - punchMagic = 0x00000000.uint32 - headerSize = 4 + 4 + 4 + 2 + 2 - ackTime = 0.250 ## Seconds to wait before sending the packet again. - connTimeout = 10.00 ## Seconds to wait until timing-out the connection. - defaultMaxUdpPacket = 508 - headerSize - defaultMaxInFlight = 25_000 + PartMagic = 0xFFDDFF33.uint32 + AckMagic = 0xFF33FF11.uint32 + DisconnectMagic = 0xFF77FF99.uint32 + PunchMagic = 0x00000000.uint32 + HeaderSize = 4 + 4 + 4 + 2 + 2 + AckTime = 0.250 ## Seconds to wait before sending the packet again. + ConnTimeout = 10.00 ## Seconds to wait until timing-out the connection. + DefaultMaxUdpPacket = 508 - HeaderSize + DefaultMaxInFlight = 25_000 + DefaultMaxConnections = 1000 + DefaultMaxRecvParts = 1000 type + NettyError* = object of CatchableError + Address* = object ## A host/port of the client. host*: string @@ -25,7 +30,6 @@ type dropRate*: float32 ## [0, 1] % simulated drop rate. readLatency*: float32 ## Min simulated read latency in seconds. sendLatency*: float32 ## Min simulated send latency in seconds. - maxUdpPacket*: int ## Max size of each outgoing UDP packet in bytes. Reactor* = ref object ## Main networking system that can open or receive connections. @@ -34,7 +38,10 @@ type address*: Address socket: Socket time: float64 - maxInFlight*: int ## Max bytes in-flight on the socket. + maxInFlight*: int ## Max bytes in-flight on the socket. + maxConnections*: int ## Max accepted or outbound connections. + maxRecvParts*: int ## Max buffered receive parts per connection. + maxUdpPacket*: int ## Max payload bytes per outgoing UDP packet. debug*: DebugConfig connections*: seq[Connection] @@ -104,8 +111,19 @@ func hash*(x: Address): Hash = ## Computes a hash for the address. hash((x.host, x.port)) +func seqLess(a, b: uint32): bool {.inline.} = + ## True if a is before b in RFC1982 serial number space. + cast[int32](a - b) < 0 + func genId(reactor: Reactor): uint32 {.inline.} = - reactor.r.rand(0u32..uint32.high).uint32 + reactor.r.rand(0u32 .. uint32.high).uint32 + +func currentTime(reactor: Reactor): float64 {.inline.} = + ## Prefer debug tick override when set so send() matches tick time. + if reactor.debug.tickTime != 0: + reactor.debug.tickTime + else: + reactor.time func newConnection(reactor: Reactor, address: Address): Connection = result = Connection() @@ -121,6 +139,23 @@ func getConn(reactor: Reactor, connId: uint32): Connection = if conn.id == connId: return conn +func updateSendStats(reactor: Reactor) = + ## Recount in-flight bytes after sends or acks. + for conn in reactor.connections: + var + inFlight: int + saturated: bool + for part in conn.sendParts: + if part.acked: + continue + if inFlight + part.data.len > reactor.maxInFlight: + saturated = true + break + if part.sentTime != 0: + inFlight += part.data.len + conn.stats.inFlight = inFlight + conn.stats.saturated = saturated + func read(reactor: Reactor, conn: Connection): (bool, Message) = if conn.recvParts.len == 0: return @@ -129,17 +164,21 @@ func read(reactor: Reactor, conn: Connection): (bool, Message) = sequenceNum = conn.recvSequenceNum numParts = conn.recvParts[0].numParts - if conn.recvParts.len < numParts.int: + if numParts == 0 or conn.recvParts.len < numParts.int: return var good = true for i in 0.uint16 ..< numParts: - if conn.recvParts[i].ackedTime + reactor.debug.readLatency > reactor.time: + if conn.recvParts[i].ackedTime + reactor.debug.readLatency > + reactor.time: + good = false break - if not(conn.recvParts[i].sequenceNum == sequenceNum and + if not( + conn.recvParts[i].sequenceNum == sequenceNum and conn.recvParts[i].numParts == numParts and - conn.recvParts[i].partNum == i): + conn.recvParts[i].partNum == i + ): good = false break @@ -154,11 +193,12 @@ func read(reactor: Reactor, conn: Connection): (bool, Message) = result[1].data.add(conn.recvParts[i].data) inc conn.recvSequenceNum - conn.recvParts.delete(0, numParts - 1) + conn.recvParts.delete(0 .. numParts.int - 1) func divideAndSend(reactor: Reactor, conn: Connection, data: string) = ## Divides a packet into parts and gets it ready to be sent. - assert data.len != 0 + if data.len == 0: + return conn.stats.inQueue += data.len var @@ -173,16 +213,17 @@ func divideAndSend(reactor: Reactor, conn: Connection, data: string) = part.partNum = partNum inc partNum - let maxAt = min(at + reactor.debug.maxUdpPacket, data.len) + let maxAt = min(at + reactor.maxUdpPacket, data.len) part.data = data[at ..< maxAt] at = maxAt parts.add(part) - assert parts.len < high(uint16).int + if parts.len > high(uint16).int: + raise newException(NettyError, "message has too many parts") for part in parts.mitems: part.numParts = parts.len.uint16 - part.queuedTime = reactor.time + part.queuedTime = reactor.currentTime() conn.sendParts.add(parts) inc conn.sendSequenceNum @@ -194,32 +235,37 @@ proc rawSend(reactor: Reactor, address: Address, packet: string) = return try: reactor.socket.sendTo(address.host, address.port, packet) - except: + except OSError: return proc sendNeededParts(reactor: Reactor) = for conn in reactor.connections: - var - inFlight: int - saturated: bool for part in conn.sendParts: - if inFlight + part.data.len > reactor.maxInFlight: - saturated = true + if part.acked: + continue + + if conn.stats.inFlight + part.data.len > reactor.maxInFlight: + conn.stats.saturated = true break - if part.acked or (part.sentTime + ackTime > reactor.time): + # Waiting for ACK / RTO; still counts as in-flight. + if part.sentTime != 0 and + part.sentTime + AckTime > reactor.time: + conn.stats.inFlight += part.data.len continue if part.queuedTime + reactor.debug.sendLatency > reactor.time: continue - inFlight += part.data.len - conn.stats.inQueue -= part.data.len + let firstSend = part.sentTime == 0 + conn.stats.inFlight += part.data.len + if firstSend: + conn.stats.inQueue -= part.data.len part.sentTime = reactor.time - var packet = newStringOfCap(headerSize + part.data.len) - packet.addUint32(partMagic) + var packet = newStringOfCap(HeaderSize + part.data.len) + packet.addUint32(PartMagic) packet.addUint32(part.sequenceNum) packet.addUint32(part.connId) packet.addUint16(part.partNum) @@ -228,8 +274,7 @@ proc sendNeededParts(reactor: Reactor) = reactor.rawSend(conn.address, packet) - conn.stats.inFlight = inFlight - conn.stats.saturated = saturated + reactor.updateSendStats() proc sendSpecial( reactor: Reactor, conn: Connection, part: Part, magic: uint32 @@ -237,7 +282,7 @@ proc sendSpecial( assert reactor.id == conn.reactorId assert conn.id == part.connId - var packet = newStringOfCap(headerSize) + var packet = newStringOfCap(HeaderSize) packet.addUint32(magic) packet.addUint32(part.sequenceNum) packet.addUint32(part.connId) @@ -262,13 +307,13 @@ func deleteAckedParts(reactor: Reactor) = minTime = min(minTime, part.queuedTime) conn.stats.latencyTs.add((reactor.time - minTime).float32) - conn.sendParts.delete(0, pos - 1) + conn.sendParts.delete(0 .. pos - 1) conn.stats.throughputTs.add(reactor.time, bytesAcked.float64) proc readParts(reactor: Reactor) = var - buf = newStringOfCap(reactor.debug.maxUdpPacket + headerSize) + buf = newStringOfCap(reactor.maxUdpPacket + HeaderSize) host: string port: Port @@ -276,17 +321,23 @@ proc readParts(reactor: Reactor) = var byteLen: int try: byteLen = reactor.socket.recvFrom( - buf, reactor.debug.maxUdpPacket + headerSize, host, port + buf, reactor.maxUdpPacket + HeaderSize, host, port ) - except: + except OSError: when defined(nettyMagicSleep): sleep(1) break + if byteLen < 4: + # Need a magic word at minimum. + continue + let address = initAddress(host, port.int) + let magic = buf.readUint32(0) - var magic = buf.readUint32(0) - if magic == disconnectMagic: + if magic == DisconnectMagic: + if byteLen < 8: + continue let connId = buf.readUint32(4) var conn = reactor.getConn(connId) if conn != nil: @@ -294,25 +345,32 @@ proc readParts(reactor: Reactor) = reactor.connections.delete(reactor.connections.find(conn)) continue - if magic == punchMagic: - # echo &"Received punch through from {address}" + if magic == PunchMagic: continue - if byteLen < headerSize: - # A valid packet will have at least the header. - # echo &"Received packet of invalid size {reactor.address}" - break + if byteLen < HeaderSize: + continue + + if magic != PartMagic and magic != AckMagic: + continue var part = Part() part.sequenceNum = buf.readUint32(4) part.connId = buf.readUint32(8) part.partNum = buf.readUint16(12) part.numParts = buf.readUint16(14) - part.data = buf.readStr(16, buf.len - 1) + part.data = buf.readStr(HeaderSize, byteLen - HeaderSize) + + if part.numParts == 0 or part.partNum >= part.numParts: + continue var conn = reactor.getConn(part.connId) if conn == nil: - if magic == partMagic and part.sequenceNum == 0 and part.partNum == 0: + if magic == PartMagic and + part.sequenceNum == 0 and + part.partNum == 0: + if reactor.connections.len >= reactor.maxConnections: + continue conn = newConnection(reactor, address) conn.id = part.connId reactor.connections.add(conn) @@ -326,35 +384,44 @@ proc readParts(reactor: Reactor) = conn.lastActiveTime = reactor.time - if magic == partMagic: - part.acked = true - part.ackedTime = reactor.time - reactor.sendSpecial(conn, part, ackMagic) - - if part.sequenceNum < conn.recvSequenceNum: + if magic == PartMagic: + if seqLess(part.sequenceNum, conn.recvSequenceNum): + # Already delivered; ACK so the sender stops retrying. + part.acked = true + part.ackedTime = reactor.time + reactor.sendSpecial(conn, part, AckMagic) continue var pos: int + var duplicate: bool for p in conn.recvParts: - if p.sequenceNum > part.sequenceNum: + if seqLess(part.sequenceNum, p.sequenceNum): break - if p.sequenceNum == part.sequenceNum: if p.partNum > part.partNum: break - if p.partNum == part.partNum: - # Duplicate - pos = -1 - assert p.data == part.data + duplicate = true break - inc pos - if pos != -1: # If not a duplicate - conn.recvParts.insert(part, pos) + if duplicate: + # Never assert on network bytes; ACK the part we already have. + part.acked = true + part.ackedTime = reactor.time + reactor.sendSpecial(conn, part, AckMagic) + continue + + if conn.recvParts.len >= reactor.maxRecvParts: + # Receive window full; skip ACK so the sender retries later. + continue + + part.acked = true + part.ackedTime = reactor.time + reactor.sendSpecial(conn, part, AckMagic) + conn.recvParts.insert(part, pos) - elif magic == ackMagic: + elif magic == AckMagic: for p in conn.sendParts: if p.sequenceNum == part.sequenceNum and p.numParts == part.numParts and @@ -363,10 +430,6 @@ proc readParts(reactor: Reactor) = p.acked = true p.ackedTime = reactor.time - else: - # Unrecognized packet - discard - func combineParts(reactor: Reactor) = for conn in reactor.connections.mitems: while true: @@ -381,7 +444,7 @@ func timeoutConnections(reactor: Reactor) = var i = 0 while i < reactor.connections.len: let conn = reactor.connections[i] - if conn.lastActiveTime + connTimeout <= reactor.time: + if conn.lastActiveTime + ConnTimeout <= reactor.time: reactor.deadConnections.add(conn) reactor.connections.delete(i) continue @@ -397,14 +460,21 @@ proc tick*(reactor: Reactor) = reactor.deadConnections.setLen(0) reactor.messages.setLen(0) + for conn in reactor.connections: + conn.stats.inFlight = 0 + conn.stats.saturated = false + reactor.sendNeededParts() reactor.readParts() reactor.combineParts() reactor.deleteAckedParts() + reactor.updateSendStats() reactor.timeoutConnections() func connect*(reactor: Reactor, address: Address): Connection = ## Starts a new connection to an address. + if reactor.connections.len >= reactor.maxConnections: + raise newException(NettyError, "max connections reached") result = newConnection(reactor, address) result.reactorId = reactor.id result.lastActiveTime = reactor.time @@ -431,13 +501,16 @@ proc sendMagic( packet.addUint32(connId) packet.addStr(extra) - reactor.socket.sendTo(address.host, address.port, packet) + try: + reactor.socket.sendTo(address.host, address.port, packet) + except OSError: + return proc disconnect*(reactor: Reactor, conn: Connection) = ## Disconnects the connection. assert reactor.id == conn.reactorId for i in 0 .. 10: - reactor.sendMagic(conn.address, disconnectMagic, conn.id) + reactor.sendMagic(conn.address, DisconnectMagic, conn.id) reactor.deadConnections.add(conn) let index = reactor.connections.find(conn) if index != -1: @@ -446,7 +519,7 @@ proc disconnect*(reactor: Reactor, conn: Connection) = proc punchThrough*(reactor: Reactor, address: Address) = ## Tries to punch through to host/port. for i in 0 .. 10: - reactor.sendMagic(address, punchMagic, 0, "punch through") + reactor.sendMagic(address, PunchMagic, 0, "punch through") proc punchThrough*(reactor: Reactor, host: string, port: int) = ## Tries to punch through to host/port. @@ -457,7 +530,9 @@ proc newReactor*(address: Address): Reactor = result = Reactor() result.r = initRand(getMonoTime().ticks) result.id = result.genId() - result.maxInFlight = defaultMaxInFlight + result.maxInFlight = DefaultMaxInFlight + result.maxConnections = DefaultMaxConnections + result.maxRecvParts = DefaultMaxRecvParts result.address = address result.socket = newSocket( @@ -472,7 +547,7 @@ proc newReactor*(address: Address): Reactor = let (_, portLocal) = result.socket.getLocalAddr() result.address.port = portLocal - result.debug.maxUdpPacket = defaultMaxUdpPacket + result.maxUdpPacket = DefaultMaxUdpPacket result.tick() diff --git a/tests/fuzz.nim b/tests/fuzz.nim new file mode 100644 index 0000000..70c419f --- /dev/null +++ b/tests/fuzz.nim @@ -0,0 +1,155 @@ +## Packet fuzzer for netty's UDP receive path. +## +## Sends adversarial and random datagrams at a reactor and requires that +## `tick` never raises. Catchable errors and silent drops are fine; process +## death or uncaught Defects are not. +## +## Run: nim r tests/fuzz.nim +## Replay: nim r tests/fuzz.nim --replay + +import + std/[os, random, strutils], + flatty/binny + +include netty + +var nextPortNumber = 5000 +proc nextPort(): int = + result = nextPortNumber + inc nextPortNumber + +proc toHex(s: string): string = + for c in s: + result.add toHex(c.ord, 2) + +proc fromHex(h: string): string = + var i = 0 + while i + 1 < h.len: + result.add chr(parseHexInt(h[i .. i + 1])) + i += 2 + +proc partPacket( + sequenceNum, connId: uint32, + partNum, numParts: uint16, + data: string +): string = + result.addUint32(PartMagic) + result.addUint32(sequenceNum) + result.addUint32(connId) + result.addUint16(partNum) + result.addUint16(numParts) + result.addStr(data) + +proc corners(): seq[string] = + ## Deterministic packets most likely to trip the parser. + result.add "" + for n in 1 .. 20: + result.add newString(n) + var d = "" + d.addUint32(DisconnectMagic) + result.add d + d.addUint32(0) + result.add d + result.add partPacket(0, 1, 0, 0, "x") + result.add partPacket(0, 1, 1, 1, "x") + result.add partPacket(0, 1, 0, 1, "") + result.add partPacket(0, 1, 0, 1, "ok") + result.add partPacket( + high(uint32), high(uint32), high(uint16), high(uint16), "z" + ) + var ack = "" + ack.addUint32(AckMagic) + ack.addUint32(0) + ack.addUint32(1) + ack.addUint16(0) + ack.addUint16(1) + result.add ack + var punch = "" + punch.addUint32(PunchMagic) + result.add punch + result.add partPacket(0, 9, 0, 100, "tiny") + +proc randPacket(r: var Rand): string = + case r.rand(0 .. 7) + of 0: + result = newString(r.rand(0 .. 64)) + for i in 0 ..< result.len: + result[i] = char(r.rand(255)) + of 1: + result = partPacket( + r.rand(uint32), + r.rand(uint32), + r.rand(uint16), + r.rand(uint16), + newString(r.rand(0 .. 200)) + ) + of 2: + result.addUint32(DisconnectMagic) + if r.rand(1.0) < 0.7: + result.addUint32(r.rand(uint32)) + of 3: + result.addUint32(AckMagic) + result.addUint32(r.rand(uint32)) + result.addUint32(r.rand(uint32)) + result.addUint16(r.rand(uint16)) + result.addUint16(r.rand(uint16)) + of 4: + result.addUint32(PunchMagic) + result.add newString(r.rand(0 .. 32)) + of 5: + result = partPacket(0, r.rand(uint32), 0, 1, "open") + of 6: + result = partPacket( + r.rand(0u32 .. 20u32), + r.rand(uint32), + 0, + r.rand(1u16 .. 8u16), + "frag" + ) + else: + result = newString(r.rand(0 .. 8)) + +proc feed(server, attacker: Reactor, packet: string) = + attacker.rawSend(server.address, packet) + attacker.tick() + server.tick() + +proc fuzzAll(packets: seq[string]) = + var server = newReactor("127.0.0.1", nextPort()) + server.maxConnections = 32 + server.maxRecvParts = 64 + var attacker = newReactor() + for packet in packets: + try: + feed(server, attacker, packet) + except Exception as e: + echo "CRASH on packet ", toHex(packet) + echo " ", e.name, ": ", e.msg + quit(1) + +proc main() = + if paramCount() >= 2 and paramStr(1) == "--replay": + fuzzAll(@[fromHex(paramStr(2))]) + echo "replay ok" + return + + echo "=== corner packets ===" + fuzzAll(corners()) + + let total = + if paramCount() >= 1: + parseInt(paramStr(1)) + else: + 2000 + + echo "=== random packets x ", total, " ===" + var + r = initRand(0x11E77) + packets = newSeq[string](total) + for i in 0 ..< total: + packets[i] = randPacket(r) + fuzzAll(packets) + + echo "=== fuzz summary: ok ===" + +main() diff --git a/tests/test.nim b/tests/test.nim index e1064a3..b6fb5d3 100644 --- a/tests/test.nim +++ b/tests/test.nim @@ -64,7 +64,7 @@ block: client.tick() doAssert len(server.messages) == 1, $server.messages.len doAssert len(server.connections) == 1, $server.connections.len - client.debug.tickTime = 1.0 + connTimeout + client.debug.tickTime = 1.0 + ConnTimeout client.tick() doAssert len(client.deadConnections) == 1 doAssert len(client.connections) == 0 @@ -75,7 +75,7 @@ block: var server = newReactor("127.0.0.1", nextPort()) var client = newReactor("127.0.0.1", nextPort()) - doAssert client.debug.maxUdpPacket == 492 + doAssert client.maxUdpPacket == 492 var buffer = "large:" for i in 0 ..< 1000: buffer.add "" @@ -175,7 +175,7 @@ block: # make sure all messages made it doAssert dataToSend.len == 0 - server.debug.tickTime = 1.0 + connTimeout + server.debug.tickTime = 1.0 + ConnTimeout server.tick() doAssert len(server.connections) == 0, $server.connections.len @@ -199,7 +199,7 @@ block: var server = newReactor("127.0.0.1", nextPort()) var client = newReactor("127.0.0.1", nextPort()) - client.debug.maxUdpPacket = 100 + client.maxUdpPacket = 100 client.maxInFlight = 10_000 var buffer = "large:" @@ -212,23 +212,26 @@ block: doAssert c2s.sendParts.len == 122 - client.tick() # can only send 100 parts due to maxInFlight and maxUdpPacket + client.tick() # can only send ~100 parts due to maxInFlight and maxUdpPacket doAssert c2s.stats.saturated == true + doAssert c2s.stats.inFlight <= client.maxInFlight, + &"stats.inFlight: {c2s.stats.inFlight}" server.tick() # receives 100 parts, sends acks back doAssert server.messages.len == 1, &"len: {server.messages.len}" - doAssert c2s.stats.inFlight < client.maxInFlight, - &"stats.inFlight: {c2s.stats.inFlight}" - doAssert c2s.stats.saturated == true - client.tick() # process the 100 acks, 22 parts left in flight + client.tick() # process the 100 acks; 22 parts still queued unsent doAssert c2s.sendParts.len == 22 - doAssert c2s.stats.inFlight == 2106, &"stats.inFlight: {c2s.stats.inFlight}" + doAssert c2s.stats.inFlight == 0, &"stats.inFlight: {c2s.stats.inFlight}" doAssert c2s.stats.saturated == false + client.tick() # send the remaining 22 parts + + doAssert c2s.stats.inFlight == 2106, &"stats.inFlight: {c2s.stats.inFlight}" + server.tick() # process the last 22 parts, send 22 acks doAssert server.messages.len == 1, &"len: {server.messages.len}" @@ -256,7 +259,7 @@ block: let firstSentTime = c2s.sendParts[0].sentTime - client.debug.tickTime = epochTime() + ackTime + client.debug.tickTime = epochTime() + AckTime client.tick() @@ -280,7 +283,7 @@ block: doAssert server.connections.len == 0 var msg = "" - msg.addUint32(partMagic) + msg.addUint32(PartMagic) msg.addStr("aasdfasdfaasdfaasdfasdfsdfsdasdfasdfsaasdfasdffsadfaasdfasdfa") client.rawSend(c2s.address, msg) diff --git a/tests/test_hardening.nim b/tests/test_hardening.nim new file mode 100644 index 0000000..edf507d --- /dev/null +++ b/tests/test_hardening.nim @@ -0,0 +1,316 @@ +## Hardening tests for untrusted UDP and accounting bugs. +## Each block forces one issue from the security / correctness list. + +import flatty/binny + +include netty + +var nextPortNumber = 4000 +proc nextPort(): int = + result = nextPortNumber + inc nextPortNumber + +proc partPacket( + sequenceNum, connId: uint32, + partNum, numParts: uint16, + data: string +): string = + result.addUint32(PartMagic) + result.addUint32(sequenceNum) + result.addUint32(connId) + result.addUint16(partNum) + result.addUint16(numParts) + result.addStr(data) + +block: + # Short datagram must not crash (header read before length check). + echo "Testing short packets do not crash tick" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + let address = server.address + + for n in 0 .. 15: + client.rawSend(address, newString(n)) + + # Disconnect magic with truncated body (4..7 bytes). + for n in 4 .. 7: + var junk = "" + junk.addUint32(DisconnectMagic) + junk.setLen(n) + client.rawSend(address, junk) + + client.tick() + server.tick() + doAssert server.connections.len == 0 + doAssert server.messages.len == 0 + +block: + # Undersized junk must continue, not break the read loop. + echo "Testing short packet does not starve later messages" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + var c2s = client.connect(server.address) + + client.rawSend(server.address, "short") + client.send(c2s, "hello") + client.tick() + server.tick() + + doAssert server.messages.len == 1, $server.messages.len + doAssert server.messages[0].data == "hello" + +block: + # Connection flood must respect maxConnections. + echo "Testing connection flood is capped" + var server = newReactor("127.0.0.1", nextPort()) + server.maxConnections = 8 + var client = newReactor() + + for i in 1u32 .. 40u32: + client.rawSend( + server.address, + partPacket(0, i, 0, 1, "x") + ) + client.tick() + server.tick() + + doAssert server.connections.len == 8, $server.connections.len + doAssert server.newConnections.len == 8, $server.newConnections.len + +block: + # Receive buffer must not grow without bound from incomplete messages. + echo "Testing recvParts receive window" + var server = newReactor("127.0.0.1", nextPort()) + server.maxRecvParts = 16 + var client = newReactor() + var c2s = client.connect(server.address) + + # Open the connection with a complete first message. + client.send(c2s, "open") + client.tick() + server.tick() + doAssert server.messages.len == 1 + doAssert server.connections.len == 1 + + let connId = server.connections[0].id + # Flood future sequence numbers, part 0 of unfinished multi-part messages. + for seqNum in 1u32 .. 200u32: + client.rawSend( + server.address, + partPacket(seqNum, connId, 0, 4, "frag") + ) + client.tick() + server.tick() + + doAssert server.connections[0].recvParts.len <= server.maxRecvParts, + $server.connections[0].recvParts.len + +block: + # Duplicate part with different bytes must not AssertionDefect. + echo "Testing spoofed duplicate does not assert" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + var c2s = client.connect(server.address) + + # Open the connection. + client.send(c2s, "open") + client.tick() + server.tick() + doAssert server.messages.len == 1 + let connId = server.connections[0].id + + # Buffer part 0 of an unfinished 2-part message. + client.rawSend( + server.address, + partPacket(1, connId, 0, 2, "real") + ) + client.tick() + server.tick() + doAssert server.connections[0].recvParts.len == 1 + + # Same seq/part, different payload — must not assert. + client.rawSend( + server.address, + partPacket(1, connId, 0, 2, "FAKE") + ) + client.tick() + server.tick() + doAssert server.connections[0].recvParts.len == 1 + doAssert server.connections[0].recvParts[0].data == "real" + doAssert server.messages.len == 0 + +block: + # Empty send must not assert; no-op is fine. + echo "Testing empty send is safe" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + var c2s = client.connect(server.address) + client.send(c2s, "") + client.tick() + server.tick() + doAssert server.messages.len == 0 + doAssert c2s.sendParts.len == 0 + +block: + # Invalid partNum / numParts must be ignored. + echo "Testing invalid part fields ignored" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + + client.rawSend( + server.address, + partPacket(0, 1, 0, 0, "bad") + ) + client.rawSend( + server.address, + partPacket(0, 2, 5, 3, "bad") + ) + client.tick() + server.tick() + doAssert server.connections.len == 0 + +block: + # Payload must use byteLen, not buf.len - 1 clamping. + echo "Testing payload length uses byteLen" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + let payload = "abcdef" + client.rawSend( + server.address, + partPacket(0, 42, 0, 1, payload) + ) + client.tick() + server.tick() + doAssert server.messages.len == 1 + doAssert server.messages[0].data == payload + +block: + # readLatency must hold delivery until latency elapses. + echo "Testing readLatency gates delivery" + var server = newReactor("127.0.0.1", nextPort()) + server.debug.tickTime = 100.0 + server.debug.readLatency = 1.0 + server.tick() + var client = newReactor() + var c2s = client.connect(server.address) + + client.send(c2s, "late") + client.tick() + server.tick() + doAssert server.messages.len == 0, "message arrived before latency" + + server.debug.tickTime = 101.1 + server.tick() + doAssert server.messages.len == 1 + doAssert server.messages[0].data == "late" + +block: + # maxInFlight must count bytes awaiting ACK across ticks. + echo "Testing maxInFlight across ticks without ack" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor("127.0.0.1", nextPort()) + client.maxUdpPacket = 100 + client.maxInFlight = 1000 + client.debug.tickTime = 10.0 + client.tick() + + var buffer = newString(5000) + for i in 0 ..< buffer.len: + buffer[i] = 'x' + + var c2s = client.connect(server.address) + client.send(c2s, buffer) + + client.tick() + var sentCount = 0 + for part in c2s.sendParts: + if part.sentTime != 0: + inc sentCount + doAssert c2s.stats.inFlight <= client.maxInFlight, + $c2s.stats.inFlight + doAssert c2s.stats.saturated == true + doAssert sentCount > 0 + + # Second tick before RTO and before any ACK must not send more. + client.tick() + var sentCount2 = 0 + for part in c2s.sendParts: + if part.sentTime != 0: + inc sentCount2 + doAssert sentCount2 == sentCount, + &"sent more without ack: {sentCount2} vs {sentCount}" + doAssert c2s.stats.inFlight <= client.maxInFlight, + $c2s.stats.inFlight + +block: + # inQueue must not go negative on retry. + echo "Testing inQueue on retry" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor("127.0.0.1", nextPort()) + client.debug.tickTime = 20.0 + client.tick() + var c2s = client.connect(server.address) + client.send(c2s, "retry-me") + client.tick() + doAssert c2s.stats.inQueue == 0, $c2s.stats.inQueue + + client.debug.tickTime = 20.0 + AckTime + client.tick() + doAssert c2s.stats.inQueue == 0, $c2s.stats.inQueue + doAssert c2s.sendParts.len == 1 + +block: + # Disconnect packet must not take down later traffic. + echo "Testing truncated disconnect then real message" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + var c2s = client.connect(server.address) + + var trunc = "" + trunc.addUint32(DisconnectMagic) + client.rawSend(server.address, trunc) + client.send(c2s, "after") + client.tick() + server.tick() + doAssert server.messages.len == 1 + doAssert server.messages[0].data == "after" + +block: + # Sequence wrap: old-looking high seq must not be treated as future forever. + echo "Testing sequence number wrap comparison" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + var c2s = client.connect(server.address) + + client.send(c2s, "first") + client.tick() + server.tick() + doAssert server.messages.len == 1 + + let conn = server.connections[0] + # Pretend we have already accepted messages up through high(uint32). + conn.recvSequenceNum = 0 + conn.recvParts.setLen(0) + + # Inject a part that is "behind" after wrap using serial comparison. + # After recvSequenceNum = 5, sequence 4 is old and must be ignored. + conn.recvSequenceNum = 5 + client.rawSend( + server.address, + partPacket(4, conn.id, 0, 1, "old") + ) + client.tick() + server.tick() + doAssert server.messages.len == 0 + doAssert conn.recvParts.len == 0 + + client.rawSend( + server.address, + partPacket(5, conn.id, 0, 1, "next") + ) + client.tick() + server.tick() + doAssert server.messages.len == 1 + doAssert server.messages[0].data == "next" + +echo "All hardening tests passed" From dcd98546154261fbcca47aee82d4a5af21d4b8c5 Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 08:05:18 -1000 Subject: [PATCH 2/8] run hardening tests in CI Co-authored-by: Cursor --- .github/workflows/build.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2501d83..94ad06e 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -23,3 +23,5 @@ jobs: cd .. nimby install -g "${{ github.event.repository.name }}/${{ github.event.repository.name }}.nimble" - run: nim r tests/tests.nim + - run: nim r tests/test_hardening.nim + - run: nim r tests/test_timeseries.nim From 22e3e07b1ce847cb9023ff3bf1d4c7e6bfb83b31 Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 08:08:28 -1000 Subject: [PATCH 3/8] Update readme. --- README.md | 15 ++++++++------- src/netty.nim | 2 +- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index ebcacfe..5875bff 100644 --- a/README.md +++ b/README.md @@ -10,15 +10,15 @@ ## About -Netty is a reliable connection over UDP aimed at games. Normally UDP packets can get duplicated, dropped, or come out of order. Netty makes sure packets are not duplicated, re-sends them if they get dropped, and all packets come in order. UDP packets might also get split if they are above 512 bytes and also can fail to be sent if they are bigger than 1-2k. Netty breaks up big packets and sends them in pieces making sure each piece comes reliably in order. Finally sometimes it's impossible for two clients to communicate direclty with TCP because of NATs, but Netty provides hole punching which allows them to connect. +Netty is a reliable connection over UDP aimed at games. Normally UDP packets can get duplicated, dropped, or come out of order. Netty makes sure packets are not duplicated, re-sends them if they get dropped, and all packets come in order. UDP packets might also get split if they are above 512 bytes and also can fail to be sent if they are bigger than 1-2k. Netty breaks up big packets and sends them in pieces making sure each piece comes reliably in order. Finally sometimes it is impossible for two clients to communicate directly with TCP because of NATs, but Netty can send hole punch probes to help them connect. ### Documentation API reference: https://treeform.github.io/netty -## Is Netty a implementation of TCP? +## Is Netty an implementation of TCP? -TCP is really bad for short latency sensitive messages. TCP was designed for throughput (downloading files) not latency (games). Netty will resend stuff faster than TCP, Netty will not buffer and you also get nat punch-through (which TCP does not have). Netty is basically "like TCP but for games". You should not be using Netty if you are will be sending large mount of data. By default Netty is capped at 250K of data in flight. +TCP is really bad for short latency sensitive messages. TCP was designed for throughput (downloading files) not latency (games). Netty will resend stuff faster than TCP, Netty will not buffer and you also get NAT punch-through probes (which TCP does not have). Netty is basically "like TCP but for games". You should not be using Netty if you will be sending large amounts of data. By default Netty is capped at 250KB of data in flight (`reactor.maxInFlight`, configurable). ## Features: @@ -30,10 +30,11 @@ TCP is really bad for short latency sensitive messages. TCP was designed for thr | packet ordering | yes | no | yes | | packet splitting | yes | no | yes | | packet retry | yes | no | yes | -| packet reduplication | yes | no | yes | +| packet deduplication | yes | no | yes | | hole punch through | no | yes | yes | | connection handling | yes | no | yes | -| congestion control | yes | no | yes | +| congestion control | yes | no | no | +| fixed in-flight byte cap | no | no | yes | # Echo Server/Client example @@ -45,7 +46,7 @@ import netty # listen for a connection on localhost port 1999 var server = newReactor("127.0.0.1", 1999) -echo "Listenting for UDP on 127.0.0.1:1999" +echo "Listening for UDP on 127.0.0.1:1999" # main loop while true: # must call tick to both read and write @@ -87,7 +88,7 @@ while true: import netty var server = newReactor("127.0.0.1", 2001) -echo "Listenting for UDP on 127.0.0.1:2001" +echo "Listening for UDP on 127.0.0.1:2001" while true: server.tick() for connection in server.newConnections: diff --git a/src/netty.nim b/src/netty.nim index 0f1c769..efb76c5 100644 --- a/src/netty.nim +++ b/src/netty.nim @@ -13,7 +13,7 @@ const AckTime = 0.250 ## Seconds to wait before sending the packet again. ConnTimeout = 10.00 ## Seconds to wait until timing-out the connection. DefaultMaxUdpPacket = 508 - HeaderSize - DefaultMaxInFlight = 25_000 + DefaultMaxInFlight = 250_000 DefaultMaxConnections = 1000 DefaultMaxRecvParts = 1000 From 682a497fce87ef46b20af1ce1840e4026b04936a Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 08:16:27 -1000 Subject: [PATCH 4/8] add reactor close for socket cleanup Co-authored-by: Cursor --- src/netty.nim | 21 +++++++++++++++++++++ tests/test_hardening.nim | 23 +++++++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/src/netty.nim b/src/netty.nim index efb76c5..aca3a6a 100644 --- a/src/netty.nim +++ b/src/netty.nim @@ -230,6 +230,8 @@ func divideAndSend(reactor: Reactor, conn: Connection, data: string) = proc rawSend(reactor: Reactor, address: Address, packet: string) = ## Low level send to a socket. + if reactor.socket == nil: + return if reactor.debug.dropRate != 0: if reactor.r.rand(1.0) <= reactor.debug.dropRate: return @@ -451,6 +453,9 @@ func timeoutConnections(reactor: Reactor) = inc i proc tick*(reactor: Reactor) = + if reactor.socket == nil: + return + if reactor.debug.tickTime != 0: reactor.time = reactor.debug.tickTime else: @@ -496,6 +501,8 @@ proc sendMagic( connId: uint32, extra = "" ) = + if reactor.socket == nil: + return var packet = newStringOfCap(4 + 4 + extra.len) packet.addUint32(magic) packet.addUint32(connId) @@ -516,6 +523,20 @@ proc disconnect*(reactor: Reactor, conn: Connection) = if index != -1: reactor.connections.delete(index) +proc close*(reactor: Reactor) = + ## Closes the UDP socket and clears local connection state. + if reactor.socket != nil: + let conns = reactor.connections + for conn in conns: + for i in 0 .. 10: + reactor.sendMagic(conn.address, DisconnectMagic, conn.id) + reactor.socket.close() + reactor.socket = nil + reactor.connections.setLen(0) + reactor.newConnections.setLen(0) + reactor.deadConnections.setLen(0) + reactor.messages.setLen(0) + proc punchThrough*(reactor: Reactor, address: Address) = ## Tries to punch through to host/port. for i in 0 .. 10: diff --git a/tests/test_hardening.nim b/tests/test_hardening.nim index edf507d..dee4570 100644 --- a/tests/test_hardening.nim +++ b/tests/test_hardening.nim @@ -313,4 +313,27 @@ block: doAssert server.messages.len == 1 doAssert server.messages[0].data == "next" +block: + # close() tears down the socket and is idempotent. + echo "Testing reactor close" + var server = newReactor("127.0.0.1", nextPort()) + var client = newReactor() + var c2s = client.connect(server.address) + client.send(c2s, "bye") + client.tick() + server.tick() + doAssert server.messages.len == 1 + + client.close() + doAssert client.socket == nil + doAssert client.connections.len == 0 + client.tick() + client.close() + + server.tick() + doAssert server.deadConnections.len == 1 + doAssert server.connections.len == 0 + server.close() + doAssert server.socket == nil + echo "All hardening tests passed" From 181f2bb661f148871d89aff89506919aa4fead9d Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 08:22:28 -1000 Subject: [PATCH 5/8] add reactor bench and baseline numbers Co-authored-by: Cursor --- docs/bench-baseline.md | 33 ++++ src/netty.nim | 2 +- tests/bench_reactor.nim | 324 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 358 insertions(+), 1 deletion(-) create mode 100644 docs/bench-baseline.md create mode 100644 tests/bench_reactor.nim diff --git a/docs/bench-baseline.md b/docs/bench-baseline.md new file mode 100644 index 0000000..4d2c2a7 --- /dev/null +++ b/docs/bench-baseline.md @@ -0,0 +1,33 @@ +# Reactor bench baseline + +Captured before connection-table / queue / pooling work. +Re-run the same command after perf changes and compare this table. + +``` +nim r -d:release -d:nettyBench tests/bench_reactor.nim 1000 +``` + +## Environment + +- Date: 2026-07-14 +- Host: macOS darwin 24.1.0 (arm64) +- Nim: 2.2.4 +- Flags: `-d:release -d:nettyBench` +- Scale: 1000 + +## Baseline + +| scenario | conns | ticks | msgs | bytes | msgs/s | mean tick us | p99 tick us | max recvParts | max sendParts | occupied MB | +|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:| +| many-idle | 1000 | 200 | 0 | 0 | 0 | 12.2 | 34 | 0 | 0 | 47.4 | +| many-active | 500 | 200 | 100000 | 400000 | 270931 | 1845.5 | 2079 | 0 | 0 | 23.8 | +| fan-in-large | 4 | 200 | 800 | 6400000 | 3490 | 1145.9 | 1363 | 0 | 0 | 0.3 | +| churn | 66 | 500 | 484 | 2420 | 74602 | 13.0 | 24 | 0 | 0 | 0.1 | + +## What should move + +| Change | Expect | +|--------|--------| +| `Table` for connections | lower many-idle / many-active mean and p99 tick | +| Ring/deque for parts | lower fan-in-large mean and p99 tick | +| Part pool / less copy | lower fan-in-large occupied MB and mean tick | diff --git a/src/netty.nim b/src/netty.nim index aca3a6a..0f9835f 100644 --- a/src/netty.nim +++ b/src/netty.nim @@ -326,7 +326,7 @@ proc readParts(reactor: Reactor) = buf, reactor.maxUdpPacket + HeaderSize, host, port ) except OSError: - when defined(nettyMagicSleep): + when defined(nettyMagicSleep) and not defined(nettyBench): sleep(1) break diff --git a/tests/bench_reactor.nim b/tests/bench_reactor.nim new file mode 100644 index 0000000..ac219d7 --- /dev/null +++ b/tests/bench_reactor.nim @@ -0,0 +1,324 @@ +## Server-side reactor benchmarks. +## +## One server, one client socket, many logical connections. Measures server +## tick cost so later perf work (tables, deques, pooling) has a baseline. +## +## Run (release, no macOS recv sleep): +## nim r -d:release -d:nettyBench tests/bench_reactor.nim +## Optional scale: +## nim r -d:release -d:nettyBench tests/bench_reactor.nim 2000 +## +## Checked-in baseline (compare after perf changes): +## docs/bench-baseline.md + +import std/[algorithm, monotimes, strformat, strutils, os] + +include netty + +type + BenchRow = object + name: string + connections: int + ticks: int + messages: int + bytes: int + serverNs: int64 + p99TickUs: int + maxRecvParts: int + maxSendParts: int + memBytes: int + +proc percentile99(samples: seq[int64]): int64 = + ## Nearest-rank p99 over ascending samples (nanoseconds). + if samples.len == 0: + return 0 + var sorted = samples + sorted.sort() + let i = min(sorted.len - 1, (sorted.len * 99) div 100) + sorted[i] + +proc drain(server, client: Reactor, rounds = 50) = + for _ in 0 ..< rounds: + client.tick() + server.tick() + +proc bumpTime(reactor: Reactor, dt = 0.001) = + reactor.debug.tickTime += dt + +proc maxParts(reactor: Reactor): (int, int) = + var recvMax, sendMax: int + for conn in reactor.connections: + recvMax = max(recvMax, conn.recvParts.len) + sendMax = max(sendMax, conn.sendParts.len) + (recvMax, sendMax) + +proc openPair(maxConns: int): (Reactor, Reactor) = + var server = newReactor("127.0.0.1", 0) + var client = newReactor() + server.maxConnections = maxConns + client.maxConnections = maxConns + server.debug.tickTime = 1_000.0 + client.debug.tickTime = 1_000.0 + server.tick() + client.tick() + (server, client) + +proc establish( + server, client: Reactor, + count: int, + payload = "x" +): seq[Connection] = + result = newSeqOfCap[Connection](count) + for i in 0 ..< count: + let conn = client.connect(server.address) + client.send(conn, payload) + result.add(conn) + # Keep time alive and flush in batches so the socket buffer stays sane. + if i mod 64 == 63: + client.bumpTime() + server.bumpTime() + drain(server, client, 4) + + var guard = 0 + while server.connections.len < count and guard < 10_000: + client.bumpTime() + server.bumpTime() + client.tick() + server.tick() + inc guard + + doAssert server.connections.len == count, + &"wanted {count} conns, got {server.connections.len}" + +proc measureServerTicks( + server, client: Reactor, + ticks: int, + beforeTick: proc() {.closure.} +): tuple[serverNs: int64, tickNs: seq[int64], messages: int, bytes: int] = + var + serverNs: int64 + tickNs = newSeqOfCap[int64](ticks) + messages, bytes: int + + for _ in 0 ..< ticks: + beforeTick() + client.tick() + let t0 = getMonoTime() + server.tick() + let dt = (getMonoTime() - t0).inNanoseconds + serverNs += dt + tickNs.add(dt) + for msg in server.messages: + inc messages + bytes += msg.data.len + + (serverNs, tickNs, messages, bytes) + +proc row( + name: string, + connections, ticks, messages, bytes: int, + serverNs: int64, + tickNs: seq[int64], + server: Reactor +): BenchRow = + let (recvMax, sendMax) = server.maxParts() + result = BenchRow( + name: name, + connections: connections, + ticks: ticks, + messages: messages, + bytes: bytes, + serverNs: serverNs, + p99TickUs: int(percentile99(tickNs) div 1000), + maxRecvParts: recvMax, + maxSendParts: sendMax, + memBytes: getOccupiedMem() + ) + +proc printRows(rows: seq[BenchRow]) = + echo "" + echo "| scenario | conns | ticks | msgs | bytes | msgs/s | mean tick us | p99 tick us | max recvParts | max sendParts | occupied MB |" + echo "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|" + for r in rows: + let + secs = r.serverNs.float64 / 1e9 + msgsPerSec = + if secs > 0: + r.messages.float64 / secs + else: + 0.0 + meanTickUs = + if r.ticks > 0: + r.serverNs.float64 / r.ticks.float64 / 1e3 + else: + 0.0 + memMb = r.memBytes.float64 / (1024.0 * 1024.0) + echo &"| {r.name} | {r.connections} | {r.ticks} | {r.messages} | {r.bytes} | {msgsPerSec.int} | {meanTickUs:.1f} | {r.p99TickUs} | {r.maxRecvParts} | {r.maxSendParts} | {memMb:.1f} |" + +proc benchIdle(connCount, measureTicks: int): BenchRow = + ## Many connections, almost no traffic after establish. + let (server, client) = openPair(connCount + 16) + discard establish(server, client, connCount) + # Drain establish traffic and ACK noise. + drain(server, client, 100) + + let measured = measureServerTicks(server, client, measureTicks) do (): + client.bumpTime() + server.bumpTime() + + result = row( + "many-idle", + connCount, + measureTicks, + measured.messages, + measured.bytes, + measured.serverNs, + measured.tickNs, + server + ) + client.close() + server.close() + +proc benchActive(connCount, measureTicks: int): BenchRow = + ## Every connection sends one small message per tick. + let (server, client) = openPair(connCount + 16) + let conns = establish(server, client, connCount) + drain(server, client, 100) + + let measured = measureServerTicks(server, client, measureTicks) do (): + for conn in conns: + client.send(conn, "ping") + client.bumpTime() + server.bumpTime() + + result = row( + "many-active", + connCount, + measureTicks, + measured.messages, + measured.bytes, + measured.serverNs, + measured.tickNs, + server + ) + client.close() + server.close() + +proc benchFanIn(connCount, measureTicks: int): BenchRow = + ## Few connections push large fragmented payloads each tick. + let (server, client) = openPair(connCount + 16) + client.maxUdpPacket = 200 + server.maxUdpPacket = 200 + let conns = establish(server, client, connCount, "open") + drain(server, client, 50) + + var big = newString(8_000) + for i in 0 ..< big.len: + big[i] = char(ord('A') + (i mod 26)) + + let measured = measureServerTicks(server, client, measureTicks) do (): + for conn in conns: + client.send(conn, big) + client.bumpTime(AckTime) + server.bumpTime(AckTime) + + result = row( + "fan-in-large", + connCount, + measureTicks, + measured.messages, + measured.bytes, + measured.serverNs, + measured.tickNs, + server + ) + client.close() + server.close() + +proc benchChurn(cycles: int): BenchRow = + ## Connect, send, disconnect repeatedly against one server. + let maxConns = max(64, cycles div 10 + 16) + let (server, client) = openPair(maxConns) + var + serverNs: int64 + tickNs: seq[int64] + messages, bytes, ticks: int + + for i in 0 ..< cycles: + let conn = client.connect(server.address) + client.send(conn, "churn") + client.bumpTime() + server.bumpTime() + client.tick() + + let t0 = getMonoTime() + server.tick() + let dt = (getMonoTime() - t0).inNanoseconds + serverNs += dt + tickNs.add(dt) + inc ticks + for msg in server.messages: + inc messages + bytes += msg.data.len + + if server.connections.len > 0: + # Prefer disconnecting the matching server-side conn if present. + var target = server.connections[0] + for c in server.connections: + if c.id == conn.id: + target = c + break + server.disconnect(target) + client.disconnect(conn) + + if i mod 32 == 31: + drain(server, client, 2) + + result = row( + "churn", + maxConns, + ticks, + messages, + bytes, + serverNs, + tickNs, + server + ) + client.close() + server.close() + +proc main() = + let scale = + if paramCount() >= 1: + parseInt(paramStr(1)) + else: + 1000 + + let + idleConns = scale + activeConns = max(1, scale div 2) + fanConns = 4 + measureTicks = 200 + churnCycles = max(100, scale div 2) + + echo "netty reactor bench" + echo &" scale={scale} idleConns={idleConns} activeConns={activeConns}" + echo &" fanConns={fanConns} measureTicks={measureTicks} churnCycles={churnCycles}" + echo " compile with: nim r -d:release -d:nettyBench tests/bench_reactor.nim" + + # Warm localhost UDP path. + block: + let (server, client) = openPair(8) + discard establish(server, client, 4) + drain(server, client, 20) + client.close() + server.close() + + var rows: seq[BenchRow] + rows.add benchIdle(idleConns, measureTicks) + rows.add benchActive(activeConns, measureTicks) + rows.add benchFanIn(fanConns, measureTicks) + rows.add benchChurn(churnCycles) + printRows(rows) + +main() From 9b1f97e956a346e7f62b4cd62a5d5dfb677ccd5b Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 09:53:03 -1000 Subject: [PATCH 6/8] speed up reactor paths, target 10k conns Co-authored-by: Cursor --- docs/bench-baseline.md | 51 +++-- src/netty.nim | 420 +++++++++++++++++++++++++++------------ tests/bench_reactor.nim | 64 ++++-- tests/test_hardening.nim | 14 +- tests/tests.nim | 10 +- 5 files changed, 383 insertions(+), 176 deletions(-) diff --git a/docs/bench-baseline.md b/docs/bench-baseline.md index 4d2c2a7..bcbacaf 100644 --- a/docs/bench-baseline.md +++ b/docs/bench-baseline.md @@ -1,33 +1,48 @@ # Reactor bench baseline -Captured before connection-table / queue / pooling work. -Re-run the same command after perf changes and compare this table. +Target: **10k connections** — Istrolid1 peaked around 3k concurrent players; +Istrolid2 should handle that easily, with 10k as headroom. ``` -nim r -d:release -d:nettyBench tests/bench_reactor.nim 1000 +nim r -d:release -d:nettyBench tests/bench_reactor.nim +# default scale is 10000 ``` ## Environment -- Date: 2026-07-14 - Host: macOS darwin 24.1.0 (arm64) - Nim: 2.2.4 - Flags: `-d:release -d:nettyBench` -- Scale: 1000 +- Default scale: 10000 -## Baseline +## Current baseline (post perf work, scale=10000) -| scenario | conns | ticks | msgs | bytes | msgs/s | mean tick us | p99 tick us | max recvParts | max sendParts | occupied MB | -|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:| -| many-idle | 1000 | 200 | 0 | 0 | 0 | 12.2 | 34 | 0 | 0 | 47.4 | -| many-active | 500 | 200 | 100000 | 400000 | 270931 | 1845.5 | 2079 | 0 | 0 | 23.8 | -| fan-in-large | 4 | 200 | 800 | 6400000 | 3490 | 1145.9 | 1363 | 0 | 0 | 0.3 | -| churn | 66 | 500 | 484 | 2420 | 74602 | 13.0 | 24 | 0 | 0 | 0.1 | +| scenario | conns | ticks | msgs | bytes | msgs/s | mean tick us | p99 tick us | occupied MB | +|---|---:|---:|---:|---:|---:|---:|---:|---:| +| many-idle | 10000 | 200 | 0 | 0 | 0 | 215.5 | 282 | 40.0 | +| many-active | 5000 | 200 | 819800 | 3279200 | 838653 | 4887.6 | 5631 | 39.4 | +| fan-in-large | 4 | 200 | 800 | 6400000 | 12748 | 313.8 | 462 | 0.1 | +| churn | 216 | 2000 | 1937 | 9685 | 91847 | 10.5 | 18 | 0.0 | -## What should move +## Reference at scale=1000 (smoke) -| Change | Expect | -|--------|--------| -| `Table` for connections | lower many-idle / many-active mean and p99 tick | -| Ring/deque for parts | lower fan-in-large mean and p99 tick | -| Part pool / less copy | lower fan-in-large occupied MB and mean tick | +| scenario | conns | mean tick us | msgs/s | +|---|---:|---:|---:| +| many-idle | 1000 | ~19 | 0 | +| many-active | 500 | ~439 | ~1.1M | + +## Reading the 10k numbers for Istrolid2 + +- **3k players** is below this bench's idle=10k / active=5k load. +- **many-active ~4.9ms mean tick** at 5k sending every tick is a stress ceiling, not a typical frame (games traffic is far sparser per conn). +- **many-idle ~0.22ms** at 10k is the "connected but quiet" cost — still scans every connection each tick; next win if needed. +- Prefer improving **10k** numbers over chasing 100k (establish is too slow/flaky on localhost UDP). + +## What already landed + +- `Table[uint32, Connection]` for O(1) `getConn` +- Per-sequence receive window (no O(n) inserts) +- `Deque` send queue + part pool +- ACK bundling (`AckBundleMagic`) +- Reused `outBuf`, `MaxUdpRecv`, `MaxPartsPerTick` +- `DefaultMaxConnections = 10_000` diff --git a/src/netty.nim b/src/netty.nim index 0f9835f..782924c 100644 --- a/src/netty.nim +++ b/src/netty.nim @@ -1,21 +1,26 @@ import flatty/binny, hashes, nativesockets, net, netty/timeseries, random, - sequtils, std/monotimes, strformat, times, os + std/[deques, monotimes, tables], strformat, times, os export Port, timeseries const PartMagic = 0xFFDDFF33.uint32 AckMagic = 0xFF33FF11.uint32 + AckBundleMagic = 0xFF33FF22.uint32 DisconnectMagic = 0xFF77FF99.uint32 PunchMagic = 0x00000000.uint32 HeaderSize = 4 + 4 + 4 + 2 + 2 + AckEntrySize = 4 + 4 + 2 + 2 AckTime = 0.250 ## Seconds to wait before sending the packet again. ConnTimeout = 10.00 ## Seconds to wait until timing-out the connection. DefaultMaxUdpPacket = 508 - HeaderSize DefaultMaxInFlight = 250_000 - DefaultMaxConnections = 1000 + DefaultMaxConnections = 10_000 DefaultMaxRecvParts = 1000 + MaxPartPool = 4096 + MaxUdpRecv = 2048 ## Socket read size; larger than part payload so ACK bundles fit. + MaxPartsPerTick = 100_000 ## Max datagrams drained per tick. type NettyError* = object of CatchableError @@ -31,6 +36,22 @@ type readLatency*: float32 ## Min simulated read latency in seconds. sendLatency*: float32 ## Min simulated send latency in seconds. + RecvSlot = object + data: string + ackedTime: float64 + present: bool + + RecvMessage = object + numParts: uint16 + got: uint16 + slots: seq[RecvSlot] + + AckEntry = object + sequenceNum: uint32 + connId: uint32 + partNum: uint16 + numParts: uint16 + Reactor* = ref object ## Main networking system that can open or receive connections. r: Rand @@ -43,6 +64,12 @@ type maxRecvParts*: int ## Max buffered receive parts per connection. maxUdpPacket*: int ## Max payload bytes per outgoing UDP packet. debug*: DebugConfig + connectionById: Table[uint32, Connection] + partPool: seq[Part] + outBuf: string + pendingAcks: seq[AckEntry] + pendingAckAddress: Address + pendingAckSet: bool connections*: seq[Connection] newConnections*: seq[Connection] ## New connections since last tick. @@ -63,8 +90,9 @@ type stats*: ConnectionStats lastActiveTime*: float64 - sendParts: seq[Part] ## Parts queued to be sent. - recvParts: seq[Part] ## Parts that have been read from the socket. + sendParts: Deque[Part] ## Parts queued to be sent. + recvPending: Table[uint32, RecvMessage] + recvPartCount: int ## Buffered receive parts (for the window cap). sendSequenceNum: uint32 ## Next message sequence num when sending. recvSequenceNum: uint32 ## Next message sequence number to receive. @@ -125,19 +153,61 @@ func currentTime(reactor: Reactor): float64 {.inline.} = else: reactor.time +func resetPart(part: Part) = + part.sequenceNum = 0 + part.connId = 0 + part.numParts = 0 + part.partNum = 0 + part.data = "" + part.queuedTime = 0 + part.sentTime = 0 + part.acked = false + part.ackedTime = 0 + +func acquirePart(reactor: Reactor): Part = + if reactor.partPool.len > 0: + result = reactor.partPool.pop() + resetPart(result) + else: + result = Part() + +func releasePart(reactor: Reactor, part: Part) = + if reactor.partPool.len < MaxPartPool: + resetPart(part) + reactor.partPool.add(part) + func newConnection(reactor: Reactor, address: Address): Connection = result = Connection() result.id = reactor.genId() result.reactorId = reactor.id result.address = address - - result.stats.latencyTs = newTimeSeries() - result.stats.throughputTs = newTimedSamples() + result.sendParts = initDeque[Part]() + when defined(nettyBench): + # Keep per-conn stats rings tiny so large-scale benches fit in RAM. + result.stats.latencyTs = newTimeSeries(16) + result.stats.throughputTs = newTimedSamples(16) + else: + result.stats.latencyTs = newTimeSeries() + result.stats.throughputTs = newTimedSamples() func getConn(reactor: Reactor, connId: uint32): Connection = - for conn in reactor.connections: - if conn.id == connId: - return conn + if connId in reactor.connectionById: + result = reactor.connectionById[connId] + +func addConnection(reactor: Reactor, conn: Connection) = + reactor.connections.add(conn) + reactor.connectionById[conn.id] = conn + +func removeConnection(reactor: Reactor, conn: Connection) = + reactor.connectionById.del(conn.id) + for i in 0 ..< reactor.connections.len: + if reactor.connections[i] == conn: + reactor.connections.del(i) + break + +func clearRecv(conn: Connection) = + conn.recvPending.clear() + conn.recvPartCount = 0 func updateSendStats(reactor: Reactor) = ## Recount in-flight bytes after sends or acks. @@ -157,43 +227,44 @@ func updateSendStats(reactor: Reactor) = conn.stats.saturated = saturated func read(reactor: Reactor, conn: Connection): (bool, Message) = - if conn.recvParts.len == 0: + if conn.recvSequenceNum notin conn.recvPending: return + let pending = conn.recvPending[conn.recvSequenceNum] let sequenceNum = conn.recvSequenceNum - numParts = conn.recvParts[0].numParts + numParts = pending.numParts - if numParts == 0 or conn.recvParts.len < numParts.int: + if numParts == 0 or pending.got < numParts: return var good = true for i in 0.uint16 ..< numParts: - if conn.recvParts[i].ackedTime + reactor.debug.readLatency > - reactor.time: + let slot = pending.slots[i] + if not slot.present: good = false break - - if not( - conn.recvParts[i].sequenceNum == sequenceNum and - conn.recvParts[i].numParts == numParts and - conn.recvParts[i].partNum == i - ): + if slot.ackedTime + reactor.debug.readLatency > reactor.time: good = false break if not good: return + var total: int + for i in 0.uint16 ..< numParts: + total += pending.slots[i].data.len + result[0] = true result[1].conn = conn result[1].sequenceNum = sequenceNum - + result[1].data = newStringOfCap(total) for i in 0.uint16 ..< numParts: - result[1].data.add(conn.recvParts[i].data) + result[1].data.add(pending.slots[i].data) + conn.recvPartCount -= numParts.int + conn.recvPending.del(sequenceNum) inc conn.recvSequenceNum - conn.recvParts.delete(0 .. numParts.int - 1) func divideAndSend(reactor: Reactor, conn: Connection, data: string) = ## Divides a packet into parts and gets it ready to be sent. @@ -207,7 +278,7 @@ func divideAndSend(reactor: Reactor, conn: Connection, data: string) = at: int while at < data.len: - var part = Part() + var part = reactor.acquirePart() part.sequenceNum = conn.sendSequenceNum part.connId = conn.id part.partNum = partNum @@ -219,30 +290,35 @@ func divideAndSend(reactor: Reactor, conn: Connection, data: string) = parts.add(part) if parts.len > high(uint16).int: + for part in parts: + reactor.releasePart(part) raise newException(NettyError, "message has too many parts") for part in parts.mitems: part.numParts = parts.len.uint16 part.queuedTime = reactor.currentTime() + conn.sendParts.addLast(part) - conn.sendParts.add(parts) inc conn.sendSequenceNum -proc rawSend(reactor: Reactor, address: Address, packet: string) = - ## Low level send to a socket. +proc rawSend(reactor: Reactor, address: Address, packet: string): bool {.discardable.} = + ## Low level send to a socket. False means try again next tick. if reactor.socket == nil: - return + return false if reactor.debug.dropRate != 0: if reactor.r.rand(1.0) <= reactor.debug.dropRate: - return + # Count as sent so simulated loss still uses RTO. + return true try: reactor.socket.sendTo(address.host, address.port, packet) + return true except OSError: - return + return false proc sendNeededParts(reactor: Reactor) = for conn in reactor.connections: - for part in conn.sendParts: + for i in 0 ..< conn.sendParts.len: + let part = conn.sendParts[i] if part.acked: continue @@ -260,70 +336,132 @@ proc sendNeededParts(reactor: Reactor) = continue let firstSend = part.sentTime == 0 - conn.stats.inFlight += part.data.len - if firstSend: - conn.stats.inQueue -= part.data.len - part.sentTime = reactor.time + reactor.outBuf.setLen(0) + reactor.outBuf.addUint32(PartMagic) + reactor.outBuf.addUint32(part.sequenceNum) + reactor.outBuf.addUint32(part.connId) + reactor.outBuf.addUint16(part.partNum) + reactor.outBuf.addUint16(part.numParts) + reactor.outBuf.addStr(part.data) - var packet = newStringOfCap(HeaderSize + part.data.len) - packet.addUint32(PartMagic) - packet.addUint32(part.sequenceNum) - packet.addUint32(part.connId) - packet.addUint16(part.partNum) - packet.addUint16(part.numParts) - packet.addStr(part.data) + if not reactor.rawSend(conn.address, reactor.outBuf): + # OS send buffer full; retry next tick without burning RTO. + conn.stats.saturated = true + break - reactor.rawSend(conn.address, packet) + if firstSend: + conn.stats.inQueue -= part.data.len + part.sentTime = reactor.time + conn.stats.inFlight += part.data.len reactor.updateSendStats() -proc sendSpecial( - reactor: Reactor, conn: Connection, part: Part, magic: uint32 -) = - assert reactor.id == conn.reactorId - assert conn.id == part.connId - - var packet = newStringOfCap(HeaderSize) - packet.addUint32(magic) - packet.addUint32(part.sequenceNum) - packet.addUint32(part.connId) - packet.addUint16(part.partNum) - packet.addUint16(part.numParts) +proc flushAcks(reactor: Reactor) = + ## Sends queued ACKs, bundling when several share a destination. + if not reactor.pendingAckSet or reactor.pendingAcks.len == 0: + reactor.pendingAcks.setLen(0) + reactor.pendingAckSet = false + return - reactor.rawSend(conn.address, packet) + let address = reactor.pendingAckAddress + if reactor.pendingAcks.len == 1: + let ack = reactor.pendingAcks[0] + reactor.outBuf.setLen(0) + reactor.outBuf.addUint32(AckMagic) + reactor.outBuf.addUint32(ack.sequenceNum) + reactor.outBuf.addUint32(ack.connId) + reactor.outBuf.addUint16(ack.partNum) + reactor.outBuf.addUint16(ack.numParts) + discard reactor.rawSend(address, reactor.outBuf) + else: + # Keep bundles within a typical UDP datagram, independent of + # maxUdpPacket (which only sizes message payloads). + let maxEntries = max(1, (MaxUdpRecv - 6) div AckEntrySize) + var at = 0 + while at < reactor.pendingAcks.len: + let n = min(maxEntries, reactor.pendingAcks.len - at) + reactor.outBuf.setLen(0) + reactor.outBuf.addUint32(AckBundleMagic) + reactor.outBuf.addUint16(n.uint16) + for i in 0 ..< n: + let ack = reactor.pendingAcks[at + i] + reactor.outBuf.addUint32(ack.sequenceNum) + reactor.outBuf.addUint32(ack.connId) + reactor.outBuf.addUint16(ack.partNum) + reactor.outBuf.addUint16(ack.numParts) + discard reactor.rawSend(address, reactor.outBuf) + at += n + + reactor.pendingAcks.setLen(0) + reactor.pendingAckSet = false + +proc queueAck( + reactor: Reactor, + address: Address, + sequenceNum, connId: uint32, + partNum, numParts: uint16 +) = + ## Queues an ACK, flushing when the destination changes. + if reactor.pendingAckSet and ( + reactor.pendingAckAddress.host != address.host or + reactor.pendingAckAddress.port != address.port + ): + reactor.flushAcks() + + if not reactor.pendingAckSet: + reactor.pendingAckAddress = address + reactor.pendingAckSet = true + + reactor.pendingAcks.add(AckEntry( + sequenceNum: sequenceNum, + connId: connId, + partNum: partNum, + numParts: numParts + )) + +func markAcked( + conn: Connection, + sequenceNum: uint32, + partNum, numParts: uint16 +) = + for i in 0 ..< conn.sendParts.len: + let p = conn.sendParts[i] + if p.sequenceNum == sequenceNum and + p.numParts == numParts and + p.partNum == partNum: + if not p.acked: + p.acked = true + return func deleteAckedParts(reactor: Reactor) = for conn in reactor.connections: - var pos, bytesAcked: int - for part in conn.sendParts: - if not part.acked: - break - inc pos + var bytesAcked: int + var minTime = float64.high + var popped: int + while conn.sendParts.len > 0 and conn.sendParts.peekFirst().acked: + let part = conn.sendParts.popFirst() bytesAcked += part.data.len + minTime = min(minTime, part.queuedTime) + inc popped + reactor.releasePart(part) - if pos > 0: - var minTime = float64.high - for i in 0 ..< pos: - let part = conn.sendParts[i] - minTime = min(minTime, part.queuedTime) - + if popped > 0: conn.stats.latencyTs.add((reactor.time - minTime).float32) - conn.sendParts.delete(0 .. pos - 1) conn.stats.throughputTs.add(reactor.time, bytesAcked.float64) proc readParts(reactor: Reactor) = var - buf = newStringOfCap(reactor.maxUdpPacket + HeaderSize) + buf = newStringOfCap(MaxUdpRecv) host: string port: Port - for _ in 0 ..< 1000: + for _ in 0 ..< MaxPartsPerTick: var byteLen: int try: byteLen = reactor.socket.recvFrom( - buf, reactor.maxUdpPacket + HeaderSize, host, port + buf, MaxUdpRecv, host, port ) except OSError: when defined(nettyMagicSleep) and not defined(nettyBench): @@ -344,38 +482,58 @@ proc readParts(reactor: Reactor) = var conn = reactor.getConn(connId) if conn != nil: reactor.deadConnections.add(conn) - reactor.connections.delete(reactor.connections.find(conn)) + reactor.removeConnection(conn) continue if magic == PunchMagic: continue + if magic == AckBundleMagic: + if byteLen < 6: + continue + let count = buf.readUint16(4).int + if count < 0 or byteLen < 6 + count * AckEntrySize: + continue + if reactor.debug.dropRate > 0.0: + if reactor.r.rand(1.0) <= reactor.debug.dropRate: + continue + var off = 6 + for _ in 0 ..< count: + let + sequenceNum = buf.readUint32(off) + connId = buf.readUint32(off + 4) + partNum = buf.readUint16(off + 8) + numParts = buf.readUint16(off + 10) + off += AckEntrySize + var conn = reactor.getConn(connId) + if conn != nil: + conn.lastActiveTime = reactor.time + conn.markAcked(sequenceNum, partNum, numParts) + continue + if byteLen < HeaderSize: continue if magic != PartMagic and magic != AckMagic: continue - var part = Part() - part.sequenceNum = buf.readUint32(4) - part.connId = buf.readUint32(8) - part.partNum = buf.readUint16(12) - part.numParts = buf.readUint16(14) - part.data = buf.readStr(HeaderSize, byteLen - HeaderSize) + let + sequenceNum = buf.readUint32(4) + connId = buf.readUint32(8) + partNum = buf.readUint16(12) + numParts = buf.readUint16(14) - if part.numParts == 0 or part.partNum >= part.numParts: + if numParts == 0 or partNum >= numParts: continue - var conn = reactor.getConn(part.connId) + var conn = reactor.getConn(connId) if conn == nil: - if magic == PartMagic and - part.sequenceNum == 0 and - part.partNum == 0: + if magic == PartMagic and sequenceNum == 0 and partNum == 0: if reactor.connections.len >= reactor.maxConnections: continue conn = newConnection(reactor, address) - conn.id = part.connId - reactor.connections.add(conn) + conn.id = connId + reactor.addConnection(conn) reactor.newConnections.add(conn) else: continue @@ -387,50 +545,57 @@ proc readParts(reactor: Reactor) = conn.lastActiveTime = reactor.time if magic == PartMagic: - if seqLess(part.sequenceNum, conn.recvSequenceNum): - # Already delivered; ACK so the sender stops retrying. - part.acked = true - part.ackedTime = reactor.time - reactor.sendSpecial(conn, part, AckMagic) + if seqLess(sequenceNum, conn.recvSequenceNum): + reactor.queueAck( + conn.address, sequenceNum, connId, partNum, numParts + ) continue - var pos: int - var duplicate: bool - for p in conn.recvParts: - if seqLess(part.sequenceNum, p.sequenceNum): - break - if p.sequenceNum == part.sequenceNum: - if p.partNum > part.partNum: - break - if p.partNum == part.partNum: - duplicate = true - break - inc pos - - if duplicate: - # Never assert on network bytes; ACK the part we already have. - part.acked = true - part.ackedTime = reactor.time - reactor.sendSpecial(conn, part, AckMagic) + if sequenceNum in conn.recvPending: + let pending = conn.recvPending[sequenceNum] + if partNum.int < pending.slots.len and + pending.slots[partNum].present: + reactor.queueAck( + conn.address, sequenceNum, connId, partNum, numParts + ) + continue + elif conn.recvPartCount >= reactor.maxRecvParts: + # Receive window full; skip ACK so the sender retries later. continue - if conn.recvParts.len >= reactor.maxRecvParts: - # Receive window full; skip ACK so the sender retries later. + if sequenceNum notin conn.recvPending: + var fresh = RecvMessage(numParts: numParts) + fresh.slots.setLen(numParts.int) + conn.recvPending[sequenceNum] = fresh + + var pending = conn.recvPending[sequenceNum] + if pending.numParts != numParts: + # Conflicting claim; ignore. + continue + if partNum.int >= pending.slots.len: + continue + if pending.slots[partNum].present: + reactor.queueAck( + conn.address, sequenceNum, connId, partNum, numParts + ) continue - part.acked = true - part.ackedTime = reactor.time - reactor.sendSpecial(conn, part, AckMagic) - conn.recvParts.insert(part, pos) + pending.slots[partNum].data = + buf.readStr(HeaderSize, byteLen - HeaderSize) + pending.slots[partNum].ackedTime = reactor.time + pending.slots[partNum].present = true + inc pending.got + conn.recvPending[sequenceNum] = pending + inc conn.recvPartCount + + reactor.queueAck( + conn.address, sequenceNum, connId, partNum, numParts + ) elif magic == AckMagic: - for p in conn.sendParts: - if p.sequenceNum == part.sequenceNum and - p.numParts == part.numParts and - p.partNum == part.partNum: - if not p.acked: - p.acked = true - p.ackedTime = reactor.time + conn.markAcked(sequenceNum, partNum, numParts) + + reactor.flushAcks() func combineParts(reactor: Reactor) = for conn in reactor.connections.mitems: @@ -448,7 +613,8 @@ func timeoutConnections(reactor: Reactor) = let conn = reactor.connections[i] if conn.lastActiveTime + ConnTimeout <= reactor.time: reactor.deadConnections.add(conn) - reactor.connections.delete(i) + reactor.connectionById.del(conn.id) + reactor.connections.del(i) continue inc i @@ -483,7 +649,7 @@ func connect*(reactor: Reactor, address: Address): Connection = result = newConnection(reactor, address) result.reactorId = reactor.id result.lastActiveTime = reactor.time - reactor.connections.add(result) + reactor.addConnection(result) reactor.newConnections.add(result) func connect*(reactor: Reactor, host: string, port: int): Connection = @@ -519,9 +685,7 @@ proc disconnect*(reactor: Reactor, conn: Connection) = for i in 0 .. 10: reactor.sendMagic(conn.address, DisconnectMagic, conn.id) reactor.deadConnections.add(conn) - let index = reactor.connections.find(conn) - if index != -1: - reactor.connections.delete(index) + reactor.removeConnection(conn) proc close*(reactor: Reactor) = ## Closes the UDP socket and clears local connection state. @@ -533,9 +697,11 @@ proc close*(reactor: Reactor) = reactor.socket.close() reactor.socket = nil reactor.connections.setLen(0) + reactor.connectionById.clear() reactor.newConnections.setLen(0) reactor.deadConnections.setLen(0) reactor.messages.setLen(0) + reactor.partPool.setLen(0) proc punchThrough*(reactor: Reactor, address: Address) = ## Tries to punch through to host/port. diff --git a/tests/bench_reactor.nim b/tests/bench_reactor.nim index ac219d7..10fa28c 100644 --- a/tests/bench_reactor.nim +++ b/tests/bench_reactor.nim @@ -1,12 +1,15 @@ ## Server-side reactor benchmarks. ## +## Target scale: 10k connections (Istrolid1 peaked ~3k concurrent players; +## Istrolid2 should clear that with headroom). +## ## One server, one client socket, many logical connections. Measures server -## tick cost so later perf work (tables, deques, pooling) has a baseline. +## tick cost so later perf work has a comparable baseline. ## ## Run (release, no macOS recv sleep): ## nim r -d:release -d:nettyBench tests/bench_reactor.nim ## Optional scale: -## nim r -d:release -d:nettyBench tests/bench_reactor.nim 2000 +## nim r -d:release -d:nettyBench tests/bench_reactor.nim 1000 ## ## Checked-in baseline (compare after perf changes): ## docs/bench-baseline.md @@ -48,7 +51,7 @@ proc bumpTime(reactor: Reactor, dt = 0.001) = proc maxParts(reactor: Reactor): (int, int) = var recvMax, sendMax: int for conn in reactor.connections: - recvMax = max(recvMax, conn.recvParts.len) + recvMax = max(recvMax, conn.recvPartCount) sendMax = max(sendMax, conn.sendParts.len) (recvMax, sendMax) @@ -63,32 +66,60 @@ proc openPair(maxConns: int): (Reactor, Reactor) = client.tick() (server, client) +proc refreshActivity(server, client: Reactor) = + let st = server.currentTime() + let ct = client.currentTime() + for conn in server.connections: + conn.lastActiveTime = st + for conn in client.connections: + conn.lastActiveTime = ct + proc establish( server, client: Reactor, count: int, payload = "x" ): seq[Connection] = result = newSeqOfCap[Connection](count) + client.maxInFlight = 1_000_000_000 + server.maxInFlight = 1_000_000_000 + let batch = 16 for i in 0 ..< count: let conn = client.connect(server.address) client.send(conn, payload) result.add(conn) - # Keep time alive and flush in batches so the socket buffer stays sane. - if i mod 64 == 63: + if i mod batch == batch - 1: client.bumpTime() server.bumpTime() - drain(server, client, 4) + drain(server, client, 16) + refreshActivity(server, client) + if count >= 1000 and (i < 64 or i mod 2000 == 1999): + echo " establish ", server.connections.len, "/", count, + " (client ", client.connections.len, ")" + + # Flush a trailing partial batch. + client.bumpTime() + server.bumpTime() + drain(server, client, 64) + refreshActivity(server, client) var guard = 0 - while server.connections.len < count and guard < 10_000: - client.bumpTime() - server.bumpTime() + let guardMax = max(10_000, count) + while server.connections.len < count and guard < guardMax: + if guard mod 32 == 31: + client.bumpTime(AckTime) + server.bumpTime(AckTime) + else: + client.bumpTime() + server.bumpTime() client.tick() server.tick() + refreshActivity(server, client) inc guard + if guard mod 5000 == 0: + echo " establish ", server.connections.len, "/", count doAssert server.connections.len == count, - &"wanted {count} conns, got {server.connections.len}" + &"wanted {count} conns, got {server.connections.len} after {guard} drains" proc measureServerTicks( server, client: Reactor, @@ -292,28 +323,21 @@ proc main() = if paramCount() >= 1: parseInt(paramStr(1)) else: - 1000 + 10_000 let idleConns = scale activeConns = max(1, scale div 2) fanConns = 4 measureTicks = 200 - churnCycles = max(100, scale div 2) + # Churn cost grows linearly; keep it bounded at large scale. + churnCycles = max(100, min(2000, scale div 2)) echo "netty reactor bench" echo &" scale={scale} idleConns={idleConns} activeConns={activeConns}" echo &" fanConns={fanConns} measureTicks={measureTicks} churnCycles={churnCycles}" echo " compile with: nim r -d:release -d:nettyBench tests/bench_reactor.nim" - # Warm localhost UDP path. - block: - let (server, client) = openPair(8) - discard establish(server, client, 4) - drain(server, client, 20) - client.close() - server.close() - var rows: seq[BenchRow] rows.add benchIdle(idleConns, measureTicks) rows.add benchActive(activeConns, measureTicks) diff --git a/tests/test_hardening.nim b/tests/test_hardening.nim index dee4570..4ed905c 100644 --- a/tests/test_hardening.nim +++ b/tests/test_hardening.nim @@ -102,8 +102,8 @@ block: client.tick() server.tick() - doAssert server.connections[0].recvParts.len <= server.maxRecvParts, - $server.connections[0].recvParts.len + doAssert server.connections[0].recvPartCount <= server.maxRecvParts, + $server.connections[0].recvPartCount block: # Duplicate part with different bytes must not AssertionDefect. @@ -126,7 +126,7 @@ block: ) client.tick() server.tick() - doAssert server.connections[0].recvParts.len == 1 + doAssert server.connections[0].recvPartCount == 1 # Same seq/part, different payload — must not assert. client.rawSend( @@ -135,8 +135,8 @@ block: ) client.tick() server.tick() - doAssert server.connections[0].recvParts.len == 1 - doAssert server.connections[0].recvParts[0].data == "real" + doAssert server.connections[0].recvPartCount == 1 + doAssert server.connections[0].recvPending[1].slots[0].data == "real" doAssert server.messages.len == 0 block: @@ -290,7 +290,7 @@ block: let conn = server.connections[0] # Pretend we have already accepted messages up through high(uint32). conn.recvSequenceNum = 0 - conn.recvParts.setLen(0) + conn.clearRecv() # Inject a part that is "behind" after wrap using serial comparison. # After recvSequenceNum = 5, sequence 4 is old and must be ignored. @@ -302,7 +302,7 @@ block: client.tick() server.tick() doAssert server.messages.len == 0 - doAssert conn.recvParts.len == 0 + doAssert conn.recvPartCount == 0 client.rawSend( server.address, diff --git a/tests/tests.nim b/tests/tests.nim index b6fb5d3..f07c7e5 100644 --- a/tests/tests.nim +++ b/tests/tests.nim @@ -39,20 +39,20 @@ block: # client should have part ACK:false doAssert client.connections[0].sendParts.len == 1 - doAssert client.connections[0].recvParts.len == 0 + doAssert client.connections[0].recvPartCount == 0 server.tick() # get message, ack message client.tick() # get ack # client should not have any parts now, acked parts deleted doAssert client.connections[0].sendParts.len == 0 - doAssert client.connections[0].recvParts.len == 0 + doAssert client.connections[0].recvPartCount == 0 # id should match doAssert server.connections[0].id == client.connections[0].id block: - # Text single client disconnect. + # Test single client disconnect. var server = newReactor("127.0.0.1", nextPort()) var client = newReactor() client.debug.tickTime = 1.0 @@ -61,9 +61,11 @@ block: client.send(c2s, "hi") client.tick() server.tick() - client.tick() doAssert len(server.messages) == 1, $server.messages.len doAssert len(server.connections) == 1, $server.connections.len + # Drain ACKs so a late packet does not refresh lastActiveTime. + client.tick() + client.tick() client.debug.tickTime = 1.0 + ConnTimeout client.tick() doAssert len(client.deadConnections) == 1 From dab8393ed9cd56d3d92692b3793cfe51d0450d36 Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 09:55:55 -1000 Subject: [PATCH 7/8] fix flaky maxInFlight test on macOS Co-authored-by: Cursor --- tests/tests.nim | 27 ++++++++++++--------------- 1 file changed, 12 insertions(+), 15 deletions(-) diff --git a/tests/tests.nim b/tests/tests.nim index f07c7e5..e3e8fd5 100644 --- a/tests/tests.nim +++ b/tests/tests.nim @@ -220,29 +220,26 @@ block: doAssert c2s.stats.inFlight <= client.maxInFlight, &"stats.inFlight: {c2s.stats.inFlight}" - server.tick() # receives 100 parts, sends acks back + server.tick() # receives first window, sends acks back doAssert server.messages.len == 1, &"len: {server.messages.len}" - client.tick() # process the 100 acks; 22 parts still queued unsent - - doAssert c2s.sendParts.len == 22 - doAssert c2s.stats.inFlight == 0, &"stats.inFlight: {c2s.stats.inFlight}" + client.tick() # process acks; remaining parts still queued unsent + doAssert c2s.sendParts.len > 0 + doAssert c2s.sendParts.len < 122 doAssert c2s.stats.saturated == false - client.tick() # send the remaining 22 parts - - doAssert c2s.stats.inFlight == 2106, &"stats.inFlight: {c2s.stats.inFlight}" - - server.tick() # process the last 22 parts, send 22 acks - - doAssert server.messages.len == 1, &"len: {server.messages.len}" - - client.tick() # receive the 22 acks + # Finish delivery; macOS localhost may need extra ticks for ACK bundles. + var guard = 0 + while c2s.sendParts.len > 0 and guard < 100: + client.tick() + server.tick() + inc guard - doAssert c2s.sendParts.len == 0 + doAssert c2s.sendParts.len == 0, &"sendParts left: {c2s.sendParts.len}" doAssert c2s.stats.inFlight == 0, &"stats.inFlight: {c2s.stats.inFlight}" doAssert c2s.stats.saturated == false + doAssert server.messages.len >= 1 doAssert c2s.stats.latencyTs.avg() > 0 doAssert c2s.stats.throughputTs.avg() > 0 From 0fbbd193b4dad649c043523844fc73171df5d73f Mon Sep 17 00:00:00 2001 From: treeform Date: Tue, 14 Jul 2026 09:56:21 -1000 Subject: [PATCH 8/8] count messages across maxInFlight drain Co-authored-by: Cursor --- tests/tests.nim | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/tests.nim b/tests/tests.nim index e3e8fd5..d436a0e 100644 --- a/tests/tests.nim +++ b/tests/tests.nim @@ -221,8 +221,8 @@ block: &"stats.inFlight: {c2s.stats.inFlight}" server.tick() # receives first window, sends acks back - - doAssert server.messages.len == 1, &"len: {server.messages.len}" + var got = server.messages.len + doAssert got >= 1, &"len: {got}" client.tick() # process acks; remaining parts still queued unsent doAssert c2s.sendParts.len > 0 @@ -234,12 +234,13 @@ block: while c2s.sendParts.len > 0 and guard < 100: client.tick() server.tick() + got += server.messages.len inc guard doAssert c2s.sendParts.len == 0, &"sendParts left: {c2s.sendParts.len}" doAssert c2s.stats.inFlight == 0, &"stats.inFlight: {c2s.stats.inFlight}" doAssert c2s.stats.saturated == false - doAssert server.messages.len >= 1 + doAssert got == 2, &"messages got: {got}" doAssert c2s.stats.latencyTs.avg() > 0 doAssert c2s.stats.throughputTs.avg() > 0