Coming from EVE
EVE separates OS backends, its callback loop, asynchronous I/O, and optional runtime facilities such as fibers and futures. The useful migration question is where execution and resource ownership move—not whether one library has “real async” and the other does not.
This guide uses the EVE source baseline at c72f75135de636987e91bcd8ec5a22d32d34a197. Its README lists Linux io_uring/epoll/poll and Windows IOCP/WSAPoll; it marks macOS/BSD kqueue support as planned. EVE is not a readiness-only library with working kqueue support.
NOTE
Each source tab imports a complete single-file DUB program. Run it with dub run --single <file> -b checked; assertions stay enabled. Each labelled output belongs to the immediately preceding implementation and is independently checked by ci --verify. See Running the examples for prerequisites, dependency versions and the validation command.
Map the execution model
| In EVE | In Event Horizon | Migration decision |
|---|---|---|
| Select a backend and runtime layer | Start LoopGroup, receive RootScope and Env | Keep the owning loop explicit; the default topology is single-threaded. |
| Register timer callbacks | Call env.clock.sleep in a fiber | Sequential code resumes after the wait; check its result. |
| Launch runtime fibers or wait on futures | Fork scoped children and join JoinHandle!T | A scope joins its children before returning. |
| Coordinate producers and consumers | Use Channel!(T, capacity) | These channels are scheduler-local, not cross-thread queues. |
| Cancel an operation or task | Use a deadline or cancel the owning scope | Cancellation is cooperative; protect only cleanup that must finish. |
The baseline's future implementation supports several callback-delivery policies. Do not replace a future wait with a thread-blocking wait inside an Event Horizon fiber: that would stall the loop's other fibers too.
The lifecycle is common; the mechanism is not. Callback registrations need an explicit owner and terminal path. Fibers make sequential control flow possible, but still need cancellation and lifetime rules. A lexical scope is useful when it actually owns the work being started.
Timers and cadence
The contract is three timer deliveries in order, followed by termination of the loop or task. The assertions check the count; they do not claim exact timing. The event-horizon program uses relative sleeps, not Ticker: work between sleeps shifts the next deadline. For absolute cadence, maintain an absolute next deadline and define how missed ticks are skipped. Cancelling a timer registration and interrupting the fiber waiting on it are different operations.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_timers"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import eve;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
int ticks = 0;
// Register a repeating timer: initial delay 10ms, interval 10ms
loop.registerTimer(10, 10, (ref EventLoop l, Token t) @safe nothrow {
ticks++;
try writeln("tick ", ticks); catch (Exception error) { assert(false, error.msg); }
if (ticks >= 3)
l.stop();
});
loop.run();
assert(ticks == 3);
}tick 1
tick 2
tick 3#!/usr/bin/env dub
/+ dub.sdl:
name "eh_timers"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
int ticks;
auto run = group.run((ref RootScope sc, ref Env env) {
foreach (i; 1 .. 4)
{
assert(!env.clock.sleep(10.msecs).hasError); // parks this fiber only
writeln("tick ", i);
++ticks;
}
return 0;
});
assert(run.hasValue);
assert(ticks == 3);
}tick 1
tick 2
tick 3Concurrency and joined lifetime
Start independent work, combine the two results as 1 * 10 + 2, and do not return until the work is complete. Event Horizon's ready/release handshake proves both children have started before either can finish. This is concurrency, not proof of CPU parallelism.
On the foreign side, keep callback state or task handles alive through completion. A lexical D block alone does not join asynchronous work. On the event-horizon side, the nested scope owns that lifetime and the join handles carry outcomes. Decide separately what failure of one child should do to its siblings; successful joining alone does not establish fail-fast cancellation semantics.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_concurrency"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import eve;
import eve.rt;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
auto pSlow = promise!int();
auto pFast = promise!int();
// Layer 1 callbacks are nothrow; Promise.resolve uses a Mutex and can throw
loop.registerTimer(30, 0, (ref EventLoop l, Token t) @safe nothrow {
try pSlow.resolve(1); catch (Exception error) { assert(false, error.msg); }
});
loop.registerTimer(5, 0, (ref EventLoop l, Token t) @safe nothrow {
try pFast.resolve(2); catch (Exception error) { assert(false, error.msg); }
});
auto joined = all([pSlow, pFast]);
int combined = 0;
joined.then((int[] vals) @safe {
combined = vals[0] * 10 + vals[1];
try writeln("joined: ", combined); catch (Exception error) { assert(false, error.msg); }
loop.stop();
});
loop.run();
assert(combined == 12);
}joined: 12#!/usr/bin/env dub
/+ dub.sdl:
name "eh_concurrency"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope root, ref Env env) {
ref Sched s = currentScheduler();
Channel!(int, 2) ready, release;
int finished;
auto outcome = withScope!((ref sc) {
JoinHandle!int slow, fast;
sc.fork(slow, () {
assert(!ready.put(s, 1).hasError);
assert(!release.take(s).hasError);
++finished;
return ioOk(1); // an IoResult: the typed success or IoError
});
sc.fork(fast, () {
assert(!ready.put(s, 2).hasError);
assert(!release.take(s).hasError);
++finished;
return ioOk(2);
});
// Both children have started, but neither can finish yet.
assert(!ready.take(s).hasError && !ready.take(s).hasError);
assert(finished == 0);
assert(!release.put(s, 0).hasError && !release.put(s, 0).hasError);
// Release both and join in either order; no scheduling-time assumption.
auto a = slow.join(s);
auto b = fast.join(s);
assert(a.hasValue && b.hasValue);
return a.value * 10 + b.value;
})(s);
assert(outcome.hasValue && outcome.value == 12);
assert(finished == 2);
writeln("joined: ", outcome.value);
return 0;
});
assert(run.hasValue);
}joined: 12Cancellation, deadlines and cleanup
The examples distinguish timeout detection from successful completion and assert that cleanup ran. Read the foreign operation's interruption mechanism carefully: cancelling a registration, signalling a flag and interrupting a fiber are not interchangeable. A timeout on a wait does not necessarily stop the operation being waited for.
Event Horizon's deadline latches cooperative cancellation. The parked sleep returns ECANCELED; protect lets cleanup cross checkpoints even with that latch set. Keep cleanup bounded. The foreign tab documents its own status representation rather than pretending its errors are event-horizon values.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_deadline"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import eve;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
CancelSource cancelSource;
CancelToken token = cancelSource.token();
bool sleepInterrupted = false;
bool timedOut = false;
bool cleanedUp = false;
// Timeout: fires at 50ms and cancels the token
loop.registerTimer(50, 0, (ref EventLoop l, Token t) @safe nothrow {
cancelSource.cancel();
});
// Simulated cancellable operation waiting on a timer
Token opToken = loop.registerTimer(10_000, 0, (ref EventLoop l, Token t) @safe nothrow {
// Will be cancelled before 10s elapses
});
// Loop watcher checking cancellation status
loop.registerTimer(10, 10, (ref EventLoop l, Token t) @safe nothrow {
if (token.isCancelled)
{
sleepInterrupted = true;
timedOut = true;
l.unregister(t);
l.unregister(opToken);
// Cleanup phase
cleanedUp = true;
try {
writeln("sleep returned: stopped by adapter");
writeln("timed out: ", timedOut);
writeln("cleaned up: ", cleanedUp);
} catch (Exception error) { assert(false, error.msg); }
l.stop();
}
});
loop.run();
assert(sleepInterrupted);
assert(timedOut);
assert(cleanedUp);
}sleep returned: stopped by adapter
timed out: true
cleaned up: true#!/usr/bin/env dub
/+ dub.sdl:
name "eh_deadline"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs, seconds;
import core.stdc.errno : ECANCELED;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope root, ref Env env) {
ref Sched s = currentScheduler();
bool cleanedUp;
auto outcome = withDeadline!((ref sc) {
auto slept = env.clock.sleep(10.seconds); // interrupted at 50 ms
writeln("sleep returned: ", slept.hasError ? "ECANCELED" : "ok");
assert(slept.hasError && slept.error.code == ECANCELED);
// Cleanup runs to completion even though the fiber is cancelled.
cast(void) protect!(() {
assert(!env.clock.sleep(5.msecs).hasError);
cleanedUp = true;
return 0;
})(s);
return 0;
})(s, 50.msecs);
assert(outcome.hasError && outcome.error.isTimeout);
assert(cleanedUp);
writeln("timed out: ", outcome.hasError && outcome.error.isTimeout);
writeln("cleaned up: ", cleanedUp);
return 0;
});
assert(run.hasValue);
}sleep returned: ECANCELED
timed out: true
cleaned up: trueTCP and buffer ownership
The fixture sends the five-byte message hello and asserts the echoed bytes. TCP is a byte stream: neither one write nor one readiness/completion notification is a message boundary. Accumulate partial receives and retain unsent suffixes, or use an API's documented all-bytes operation. Binding loopback port zero lets the OS reserve a port for this run without a port-selection race.
The foreign callbacks borrow buffers according to their library's contract. Copy data that must outlive a callback, or retain the documented owner. Event Horizon moves an owned buffer into the operation and returns ownership on completion. That lifetime discipline is not a promise of kernel or network zero-copy. A real protocol also needs framing, maximum message sizes and per-connection deadlines.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_tcp_echo"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import eve;
import eve.backend.common : Handle;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
auto listener = TcpListener.create();
auto server = TcpConnection.create();
auto client = TcpConnection.create();
string echoed;
// Server echoes incoming borrowed data back
server.onData = (ref TcpConnection c, scope const(ubyte)[] data) {
assert(c.send(data) == SendResult.OK);
};
// Client sends "hello" on connect
client.onConnect = (ref TcpConnection c) {
assert(c.send(cast(const(ubyte)[]) "hello") == SendResult.OK);
};
// Client receives echoed data
client.onData = (ref TcpConnection c, scope const(ubyte)[] data) {
echoed ~= cast(string) data.idup; // accumulate arbitrary TCP chunks
assert(echoed.length <= 5);
if (echoed.length < 5) return;
try writeln("echoed: ", echoed); catch (Exception error) { assert(false, error.msg); }
c.close();
server.close();
listener.close();
loop.stop();
};
// Listener adopts accepted client handle
listener.onAccept = (ref TcpListener l, Handle clientHandle) {
assert(server.adopt(loop, clientHandle) == AdoptResult.OK);
};
assert(listener.listen(loop, "127.0.0.1", 0) == ListenResult.OK);
assert(client.connect(loop, "127.0.0.1", listener.localPort) == ConnectResult.OK);
loop.run();
assert(echoed == "hello");
}echoed: hello#!/usr/bin/env dub
/+ dub.sdl:
name "eh_tcp_echo"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.lifetime : move;
import std.stdio : writeln;
import sparkles.base.buffer : UniqueBuffer;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
auto listening = env.net.listen(ipv4("127.0.0.1", 0));
assert(listening.hasValue);
auto listener = move(listening.value);
scope (exit) listener.close();
const port = boundPort(listener);
// The server is a joined child, handling one five-byte exchange.
assert(sc.spawn({
auto conn = listener.accept; // parks until a client arrives
assert(conn.hasValue);
echo(conn.value);
}));
// The client, in the same loop.
auto connected = env.net.connect(ipv4("127.0.0.1", port));
assert(connected.hasValue);
auto client = move(connected.value);
scope (exit) client.close();
sendText(client, "hello");
UniqueBuffer!(ubyte, 64) back;
string response;
while (response.length < 5)
{
back.length = 64;
auto got = client.recv(move(back));
back = move(got.buf);
assert(got.res.hasValue && got.res.value > 0);
response ~= cast(const(char)[]) back[][0 .. got.res.value];
}
assert(response == "hello");
writeln("echoed: ", response);
return 0;
});
assert(run.hasValue);
}
void echo(Stream conn)
{
scope (exit) conn.close();
UniqueBuffer!(ubyte, 4096) buf;
size_t received;
while (received < 5)
{
buf.length = 4096;
auto r = conn.recv(move(buf));
buf = move(r.buf);
assert(r.res.hasValue && r.res.value > 0);
received += r.res.value;
sendText(conn, cast(const(char)[]) buf[][0 .. r.res.value]);
}
}
// Copies the unsent suffix into an owned buffer on each attempt. This is a
// small correctness example, not a claim of zero-copy networking.
void sendText(ref Stream conn, scope const(char)[] bytes)
{
size_t offset;
while (offset < bytes.length)
{
UniqueBuffer!(ubyte, 64) buf;
buf ~= cast(const(ubyte)[]) bytes[offset .. $];
auto sent = conn.send(move(buf));
assert(sent.res.hasValue && sent.res.value > 0);
offset += sent.res.value;
}
}
ushort boundPort(ref Listener l) @trusted
{
import core.sys.posix.arpa.inet : ntohs;
import core.sys.posix.netinet.in_ : sockaddr_in;
import core.sys.posix.sys.socket : getsockname, sockaddr, socklen_t;
sockaddr_in a;
socklen_t len = a.sizeof;
assert(getsockname(l.fd, cast(sockaddr*) &a, &len) == 0);
return ntohs(a.sin_port);
}echoed: helloQueues, channels and backpressure
The shared payload contract is the ordered sequence 1 through 5, with sum 15. The flow-control contracts can differ. The event-horizon program fills capacity two, proves the third put is pending, frees one slot, and proves exactly one additional put progresses before draining the remaining values and checking the closed-channel result. No short sleep is used as the backpressure proof.
Read the foreign implementation before translating put: an unbounded queue does not acquire a capacity bound merely because the consumer is slow. A callback adapter must stop producing and arrange a wakeup when space returns; blocking an event-loop thread on an OS semaphore can stop unrelated connections. Event Horizon's Channel!(T, capacity) belongs to one scheduler and is not a drop-in cross-thread notification or broadcast mechanism.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_channel"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import eve;
import eve.rt;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
auto ch = channel!int(2); // capacity 2: bounded channel
assert(ch.trySend(1) == ChannelStatus.OK);
assert(ch.trySend(2) == ChannelStatus.OK);
assert(ch.trySend(3) == ChannelStatus.WOULD_BLOCK);
int first;
assert(ch.tryReceive(first) == ChannelStatus.OK && first == 1);
assert(ch.trySend(3) == ChannelStatus.OK);
assert(ch.trySend(4) == ChannelStatus.WOULD_BLOCK);
int sum = 1;
int nextToSend = 4;
int expected = 2;
// Producer timer on the event loop: pushes 1 .. 5 respecting backpressure
loop.registerTimer(1, 1, (ref EventLoop l, Token t) @safe nothrow {
try {
while (nextToSend <= 5)
{
const st = ch.trySend(nextToSend);
if (st == ChannelStatus.OK)
nextToSend++;
else if (st == ChannelStatus.WOULD_BLOCK)
break; // backpressure: channel full, yield to event loop
else
break;
}
if (nextToSend > 5)
{
ch.close();
l.unregister(t);
}
} catch (Exception error) { assert(false, error.msg); }
});
// Consumer timer on the event loop: drains items
loop.registerTimer(2, 2, (ref EventLoop l, Token t) @safe nothrow {
try {
int val;
for (;;)
{
const st = ch.tryReceive(val);
if (st == ChannelStatus.OK)
{
assert(val == expected++);
sum += val;
}
else if (st == ChannelStatus.WOULD_BLOCK)
{
break; // empty for now
}
else if (st == ChannelStatus.CLOSED)
{
l.unregister(t);
writeln("consumed: ", sum);
l.stop();
break;
}
}
} catch (Exception error) { assert(false, error.msg); }
});
loop.run();
assert(sum == 15);
assert(expected == 6);
}consumed: 15#!/usr/bin/env dub
/+ dub.sdl:
name "eh_channel"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import core.stdc.errno : EPIPE;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope root, ref Env env) {
ref Sched s = currentScheduler();
Channel!(int, 2) items; // capacity 2: the producer parks on the third put
int produced;
bool attemptingThird;
auto outcome = withScope!((ref sc) {
assert(sc.spawn(() {
foreach (i; 1 .. 6)
{
if (i == 3) attemptingThird = true;
assert(!items.put(s, i).hasError);
produced = i;
}
items.close();
}));
while (!attemptingThird) s.yieldNow();
assert(produced == 2, "third put must remain pending while full");
auto first = items.take(s);
assert(first.hasValue && first.value == 1);
while (produced < 3) s.yieldNow();
assert(produced == 3, "one freed slot permits exactly one put");
int sum = 1;
int expected = 2;
for (;;)
{
auto next = items.take(s); // parks while empty
if (next.hasError)
{
assert(next.error.code == EPIPE);
break; // closed and drained: EPIPE
}
assert(next.value == expected++);
assert(!env.clock.sleep(2.msecs).hasError); // the slow consumer
sum += next.value;
}
assert(expected == 6 && produced == 5);
return sum;
})(s);
assert(outcome.hasValue && outcome.value == 15);
writeln("consumed: ", outcome.value);
return 0;
});
assert(run.hasValue);
}consumed: 15Files and operation lifetime
The fixture owns a fresh temporary directory and a file containing exactly hello from a file\n. Assertions check the byte count and contents. Keep file and buffer ownership until all pending operations finish; then close the file before removing the fixture. Never classify an arbitrary read failure as EOF.
The event-horizon example reads until EOF using offsets and an owned buffer. Its Linux io_uring implementation submits file operations to the kernel. This does not establish the same behavior for every backend: kqueue synthesizes completions over readiness and its regular-file worker-pool refinement is separate. Where the foreign example uses a worker adapter, it is explicitly application code, not evidence of a native filesystem capability.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_file_read"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.fcntl : O_RDONLY;
import std.file : remove, tempDir, write;
import std.path : buildPath;
import std.stdio : writeln;
import eve;
void main() @safe
{
const directory = fixtureDirectory();
scope(exit) { import std.file : rmdir; rmdir(directory); }
const path = buildPath(directory, "input");
write(path, "hello from a file\n");
scope (exit) remove(path);
auto loop = EventLoop.create();
scope (exit) loop.dispose();
auto file = AsyncFile.create();
scope (exit) file.dispose();
assert(file.open(loop, path, O_RDONLY) == OpenResult.OK);
ubyte[128] buffer;
size_t bytesRead = 0;
string text;
file.onRead = (ref AsyncFile f, ubyte[] data, int error) @safe {
if (error == 0 && data !is null)
{
bytesRead = data.length;
text = cast(string) data.idup; // borrowed slice
try {
writeln("read ", bytesRead, " bytes: ", text[0 .. $ - 1]);
} catch (Exception error) { assert(false, error.msg); }
}
f.close();
loop.stop();
};
assert(file.read(buffer[], 0) == FileResult.PENDING);
loop.run();
assert(bytesRead == 18);
assert(text == "hello from a file\n");
}
// A fresh directory prevents interference between parallel tutorial runs.
string fixtureDirectory() @trusted
{
import core.sys.posix.stdlib : mkdtemp;
import std.string : fromStringz;
auto pattern = (buildPath(tempDir(), "eh-comparison-XXXXXX") ~ '\0').dup;
auto created = mkdtemp(pattern.ptr);
assert(created !is null);
return fromStringz(created).idup;
}read 18 bytes: hello from a file#!/usr/bin/env dub
/+ dub.sdl:
name "eh_file_read"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.lifetime : move;
import core.sys.posix.stdlib : mkdtemp;
import std.file : remove, rmdir, tempDir, write;
import std.path : buildPath;
import std.stdio : writeln;
import sparkles.base.buffer : UniqueBuffer;
import sparkles.base.vfs : ErrorKind, OpenMode, Rights, ambientAuthority, openRoot;
import sparkles.event_horizon;
void main()
{
auto template_ = (buildPath(tempDir(), "eh-file-XXXXXX") ~ '\0').dup;
assert(mkdtemp(template_.ptr) !is null);
const dir = template_[0 .. $ - 1].idup;
scope (exit) rmdir(dir);
const path = buildPath(dir, "input.txt");
write(path, "hello from a file\n");
scope (exit) remove(path);
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
// Files are reached through a directory capability, and close
// themselves; `read` borrows the file's descriptor.
auto root = openRoot!(Rights.readOnly)(&env.fs(), dir, ambientAuthority());
assert(root.hasValue);
auto missing = root.value.openFile!(OpenMode.read)("missing");
assert(missing.hasError && missing.error.kind == ErrorKind.notFound);
auto opened = root.value.openFile!(OpenMode.read)("input.txt");
assert(opened.hasValue);
auto f = move(opened.value);
UniqueBuffer!(ubyte, 128) buf;
string contents;
for (;;)
{
buf.length = 128;
auto got = read(f, move(buf), contents.length);
buf = move(got.buf);
assert(got.res.hasValue);
if (got.res.value == 0)
break;
contents ~= cast(const(char)[]) buf[][0 .. got.res.value];
}
assert(contents == "hello from a file\n");
writeln("read ", contents.length, " bytes: ", contents);
return 0;
});
assert(run.hasValue);
}read 18 bytes: hello from a fileCapturing a child process
The child writes different text to stdout and stderr, then exits with code 3. Both streams and the nonzero status are asserted: a child command failing is not necessarily failure to launch or observe it. Root reaping is part of the operation's lifetime.
A general capture implementation must drain both pipes concurrently, even when one is quiet. Reading all stdout before stderr can deadlock when the stderr pipe fills. Tiny fixed fixtures do not prove a hand-written adapter safe for unbounded output. capture supplies the drain/reap coordination; adapters need their own output limits, cancellation and error handling.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_exec"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import std.process : pipeProcess, Redirect, wait;
import core.thread : Thread;
void main()
{
runWorker({
auto child = pipeProcess(["sh", "-c", "echo out; echo err >&2; exit 3"], Redirect.stdout | Redirect.stderr);
string stdoutText, stderrText;
auto drainError = new Thread({
foreach (chunk; child.stderr.byChunk(1024)) stderrText ~= cast(const(char)[]) chunk;
child.stderr.close();
});
drainError.start();
foreach (chunk; child.stdout.byChunk(1024)) stdoutText ~= cast(const(char)[]) chunk;
child.stdout.close();
const code = child.pid.wait();
drainError.join();
assert(stdoutText == "out\n" && stderrText == "err\n" && code == 3);
writeln("stdout: ", stdoutText);
writeln("stderr: ", stderrText);
writeln("exit code: ", code);
});
}
void runWorker(void delegate() work)
{
import eve;
import core.atomic : atomicLoad, atomicStore;
import core.thread : Thread;
auto loop = EventLoop.create();
scope(exit) loop.dispose();
shared bool finished;
auto worker = new Thread({ work(); atomicStore(finished, true); });
worker.start();
loop.registerTimer(1, 1, (ref EventLoop l, Token t) @safe nothrow {
if (atomicLoad(finished)) l.stop();
});
loop.run();
worker.join();
assert(atomicLoad(finished));
}stdout: out
stderr: err
exit code: 3#!/usr/bin/env dub
/+ dub.sdl:
name "eh_exec"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
ref Sched s = currentScheduler();
ProcessConfig cfg;
cfg.stderrSpec = StdioSpec(StdioMode.pipe);
auto got = capture(s, ["sh", "-c", "echo out; echo err >&2; exit 3"], cfg);
assert(got.hasValue);
assert(got.value.status.code == 3);
assert(cast(const(char)[]) got.value.stdout_[] == "out\n");
assert(cast(const(char)[]) got.value.stderr_[] == "err\n");
writeln("stdout: ", cast(const(char)[]) got.value.stdout_[]);
writeln("stderr: ", cast(const(char)[]) got.value.stderr_[]);
writeln("exit code: ", got.value.status.code);
return 0;
});
assert(run.hasValue);
}stdout: out
stderr: err
exit code: 3POSIX signals and loop notifications
Install the signal receiver before sending SIGUSR1 to this process and assert that the expected signal was observed. Linux signal masks and handler installation are process/thread integration decisions; they are not interchangeable with an ordinary callback queue. Keep signal-handler work async-signal-safe.
SignalFd receives masked POSIX signals through the event-horizon loop. A foreign notification primitive may only wake an owner loop from another thread. In particular, libasync's AsyncSignal is not a POSIX signal subscription; an OS signal bridge must supply that integration explicitly. These programs are Linux fixtures, not Windows signal portability examples.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_signal"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.signal : SIGUSR1, kill;
import core.sys.posix.unistd : getpid;
import std.stdio : writeln;
import eve;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
bool gotSignal = false;
loop.registerSignal(Signal.USER1, (ref EventLoop l, Token t, SignalInfo info) @safe nothrow {
if (info.signal == Signal.USER1)
{
gotSignal = true;
try writeln("got signal SIGUSR1"); catch (Exception error) { assert(false, error.msg); }
}
l.stop();
});
// Send SIGUSR1 to own process
() @trusted { assert(kill(getpid(), SIGUSR1) == 0); }();
loop.run();
assert(gotSignal);
}got signal SIGUSR1#!/usr/bin/env dub
/+ dub.sdl:
name "eh_signal"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.signal : SIGUSR1, kill;
import core.sys.posix.unistd : getpid;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
SignalFd signals;
if (SignalFd.create(signals, [SIGUSR1]).hasError)
assert(false, "signalfd unavailable");
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
ref Sched s = currentScheduler();
assert(kill(getpid(), SIGUSR1) == 0);
auto got = signals.nextSignal(s); // parks until the signal completes
assert(got.hasValue && got.value == SIGUSR1);
writeln("got signal ", got.value == SIGUSR1 ? "SIGUSR1" : "other");
return 0;
});
assert(run.hasValue);
}got signal SIGUSR1Streaming and root termination: an explicit adapter
This foreign program is an application-side process adapter, not a native supervision API. It owns an OS worker, reads complete lines with Phobos, sends TERM only after the second line establishes readiness, and asserts the root's reaped status. exec sleep avoids leaving a shell child holding the output pipe. The library-specific runWorker code makes completion delivery and joining visible. EVE's adapter polls an atomic completion flag; Collie posts a task; libasync triggers an owner-loop notifier; Hunt uses an explicitly joined Task.
The event-horizon program requests scope cancellation after the same readiness point. It asserts the final cancelled result, root reap status and natural EOF. Neither fixture demonstrates unescapable descendants. Timeouts and hard-kill escalation are separate policies; see the deadline pair and containment notes.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_spawn"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import std.process : pipeProcess, Redirect, wait, kill;
import core.sys.posix.signal : SIGTERM;
void main()
{
// Application process adapter: the networking runtime delivers completion,
// while this worker owns pipes, signalling and root reaping. No tree claim.
runWorker({
auto child = pipeProcess(["sh", "-c", "echo one; echo two; exec sleep 30"], Redirect.stdout);
string[] lines;
foreach (line; child.stdout.byLine)
{
lines ~= line.idup;
assert(lines.length <= 2);
if (lines.length == 2) child.pid.kill(SIGTERM); // child-ready handshake
}
const status = child.pid.wait();
child.stdout.close();
assert(lines == ["one", "two"]);
assert(status == -SIGTERM);
foreach(line; lines) writeln("line: ", line);
writeln("exited: SIGTERM");
writeln("end: reaped");
});
}
void runWorker(void delegate() work)
{
import eve;
import core.atomic : atomicLoad, atomicStore;
import core.thread : Thread;
auto loop = EventLoop.create();
scope(exit) loop.dispose();
shared bool finished;
auto worker = new Thread({ work(); atomicStore(finished, true); });
worker.start();
loop.registerTimer(1, 1, (ref EventLoop l, Token t) @safe nothrow {
if (atomicLoad(finished)) l.stop();
});
loop.run();
worker.join();
assert(atomicLoad(finished));
}line: one
line: two
exited: SIGTERM
end: reaped#!/usr/bin/env dub
/+ dub.sdl:
name "eh_spawn"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
ref Sched s = currentScheduler();
SupervisedProcessConfig cfg;
cfg.terminateGrace = 500.msecs; // cancellation: TERM, then bounded KILL grace
cfg.process.stdinSpec = StdioSpec(StdioMode.nullDev);
int linesSeen;
SupervisedProcessResult result;
bool returned;
auto scopeResult = withScope!((ref outer) {
JoinHandle!void task;
outer.fork(task, () {
auto got = supervise(s, ["sh", "-c", "echo one; echo two; exec sleep 30"], cfg,
null, (in ProcessEvent ev) {
final switch (ev.kind)
{
case ProcessEventKind.line:
assert(cast(const(char)[]) ev.line.bytes == (linesSeen == 0 ? "one" : "two"));
writeln("line: ", cast(const(char)[]) ev.line.bytes);
++linesSeen;
if (linesSeen == 2) outer.cancel(); // readiness handshake, no startup-time race
break;
case ProcessEventKind.exited:
writeln("exited: ", ev.end, ", signaled by ",
ev.status.signaled ? ev.status.code : 0);
break;
case ProcessEventKind.sample:
break; // periodic resource samples; not printed here
}
});
assert(got.hasValue);
result = got.value;
returned = true;
return ioOk();
});
task.join(s);
})(s);
assert(!scopeResult.hasError && returned);
assert(linesSeen == 2);
assert(result.end == ProcessEnd.cancelled);
assert(result.reap == ReapOutcome.reaped);
assert(result.status.signaled && result.status.code == 15);
assert(!result.eofForced);
writeln("end: ", result.end, ", reap: ", result.reap);
return 0;
});
assert(run.hasValue);
}line: one
line: two
exited: cancelled, signaled by 15
end: cancelled, reap: reapedRetries and finite policy
The first two attempts represent transient failure; the third succeeds. The foreign timer adapter uses a fixed interval (Photon uses explicit exponential delays), whereas event-horizon composes exponential with recurs. Assert the attempt budget and final value rather than exact elapsed time. In production, retry only selected transient errors, propagate cancellation and permanent failures, and ensure repeated side effects are safe. A timer callback becoming ready is not itself permission to retry a non-idempotent operation.
#!/usr/bin/env dub
/+ dub.sdl:
name "eve_retry"
dependency "eve" repository="git+https://codeberg.org/ddn/eve.git" version="c72f75135de636987e91bcd8ec5a22d32d34a197"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import eve;
void main() @safe
{
auto loop = EventLoop.create();
scope (exit) loop.dispose();
int ticks = 0;
int failures, value;
// Register a repeating timer: initial delay 10ms, interval 10ms
loop.registerTimer(10, 10, (ref EventLoop l, Token t) @safe nothrow {
ticks++;
auto error = tryOperation(ticks, value);
if (error) { assert(error == 11); ++failures; return; }
try if (ticks == 3) writeln("succeeded on attempt ", ticks); catch (Exception error) { assert(false, error.msg); }
if (ticks >= 3)
l.stop();
});
loop.run();
assert(ticks == 3);
assert(failures == 2 && value == 42);
}
int tryOperation(int attempt, out int value) @safe nothrow
{
if (attempt < 3) return 11; // deterministic Linux EAGAIN injection
value = 42;
return 0;
}succeeded on attempt 3#!/usr/bin/env dub
/+ dub.sdl:
name "eh_retry"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import core.stdc.errno : EAGAIN;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
ref Sched s = currentScheduler();
int attempts;
auto outcome = withScope!((ref inner) {
enum policy = exponential(5.msecs) & recurs(4);
auto r = retry(inner, env.clock, policy, () {
++attempts;
if (attempts < 3)
return ioErr!int(EAGAIN, OpKind.none, IoErrorStage.submit,
"flaky");
return ioOk(attempts);
});
return r.hasValue ? r.value : -1;
})(s);
assert(outcome.hasValue && outcome.value == 3);
writeln("succeeded on attempt ", outcome.value);
return 0;
});
assert(run.hasValue);
}succeeded on attempt 3Errors and migration decisions
| Boundary | Migration decision |
|---|---|
| Ordinary operation failure | Check the foreign status/exception and the event-horizon IoResult; do not silently treat failure as success or EOF. |
| Task failure | Decide who observes the error and who joins remaining work. Scope outcomes and defects are distinct from ordinary I/O errors. |
| Buffer lifetime | Retain borrowed data until the foreign operation finishes; recover moved buffers from event-horizon completions. |
| Cross-thread work | Use a documented cross-thread bridge. Scheduler-local channels do not replace it. |
| Missing native primitive | Treat a Phobos/POSIX adapter as application integration, not as a feature supplied by the networking library. |
Next steps
Start with one independently owned connection or operation. Preserve its framing, error handling and shutdown behavior before changing scheduler topology. The Node.js comparison provides another fully executable view of the same concerns, including explicit retries. The execution guide records platform and containment limits.