sparkles · pipe.d

#!/usr/bin/env dub
/+ dub.sdl:
    name "io_uring_pipe"
    dependency "during" version="~>0.5.0"
    platforms "linux"
    targetPath "build"
+/
/**
 * `io_uring` — async pipe creation (`IORING_OP_PIPE`, Linux 6.16).
 *
 * Before 6.16, creating a pipe meant a synchronous `pipe2(2)` syscall outside
 * the ring. `IORING_OP_PIPE` lets the ring itself manufacture the pipe pair:
 * submit one SQE pointing at an `int[2]` buffer and, on completion, the kernel
 * has filled it with the read end (`fds[0]`) and write end (`fds[1]`) — exactly
 * like `pipe2(2)`, but folded into the same submit/complete batch as the rest of
 * your I/O (so a pipe can be created and immediately used in a single linked
 * chain without a syscall round-trip).
 *
 * This example creates a pipe through the ring, then proves it actually works by
 * writing a few bytes into the write end and reading them back out of the read
 * end (plain libc `write`/`read` — the point here is the *creation* op, not the
 * transfer), asserting the bytes round-trip before closing both ends.
 *
 * Companion to the io_uring chronology:
 * see docs/research/async-io/io-uring/timeline.md § "6.16 — Async pipe".
 *
 * Run with: `dub run --single pipe.d`
 *
 * Portability: prints a `SKIP:` line and exits 0 if io_uring is unavailable, or
 * if the `PIPE` op is unsupported (kernel older than the op's introduction —
 * detected via the probe or an `-EINVAL`/`-EOPNOTSUPP` completion), so it stays
 * green on CI hosts running older kernels.
 */
module 
(module) io_uring_pipeio_uring — async pipe creation (IORING_OP_PIPE, Linux 6.16).
Before 6.16, creating a pipe meant a synchronous pipe2(2) syscall outside
the ring. IORING_OP_PIPE lets the ring itself manufacture the pipe pair:
submit one SQE pointing at an int[2] buffer and, on completion, the kernel
has filled it with the read end (fds[0]) and write end (fds[1]) — exactly
like pipe2(2), but folded into the same submit/complete batch as the rest of
your I/O (so a pipe can be created and immediately used in a single linked
chain without a syscall round-trip).
This example creates a pipe through the ring, then proves it actually works by
writing a few bytes into the write end and reading them back out of the read
end (plain libc write/read — the point here is the creation op, not the
transfer), asserting the bytes round-trip before closing both ends.
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md § "6.16 — Async pipe".
Run with: dub run --single pipe.d
Portability
prints a SKIP: line and exits 0 if io_uring is unavailable, or
if the PIPE op is unsupported (kernel older than the op's introduction —
detected via the probe or an -EINVAL/-EOPNOTSUPP completion), so it stays
green on CI hosts running older kernels.io_uring_pipe;

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

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

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_pipe.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL, 
(alias constant) io_uring_pipe.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP, 
(alias constant) io_uring_pipe.ENOSYS = int core.stdc.errno.ENOSYS = 38ENOSYS;
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_pipe.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose, 
(alias) io_uring_pipe.read = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread, 
(alias) io_uring_pipe.write = long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite;

// O_CLOEXEC: ask the kernel to mark both pipe ends close-on-exec, the sane
// default for fds we never intend to leak across an exec. Linux value is octal
// 02000000, identical across the architectures this example targets.
enum int 
(constant) int io_uring_pipe.O_CLOEXEC = 524288O_CLOEXEC = 0x80000;

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;
    }

    // Fast path: if the kernel's op probe answers, trust it. A kernel too old to
    // know about PIPE simply reports it unsupported.
    auto 
(local variable) during.Probe probeprobe = 
(local variable) during.Uring ioio.
during.Probe during.Uring.probe() nothrow @nogc @safeProbes supported operationsprobe();
    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 @safeIs operation supported?isSupported(
(enum) during.io_uring.OperationDescribes the operation to be performed
@seeio_uring_enter(2)Operation.
(enum value) during.io_uring.Operation.PIPE = cast(ubyte)62uIORING_OP_PIPE - async pipe(2)/pipe2(2)PIPE))
    {
        
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent to writef(fmt, args, '\n').writefln("SKIP: IORING_OP_PIPE unsupported on this kernel (needs Linux 6.16+)");
        return 0;
    }

    // The kernel writes the read end into fds[0] and the write end into fds[1].
    // We pass the buffer by pointer through `putWith` and dereference inside the
    // lambda so `prepPipe` binds it by reference.
    int[2] 
(local variable) int[2] fdsfds = [-1, -1];
    
(local variable) during.Uring ioio.putWith!(
        (ref SubmissionEntry e, int[2]* out_)
        {
            e.prepPipe(*out_, O_CLOEXEC);
            e.user_data = 1;
        })(&
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int[2]* out_) nothrow @nogc @safe
{
prepPipe(e, *out_, 524288);
e.user_data = 1LU;
}
, int[2]*)(int[2]* __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.fds);

    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(1);
    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;
    }

    
(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);
    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;
    const 
(local variable) const(ulong) cookiecookie = 
(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;
    
(local variable) during.Uring ioio.
void during.Uring.popFront() pure nothrow @nogc @safeMove to next CompletionEntrypopFront();

    // Runtime fallback: on a kernel without the op, the CQE itself reports the
    // failure. Treat those as "unsupported", anything else as a real error.
    if (
(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: IORING_OP_PIPE rejected (errno %d) — kernel predates the op", -
(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("PIPE completed with error: errno %d", -
(local variable) const(int) resres);
        return 1;
    }
    if (
(local variable) const(ulong) cookiecookie != 1)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("user_data mismatch: expected 1, got %d", 
(local variable) const(ulong) cookiecookie);
        return 1;
    }
    if (
(local variable) int[2] fdsfds[0] < 0 || 
(local variable) int[2] fdsfds[1] < 0)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("PIPE did not fill the fd pair: fds=[%d, %d]", 
(local variable) int[2] fdsfds[0], 
(local variable) int[2] fdsfds[1]);
        return 1;
    }

    // Prove the freshly minted pipe works: a short round-trip through it.
    scope (exit) { 
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose(
(local variable) int[2] fdsfds[0]); 
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose(
(local variable) int[2] fdsfds[1]); }
    ubyte[8] 
(local variable) ubyte[8] txtx = [0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04];
    ubyte[8] 
(local variable) ubyte[8] rxrx = 0;

    const 
(local variable) const(long) wrotewrote = 
long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite(
(local variable) int[2] fdsfds[1], &
(local variable) ubyte[8] txtx[0], 
(local variable) ubyte[8] txtx.
(constant) ulong ubyte[8].length = 8LUlength);
    if (
(local variable) const(long) wrotewrote != 
(local variable) ubyte[8] txtx.
(constant) ulong ubyte[8].length = 8LUlength)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("write to pipe failed: returned %d (errno-side)", 
(local variable) const(long) wrotewrote);
        return 1;
    }
    const 
(local variable) const(long) gotgot = 
long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread(
(local variable) int[2] fdsfds[0], &
(local variable) ubyte[8] rxrx[0], 
(local variable) ubyte[8] rxrx.
(constant) ulong ubyte[8].length = 8LUlength);
    if (
(local variable) const(long) gotgot != 
(local variable) ubyte[8] txtx.
(constant) ulong ubyte[8].length = 8LUlength)
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("read from pipe short: returned %d, expected %d", 
(local variable) const(long) gotgot, 
(local variable) ubyte[8] txtx.
(constant) ulong ubyte[8].length = 8LUlength);
        return 1;
    }
    if (
(local variable) ubyte[8] rxrx[] != 
(local variable) ubyte[8] txtx[])
    {
        stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("pipe round-trip mismatch: tx=%(%02X %), rx=%(%02X %)", 
(local variable) ubyte[8] txtx[], 
(local variable) ubyte[8] rxrx[]);
        return 1;
    }

    
void std.stdio.writefln!(char, int, int, const(long))(in char[] fmt, int __param_1, int __param_2, const(long) __param_3) @safeEquivalent to writef(fmt, args, '\n').writefln("ok: PIPE created fds=[%d, %d] through the ring; %d bytes round-tripped",
        
(local variable) int[2] fdsfds[0], 
(local variable) int[2] fdsfds[1], 
(local variable) const(long) gotgot);
    return 0;
}