From 37838c9ca53628be518db447a0d52f4605d70ca3 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 09:45:13 -0400 Subject: [PATCH 01/15] fix(transport): preserve wildcard socket source addresses --- lsquic/context/io.nim | 117 ++++++++++++++++++++++++++++++++++++-- lsquic/endpoint.nim | 106 +++++++++++++++++++++++----------- tests/test_connection.nim | 23 ++++++++ tests/test_send.nim | 52 +++++++++++++++++ 4 files changed, 259 insertions(+), 39 deletions(-) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 3d4ab1d..a50ad45 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -30,10 +30,111 @@ when defined(linux): msg_hdr: Tmsghdr msg_len: cuint + type + InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = + object + ipi_ifindex: cint + ipi_spec_dst: InAddr + ipi_addr: InAddr + + In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = + object + ipi6_addr: In6Addr + ipi6_ifindex: cuint + + var + IP_PKTINFO {.importc, header: "".}: cint + IPV6_PKTINFO {.importc, header: "".}: cint + proc sendmmsg( sockfd: SocketHandle, msgvec: ptr MMsgHdr, vlen: cuint, flags: cint ): cint {.importc, header: "".} +when not defined(windows): + type ControlBuffer {.union.} = object + alignment: clong + data: array[128, byte] + + proc prepareSourceAddr( + localSa: ptr SockAddr, control: var ControlBuffer, msg: var Tmsghdr + ) = + when defined(linux): + let local = localSa.toTransportAddress() + if local.isAnyLocal(): + return + + zeroMem(addr control.data[0], control.data.len) + msg.msg_control = addr control.data[0] + + let cmsg = cast[ptr Tcmsghdr](msg.msg_control) + if local.family == AddressFamily.IPv4: + msg.msg_controllen = CMSG_SPACE(sizeof(InPktInfo).csize_t) + cmsg.cmsg_len = CMSG_LEN(sizeof(InPktInfo).csize_t) + cmsg.cmsg_level = IPPROTO_IP + cmsg.cmsg_type = IP_PKTINFO + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + copyMem( + addr info.ipi_spec_dst, + unsafeAddr local.address_v4[0], + local.address_v4.len, + ) + elif local.family == AddressFamily.IPv6: + msg.msg_controllen = CMSG_SPACE(sizeof(In6PktInfo).csize_t) + cmsg.cmsg_len = CMSG_LEN(sizeof(In6PktInfo).csize_t) + cmsg.cmsg_level = IPPROTO_IPV6 + cmsg.cmsg_type = IPV6_PKTINFO + let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) + copyMem( + addr info.ipi6_addr, unsafeAddr local.address_v6[0], local.address_v6.len + ) + +when defined(linux): + proc recvPacket*( + fd: SocketHandle, + buf: var seq[byte], + boundLocal: TransportAddress, + local, remote: var TransportAddress, + ): int {.raises: [].} = + var + remoteStorage: Sockaddr_storage + iov = IOVec(iov_base: addr buf[0], iov_len: buf.len.csize_t) + control: ControlBuffer + msg = Tmsghdr( + msg_name: addr remoteStorage, + msg_namelen: SockLen(sizeof(remoteStorage)), + msg_iov: addr iov, + msg_iovlen: 1, + msg_control: addr control.data[0], + msg_controllen: control.data.len.csize_t, + ) + + result = recvmsg(fd, addr msg, 0) + if result < 0: + return + + remote = toTransportAddress(cast[ptr SockAddr](addr remoteStorage)) + if boundLocal.family == AddressFamily.IPv6 and remote.isV4Mapped(): + remote = remote.toIPv4() + local = boundLocal + + var cmsg = CMSG_FIRSTHDR(addr msg) + while not cmsg.isNil: + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + local = TransportAddress( + family: AddressFamily.IPv4, port: boundLocal.port + ) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break + elif cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: + let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) + local = TransportAddress( + family: AddressFamily.IPv6, port: boundLocal.port + ) + copyMem(addr local.address_v6[0], addr info.ipi6_addr, local.address_v6.len) + break + cmsg = CMSG_NXTHDR(addr msg, cmsg) + when defined(windows): import std/winlean @@ -77,9 +178,10 @@ when not defined(windows): spec: struct_lsquic_out_spec, destStorage: var Sockaddr_storage, destAddrLen: SockLen, + control: var ControlBuffer, ): Tmsghdr = when defined(linux) and defined(x86_64) and not defined(android): - Tmsghdr( + result = Tmsghdr( msg_name: cast[pointer](addr destStorage), msg_namelen: destAddrLen, msg_iov: cast[ptr IOVec](spec.iov), @@ -89,7 +191,7 @@ when not defined(windows): msg_flags: 0, ) else: - Tmsghdr( + result = Tmsghdr( msg_name: cast[pointer](addr destStorage), msg_namelen: destAddrLen, msg_iov: cast[ptr IOVec](spec.iov), @@ -98,6 +200,7 @@ when not defined(windows): msg_controllen: 0, msg_flags: 0, ) + prepareSourceAddr(spec.local_sa, control, result) proc packetIn*( ctx: QuicContext, @@ -144,6 +247,7 @@ proc sendPacketsOut*( when defined(linux): var destStorages {.noinit.}: array[SendmmsgBatchSize, Sockaddr_storage] + controls {.noinit.}: array[SendmmsgBatchSize, ControlBuffer] msgs {.noinit.}: array[SendmmsgBatchSize, MMsgHdr] sent = 0 @@ -153,8 +257,10 @@ proc sendPacketsOut*( let curr = specsArr[sent + i] var destAddrLen: SockLen prepareDestAddr(curr.local_sa, curr.dest_sa, destStorages[i], destAddrLen) - msgs[i] = - MMsgHdr(msg_hdr: makeMsgHdr(curr, destStorages[i], destAddrLen), msg_len: 0) + msgs[i] = MMsgHdr( + msg_hdr: makeMsgHdr(curr, destStorages[i], destAddrLen, controls[i]), + msg_len: 0, + ) let res = sendmmsg(SocketHandle(quicCtx.fd), addr msgs[0], nmsgs.cuint, 0) if res < 0: @@ -223,7 +329,8 @@ proc sendPacketsOut*( return -1 break else: - let msg = makeMsgHdr(curr, destStorage, destAddrLen) + var control: ControlBuffer + let msg = makeMsgHdr(curr, destStorage, destAddrLen, control) let res = sendmsg(SocketHandle(quicCtx.fd), msg.addr, 0) if res < 0: diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index f6a4416..ba3e166 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -12,7 +12,9 @@ import socketconfig, engine_config, ] import ./context/[server, client, context, io] -import ./helpers/[logging, transportaddr] +import ./helpers/logging +when not defined(linux): + import ./helpers/transportaddr from chronos/osdefs import Sockaddr_storage, SockAddr, SockLen, SocketHandle when defined(windows): from std/winlean import recvfrom @@ -197,30 +199,40 @@ proc routeDatagram( proc recvDatagram( fd: SocketHandle, buf: var seq[byte], - remoteAddress: var Sockaddr_storage, - remoteAddrLen: var SockLen, + boundLocal: TransportAddress, + local, remote: var TransportAddress, ): int {.raises: [].} = - when defined(windows): - recvfrom( - fd, - cast[cstring](addr buf[0]), - cint(buf.len), - cint(0), - cast[ptr SockAddr](addr remoteAddress), - addr remoteAddrLen, - ).int + when defined(linux): + recvPacket(fd, buf, boundLocal, local, remote) else: - recvfrom( - fd, - addr buf[0], - buf.len, - cint(0), - cast[ptr SockAddr](addr remoteAddress), - addr remoteAddrLen, - ).int + var + remoteAddress: Sockaddr_storage + remoteAddrLen = SockLen(sizeof(Sockaddr_storage)) + result = + when defined(windows): + recvfrom( + fd, + cast[cstring](addr buf[0]), + cint(buf.len), + cint(0), + cast[ptr SockAddr](addr remoteAddress), + addr remoteAddrLen, + ).int + else: + recvfrom( + fd, + addr buf[0], + buf.len, + cint(0), + cast[ptr SockAddr](addr remoteAddress), + addr remoteAddrLen, + ).int + if result >= 0: + local = boundLocal + remote = toTransportAddress(cast[ptr SockAddr](addr remoteAddress)) proc drainDatagrams( - endpoint: QuicEndpoint, udp: DatagramTransport, local: TransportAddress + endpoint: QuicEndpoint, udp: DatagramTransport ): set[RouteTarget] {.raises: [].} = ## Chronos hands this callback a single datagram per event-loop wakeup, so ## whatever else has already arrived is read here rather than one wakeup, and @@ -228,13 +240,18 @@ proc drainDatagrams( if endpoint.drainBuf.len == 0: endpoint.drainBuf = newSeq[byte](DefaultDatagramBufferSize) - var targets: set[RouteTarget] + var + targets: set[RouteTarget] + boundLocal: TransportAddress + try: + boundLocal = udp.localAddress() + except TransportOsError: + return + for _ in 0 ..< MaxDatagramsPerWakeup: - var - remoteAddress: Sockaddr_storage - remoteAddrLen = SockLen(sizeof(Sockaddr_storage)) + var local, remote: TransportAddress let res = recvDatagram( - SocketHandle(udp.fd), endpoint.drainBuf, remoteAddress, remoteAddrLen + SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote ) if res < 0: # Empty, or an error the transport will report again on the next wakeup. @@ -242,9 +259,7 @@ proc drainDatagrams( if res > 0: targets.incl endpoint.routeDatagram( - endpoint.drainBuf.toOpenArray(0, res - 1), - local, - toTransportAddress(cast[ptr SockAddr](addr remoteAddress)), + endpoint.drainBuf.toOpenArray(0, res - 1), local, remote ) targets @@ -272,7 +287,7 @@ proc receiveFromUdp( var msg: seq[byte] msgLen: int - local = udp.localAddress() + local = udp.receivedLocalAddress() readIncoming(udp, msg, msgLen) if msgLen > 0: targets = endpoint.routeDatagram(msg.toOpenArray(0, msgLen - 1), local, remote) @@ -280,7 +295,7 @@ proc receiveFromUdp( warn "Failed to read UDP datagram", error = shortLog(e.msg) return - targets = targets + endpoint.drainDatagrams(udp, local) + targets = targets + endpoint.drainDatagrams(udp) if rtClient in targets: endpoint.clientContext.processWhenReady() @@ -298,9 +313,13 @@ proc createUdp( let udp = case address.family of AddressFamily.IPv4: - newDatagramTransport(onReceive, local = address) + newDatagramTransport( + onReceive, local = address, flags = {ServerFlags.PacketInfo} + ) of AddressFamily.IPv6: - newDatagramTransport6(onReceive, local = address) + newDatagramTransport6( + onReceive, local = address, flags = {ServerFlags.PacketInfo} + ) else: raise newException(QuicError, "only IPv4/IPv6 address is supported") @@ -380,6 +399,25 @@ proc ensureClientContext( endpoint.clientContext +proc dialLocalAddress( + endpoint: QuicEndpoint, remote: TransportAddress +): TransportAddress {.raises: [TransportOsError].} = + let bound = endpoint.udp.localAddress() + when not defined(linux): + return bound + + if not bound.isAnyLocal(): + return bound + + var source = getBestRoute(remote).source + if source.family == AddressFamily.None: + return bound + + if bound.family == AddressFamily.IPv6 and source.family == AddressFamily.IPv4: + source = source.toIPv6() + source.port = bound.port + source + proc accept*( endpoint: QuicEndpoint ): Future[Connection] {.async: (raises: [CancelledError, TransportError]).} = @@ -418,7 +456,7 @@ proc dial( .} = let ctx = endpoint.ensureClientContext() let connection = newOutgoingConnection( - ctx, endpoint.udp.localAddress(), address, serverName, certVerifier + ctx, endpoint.dialLocalAddress(address), address, serverName, certVerifier ) endpoint.connman.addConnection(connection) var connected = false diff --git a/tests/test_connection.nim b/tests/test_connection.nim index 037d895..3c1e91a 100644 --- a/tests/test_connection.nim +++ b/tests/test_connection.nim @@ -44,6 +44,9 @@ proc runConnectionTest( incomingConn.localAddress().port == boundAddress.port outgoingConn.localAddress().port == incomingConn.remoteAddress().port + if boundAddress.isAnyLocal(): + check not incomingConn.localAddress().isAnyLocal() + let outgoingBehaviour = proc() {.async.} = let stream = await outgoingConn.openStream() @@ -180,6 +183,23 @@ proc runEndpointSharedSocketDialTest(address: TransportAddress) {.async.} = incomingConn.close() await allFutures(outgoingConn.closedFuture(), incomingConn.closedFuture()) +proc runEndpointDualStackDialTest() {.async.} = + let dialer = makeEndpoint(WildcardIP6) + let server = makeEndpoint(AutoAddressIP4, {CanListen}) + let dialerPort = dialer.localAddress().port + defer: + await allFutures(dialer.stop(), server.stop()) + + let accepting = server.accept() + let outgoingConn = await dialer.dial(server.localAddress()).wait(dialTimeout) + let incomingConn = await accepting.wait(dialTimeout) + + check incomingConn.remoteAddress().port == dialerPort + + outgoingConn.close() + incomingConn.close() + await allFutures(outgoingConn.closedFuture(), incomingConn.closedFuture()) + proc runEndpointDialOnlyTest(address: TransportAddress) {.async.} = let server = makeServer() let listener = server.listen(address) @@ -392,6 +412,9 @@ suite "connection": asyncTest "endpoint dials from listener socket": await runEndpointSharedSocketDialTest(AutoAddressIP4) + asyncTest "dual-stack endpoint dials IPv4 from listener socket": + await runEndpointDualStackDialTest() + asyncTest "dial-only endpoint works without listener": await runEndpointDialOnlyTest(AutoAddressIP4) diff --git a/tests/test_send.nim b/tests/test_send.nim index 96b52b6..3ad785b 100644 --- a/tests/test_send.nim +++ b/tests/test_send.nim @@ -85,3 +85,55 @@ suite "packets out": spec = makeOutSpec(addr iov, addr localStorage, addr destStorage) check sendPacketsOut(cast[pointer](ctx), addr specs[0], SpecCount.cuint) == SpecCount + + test "packet info selects the requested IPv4 source address": + when defined(linux): + let + senderFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + receiverFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + defer: + nativesockets.close(senderFd) + nativesockets.close(receiverFd) + + var + senderBind = toSockaddrStorage(initTAddress("0.0.0.0:0")) + receiverBind = toSockaddrStorage(initTAddress("127.0.0.2:0")) + check bindSocket( + senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in).SockLen + ) == 0 + check bindSocket( + receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in).SockLen + ) == 0 + + var receiverLen = sizeof(receiverBind).SockLen + check getsockname( + receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen + ) == 0 + + let ctx = QuicContext(fd: cint(senderFd)) + var + payload = @[byte(1), 2, 3] + localStorage = toSockaddrStorage(initTAddress("127.0.0.2:0")) + iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) + spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) + + check sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + + var + received: array[3, byte] + remoteStorage: Sockaddr_storage + remoteLen = sizeof(remoteStorage).SockLen + check recvfrom( + receiverFd, + addr received[0], + received.len, + 0, + cast[ptr SockAddr](addr remoteStorage), + addr remoteLen, + ) == received.len + + var remote: TransportAddress + fromSAddr(addr remoteStorage, remoteLen, remote) + check remote.toIpAddress() == parseIpAddress("127.0.0.2") + else: + skip() From 05ff3f34a2245f25203cd8d6a37c4198fb0a5535 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 09:48:10 -0400 Subject: [PATCH 02/15] chore: format packet info changes --- lsquic/context/io.nim | 28 ++++++++++------------------ lsquic/endpoint.nim | 9 +++------ 2 files changed, 13 insertions(+), 24 deletions(-) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index a50ad45..64d3d17 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -31,16 +31,14 @@ when defined(linux): msg_len: cuint type - InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = - object - ipi_ifindex: cint - ipi_spec_dst: InAddr - ipi_addr: InAddr + InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object + ipi_ifindex: cint + ipi_spec_dst: InAddr + ipi_addr: InAddr - In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = - object - ipi6_addr: In6Addr - ipi6_ifindex: cuint + In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = object + ipi6_addr: In6Addr + ipi6_ifindex: cuint var IP_PKTINFO {.importc, header: "".}: cint @@ -74,9 +72,7 @@ when not defined(windows): cmsg.cmsg_type = IP_PKTINFO let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) copyMem( - addr info.ipi_spec_dst, - unsafeAddr local.address_v4[0], - local.address_v4.len, + addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len ) elif local.family == AddressFamily.IPv6: msg.msg_controllen = CMSG_SPACE(sizeof(In6PktInfo).csize_t) @@ -121,16 +117,12 @@ when defined(linux): while not cmsg.isNil: if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) - local = TransportAddress( - family: AddressFamily.IPv4, port: boundLocal.port - ) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) break elif cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) - local = TransportAddress( - family: AddressFamily.IPv6, port: boundLocal.port - ) + local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) copyMem(addr local.address_v6[0], addr info.ipi6_addr, local.address_v6.len) break cmsg = CMSG_NXTHDR(addr msg, cmsg) diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index ba3e166..3d4f1ba 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -250,9 +250,8 @@ proc drainDatagrams( for _ in 0 ..< MaxDatagramsPerWakeup: var local, remote: TransportAddress - let res = recvDatagram( - SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote - ) + let res = + recvDatagram(SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote) if res < 0: # Empty, or an error the transport will report again on the next wakeup. break @@ -313,9 +312,7 @@ proc createUdp( let udp = case address.family of AddressFamily.IPv4: - newDatagramTransport( - onReceive, local = address, flags = {ServerFlags.PacketInfo} - ) + newDatagramTransport(onReceive, local = address, flags = {ServerFlags.PacketInfo}) of AddressFamily.IPv6: newDatagramTransport6( onReceive, local = address, flags = {ServerFlags.PacketInfo} From 89de7c34b91fc616d48cf6ea49f07fbf7841a77f Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 09:50:11 -0400 Subject: [PATCH 03/15] fix(io): use portable control length types --- lsquic/context/io.nim | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 64d3d17..1f3f0ca 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -66,8 +66,9 @@ when not defined(windows): let cmsg = cast[ptr Tcmsghdr](msg.msg_control) if local.family == AddressFamily.IPv4: - msg.msg_controllen = CMSG_SPACE(sizeof(InPktInfo).csize_t) - cmsg.cmsg_len = CMSG_LEN(sizeof(InPktInfo).csize_t) + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InPktInfo).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InPktInfo).csize_t)) cmsg.cmsg_level = IPPROTO_IP cmsg.cmsg_type = IP_PKTINFO let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) @@ -75,8 +76,9 @@ when not defined(windows): addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len ) elif local.family == AddressFamily.IPv6: - msg.msg_controllen = CMSG_SPACE(sizeof(In6PktInfo).csize_t) - cmsg.cmsg_len = CMSG_LEN(sizeof(In6PktInfo).csize_t) + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(In6PktInfo).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(In6PktInfo).csize_t)) cmsg.cmsg_level = IPPROTO_IPV6 cmsg.cmsg_type = IPV6_PKTINFO let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) @@ -101,8 +103,8 @@ when defined(linux): msg_iov: addr iov, msg_iovlen: 1, msg_control: addr control.data[0], - msg_controllen: control.data.len.csize_t, ) + msg.msg_controllen = typeof(msg.msg_controllen)(control.data.len) result = recvmsg(fd, addr msg, 0) if result < 0: From 365a224c6b3cea7dc024af25c5117918d0b778f4 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 10:10:06 -0400 Subject: [PATCH 04/15] fix(endpoint): preserve dual-stack path family --- lsquic/endpoint.nim | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index 3d4f1ba..735b416 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -287,6 +287,10 @@ proc receiveFromUdp( msg: seq[byte] msgLen: int local = udp.receivedLocalAddress() + if udp.localAddress().family == AddressFamily.IPv6 and + local.family == AddressFamily.IPv4: + # LSQUIC identifies paths using the socket address family. + local = local.toIPv6() readIncoming(udp, msg, msgLen) if msgLen > 0: targets = endpoint.routeDatagram(msg.toOpenArray(0, msgLen - 1), local, remote) From e6db43a95d1c7680ab6a433e15281eb0c6bf66ab Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 11:02:42 -0400 Subject: [PATCH 05/15] build: pin packet-info Chronos revision --- lsquic.nimble | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lsquic.nimble b/lsquic.nimble index fef1735..ca79857 100644 --- a/lsquic.nimble +++ b/lsquic.nimble @@ -9,7 +9,7 @@ installFiles = @["lsquic.nim"] requires "nim >= 2.0.0" requires "zlib" requires "stew >= 0.4.0" -requires "chronos >= 4.0.4" +requires "https://github.com/status-im/nim-chronos#96eeb8a0d76d609d2ecde9d45fe1a16bbfa0b3ea" requires "unittest2" requires "chronicles >= 0.11.0" requires "https://github.com/vacp2p/nim-boringssl >= 0.0.11" From 6b7071d64d37344f99bb97038ab71ba07fd28083 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 11:43:55 -0400 Subject: [PATCH 06/15] support wildcard UDP paths across platforms --- lsquic.nimble | 2 +- lsquic/context/io.nim | 135 +++++++++++++++++++++++++++-------------- lsquic/endpoint.nim | 115 ++++++++++++----------------------- tests/test_all.nim | 2 +- tests/test_receive.nim | 58 ++++++++++++++++++ 5 files changed, 186 insertions(+), 126 deletions(-) create mode 100644 tests/test_receive.nim diff --git a/lsquic.nimble b/lsquic.nimble index ca79857..6983d11 100644 --- a/lsquic.nimble +++ b/lsquic.nimble @@ -9,7 +9,7 @@ installFiles = @["lsquic.nim"] requires "nim >= 2.0.0" requires "zlib" requires "stew >= 0.4.0" -requires "https://github.com/status-im/nim-chronos#96eeb8a0d76d609d2ecde9d45fe1a16bbfa0b3ea" +requires "https://github.com/status-im/nim-chronos#a1f47ec2ae6ad366ebc15a17693f595bcb385502" requires "unittest2" requires "chronicles >= 0.11.0" requires "https://github.com/vacp2p/nim-boringssl >= 0.0.11" diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 1f3f0ca..ca2dfca 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -16,10 +16,11 @@ when not defined(windows): logScope: topics = "lsquic" -const MaxBatch = 1024 - ## Upper bound on the stack WSABUF array in the Windows send path (`sendPacketsOut`). - ## In practice `iovlen` is small; if it exceeds this bound, `sendPacketsOut` - ## falls back to a heap-allocated seq. +when defined(windows): + const MaxBatch = 1024 + ## Upper bound on the stack WSABUF array in the Windows send path (`sendPacketsOut`). + ## In practice `iovlen` is small; if it exceeds this bound, `sendPacketsOut` + ## falls back to a heap-allocated seq. when defined(linux): {.passc: "-D_GNU_SOURCE".} @@ -30,6 +31,14 @@ when defined(linux): msg_hdr: Tmsghdr msg_len: cuint + proc sendmmsg( + sockfd: SocketHandle, msgvec: ptr MMsgHdr, vlen: cuint, flags: cint + ): cint {.importc, header: "".} + +when defined(linux) or defined(macosx): + when defined(macosx): + {.passc: "-D__APPLE_USE_RFC_3542".} + type InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object ipi_ifindex: cint @@ -44,10 +53,6 @@ when defined(linux): IP_PKTINFO {.importc, header: "".}: cint IPV6_PKTINFO {.importc, header: "".}: cint - proc sendmmsg( - sockfd: SocketHandle, msgvec: ptr MMsgHdr, vlen: cuint, flags: cint - ): cint {.importc, header: "".} - when not defined(windows): type ControlBuffer {.union.} = object alignment: clong @@ -56,8 +61,10 @@ when not defined(windows): proc prepareSourceAddr( localSa: ptr SockAddr, control: var ControlBuffer, msg: var Tmsghdr ) = - when defined(linux): - let local = localSa.toTransportAddress() + when defined(linux) or defined(macosx): + var local = localSa.toTransportAddress() + if local.isV4Mapped(): + local = local.toIPv4() if local.isAnyLocal(): return @@ -129,26 +136,43 @@ when defined(linux): break cmsg = CMSG_NXTHDR(addr msg, cmsg) + if boundLocal.family == AddressFamily.IPv6 and local.family == AddressFamily.IPv4: + local = local.toIPv6() + when defined(windows): - import std/winlean - - {.pragma: wsa, stdcall, dynlib: "ws2_32.dll".} - - type WSABUF* = object - len*: culong - buf*: ptr char - - proc WSASendTo*( - s: SocketHandle, - lpBuffers: ptr WSABUF, - dwBufferCount: culong, - lpNumberOfBytesSent: ptr culong, - dwFlags: culong, - lpTo: ptr SockAddr, - iToLen: cint, - lpOverlapped: pointer, - lpCompletionRoutine: pointer, - ): cint {.wsa, importc: "WSASendTo".} + func wsaCmsgAlign(value: uint): uint = + (value + uint(sizeof(uint) - 1)) and not uint(sizeof(uint) - 1) + + proc prepareSourceAddr( + localSa: ptr SockAddr, control: var array[128, byte], msg: var osdefs.WSAMSG + ) = + var local = localSa.toTransportAddress() + if local.isV4Mapped(): + local = local.toIPv4() + if local.isAnyLocal(): + return + + zeroMem(addr control[0], control.len) + let + headerLen = wsaCmsgAlign(uint(sizeof(osdefs.WSACMSGHDR))) + header = cast[ptr osdefs.WSACMSGHDR](addr control[0]) + data = cast[pointer](cast[uint](header) + headerLen) + + msg.control.buf = cast[cstring](addr control[0]) + if local.family == AddressFamily.IPv4: + header.cmsg_len = headerLen + uint(sizeof(osdefs.WinInPktInfo)) + header.cmsg_level = osdefs.IPPROTO_IP + header.cmsg_type = osdefs.IP_PKTINFO + msg.control.len = ULONG(wsaCmsgAlign(header.cmsg_len)) + let info = cast[ptr osdefs.WinInPktInfo](data) + copyMem(addr info.ipi_addr, unsafeAddr local.address_v4[0], local.address_v4.len) + elif local.family == AddressFamily.IPv6: + header.cmsg_len = headerLen + uint(sizeof(osdefs.WinIn6PktInfo)) + header.cmsg_level = osdefs.IPPROTO_IPV6 + header.cmsg_type = osdefs.IPV6_PKTINFO + msg.control.len = ULONG(wsaCmsgAlign(header.cmsg_len)) + let info = cast[ptr osdefs.WinIn6PktInfo](data) + copyMem(addr info.ipi6_addr, unsafeAddr local.address_v6[0], local.address_v6.len) proc prepareDestAddr( localSa: ptr SockAddr, @@ -277,8 +301,24 @@ proc sendPacketsOut*( else: when defined(windows): var - bufs {.noinit.}: array[MaxBatch, WSABUF] - overflow: seq[WSABUF] + bufs {.noinit.}: array[MaxBatch, osdefs.WSABUF] + overflow: seq[osdefs.WSABUF] + extension: pointer + extensionBytesRet: DWORD + sendMsgGuid = osdefs.WSAID_WSASENDMSG + if wsaIoctl( + SocketHandle(quicCtx.fd), + osdefs.SIO_GET_EXTENSION_FUNCTION_POINTER, + addr sendMsgGuid, + DWORD(sizeof(sendMsgGuid)), + addr extension, + DWORD(sizeof(extension)), + addr extensionBytesRet, + nil, + nil, + ) != 0: + return -1 + let wsaSendMsg = cast[osdefs.LPFN_WSASENDMSG](extension) var sent = 0 for i in 0 ..< nspecs.int: let curr = specsArr[i] @@ -293,27 +333,28 @@ proc sendPacketsOut*( iovlen = curr.iovlen.int dst = if iovlen <= bufs.len: - cast[ptr UncheckedArray[WSABUF]](addr bufs[0]) + cast[ptr UncheckedArray[osdefs.WSABUF]](addr bufs[0]) else: overflow.setLen(iovlen) - cast[ptr UncheckedArray[WSABUF]](addr overflow[0]) + cast[ptr UncheckedArray[osdefs.WSABUF]](addr overflow[0]) for j in 0 ..< iovlen: let src = iovArr[j] - dst[j].len = culong(src.iov_len) - dst[j].buf = cast[ptr char](src.iov_base) - - var bytesSent: culong = 0 - let res = WSASendTo( - SocketHandle(quicCtx.fd), - addr dst[0], - culong(iovlen), - addr bytesSent, - 0, # flags - cast[ptr SockAddr](addr destStorage), - cint(destAddrLen), - nil, - nil, # no overlapped + dst[j].len = ULONG(src.iov_len) + dst[j].buf = cast[cstring](src.iov_base) + + var + control: array[128, byte] + bytesSent: DWORD + msg = osdefs.WSAMSG( + name: cast[ptr SockAddr](addr destStorage), + namelen: cint(destAddrLen), + lpBuffers: addr dst[0], + dwBufferCount: DWORD(iovlen), + ) + prepareSourceAddr(curr.local_sa, control, msg) + let res = wsaSendMsg( + SocketHandle(quicCtx.fd), addr msg, DWORD(0), addr bytesSent, nil, nil ) if res != 0: let errorCode = osdefs.wsaGetLastError() diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index 735b416..f556ed8 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -13,13 +13,8 @@ import ] import ./context/[server, client, context, io] import ./helpers/logging -when not defined(linux): - import ./helpers/transportaddr -from chronos/osdefs import Sockaddr_storage, SockAddr, SockLen, SocketHandle -when defined(windows): - from std/winlean import recvfrom -else: - from chronos/osdefs import recvfrom +when defined(linux): + from chronos/osdefs import SocketHandle logScope: topics = "lsquic" @@ -46,10 +41,9 @@ type drainBuf: seq[byte] engineConfig: QuicEngineConfig -const - CloseWait: Duration = 300.milliseconds - - MaxDatagramsPerWakeup = 64 +const CloseWait: Duration = 300.milliseconds +when defined(linux): + const MaxDatagramsPerWakeup = 64 ## Capped so that a busy socket cannot starve the rest of the event loop. proc socketReceiveBufferBytes( @@ -196,72 +190,38 @@ proc routeDatagram( trace "Dropping packet with unknown connection ID", bytes = data.len, local, remote {} -proc recvDatagram( - fd: SocketHandle, - buf: var seq[byte], - boundLocal: TransportAddress, - local, remote: var TransportAddress, -): int {.raises: [].} = - when defined(linux): - recvPacket(fd, buf, boundLocal, local, remote) - else: - var - remoteAddress: Sockaddr_storage - remoteAddrLen = SockLen(sizeof(Sockaddr_storage)) - result = - when defined(windows): - recvfrom( - fd, - cast[cstring](addr buf[0]), - cint(buf.len), - cint(0), - cast[ptr SockAddr](addr remoteAddress), - addr remoteAddrLen, - ).int - else: - recvfrom( - fd, - addr buf[0], - buf.len, - cint(0), - cast[ptr SockAddr](addr remoteAddress), - addr remoteAddrLen, - ).int - if result >= 0: - local = boundLocal - remote = toTransportAddress(cast[ptr SockAddr](addr remoteAddress)) - -proc drainDatagrams( - endpoint: QuicEndpoint, udp: DatagramTransport -): set[RouteTarget] {.raises: [].} = - ## Chronos hands this callback a single datagram per event-loop wakeup, so - ## whatever else has already arrived is read here rather than one wakeup, and - ## one engine tick, at a time. - if endpoint.drainBuf.len == 0: - endpoint.drainBuf = newSeq[byte](DefaultDatagramBufferSize) +when defined(linux): + proc drainDatagrams( + endpoint: QuicEndpoint, udp: DatagramTransport + ): set[RouteTarget] {.raises: [].} = + ## Chronos hands this callback a single datagram per event-loop wakeup, so + ## whatever else has already arrived is read here rather than one wakeup, and + ## one engine tick, at a time. + if endpoint.drainBuf.len == 0: + endpoint.drainBuf = newSeq[byte](DefaultDatagramBufferSize) - var - targets: set[RouteTarget] - boundLocal: TransportAddress - try: - boundLocal = udp.localAddress() - except TransportOsError: - return - - for _ in 0 ..< MaxDatagramsPerWakeup: - var local, remote: TransportAddress - let res = - recvDatagram(SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote) - if res < 0: - # Empty, or an error the transport will report again on the next wakeup. - break - - if res > 0: - targets.incl endpoint.routeDatagram( - endpoint.drainBuf.toOpenArray(0, res - 1), local, remote - ) - - targets + var + targets: set[RouteTarget] + boundLocal: TransportAddress + try: + boundLocal = udp.localAddress() + except TransportOsError: + return + + for _ in 0 ..< MaxDatagramsPerWakeup: + var local, remote: TransportAddress + let res = + recvPacket(SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote) + if res < 0: + # Empty, or an error the transport will report again on the next wakeup. + break + + if res > 0: + targets.incl endpoint.routeDatagram( + endpoint.drainBuf.toOpenArray(0, res - 1), local, remote + ) + + targets proc readIncoming( udp: DatagramTransport, msg: var seq[byte], msgLen: var int @@ -298,7 +258,8 @@ proc receiveFromUdp( warn "Failed to read UDP datagram", error = shortLog(e.msg) return - targets = targets + endpoint.drainDatagrams(udp) + when defined(linux): + targets = targets + endpoint.drainDatagrams(udp) if rtClient in targets: endpoint.clientContext.processWhenReady() diff --git a/tests/test_all.nim b/tests/test_all.nim index d86d5cc..6631ca4 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -6,7 +6,7 @@ import test_connection, test_perf, test_tlsconfig, test_verifier, test_lifecycle, test_timeout, test_stress, test_runtime, test_routing, test_destaddr, test_send, - test_engine_config + test_engine_config, test_receive from ./helpers/trackers import finalCheckTrackers finalCheckTrackers() diff --git a/tests/test_receive.nim b/tests/test_receive.nim new file mode 100644 index 0000000..05c5817 --- /dev/null +++ b/tests/test_receive.nim @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: Apache-2.0 OR MIT +# Copyright (c) Status Research & Development GmbH + +{.used.} + +import chronos, chronos/unittest2/asynctests +when defined(linux): + import chronos/osdefs + import ../lsquic/context/io +import ./helpers/trackers + +suite "UDP receive": + teardown: + checkTrackers() + + asyncTest "drained IPv4 burst keeps the dual-stack socket family": + when defined(linux): + proc ignore( + udp: DatagramTransport, remote: TransportAddress + ): Future[void] {.async: (raises: []).} = + discard + + let + server = newDatagramTransport6( + ignore, + local = initTAddress("[::]:0"), + flags = {ServerFlags.NoAutoRead, ServerFlags.PacketInfo}, + ) + client = newDatagramTransport(ignore) + defer: + await allFutures(server.closeWait(), client.closeWait()) + + var destination = initTAddress("127.0.0.1:0") + destination.port = server.localAddress().port + await client.sendTo(destination, @[1.byte]) + await client.sendTo(destination, @[2.byte]) + + var buf = newSeq[byte](DefaultDatagramBufferSize) + for expected in [1.byte, 2.byte]: + var + local, remote: TransportAddress + received = -1 + + for _ in 0 ..< 100: + received = recvPacket( + SocketHandle(server.fd), buf, server.localAddress(), local, remote + ) + if received >= 0: + break + await sleepAsync(10.milliseconds) + + check: + received == 1 + buf[0] == expected + local == destination.toIPv6() + remote.family == AddressFamily.IPv4 + else: + skip() From 411825f8b9f7b8933ce4f26510c12183416b3af7 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 11:46:38 -0400 Subject: [PATCH 07/15] build: update packet info Chronos revision --- lsquic.nimble | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lsquic.nimble b/lsquic.nimble index 6983d11..2af469c 100644 --- a/lsquic.nimble +++ b/lsquic.nimble @@ -9,7 +9,7 @@ installFiles = @["lsquic.nim"] requires "nim >= 2.0.0" requires "zlib" requires "stew >= 0.4.0" -requires "https://github.com/status-im/nim-chronos#a1f47ec2ae6ad366ebc15a17693f595bcb385502" +requires "https://github.com/status-im/nim-chronos#012cd2c92e9e62b11976ad15b5fd1404e279cf9a" requires "unittest2" requires "chronicles >= 0.11.0" requires "https://github.com/vacp2p/nim-boringssl >= 0.0.11" From 1e5a9f73cc7c861af3dc97673150916859d9b575 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 12:06:42 -0400 Subject: [PATCH 08/15] fix cross-platform wildcard path selection --- lsquic/context/context.nim | 4 ++++ lsquic/context/io.nim | 38 ++++++++++++++++++++++---------------- lsquic/endpoint.nim | 3 --- tests/test_receive.nim | 1 + 4 files changed, 27 insertions(+), 19 deletions(-) diff --git a/lsquic/context/context.nim b/lsquic/context/context.nim index a565c5b..4e914e0 100644 --- a/lsquic/context/context.nim +++ b/lsquic/context/context.nim @@ -40,6 +40,10 @@ type running*: bool ownedCids: HashSet[CidKey] + when defined(windows): + wsaSendMsg*: osdefs.LPFN_WSASENDMSG + wsaSendMsgResolved*: bool + func hash*(cid: CidKey): Hash = var h = hash(cid.len) for i in 0 ..< cid.len.int: diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index ca2dfca..294e378 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -116,6 +116,8 @@ when defined(linux): result = recvmsg(fd, addr msg, 0) if result < 0: return + if (msg.msg_flags and MSG_CTRUNC) != 0: + return -1 remote = toTransportAddress(cast[ptr SockAddr](addr remoteStorage)) if boundLocal.family == AddressFamily.IPv6 and remote.isV4Mapped(): @@ -303,22 +305,26 @@ proc sendPacketsOut*( var bufs {.noinit.}: array[MaxBatch, osdefs.WSABUF] overflow: seq[osdefs.WSABUF] - extension: pointer - extensionBytesRet: DWORD - sendMsgGuid = osdefs.WSAID_WSASENDMSG - if wsaIoctl( - SocketHandle(quicCtx.fd), - osdefs.SIO_GET_EXTENSION_FUNCTION_POINTER, - addr sendMsgGuid, - DWORD(sizeof(sendMsgGuid)), - addr extension, - DWORD(sizeof(extension)), - addr extensionBytesRet, - nil, - nil, - ) != 0: + if not quicCtx.wsaSendMsgResolved: + var + extension: pointer + extensionBytesRet: DWORD + sendMsgGuid = osdefs.WSAID_WSASENDMSG + if wsaIoctl( + SocketHandle(quicCtx.fd), + osdefs.SIO_GET_EXTENSION_FUNCTION_POINTER, + addr sendMsgGuid, + DWORD(sizeof(sendMsgGuid)), + addr extension, + DWORD(sizeof(extension)), + addr extensionBytesRet, + nil, + nil, + ) == 0: + quicCtx.wsaSendMsg = cast[osdefs.LPFN_WSASENDMSG](extension) + quicCtx.wsaSendMsgResolved = true + if quicCtx.wsaSendMsg.isNil: return -1 - let wsaSendMsg = cast[osdefs.LPFN_WSASENDMSG](extension) var sent = 0 for i in 0 ..< nspecs.int: let curr = specsArr[i] @@ -353,7 +359,7 @@ proc sendPacketsOut*( dwBufferCount: DWORD(iovlen), ) prepareSourceAddr(curr.local_sa, control, msg) - let res = wsaSendMsg( + let res = quicCtx.wsaSendMsg( SocketHandle(quicCtx.fd), addr msg, DWORD(0), addr bytesSent, nil, nil ) if res != 0: diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index f556ed8..ea03bb8 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -365,9 +365,6 @@ proc dialLocalAddress( endpoint: QuicEndpoint, remote: TransportAddress ): TransportAddress {.raises: [TransportOsError].} = let bound = endpoint.udp.localAddress() - when not defined(linux): - return bound - if not bound.isAnyLocal(): return bound diff --git a/tests/test_receive.nim b/tests/test_receive.nim index 05c5817..fd685ec 100644 --- a/tests/test_receive.nim +++ b/tests/test_receive.nim @@ -47,6 +47,7 @@ suite "UDP receive": ) if received >= 0: break + # The socket is non-blocking and send completion may precede readability. await sleepAsync(10.milliseconds) check: From a917e0d8ffd2ddb440f69945a57cafa4d4174208 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 12:22:22 -0400 Subject: [PATCH 09/15] fix macOS IPv6 source selection --- lsquic/endpoint.nim | 33 +++++++++++++++++++++++++++++---- 1 file changed, 29 insertions(+), 4 deletions(-) diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index ea03bb8..c18c468 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -2,10 +2,12 @@ # Copyright (c) Status Research & Development GmbH import chronos, chronicles, results +import std/nativesockets except SOL_SOCKET, SO_RCVBUF when defined(windows): - from chronos/osdefs import SOL_SOCKET, SO_RCVBUF + from chronos/osdefs import SOL_SOCKET, SO_RCVBUF, getsockname else: from posix import SOL_SOCKET, SO_RCVBUF + from chronos/osdefs import getsockname import ./[ errors, connection, tlsconfig, connectionmanager, lsquic_ffi, certificateverifier, @@ -13,8 +15,6 @@ import ] import ./context/[server, client, context, io] import ./helpers/logging -when defined(linux): - from chronos/osdefs import SocketHandle logScope: topics = "lsquic" @@ -361,6 +361,31 @@ proc ensureClientContext( endpoint.clientContext +proc selectSourceAddress(remote: TransportAddress): TransportAddress {.raises: [].} = + let domain = + case remote.family + of AddressFamily.IPv4: + AF_INET + of AddressFamily.IPv6: + AF_INET6 + else: + return + let fd = createNativeSocket(domain, SOCK_DGRAM, IPPROTO_UDP) + if fd.int == osInvalidSocket.int: + return + defer: + nativesockets.close(fd) + + var + remoteStorage, localStorage: Sockaddr_storage + remoteLen, localLen: SockLen + remote.toSAddr(remoteStorage, remoteLen) + if connect(fd, cast[ptr SockAddr](addr remoteStorage), remoteLen) != 0: + return + localLen = SockLen(sizeof(localStorage)) + if getsockname(fd, cast[ptr SockAddr](addr localStorage), addr localLen) == 0: + fromSAddr(addr localStorage, localLen, result) + proc dialLocalAddress( endpoint: QuicEndpoint, remote: TransportAddress ): TransportAddress {.raises: [TransportOsError].} = @@ -368,7 +393,7 @@ proc dialLocalAddress( if not bound.isAnyLocal(): return bound - var source = getBestRoute(remote).source + var source = selectSourceAddress(remote) if source.family == AddressFamily.None: return bound From 4f2ddbf0632d4eef89bb644a7b8091c6933e2f4c Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 14:19:58 -0400 Subject: [PATCH 10/15] fix: do not use `result` --- lsquic/context/io.nim | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 294e378..b4cb666 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -113,9 +113,9 @@ when defined(linux): ) msg.msg_controllen = typeof(msg.msg_controllen)(control.data.len) - result = recvmsg(fd, addr msg, 0) - if result < 0: - return + var response = recvmsg(fd, addr msg, 0) + if response < 0: + return response if (msg.msg_flags and MSG_CTRUNC) != 0: return -1 @@ -141,6 +141,8 @@ when defined(linux): if boundLocal.family == AddressFamily.IPv6 and local.family == AddressFamily.IPv4: local = local.toIPv6() + return response + when defined(windows): func wsaCmsgAlign(value: uint): uint = (value + uint(sizeof(uint) - 1)) and not uint(sizeof(uint) - 1) @@ -200,8 +202,9 @@ when not defined(windows): destAddrLen: SockLen, control: var ControlBuffer, ): Tmsghdr = + var response: Tmsghdr when defined(linux) and defined(x86_64) and not defined(android): - result = Tmsghdr( + response = Tmsghdr( msg_name: cast[pointer](addr destStorage), msg_namelen: destAddrLen, msg_iov: cast[ptr IOVec](spec.iov), @@ -211,7 +214,7 @@ when not defined(windows): msg_flags: 0, ) else: - result = Tmsghdr( + response = Tmsghdr( msg_name: cast[pointer](addr destStorage), msg_namelen: destAddrLen, msg_iov: cast[ptr IOVec](spec.iov), @@ -220,7 +223,8 @@ when not defined(windows): msg_controllen: 0, msg_flags: 0, ) - prepareSourceAddr(spec.local_sa, control, result) + prepareSourceAddr(spec.local_sa, control, response) + response proc packetIn*( ctx: QuicContext, From ca2486d46ef0976de1574438ec5f4d607270c560 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 17:33:52 -0400 Subject: [PATCH 11/15] Address cross-platform packet I/O review --- lsquic/context/context.nim | 50 ++++++++- lsquic/context/io.nim | 174 ++++++++++++++++++++----------- lsquic/endpoint.nim | 94 ++++++++++------- lsquic/helpers/transportaddr.nim | 6 ++ tests/test_receive.nim | 101 +++++++++--------- tests/test_send.nim | 166 ++++++++++++++++++++--------- 6 files changed, 390 insertions(+), 201 deletions(-) diff --git a/lsquic/context/context.nim b/lsquic/context/context.nim index 4e914e0..6191c32 100644 --- a/lsquic/context/context.nim +++ b/lsquic/context/context.nim @@ -41,8 +41,54 @@ type ownedCids: HashSet[CidKey] when defined(windows): - wsaSendMsg*: osdefs.LPFN_WSASENDMSG - wsaSendMsgResolved*: bool + wsaSendMsg: osdefs.LPFN_WSASENDMSG + wsaRecvMsg: osdefs.LPFN_WSARECVMSG + +when defined(windows): + proc resolveExtension( + fd: SocketHandle, guid: var GUID, extension: var pointer + ): bool {.raises: [].} = + var bytesReturned: DWORD + wsaIoctl( + fd, + osdefs.SIO_GET_EXTENSION_FUNCTION_POINTER, + addr guid, + DWORD(sizeof(guid)), + addr extension, + DWORD(sizeof(extension)), + addr bytesReturned, + nil, + nil, + ) == 0 + + proc initPacketIo*(ctx: QuicContext, fd: SocketHandle): bool {.raises: [].} = + var + sendExtension, recvExtension: pointer + sendGuid = osdefs.WSAID_WSASENDMSG + recvGuid = osdefs.WSAID_WSARECVMSG + if not resolveExtension(fd, sendGuid, sendExtension) or + not resolveExtension(fd, recvGuid, recvExtension): + return false + ctx.wsaSendMsg = cast[osdefs.LPFN_WSASENDMSG](sendExtension) + ctx.wsaRecvMsg = cast[osdefs.LPFN_WSARECVMSG](recvExtension) + true + + proc sendMsg*( + ctx: QuicContext, fd: SocketHandle, msg: ptr osdefs.WSAMSG, bytesSent: ptr DWORD + ): cint {.inline, raises: [].} = + if ctx.wsaSendMsg.isNil: + return -1 + ctx.wsaSendMsg(fd, msg, DWORD(0), bytesSent, nil, nil) + + proc recvMsg*( + ctx: QuicContext, + fd: SocketHandle, + msg: ptr osdefs.WSAMSG, + bytesReceived: ptr DWORD, + ): cint {.inline, raises: [].} = + if ctx.wsaRecvMsg.isNil: + return -1 + ctx.wsaRecvMsg(fd, msg, bytesReceived, nil, nil) func hash*(cid: CidKey): Hash = var h = hash(cid.len) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index b4cb666..968152b 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -16,6 +16,10 @@ when not defined(windows): logScope: topics = "lsquic" +type ControlBuffer {.union.} = object + alignment: uint + data: array[128, byte] + when defined(windows): const MaxBatch = 1024 ## Upper bound on the stack WSABUF array in the Windows send path (`sendPacketsOut`). @@ -39,25 +43,25 @@ when defined(linux) or defined(macosx): when defined(macosx): {.passc: "-D__APPLE_USE_RFC_3542".} - type - InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object + type In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = object + ipi6_addr: In6Addr + ipi6_ifindex: cuint + + when defined(linux): + type InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object ipi_ifindex: cint ipi_spec_dst: InAddr ipi_addr: InAddr - In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = object - ipi6_addr: In6Addr - ipi6_ifindex: cuint + var IP_PKTINFO {.importc, header: "".}: cint + else: + var + IP_RECVDSTADDR {.importc, header: "".}: cint + IP_SENDSRCADDR {.importc, header: "".}: cint - var - IP_PKTINFO {.importc, header: "".}: cint - IPV6_PKTINFO {.importc, header: "".}: cint + var IPV6_PKTINFO {.importc, header: "".}: cint when not defined(windows): - type ControlBuffer {.union.} = object - alignment: clong - data: array[128, byte] - proc prepareSourceAddr( localSa: ptr SockAddr, control: var ControlBuffer, msg: var Tmsghdr ) = @@ -73,15 +77,22 @@ when not defined(windows): let cmsg = cast[ptr Tcmsghdr](msg.msg_control) if local.family == AddressFamily.IPv4: - msg.msg_controllen = - typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InPktInfo).csize_t)) - cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InPktInfo).csize_t)) cmsg.cmsg_level = IPPROTO_IP - cmsg.cmsg_type = IP_PKTINFO - let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) - copyMem( - addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len - ) + when defined(linux): + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InPktInfo).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InPktInfo).csize_t)) + cmsg.cmsg_type = IP_PKTINFO + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + copyMem( + addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len + ) + else: + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InAddr).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InAddr).csize_t)) + cmsg.cmsg_type = IP_SENDSRCADDR + copyMem(CMSG_DATA(cmsg), unsafeAddr local.address_v4[0], local.address_v4.len) elif local.family == AddressFamily.IPv6: msg.msg_controllen = typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(In6PktInfo).csize_t)) @@ -93,7 +104,7 @@ when not defined(windows): addr info.ipi6_addr, unsafeAddr local.address_v6[0], local.address_v6.len ) -when defined(linux): +when defined(linux) or defined(macosx): proc recvPacket*( fd: SocketHandle, buf: var seq[byte], @@ -113,10 +124,11 @@ when defined(linux): ) msg.msg_controllen = typeof(msg.msg_controllen)(control.data.len) - var response = recvmsg(fd, addr msg, 0) + let response = recvmsg(fd, addr msg, 0) if response < 0: return response if (msg.msg_flags and MSG_CTRUNC) != 0: + trace "Dropping UDP datagram with truncated packet info", boundLocal return -1 remote = toTransportAddress(cast[ptr SockAddr](addr remoteStorage)) @@ -126,29 +138,35 @@ when defined(linux): var cmsg = CMSG_FIRSTHDR(addr msg) while not cmsg.isNil: - if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: - let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) - local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) - copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) - break - elif cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: + when defined(linux): + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break + else: + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_RECVDSTADDR: + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], CMSG_DATA(cmsg), local.address_v4.len) + break + if cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) copyMem(addr local.address_v6[0], addr info.ipi6_addr, local.address_v6.len) break cmsg = CMSG_NXTHDR(addr msg, cmsg) - if boundLocal.family == AddressFamily.IPv6 and local.family == AddressFamily.IPv4: - local = local.toIPv6() - - return response + local = local.matchSocketFamily(boundLocal) + response when defined(windows): + const WsaMsgCtrunc = DWORD(0x0200) + func wsaCmsgAlign(value: uint): uint = (value + uint(sizeof(uint) - 1)) and not uint(sizeof(uint) - 1) proc prepareSourceAddr( - localSa: ptr SockAddr, control: var array[128, byte], msg: var osdefs.WSAMSG + localSa: ptr SockAddr, control: var ControlBuffer, msg: var osdefs.WSAMSG ) = var local = localSa.toTransportAddress() if local.isV4Mapped(): @@ -156,13 +174,13 @@ when defined(windows): if local.isAnyLocal(): return - zeroMem(addr control[0], control.len) + zeroMem(addr control.data[0], control.data.len) let headerLen = wsaCmsgAlign(uint(sizeof(osdefs.WSACMSGHDR))) - header = cast[ptr osdefs.WSACMSGHDR](addr control[0]) + header = cast[ptr osdefs.WSACMSGHDR](addr control.data[0]) data = cast[pointer](cast[uint](header) + headerLen) - msg.control.buf = cast[cstring](addr control[0]) + msg.control.buf = cast[cstring](addr control.data[0]) if local.family == AddressFamily.IPv4: header.cmsg_len = headerLen + uint(sizeof(osdefs.WinInPktInfo)) header.cmsg_level = osdefs.IPPROTO_IP @@ -178,6 +196,64 @@ when defined(windows): let info = cast[ptr osdefs.WinIn6PktInfo](data) copyMem(addr info.ipi6_addr, unsafeAddr local.address_v6[0], local.address_v6.len) + proc recvPacket*( + ctx: QuicContext, + fd: SocketHandle, + buf: var seq[byte], + boundLocal: TransportAddress, + local, remote: var TransportAddress, + ): int {.raises: [].} = + var + remoteStorage: Sockaddr_storage + control: ControlBuffer + data = osdefs.WSABUF(len: ULONG(buf.len), buf: cast[cstring](addr buf[0])) + msg = osdefs.WSAMSG( + name: cast[ptr SockAddr](addr remoteStorage), + namelen: cint(sizeof(remoteStorage)), + lpBuffers: addr data, + dwBufferCount: DWORD(1), + control: osdefs.WSABUF( + len: ULONG(control.data.len), buf: cast[cstring](addr control.data[0]) + ), + ) + bytesReceived: DWORD + + if ctx.recvMsg(fd, addr msg, addr bytesReceived) != 0: + return -1 + if (msg.dwFlags and WsaMsgCtrunc) != 0: + trace "Dropping UDP datagram with truncated packet info", boundLocal + return -1 + + remote = toTransportAddress(cast[ptr SockAddr](addr remoteStorage)) + if boundLocal.family == AddressFamily.IPv6 and remote.isV4Mapped(): + remote = remote.toIPv4() + local = boundLocal + + var + cursor = cast[uint](msg.control.buf) + finish = cursor + uint(msg.control.len) + while cursor + uint(sizeof(osdefs.WSACMSGHDR)) <= finish: + let header = cast[ptr osdefs.WSACMSGHDR](cursor) + if header.cmsg_len < uint(sizeof(osdefs.WSACMSGHDR)) or + cursor + header.cmsg_len > finish: + break + let data = cast[pointer](cursor + wsaCmsgAlign(uint(sizeof(osdefs.WSACMSGHDR)))) + if header.cmsg_level == osdefs.IPPROTO_IP and header.cmsg_type == osdefs.IP_PKTINFO: + let info = cast[ptr osdefs.WinInPktInfo](data) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break + if header.cmsg_level == osdefs.IPPROTO_IPV6 and + header.cmsg_type == osdefs.IPV6_PKTINFO: + let info = cast[ptr osdefs.WinIn6PktInfo](data) + local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) + copyMem(addr local.address_v6[0], addr info.ipi6_addr, local.address_v6.len) + break + cursor += wsaCmsgAlign(header.cmsg_len) + + local = local.matchSocketFamily(boundLocal) + bytesReceived.int + proc prepareDestAddr( localSa: ptr SockAddr, destSa: ptr SockAddr, @@ -309,26 +385,6 @@ proc sendPacketsOut*( var bufs {.noinit.}: array[MaxBatch, osdefs.WSABUF] overflow: seq[osdefs.WSABUF] - if not quicCtx.wsaSendMsgResolved: - var - extension: pointer - extensionBytesRet: DWORD - sendMsgGuid = osdefs.WSAID_WSASENDMSG - if wsaIoctl( - SocketHandle(quicCtx.fd), - osdefs.SIO_GET_EXTENSION_FUNCTION_POINTER, - addr sendMsgGuid, - DWORD(sizeof(sendMsgGuid)), - addr extension, - DWORD(sizeof(extension)), - addr extensionBytesRet, - nil, - nil, - ) == 0: - quicCtx.wsaSendMsg = cast[osdefs.LPFN_WSASENDMSG](extension) - quicCtx.wsaSendMsgResolved = true - if quicCtx.wsaSendMsg.isNil: - return -1 var sent = 0 for i in 0 ..< nspecs.int: let curr = specsArr[i] @@ -354,7 +410,7 @@ proc sendPacketsOut*( dst[j].buf = cast[cstring](src.iov_base) var - control: array[128, byte] + control: ControlBuffer bytesSent: DWORD msg = osdefs.WSAMSG( name: cast[ptr SockAddr](addr destStorage), @@ -363,9 +419,7 @@ proc sendPacketsOut*( dwBufferCount: DWORD(iovlen), ) prepareSourceAddr(curr.local_sa, control, msg) - let res = quicCtx.wsaSendMsg( - SocketHandle(quicCtx.fd), addr msg, DWORD(0), addr bytesSent, nil, nil - ) + let res = quicCtx.sendMsg(SocketHandle(quicCtx.fd), addr msg, addr bytesSent) if res != 0: let errorCode = osdefs.wsaGetLastError() trace "Failed to send UDP datagram", diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index c18c468..2e0b17d 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -14,7 +14,7 @@ import socketconfig, engine_config, ] import ./context/[server, client, context, io] -import ./helpers/logging +import ./helpers/[logging, transportaddr] logScope: topics = "lsquic" @@ -41,9 +41,9 @@ type drainBuf: seq[byte] engineConfig: QuicEngineConfig -const CloseWait: Duration = 300.milliseconds -when defined(linux): - const MaxDatagramsPerWakeup = 64 +const + CloseWait: Duration = 300.milliseconds + MaxDatagramsPerWakeup = 64 ## Capped so that a busy socket cannot starve the rest of the event loop. proc socketReceiveBufferBytes( @@ -86,6 +86,10 @@ proc createServerContext( var context = ServerContext.new(tlsConfig, engineConfig).valueOr: raise newException(QuicError, error) context.fd = fd + when defined(windows): + if not context.initPacketIo(SocketHandle(fd)): + context.destroy() + raise newException(QuicError, "could not initialize Windows packet I/O") context proc createClientContext( @@ -94,6 +98,10 @@ proc createClientContext( var context = ClientContext.new(tlsConfig, engineConfig).valueOr: raise newException(QuicError, error) context.fd = fd + when defined(windows): + if not context.initPacketIo(SocketHandle(fd)): + context.destroy() + raise newException(QuicError, "could not initialize Windows packet I/O") context proc scidLen(endpoint: QuicEndpoint): cuint {.raises: [].} = @@ -190,38 +198,50 @@ proc routeDatagram( trace "Dropping packet with unknown connection ID", bytes = data.len, local, remote {} -when defined(linux): - proc drainDatagrams( - endpoint: QuicEndpoint, udp: DatagramTransport - ): set[RouteTarget] {.raises: [].} = - ## Chronos hands this callback a single datagram per event-loop wakeup, so - ## whatever else has already arrived is read here rather than one wakeup, and - ## one engine tick, at a time. - if endpoint.drainBuf.len == 0: - endpoint.drainBuf = newSeq[byte](DefaultDatagramBufferSize) +proc drainDatagrams( + endpoint: QuicEndpoint, udp: DatagramTransport +): set[RouteTarget] {.raises: [].} = + ## Chronos hands this callback a single datagram per event-loop wakeup, so + ## whatever else has already arrived is read here rather than one wakeup, and + ## one engine tick, at a time. + if endpoint.drainBuf.len == 0: + endpoint.drainBuf = newSeq[byte](DefaultDatagramBufferSize) - var - targets: set[RouteTarget] - boundLocal: TransportAddress - try: - boundLocal = udp.localAddress() - except TransportOsError: - return + var + targets: set[RouteTarget] + boundLocal: TransportAddress + try: + boundLocal = udp.localAddress() + except TransportOsError: + return - for _ in 0 ..< MaxDatagramsPerWakeup: - var local, remote: TransportAddress - let res = + for _ in 0 ..< MaxDatagramsPerWakeup: + var local, remote: TransportAddress + let res = + when defined(windows): + let ctx = + if not endpoint.clientContext.isNil: + endpoint.clientContext + else: + endpoint.serverContext + if ctx.isNil: + -1 + else: + recvPacket( + ctx, SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote + ) + else: recvPacket(SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote) - if res < 0: - # Empty, or an error the transport will report again on the next wakeup. - break + if res < 0: + # Empty, or an error the transport will report again on the next wakeup. + break - if res > 0: - targets.incl endpoint.routeDatagram( - endpoint.drainBuf.toOpenArray(0, res - 1), local, remote - ) + if res > 0: + targets.incl endpoint.routeDatagram( + endpoint.drainBuf.toOpenArray(0, res - 1), local, remote + ) - targets + targets proc readIncoming( udp: DatagramTransport, msg: var seq[byte], msgLen: var int @@ -246,11 +266,7 @@ proc receiveFromUdp( var msg: seq[byte] msgLen: int - local = udp.receivedLocalAddress() - if udp.localAddress().family == AddressFamily.IPv6 and - local.family == AddressFamily.IPv4: - # LSQUIC identifies paths using the socket address family. - local = local.toIPv6() + local = udp.receivedLocalAddress().matchSocketFamily(udp.localAddress()) readIncoming(udp, msg, msgLen) if msgLen > 0: targets = endpoint.routeDatagram(msg.toOpenArray(0, msgLen - 1), local, remote) @@ -258,8 +274,7 @@ proc receiveFromUdp( warn "Failed to read UDP datagram", error = shortLog(e.msg) return - when defined(linux): - targets = targets + endpoint.drainDatagrams(udp) + targets = targets + endpoint.drainDatagrams(udp) if rtClient in targets: endpoint.clientContext.processWhenReady() @@ -397,8 +412,7 @@ proc dialLocalAddress( if source.family == AddressFamily.None: return bound - if bound.family == AddressFamily.IPv6 and source.family == AddressFamily.IPv4: - source = source.toIPv6() + source = source.matchSocketFamily(bound) source.port = bound.port source diff --git a/lsquic/helpers/transportaddr.nim b/lsquic/helpers/transportaddr.nim index f7e1a08..dcffdde 100644 --- a/lsquic/helpers/transportaddr.nim +++ b/lsquic/helpers/transportaddr.nim @@ -37,3 +37,9 @@ proc isIPv6Family*(sock: ptr SockAddr): bool {.inline.} = proc isIPv4Family*(sock: ptr SockAddr): bool {.inline.} = sock.sa_family.int == AF_INET.int + +func matchSocketFamily*(address, bound: TransportAddress): TransportAddress {.inline.} = + if bound.family == AddressFamily.IPv6 and address.family == AddressFamily.IPv4: + address.toIPv6() + else: + address diff --git a/tests/test_receive.nim b/tests/test_receive.nim index fd685ec..3075427 100644 --- a/tests/test_receive.nim +++ b/tests/test_receive.nim @@ -3,10 +3,10 @@ {.used.} -import chronos, chronos/unittest2/asynctests -when defined(linux): - import chronos/osdefs - import ../lsquic/context/io +import chronos, chronos/osdefs, chronos/unittest2/asynctests +import ../lsquic/context/io +when defined(windows): + import ../lsquic/context/context import ./helpers/trackers suite "UDP receive": @@ -14,46 +14,53 @@ suite "UDP receive": checkTrackers() asyncTest "drained IPv4 burst keeps the dual-stack socket family": - when defined(linux): - proc ignore( - udp: DatagramTransport, remote: TransportAddress - ): Future[void] {.async: (raises: []).} = - discard - - let - server = newDatagramTransport6( - ignore, - local = initTAddress("[::]:0"), - flags = {ServerFlags.NoAutoRead, ServerFlags.PacketInfo}, - ) - client = newDatagramTransport(ignore) - defer: - await allFutures(server.closeWait(), client.closeWait()) - - var destination = initTAddress("127.0.0.1:0") - destination.port = server.localAddress().port - await client.sendTo(destination, @[1.byte]) - await client.sendTo(destination, @[2.byte]) - - var buf = newSeq[byte](DefaultDatagramBufferSize) - for expected in [1.byte, 2.byte]: - var - local, remote: TransportAddress - received = -1 - - for _ in 0 ..< 100: - received = recvPacket( - SocketHandle(server.fd), buf, server.localAddress(), local, remote - ) - if received >= 0: - break - # The socket is non-blocking and send completion may precede readability. - await sleepAsync(10.milliseconds) - - check: - received == 1 - buf[0] == expected - local == destination.toIPv6() - remote.family == AddressFamily.IPv4 - else: - skip() + proc ignore( + udp: DatagramTransport, remote: TransportAddress + ): Future[void] {.async: (raises: []).} = + discard + + let + server = newDatagramTransport6( + ignore, + local = initTAddress("[::]:0"), + flags = {ServerFlags.NoAutoRead, ServerFlags.PacketInfo}, + ) + client = newDatagramTransport(ignore) + defer: + await allFutures(server.closeWait(), client.closeWait()) + + when defined(windows): + let ctx = QuicContext() + check ctx.initPacketIo(SocketHandle(server.fd)) + + var destination = initTAddress("127.0.0.1:0") + destination.port = server.localAddress().port + await client.sendTo(destination, @[1.byte]) + await client.sendTo(destination, @[2.byte]) + + var buf = newSeq[byte](DefaultDatagramBufferSize) + for expected in [1.byte, 2.byte]: + var + local, remote: TransportAddress + received = -1 + + for _ in 0 ..< 100: + received = + when defined(windows): + recvPacket( + ctx, SocketHandle(server.fd), buf, server.localAddress(), local, remote + ) + else: + recvPacket( + SocketHandle(server.fd), buf, server.localAddress(), local, remote + ) + if received >= 0: + break + # The socket is non-blocking and send completion may precede readability. + await sleepAsync(10.milliseconds) + + check: + received == 1 + buf[0] == expected + local == destination.toIPv6() + remote.family == AddressFamily.IPv4 diff --git a/tests/test_send.nim b/tests/test_send.nim index 3ad785b..ddf65e5 100644 --- a/tests/test_send.nim +++ b/tests/test_send.nim @@ -20,6 +20,11 @@ proc makeOutSpec( dest_sa: cast[ptr SockAddr](dest), ) +proc makeContext(fd: SocketHandle): QuicContext = + result = QuicContext(fd: cint(fd)) + when defined(windows): + doAssert result.initPacketIo(fd) + suite "packets out": teardown: checkTrackers() @@ -46,7 +51,7 @@ suite "packets out": defer: nativesockets.close(fd) - let ctx = QuicContext(fd: cint(fd)) + let ctx = makeContext(fd) var payload = @[1'u8, 2, 3] localStorage = toSockaddrStorage(initTAddress("127.0.0.1:1000")) @@ -74,7 +79,7 @@ suite "packets out": defer: nativesockets.close(fd) - let ctx = QuicContext(fd: cint(fd)) + let ctx = makeContext(fd) var payload = @[1'u8, 2, 3] localStorage = toSockaddrStorage(initTAddress("127.0.0.1:1000")) @@ -87,53 +92,110 @@ suite "packets out": check sendPacketsOut(cast[pointer](ctx), addr specs[0], SpecCount.cuint) == SpecCount test "packet info selects the requested IPv4 source address": - when defined(linux): - let - senderFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) - receiverFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) - defer: - nativesockets.close(senderFd) - nativesockets.close(receiverFd) - - var - senderBind = toSockaddrStorage(initTAddress("0.0.0.0:0")) - receiverBind = toSockaddrStorage(initTAddress("127.0.0.2:0")) - check bindSocket( - senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in).SockLen - ) == 0 - check bindSocket( - receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in).SockLen - ) == 0 - - var receiverLen = sizeof(receiverBind).SockLen - check getsockname( - receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen - ) == 0 - - let ctx = QuicContext(fd: cint(senderFd)) - var - payload = @[byte(1), 2, 3] - localStorage = toSockaddrStorage(initTAddress("127.0.0.2:0")) - iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) - spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) - - check sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 - - var - received: array[3, byte] - remoteStorage: Sockaddr_storage - remoteLen = sizeof(remoteStorage).SockLen - check recvfrom( - receiverFd, - addr received[0], - received.len, - 0, - cast[ptr SockAddr](addr remoteStorage), - addr remoteLen, - ) == received.len - - var remote: TransportAddress - fromSAddr(addr remoteStorage, remoteLen, remote) - check remote.toIpAddress() == parseIpAddress("127.0.0.2") - else: - skip() + let + senderFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + receiverFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + requestedSource = when defined(windows): "127.0.0.1" else: "127.0.0.2" + defer: + nativesockets.close(senderFd) + nativesockets.close(receiverFd) + + var + senderBind = toSockaddrStorage(initTAddress("0.0.0.0:0")) + receiverBind = toSockaddrStorage(initTAddress(requestedSource & ":0")) + check bindSocket( + senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in).SockLen + ) == 0 + check bindSocket( + receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in).SockLen + ) == 0 + + var receiverLen = sizeof(receiverBind).SockLen + check getsockname( + receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen + ) == 0 + + let ctx = makeContext(senderFd) + var + payload = @[byte(1), 2, 3] + localStorage = toSockaddrStorage(initTAddress(requestedSource & ":0")) + iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) + spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) + + check sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + + var + received: array[3, byte] + remoteStorage: Sockaddr_storage + remoteLen = sizeof(remoteStorage).SockLen + receiveBuffer = + when defined(windows): + cast[cstring](addr received[0]) + else: + addr received[0] + check recvfrom( + receiverFd, + receiveBuffer, + received.len.cint, + 0, + cast[ptr SockAddr](addr remoteStorage), + addr remoteLen, + ) == received.len + + var remote: TransportAddress + fromSAddr(addr remoteStorage, remoteLen, remote) + check remote.toIpAddress() == parseIpAddress(requestedSource) + + test "packet info selects the requested IPv6 source address": + let + senderFd = createNativeSocket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP) + receiverFd = createNativeSocket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP) + defer: + nativesockets.close(senderFd) + nativesockets.close(receiverFd) + + var + senderBind = toSockaddrStorage(initTAddress("[::]:0")) + receiverBind = toSockaddrStorage(initTAddress("[::1]:0")) + check bindSocket( + senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in6).SockLen + ) == 0 + check bindSocket( + receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in6).SockLen + ) == 0 + + var receiverLen = sizeof(receiverBind).SockLen + check getsockname( + receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen + ) == 0 + + let ctx = makeContext(senderFd) + var + payload = @[byte(1), 2, 3] + localStorage = toSockaddrStorage(initTAddress("[::1]:0")) + iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) + spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) + + check sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + + var + received: array[3, byte] + remoteStorage: Sockaddr_storage + remoteLen = sizeof(remoteStorage).SockLen + receiveBuffer = + when defined(windows): + cast[cstring](addr received[0]) + else: + addr received[0] + check recvfrom( + receiverFd, + receiveBuffer, + received.len.cint, + 0, + cast[ptr SockAddr](addr remoteStorage), + addr remoteLen, + ) == received.len + + var remote: TransportAddress + fromSAddr(addr remoteStorage, remoteLen, remote) + check remote.toIpAddress() == parseIpAddress("::1") From a6fd05db1306a10b4a195b10e40fe91d965e12aa Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 17:45:12 -0400 Subject: [PATCH 12/15] Fix macOS packet info control messages --- lsquic/context/io.nim | 59 +++++++++++++++---------------------------- 1 file changed, 21 insertions(+), 38 deletions(-) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 968152b..86c7553 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -43,23 +43,19 @@ when defined(linux) or defined(macosx): when defined(macosx): {.passc: "-D__APPLE_USE_RFC_3542".} - type In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = object - ipi6_addr: In6Addr - ipi6_ifindex: cuint - - when defined(linux): - type InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object + type + InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object ipi_ifindex: cint ipi_spec_dst: InAddr ipi_addr: InAddr - var IP_PKTINFO {.importc, header: "".}: cint - else: - var - IP_RECVDSTADDR {.importc, header: "".}: cint - IP_SENDSRCADDR {.importc, header: "".}: cint + In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = object + ipi6_addr: In6Addr + ipi6_ifindex: cuint - var IPV6_PKTINFO {.importc, header: "".}: cint + var + IP_PKTINFO {.importc, header: "".}: cint + IPV6_PKTINFO {.importc, header: "".}: cint when not defined(windows): proc prepareSourceAddr( @@ -77,22 +73,15 @@ when not defined(windows): let cmsg = cast[ptr Tcmsghdr](msg.msg_control) if local.family == AddressFamily.IPv4: + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InPktInfo).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InPktInfo).csize_t)) cmsg.cmsg_level = IPPROTO_IP - when defined(linux): - msg.msg_controllen = - typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InPktInfo).csize_t)) - cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InPktInfo).csize_t)) - cmsg.cmsg_type = IP_PKTINFO - let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) - copyMem( - addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len - ) - else: - msg.msg_controllen = - typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InAddr).csize_t)) - cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InAddr).csize_t)) - cmsg.cmsg_type = IP_SENDSRCADDR - copyMem(CMSG_DATA(cmsg), unsafeAddr local.address_v4[0], local.address_v4.len) + cmsg.cmsg_type = IP_PKTINFO + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + copyMem( + addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len + ) elif local.family == AddressFamily.IPv6: msg.msg_controllen = typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(In6PktInfo).csize_t)) @@ -138,17 +127,11 @@ when defined(linux) or defined(macosx): var cmsg = CMSG_FIRSTHDR(addr msg) while not cmsg.isNil: - when defined(linux): - if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: - let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) - local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) - copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) - break - else: - if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_RECVDSTADDR: - local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) - copyMem(addr local.address_v4[0], CMSG_DATA(cmsg), local.address_v4.len) - break + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break if cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) From 5778bd5e26559f9e9f505504d73287446bd73891 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 17:51:03 -0400 Subject: [PATCH 13/15] Fix macOS packet info receive parsing --- lsquic/context/io.nim | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 86c7553..7569638 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -57,6 +57,9 @@ when defined(linux) or defined(macosx): IP_PKTINFO {.importc, header: "".}: cint IPV6_PKTINFO {.importc, header: "".}: cint + when defined(macosx): + var IP_RECVDSTADDR {.importc, header: "".}: cint + when not defined(windows): proc prepareSourceAddr( localSa: ptr SockAddr, control: var ControlBuffer, msg: var Tmsghdr @@ -127,11 +130,17 @@ when defined(linux) or defined(macosx): var cmsg = CMSG_FIRSTHDR(addr msg) while not cmsg.isNil: - if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: - let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) - local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) - copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) - break + when defined(linux): + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break + else: + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_RECVDSTADDR: + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], CMSG_DATA(cmsg), local.address_v4.len) + break if cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) From 4c00d138345cd3ae72eab84c27c4e0d5a6cff692 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 22 Sep 2026 18:02:39 -0400 Subject: [PATCH 14/15] Prevent packet info tests from blocking --- tests/test_send.nim | 78 +++++++++++++++++++++++++-------------------- 1 file changed, 43 insertions(+), 35 deletions(-) diff --git a/tests/test_send.nim b/tests/test_send.nim index ddf65e5..c8b3d8a 100644 --- a/tests/test_send.nim +++ b/tests/test_send.nim @@ -4,8 +4,10 @@ {.used.} import chronos, chronos/osdefs, nativesockets, unittest2 +import chronos/osutils import lsquic/[lsquic_ffi, context/context, context/io] import ./helpers/[address, trackers] +import std/os when not defined(windows): from posix import EAGAIN, EBADF, errno @@ -25,6 +27,34 @@ proc makeContext(fd: SocketHandle): QuicContext = when defined(windows): doAssert result.initPacketIo(fd) +proc receiveWithTimeout( + fd: SocketHandle, + received: var array[3, byte], + remoteStorage: var Sockaddr_storage, + remoteLen: var SockLen, +): int = + doAssert setDescriptorBlocking(fd, false).isOk() + let buffer = + when defined(windows): + cast[cstring](addr received[0]) + else: + addr received[0] + + result = -1 + for _ in 0 ..< 100: + remoteLen = sizeof(remoteStorage).SockLen + result = recvfrom( + fd, + buffer, + received.len.cint, + 0, + cast[ptr SockAddr](addr remoteStorage), + addr remoteLen, + ).int + if result >= 0: + return + os.sleep(10) + suite "packets out": teardown: checkTrackers() @@ -95,7 +125,7 @@ suite "packets out": let senderFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) receiverFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) - requestedSource = when defined(windows): "127.0.0.1" else: "127.0.0.2" + requestedSource = when defined(linux): "127.0.0.2" else: "127.0.0.1" defer: nativesockets.close(senderFd) nativesockets.close(receiverFd) @@ -103,15 +133,15 @@ suite "packets out": var senderBind = toSockaddrStorage(initTAddress("0.0.0.0:0")) receiverBind = toSockaddrStorage(initTAddress(requestedSource & ":0")) - check bindSocket( + require bindSocket( senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in).SockLen ) == 0 - check bindSocket( + require bindSocket( receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in).SockLen ) == 0 var receiverLen = sizeof(receiverBind).SockLen - check getsockname( + require getsockname( receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen ) == 0 @@ -122,25 +152,14 @@ suite "packets out": iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) - check sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + require sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 var received: array[3, byte] remoteStorage: Sockaddr_storage remoteLen = sizeof(remoteStorage).SockLen - receiveBuffer = - when defined(windows): - cast[cstring](addr received[0]) - else: - addr received[0] - check recvfrom( - receiverFd, - receiveBuffer, - received.len.cint, - 0, - cast[ptr SockAddr](addr remoteStorage), - addr remoteLen, - ) == received.len + check receiveWithTimeout(receiverFd, received, remoteStorage, remoteLen) == + received.len var remote: TransportAddress fromSAddr(addr remoteStorage, remoteLen, remote) @@ -157,15 +176,15 @@ suite "packets out": var senderBind = toSockaddrStorage(initTAddress("[::]:0")) receiverBind = toSockaddrStorage(initTAddress("[::1]:0")) - check bindSocket( + require bindSocket( senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in6).SockLen ) == 0 - check bindSocket( + require bindSocket( receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in6).SockLen ) == 0 var receiverLen = sizeof(receiverBind).SockLen - check getsockname( + require getsockname( receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen ) == 0 @@ -176,25 +195,14 @@ suite "packets out": iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) - check sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + require sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 var received: array[3, byte] remoteStorage: Sockaddr_storage remoteLen = sizeof(remoteStorage).SockLen - receiveBuffer = - when defined(windows): - cast[cstring](addr received[0]) - else: - addr received[0] - check recvfrom( - receiverFd, - receiveBuffer, - received.len.cint, - 0, - cast[ptr SockAddr](addr remoteStorage), - addr remoteLen, - ) == received.len + check receiveWithTimeout(receiverFd, received, remoteStorage, remoteLen) == + received.len var remote: TransportAddress fromSAddr(addr remoteStorage, remoteLen, remote) From bc7fdeb25cd42af4a3da9adae14a77ab9b7666bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?rich=CE=9Brd?= Date: Tue, 22 Sep 2026 18:11:01 -0400 Subject: [PATCH 15/15] chore: remove usage of result keyword Refactor receiveWithTimeout and makeContext to use local variables for clarity. --- tests/test_send.nim | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/tests/test_send.nim b/tests/test_send.nim index c8b3d8a..7e4711d 100644 --- a/tests/test_send.nim +++ b/tests/test_send.nim @@ -23,9 +23,10 @@ proc makeOutSpec( ) proc makeContext(fd: SocketHandle): QuicContext = - result = QuicContext(fd: cint(fd)) + var response = QuicContext(fd: cint(fd)) when defined(windows): - doAssert result.initPacketIo(fd) + doAssert response.initPacketIo(fd) + response proc receiveWithTimeout( fd: SocketHandle, @@ -40,10 +41,10 @@ proc receiveWithTimeout( else: addr received[0] - result = -1 + var response: int = -1 for _ in 0 ..< 100: remoteLen = sizeof(remoteStorage).SockLen - result = recvfrom( + response = recvfrom( fd, buffer, received.len.cint, @@ -51,9 +52,10 @@ proc receiveWithTimeout( cast[ptr SockAddr](addr remoteStorage), addr remoteLen, ).int - if result >= 0: - return + if response >= 0: + return response os.sleep(10) + response suite "packets out": teardown: