#!/usr/bin/env dub
/+ dub.sdl:
name "io_uring_multishot_timeout"
dependency "during" version="~>0.5.0"
platforms "linux"
targetPath "build"
+/
/**
* `io_uring` — multishot timeout (`IORING_TIMEOUT_MULTISHOT`, Linux 6.4).
*
* Before 6.4 a `TIMEOUT` SQE fired exactly once: it posted a single `-ETIME`
* completion and then disarmed. Multishot timeout lets one submitted SQE act as
* a recurring tick — the kernel re-arms it after each expiry and keeps posting a
* fresh CQE every interval, each flagged with `CQEFlags.MORE` to mean "more
* completions for this user_data are still coming". The request stays armed until
* its `count` of fires is reached, or until you explicitly remove it — at which
* point the terminating CQE arrives with `MORE` cleared.
*
* This example arms a ~20 ms multishot timer (`TimeoutFlags.MULTISHOT`,
* `count == 0` => unbounded), collects 3 ticks (each `res == -ETIME`, each with
* `MORE` set), then issues an `ASYNC_CANCEL` keyed by the timer's `user_data` to
* stop the repeats and drains the terminating CQE (`res == -ECANCELED`, `MORE`
* cleared). Total runtime ~60 ms.
*
* Implementation note: `during` 0.5.0's `prepCancel` helper does not compile
* (it assigns a `uint` to a `CancelFlags` field without a cast), so we build the
* `ASYNC_CANCEL` SQE by hand — the kernel matches the request whose `user_data`
* equals the cancel SQE's `addr`. (The legacy `TIMEOUT_REMOVE` op returns
* `-ENOENT` against a multishot timer, so async-cancel is the right tool here.)
*
* Companion to the io_uring chronology:
* see docs/research/async-io/io-uring/timeline.md § "6.4 — Multishot timeout".
*
* Run with: `dub run --single multishot-timeout.d`
*
* Portability: if the running kernel has no `io_uring` (too old, or blocked by a
* seccomp/container policy), or if multishot timeout is unsupported (kernel < 6.4,
* surfaced as a `-EINVAL`/`-EOPNOTSUPP` on the first CQE), the program prints a
* `SKIP:` line and exits 0 so it stays green in CI regardless of the host kernel.
*/
module (module) io_uring_multishot_timeoutio_uring — multishot timeout (IORING_TIMEOUT_MULTISHOT, Linux 6.4).
Before 6.4 a TIMEOUT SQE fired exactly once: it posted a single -ETIME
completion and then disarmed. Multishot timeout lets one submitted SQE act as
a recurring tick — the kernel re-arms it after each expiry and keeps posting a
fresh CQE every interval, each flagged with CQEFlags.MORE to mean "more
completions for this user_data are still coming". The request stays armed until
its count of fires is reached, or until you explicitly remove it — at which
point the terminating CQE arrives with MORE cleared.
This example arms a ~20 ms multishot timer (TimeoutFlags.MULTISHOT,
count == 0 => unbounded), collects 3 ticks (each res == -ETIME, each with
MORE set), then issues an ASYNC_CANCEL keyed by the timer's user_data to
stop the repeats and drains the terminating CQE (res == -ECANCELED, MORE
cleared). Total runtime ~60 ms.
Implementation note: during 0.5.0's prepCancel helper does not compile
(it assigns a uint to a CancelFlags field without a cast), so we build the
ASYNC_CANCEL SQE by hand — the kernel matches the request whose user_data
equals the cancel SQE's addr. (The legacy TIMEOUT_REMOVE op returns
-ENOENT against a multishot timer, so async-cancel is the right tool here.)
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md § "6.4 — Multishot timeout".
Run with: dub run --single multishot-timeout.d
Portability
if the running kernel has no io_uring (too old, or blocked by a
seccomp/container policy), or if multishot timeout is unsupported (kernel < 6.4,
surfaced as a -EINVAL/-EOPNOTSUPP on the first CQE), the program prints a
SKIP: line and exits 0 so it stays green in CI regardless of the host kernel.
io_uring_multishot_timeout;
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_multishot_timeout.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_multishot_timeout.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL, (alias constant) io_uring_multishot_timeout.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP, (alias constant) io_uring_multishot_timeout.ETIME = int core.stdc.errno.ETIME = 62ETIME, (alias constant) io_uring_multishot_timeout.ECANCELED = int core.stdc.errno.ECANCELED = 125ECANCELED;
int int D main()main()
{
// Distinct cookies so we can tell the timer's ticks apart from the
// cancel request's own completion when both are in flight.
enum ulong (constant) ulong io_uring_multishot_timeout.main.timerData = 1LUtimerData = 1;
enum ulong (constant) ulong io_uring_multishot_timeout.main.cancelData = 2LUcancelData = 2;
(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;
}
// A short relative interval; with MULTISHOT and count==0 the kernel re-arms
// this same timer after every expiry, posting one CQE per ~20 ms tick.
(struct) during.io_uring.KernelTimespecTime specification as defined in kernel headers (used by TIMEOUT operations)
KernelTimespec (local variable) during.io_uring.KernelTimespec tsts;
(local variable) during.io_uring.KernelTimespec tsts.(field) long during.io_uring.KernelTimespec.tv_secseconds
tv_sec = 0;
(local variable) during.io_uring.KernelTimespec tsts.(field) long during.io_uring.KernelTimespec.tv_nsecnanoseconds
tv_nsec = 20_000_000; // 20 ms
(local variable) during.Uring ioio.putWith!(
(ref SubmissionEntry e, ref KernelTimespec t)
{
e.prepTimeout(t, /*count*/ 0, TimeoutFlags.MULTISHOT);
e.user_data = timerData;
})(during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, ref during.io_uring.KernelTimespec t) nothrow @nogc @safe
{
prepTimeout(e, t, 0LU, TimeoutFlags.MULTISHOT);
e.user_data = 1LU;
}
, during.io_uring.KernelTimespec)(ref during.io_uring.KernelTimespec __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().
ts);
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(0); // flush the SQ; the timer arms in the kernel
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;
}
enum int (constant) int io_uring_multishot_timeout.main.wantTicks = 3wantTicks = 3;
int (local variable) int seenseen;
bool (local variable) bool lastHadMorelastHadMore = false;
// Collect 3 recurring ticks. We bound the wait by only ever asking for one
// completion at a time and stopping after wantTicks fires.
while ((local variable) int seenseen < (constant) int io_uring_multishot_timeout.main.wantTicks = 3wantTicks)
{
(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(1);
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;
const (local variable) const(int) resres = (local variable) const(during.io_uring.CompletionEntry) cc.(field) int during.io_uring.CompletionEntry.resresult code for this event
res;
const (local variable) const(bool) moremore = ((local variable) const(during.io_uring.CompletionEntry) cc.(field) during.io_uring.CQEFlags during.io_uring.CompletionEntry.flagsflags & (enum) during.io_uring.CQEFlagsFlags used with CompletionEntry
CQEFlags.(enum value) during.io_uring.CQEFlags.MORE = 2uIORING_CQE_F_MORE (from Linux 5.13)
If set, parent SQE will generate more CQE entries
MORE) != 0;
(local variable) during.Uring ioio.void during.Uring.popFront() pure nothrow @nogc @safeMove to next CompletionEntry
popFront();
// First CQE of -EINVAL/-EOPNOTSUPP => kernel predates multishot timeout.
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)
{
void std.stdio.writefln!(char, const(int))(in char[] fmt, const(int) __param_1) @safeEquivalent to writef(fmt, args, '\n').
writefln("SKIP: multishot timeout unsupported on this kernel (errno %d) — needs Linux 6.4+", -(local variable) const(int) resres);
return 0;
}
if ((local variable) const(int) resres != -(constant) int core.stdc.errno.ETIME = 62ETIME)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("unexpected tick result: errno %d (expected -ETIME)", -(local variable) const(int) resres);
return 1;
}
if (!(local variable) const(bool) moremore)
{
// An unbounded multishot timer must keep MORE set while it re-arms.
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("tick %d cleared CQEFlags.MORE while timer should still be armed", (local variable) int seenseen);
return 1;
}
(local variable) int seenseen++;
(local variable) bool lastHadMorelastHadMore = (local variable) const(bool) moremore;
}
// Stop the recurring timer with an ASYNC_CANCEL keyed by its user_data.
// `prepCancel` is broken in during 0.5.0, so fill the SQE by hand: the kernel
// matches the in-flight request whose user_data == this SQE's `addr`.
auto (local variable) during.io_uring.SubmissionEntry* sqesqe = &(local variable) during.Uring ioio.during.io_uring.SubmissionEntry during.Uring.next!()() pure nothrow @nogc ref @safeAdvances the userspace submision queue and returns last SubmissionEntry.
next();
*(local variable) during.io_uring.SubmissionEntry* sqesqe = (struct) during.io_uring.SubmissionEntryIO operation submission data structure (Submission queue entry).
C API: struct io_uring_sqe
SubmissionEntry.(constant) during.io_uring.SubmissionEntry during.io_uring.SubmissionEntry.init = SubmissionEntry(Operation.NOP, SubmissionEntryFlags.NONE, cast(ushort)0u, 0, 0LU, , , , 0LU, , , , 0u, ReadWriteFlags.NONE, , , , , , , , , , , , , , , , , , , , , , , , 0LU, cast(ushort)0u, , cast(ushort)0u, 0, , , , , , , , 0LU, [0LU], , )init;
(local variable) during.io_uring.SubmissionEntry* sqesqe.(field) during.io_uring.Operation during.io_uring.SubmissionEntry.opcodetype of operation for this sqe
opcode = (enum) during.io_uring.OperationDescribes the operation to be performed
Operation.(enum value) during.io_uring.Operation.ASYNC_CANCEL = cast(ubyte)14uIORING_OP_ASYNC_CANCEL
ASYNC_CANCEL;
(local variable) during.io_uring.SubmissionEntry* sqesqe.(field) int during.io_uring.SubmissionEntry.fdfile descriptor to do IO on
fd = -1;
(local variable) during.io_uring.SubmissionEntry* sqesqe.(field) ulong during.io_uring.SubmissionEntry.addrpointer to buffer or iovecs
addr = (constant) ulong io_uring_multishot_timeout.main.timerData = 1LUtimerData; // key: cancel the request with this user_data
(local variable) during.io_uring.SubmissionEntry* sqesqe.(field) ulong during.io_uring.SubmissionEntry.user_datadata to be passed back at completion time
user_data = (constant) ulong io_uring_multishot_timeout.main.cancelData = 2LUcancelData;
const (local variable) const(int) cancelSubmittedcancelSubmitted = (local variable) during.Uring ioio.int during.Uring.submit(uint want) nothrow @nogc @safeSubmits qued SubmissionEntry to be processed by kernel.
submit(0);
if ((local variable) const(int) cancelSubmittedcancelSubmitted < 0)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("submit (cancel) failed: errno %d", -(local variable) const(int) cancelSubmittedcancelSubmitted);
return 1;
}
// Drain until we've observed BOTH the cancel request's own CQE and the timer's
// terminating CQE (res == -ECANCELED, MORE cleared). A tick already in flight
// may slip in before the cancel lands; we just consume it. The loop is bounded
// by iteration count so it can never block indefinitely.
bool (local variable) bool sawCancelsawCancel = false;
bool (local variable) bool sawTimerTerminationsawTimerTermination = false;
foreach ((local variable) int __; 0 .. 8)
{
if ((local variable) bool sawCancelsawCancel && (local variable) bool sawTimerTerminationsawTimerTermination)
break;
(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(1);
// Consume every CQE currently ready, not just one — the cancel and the
// timer termination often arrive in the same batch.
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;
const (local variable) const(int) resres = (local variable) const(during.io_uring.CompletionEntry) cc.(field) int during.io_uring.CompletionEntry.resresult code for this event
res;
const (local variable) const(ulong) datadata = (local variable) const(during.io_uring.CompletionEntry) cc.(field) ulong during.io_uring.CompletionEntry.user_datasqe->data submission passed back
user_data;
const (local variable) const(bool) moremore = ((local variable) const(during.io_uring.CompletionEntry) cc.(field) during.io_uring.CQEFlags during.io_uring.CompletionEntry.flagsflags & (enum) during.io_uring.CQEFlagsFlags used with CompletionEntry
CQEFlags.(enum value) during.io_uring.CQEFlags.MORE = 2uIORING_CQE_F_MORE (from Linux 5.13)
If set, parent SQE will generate more CQE entries
MORE) != 0;
(local variable) during.Uring ioio.void during.Uring.popFront() pure nothrow @nogc @safeMove to next CompletionEntry
popFront();
if ((local variable) const(ulong) datadata == (constant) ulong io_uring_multishot_timeout.main.timerData = 1LUtimerData)
{
// A late tick (-ETIME, MORE set) or the termination
// (-ECANCELED, MORE cleared).
if ((local variable) const(int) resres == -(constant) int core.stdc.errno.ECANCELED = 125ECANCELED || !(local variable) const(bool) moremore)
(local variable) bool sawTimerTerminationsawTimerTermination = true;
}
else if ((local variable) const(ulong) datadata == (constant) ulong io_uring_multishot_timeout.main.cancelData = 2LUcancelData)
{
// 0 == cancelled successfully. A benign race (timer fired as the
// cancel ran) can report -ENOENT/-EALREADY; either way the timer
// is gone, so don't treat those as failures.
(local variable) bool sawCancelsawCancel = true;
}
}
}
if (!(local variable) bool sawCancelsawCancel || !(local variable) bool sawTimerTerminationsawTimerTermination)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("did not observe both the cancel (%s) and the timer termination (%s)",
(local variable) bool sawCancelsawCancel, (local variable) bool sawTimerTerminationsawTimerTermination);
return 1;
}
void std.stdio.writefln!(char, int, bool)(in char[] fmt, int __param_1, bool __param_2) @safeEquivalent to writef(fmt, args, '\n').
writefln("ok: multishot timeout fired %d times (each -ETIME with CQEFlags.MORE), "
~ "then stopped via ASYNC_CANCEL (last fire MORE=%s)", (local variable) int seenseen, (local variable) bool lastHadMorelastHadMore);
return 0;
}