#!/usr/bin/env dub
/+ dub.sdl:
name "io_uring_splice_tee"
dependency "during" version="~>0.5.0"
platforms "linux"
targetPath "build"
+/
/**
* `io_uring` — zero-copy `IORING_OP_SPLICE` (Linux 5.7) + `IORING_OP_TEE` (5.8).
*
* `splice(2)` moves bytes between two file descriptors *without* a round trip
* through a userspace buffer, provided one end is a pipe — the kernel just moves
* page references between the pipe buffers. `tee(2)` is its sibling: it
* *duplicates* bytes from one pipe to another without consuming the source, so
* the same data can still be read from the original pipe afterwards. 5.7 brought
* `splice` into `io_uring` (`IORING_OP_SPLICE`); 5.8 added `IORING_OP_TEE`.
*
* This program wires up three pipes and runs the two ops back-to-back:
* 1. Write a known payload into pipe A's write end (a plain `write(2)`).
* 2. `SPLICE` A.read -> B.write — zero-copy hand-off, no userspace copy.
* 3. `TEE` B.read -> C.write — duplicate B's bytes into C *without* draining B.
* 4. `read(2)` from C (the tee'd copy) and then from B (the original) and
* verify both still carry the full payload.
*
* The load-bearing detail is the offset convention: for a pipe fd the splice
* offset *must* be `-1` (`off_in`/`off_out`), which `during`'s `prepSplice`
* takes as a `ulong`, so we pass `cast(ulong)-1` (== `ulong.max`). `prepTee` has
* no offsets at all — pipes are inherently offset-less streams.
*
* The two SQEs are submitted as a single batch but ordered with `IO_LINK` so the
* TEE cannot start reading pipe B until the SPLICE that fills it has completed.
*
* Companion to the io_uring chronology:
* see docs/research/async-io/io-uring/timeline.md
* § "5.7 — Splice, provided buffers, fast poll (May 2020)".
*
* Run with: `dub run --single splice-tee.d`
*
* Portability: prints `SKIP:` and exits 0 when io_uring is unavailable or the
* kernel lacks SPLICE/TEE (probe miss, or an op result of -EINVAL/-EOPNOTSUPP/
* -ENOSYS). Exits nonzero only on a genuinely unexpected syscall failure.
*/
module (module) io_uring_splice_teeio_uring — zero-copy IORING_OP_SPLICE (Linux 5.7) + IORING_OP_TEE (5.8).
splice(2) moves bytes between two file descriptors without a round trip
through a userspace buffer, provided one end is a pipe — the kernel just moves
page references between the pipe buffers. tee(2) is its sibling: it
duplicates bytes from one pipe to another without consuming the source, so
the same data can still be read from the original pipe afterwards. 5.7 brought
splice into io_uring (IORING_OP_SPLICE); 5.8 added IORING_OP_TEE.
This program wires up three pipes and runs the two ops back-to-back:
Write a known payload into pipe A's write end (a plain write(2)).
SPLICE A.read -> B.write — zero-copy hand-off, no userspace copy.
TEE B.read -> C.write — duplicate B's bytes into C without draining B.
read(2) from C (the tee'd copy) and then from B (the original) and
verify both still carry the full payload.
The load-bearing detail is the offset convention: for a pipe fd the splice
offset must be -1 (off_in/off_out), which during's prepSplice
takes as a ulong, so we pass cast(ulong)-1 (== ulong.max). prepTee has
no offsets at all — pipes are inherently offset-less streams.
The two SQEs are submitted as a single batch but ordered with IO_LINK so the
TEE cannot start reading pipe B until the SPLICE that fills it has completed.
Companion to the io_uring chronology:
see docs/research/async-io/io-uring/timeline.md
§ "5.7 — Splice, provided buffers, fast poll (May 2020)".
Run with: dub run --single splice-tee.d
Portability
prints SKIP: and exits 0 when io_uring is unavailable or the
kernel lacks SPLICE/TEE (probe miss, or an op result of -EINVAL/-EOPNOTSUPP/
-ENOSYS). Exits nonzero only on a genuinely unexpected syscall failure.
io_uring_splice_tee;
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.stdcstdc.(module) core.stdc.errnoD header file for C99.
pubs.opengroup.org/onlinepubs/009695399/basedefs/errno.h.html, errno.h
Source
core/stdc/errno.d
errno : (alias constant) io_uring_splice_tee.EINVAL = int core.stdc.errno.EINVAL = 22EINVAL, (alias constant) io_uring_splice_tee.ENOSYS = int core.stdc.errno.ENOSYS = 38ENOSYS, (alias constant) io_uring_splice_tee.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP;
import (package) corecore.(package) core.syssys.(package) core.sys.posixposix.(module) core.sys.posix.unistdD header file for POSIX.
unistd : (alias) io_uring_splice_tee.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose, (alias) io_uring_splice_tee.pipe = int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe, (alias) io_uring_splice_tee.read = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread, (alias) io_uring_splice_tee.write = long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite;
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 : stderr, (alias template) io_uring_splice_tee.writefln = std.stdio.writefln(alias fmt, A...)(A args) if (isSomeString!(typeof(fmt)))Equivalent to writef(fmt, args, '\n').
writefln;
// User-data cookies so we can tell the two completions apart regardless of the
// order the kernel posts them.
enum ulong (constant) ulong io_uring_splice_tee.UD_SPLICE = 1LUUD_SPLICE = 1;
enum ulong (constant) ulong io_uring_splice_tee.UD_TEE = 2LUUD_TEE = 2;
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;
}
// Cheap, kernel-version-agnostic capability gate: ask the ring's op probe
// whether SPLICE and TEE are advertised. On a host that predates them this
// short-circuits to a clean SKIP before we touch any pipes.
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.SPLICE = cast(ubyte)30uIORING_OP_SPLICE
SPLICE) && (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.TEE = cast(ubyte)33uIORING_OP_TEE
TEE)))
{
void std.stdio.writefln!char(in char[] fmt) @safeEquivalent to writef(fmt, args, '\n').
writefln("SKIP: kernel io_uring lacks SPLICE/TEE (probe miss) — feature added in 5.7/5.8");
return 0;
}
enum (alias) object.string = stringstring (constant) string io_uring_splice_tee.main.payload = "io_uring zero-copy splice+tee\n"payload = "io_uring zero-copy splice+tee\n";
immutable(ubyte)[] (local variable) immutable(ubyte)[] txtx = cast(immutable(ubyte)[]) (constant) string io_uring_splice_tee.main.payload = "io_uring zero-copy splice+tee\n"payload;
// Three pipes: A is the source, B receives the SPLICE, C receives the TEE.
int[2] (local variable) int[2] aa = [-1, -1], (local variable) int[2] bb = [-1, -1], (local variable) int[2] cc = [-1, -1];
if (int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe((local variable) int[2] aa) != 0 || int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe((local variable) int[2] bb) != 0 || int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe((local variable) int[2] cc) != 0)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("pipe(2) failed");
return 1;
}
scope (exit)
foreach ((parameter) int[2] pp; [(local variable) int[2] aa, (local variable) int[2] bb, (local variable) int[2] cc])
foreach ((parameter) int fdfd; (local variable) int[2] pp)
if ((local variable) int fdfd >= 0)
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose((local variable) int fdfd);
// Seed pipe A with the payload via an ordinary blocking write — small enough
// to fit comfortably in the default 64 KiB pipe buffer, so this never blocks.
const (local variable) const(long) wrotewrote = long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite((local variable) int[2] aa[1], (local variable) immutable(ubyte)[] txtx.(field) immutable(ubyte)* immutable(ubyte)[].ptrptr, (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength);
if ((local variable) const(long) wrotewrote != cast(long) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("seed write to pipe A failed: wrote %d of %d", (local variable) const(long) wrotewrote, (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength);
return 1;
}
// SQE 1: SPLICE A.read -> B.write. Pipe offsets must be -1 (ulong.max here).
// IO_LINK makes the following TEE wait for this to finish (and succeed).
(local variable) during.Uring ioio.putWith!((ref SubmissionEntry e, int fdIn, int fdOut, uint len) {
e.prepSplice(fdIn, cast(ulong)-1, fdOut, cast(ulong)-1, len, 0);
e.flags |= SubmissionEntryFlags.IO_LINK;
e.user_data = UD_SPLICE;
})(during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int fdIn, int fdOut, uint len) nothrow @nogc @safe
{
prepSplice(e, fdIn, 18446744073709551615LU, fdOut, 18446744073709551615LU, len, 0u);
cast(int)e.flags |= 4;
e.user_data = 1LU;
}
, int, int, uint)(ref int __param_0, ref int __param_1, uint __param_2) 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().
a[0], (local variable) int[2] bb[1], cast(uint) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength);
// SQE 2: TEE B.read -> C.write. TEE duplicates without consuming B, so the
// bytes remain readable from B afterwards. No offsets for tee.
(local variable) during.Uring ioio.putWith!((ref SubmissionEntry e, int fdIn, int fdOut, uint len) {
e.prepTee(fdIn, fdOut, len, 0);
e.user_data = UD_TEE;
})(during.Uring during.Uring.putWith!(function (ref during.io_uring.SubmissionEntry e, int fdIn, int fdOut, uint len) nothrow @nogc @safe
{
prepTee(e, fdIn, fdOut, len, 0u);
e.user_data = 2LU;
}
, int, int, uint)(ref int __param_0, ref int __param_1, uint __param_2) 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().
b[0], (local variable) int[2] cc[1], cast(uint) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength);
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;
}
// Collect both completions (bounded — exactly two SQEs were submitted).
int (local variable) int spliceResspliceRes = int.(constant) int int.min = -2147483648min, (local variable) int teeResteeRes = int.(constant) int int.min = -2147483648min;
foreach ((local variable) int __; 0 .. 2)
{
(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) cqecqe = (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) cqecqe.(field) ulong during.io_uring.CompletionEntry.user_datasqe->data submission passed back
user_data == (constant) ulong io_uring_splice_tee.UD_SPLICE = 1LUUD_SPLICE)
(local variable) int spliceResspliceRes = (local variable) const(during.io_uring.CompletionEntry) cqecqe.(field) int during.io_uring.CompletionEntry.resresult code for this event
res;
else if ((local variable) const(during.io_uring.CompletionEntry) cqecqe.(field) ulong during.io_uring.CompletionEntry.user_datasqe->data submission passed back
user_data == (constant) ulong io_uring_splice_tee.UD_TEE = 2LUUD_TEE)
(local variable) int teeResteeRes = (local variable) const(during.io_uring.CompletionEntry) cqecqe.(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 linked op that the kernel skips because its predecessor failed reports
// -ECANCELED; treat an unsupported SPLICE/TEE as a clean SKIP either way.
bool bool io_uring_splice_tee.main.unsupported(int r) pure nothrow @nogc @safeunsupported(int (parameter) int rr)
{
return (parameter) int rr == -(constant) int core.stdc.errno.EINVAL = 22EINVAL || (parameter) int rr == -(constant) int core.stdc.errno.EOPNOTSUPP = 95EOPNOTSUPP || (parameter) int rr == -(constant) int core.stdc.errno.ENOSYS = 38ENOSYS;
}
if (bool io_uring_splice_tee.main.unsupported(int r) pure nothrow @nogc @safeunsupported((local variable) int spliceResspliceRes) || bool io_uring_splice_tee.main.unsupported(int r) pure nothrow @nogc @safeunsupported((local variable) int teeResteeRes))
{
void std.stdio.writefln!(char, int, int)(in char[] fmt, int __param_1, int __param_2) @safeEquivalent to writef(fmt, args, '\n').
writefln("SKIP: SPLICE/TEE rejected by kernel (splice res=%d, tee res=%d)", (local variable) int spliceResspliceRes, (local variable) int teeResteeRes);
return 0;
}
if ((local variable) int spliceResspliceRes != cast(int) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("SPLICE moved %d bytes, expected %d", (local variable) int spliceResspliceRes, (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength);
return 1;
}
if ((local variable) int teeResteeRes != cast(int) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength)
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("TEE duplicated %d bytes, expected %d", (local variable) int teeResteeRes, (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength);
return 1;
}
// The tee'd copy lands in C; the original is still queued in B because TEE
// does not consume. Read both and verify the payload survived both hops.
ubyte[256] (local variable) ubyte[256] rxCrxC, (local variable) ubyte[256] rxBrxB;
const (local variable) const(long) rdCrdC = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread((local variable) int[2] cc[0], (local variable) ubyte[256] rxCrxC.(constant) ubyte* ubyte[256].ptr = &rxCptr, (local variable) ubyte[256] rxCrxC.(constant) ulong ubyte[256].length = 256LUlength);
const (local variable) const(long) rdBrdB = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread((local variable) int[2] bb[0], (local variable) ubyte[256] rxBrxB.(constant) ubyte* ubyte[256].ptr = &rxBptr, (local variable) ubyte[256] rxBrxB.(constant) ulong ubyte[256].length = 256LUlength);
if ((local variable) const(long) rdCrdC != cast(long) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength || (local variable) ubyte[256] rxCrxC[0 .. (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength] != (local variable) immutable(ubyte)[] txtx[])
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("tee'd copy in pipe C mismatch (read %d bytes)", (local variable) const(long) rdCrdC);
return 1;
}
if ((local variable) const(long) rdBrdB != cast(long) (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength || (local variable) ubyte[256] rxBrxB[0 .. (local variable) immutable(ubyte)[] txtx.(field) ulong immutable(ubyte)[].lengthlength] != (local variable) immutable(ubyte)[] txtx[])
{
stderr.std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @systemwritefln("original bytes in pipe B were consumed by TEE (read %d bytes)", (local variable) const(long) rdBrdB);
return 1;
}
void std.stdio.writefln!(char, int)(in char[] fmt, int __param_1) @safeEquivalent to writef(fmt, args, '\n').
writefln("ok: SPLICE moved %d bytes A->B zero-copy, TEE duplicated them B->C "
~ "without consuming (both pipes still held the payload)", (local variable) int spliceResspliceRes);
return 0;
}