async-cancel.dhover×121all
#!/usr/bin/env dub
/+ dub.sdl:
    name "io_uring_async_cancel"
    dependency "during" version="~>0.5.0"
    platforms "linux"
    targetPath "build"
+/
/**
 * `io_uring` — cancel an in-flight request (`IORING_OP_ASYNC_CANCEL`, Linux 5.5).
 *
 * 5.5 gave `io_uring` the ability to cancel a *pending* submission by its
 * `user_data` cookie. This example arms a `POLL_ADD` on the read end of a pipe
 * that nobody ever writes to — so the poll can never complete on its own — then
 * submits an `ASYNC_CANCEL` keyed to that same request. The kernel tears the
 * poll down and reports two completions:
 *
 *   - the poll CQE completes with `res == -ECANCELED` (it was cancelled), and
 *   - the cancel CQE completes with `res >= 0` — historically `0`, but newer
 *     kernels report the count of requests found & cancelled — or `-EALREADY`
 *     if the kernel had already started completing it.
 *
 * `during`'s `prepPollAdd`/`prepCancel` key off the *address* of a stable
 * variable: `e.setUserData(key)` stores `&key` in the SQE's `user_data`, and
 * `e.prepCancel(key)` puts that same `&key` into the cancel's match field — so
 * the two refer to the same in-flight request. The cancel SQE carries its own
 * distinct `user_data` cookie so we can tell the two CQEs apart.
 *
 * Companion to the io_uring chronology:
 * see docs/research/async-io/io-uring/timeline.md
 *   § "5.5 — Accept/connect, cancel, link-timeout (January 2020)".
 *
 * Run with: `dub run --single async-cancel.d`
 *
 * Portability: prints a `SKIP:` line and exits 0 if `io_uring` is unavailable
 * (old kernel / sandbox) or if `POLL_ADD`/`ASYNC_CANCEL` are not supported on
 * the running kernel (both predate the probe, but we guard defensively), so it
 * stays green in CI regardless of host kernel.
 */
module 
(module) io_uring_async_cancel

io_uring — cancel an in-flight request (IORING_OP_ASYNC_CANCEL, Linux 5.5).

5.5 gave io_uring the ability to cancel a pending submission by its user_data cookie. This example arms a POLL_ADD on the read end of a pipe that nobody ever writes to — so the poll can never complete on its own — then submits an ASYNC_CANCEL keyed to that same request. The kernel tears the poll down and reports two completions:

  • the poll CQE completes with res == -ECANCELED (it was cancelled), and

  • the cancel CQE completes with res >= 0 — historically 0, but newer kernels report the count of requests found & cancelled — or -EALREADY if the kernel had already started completing it.

during's prepPollAdd/prepCancel key off the address of a stable variable: e.setUserData(key) stores &key in the SQE's user_data, and e.prepCancel(key) puts that same &key into the cancel's match field — so the two refer to the same in-flight request. The cancel SQE carries its own distinct user_data cookie so we can tell the two CQEs apart.

Companion to the io_uring chronology: see docs/research/async-io/io-uring/timeline.md § "5.5 — Accept/connect, cancel, link-timeout (January 2020)".

Run with: dub run --single async-cancel.d

Portability

prints a SKIP: line and exits 0 if io_uring is unavailable (old kernel / sandbox) or if POLL_ADD/ASYNC_CANCEL are not supported on the running kernel (both predate the probe, but we guard defensively), so it stays green in CI regardless of host kernel.

io_uring_async_cancel
;
import
(module) during

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

during
;
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_async_cancel.ECANCELED = int core.stdc.errno.ECANCELED = 125
ECANCELED
,
(alias constant) io_uring_async_cancel.EALREADY = int core.stdc.errno.EALREADY = 114
EALREADY
,
(alias constant) io_uring_async_cancel.EINVAL = int core.stdc.errno.EINVAL = 22
EINVAL
,
(alias constant) io_uring_async_cancel.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95
EOPNOTSUPP
,
(alias constant) io_uring_async_cancel.ENOSYS = int core.stdc.errno.ENOSYS = 38
ENOSYS
;
import
(package) core
core
.
(package) core.sys
sys
.
(package) core.sys.posix
posix
.
(module) core.sys.posix.unistd

D 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 Edition
unistd
:
(alias) io_uring_async_cancel.pipe = int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trusted
pipe
,
(alias) io_uring_async_cancel.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trusted
close
,
(alias) io_uring_async_cancel.read = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogc
read
;
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_async_cancel.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
()
{
(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; } // Defensive capability check: POLL_ADD (5.1) and ASYNC_CANCEL (5.5) both // predate the operation probe, but if the kernel reports them unsupported we // skip rather than fail. auto
(local variable) during.Probe probe
probe
=
(local variable) during.Uring io
io
.
during.Probe during.Uring.probe() nothrow @nogc @safe

Probes supported operations

probe
();
if (cast(bool)
(local variable) during.Probe probe
probe
&& (!
(local variable) during.Probe probe
probe
.
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.POLL_ADD = cast(ubyte)6u

IORING_OP_POLL_ADD

POLL_ADD
) || !
(local variable) during.Probe probe
probe
.
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.ASYNC_CANCEL = cast(ubyte)14u

IORING_OP_ASYNC_CANCEL

ASYNC_CANCEL
)))
{
void std.stdio.writefln!char(in char[] fmt) @safe

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

writefln
("SKIP: POLL_ADD/ASYNC_CANCEL not supported by this kernel's io_uring");
return 0; } // A pipe whose read end never becomes readable (we never write to the write // end): the perfect target for a poll that we intend to cancel. int[2]
(local variable) int[2] fds
fds
;
if (
int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trusted
pipe
(
(local variable) int[2] fds
fds
) != 0)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("pipe() failed");
return 1; } scope (exit) {
int core.sys.posix.unistd.close(int) nothrow @nogc @trusted
close
(
(local variable) int[2] fds
fds
[0]);
int core.sys.posix.unistd.close(int) nothrow @nogc @trusted
close
(
(local variable) int[2] fds
fds
[1]); }
const
(local variable) const(int) readFd
readFd
=
(local variable) int[2] fds
fds
[0];
// `pollKey`'s address is the request's user_data; `prepCancel` keys off the // very same address, so it matches this poll. Must outlive the operation. int
(local variable) int pollKey
pollKey
;
enum ulong
(constant) ulong io_uring_async_cancel.main.cancelCookie = 2LU
cancelCookie
= 2;
// SQE #1: poll the read end for readability. It will never fire on its own.
(local variable) during.Uring io
io
.putWith!((ref SubmissionEntry e, ref int key, int fd) {
e.prepPollAdd(fd, PollEvents.IN); e.setUserData(key); // user_data := &key })(
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, ref int key, int fd) nothrow @nogc @safe { prepPollAdd(e, fd, PollEvents.IN, PollFlags.NONE); setUserData(e, key); } , int, const(int))(ref int __param_0, ref const(int) __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.
pollKey
,
(local variable) const(int) readFd
readFd
);
// SQE #2: cancel the request identified by &pollKey. // // We hand-roll what `during`'s `prepCancel` does internally — set the // ASYNC_CANCEL opcode with `addr` pointing at the match key — because // `prepCancel`'s default-flag path is mis-typed in 0.5.0 (it assigns a bare // `uint` to the `CancelFlags`-typed union field and fails to compile). The // match field (`addr`) must equal the poll's `user_data`, i.e. `&pollKey`.
(local variable) during.Uring io
io
.putWith!((ref SubmissionEntry e, ref int key) {
e.prepRW(Operation.ASYNC_CANCEL, -1, cast(void*)&key); e.cancel_flags = CancelFlags.init; // no CANCEL_ALL/FD/etc — key off user_data e.user_data = cancelCookie; })(
during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, ref int key) nothrow @nogc @system { prepRW(e, Operation.ASYNC_CANCEL, -1, cast(void*)&key, 0u, 0LU); e.cancel_flags = CancelFlags.CANCEL_ALL; e.user_data = 2LU; } , int)(ref int __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.
pollKey
);
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
(2);
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; } // Both the poll (cancelled) and the cancel op produce a completion.
(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
(2);
bool
(local variable) bool sawPoll
sawPoll
,
(local variable) bool sawCancel
sawCancel
;
int
(local variable) int pollRes
pollRes
,
(local variable) int cancelRes
cancelRes
;
const ulong
(local variable) const(ulong) pollUserData
pollUserData
= cast(ulong)cast(void*)&
(local variable) int pollKey
pollKey
;
foreach (
(local variable) int _
_
; 0 .. 2)
{ if (
(local variable) during.Uring io
io
.
bool during.Uring.empty() const pure nothrow @nogc @safe

Check if there is some CompletionEntry to process.

empty
) break;
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
;
if (
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) ulong during.io_uring.CompletionEntry.user_data

sqe->data submission passed back

user_data
==
(local variable) const(ulong) pollUserData
pollUserData
)
{
(local variable) bool sawPoll
sawPoll
= true;
(local variable) int pollRes
pollRes
=
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) int during.io_uring.CompletionEntry.res

result code for this event

res
;
} else if (
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) ulong during.io_uring.CompletionEntry.user_data

sqe->data submission passed back

user_data
==
(constant) ulong io_uring_async_cancel.main.cancelCookie = 2LU
cancelCookie
)
{
(local variable) bool sawCancel
sawCancel
= true;
(local variable) int cancelRes
cancelRes
=
(local variable) const(during.io_uring.CompletionEntry) c
c
.
(field) int during.io_uring.CompletionEntry.res

result code for this event

res
;
}
(local variable) during.Uring io
io
.
void during.Uring.popFront() pure nothrow @nogc @safe

Move to next CompletionEntry

popFront
();
} if (!
(local variable) bool sawPoll
sawPoll
|| !
(local variable) bool sawCancel
sawCancel
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("missing completion(s): sawPoll=%s sawCancel=%s",
(local variable) bool sawPoll
sawPoll
,
(local variable) bool sawCancel
sawCancel
);
return 1; } // The cancelled poll must report -ECANCELED. if (
(local variable) int pollRes
pollRes
!= -
(constant) int core.stdc.errno.ECANCELED = 125
ECANCELED
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("poll CQE: expected -ECANCELED, got res=%d",
(local variable) int pollRes
pollRes
);
return 1; } // The cancel op succeeds with res >= 0 — historically 0, but newer kernels // report the *count* of requests found and cancelled (here 1). -EALREADY // means the request had already started completing (also a success: the // poll still ends up -ECANCELED). if (
(local variable) int cancelRes
cancelRes
< 0 &&
(local variable) int cancelRes
cancelRes
!= -
(constant) int core.stdc.errno.EALREADY = 114
EALREADY
)
{ // -EINVAL/-EOPNOTSUPP/-ENOSYS here would mean the op isn't really // supported despite the probe — treat as SKIP, not failure. if (
(local variable) int cancelRes
cancelRes
== -
(constant) int core.stdc.errno.EINVAL = 22
EINVAL
||
(local variable) int cancelRes
cancelRes
== -
(constant) int core.stdc.errno.EOPNOTSUPP = 95
EOPNOTSUPP
||
(local variable) int cancelRes
cancelRes
== -
(constant) int core.stdc.errno.ENOSYS = 38
ENOSYS
)
{
void std.stdio.writefln!(char, int)(in char[] fmt, int __param_1) @safe

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

writefln
("SKIP: ASYNC_CANCEL returned errno %d — unsupported on this kernel", -
(local variable) int cancelRes
cancelRes
);
return 0; } stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("cancel CQE: expected res >= 0 or -EALREADY, got res=%d",
(local variable) int cancelRes
cancelRes
);
return 1; }
void std.stdio.writefln!(char, int, int)(in char[] fmt, int __param_1, int __param_2) @safe

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

writefln
("ok: ASYNC_CANCEL torn down the poll (poll res=%d=-ECANCELED, cancel res=%d)",
(local variable) int pollRes
pollRes
,
(local variable) int cancelRes
cancelRes
);
return 0; }