#!/usr/bin/env dub
/+ dub.sdl:
name "io_uring_linked_sqes"
dependency "during" version="~>0.5.0"
platforms "linux"
targetPath "build"
+/
/**
* `io_uring` — linked SQEs via `IOSQE_IO_LINK` (Linux 5.3).
*
* `IOSQE_IO_LINK` turns an otherwise unordered batch of submission queue
* entries into an ordered dependency chain: when an SQE carries the
* `IO_LINK` flag, the *next* SQE in the batch will not start until the
* flagged one has completed successfully. This lets you express
* "do B only after A" — e.g. write-then-fsync — without an extra
* userspace submit/wait round-trip.
*
* This example opens a temp file and submits TWO SQEs in one batch:
* 1. WRITE a known payload (user_data = 1), flagged `IO_LINK`.
* 2. FSYNC the same fd (user_data = 2), the link target.
* Because the WRITE is `IO_LINK`-flagged, the kernel guarantees the FSYNC
* runs strictly *after* the WRITE has completed — without the link the two
* could be reordered/run concurrently, and the fsync might flush *before*
* the bytes ever reached the file.
*
* Failure propagation (explained, not triggered here): if a linked SQE
* fails (or is short), the kernel breaks the chain — every *subsequent*
* linked SQE is cancelled with `res == -ECANCELED` and never runs. So if
* the WRITE had failed, the FSYNC would never touch the disk: the chain
* fails closed. (`IO_HARDLINK` is the variant that keeps going regardless
* of the predecessor's result.)
*
* Companion to the io_uring chronology:
* see docs/research/async-io/io-uring/timeline.md
* § "5.3 — Network message ops (September 2019)".
*
* Run with: `dub run --single linked-sqes.d`
*
* Portability: if the running kernel has no `io_uring`, or is too old for
* linked SQEs (pre-5.3), the program prints a `SKIP:` line and exits 0 so
* it stays green in CI regardless of the host kernel.
*/
module (module) io_uring_linked_sqesio_uring — linked SQEs via IOSQE_IO_LINK (Linux 5.3).
IOSQE_IO_LINK turns an otherwise unordered batch of submission queue
entries into an ordered dependency chain: when an SQE carries the
IO_LINK flag, the next SQE in the batch will not start until the
flagged one has completed successfully. This lets you express
"do B only after A" — e.g. write-then-fsync — without an extra
userspace submit/wait round-trip.
This example opens a temp file and submits TWO SQEs in one batch:
WRITE a known payload (user_data = 1), flagged IO_LINK.
FSYNC the same fd (user_data = 2), the link target.
Because the WRITE is IO_LINK-flagged, the kernel guarantees the FSYNC
runs strictly after the WRITE has completed — without the link the two
could be reordered/run concurrently, and the fsync might flush before
the bytes ever reached the file.
Failure propagation (explained, not triggered here): if a linked SQE
fails (or is short), the kernel breaks the chain — every subsequent
linked SQE is cancelled with res == -ECANCELED and never runs. So if
the WRITE had failed, the FSYNC would never touch the disk: the chain
fails closed. (IO_HARDLINK is the variant that keeps going regardless
of the predecessor's result.)
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md
§ "5.3 — Network message ops (September 2019)".
Run with: dub run --single linked-sqes.d
Portability
if the running kernel has no io_uring, or is too old for
linked SQEs (pre-5.3), the program prints a SKIP: line and exits 0 so
it stays green in CI regardless of the host kernel.
io_uring_linked_sqes;
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
stdio : (alias template) io_uring_linked_sqes.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
errno : (alias constant) io_uring_linked_sqes.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL, (alias constant) io_uring_linked_sqes.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP, (alias constant) io_uring_linked_sqes.ENOSYS = int core.stdc.errno.ENOSYS = 38ENOSYS, (alias constant) io_uring_linked_sqes.ECANCELED = int core.stdc.errno.ECANCELED = 125ECANCELED;
import (package) corecore.(package) core.syssys.(package) core.sys.posixposix.(module) core.sys.posix.stdlibD header file for POSIX.
stdlib : mkstemp;
import (package) corecore.(package) core.syssys.(package) core.sys.posixposix.(module) core.sys.posix.unistdD header file for POSIX.
unistd : (alias) io_uring_linked_sqes.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose, (alias) io_uring_linked_sqes.unlink = int core.sys.posix.unistd.unlink(scope const(char*)) nothrow @nogcunlink;
int int D main()main()
{
enum ulong (constant) ulong io_uring_linked_sqes.main.writeTag = 1LUwriteTag = 1; // the link head (runs first)
enum ulong (constant) ulong io_uring_linked_sqes.main.fsyncTag = 2LUfsyncTag = 2; // the link target (runs only after the write)
(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.
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;
}
// Create a private temp file to write+fsync. mkstemp replaces the XXXXXX
// template in place and returns an open fd.
char[] (local variable) char[] tmpltmpl = "/tmp/io_uring_linked_sqes_XXXXXX\0".char[] object.dup!char(const(char)[] a) pure nothrow @property @safedup;
const (local variable) const(int) fdfd = mkstemp(&(local variable) char[] tmpltmpl[0]);
if ((local variable) const(int) fdfd < 0)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("mkstemp failed");
return 1;
}
// Unlink immediately: the file stays alive via the open fd but leaves no
// litter behind when we close it.
int core.sys.posix.unistd.unlink(scope const(char*)) nothrow @nogcunlink(&(local variable) char[] tmpltmpl[0]);
scope (exit) int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose((local variable) const(int) fdfd);
immutable ubyte[] (local variable) immutable(ubyte[]) payloadpayload = cast(immutable(ubyte[])) "linked-sqes: write then fsync, in order\n";
// ---- Submit the linked batch (two SQEs, one submit) -------------------
// SQE #1: WRITE the payload, flagged IO_LINK so SQE #2 depends on it.
(local variable) during.Uring ioio.putWith!((ref SubmissionEntry e, int f, const(ubyte)[] buf)
{
e.prepWrite(f, buf, 0);
e.user_data = writeTag;
// IO_LINK: the *next* SQE in this submission won't start until this
// write completes successfully — the heart of the ordering guarantee.
e.flags |= SubmissionEntryFlags.IO_LINK;
})(during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int f, const(ubyte)[] buf) nothrow @nogc @safe
{
prepWrite(e, f, buf, 0L);
e.user_data = 1LU;
cast(int)e.flags |= 4;
}
, const(int), immutable(ubyte[]))(ref const(int) __param_0, ref immutable(ubyte[]) __param_1) 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().
fd, during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int f, const(ubyte)[] buf) nothrow @nogc @safe
{
prepWrite(e, f, buf, 0L);
e.user_data = 1LU;
cast(int)e.flags |= 4;
}
, const(int), immutable(ubyte[]))(ref const(int) __param_0, ref immutable(ubyte[]) __param_1) 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().
payload);
// SQE #2: FSYNC the same fd — the link target. No IO_LINK here: it ends
// the chain.
(local variable) during.Uring ioio.putWith!((ref SubmissionEntry e, int f)
{
e.prepFsync(f);
e.user_data = fsyncTag;
})(during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int f) nothrow @nogc @safe
{
prepFsync(e, f, FsyncFlags.NORMAL);
e.user_data = 2LU;
}
, 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().
fd);
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.
submit(2);
if ((local variable) const(int) submittedsubmitted < 0)
{
// Linked SQEs arrived in 5.3; a pre-5.3 kernel rejects the batch.
if ((local variable) const(int) submittedsubmitted == -(constant) int core.stdc.errno.EINVAL = 22EINVAL || (local variable) const(int) submittedsubmitted == -(constant) int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP || (local variable) const(int) submittedsubmitted == -(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: linked SQEs (IOSQE_IO_LINK) unsupported on this kernel (errno %d)", -(local variable) const(int) submittedsubmitted);
return 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;
}
// Block until BOTH completions are ready.
(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.
wait(2);
// Drain the completion queue. Completions arrive in chain order: the
// write (head) first, then the fsync (target).
int (local variable) int writeReswriteRes = int.(constant) int int.min = -2147483648min;
int (local variable) int fsyncResfsyncRes = int.(constant) int int.min = -2147483648min;
while (!(local variable) during.Uring ioio.bool during.Uring.empty() const pure nothrow @nogc @safeCheck if there is some CompletionEntry to process.
empty)
{
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 ring
front;
if ((local variable) const(during.io_uring.CompletionEntry) cc.(field) ulong during.io_uring.CompletionEntry.user_datasqe->data submission passed back
user_data == (constant) ulong io_uring_linked_sqes.main.writeTag = 1LUwriteTag)
(local variable) int writeReswriteRes = (local variable) const(during.io_uring.CompletionEntry) cc.(field) int during.io_uring.CompletionEntry.resresult code for this event
res;
else if ((local variable) const(during.io_uring.CompletionEntry) cc.(field) ulong during.io_uring.CompletionEntry.user_datasqe->data submission passed back
user_data == (constant) ulong io_uring_linked_sqes.main.fsyncTag = 2LUfsyncTag)
(local variable) int fsyncResfsyncRes = (local variable) const(during.io_uring.CompletionEntry) cc.(field) int during.io_uring.CompletionEntry.resresult code for this event
res;
(local variable) during.Uring ioio.void during.Uring.popFront() pure nothrow @nogc @safeMove to next CompletionEntry
popFront();
}
// A pre-5.3 kernel that *accepted* the batch but ignored the link flag
// could still reject the flag at completion time — treat that as a SKIP.
if ((local variable) int writeReswriteRes == -(constant) int core.stdc.errno.EINVAL = 22EINVAL || (local variable) int fsyncResfsyncRes == -(constant) int core.stdc.errno.EINVAL = 22EINVAL
|| (local variable) int writeReswriteRes == -(constant) int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP || (local variable) int fsyncResfsyncRes == -(constant) int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP)
{
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent to writef(fmt, args, '\n').
writefln("SKIP: linked SQEs (IOSQE_IO_LINK) unsupported on this kernel");
return 0;
}
// If the link had broken (write failed), the kernel would have cancelled
// the fsync with -ECANCELED. Surface that explicitly for clarity.
if ((local variable) int fsyncResfsyncRes == -(constant) int core.stdc.errno.ECANCELED = 125ECANCELED)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("fsync was cancelled (-ECANCELED): the linked write failed (res=%d)", (local variable) int writeReswriteRes);
return 1;
}
if ((local variable) int writeReswriteRes < 0)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("WRITE completed with error: errno %d", -(local variable) int writeReswriteRes);
return 1;
}
if ((local variable) int fsyncResfsyncRes < 0)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("FSYNC completed with error: errno %d", -(local variable) int fsyncResfsyncRes);
return 1;
}
// The write must have transferred the full payload, and the fsync must
// have returned 0 — and, by the IO_LINK contract, the fsync only ran
// because the write succeeded first.
if ((local variable) int writeReswriteRes != cast(int) (local variable) immutable(ubyte[]) payloadpayload.(field) ulong immutable(ubyte[]).lengthlength)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("short write: wrote %d of %d bytes", (local variable) int writeReswriteRes, (local variable) immutable(ubyte[]) payloadpayload.(field) ulong immutable(ubyte[]).lengthlength);
return 1;
}
void std.stdio.writefln!(char, int, ulong, int)(in char[] fmt, int __param_1, ulong __param_2, int __param_3) @safeEquivalent to writef(fmt, args, '\n').
writefln("ok: linked SQEs ordered write-before-fsync — WRITE res=%d (%d bytes), FSYNC res=%d",
(local variable) int writeReswriteRes, (local variable) immutable(ubyte[]) payloadpayload.(field) ulong immutable(ubyte[]).lengthlength, (local variable) int fsyncResfsyncRes);
return 0;
}