msg-ring.dhover×88all
#!/usr/bin/env dub
/+ dub.sdl:
    name "io_uring_msg_ring"
    dependency "during" version="~>0.5.0"
    platforms "linux"
    targetPath "build"
+/
/**
 * `io_uring` — cross-ring messaging (`IORING_OP_MSG_RING`, Linux 5.18).
 *
 * `MSG_RING` lets one ring post a u64 cookie straight into *another* ring's
 * completion queue. It is the in-kernel wakeup/handoff primitive that powers
 * multi-threaded designs: a worker holding ring A can wake a peer on ring B
 * (often pinned to another core) without any syscall, eventfd, or shared lock —
 * the kernel synthesises a CQE on the destination ring directly.
 *
 * This example creates TWO `Uring` instances in a single process. On ring A it
 * submits `prepMsgRing(ringB.fd, 0, MESSAGE, 0)`: the SQE targets ring B's file
 * descriptor and carries the payload that becomes B's CQE `user_data`. We reap
 * ring A's own send-completion (status 0 = delivered), then wait on ring B and
 * assert it received an unsolicited CQE whose `user_data == MESSAGE`.
 *
 * Companion to the io_uring chronology:
 * see docs/research/async-io/io-uring/timeline.md
 *   § "5.18 — Ring-fd registration, msg-ring, linked-file (May 2022)".
 *
 * Run with: `dub run --single msg-ring.d`
 *
 * Portability: prints a `SKIP:` line and exits 0 if io_uring is unavailable, or
 * if the running kernel is older than 5.18 (no `MSG_RING` op). Demonstrates the
 * feature and prints an `ok:` line on a 5.18+ kernel.
 */
module 
(module) io_uring_msg_ring
io_uring — cross-ring messaging (IORING_OP_MSG_RING, Linux 5.18).
MSG_RING lets one ring post a u64 cookie straight into another ring's
completion queue. It is the in-kernel wakeup/handoff primitive that powers
multi-threaded designs: a worker holding ring A can wake a peer on ring B
(often pinned to another core) without any syscall, eventfd, or shared lock —
the kernel synthesises a CQE on the destination ring directly.
This example creates TWO Uring instances in a single process. On ring A it
submits prepMsgRing(ringB.fd, 0, MESSAGE, 0): the SQE targets ring B's file
descriptor and carries the payload that becomes B's CQE user_data. We reap
ring A's own send-completion (status 0 = delivered), then wait on ring B and
assert it received an unsolicited CQE whose user_data == MESSAGE.
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md
§ "5.18 — Ring-fd registration, msg-ring, linked-file (May 2022)".
Run with: dub run --single msg-ring.d
Portability
prints a SKIP: line and exits 0 if io_uring is unavailable, or
if the running kernel is older than 5.18 (no MSG_RING op). Demonstrates the
feature and prints an ok: line on a 5.18+ kernel.
io_uring_msg_ring;

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

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_msg_ring.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()
{
    // The u64 that ring A pushes into ring B's completion queue. On the target
    // ring it surfaces as the CQE's `user_data`, so the receiver can dispatch on
    // it exactly like a locally-submitted op's cookie.
    enum ulong 
(constant) ulong io_uring_msg_ring.main.message = 5859554002255855343LUmessage = 0x5151_5151_DEAD_BEEF;

    // Two independent rings in one process. In a real design these would belong
    // to different worker threads (often on different cores); MSG_RING is the
    // wakeup edge between them.
    
(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 ringAringA, 
(local variable) during.Uring ringBringB;

    if (
(local variable) during.Uring ringAringA.
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) < 0)
    {
        
void std.stdio.writefln!char(in char[] fmt) @safe
Equivalent to writef(fmt, args, '\n').
writefln("SKIP: io_uring_setup failed — io_uring unavailable on this host");
        return 0;
    }
    if (
(local variable) during.Uring ringBringB.
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) < 0)
    {
        
void std.stdio.writefln!char(in char[] fmt) @safe
Equivalent to writef(fmt, args, '\n').
writefln("SKIP: io_uring_setup failed for second ring — io_uring unavailable");
        return 0;
    }

    // Feature gate: MSG_RING arrived in 5.18. On older kernels probe() reports it
    // unsupported (and a submitted SQE would complete with -EINVAL); skip cleanly.
    auto 
(local variable) during.Probe probeprobe = 
(local variable) during.Uring ringAringA.
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.MSG_RING = cast(ubyte)40u
IORING_OP_MSG_RING - allows an SQE to signal another ring
MSG_RING))
    {
        
void std.stdio.writefln!char(in char[] fmt) @safe
Equivalent to writef(fmt, args, '\n').
writefln("SKIP: IORING_OP_MSG_RING unsupported on this kernel (needs Linux 5.18+)");
        return 0;
    }

    // Post the message from ring A into ring B. `prepMsgRing(fd, len, data, flags)`:
    //   fd   = the *destination* ring's fd (ringB.fd) — that is how the kernel
    //          knows which ring's CQ to push into,
    //   len  = a value handed to the target as its CQE `res` (0 here),
    //   data = the payload delivered as the target CQE's `user_data`.
    enum ulong 
(constant) ulong io_uring_msg_ring.main.sendCookie = 10LUsendCookie = 0xA;
    
(local variable) during.Uring ringAringA.putWith!((ref SubmissionEntry e, int targetFd, ulong data) {
        e.prepMsgRing(targetFd, 0, data, 0);
        e.user_data = sendCookie;
    })(
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int targetFd, ulong data) nothrow @nogc @system
{
prepMsgRing(e, targetFd, 0u, data, 0u);
e.user_data = 10LU;
}
, int, ulong)(int __param_0, ulong __param_1) 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.
ringB.
int during.Uring.fd() const pure nothrow @nogc @safe
Native io_uring file descriptor
fd, 
(constant) ulong io_uring_msg_ring.main.message = 5859554002255855343LUmessage);

    const 
(local variable) const(int) submittedsubmitted = 
(local variable) during.Uring ringAringA.
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 < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("ring A submit failed: errno %d", -
(local variable) const(int) submittedsubmitted);
        return 1;
    }

    // Reap ring A's *own* completion for the send. A 5.18 kernel that recognises
    // the op but cannot deliver returns -EINVAL/-EOPNOTSUPP here — treat that as
    // "feature unsupported" and skip rather than fail.
    
(local variable) during.Uring ringAringA.
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) sendRessendRes = 
(local variable) during.Uring ringAringA.
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) sendEchosendEcho = 
(local variable) during.Uring ringAringA.
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 ringAringA.
void during.Uring.popFront() pure nothrow @nogc @safe
Move to next CompletionEntry
popFront();

    if (
(local variable) const(int) sendRessendRes == -
(constant) int io_uring_msg_ring.EINVAL = 22EINVAL || 
(local variable) const(int) sendRessendRes == -
(constant) int io_uring_msg_ring.EOPNOTSUPP = 95EOPNOTSUPP || 
(local variable) const(int) sendRessendRes == -
(constant) int io_uring_msg_ring.ENOSYS = 38ENOSYS)
    {
        
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_MSG_RING rejected (errno %d) — needs Linux 5.18+", -
(local variable) const(int) sendRessendRes);
        return 0;
    }
    if (
(local variable) const(int) sendRessendRes < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("MSG_RING send completed with error: errno %d", -
(local variable) const(int) sendRessendRes);
        return 1;
    }
    if (
(local variable) const(ulong) sendEchosendEcho != 
(constant) ulong io_uring_msg_ring.main.sendCookie = 10LUsendCookie)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("ring A user_data mismatch: expected 0x%X, got 0x%X", 
(constant) ulong io_uring_msg_ring.main.sendCookie = 10LUsendCookie, 
(local variable) const(ulong) sendEchosendEcho);
        return 1;
    }

    // Now ring B must observe an unsolicited CQE — one we never submitted on B —
    // carrying the payload from A. This is the whole point: a cross-ring wakeup.
    
(local variable) during.Uring ringBringB.
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(ulong) recvDatarecvData = 
(local variable) during.Uring ringBringB.
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;
    const 
(local variable) const(int) recvResrecvRes = 
(local variable) during.Uring ringBringB.
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;
    
(local variable) during.Uring ringBringB.
void during.Uring.popFront() pure nothrow @nogc @safe
Move to next CompletionEntry
popFront();

    if (
(local variable) const(ulong) recvDatarecvData != 
(constant) ulong io_uring_msg_ring.main.message = 5859554002255855343LUmessage)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("ring B got wrong payload: expected 0x%X, got 0x%X", 
(constant) ulong io_uring_msg_ring.main.message = 5859554002255855343LUmessage, 
(local variable) const(ulong) recvDatarecvData);
        return 1;
    }

    
void std.stdio.writefln!(char, const(ulong), const(int), const(int))(in char[] fmt, const(ulong) __param_1, const(int) __param_2, const(int) __param_3) @safe
Equivalent to writef(fmt, args, '\n').
writefln(
        "ok: MSG_RING delivered 0x%X from ring A (send res=%d) into ring B's CQ (res=%d) — cross-ring wakeup with no syscall on B",
        
(local variable) const(ulong) recvDatarecvData, 
(local variable) const(int) sendRessendRes, 
(local variable) const(int) recvResrecvRes);
    return 0;
}

// errno constants used for the feature-unsupported gate above.
private enum int 
(constant) int io_uring_msg_ring.EINVAL = 22EINVAL = 22;
private enum int 
(constant) int io_uring_msg_ring.ENOSYS = 38ENOSYS = 38;
private enum int 
(constant) int io_uring_msg_ring.EOPNOTSUPP = 95EOPNOTSUPP = 95;