multishot-timeout.dhover×140all
#!/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_timeout

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.

io_uring_multishot_timeout
;
import
(module) during

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

during
;
import
(package) std
std
.
(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_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) core
core
.
(package) core.sys
sys
.
(package) core.sys.linux
linux
.
(module) core.sys.linux.errno

D header file for GNU/Linux

glibc stdlib/errno.h

errno
:
(alias constant) io_uring_multishot_timeout.EINVAL = int core.stdc.errno.EINVAL = 22
EINVAL
,
(alias constant) io_uring_multishot_timeout.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95
EOPNOTSUPP
,
(alias constant) io_uring_multishot_timeout.ETIME = int core.stdc.errno.ETIME = 62
ETIME
,
(alias constant) io_uring_multishot_timeout.ECANCELED = int core.stdc.errno.ECANCELED = 125
ECANCELED
;
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 = 1LU
timerData
= 1;
enum ulong
(constant) ulong io_uring_multishot_timeout.main.cancelData = 2LU
cancelData
= 2;
(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 io
io
;
const
(local variable) const(int) setupRet
setupRet
=
(local variable) during.Uring io
io
.
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) setupRet
setupRet
< 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) setupRet
setupRet
);
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.KernelTimespec

Time specification as defined in kernel headers (used by TIMEOUT operations)

KernelTimespec
(local variable) during.io_uring.KernelTimespec ts
ts
;
(local variable) during.io_uring.KernelTimespec ts
ts
.
(field) long during.io_uring.KernelTimespec.tv_sec

seconds

tv_sec
= 0;
(local variable) during.io_uring.KernelTimespec ts
ts
.
(field) long during.io_uring.KernelTimespec.tv_nsec

nanoseconds

tv_nsec
= 20_000_000; // 20 ms
(local variable) during.Uring io
io
.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 @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.
ts
);
const
(local variable) const(int) submitted
submitted
=
(local variable) during.Uring io
io
.
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
(0); // flush the SQ; the timer arms in the kernel
if (
(local variable) const(int) submitted
submitted
< 0)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("submit failed: errno %d", -
(local variable) const(int) submitted
submitted
);
return 1; } enum int
(constant) int io_uring_multishot_timeout.main.wantTicks = 3
wantTicks
= 3;
int
(local variable) int seen
seen
;
bool
(local variable) bool lastHadMore
lastHadMore
= 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 seen
seen
<
(constant) int io_uring_multishot_timeout.main.wantTicks = 3
wantTicks
)
{
(local variable) during.Uring io
io
.
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(during.io_uring.CompletionEntry) c
c
=
(local variable) during.Uring io
io
.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safe

Get first CompletionEntry from cq ring

front
;
const
(local variable) const(int) res
res
=
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) int during.io_uring.CompletionEntry.res

result code for this event

res
;
const
(local variable) const(bool) more
more
= (
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) during.io_uring.CQEFlags during.io_uring.CompletionEntry.flags
flags
&
(enum) during.io_uring.CQEFlags

Flags used with CompletionEntry

CQEFlags
.
(enum value) during.io_uring.CQEFlags.MORE = 2u

IORING_CQE_F_MORE (from Linux 5.13) If set, parent SQE will generate more CQE entries

MORE
) != 0;
(local variable) during.Uring io
io
.
void during.Uring.popFront() pure nothrow @nogc @safe

Move to next CompletionEntry

popFront
();
// First CQE of -EINVAL/-EOPNOTSUPP => kernel predates multishot timeout. if (
(local variable) const(int) res
res
== -
(constant) int core.stdc.errno.EINVAL = 22
EINVAL
||
(local variable) const(int) res
res
== -
(constant) int core.stdc.errno.EOPNOTSUPP = 95
EOPNOTSUPP
)
{
void std.stdio.writefln!(char, const(int))(in char[] fmt, const(int) __param_1) @safe

Equivalent to writef(fmt, args, '\n').

writefln
("SKIP: multishot timeout unsupported on this kernel (errno %d) — needs Linux 6.4+", -
(local variable) const(int) res
res
);
return 0; } if (
(local variable) const(int) res
res
!= -
(constant) int core.stdc.errno.ETIME = 62
ETIME
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("unexpected tick result: errno %d (expected -ETIME)", -
(local variable) const(int) res
res
);
return 1; } if (!
(local variable) const(bool) more
more
)
{ // 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 @system
writefln
("tick %d cleared CQEFlags.MORE while timer should still be armed",
(local variable) int seen
seen
);
return 1; }
(local variable) int seen
seen
++;
(local variable) bool lastHadMore
lastHadMore
=
(local variable) const(bool) more
more
;
} // 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* sqe
sqe
= &
(local variable) during.Uring io
io
.
during.io_uring.SubmissionEntry during.Uring.next!()() pure nothrow @nogc ref @safe

Advances the userspace submision queue and returns last SubmissionEntry.

next
();
*
(local variable) during.io_uring.SubmissionEntry* sqe
sqe
=
(struct) during.io_uring.SubmissionEntry

IO 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* sqe
sqe
.
(field) during.io_uring.Operation during.io_uring.SubmissionEntry.opcode

type of operation for this sqe

opcode
=
(enum) during.io_uring.Operation

Describes the operation to be performed

@seeio_uring_enter(2)
Operation
.
(enum value) during.io_uring.Operation.ASYNC_CANCEL = cast(ubyte)14u

IORING_OP_ASYNC_CANCEL

ASYNC_CANCEL
;
(local variable) during.io_uring.SubmissionEntry* sqe
sqe
.
(field) int during.io_uring.SubmissionEntry.fd

file descriptor to do IO on

fd
= -1;
(local variable) during.io_uring.SubmissionEntry* sqe
sqe
.
(field) ulong during.io_uring.SubmissionEntry.addr

pointer to buffer or iovecs

addr
=
(constant) ulong io_uring_multishot_timeout.main.timerData = 1LU
timerData
; // key: cancel the request with this user_data
(local variable) during.io_uring.SubmissionEntry* sqe
sqe
.
(field) ulong during.io_uring.SubmissionEntry.user_data

data to be passed back at completion time

user_data
=
(constant) ulong io_uring_multishot_timeout.main.cancelData = 2LU
cancelData
;
const
(local variable) const(int) cancelSubmitted
cancelSubmitted
=
(local variable) during.Uring io
io
.
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
(0);
if (
(local variable) const(int) cancelSubmitted
cancelSubmitted
< 0)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("submit (cancel) failed: errno %d", -
(local variable) const(int) cancelSubmitted
cancelSubmitted
);
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 sawCancel
sawCancel
= false;
bool
(local variable) bool sawTimerTermination
sawTimerTermination
= false;
foreach (
(local variable) int _
_
; 0 .. 8)
{ if (
(local variable) bool sawCancel
sawCancel
&&
(local variable) bool sawTimerTermination
sawTimerTermination
)
break;
(local variable) during.Uring io
io
.
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);
// 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 io
io
.
bool during.Uring.empty() const pure nothrow @nogc @safe

Check if there is some CompletionEntry to process.

empty
)
{ const
(local variable) const(during.io_uring.CompletionEntry) c
c
=
(local variable) during.Uring io
io
.
during.io_uring.CompletionEntry during.Uring.front() pure nothrow @nogc return ref @safe

Get first CompletionEntry from cq ring

front
;
const
(local variable) const(int) res
res
=
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) int during.io_uring.CompletionEntry.res

result code for this event

res
;
const
(local variable) const(ulong) data
data
=
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) ulong during.io_uring.CompletionEntry.user_data

sqe->data submission passed back

user_data
;
const
(local variable) const(bool) more
more
= (
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) during.io_uring.CQEFlags during.io_uring.CompletionEntry.flags
flags
&
(enum) during.io_uring.CQEFlags

Flags used with CompletionEntry

CQEFlags
.
(enum value) during.io_uring.CQEFlags.MORE = 2u

IORING_CQE_F_MORE (from Linux 5.13) If set, parent SQE will generate more CQE entries

MORE
) != 0;
(local variable) during.Uring io
io
.
void during.Uring.popFront() pure nothrow @nogc @safe

Move to next CompletionEntry

popFront
();
if (
(local variable) const(ulong) data
data
==
(constant) ulong io_uring_multishot_timeout.main.timerData = 1LU
timerData
)
{ // A late tick (-ETIME, MORE set) or the termination // (-ECANCELED, MORE cleared). if (
(local variable) const(int) res
res
== -
(constant) int core.stdc.errno.ECANCELED = 125
ECANCELED
|| !
(local variable) const(bool) more
more
)
(local variable) bool sawTimerTermination
sawTimerTermination
= true;
} else if (
(local variable) const(ulong) data
data
==
(constant) ulong io_uring_multishot_timeout.main.cancelData = 2LU
cancelData
)
{ // 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 sawCancel
sawCancel
= true;
} } } if (!
(local variable) bool sawCancel
sawCancel
|| !
(local variable) bool sawTimerTermination
sawTimerTermination
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("did not observe both the cancel (%s) and the timer termination (%s)",
(local variable) bool sawCancel
sawCancel
,
(local variable) bool sawTimerTermination
sawTimerTermination
);
return 1; }
void std.stdio.writefln!(char, int, bool)(in char[] fmt, int __param_1, bool __param_2) @safe

Equivalent 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 seen
seen
,
(local variable) bool lastHadMore
lastHadMore
);
return 0; }