splice-tee.dhover×185all
#!/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_tee

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.

io_uring_splice_tee
;
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.stdc
stdc
.
(module) core.stdc.errno

D header file for C99.

pubs.opengroup.org/onlinepubs/009695399/basedefs/errno.h.html, errno.h

Source

core/stdc/errno.d

@copyrightCopyright Sean Kelly 2005 - 2009.@licenseDistributed under the Boost Software License 1.0. (See accompanying file LICENSE)@authorsSean Kelly, Alex Rønne Petersen@standardsISO/IEC 9899:1999 (E)
errno
:
(alias constant) io_uring_splice_tee.EINVAL = int core.stdc.errno.EINVAL = 22
EINVAL
,
(alias constant) io_uring_splice_tee.ENOSYS = int core.stdc.errno.ENOSYS = 38
ENOSYS
,
(alias constant) io_uring_splice_tee.EOPNOTSUPP = int core.stdc.errno.EOPNOTSUPP = 95
EOPNOTSUPP
;
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_splice_tee.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trusted
close
,
(alias) io_uring_splice_tee.pipe = int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trusted
pipe
,
(alias) io_uring_splice_tee.read = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogc
read
,
(alias) io_uring_splice_tee.write = long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogc
write
;
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
: 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 = 1LU
UD_SPLICE
= 1;
enum ulong
(constant) ulong io_uring_splice_tee.UD_TEE = 2LU
UD_TEE
= 2;
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; } // 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 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.SPLICE = cast(ubyte)30u

IORING_OP_SPLICE

SPLICE
) &&
(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.TEE = cast(ubyte)33u

IORING_OP_TEE

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

Equivalent 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 = string
string
(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)[] tx
tx
= 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] a
a
= [-1, -1],
(local variable) int[2] b
b
= [-1, -1],
(local variable) int[2] c
c
= [-1, -1];
if (
int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trusted
pipe
(
(local variable) int[2] a
a
) != 0 ||
int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trusted
pipe
(
(local variable) int[2] b
b
) != 0 ||
int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trusted
pipe
(
(local variable) int[2] c
c
) != 0)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("pipe(2) failed");
return 1; } scope (exit) foreach (
(parameter) int[2] p
p
; [
(local variable) int[2] a
a
,
(local variable) int[2] b
b
,
(local variable) int[2] c
c
])
foreach (
(parameter) int fd
fd
;
(local variable) int[2] p
p
)
if (
(local variable) int fd
fd
>= 0)
int core.sys.posix.unistd.close(int) nothrow @nogc @trusted
close
(
(local variable) int fd
fd
);
// 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) wrote
wrote
=
long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogc
write
(
(local variable) int[2] a
a
[1],
(local variable) immutable(ubyte)[] tx
tx
.
(field) immutable(ubyte)* immutable(ubyte)[].ptr
ptr
,
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
);
if (
(local variable) const(long) wrote
wrote
!= cast(long)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("seed write to pipe A failed: wrote %d of %d",
(local variable) const(long) wrote
wrote
,
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
);
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 io
io
.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 @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.
a
[0],
(local variable) int[2] b
b
[1], cast(uint)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
);
// 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 io
io
.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 @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.
b
[0],
(local variable) int[2] c
c
[1], cast(uint)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
);
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; } // Collect both completions (bounded — exactly two SQEs were submitted). int
(local variable) int spliceRes
spliceRes
= int.
(constant) int int.min = -2147483648
min
,
(local variable) int teeRes
teeRes
= int.
(constant) int int.min = -2147483648
min
;
foreach (
(local variable) int _
_
; 0 .. 2)
{
(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) cqe
cqe
=
(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) cqe
cqe
.
(field) ulong during.io_uring.CompletionEntry.user_data

sqe->data submission passed back

user_data
==
(constant) ulong io_uring_splice_tee.UD_SPLICE = 1LU
UD_SPLICE
)
(local variable) int spliceRes
spliceRes
=
(local variable) const(during.io_uring.CompletionEntry) cqe
cqe
.
(field) int during.io_uring.CompletionEntry.res

result code for this event

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

sqe->data submission passed back

user_data
==
(constant) ulong io_uring_splice_tee.UD_TEE = 2LU
UD_TEE
)
(local variable) int teeRes
teeRes
=
(local variable) const(during.io_uring.CompletionEntry) cqe
cqe
.
(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
();
} // 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 @safe
unsupported
(int
(parameter) int r
r
)
{ return
(parameter) int r
r
== -
(constant) int core.stdc.errno.EINVAL = 22
EINVAL
||
(parameter) int r
r
== -
(constant) int core.stdc.errno.EOPNOTSUPP = 95
EOPNOTSUPP
||
(parameter) int r
r
== -
(constant) int core.stdc.errno.ENOSYS = 38
ENOSYS
;
} if (
bool io_uring_splice_tee.main.unsupported(int r) pure nothrow @nogc @safe
unsupported
(
(local variable) int spliceRes
spliceRes
) ||
bool io_uring_splice_tee.main.unsupported(int r) pure nothrow @nogc @safe
unsupported
(
(local variable) int teeRes
teeRes
))
{
void std.stdio.writefln!(char, int, int)(in char[] fmt, int __param_1, int __param_2) @safe

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

writefln
("SKIP: SPLICE/TEE rejected by kernel (splice res=%d, tee res=%d)",
(local variable) int spliceRes
spliceRes
,
(local variable) int teeRes
teeRes
);
return 0; } if (
(local variable) int spliceRes
spliceRes
!= cast(int)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("SPLICE moved %d bytes, expected %d",
(local variable) int spliceRes
spliceRes
,
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
);
return 1; } if (
(local variable) int teeRes
teeRes
!= cast(int)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
)
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("TEE duplicated %d bytes, expected %d",
(local variable) int teeRes
teeRes
,
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
);
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] rxC
rxC
,
(local variable) ubyte[256] rxB
rxB
;
const
(local variable) const(long) rdC
rdC
=
long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogc
read
(
(local variable) int[2] c
c
[0],
(local variable) ubyte[256] rxC
rxC
.
(constant) ubyte* ubyte[256].ptr = &rxC
ptr
,
(local variable) ubyte[256] rxC
rxC
.
(constant) ulong ubyte[256].length = 256LU
length
);
const
(local variable) const(long) rdB
rdB
=
long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogc
read
(
(local variable) int[2] b
b
[0],
(local variable) ubyte[256] rxB
rxB
.
(constant) ubyte* ubyte[256].ptr = &rxB
ptr
,
(local variable) ubyte[256] rxB
rxB
.
(constant) ulong ubyte[256].length = 256LU
length
);
if (
(local variable) const(long) rdC
rdC
!= cast(long)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
||
(local variable) ubyte[256] rxC
rxC
[0 ..
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
] !=
(local variable) immutable(ubyte)[] tx
tx
[])
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("tee'd copy in pipe C mismatch (read %d bytes)",
(local variable) const(long) rdC
rdC
);
return 1; } if (
(local variable) const(long) rdB
rdB
!= cast(long)
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
||
(local variable) ubyte[256] rxB
rxB
[0 ..
(local variable) immutable(ubyte)[] tx
tx
.
(field) ulong immutable(ubyte)[].length
length
] !=
(local variable) immutable(ubyte)[] tx
tx
[])
{ stderr.
std.stdio.File std.stdio.makeGlobal!"core.stdc.stdio.stderr"() nothrow @nogc @property ref @system
writefln
("original bytes in pipe B were consumed by TEE (read %d bytes)",
(local variable) const(long) rdB
rdB
);
return 1; }
void std.stdio.writefln!(char, int)(in char[] fmt, int __param_1) @safe

Equivalent 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 spliceRes
spliceRes
);
return 0; }