Network helpers

class Msghdr(name=None, iov=None, control=None, flags=None, namelen=None, controllen=None)#

Message header for io_uring_prep_sendmsg / io_uring_prep_recvmsg: an address, a vector of buffers and ancillary data.

Example #1

# send: an address and the buffers to send
>>> msg = Msghdr(Sockaddr(AF_INET, "127.0.0.1", 9000), [b"hello"], flags=0)
>>> io_uring_prep_sendmsg(sqe, sockfd, msg)
True

def net(ns):
    '''`net_ring`, with an `sqe` of the ring.'''
    net_ring(ns)
    ns['sqe'] = io_uring_get_sqe(ns['ring'])

Example #2

# receive: the kernel fills `name`, the `iov` buffers and `control`
>>> name, buf, control = Sockaddr(), bytearray(1024), bytearray(64)
>>> msg = Msghdr(name, [buf], control)
>>> io_uring_prep_recvmsg(sqe, sockfd, msg)
True

Example #3

# multishot receive with provided buffers: only the lengths are given
>>> msg = Msghdr(namelen=16, controllen=0)
>>> io_uring_prep_recvmsg_multishot(sqe, sockfd, msg)
True

Note

• A request holds msg (and so what it was made from) until its completion: io_uring reads it when submitting and writes namelen, controllen and flags back when done.
• iov is a list of buffers (bytes, bytearray, memoryview, ...); io_uring_prep_recvmsg* needs them and control writable (TypeError).
• name, iov and control are kept alive and locked (no resize) by the Msghdr; namelen / controllen start as the size of the Sockaddr / control given and can be set (e.g. for a multishot receive, where the kernel puts both, with the data, into a provided buffer).

class Cmsghdr#

One control message (struct cmsghdr) of a received Msghdr, from io_uring_recvmsg_cmsg_firsthdr / io_uring_recvmsg_cmsg_nexthdr.

Example

# `out = io_uring_recvmsg_validate(buf, msg)` of a message that came with one `SCM_RIGHTS` fd
>>> cmsg = io_uring_recvmsg_cmsg_firsthdr(out, msg)
>>> while cmsg:
...     cmsg.level, cmsg.type, cmsg.data
...     cmsg = io_uring_recvmsg_cmsg_nexthdr(out, msg, cmsg)
...
(1, 1, b'\x07\x00\x00\x00')

def received_fd(ns):
    '''`out` and `msg` of a received message with one `SCM_RIGHTS` control message (fd `7`), as the kernel lays it
    out in a provided buffer: `[header][name][control][payload]`.'''
    name = bytes(16)
    cmsg = struct.pack('=QII', struct.calcsize('=QII') + 4, socket.SOL_SOCKET, socket.SCM_RIGHTS)
    cmsg += struct.pack('=i', 7) + bytes(4)
    header = struct.pack('=IIII', len(name), len(cmsg), 3, 0)
    ns['msg'] = msg = Msghdr(namelen=len(name), controllen=len(cmsg))
    ns['out'] = io_uring_recvmsg_validate(bytearray(header + name + cmsg + b'abc'), msg)

class RecvmsgOut#

The result of io_uring_recvmsg_validate: the header the kernel put in front of a message received with io_uring_prep_recvmsg_multishot into a provided buffer: [header][name][control][payload]. The header is 16 bytes (four u32), then msg.namelen bytes of address and msg.controllen bytes of control data (the sizes given in the Msghdr, whatever the message had), then the payload; io_uring_recvmsg_name, io_uring_recvmsg_cmsg_firsthdr / _nexthdr and io_uring_recvmsg_payload find each part.

Example

# `buf`: the provided buffer a multishot receive (`msg = Msghdr(namelen=16, controllen=0)`) filled
>>> out = io_uring_recvmsg_validate(buf, msg)
>>> out.namelen, out.controllen, out.payloadlen, out.flags
(16, 0, 5, 0)
>>> io_uring_recvmsg_payload(out, msg)
b'hello'

def received(ns):
    '''`buf` and `msg` of a multishot receive (`Msghdr(namelen=16, controllen=0)`) of `b'hello'` over UDP.'''
    ring = make_ring(ns, entries=8)
    cqe = Cqe()
    rx = keep(ns, socket.socket(socket.AF_INET, socket.SOCK_DGRAM))
    tx = keep(ns, socket.socket(socket.AF_INET, socket.SOCK_DGRAM))
    rx.bind(('127.0.0.1', 0))
    area = bytearray(256)
    sqe = io_uring_get_sqe(ring)
    io_uring_prep_provide_buffers(sqe, area, 1, 1, 0)  # one buffer of 256: group 1, id 0
    io_uring_submit_and_wait(ring, 1)
    io_uring_cq_advance(ring, 1)
    ns['msg'] = msg = Msghdr(namelen=16, controllen=0)
    sqe = io_uring_get_sqe(ring)
    io_uring_prep_recvmsg_multishot(sqe, rx.fileno(), msg)
    sqe.flags = IOSQE_BUFFER_SELECT
    io_uring_sqe_set_buf_group(sqe, 1)
    io_uring_submit(ring)
    tx.sendto(b'hello', rx.getsockname())
    io_uring_wait_cqe_timeout(ring, cqe, timespec(2))
    io_uring_cqe_seen(ring, cqe[0])
    ns['buf'] = area

class Sockaddr(family=None, addr=None, port=None)#

Generic Socket Address. Sockaddr() (no arguments) is empty: room for the kernel to fill in, e.g. for io_uring_prep_accept.

Type

Example

>>> addr = Sockaddr(AF_UNIX, './path')
>>> addr = Sockaddr(AF_INET, '0.0.0.0', 12345)
>>> addr = Sockaddr(AF_INET6, '::1', 12345)
>>> io_uring_prep_bind(sqe, sockfd, addr)
True

def net(ns):
    '''`net_ring`, with an `sqe` of the ring.'''
    net_ring(ns)
    ns['sqe'] = io_uring_get_sqe(ns['ring'])

Note

• IPv6 scope_id (an index or an interface name) can be added like so "ff01::fb%123", "fe80::1%eth0"
• ValueError for a missing/invalid addr or port, OverflowError for port > 65_535, NotImplementedError for any other family.
• Sockaddr() is low level setup, letting you serve/connect directly using path/ip. If you need higher level features you can use getaddrinfo() this lets you connect using domain names, ...

>>> sock = Sockaddr(AF_UNIX, './path')
>>> sock.path
'./path'

>>> Sockaddr(AF_INET, '1.2.3.4', 80).ip
'1.2.3.4'
>>> Sockaddr(AF_INET6, '2001:db8:0:0:0:0:0:1', 80).ip
'2001:db8::1'
>>> Sockaddr(AF_INET6, 'fe80::1%2', 80).ip
'fe80::1%2'

>>> sock = Sockaddr(AF_INET, '127.0.0.1', 1234)
>>> sock.port
1234

>>> sock = Sockaddr(AF_UNIX, './path')
>>> sock.family == AF_UNIX
True

io_uring_recvmsg_validate(buf: Any, msgh: Msghdr) -> RecvmsgOut#

Checks that buf holds a message received by io_uring_prep_recvmsg_multishot (ValueError if it is too short for the namelen and controllen of msgh) and gives its parts.

Type

Note

• The result's namelen, controllen, payloadlen and flags are what recvmsg would have reported for the message; msgh is the Msghdr the request was submitted with.

io_uring_recvmsg_name(o: RecvmsgOut) -> bytes#

The socket address the message came from (bytes: the start of a sockaddr, out.namelen long).

io_uring_recvmsg_cmsg_firsthdr(o: RecvmsgOut, msgh: Msghdr) -> Cmsghdr | None#

The first control message, or None.

Type

io_uring_recvmsg_cmsg_nexthdr(o: RecvmsgOut, msgh: Msghdr, cmsg: Cmsghdr) -> Cmsghdr | None#

The control message after cmsg, or None.

Type

io_uring_recvmsg_payload(o: RecvmsgOut, msgh: Msghdr) -> bytes#

The data of the message (bytes, out.payloadlen long, or less if the buffer ends first).

io_uring_recvmsg_payload_length(o: RecvmsgOut, buf_len: int, msgh: Msghdr) -> int#

How many bytes the payload may take in a buffer of buf_len.

setsockopt(sqe: SQE, sockfd: int, level: int, optname: int, optval: Buffer) -> bool#

Sets the socket option optname of level on sockfd to optval (SOCKET_URING_OP_SETSOCKOPT).

Example #1

>>> val = (1).to_bytes(4, "big")
>>> sqe = io_uring_get_sqe(ring)
>>> setsockopt(sqe, sockfd, SOL_SOCKET, SO_KEEPALIVE, val)
True

>>> val = (0).to_bytes(4, "big")
>>> sqe = io_uring_get_sqe(ring)
>>> setsockopt(sqe, sockfd, SOL_SOCKET, SO_KEEPALIVE, val)
True

def net_ring(ns):
    '''A ring, `sockfd` (a TCP socket, `SO_KEEPALIVE` on) and `fd` (a listening one).'''
    make_ring(ns)
    sock = keep(ns, socket.socket())
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    ns['sockfd'] = sock.fileno()
    ns['fd'] = keep(ns, socket.create_server(('127.0.0.1', 0))).fileno()

Example #2

# a bytes option
>>> val = b"eth1"
>>> sqe = io_uring_get_sqe(ring)
>>> setsockopt(sqe, sockfd, SOL_SOCKET, SO_BINDTODEVICE, val)
True

Note

• the kernel checks len(val) for the option (an int option needs 4 bytes, else cqe.res is -EINVAL).
• watch out for "big" or "little" endian, keep it same or it will switch to systems default.
• Needs Linux 6.7+ (SOCKET_URING_OP_SETSOCKOPT). cqe.res is 0 on success; optval is held until the completion.

getsockopt(sqe: SQE, sockfd: int, level: int, optname: int, optval: Buffer) -> bool#

Gets the socket option optname of level of sockfd into optval (SOCKET_URING_OP_GETSOCKOPT).

Example

# assuming `SO_KEEPALIVE` was previous set to `1`
>>> buf = bytearray(4)
>>> sqe = io_uring_get_sqe(ring)
>>> getsockopt(sqe, sockfd, SOL_SOCKET, SO_KEEPALIVE, buf)
True
>>> io_uring_submit_and_wait(ring, 1)
1
>>> int.from_bytes(buf, "little")  # the kernel writes the system's byte order
1

def net_ring(ns):
    '''A ring, `sockfd` (a TCP socket, `SO_KEEPALIVE` on) and `fd` (a listening one).'''
    make_ring(ns)
    sock = keep(ns, socket.socket())
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    ns['sockfd'] = sock.fileno()
    ns['fd'] = keep(ns, socket.create_server(('127.0.0.1', 0))).fileno()

Note

• cqe.res will return len() of populating data(buf) (the updated optlen, unlike getsockopt(2)).
• the kernel checks len(buf) for the option (an int option needs 4 bytes, else cqe.res is -EINVAL).
• watch out for "big" or "little" endian, keep it same or it will switch to systems default.
• Needs Linux 6.7+ (SOCKET_URING_OP_GETSOCKOPT), and only level SOL_SOCKET is supported. buf is held until the completion.

getsockname(sqe: SQE, fd: int, sockaddr: Sockaddr, peer: int | None = None) -> bool#

Gets the local address of fd (peer=0, the default) or of its peer (peer=1) into sockaddr, written at the completion; sockaddr is held until then.

Example

>>> sockaddr = Sockaddr()
>>> sqe = io_uring_get_sqe(ring)
>>> getsockname(sqe, sockfd, sockaddr)
True

def net_ring(ns):
    '''A ring, `sockfd` (a TCP socket, `SO_KEEPALIVE` on) and `fd` (a listening one).'''
    make_ring(ns)
    sock = keep(ns, socket.socket())
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    ns['sockfd'] = sock.fileno()
    ns['fd'] = keep(ns, socket.create_server(('127.0.0.1', 0))).fileno()

Note

• This function is an alias of io_uring_prep_cmd_getsockname
• Needs Linux 6.19+ (SOCKET_URING_OP_GETSOCKNAME). cqe.res is the real size of the address (more than a Sockaddr holds: truncated); sockaddr.family, .ip, .port / .path then read it.