Skip to content
Open
Show file tree
Hide file tree
Changes from 11 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion lsquic.nimble
Original file line number Diff line number Diff line change
Expand Up @@ -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#012cd2c92e9e62b11976ad15b5fd1404e279cf9a"
requires "unittest2"
requires "chronicles >= 0.11.0"
requires "https://github.com/vacp2p/nim-boringssl >= 0.0.11"
Expand Down
4 changes: 4 additions & 0 deletions lsquic/context/context.nim
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
244 changes: 198 additions & 46 deletions lsquic/context/io.nim
Original file line number Diff line number Diff line change
Expand Up @@ -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".}
Expand All @@ -34,26 +35,148 @@ when defined(linux):
sockfd: SocketHandle, msgvec: ptr MMsgHdr, vlen: cuint, flags: cint
): cint {.importc, header: "<sys/socket.h>".}

when defined(linux) or defined(macosx):
when defined(macosx):
{.passc: "-D__APPLE_USE_RFC_3542".}

type
InPktInfo {.importc: "struct in_pktinfo", header: "<netinet/in.h>", bycopy.} = object
ipi_ifindex: cint
ipi_spec_dst: InAddr
ipi_addr: InAddr
Comment thread
richard-ramos marked this conversation as resolved.
Resolved

In6PktInfo {.importc: "struct in6_pktinfo", header: "<netinet/in.h>", bycopy.} = object
ipi6_addr: In6Addr
ipi6_ifindex: cuint

var
IP_PKTINFO {.importc, header: "<netinet/in.h>".}: cint
IPV6_PKTINFO {.importc, header: "<netinet/in.h>".}: 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
) =
when defined(linux) or defined(macosx):
var local = localSa.toTransportAddress()
if local.isV4Mapped():
local = local.toIPv4()
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 =
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
)
elif local.family == AddressFamily.IPv6:
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))
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.msg_controllen = typeof(msg.msg_controllen)(control.data.len)

var response = recvmsg(fd, addr msg, 0)
if response < 0:
return response
if (msg.msg_flags and MSG_CTRUNC) != 0:
return -1
Comment thread
richard-ramos marked this conversation as resolved.
Resolved

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:
Comment thread
richard-ramos marked this conversation as resolved.
Resolved
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)

if boundLocal.family == AddressFamily.IPv6 and local.family == AddressFamily.IPv4:
local = local.toIPv6()

return response

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)
Comment thread
richard-ramos marked this conversation as resolved.
Resolved

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])
Comment thread
richard-ramos marked this conversation as resolved.
Resolved
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,
Expand All @@ -77,9 +200,11 @@ when not defined(windows):
spec: struct_lsquic_out_spec,
destStorage: var Sockaddr_storage,
destAddrLen: SockLen,
control: var ControlBuffer,
): Tmsghdr =
var response: Tmsghdr
when defined(linux) and defined(x86_64) and not defined(android):
Tmsghdr(
response = Tmsghdr(
msg_name: cast[pointer](addr destStorage),
msg_namelen: destAddrLen,
msg_iov: cast[ptr IOVec](spec.iov),
Expand All @@ -89,7 +214,7 @@ when not defined(windows):
msg_flags: 0,
)
else:
Tmsghdr(
response = Tmsghdr(
msg_name: cast[pointer](addr destStorage),
msg_namelen: destAddrLen,
msg_iov: cast[ptr IOVec](spec.iov),
Expand All @@ -98,6 +223,8 @@ when not defined(windows):
msg_controllen: 0,
msg_flags: 0,
)
prepareSourceAddr(spec.local_sa, control, response)
response

proc packetIn*(
ctx: QuicContext,
Expand Down Expand Up @@ -144,6 +271,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

Expand All @@ -153,8 +281,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:
Expand All @@ -177,8 +307,28 @@ 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]
if not quicCtx.wsaSendMsgResolved:
var
extension: pointer
extensionBytesRet: DWORD
sendMsgGuid = osdefs.WSAID_WSASENDMSG
if wsaIoctl(
Comment thread
richard-ramos marked this conversation as resolved.
Resolved
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]
Expand All @@ -193,27 +343,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 = quicCtx.wsaSendMsg(
SocketHandle(quicCtx.fd), addr msg, DWORD(0), addr bytesSent, nil, nil
)
if res != 0:
let errorCode = osdefs.wsaGetLastError()
Expand All @@ -223,7 +374,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:
Expand Down
Loading
Loading