futex.dhover×138all
#!/usr/bin/env dub
/+ dub.sdl:
    name "io_uring_futex"
    dependency "during" version="~>0.5.0"
    platforms "linux"
    targetPath "build"
+/
/**
 * `io_uring` — async futex wait/wake (`IORING_OP_FUTEX_WAIT`, Linux 6.7).
 *
 * Before 6.7 a thread that wanted to block on a futex had to call `futex(2)`
 * directly, taking it out of the io_uring completion-driven event loop. The
 * 6.7 `FUTEX_WAIT` / `FUTEX_WAKE` ops let a ring park on a 32-bit futex word
 * asynchronously: the wait turns into a CQE that lands whenever the word is
 * woken, so a futex hand-off composes with every other queued operation.
 *
 * This example is a two-thread ping-pong. The main thread submits a
 * `FUTEX_WAIT` against a private 32-bit futex (`FUTEX2_SIZE_U32 |
 * FUTEX2_PRIVATE`, matching any bit via `FUTEX_BITSET_MATCH_ANY`). A helper
 * pthread issues a *legacy* `futex(2)` `FUTEX_WAKE_PRIVATE` to wake it — the
 * io_uring FUTEX2 waiter and the classic futex(2) waker share the same kernel
 * hash bucket, so they interoperate. The helper retries on a short interval
 * (bounded to ~1s) to defeat the inherent race between SQE submission and the
 * kernel actually parking the waiter. We assert the wait CQE `res == 0`.
 *
 * Companion to the io_uring chronology:
 * see docs/research/async-io/io-uring/timeline.md
 *   § "6.7 — Futex, waitid, read-multishot (January 2024)".
 *
 * Run with: `dub run --single futex.d`
 *
 * Portability: if the running kernel has no `io_uring`, or is older than 6.7
 * (no `FUTEX_WAIT` op — detected via `io.probe()` or a `-EINVAL`/`-EOPNOTSUPP`
 * completion), the program prints a `SKIP:` line and exits 0 so it stays green
 * in CI regardless of the host kernel.
 */
module 
(module) io_uring_futex
io_uring — async futex wait/wake (IORING_OP_FUTEX_WAIT, Linux 6.7).
Before 6.7 a thread that wanted to block on a futex had to call futex(2)
directly, taking it out of the io_uring completion-driven event loop. The
6.7 FUTEX_WAIT / FUTEX_WAKE ops let a ring park on a 32-bit futex word
asynchronously: the wait turns into a CQE that lands whenever the word is
woken, so a futex hand-off composes with every other queued operation.
This example is a two-thread ping-pong. The main thread submits a
FUTEX_WAIT against a private 32-bit futex (FUTEX2_SIZE_U32 |
FUTEX2_PRIVATE, matching any bit via FUTEX_BITSET_MATCH_ANY). A helper
pthread issues a legacy futex(2) FUTEX_WAKE_PRIVATE to wake it — the
io_uring FUTEX2 waiter and the classic futex(2) waker share the same kernel
hash bucket, so they interoperate. The helper retries on a short interval
(bounded to ~1s) to defeat the inherent race between SQE submission and the
kernel actually parking the waiter. We assert the wait CQE res == 0.
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md
§ "6.7 — Futex, waitid, read-multishot (January 2024)".
Run with: dub run --single futex.d
Portability
if the running kernel has no io_uring, or is older than 6.7
(no FUTEX_WAIT op — detected via io.probe() or a -EINVAL/-EOPNOTSUPP
completion), the program prints a SKIP: line and exits 0 so it stays green
in CI regardless of the host kernel.
io_uring_futex;

import 
(module) during
Simple 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.posixposix.
(module) core.sys.posix.pthread
D 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 Edition
pthread;
import 
(package) corecore.
(package) core.syssys.
(package) core.sys.posixposix.
(module) core.sys.posix.unistd
D 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 Edition
unistd : 
(alias) io_uring_futex.usleep = int core.sys.posix.unistd.usleep(uint) nothrow @nogc @trustedusleep;
import 
(package) corecore.
(module) core.atomic
The atomic module provides basic support for lock-free
concurrent programming.
Use the -preview=nosharedaccess compiler flag to detect
unsafe individual read or write operations on shared data.
Source
core/atomic.d
Examples
int y = 2;
shared int x = y; // OK

//x++; // read modify write error
x.atomicOp!"+="(1); // OK
//y = x; // read error with preview flag
y = x.atomicLoad(); // OK
assert(y == 3);
//x = 5; // write error with preview flag
x.atomicStore(5); // OK
assert(x.atomicLoad() == 5);

@copyrightCopyright Sean Kelly 2005 - 2016.@licenseBoost License 1.0@authorsSean Kelly, Alex Rønne Petersen, Manu Evans
atomic : 
(alias template) io_uring_futex.atomicLoad = core.atomic.atomicLoad(MemoryOrder ms = MemoryOrder.seq, T)(auto ref return scope const T val) if (!is(T == shared(U), U) && !is(T == shared(inout(U)), U) && !is(T == shared(const(U)), U))
Loads 'val' from memory and returns it.  The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default.  Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
@paramval The target variable.@returnsThe value of 'val'.
atomicLoad, 
(alias template) io_uring_futex.atomicStore = core.atomic.atomicStore(MemoryOrder ms = MemoryOrder.seq, T, V)(ref T val, V newval) if (!is(T == shared) && !is(V == shared))
Writes 'newval' into 'val'.  The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
@paramval The target variable.@paramnewval The value to store.
atomicStore, 
(enum) core.atomic.MemoryOrder
Specifies the memory ordering semantics of an atomic operation.
@see
MemoryOrder;

import 
(package) stdstd.
(module) std.stdio
Category 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:
  1. The lowest layer is the operating system layer. The two main schemes are Windows and Posix.
  2. C's stdio.h which unifies the two operating system schemes.
  3. 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 Petersen
stdio : 
(alias template) io_uring_futex.writefln = std.stdio.writefln(alias fmt, A...)(A args) if (isSomeString!(typeof(fmt)))
Equivalent to writef(fmt, args, '\n').
writefln, stderr;

// errno values we treat as "feature unsupported" rather than a hard failure.
private enum 
(constant) int io_uring_futex.EINVAL = 22EINVAL = 22;
private enum 
(constant) int io_uring_futex.EOPNOTSUPP = 95EOPNOTSUPP = 95;

// Legacy futex(2) syscall number + flags, used by the waker thread.
version (
X86_64X86_64) private enum 
(constant) int io_uring_futex.SYS_futex = 202SYS_futex = 202;
else version (AArch64) private enum SYS_futex = 98;
else version (X86) private enum SYS_futex = 240;
else static assert(0, "Unsupported platform for the futex(2) waker");

private enum 
(constant) int io_uring_futex.FUTEX_WAKE = 1FUTEX_WAKE = 1;
private enum 
(constant) int io_uring_futex.FUTEX_PRIVATE_FLAG = 128FUTEX_PRIVATE_FLAG = 128;
private enum 
(constant) int io_uring_futex.FUTEX_WAKE_PRIVATE = 129FUTEX_WAKE_PRIVATE = 
(constant) int io_uring_futex.FUTEX_WAKE = 1FUTEX_WAKE | 
(constant) int io_uring_futex.FUTEX_PRIVATE_FLAG = 128FUTEX_PRIVATE_FLAG;

private extern (C) int 
int io_uring_futex.syscall(int sysno, ...) nothrow @nogc @systemsyscall(int 
(parameter) int sysnosysno, ...) nothrow @nogc @system;

// Shared state between the main thread (waiter) and the helper (waker).
private struct 
(struct) io_uring_futex.WakeCtxWakeCtx
{
    uint* 
(field) uint* io_uring_futex.WakeCtx.wordword; // the futex word both threads agree on
    shared int 
(field) shared(int) io_uring_futex.WakeCtx.stopstop; // main thread sets this once the CQE arrives
    int 
(field) int io_uring_futex.WakeCtx.attemptsattempts; // how many wake retries it took (diagnostic)
}

// Helper thread: repeatedly issue a legacy FUTEX_WAKE on the shared word until
// the main thread signals `stop` (it got its completion) or we hit the retry
// cap. Retrying defeats the race where the wake fires before the kernel has
// parked the io_uring waiter. @nogc/nothrow so it is safe as a raw pthread fn.
private extern (C) void* 
void* io_uring_futex.wakeWorker(void* arg) nothrow @nogc @systemwakeWorker(void* 
(parameter) void* argarg) @system nothrow @nogc
{
    auto 
(local variable) io_uring_futex.WakeCtx* ctxctx = cast(
(struct) io_uring_futex.WakeCtxWakeCtx*) 
(parameter) void* argarg;
    foreach (
(local variable) int ii; 0 .. 200) // 200 * 5ms = ~1s upper bound
    {
        if (
int core.atomic.atomicLoad!(MemoryOrder.acq, int)(ref return scope shared(const(int)) val) pure nothrow @nogc @trusted
Loads 'val' from memory and returns it.  The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default.  Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
@paramval The target variable.@returnsThe value of 'val'.
atomicLoad!(
(enum) core.atomic.MemoryOrder
Specifies the memory ordering semantics of an atomic operation.
@see
MemoryOrder.
(enum value) core.atomic.MemoryOrder.acq = 2
Hoist-load + hoist-store barrier.
Corresponds to LLVM AtomicOrdering.Acquire
and C++11/C11 memory_order_acquire.
acq)(
(local variable) io_uring_futex.WakeCtx* ctxctx.
(field) shared(int) io_uring_futex.WakeCtx.stopstop))
            break;
        
int core.sys.posix.unistd.usleep(uint) nothrow @nogc @trustedusleep(5_000);
        
(local variable) io_uring_futex.WakeCtx* ctxctx.
(field) int io_uring_futex.WakeCtx.attemptsattempts = 
(local variable) int ii + 1;
        
int io_uring_futex.syscall(int sysno, ...) nothrow @nogc @systemsyscall(
(constant) int io_uring_futex.SYS_futex = 202SYS_futex, cast(void*) 
(local variable) io_uring_futex.WakeCtx* ctxctx.
(field) uint* io_uring_futex.WakeCtx.wordword, 
(constant) int io_uring_futex.FUTEX_WAKE_PRIVATE = 129FUTEX_WAKE_PRIVATE, 1, null, null, 0);
    }
    return null;
}

int 
int D main()main()
{
    enum ulong 
(constant) ulong io_uring_futex.main.cookie = 64222LUcookie = 0xFADE;

    
(struct) during.Uring
Main 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 @safe
Setup 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) @safe
Equivalent 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;
    }

    // Static capability check: kernels < 6.7 don't advertise FUTEX_WAIT in the
    // probe, so we can skip cleanly before ever submitting an SQE.
    auto 
(local variable) during.Probe probeprobe = 
(local variable) during.Uring ioio.
during.Probe during.Uring.probe() nothrow @nogc @safe
Probes supported operations
probe();
    if (!cast(bool) 
(local variable) during.Probe probeprobe || !
(local variable) during.Probe probeprobe.
bool during.Probe.isSupported(during.io_uring.Operation op) const pure nothrow @nogc @safe
Is operation supported?
isSupported(
(enum) during.io_uring.Operation
Describes the operation to be performed
@seeio_uring_enter(2)
Operation.
(enum value) during.io_uring.Operation.FUTEX_WAIT = cast(ubyte)51u
IORING_OP_FUTEX_WAIT - async futex(2) FUTEX_WAIT
FUTEX_WAIT))
    {
        
void std.stdio.writefln!char(in char[] fmt) @safe
Equivalent to writef(fmt, args, '\n').
writefln("SKIP: IORING_OP_FUTEX_WAIT unsupported (kernel < 6.7)");
        return 0;
    }

    // The futex word starts at 0; FUTEX_WAIT below uses expected value 0, so the
    // kernel parks the ring until someone (our helper) wakes the word.
    uint 
(local variable) uint wordword = 0;
    
(struct) io_uring_futex.WakeCtxWakeCtx 
(local variable) io_uring_futex.WakeCtx ctxctx;
    
(local variable) io_uring_futex.WakeCtx ctxctx.
(field) uint* io_uring_futex.WakeCtx.wordword = &
(local variable) uint wordword;

    // Spawn the waker before submitting so it is already retrying by the time
    // the kernel parks our waiter.
    
(alias) core.sys.posix.sys.types.pthread_t = ulongpthread_t 
(local variable) ulong tidtid;
    if (
int core.sys.posix.pthread.pthread_create(ulong*, scope const(core.sys.posix.sys.types.pthread_attr_t*), extern (C) void* function(void*), void*) nothrow @nogcpthread_create(&
(local variable) ulong tidtid, null, &
void* io_uring_futex.wakeWorker(void* arg) nothrow @nogc @systemwakeWorker, &
(local variable) io_uring_futex.WakeCtx ctxctx) != 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("pthread_create failed");
        return 1;
    }

    // Submit the async FUTEX_WAIT. `val` is the expected current value (0); the
    // request completes when the word is woken (or differs from `val`).
    
(local variable) during.Uring ioio.putWith!(
        (ref SubmissionEntry e, uint* w) {
        e.prepFutexWait(w, 0, FUTEX_BITSET_MATCH_ANY, FUTEX2_SIZE_U32 | FUTEX2_PRIVATE, 0);
        e.user_data = cookie;
    })(&
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, uint* w) nothrow @nogc @safe
{
prepFutexWait(e, cast(const(uint)*)w, 0LU, 4294967295LU, 130u, 0u);
e.user_data = 64222LU;
}
, uint*)(uint* __param_0) nothrow @nogc return ref @safe
Adds 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.
word);

    const 
(local variable) const(int) submittedsubmitted = 
(local variable) during.Uring ioio.
int during.Uring.submit(uint want) nothrow @nogc @safe
Submits 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 error
submit(1);
    if (
(local variable) const(int) submittedsubmitted != 1)
    {
        
void core.atomic.atomicStore!(MemoryOrder.rel, int, int)(ref shared(int) val, int newval) pure nothrow @nogc @trusted
Writes 'newval' into 'val'.  The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
@paramval The target variable.@paramnewval The value to store.
atomicStore!(
(enum) core.atomic.MemoryOrder
Specifies the memory ordering semantics of an atomic operation.
@see
MemoryOrder.
(enum value) core.atomic.MemoryOrder.rel = 3
Sink-load + sink-store barrier.
Corresponds to LLVM AtomicOrdering.Release
and C++11/C11 memory_order_release.
rel)(
(local variable) io_uring_futex.WakeCtx ctxctx.
(field) shared(int) io_uring_futex.WakeCtx.stopstop, 1);
        
int core.sys.posix.pthread.pthread_join(ulong, void**) nothrow @nogcpthread_join(
(local variable) ulong tidtid, null);
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("submit failed: returned %d", 
(local variable) const(int) submittedsubmitted);
        return 1;
    }

    // Block for the wait completion. The helper thread is hammering FUTEX_WAKE,
    // so this is bounded by the helper's ~1s retry budget.
    
(local variable) during.Uring ioio.
int during.Uring.wait(uint want = 1u) nothrow @nogc
Simmilar to submit but with this method we just wait for required number
of CompletionEntries.
@returns0 on success, -errno on error
wait(1);
    const 
(local variable) const(int) resres = 
(local variable) during.Uring ioio.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safe
Get first CompletionEntry from cq ring
front.
(field) int during.io_uring.CompletionEntry.res
result code for this event
res;
    const 
(local variable) const(ulong) echoedechoed = 
(local variable) during.Uring ioio.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safe
Get first CompletionEntry from cq ring
front.
(field) ulong during.io_uring.CompletionEntry.user_data
sqe->data submission passed back
user_data;
    
(local variable) during.Uring ioio.
void during.Uring.popFront() pure nothrow @nogc @safe
Move to next CompletionEntry
popFront();

    // Tell the helper to stop and reap it.
    
void core.atomic.atomicStore!(MemoryOrder.rel, int, int)(ref shared(int) val, int newval) pure nothrow @nogc @trusted
Writes 'newval' into 'val'.  The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
@paramval The target variable.@paramnewval The value to store.
atomicStore!(
(enum) core.atomic.MemoryOrder
Specifies the memory ordering semantics of an atomic operation.
@see
MemoryOrder.
(enum value) core.atomic.MemoryOrder.rel = 3
Sink-load + sink-store barrier.
Corresponds to LLVM AtomicOrdering.Release
and C++11/C11 memory_order_release.
rel)(
(local variable) io_uring_futex.WakeCtx ctxctx.
(field) shared(int) io_uring_futex.WakeCtx.stopstop, 1);
    
int core.sys.posix.pthread.pthread_join(ulong, void**) nothrow @nogcpthread_join(
(local variable) ulong tidtid, null);

    // Runtime fallback: even if the probe lied, an unsupported op reports these.
    if (
(local variable) const(int) resres == -
(constant) int io_uring_futex.EINVAL = 22EINVAL || 
(local variable) const(int) resres == -
(constant) int io_uring_futex.EOPNOTSUPP = 95EOPNOTSUPP)
    {
        
void std.stdio.writefln!(char, const(int))(in char[] fmt, const(int) __param_1) @safe
Equivalent to writef(fmt, args, '\n').
writefln("SKIP: IORING_OP_FUTEX_WAIT rejected at runtime (res=%d) — kernel < 6.7", 
(local variable) const(int) resres);
        return 0;
    }

    if (
(local variable) const(int) resres < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("FUTEX_WAIT completed with error: errno %d", -
(local variable) const(int) resres);
        return 1;
    }

    if (
(local variable) const(ulong) echoedechoed != 
(constant) ulong io_uring_futex.main.cookie = 64222LUcookie)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("user_data mismatch: expected 0x%X, got 0x%X", 
(constant) ulong io_uring_futex.main.cookie = 64222LUcookie, 
(local variable) const(ulong) echoedechoed);
        return 1;
    }

    
void std.stdio.writefln!(char, const(int), int)(in char[] fmt, const(int) __param_1, int __param_2) @safe
Equivalent to writef(fmt, args, '\n').
writefln("ok: io_uring FUTEX_WAIT (res=%d) woken by a legacy futex(2) waker after %d attempt(s)",
        
(local variable) const(int) resres, 
(local variable) io_uring_futex.WakeCtx ctxctx.
(field) int io_uring_futex.WakeCtx.attemptsattempts);
    return 0;
}