Coming from libasync
Libasync attaches asynchronous objects to an EventLoop and reports results through callbacks. Event Horizon's fiber layer keeps the continuation in the calling function: an I/O verb parks, then returns a result and its buffer.
The libasync baseline is 20c29c20cecd367ff38bf4cf9d7f210cd97c04ac. Its README explicitly describes thread-pool file and DNS operations, one event loop per thread, and manual error management. Those are useful migration boundaries, not reasons to assume every backend or workload behaves identically.
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.
Translate notification semantics carefully
| libasync concept | Event Horizon counterpart | Caveat |
|---|---|---|
EventLoop and its dispatch | LoopGroup.run | The default group owns one calling-thread scheduler. |
| Socket callback | accept, recv, send in a fiber | Partial transfers still need loops. |
AsyncTimer | env.clock.sleep, Ticker | Choose a delay or absolute cadence. |
| Thread-pool file callback | openRoot, openFile, read | Files are reached through a directory capability; env.fs runs each call on the ring or a worker. |
AsyncSignal | A cross-thread notification design | It is not equivalent to SignalFd. |
| Callback status | IoResult!T | Check errors explicitly instead of treating all failures as EOF. |
The distinction in the signal row matters: libasync's AsyncSignal enqueues a callback on its owner's loop when triggered from another thread. Event Horizon's Linux SignalFd receives POSIX signals such as SIGUSR1. Likewise, a scheduler-local Channel must not be presented as a drop-in cross-thread notification mechanism.
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 "libasync_timers"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import libasync;
void main()
{
auto evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
int ticks = 0;
auto timer = new AsyncTimer(evl);
timer.periodic().duration(10.msecs).run({
ticks++;
writeln("tick ", ticks);
if (ticks >= 3)
timer.kill();
});
while (ticks < 3)
evl.loop();
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 "libasync_concurrency"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import libasync;
void main()
{
auto evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
int a = 0;
int b = 0;
bool doneA = false;
bool doneB = false;
// Dispatch slow operation (30ms)
auto slow = new AsyncTimer(evl);
slow.duration(30.msecs).run({
a = 1;
doneA = true;
slow.kill();
});
// Dispatch fast operation (5ms)
auto fast = new AsyncTimer(evl);
fast.duration(5.msecs).run({
b = 2;
doneB = true;
fast.kill();
});
// Manual event loop crank until both callbacks complete
while (!doneA || !doneB)
evl.loop();
const joined = a * 10 + b;
assert(doneA && doneB);
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 "libasync_deadline"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs, seconds;
import std.stdio : writeln;
import libasync;
void main()
{
auto evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
bool timedOut = false;
bool cleanedUp = false;
bool cancelled = false;
// The long-running operation (10s)
auto slow = new AsyncTimer(evl);
slow.duration(10.seconds).run({
assert(false, "Should not fire");
});
// Cleanup timer (simulating protected cleanup)
AsyncTimer cleanupTimer;
// The deadline timer (50ms)
auto deadline = new AsyncTimer(evl);
deadline.duration(50.msecs).run({
timedOut = true;
// In libasync, cancellation is manual: kill the pending timer
slow.kill();
cancelled = true;
writeln("sleep returned: ", cancelled ? "stopped by adapter" : "ok");
// Protected cleanup must be manually scheduled and tracked
cleanupTimer = new AsyncTimer(evl);
cleanupTimer.duration(5.msecs).run({
cleanedUp = true;
cleanupTimer.kill();
});
deadline.kill();
});
while (!cleanedUp)
evl.loop();
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 "libasync_tcp_echo"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import libasync;
void main()
{
auto evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
AsyncTCPConnection serverConn;
string echoedMsg;
bool clientFinished = false;
ubyte[] pending;
size_t serverSent, clientSent;
void flush(AsyncTCPConnection connection, const(ubyte)[] data, ref size_t offset)
{
while (offset < data.length)
{
const n = connection.send(data[offset .. $]);
if (n == 0) break; // resume on WRITE; ERROR is handled by the callback
assert(n <= data.length - offset);
offset += n;
}
}
// Server listener
auto listener = new AsyncTCPListener(evl);
void delegate(TCPEvent) serverHandler(AsyncTCPConnection conn)
{
serverConn = conn;
return (TCPEvent ev) {
if (ev == TCPEvent.READ)
{
ubyte[] buf = new ubyte[64];
const len = serverConn.recv(buf);
if (len > 0)
{
pending ~= buf[0 .. len];
flush(serverConn, pending, serverSent);
}
}
else if (ev == TCPEvent.WRITE) flush(serverConn, pending, serverSent);
else if (ev == TCPEvent.ERROR) assert(false, "server I/O failed");
};
}
const listenOk = listener.host("127.0.0.1", 0).run(&serverHandler);
assert(listenOk);
const port = listener.local.port;
// Client connection
auto client = new AsyncTCPConnection(evl);
client.peer = evl.resolveIP("127.0.0.1", port);
client.run((TCPEvent ev) {
final switch (ev)
{
case TCPEvent.CONNECT:
flush(client, cast(const(ubyte)[]) "hello", clientSent);
break;
case TCPEvent.READ:
ubyte[] buf = new ubyte[64];
const len = client.recv(buf);
if (len > 0)
{
echoedMsg ~= cast(string) buf[0 .. len].idup;
assert(echoedMsg.length <= 5);
clientFinished = echoedMsg.length == 5;
}
break;
case TCPEvent.WRITE:
flush(client, cast(const(ubyte)[]) "hello", clientSent);
break;
case TCPEvent.ERROR:
assert(false, "client I/O failed");
case TCPEvent.CLOSE:
assert(clientFinished, "connection closed before the complete reply");
break;
case TCPEvent.DESTROY:
break;
}
});
while (!clientFinished)
evl.loop();
assert(clientSent == 5 && serverSent == 5);
client.kill();
if (serverConn)
serverConn.kill();
listener.kill();
assert(echoedMsg == "hello");
writeln("echoed: ", echoedMsg);
}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 "libasync_channel"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import libasync;
void main()
{
auto evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
// libasync lacks built-in channels with backpressure.
// Developers had to hand-roll bounded ring buffers with manual notifier signaling.
int[2] buffer;
size_t count = 0;
int nextItem = 1;
int sum = 0;
bool done = false;
auto notifier = new AsyncNotifier(evl);
// Consumer step: runs on the event loop when triggered
notifier.run({
if (count > 0)
{
// Take one item (slow consumer)
sum += buffer[0];
buffer[0] = buffer[1];
count--;
// Produce more if space is available
while (count < 2 && nextItem <= 5)
{
buffer[count++] = nextItem++;
}
if (count == 0 && nextItem > 5)
{
done = true;
notifier.kill();
return;
}
// Trigger next consumption iteration
notifier.trigger();
}
});
// Fill initial bounded capacity (2 items)
while (count < 2 && nextItem <= 5)
buffer[count++] = nextItem++;
assert(count == 2 && nextItem == 3, "bounded adapter admits only two items before dispatch");
notifier.trigger();
while (!done)
evl.loop();
assert(sum == 15);
writeln("consumed: ", sum);
}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 "libasync_file_read"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.file : remove, tempDir, write;
import std.path : buildPath;
import std.stdio : writeln;
import libasync;
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 evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
auto file = new shared AsyncFile(evl);
bool done = false;
size_t bytesRead = 0;
string content;
file.onReady({
auto buf = cast(ubyte[]) file.buffer;
bytesRead = buf.length;
content = cast(string) buf.idup;
done = true;
file.kill();
}).read(path, 18);
while (!done)
evl.loop();
assert(bytesRead == 18);
assert(content == "hello from a file\n");
writeln("read ", bytesRead, " 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;
}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 filePOSIX 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 "libasync_signal"
dependency "libasync" version="0.9.8"
platforms "linux"
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 libasync;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) { loop.exit(); loop.destroy(); }
auto notifier = new AsyncNotifier(loop);
bool delivered;
notifier.run({ delivered = true; notifier.kill(); });
auto worker = new Thread({ work(); notifier.trigger(); });
worker.start();
while (!delivered) loop.loop();
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 SIGUSR1Capturing 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 "libasync_exec"
dependency "libasync" version="0.9.8"
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 libasync;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) { loop.exit(); loop.destroy(); }
auto notifier = new AsyncNotifier(loop);
bool delivered;
notifier.run({ delivered = true; notifier.kill(); });
auto worker = new Thread({ work(); notifier.trigger(); });
worker.start();
while (!delivered) loop.loop();
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: 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 "libasync_spawn"
dependency "libasync" version="0.9.8"
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 libasync;
import core.thread : Thread;
auto loop = new EventLoop();
scope(exit) { loop.exit(); loop.destroy(); }
auto notifier = new AsyncNotifier(loop);
bool delivered;
notifier.run({ delivered = true; notifier.kill(); });
auto worker = new Thread({ work(); notifier.trigger(); });
worker.start();
while (!delivered) loop.loop();
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 "libasync_retry"
dependency "libasync" version="0.9.8"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import libasync;
void main()
{
auto evl = new EventLoop;
scope (exit) { evl.exit(); evl.destroy(); }
int ticks = 0;
int failures, value;
auto timer = new AsyncTimer(evl);
timer.periodic().duration(10.msecs).run({
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.kill();
});
while (ticks < 3)
evl.loop();
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.