sparkles · cqe-skip.d

#!/usr/bin/env dub
/+ dub.sdl:
    name "io_uring_cqe_skip"
    dependency "during" version="~>0.5.0"
    platforms "linux"
    targetPath "build"
+/
/**
 * `io_uring` — skip the completion for a successful SQE
 * (`IOSQE_CQE_SKIP_SUCCESS`, Linux 5.17).
 *
 * Normally every submitted SQE posts exactly one CQE. Setting
 * `IOSQE_CQE_SKIP_SUCCESS` on an SQE flips that: if the request *succeeds* the
 * kernel posts **no** completion for it (an *error* still posts one, so failures
 * are never silently lost). This is a throughput win for fire-and-forget links
 * where the app only cares that the chain finished, not about the intermediate
 * steps.
 *
 * This program builds a two-SQE hard-ordered link of writes to a temp file:
 *   - SQE #1: `prepWrite` with `IO_LINK | CQE_SKIP_SUCCESS` — on success it is
 *     silent, so its CQE is suppressed;
 *   - SQE #2: `prepWrite`, normal flags — it terminates the link and posts the
 *     one CQE we expect to see.
 * It submits both, drains the completion queue with a bounded wait, and asserts
 * that exactly **one** CQE arrived (the last write) — direct proof that the first
 * write's success CQE was skipped. Both writes are also verified to have landed
 * in the file.
 *
 * The `IO_LINK` matters here: `CQE_SKIP_SUCCESS` only suppresses the completion;
 * we still want the first write to actually run, so the two SQEs are chained so
 * the kernel executes #1 then #2 in order, and we can reason about which single
 * CQE survives.
 *
 * Companion to the io_uring chronology:
 * see docs/research/async-io/io-uring/timeline.md
 *   § "5.17 — CQE skip, faster cancel (March 2022)".
 *
 * Run with: `dub run --single cqe-skip.d`
 *
 * Portability: prints `SKIP:` and exits 0 when io_uring is unavailable, or when
 * the running kernel predates `CQE_SKIP_SUCCESS` (detected by the link still
 * posting two CQEs, or a write completing with -EINVAL). Exits nonzero only on a
 * genuinely unexpected syscall failure.
 */
module 
(module) io_uring_cqe_skipio_uring — skip the completion for a successful SQE
(IOSQE_CQE_SKIP_SUCCESS, Linux 5.17).
Normally every submitted SQE posts exactly one CQE. Setting
IOSQE_CQE_SKIP_SUCCESS on an SQE flips that: if the request succeeds the
kernel posts no completion for it (an error still posts one, so failures
are never silently lost). This is a throughput win for fire-and-forget links
where the app only cares that the chain finished, not about the intermediate
steps.
This program builds a two-SQE hard-ordered link of writes to a temp file:
SQE #1: prepWrite with IO_LINK | CQE_SKIP_SUCCESS — on success it is
    silent, so its CQE is suppressed;
SQE #2: prepWrite, normal flags — it terminates the link and posts the
    one CQE we expect to see.
It submits both, drains the completion queue with a bounded wait, and asserts
that exactly one CQE arrived (the last write) — direct proof that the first
write's success CQE was skipped. Both writes are also verified to have landed
in the file.
The IO_LINK matters here: CQE_SKIP_SUCCESS only suppresses the completion;
we still want the first write to actually run, so the two SQEs are chained so
the kernel executes #1 then #2 in order, and we can reason about which single
CQE survives.
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md
§ "5.17 — CQE skip, faster cancel (March 2022)".
Run with: dub run --single cqe-skip.d
Portability
prints SKIP: and exits 0 when io_uring is unavailable, or when
the running kernel predates CQE_SKIP_SUCCESS (detected by the link still
posting two CQEs, or a write completing with -EINVAL). Exits nonzero only on a
genuinely unexpected syscall failure.io_uring_cqe_skip;

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.stdcstdc.
(module) core.stdc.errnoD header file for C99.
pubs.opengroup.org/onlinepubs/009695399/basedefs/errno.h.html, errno.h
Source
core/stdc/errno.d
@copyrightCopyright Sean Kelly 2005 - 2009.@licenseDistributed under the
Boost Software License 1.0.
(See accompanying file LICENSE)@authorsSean Kelly, Alex Rønne Petersen@standardsISO/IEC 9899:1999 (E)errno : 
(alias constant) io_uring_cqe_skip.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL;
import 
(package) corecore.
(package) core.syssys.
(package) core.sys.posixposix.
(module) core.sys.posix.fcntlD 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 Editionfcntl : 
(alias constant) io_uring_cqe_skip.O_CREAT = int core.sys.posix.fcntl.O_CREAT = 64O_CREAT, 
(alias constant) io_uring_cqe_skip.O_RDWR = int core.sys.posix.fcntl.O_RDWR = 2O_RDWR, open;
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_cqe_skip.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose, pread, 
(alias) io_uring_cqe_skip.unlink = int core.sys.posix.unistd.unlink(scope const(char*)) nothrow @nogcunlink;

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 : stderr, 
(alias template) io_uring_cqe_skip.writefln = std.stdio.writefln(alias fmt, A...)(A args) if (isSomeString!(typeof(fmt)))Equivalent to writef(fmt, args, '\n').writefln;

// CQE_SKIP_SUCCESS is enum value 1<<6 in SubmissionEntryFlags; we OR it with
// IO_LINK so the first write both runs *before* the second and stays silent.
private enum ubyte 
(constant) ubyte io_uring_cqe_skip.SKIP_AND_LINK = cast(ubyte)68uSKIP_AND_LINK =
    cast(ubyte)(
(enum) during.io_uring.SubmissionEntryFlagssqe->flagsSubmissionEntryFlags.
(enum value) during.io_uring.SubmissionEntryFlags.IO_LINK = cast(ubyte)4uIOSQE_IO_LINK
If set, the next SQE in the ring will depend on this SQE. A dependent SQE will not be started
until the parent SQE has completed. If the parent SQE fails, then a dependent SQE will be
failed without being started. Link chains can be arbitrarily long, the chain spans any new
SQE that continues tohave the IOSQE_IO_LINK flag set. Once an SQE is encountered that does
not have this flag set, that defines the end of the chain. This features allows to form
dependencies between individual SQEs.
Note
available from Linux 5.3IO_LINK | 
(enum) during.io_uring.SubmissionEntryFlagssqe->flagsSubmissionEntryFlags.
(enum value) during.io_uring.SubmissionEntryFlags.CQE_SKIP_SUCCESS = cast(ubyte)64uIOSQE_CQE_SKIP_SUCCESS - don't post CQE if request succeeded.
Emitting a CQE is expensive from the kernel perspective. Often, it's also not convenient for
the userspace, spends some cycles on processing and just complicates the logic. A similar
problems goes for linked requests, where we post an CQE for each request in the link.
Introduce a new flags, IOSQE_CQE_SKIP_SUCCESS, trying to help with it. When set and a request
completed successfully, it won't generate a CQE. When fails, it produces an CQE, but all
following linked requests will be CQE-less, regardless whether they have
IOSQE_CQE_SKIP_SUCCESS or not. The notion of "fail" is the same as for link
failing-cancellation, where it's opcode dependent, and usually_ result >= 0 is a success,
but not always.
Linked timeouts are a bit special. When the requests it's linked to was not attempted to be
executed, e.g. failing linked requests, it follows the description above. Otherwise, whether
a linked timeout will post a completion or not solely depends on IOSQE_CQE_SKIP_SUCCESS of
that linked timeout request. Linked timeout never "fail" during execution, so for them it's
unconditional. It's expected for users to not really care about the result of it but rely
solely on the result of the master request. Another reason for such a treatment is that it's
racy, and the timeout callback may be running awhile the master request posts its completion.
use case 1: If one doesn't care about results of some requests, e.g. normal timeouts, just
set IOSQE_CQE_SKIP_SUCCESS. Error result will still be posted and need to be handled.
use case 2: Set IOSQE_CQE_SKIP_SUCCESS for all requests of a link but the last, and it'll
post a completion only for the last one if everything goes right, otherwise there will be one
only one CQE for the first failed request.
Note
available from Linux 5.17CQE_SKIP_SUCCESS);

int 
int D main()main()
{
    
(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;
    }

    // A throwaway file in /tmp we read back at the end. O_RDWR so we can pread().
    const(char)* 
(local variable) const(char)* pathpath = "/tmp/io_uring_cqe_skip.tmp";
    const 
(local variable) const(int) fdfd = open(
int core.sys.posix.fcntl.open64(scope const(char*), int, ...) nothrow @nogcpath, 
(constant) int core.sys.posix.fcntl.O_CREAT = 64O_CREAT | 
(constant) int core.sys.posix.fcntl.O_RDWR = 2O_RDWR, 384); // mode 0600 (rw-------)
    if (
(local variable) const(int) fdfd < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("open(%s) failed", 
(local variable) const(char)* pathpath);
        return 1;
    }
    scope (exit)
    {
        
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose(
(local variable) const(int) fdfd);
        
int core.sys.posix.unistd.unlink(scope const(char*)) nothrow @nogcunlink(
(local variable) const(char)* pathpath);
    }

    // Two disjoint payloads written at different offsets so we can verify both
    // actually landed even though only the second reports a completion.
    static immutable ubyte[6] 
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.firstfirst = ['s', 'k', 'i', 'p', 'p', 'd'];
    static immutable ubyte[6] 
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.secondsecond = ['l', 'a', 's', 't', '!', '!'];
    enum ulong 
(constant) ulong io_uring_cqe_skip.main.UD_FIRST = 17LUUD_FIRST = 0x11;
    enum ulong 
(constant) ulong io_uring_cqe_skip.main.UD_SECOND = 34LUUD_SECOND = 0x22;

    // SQE #1: silent-on-success, linked to the next SQE.
    
(local variable) during.Uring ioio.putWith!((ref SubmissionEntry e, int f) {
        e.prepWrite(f, first[], 0);
        e.user_data = UD_FIRST;
        e.flags |= SKIP_AND_LINK;
    })(
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int f) nothrow @nogc @safe
{
prepWrite(e, f, first[], 0L);
e.user_data = 17LU;
cast(int)e.flags |= 68;
}
, 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.fd);

    // SQE #2: terminates the link, normal completion — this is the CQE we expect.
    
(local variable) during.Uring ioio.putWith!((ref SubmissionEntry e, int f) {
        e.prepWrite(f, second[], 16);
        e.user_data = UD_SECOND;
    })(
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int f) nothrow @nogc @safe
{
prepWrite(e, f, second[], 16L);
e.user_data = 34LU;
}
, 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.fd);

    // Submit with `want = 0`: enqueue both SQEs but do *not* block for N
    // completions. This matters here — `submit(2)` would call `submitAndWait(2)`
    // and block forever, because CQE_SKIP_SUCCESS means only ONE CQE is ever
    // posted for this link. We submit, then wait for exactly the 1 we expect.
    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);
    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;
    }
    if (
(local variable) const(int) submittedsubmitted != 2)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("expected to submit 2 SQEs, submitted %d", 
(local variable) const(int) submittedsubmitted);
        return 1;
    }

    // Block for the link's single terminal completion.
    
(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);

    int 
(local variable) int cqeCountcqeCount;
    bool 
(local variable) bool sawFirstsawFirst, 
(local variable) bool sawSecondsawSecond;
    int 
(local variable) int firstResfirstRes, 
(local variable) int secondRessecondRes;
    // Bounded drain: at most the 2 ops we submitted could ever surface.
    foreach (
(local variable) int __; 0 .. 2)
    {
        if (
(local variable) during.Uring ioio.
bool during.Uring.empty() const pure nothrow @nogc @safeCheck if there is some CompletionEntry to process.empty)
            break;
        const 
(local variable) const(ulong) udud = 
(local variable) during.Uring ioio.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safeGet first CompletionEntry from cq ringfront.
(field) ulong during.io_uring.CompletionEntry.user_datasqe->data submission passed backuser_data;
        const 
(local variable) const(int) resres = 
(local variable) during.Uring ioio.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safeGet first CompletionEntry from cq ringfront.
(field) int during.io_uring.CompletionEntry.resresult code for this eventres;
        
(local variable) during.Uring ioio.
void during.Uring.popFront() pure nothrow @nogc @safeMove to next CompletionEntrypopFront();
        
(local variable) int cqeCountcqeCount++;
        if (
(local variable) const(ulong) udud == 
(constant) ulong io_uring_cqe_skip.main.UD_FIRST = 17LUUD_FIRST) { 
(local variable) bool sawFirstsawFirst = true; 
(local variable) int firstResfirstRes = 
(local variable) const(int) resres; }
        else if (
(local variable) const(ulong) udud == 
(constant) ulong io_uring_cqe_skip.main.UD_SECOND = 34LUUD_SECOND) { 
(local variable) bool sawSecondsawSecond = true; 
(local variable) int secondRessecondRes = 
(local variable) const(int) resres; }
    }

    // If the first write reported an error CQE, that error always posts (skip
    // only suppresses *success*). A -EINVAL there is the classic "kernel doesn't
    // understand this flag" signal — treat as feature-unsupported.
    if (
(local variable) bool sawFirstsawFirst && 
(local variable) int firstResfirstRes == -
(constant) int core.stdc.errno.EINVAL = 22EINVAL)
    {
        
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent to writef(fmt, args, '\n').writefln("SKIP: write rejected with -EINVAL — CQE_SKIP_SUCCESS unsupported on this kernel");
        return 0;
    }

    // Pre-5.17 kernels ignore the flag and post a CQE for the (successful) first
    // write too: two completions instead of one. Not an error — just unsupported.
    if (
(local variable) int cqeCountcqeCount == 2 && 
(local variable) bool sawFirstsawFirst && 
(local variable) int firstResfirstRes >= 0)
    {
        
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent to writef(fmt, args, '\n').writefln("SKIP: link posted 2 CQEs (first write completed normally) — "
            ~ "CQE_SKIP_SUCCESS not honored, kernel predates 5.17");
        return 0;
    }

    // From here on the feature is in play: we must have seen exactly the terminal
    // CQE and nothing for the skipped first write.
    if (
(local variable) int cqeCountcqeCount != 1 || !
(local variable) bool sawSecondsawSecond || 
(local variable) bool sawFirstsawFirst)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("unexpected CQE pattern: count=%d sawFirst=%s sawSecond=%s",
            
(local variable) int cqeCountcqeCount, 
(local variable) bool sawFirstsawFirst, 
(local variable) bool sawSecondsawSecond);
        return 1;
    }

    if (
(local variable) int secondRessecondRes != cast(int)
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.secondsecond.
(constant) ulong immutable(ubyte[6]).length = 6LUlength)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("terminal write returned %d, expected %d", 
(local variable) int secondRessecondRes, cast(int)
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.secondsecond.
(constant) ulong immutable(ubyte[6]).length = 6LUlength);
        return 1;
    }

    // The skipped write produced no CQE, but it must still have *run* — read both
    // regions back and confirm the bytes are on disk.
    ubyte[6] 
(local variable) ubyte[6] backback = void;
    if (pread(
long core.sys.posix.unistd.pread64(int, void*, ulong, long) nothrow @nogcfd, &
(local variable) ubyte[6] backback[0], 
(local variable) ubyte[6] backback.
(constant) ulong ubyte[6].length = 6LUlength, 0) != cast(long)
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.firstfirst.
(constant) ulong immutable(ubyte[6]).length = 6LUlength || 
(local variable) ubyte[6] backback[] != 
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.firstfirst[])
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("first (skipped) write did not land on disk");
        return 1;
    }
    if (pread(
long core.sys.posix.unistd.pread64(int, void*, ulong, long) nothrow @nogcfd, &
(local variable) ubyte[6] backback[0], 
(local variable) ubyte[6] backback.
(constant) ulong ubyte[6].length = 6LUlength, 16) != cast(long)
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.secondsecond.
(constant) ulong immutable(ubyte[6]).length = 6LUlength || 
(local variable) ubyte[6] backback[] != 
(immutable global) immutable(ubyte[6]) io_uring_cqe_skip.main.secondsecond[])
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("second write did not land on disk");
        return 1;
    }

    
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent to writef(fmt, args, '\n').writefln("ok: linked writes ran, but only 1 CQE arrived (the terminal op) — "
        ~ "CQE_SKIP_SUCCESS suppressed the first write's success completion");
    return 0;
}