sparkles · multishot-accept.d

#!/usr/bin/env dub
/+ dub.sdl:
    name "io_uring_multishot_accept"
    dependency "during" version="~>0.5.0"
    platforms "linux"
    targetPath "build"
+/
/**
 * `io_uring` — multishot accept (`IORING_ACCEPT_MULTISHOT`, Linux 5.19).
 *
 * Classic `IORING_OP_ACCEPT` (5.5) consumes one SQE per connection: you must
 * re-arm an accept after every client. Multishot accept inverts that — a
 * *single* accept SQE stays armed and posts one CQE per incoming connection,
 * each carrying a fresh accepted fd in `res` and `CQEFlags.MORE` set to signal
 * "this completion is not the last; the request lives on". This removes the
 * per-connection submit round-trip from the accept loop of a server.
 *
 * This program builds a libc TCP listener on `127.0.0.1:<ephemeral>`, arms one
 * `prepMultishotAccept` SQE, then opens and connects two loopback clients. It
 * waits for the two accept CQEs and verifies each yields a valid fd (`res >= 0`)
 * with `CQEFlags.MORE` set, proving the one SQE served multiple connections.
 *
 * Companion to the io_uring chronology: see
 * docs/research/async-io/io-uring/timeline.md
 * § "5.19 — Buffer rings, zero-copy groundwork, big SQE/CQE (July 2022)".
 *
 * Run with: `dub run --single multishot-accept.d`
 *
 * Portability: if `io_uring` is unavailable (too old / sandboxed) the program
 * prints `SKIP:` and exits 0. If multishot accept itself is unsupported
 * (kernel < 5.19 — the accept CQE comes back `-EINVAL`), it likewise prints
 * `SKIP:` and exits 0, so it stays green on the older kernels CI runs on.
 */
module 
(module) io_uring_multishot_acceptio_uring — multishot accept (IORING_ACCEPT_MULTISHOT, Linux 5.19).
Classic IORING_OP_ACCEPT (5.5) consumes one SQE per connection: you must
re-arm an accept after every client. Multishot accept inverts that — a
single accept SQE stays armed and posts one CQE per incoming connection,
each carrying a fresh accepted fd in res and CQEFlags.MORE set to signal
"this completion is not the last; the request lives on". This removes the
per-connection submit round-trip from the accept loop of a server.
This program builds a libc TCP listener on 127.0.0.1:<ephemeral>, arms one
prepMultishotAccept SQE, then opens and connects two loopback clients. It
waits for the two accept CQEs and verifies each yields a valid fd (res >= 0)
with CQEFlags.MORE set, proving the one SQE served multiple connections.
Companion to the io_uring chronology: see
docs/research/async-io/io-uring/timeline.md
§ "5.19 — Buffer rings, zero-copy groundwork, big SQE/CQE (July 2022)".
Run with: dub run --single multishot-accept.d
Portability
if io_uring is unavailable (too old / sandboxed) the program
prints SKIP: and exits 0. If multishot accept itself is unsupported
(kernel < 5.19 — the accept CQE comes back -EINVAL), it likewise prints
SKIP: and exits 0, so it stays green on the older kernels CI runs on.io_uring_multishot_accept;

import 
(module) duringSimple idiomatic dlang wrapper around linux io_uring
(see: https://kernel.dk/io_uring.pdf) asynchronous API.during;

import 
(package) corecore.
(package) core.syssys.
(package) core.sys.linuxlinux.
(module) core.sys.linux.errnoD header file for GNU/Linux
glibc stdlib/errno.herrno : 
(alias constant) io_uring_multishot_accept.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL, 
(alias constant) io_uring_multishot_accept.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP, 
(alias constant) io_uring_multishot_accept.ENOSYS = int core.stdc.errno.ENOSYS = 38ENOSYS;
import 
(package) corecore.
(package) core.syssys.
(package) core.sys.posixposix.
(package) core.sys.posix.arpaarpa.
(module) core.sys.posix.arpa.inetD header file for POSIX.
@copyrightCopyright Sean Kelly 2005 - 2009.@licenseBoost License 1.0.@authorsSean Kelly@standardsThe Open Group Base Specifications Issue 6, IEEE Std 1003.1, 2004 Editioninet : 
(alias) io_uring_multishot_accept.htonl = uint core.sys.posix.arpa.inet.htonl(uint) pure nothrow @nogc @trustedhtonl;
import 
(package) corecore.
(package) core.syssys.
(package) core.sys.posixposix.
(package) core.sys.posix.netinetnetinet.
(module) core.sys.posix.netinet.in_D header file for POSIX.
@copyrightCopyright Sean Kelly 2005 - 2009.@licenseBoost License 1.0.@authorsSean Kelly@standardsThe Open Group Base Specifications Issue 6, IEEE Std 1003.1, 2004 Editionin_ : 
(struct) core.sys.posix.netinet.in_.sockaddr_insockaddr_in, 
(alias enum value) io_uring_multishot_accept.AF_INET = core.sys.posix.sys.socket.AF_INET = 2AF_INET;
import 
(package) corecore.
(package) core.syssys.
(package) core.sys.posixposix.
(package) core.sys.posix.syssys.
(module) core.sys.posix.sys.socketD header file for POSIX.
@copyrightCopyright Sean Kelly 2005 - 2009.@licenseBoost License 1.0.@authorsSean Kelly, Alex Rønne Petersen@standardsThe Open Group Base Specifications Issue 6, IEEE Std 1003.1, 2004 Editionsocket : 
(alias) io_uring_multishot_accept.socket = int core.sys.posix.sys.socket.socket(int, int, int) nothrow @nogc @safesocket, 
(alias) io_uring_multishot_accept.setsockopt = int core.sys.posix.sys.socket.setsockopt(int, int, int, scope const(void*), uint) nothrow @nogcsetsockopt, 
(alias) io_uring_multishot_accept.bind = int core.sys.posix.sys.socket.bind(int, scope const(core.sys.posix.sys.socket.sockaddr*), uint) nothrow @nogcbind, 
(alias) io_uring_multishot_accept.listen = int core.sys.posix.sys.socket.listen(int, int) nothrow @nogc @safelisten, 
(alias) io_uring_multishot_accept.connect = int core.sys.posix.sys.socket.connect(int, scope const(core.sys.posix.sys.socket.sockaddr*), uint) nothrow @nogcconnect,
    
(alias) io_uring_multishot_accept.getsockname = int core.sys.posix.sys.socket.getsockname(int, scope core.sys.posix.sys.socket.sockaddr*, scope uint*) nothrow @nogcgetsockname, 
(struct) core.sys.posix.sys.socket.sockaddrsockaddr, socklen_t, 
(alias enum value) io_uring_multishot_accept.SOCK_STREAM = core.sys.posix.sys.socket.SOCK_STREAM = 1SOCK_STREAM, 
(alias enum value) io_uring_multishot_accept.SOL_SOCKET = core.sys.posix.sys.socket.SOL_SOCKET = 1SOL_SOCKET, 
(alias enum value) io_uring_multishot_accept.SO_REUSEADDR = core.sys.posix.sys.socket.SO_REUSEADDR = 2SO_REUSEADDR;
import 
(package) corecore.
(package) core.syssys.
(package) core.sys.posixposix.
(module) core.sys.posix.unistdD header file for POSIX.
@copyrightCopyright Sean Kelly 2005 - 2009.@licenseBoost License 1.0.@authorsSean Kelly@standardsThe Open Group Base Specifications Issue 8, IEEE Std 1003.1, 2024 Editionunistd : 
(alias) io_uring_multishot_accept.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose;

import 
(package) stdstd.
(module) std.stdioCategory Symbols 
File handles _popen File isFileHandle openNetwork stderr stdin stdout 
Reading chunks lines readf readfln readln 
Writing toFile write writef writefln writeln 
Misc KeepTerminator LockType StdioException 
Standard I/O functions that extend core.stdc.stdio.  core.stdc.stdio
is publically imported when importing std.stdio.
There are three layers of I/O:
The lowest layer is the operating system layer. The two main schemes are Windows and Posix.
C's stdio.h which unifies the two operating system schemes.
std.stdio, this module, unifies the various stdio.h implementations into
    a high level package for D programs.
Source
std/stdio.d
@copyrightCopyright The D Language Foundation 2007-.@licenseBoost License 1.0.@authorsWalter Bright,
Andrei Alexandrescu,
Alex Rønne Petersenstdio : 
(alias template) io_uring_multishot_accept.writefln = std.stdio.writefln(alias fmt, A...)(A args) if (isSomeString!(typeof(fmt)))Equivalent to writef(fmt, args, '\n').writefln, stderr;

int 
int D main()main()
{
    enum 
(constant) int io_uring_multishot_accept.main.ACCEPT_TAG = 1ACCEPT_TAG = 1; // user_data cookie for the multishot accept SQE
    enum 
(constant) int io_uring_multishot_accept.main.CLIENTS = 2CLIENTS = 2; // how many loopback connections we drive through

    
(struct) during.UringMain entry point to work with io_uring.
It hides SubmissionQueue and CompletionQueue behind standard range interface.
We put in SubmissionEntry entries and take out CompletionEntry entries.
Use predefined prepXX methods to fill required fields of SubmissionEntry before put or during putWith.
Note
prepXX functions doesn't touch previous entry state, just fills in operation properties. This is because for
less error prone interface it is cleared automatically when prepared using putWith. So when using on own SubmissionEntry
(outside submission queue), that would be added to the submission queue using put, be sure its cleared if it's
reused for multiple operations.Uring 
(local variable) during.Uring ioio;
    const 
(local variable) const(int) setupRetsetupRet = 
(local variable) during.Uring ioio.
int during.setup(ref during.Uring uring, uint entries = 128u, during.io_uring.SetupFlags flags = SetupFlags.NONE) nothrow @nogc @safeSetup new instance of io_uring into provided Uring structure.
@paramuring Uring structure to be initialized (must not be already initialized)@paramentries Number of entries to initialize uring with@paramflags SetupFlags to use to initialize uring.@returnsOn succes it returns 0, -errno otherwise.setup(8);
    if (
(local variable) const(int) setupRetsetupRet < 0)
    {
        
void std.stdio.writefln!(char, const(int))(in char[] fmt, const(int) __param_1) @safeEquivalent to writef(fmt, args, '\n').writefln("SKIP: io_uring_setup failed (errno %d) — io_uring unavailable on this host",
            -
(local variable) const(int) setupRetsetupRet);
        return 0;
    }

    // --- libc TCP listener on 127.0.0.1:<ephemeral> -------------------------
    const int 
(local variable) const(int) srvsrv = 
int core.sys.posix.sys.socket.socket(int, int, int) nothrow @nogc @safesocket(
(enum value) core.sys.posix.sys.socket.AF_INET = 2AF_INET, 
(enum value) core.sys.posix.sys.socket.SOCK_STREAM = 1SOCK_STREAM, 0);
    if (
(local variable) const(int) srvsrv < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("socket(srv) failed");
        return 1;
    }
    scope (exit) 
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose(
(local variable) const(int) srvsrv);

    int 
(local variable) int oneone = 1;
    
int core.sys.posix.sys.socket.setsockopt(int, int, int, scope const(void*), uint) nothrow @nogcsetsockopt(
(local variable) const(int) srvsrv, 
(enum value) core.sys.posix.sys.socket.SOL_SOCKET = 1SOL_SOCKET, 
(enum value) core.sys.posix.sys.socket.SO_REUSEADDR = 2SO_REUSEADDR, &
(local variable) int oneone, 
(local variable) int oneone.
(constant) ulong int.sizeof = 4LUsizeof);

    
(struct) core.sys.posix.netinet.in_.sockaddr_insockaddr_in 
(local variable) core.sys.posix.netinet.in_.sockaddr_in saddrsaddr;
    
(local variable) core.sys.posix.netinet.in_.sockaddr_in saddrsaddr.
(field) ushort core.sys.posix.netinet.in_.sockaddr_in.sin_familysin_family = 
(enum value) core.sys.posix.sys.socket.AF_INET = 2AF_INET;
    
(local variable) core.sys.posix.netinet.in_.sockaddr_in saddrsaddr.
(field) ushort core.sys.posix.netinet.in_.sockaddr_in.sin_portsin_port = 0; // ask the kernel for an ephemeral port
    
(local variable) core.sys.posix.netinet.in_.sockaddr_in saddrsaddr.
(field) core.sys.posix.arpa.inet.in_addr core.sys.posix.netinet.in_.sockaddr_in.sin_addrsin_addr.
(field) uint core.sys.posix.arpa.inet.in_addr.s_addrs_addr = 
uint core.sys.posix.arpa.inet.htonl(uint) pure nothrow @nogc @trustedhtonl(0x7f00_0001); // 127.0.0.1
    if (
int core.sys.posix.sys.socket.bind(int, scope const(core.sys.posix.sys.socket.sockaddr*), uint) nothrow @nogcbind(
(local variable) const(int) srvsrv, cast(
(struct) core.sys.posix.sys.socket.sockaddrsockaddr*)&
(local variable) core.sys.posix.netinet.in_.sockaddr_in saddrsaddr, 
(local variable) core.sys.posix.netinet.in_.sockaddr_in saddrsaddr.
(constant) ulong core.sys.posix.netinet.in_.sockaddr_in.sizeof = 16LUsizeof) != 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("bind() failed");
        return 1;
    }
    if (
int core.sys.posix.sys.socket.listen(int, int) nothrow @nogc @safelisten(
(local variable) const(int) srvsrv, 
(constant) int io_uring_multishot_accept.main.CLIENTS = 2CLIENTS) != 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("listen() failed");
        return 1;
    }

    // Read back the kernel-assigned port so the clients can connect to it.
    
(struct) core.sys.posix.netinet.in_.sockaddr_insockaddr_in 
(local variable) core.sys.posix.netinet.in_.sockaddr_in actualactual;
    socklen_t 
(local variable) uint alenalen = 
(local variable) core.sys.posix.netinet.in_.sockaddr_in actualactual.
(constant) ulong core.sys.posix.netinet.in_.sockaddr_in.sizeof = 16LUsizeof;
    if (
int core.sys.posix.sys.socket.getsockname(int, scope core.sys.posix.sys.socket.sockaddr*, scope uint*) nothrow @nogcgetsockname(
(local variable) const(int) srvsrv, cast(
(struct) core.sys.posix.sys.socket.sockaddrsockaddr*)&
(local variable) core.sys.posix.netinet.in_.sockaddr_in actualactual, &
(local variable) uint alenalen) != 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("getsockname() failed");
        return 1;
    }

    // --- arm ONE multishot accept SQE ---------------------------------------
    // The kernel fills `peer`/`peerLen` for each accepted connection; one SQE
    // keeps posting CQEs (each with CQEFlags.MORE) until cancelled or it errors.
    static 
(struct) core.sys.posix.netinet.in_.sockaddr_insockaddr_in 
(thread local global) core.sys.posix.netinet.in_.sockaddr_in io_uring_multishot_accept.main.peerpeer;
    static socklen_t 
(thread local global) uint io_uring_multishot_accept.main.peerLenpeerLen = 
(thread local global) core.sys.posix.netinet.in_.sockaddr_in io_uring_multishot_accept.main.peerpeer.
(constant) ulong core.sys.posix.netinet.in_.sockaddr_in.sizeof = 16LUsizeof;
    
(local variable) during.Uring ioio.putWith!(
        (ref SubmissionEntry e, int fd)
        {
            e.prepMultishotAccept(fd, peer, peerLen);
            e.user_data = ACCEPT_TAG;
        })(
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int fd) nothrow @nogc @safe
{
prepMultishotAccept(e, fd, peer, peerLen, AcceptFlags.NONE);
e.user_data = 1LU;
}
, const(int))(ref const(int) __param_0) nothrow @nogc return ref @safeAdds new entry to the SubmissionQueue.
Note that this just adds entry to the queue and doesn't advance the tail
marker kernel sees. For that finishSq() is needed to be called next.
Also note that to actually enter new entries to kernel,
it's needed to call submit().
@paramFN Function to fill next entry in queue by ref (should be faster). It is expected to be in a form of void function(ARGS)(ref SubmissionEntry, auto ref ARGS). Note that in this case queue entry is cleaned first before function is called.@paramentry Custom built SubmissionEntry to be posted as is. Note that in this case it is copied whole over one in the SubmissionQueue.@paramargs Optional arguments passed to the function@returnsreference to Uring structure so it's possible to chain multiple commands.srv);

    const 
(local variable) const(int) submittedsubmitted = 
(local variable) during.Uring ioio.
int during.Uring.submit(uint want) nothrow @nogc @safeSubmits qued SubmissionEntry to be processed by kernel.
@paramwant number of CompletionEntries to wait for. If 0, this just submits queued entries and returns. If > 0, it blocks until at least wanted number of entries were completed.@paramsig See io_uring_enter(2) man page@returnsNumber of submitted entries on success, -errno on errorsubmit(0); // just flush the SQ; we wait explicitly below
    if (
(local variable) const(int) submittedsubmitted < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("submit failed: errno %d", -
(local variable) const(int) submittedsubmitted);
        return 1;
    }

    // --- drive CLIENTS loopback connections, collecting one CQE each --------
    int[
(constant) int io_uring_multishot_accept.main.CLIENTS = 2CLIENTS] 
(local variable) int[2] clientFdsclientFds = -1;
    scope (exit)
        foreach (
(parameter) int cfdcfd; 
(local variable) int[2] clientFdsclientFds)
            if (
(local variable) int cfdcfd >= 0) 
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose(
(local variable) int cfdcfd);

    int 
(local variable) int acceptedaccepted; // count of successfully accepted server-side fds
    bool 
(local variable) bool sawMoresawMore; // did at least one accept CQE carry CQEFlags.MORE?

    foreach (
(local variable) int ii; 0 .. 
(constant) int io_uring_multishot_accept.main.CLIENTS = 2CLIENTS)
    {
        const int 
(local variable) const(int) clicli = 
int core.sys.posix.sys.socket.socket(int, int, int) nothrow @nogc @safesocket(
(enum value) core.sys.posix.sys.socket.AF_INET = 2AF_INET, 
(enum value) core.sys.posix.sys.socket.SOCK_STREAM = 1SOCK_STREAM, 0);
        if (
(local variable) const(int) clicli < 0)
        {
            stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("socket(cli %d) failed", 
(local variable) int ii);
            return 1;
        }
        
(local variable) int[2] clientFdsclientFds[
(local variable) int ii] = 
(local variable) const(int) clicli;

        // Loopback connect completes synchronously enough that the matching
        // accept CQE is ready by the time we wait for it.
        if (
int core.sys.posix.sys.socket.connect(int, scope const(core.sys.posix.sys.socket.sockaddr*), uint) nothrow @nogcconnect(
(local variable) const(int) clicli, cast(
(struct) core.sys.posix.sys.socket.sockaddrsockaddr*)&
(local variable) core.sys.posix.netinet.in_.sockaddr_in actualactual, 
(local variable) uint alenalen) != 0)
        {
            stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("connect(cli %d) failed", 
(local variable) int ii);
            return 1;
        }

        
(local variable) during.Uring ioio.
int during.Uring.wait(uint want = 1u) nothrow @nogcSimmilar to submit but with this method we just wait for required number
of CompletionEntries.
@returns0 on success, -errno on errorwait(1); // bounded: exactly one CQE per connection we just made
        const 
(local variable) const(during.io_uring.CompletionEntry) cc = 
(local variable) during.Uring ioio.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safeGet first CompletionEntry from cq ringfront;
        const 
(local variable) const(int) resres = 
(local variable) const(during.io_uring.CompletionEntry) cc.
(field) int during.io_uring.CompletionEntry.resresult code for this eventres;
        const 
(local variable) const(during.io_uring.CQEFlags) flagsflags = 
(local variable) const(during.io_uring.CompletionEntry) cc.
(field) during.io_uring.CQEFlags during.io_uring.CompletionEntry.flagsflags;
        
(local variable) during.Uring ioio.
void during.Uring.popFront() pure nothrow @nogc @safeMove to next CompletionEntrypopFront();

        if (
(local variable) const(int) resres < 0)
        {
            // -EINVAL/-EOPNOTSUPP/-ENOSYS on the very first accept => the kernel
            // does not implement multishot accept (predates 5.19). Skip cleanly.
            if (
(local variable) int acceptedaccepted == 0 && (
(local variable) const(int) resres == -
(constant) int core.stdc.errno.EINVAL = 22EINVAL || 
(local variable) const(int) resres == -
(constant) int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP || 
(local variable) const(int) resres == -
(constant) int core.stdc.errno.ENOSYS = 38ENOSYS))
            {
                
void std.stdio.writefln!(char, const(int))(in char[] fmt, const(int) __param_1) @safeEquivalent to writef(fmt, args, '\n').writefln("SKIP: multishot accept unsupported (accept CQE res=%d) — needs Linux >= 5.19",
                    
(local variable) const(int) resres);
                return 0;
            }
            stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("accept CQE %d failed: errno %d", 
(local variable) int ii, -
(local variable) const(int) resres);
            return 1;
        }

        // `res` is a freshly accepted server-side fd — close it once observed.
        
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose(
(local variable) const(int) resres);
        
(local variable) int acceptedaccepted++;
        if (
(local variable) const(during.io_uring.CQEFlags) flagsflags & 
(enum) during.io_uring.CQEFlagsFlags used with CompletionEntryCQEFlags.
(enum value) during.io_uring.CQEFlags.MORE = 2uIORING_CQE_F_MORE (from Linux 5.13)
If set, parent SQE will generate more CQE entriesMORE)
            
(local variable) bool sawMoresawMore = true;
    }

    if (
(local variable) int acceptedaccepted != 
(constant) int io_uring_multishot_accept.main.CLIENTS = 2CLIENTS)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("expected %d accepted connections, got %d", 
(constant) int io_uring_multishot_accept.main.CLIENTS = 2CLIENTS, 
(local variable) int acceptedaccepted);
        return 1;
    }

    // CQEFlags.MORE on each completion is the defining signal of a multishot
    // request: the one accept SQE remains armed for further connections.
    if (!
(local variable) bool sawMoresawMore)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("no accept CQE carried CQEFlags.MORE — request was not multishot");
        return 1;
    }

    
void std.stdio.writefln!(char, int)(in char[] fmt, int __param_1) @safeEquivalent to writef(fmt, args, '\n').writefln("ok: one multishot accept SQE served %d loopback connections (CQEFlags.MORE set, request stays armed)",
        
(local variable) int acceptedaccepted);
    return 0;
}