Coming from collie
Collie presents networking through a Netty-style channel and pipeline framework. Event Horizon supplies lower-level I/O verbs and scoped fibers; it does not replace a codec or handler pipeline automatically.
The Collie baseline is f1e58e38a2c36366766e4778d3ea655ebac6962c. Its support matrix lists epoll, kqueue, IOCP, and select. IOCP is a completion interface, so calling every Collie backend a readiness reactor would erase an important platform distinction.
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.
Move one connection at a time
| Collie concept | Event Horizon building block | Work still owned by the application |
|---|---|---|
| Event-loop-managed channel | Listener, Stream, owning scope | Decide connection lifetime and shutdown policy. |
| Inbound handler pipeline | A fiber's receive/decode loop | Preserve protocol state and codec boundaries. |
| Outbound writes | send(move(buf)) | Handle partial writes; a completion need not send everything. |
| Timer callback | env.clock.sleep, Ticker | Choose relative delay or absolute cadence. |
| Handler error path | IoResult!T and scope outcome | Distinguish I/O failure, cancellation, and defects. |
Do not infer that moving buffers eliminates application-level copying or all aliasing. It expresses ownership through an operation; protocol framing remains a separate responsibility.
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
Collie's event engine is its kiss dependency; the timer and transport examples use that lower layer explicitly. The pinned Collie source requires the documented compatibility preparation on the current compiler. Queue and process adapters are application code, not additional native Collie APIs.
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 "collie_timers"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import kiss.event;
import kiss.util.timer;
void main()
{
auto loop = new EventLoop();
scope(exit) loop.dispose();
int ticks = 0;
Timer timer = new Timer(loop, 10.msecs);
timer.onTick((Object sender) {
++ticks;
writeln("tick ", ticks);
if (ticks >= 3)
{
timer.stop();
loop.stop();
}
});
timer.start();
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 "collie_concurrency"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import kiss.event;
import kiss.util.timer;
void main()
{
auto loop = new EventLoop();
scope(exit) loop.dispose();
int a = 0;
int b = 0;
int doneCount = 0;
// Slow task (30 ms)
Timer slowTimer = new Timer(loop, 30.msecs);
slowTimer.onTick((Object) {
slowTimer.stop();
a = 1;
++doneCount;
if (doneCount == 2)
loop.stop();
});
slowTimer.start(false, true);
// Fast task (5 ms)
Timer fastTimer = new Timer(loop, 5.msecs);
fastTimer.onTick((Object) {
fastTimer.stop();
b = 2;
++doneCount;
if (doneCount == 2)
loop.stop();
});
fastTimer.start(false, true);
loop.run();
const joined = a * 10 + b;
assert(joined == 12);
writeln("joined: ", joined);
}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 "collie_deadline"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs, seconds;
import std.stdio : writeln;
import kiss.event;
import kiss.util.timer;
void main()
{
auto loop = new EventLoop();
scope(exit) loop.dispose();
bool timedOut = false;
bool cleanedUp = false;
string sleepResult = "ok";
// Long-running operation (10s)
Timer slowOp = new Timer(loop, 10.seconds);
slowOp.onTick((Object) {
sleepResult = "ok";
});
slowOp.start();
// Deadline timer (50ms)
Timer deadline = new Timer(loop, 50.msecs);
deadline.onTick((Object) {
deadline.stop();
timedOut = true;
// In collie, cancellation requires manually stopping operations
slowOp.stop();
sleepResult = "stopped by adapter";
writeln("sleep returned: ", sleepResult);
// Run manual cleanup
cleanedUp = true;
loop.stop();
});
deadline.start();
loop.run();
assert(sleepResult == "stopped by adapter");
assert(timedOut);
assert(cleanedUp);
writeln("timed out: ", timedOut);
writeln("cleaned up: ", 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 "collie_tcp_echo"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.socket : InternetAddress, AddressFamily;
import std.stdio : writeln;
import kiss.event;
import kiss.net.TcpListener;
import kiss.net.TcpStream;
void main()
{
auto loop = new EventLoop();
scope(exit) loop.dispose();
string response;
// Server: listen and accept
auto listener = new TcpListener(loop, AddressFamily.INET);
listener.onConnectionAccepted((TcpListener sender, TcpStream serverStream) {
serverStream.onDataReceived((in ubyte[] data) {
// Collie borrows the slice during callback;
// writing requires heap duplication or a ref-counted buffer.
serverStream.write(cast(const(ubyte)[]) data.dup);
});
serverStream.start();
});
listener.bind(new InternetAddress("127.0.0.1", 0));
const port = (cast(InternetAddress) listener.localAddress).port;
listener.listen(128);
listener.start();
// Client: connect, send, receive
auto client = new TcpStream(loop);
client.onConnected((bool ok) {
assert(ok);
if (ok)
client.write(cast(const(ubyte)[]) "hello");
});
client.onDataReceived((in ubyte[] data) {
response ~= cast(string) data.dup;
assert(response.length <= 5);
if (response.length < 5) return;
writeln("echoed: ", response);
client.close();
listener.close();
loop.stop();
});
client.connect("127.0.0.1", port);
loop.run();
assert(response == "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 "collie_channel"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sync.semaphore : Semaphore;
import core.thread : Thread;
import core.time : msecs;
import std.stdio : writeln;
// In collie/Netty, bounded backpressure without fibers requires explicit
// synchronization (e.g. semaphores or watermark events) around a shared buffer.
final class BoundedQueue(T, size_t Capacity)
{
private T[Capacity] _items;
private size_t _head;
private size_t _tail;
private size_t _count;
private Semaphore _emptySlots;
private Semaphore _fullSlots;
private bool _closed;
this()
{
_emptySlots = new Semaphore(Capacity);
_fullSlots = new Semaphore(0);
}
void put(T item)
{
_emptySlots.wait();
synchronized (this)
{
_items[_tail] = item;
_tail = (_tail + 1) % Capacity;
++_count;
}
_fullSlots.notify();
}
void close()
{
synchronized (this)
{
_closed = true;
}
_fullSlots.notify();
}
bool take(out T item)
{
_fullSlots.wait();
synchronized (this)
{
if (_count == 0 && _closed)
return false;
item = _items[_head];
_head = (_head + 1) % Capacity;
--_count;
}
_emptySlots.notify();
return true;
}
}
void main()
{
runWorker({
auto queue = new BoundedQueue!(int, 2)();
auto readyForThird = new Semaphore(0);
shared int produced;
import core.atomic : atomicLoad, atomicStore;
auto producer = new Thread({
foreach (i; 1 .. 6)
{
if (i == 3) readyForThird.notify();
queue.put(i);
atomicStore(produced, i);
}
queue.close();
});
producer.start();
readyForThird.wait();
assert(atomicLoad(produced) == 2, "full queue prevents third put from completing");
int sum = 0;
int val;
int expected = 1;
while (queue.take(val))
{
assert(val == expected++);
sum += val;
}
producer.join();
assert(sum == 15);
assert(expected == 6 && atomicLoad(produced) == 5);
writeln("consumed: ", sum);
});
}
void runWorker(void delegate() work)
{
import kiss.event;
import kiss.event.task : newTask;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) loop.dispose();
bool delivered;
auto worker = new Thread({
work();
loop.postTask(newTask({ delivered = true; loop.stop(); }));
});
worker.start();
loop.run();
worker.join();
assert(delivered);
}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 "collie_file_read"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.thread : Thread;
import std.file : remove, tempDir, write, read;
import std.path : buildPath;
import std.stdio : writeln;
import kiss.event;
import kiss.event.task : newTask;
void main()
{
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 = new EventLoop();
scope(exit) loop.dispose();
string content;
// Collie has no async filesystem I/O: epoll cannot poll regular files.
// Applications must delegate disk I/O to a worker thread and post back.
auto worker = new Thread({
auto data = cast(string) read(path);
loop.postTask(newTask({
content = data;
writeln("read ", content.length, " bytes: ", content);
loop.stop();
}));
});
worker.start();
loop.run();
worker.join();
assert(content == "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 "collie_exec"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
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 kiss.event;
import kiss.event.task : newTask;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) loop.dispose();
bool delivered;
auto worker = new Thread({
work();
loop.postTask(newTask({ delivered = true; loop.stop(); }));
});
worker.start();
loop.run();
worker.join();
assert(delivered);
}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
The foreign tab uses sigwait in an ordinary worker and then delivers completion through its runtime bridge. Unlike calling an event method from a signal handler, this does not assume the event method is async-signal-safe. The signal is thread-directed for an isolated, reproducible fixture.
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 "collie_signal"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.signal;
import core.sys.posix.pthread : pthread_self;
import std.stdio : writeln;
void main()
{
int received;
// Application POSIX adapter. sigwait runs in ordinary thread context, so
// no event, allocation or mutex operation is called from a signal handler.
runWorker({
sigset_t signals, previous;
assert(sigemptyset(&signals) == 0);
assert(sigaddset(&signals, SIGUSR1) == 0);
assert(pthread_sigmask(SIG_BLOCK, &signals, &previous) == 0);
scope(exit) assert(pthread_sigmask(SIG_SETMASK, &previous, null) == 0);
assert(pthread_kill(pthread_self(), SIGUSR1) == 0);
assert(sigwait(&signals, &received) == 0);
});
assert(received == SIGUSR1);
writeln("got signal SIGUSR1");
}
void runWorker(void delegate() work)
{
import kiss.event;
import kiss.event.task : newTask;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) loop.dispose();
bool delivered;
auto worker = new Thread({
work();
loop.postTask(newTask({ delivered = true; loop.stop(); }));
});
worker.start();
loop.run();
worker.join();
assert(delivered);
}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 "collie_spawn"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
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 kiss.event;
import kiss.event.task : newTask;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) loop.dispose();
bool delivered;
auto worker = new Thread({
work();
loop.postTask(newTask({ delivered = true; loop.stop(); }));
});
worker.start();
loop.run();
worker.join();
assert(delivered);
}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 "collie_retry"
dependency "collie" path=".deps/collie"
dependency "kiss" path=".deps/kiss"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import kiss.event;
import kiss.util.timer;
void main()
{
auto loop = new EventLoop();
scope(exit) loop.dispose();
int ticks = 0;
int failures, value;
Timer timer = new Timer(loop, 10.msecs);
timer.onTick((Object sender) {
++ticks;
auto error = tryOperation(ticks, value);
if (error) { assert(error == 11); ++failures; return; }
if (ticks == 3) writeln("succeeded on attempt ", ticks);
if (ticks >= 3)
{
timer.stop();
loop.stop();
}
});
timer.start();
loop.run();
assert(ticks == 3);
assert(failures == 2 && value == 42);
}
// Deterministic dependency failure: EAGAIN twice, then a value. The timer
// supplies fixed-interval backoff; production code also needs a retry budget.
int tryOperation(int attempt, out int value) @safe nothrow
{
if (attempt < 3) return 11; // EAGAIN on the Linux fixture platform
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.