Coming from hunt-net
Hunt-net's README examples configure a server or client, install a TextLineCodec, and handle decoded messages through AbstractNetConnectionHandler.messageReceived. Event Horizon exposes the stream and scope primitives below that framework level.
This comparison uses revision 81fd38deebe6a5ec093ef9e86770cd0dce22b355. It is a source-based migration guide, not a claim of feature or performance parity.
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.
Keep the codec boundary explicit
| hunt-net idiom | Event Horizon approach | Important difference |
|---|---|---|
NetUtil.createNetServer and .listen | env.net.listen, then accept | Give each accepted connection an owning scope or child fiber. |
TextLineCodec | An incremental decoder over received bytes | TCP can split or combine application messages. |
messageReceived | Resume the waiting receive fiber | Local variables can hold per-connection parser state. |
connection.write | send(move(buf)) | Loop over partial sends; do not reuse an in-flight buffer. |
| Server/client lifecycle callbacks | Scope lifetime and explicit close | Cancellation and cleanup need a deliberate policy. |
The process supervisor's LineFramer contract does not make every socket a line stream. Decide whether to reuse a compatible framing primitive or retain your existing protocol codec when porting a service.
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 "hunt_timers"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import core.time : Duration, msecs;
import core.sync.semaphore : Semaphore;
import core.stdc.stdlib : exit;
import core.sys.posix.unistd : read;
import hunt.event.EventLoop;
import hunt.event.selector.Selector;
import hunt.util.Timer;
import hunt.net.EventLoopPool : EventLoopObjectFactory;
import hunt.Functions;
// Hunt's AbstractTimer has an upstream 4-byte buffer overflow bug reading
// timerfd (reading 8 bytes into a 32-bit uint). We subclass it safely here.
class SafeTimer : Timer {
this(Selector loop, Duration dur) {
super(loop, dur);
}
override bool readTimer(scope SimpleActionHandler handler) {
this.clearError();
ulong value;
read(this.handle, &value, 8);
this._readBuffer.data = cast(uint) value;
if (handler)
handler(this._readBuffer);
return false;
}
}
void main()
{
stopStartupClock();
EventLoop loop = EventLoopObjectFactory.buildEventLoop();
scope(exit) loop.stop();
auto sem = new Semaphore(0);
int count = 0;
auto timer = new SafeTimer(loop, 10.msecs);
timer.onTick((Object sender) {
count++;
writeln("tick ", count);
if (count == 3) {
timer.stop();
sem.notify();
}
});
timer.start();
sem.wait();
assert(count == 3);
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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
Hunt's tab wraps an explicit Phobos worker arrangement in a Hunt Task, checks its final state and joins the OS thread. This is an application adapter; it does not turn Task.stop() into preemptive interruption or structured joining.
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 "hunt_concurrency"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import core.time : msecs;
import core.thread : Thread;
import core.sync.mutex : Mutex;
import core.sync.condition : Condition;
import core.stdc.stdlib : exit;
import std.parallelism : task, taskPool;
void main()
{
stopStartupClock();
runWorker({
// Application-side Phobos worker adapter, not a hunt-net task-scope API.
auto mtx = new Mutex();
auto cond = new Condition(mtx);
int a = 0;
int b = 0;
int finished = 0;
auto taskA = task({
Thread.sleep(30.msecs);
synchronized (mtx) {
a = 1;
finished++;
if (finished == 2) cond.notify();
}
});
auto taskB = task({
Thread.sleep(5.msecs);
synchronized (mtx) {
b = 2;
finished++;
if (finished == 2) cond.notify();
}
});
taskPool.put(taskA);
taskPool.put(taskB);
synchronized (mtx) {
while (finished < 2)
cond.wait();
}
int joined = a * 10 + b;
assert(joined == 12);
writeln("joined: ", joined);
});
}
void runWorker(void delegate() work)
{
import hunt.util.worker.Task : Task;
import core.thread : Thread;
auto job = new class Task {
override protected void doExecute() { work(); }
};
auto thread = new Thread(&job.execute);
thread.start();
thread.join();
assert(job.isDone());
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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 Hunt tab demonstrates a condition-variable timeout inside an explicitly joined Hunt Task. The condition returns false: no kernel operation returns ECANCELED, and no child task has been interrupted. The printed adapter status is deliberately different from event-horizon's cancellation result.
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 "hunt_deadline"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import core.time : msecs;
import core.thread : Thread;
import core.sync.mutex : Mutex;
import core.sync.condition : Condition;
import core.stdc.stdlib : exit;
void main()
{
stopStartupClock();
runWorker({
// Application-side condition timeout; this is not a hunt-net cancellation API.
// there are no structured cancel scopes, so cancellation is simulated
// by signaling a condition variable or closing the connection session.
auto mtx = new Mutex();
auto cond = new Condition(mtx);
bool timedOut = false;
bool cleanedUp = false;
string sleepStatus;
// Simulate an operation waiting with a 50ms deadline
synchronized (mtx) {
bool notified = cond.wait(50.msecs);
if (!notified) {
timedOut = true;
sleepStatus = "stopped by adapter"; // deadline expired before 10s passed
} else {
sleepStatus = "ok";
}
}
writeln("sleep returned: ", sleepStatus);
// In hunt-net, cleanup must be executed explicitly in try/finally or closed handlers
try {
Thread.sleep(5.msecs);
cleanedUp = true;
} finally {
assert(timedOut);
assert(cleanedUp);
}
writeln("timed out: ", timedOut);
writeln("cleaned up: ", cleanedUp);
});
}
void runWorker(void delegate() work)
{
import hunt.util.worker.Task : Task;
import core.thread : Thread;
auto job = new class Task {
override protected void doExecute() { work(); }
};
auto thread = new Thread(&job.execute);
thread.start();
thread.join();
assert(job.isDone());
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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
In the pinned 0.7.1 implementation, NetServer.actualPort() returns the configured port, not the kernel-assigned port when configured with zero. This fixture uses a Phobos server bound to port zero and exercises the real Hunt-net client and its ByteBuffer callback. The fixture server is not presented as Hunt-net code. The client owns an initially stopped event loop, so connection initialization runs on the loop's thread. The main thread waits for the close notification, not just the last data callback, before tearing down that loop. This pinned dependency graph also starts a DateTime daemon before main but does not join it at shutdown. Each standalone Hunt fixture explicitly stops and joins the sole startup daemon before starting application threads. Its assertion fails if that startup ownership changes; this process-wide workaround is not a recipe for shutting down someone else's threads in an embedded application.
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 "hunt_tcp_echo"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.thread : Thread;
import core.time : seconds;
import core.sync.semaphore : Semaphore;
import std.socket : TcpSocket, InternetAddress, SocketOptionLevel, SocketOption;
import std.stdio : writeln;
import hunt.net.Connection : Connection, NetConnectionHandler;
import hunt.net.NetClientOptions : NetClientOptions;
import hunt.io.channel.Common : DataHandleStatus;
import hunt.net.NetClientImpl : NetClientImpl;
import hunt.event.EventLoop : EventLoop;
import hunt.io.channel.posix.EpollEventChannel : EpollEventChannel;
import hunt.io.ByteBuffer;
void main()
{
// This pinned dependency graph starts a DateTime daemon before main. Its
// static destructor only sets a flag; it does not join before GC teardown.
// This standalone fixture owns the process: stop and join that sole startup
// thread before creating any application threads. No arbitrary sleep.
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if the dependency graph changes");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
// A Phobos fixture server gives the actual bound ephemeral port. In 0.7.1,
// NetServer.actualPort returns the configured port (zero), not the bound one.
auto listener = new TcpSocket();
scope(exit) listener.close();
listener.bind(new InternetAddress("127.0.0.1", 0));
listener.listen(1);
listener.setOption(SocketOptionLevel.SOCKET, SocketOption.RCVTIMEO, 10.seconds);
const port = (cast(InternetAddress) listener.localAddress).port;
auto server = new Thread({
auto socket = listener.accept();
scope(exit) socket.close();
socket.setOption(SocketOptionLevel.SOCKET, SocketOption.RCVTIMEO, 10.seconds);
socket.setOption(SocketOptionLevel.SOCKET, SocketOption.SNDTIMEO, 10.seconds);
char[5] bytes;
size_t received;
while (received < bytes.length)
{
auto n = socket.receive(bytes[received .. $]);
assert(n > 0);
received += n;
}
assert(bytes[] == "hello");
size_t sent;
while (sent < bytes.length)
{
auto n = socket.send(bytes[sent .. $]);
assert(n > 0);
sent += n;
}
});
server.start();
scope(exit) server.join(false);
auto done = new Semaphore(0);
string received;
// A dedicated, initially stopped loop initializes the connection on its
// owning thread. Do not mutate a pooled, already-running loop from main.
auto loop = new EventLoop();
scope(exit) loop.stop();
auto client = new NetClientImpl(loop, new NetClientOptions());
scope(exit) client.close();
client.onClosed = { done.notify(); };
client.setHandler(new class NetConnectionHandler {
override void connectionOpened(Connection connection) {
connection.write(cast(const(ubyte)[]) "hello");
}
override void connectionClosed(Connection connection) {}
override DataHandleStatus messageReceived(Connection connection, Object message) {
auto buffer = cast(ByteBuffer) message;
assert(buffer !is null);
received ~= (cast(const(char)[]) buffer.peekRemaining()).idup;
assert(received.length <= 5);
return DataHandleStatus.Done;
}
override void exceptionCaught(Connection connection, Throwable error) {
assert(false, error.msg);
}
});
// Dispatch initialization after the selector declares itself ready. Hunt's
// runAsync initializer fires before that point, so NetClient would try to
// start an already-running loop and lose its connection initialization.
auto start = new class EpollEventChannel {
this() { super(loop); }
override void onRead() {
super.onRead();
client.connect("127.0.0.1", port);
close();
}
};
assert(loop.register(start));
start.trigger();
auto loopThread = new Thread({
scope(exit) done.notify();
loop.run(10);
});
loopThread.start();
const completed = done.wait(15.seconds);
loop.stop();
// stop() may have been initiated on the loop itself and handed to a worker.
// Join the actual loop thread even if a second stop() returns early.
loopThread.join();
assert(completed, "Hunt event loop did not finish within the fixture deadline");
server.join();
assert(received == "hello");
writeln("echoed: ", received);
}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
Hunt's SimpleQueue has no capacity bound. Its example intentionally lets the producer finish before the consumer starts: this proves the absence of a capacity-two backpressure guarantee. Add a bounded application adapter before using this design for an unbounded event source.
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 "hunt_channel"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import core.time : msecs;
import core.thread : Thread;
import core.stdc.stdlib : exit;
import std.parallelism : task, taskPool;
import hunt.util.queue.SimpleQueue;
void main()
{
stopStartupClock();
// hunt relies on heap-allocated queues (SimpleQueue) protected by Mutex/Condition;
// unlike event-horizon's bounded Channel!T which parks fibers on full/empty,
// SimpleQueue is an unbounded GC-allocated list with no producer backpressure.
auto queue = new SimpleQueue!int(500.msecs);
int sum = 0;
auto producer = task({
foreach (i; 1 .. 6) {
queue.push(i);
}
});
taskPool.put(producer);
producer.yieldForce(); // unbounded queue: producer finishes before any consumer
assert(!queue.isEmpty());
foreach (_; 1 .. 6) {
int item = queue.pop();
assert(item == _);
Thread.sleep(2.msecs); // simulate consumer processing
sum += item;
}
assert(sum == 15);
assert(queue.isEmpty());
writeln("consumed: ", sum);
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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 "hunt_file_read"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : File, writeln;
import std.file : write, remove, tempDir;
import std.path : buildPath;
import core.stdc.stdlib : exit;
import std.parallelism : task, taskPool;
void main()
{
stopStartupClock();
runWorker({
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);
// hunt has no async file channel because epoll cannot poll regular files.
// In hunt applications, file I/O must either block the event loop or be
// delegated to an OS worker threadpool (taskPool).
auto readTask = task({
auto f = File(path, "r");
char[128] buf;
auto slice = f.rawRead(buf[]);
return slice.idup;
});
taskPool.put(readTask);
string content = readTask.yieldForce();
assert(content == "hello from a file\n");
writeln("read ", content.length, " bytes: ", content);
});
}
// 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;
}
void runWorker(void delegate() work)
{
import hunt.util.worker.Task : Task;
import core.thread : Thread;
auto job = new class Task {
override protected void doExecute() { work(); }
};
auto thread = new Thread(&job.execute);
thread.start();
thread.join();
assert(job.isDone());
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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 "hunt_exec"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import std.process : pipeProcess, Redirect, wait;
import core.thread : Thread;
void main()
{
stopStartupClock();
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);
});
}
// Hunt Task supplies state transitions, not an owning lexical task scope.
// This adapter explicitly owns the OS thread and joins it.
void runWorker(void delegate() work)
{
import hunt.util.worker.Task : Task;
import core.thread : Thread;
auto job = new class Task {
override protected void doExecute() { work(); }
};
auto thread = new Thread(&job.execute);
thread.start();
thread.join();
assert(job.isDone());
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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: 3Streaming 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 "hunt_spawn"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
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()
{
stopStartupClock();
// 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");
});
}
// Hunt Task supplies state transitions, not an owning lexical task scope.
// This adapter explicitly owns the OS thread and joins it.
void runWorker(void delegate() work)
{
import hunt.util.worker.Task : Task;
import core.thread : Thread;
auto job = new class Task {
override protected void doExecute() { work(); }
};
auto thread = new Thread(&job.execute);
thread.start();
thread.join();
assert(job.isDone());
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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: reapedPOSIX signals through an explicit worker adapter
Hunt's Task state machine is not a POSIX signal subscription API. This adapter blocks SIGUSR1 in its worker, sends a thread-directed signal, receives it with sigwait, restores the previous mask and joins the worker. It asserts both the signal identity and the Task's completed state. No mutex, allocation or notification operation runs inside an asynchronous signal handler. Event Horizon's counterpart instead consumes masked signals through SignalFd.
#!/usr/bin/env dub
/+ dub.sdl:
name "hunt_signal"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import hunt.util.worker.Task : Task;
import core.thread : Thread;
import core.sys.posix.signal;
import core.sys.posix.pthread : pthread_self;
import std.stdio : writeln;
void main()
{
stopStartupClock();
// POSIX adapter in an explicitly joined Hunt Task, not a native Hunt signal API.
int received;
auto job = new class Task {
override protected void doExecute() {
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);
}
};
auto worker = new Thread(&job.execute);
worker.start();
worker.join();
assert(job.isDone() && received == SIGUSR1);
writeln("got signal SIGUSR1");
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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 SIGUSR1Retries 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 "hunt_retry"
dependency "hunt-net" version="0.7.1"
dependency "hunt" path=".deps/hunt"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import core.time : Duration, msecs;
import core.sync.semaphore : Semaphore;
import core.stdc.stdlib : exit;
import core.sys.posix.unistd : read;
import hunt.event.EventLoop;
import hunt.event.selector.Selector;
import hunt.util.Timer;
import hunt.net.EventLoopPool : EventLoopObjectFactory;
import hunt.Functions;
// Hunt's AbstractTimer has an upstream 4-byte buffer overflow bug reading
// timerfd (reading 8 bytes into a 32-bit uint). We subclass it safely here.
class SafeTimer : Timer {
this(Selector loop, Duration dur) {
super(loop, dur);
}
override bool readTimer(scope SimpleActionHandler handler) {
this.clearError();
ulong value;
read(this.handle, &value, 8);
this._readBuffer.data = cast(uint) value;
if (handler)
handler(this._readBuffer);
return false;
}
}
void main()
{
stopStartupClock();
EventLoop loop = EventLoopObjectFactory.buildEventLoop();
scope(exit) loop.stop();
auto sem = new Semaphore(0);
int count = 0;
int failures, value;
auto timer = new SafeTimer(loop, 10.msecs);
timer.onTick((Object sender) {
count++;
auto error = tryOperation(count, value);
if (error) { assert(error == 11); ++failures; return; }
if (count == 3) writeln("succeeded on attempt ", count);
if (count == 3) {
timer.stop();
sem.notify();
}
});
timer.start();
sem.wait();
assert(count == 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;
}
// Standalone-process workaround for the pinned Hunt DateTime daemon: its
// destructor stops but does not join. Join before creating application threads.
void stopStartupClock()
{
import core.thread : Thread;
import hunt.util.DateTime : DateTime;
auto startupThreads = Thread.getAll();
assert(startupThreads.length <= 2, "review startup ownership if dependencies change");
DateTime.stopClock();
foreach (thread; startupThreads)
if (thread !is Thread.getThis())
{
assert(thread.isDaemon);
thread.join();
}
}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.