#!/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_cancelio_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) duringSimple idiomatic dlang wrapper around linux io_uring
(see: https://kernel.dk/io_uring.pdf) asynchronous API.
during;
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_async_cancel.ECANCELED = int core.stdc.errno.ECANCELED = 125ECANCELED, (alias constant) io_uring_async_cancel.EALREADY = int core.stdc.errno.EALREADY = 114EALREADY, (alias constant) io_uring_async_cancel.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL, (alias constant) io_uring_async_cancel.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP, (alias constant) io_uring_async_cancel.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.
unistd : (alias) io_uring_async_cancel.pipe = int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe, (alias) io_uring_async_cancel.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose, (alias) io_uring_async_cancel.read = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread;
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_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.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;
}
// 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 probeprobe = (local variable) during.Uring ioio.during.Probe during.Uring.probe() nothrow @nogc @safeProbes supported operations
probe();
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
Operation.(enum value) during.io_uring.Operation.POLL_ADD = cast(ubyte)6uIORING_OP_POLL_ADD
POLL_ADD) || !(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
Operation.(enum value) during.io_uring.Operation.ASYNC_CANCEL = cast(ubyte)14uIORING_OP_ASYNC_CANCEL
ASYNC_CANCEL)))
{
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent 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] fdsfds;
if (int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe((local variable) int[2] fdsfds) != 0)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("pipe() failed");
return 1;
}
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]); }
const (local variable) const(int) readFdreadFd = (local variable) int[2] fdsfds[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 pollKeypollKey;
enum ulong (constant) ulong io_uring_async_cancel.main.cancelCookie = 2LUcancelCookie = 2;
// SQE #1: poll the read end for readability. It will never fire on its own.
(local variable) during.Uring ioio.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 @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().
pollKey, (local variable) const(int) readFdreadFd);
// 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 ioio.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 @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().
pollKey);
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)
{
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;
}
// Both the poll (cancelled) and the cancel op produce a completion.
(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);
bool (local variable) bool sawPollsawPoll, (local variable) bool sawCancelsawCancel;
int (local variable) int pollRespollRes, (local variable) int cancelRescancelRes;
const ulong (local variable) const(ulong) pollUserDatapollUserData = cast(ulong)cast(void*)&(local variable) int pollKeypollKey;
foreach ((local variable) int __; 0 .. 2)
{
if ((local variable) during.Uring ioio.bool during.Uring.empty() const pure nothrow @nogc @safeCheck if there is some CompletionEntry to process.
empty) break;
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 == (local variable) const(ulong) pollUserDatapollUserData)
{
(local variable) bool sawPollsawPoll = true;
(local variable) int pollRespollRes = (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_async_cancel.main.cancelCookie = 2LUcancelCookie)
{
(local variable) bool sawCancelsawCancel = true;
(local variable) int cancelRescancelRes = (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();
}
if (!(local variable) bool sawPollsawPoll || !(local variable) bool sawCancelsawCancel)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("missing completion(s): sawPoll=%s sawCancel=%s", (local variable) bool sawPollsawPoll, (local variable) bool sawCancelsawCancel);
return 1;
}
// The cancelled poll must report -ECANCELED.
if ((local variable) int pollRespollRes != -(constant) int core.stdc.errno.ECANCELED = 125ECANCELED)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("poll CQE: expected -ECANCELED, got res=%d", (local variable) int pollRespollRes);
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 cancelRescancelRes < 0 && (local variable) int cancelRescancelRes != -(constant) int core.stdc.errno.EALREADY = 114EALREADY)
{
// -EINVAL/-EOPNOTSUPP/-ENOSYS here would mean the op isn't really
// supported despite the probe — treat as SKIP, not failure.
if ((local variable) int cancelRescancelRes == -(constant) int core.stdc.errno.EINVAL = 22EINVAL || (local variable) int cancelRescancelRes == -(constant) int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP || (local variable) int cancelRescancelRes == -(constant) int core.stdc.errno.ENOSYS = 38ENOSYS)
{
void std.stdio.writefln!(char, int)(in char[] fmt, int __param_1) @safeEquivalent to writef(fmt, args, '\n').
writefln("SKIP: ASYNC_CANCEL returned errno %d — unsupported on this kernel", -(local variable) int cancelRescancelRes);
return 0;
}
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("cancel CQE: expected res >= 0 or -EALREADY, got res=%d", (local variable) int cancelRescancelRes);
return 1;
}
void std.stdio.writefln!(char, int, int)(in char[] fmt, int __param_1, int __param_2) @safeEquivalent to writef(fmt, args, '\n').
writefln("ok: ASYNC_CANCEL torn down the poll (poll res=%d=-ECANCELED, cancel res=%d)",
(local variable) int pollRespollRes, (local variable) int cancelRescancelRes);
return 0;
}