Coming from eventcore

If you use eventDriver, typed descriptor handles, and completion callbacks, you already know the lower-level responsibilities of an event loop. Event Horizon offers both callback and fiber layers. These examples compare explicit callback state with continuations carried by suspended fibers.

The eventcore baseline is ced22593f38dd9b4514aff58a390ce313e62d283. Its driver matrix includes epoll, kqueue, WinAPI, and an experimental io_uring driver. It is a callback-based abstraction, not simply “readiness pretending to be completion.” Supported operations vary by backend in both libraries.

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.

Choose the continuation style, not a different I/O engine

The six three-way groups below show the foreign library, event-horizon fibers, and event-horizon callbacks, each with its own asserted output. Both event-horizon styles submit completion operations; callbacks are Tier A, and the scheduler builds the fiber style above that core. Callback programs here create DefaultLoop directly: they do not hide a fiber behind a delegate.

Responsibility	Tier A: callbacks	Fiber style
Continue after I/O	void function(void*, ref Completion) nothrow @nogc	Resume after a verb returns an IoResult
Keep state alive	Explicit context until terminal completion	Locals in a suspended fiber, owned by its scope
Start an operation	Check submit/submitAfter for admission failure	Check the operation's returned result
Observe completion	Completion.res is a value or negative errno	Typed IoResult!T; scope outcomes are separate
Own a buffer	Move into the op; recover/move Completion.buf	Move through a sequential I/O verb
Cancel	Request cancellation, then keep driving to terminal completion	Cancel the owning scope and join protected cleanup
Combine operations	Explicit pending count/state machine	fork/join within withScope

A submission failure produces no callback. An accepted cancellation request is not permission to free the context: the target's terminal callback still runs, and a real completion may win the race. Do not call run/runOnce from a callback. The examples keep contexts and pools alive until inFlight == 0 and print after dispatch, outside the nothrow @nogc callback boundary.

Where a callback comparison stops being equivalent

The channel, signal, capture and supervision sections retain their fiber programs. Tier A has raw read/wait/process-related operations, but no drop-in callback version of scheduler-local Channel, scoped SignalFd consumption, execCapture, or supervise. Rebuilding their buffering, cancellation, drain, and reaping policies would be an application adapter, not a second spelling of the same high-level API. No such equivalence is implied by the six native operation comparisons.

Map ownership before syntax

For an eventcore port, Tier A is the closer control-flow translation: explicit registrations become explicit submissions, and callback state remains callback state. Moving to fibers is a separate choice about where to store continuations and how to own related work.

eventcore entry point	Event Horizon callbacks	Event Horizon fibers
eventDriver.core.processEvents	DefaultLoop.run / runOnce	LoopGroup.run drives the scheduler
Timer set plus one-shot wait	submitAfter followed by explicit rearm	A loop around env.clock.sleep
Socket read/write callbacks	submit(OpRecv/OpSend, callback, ctx)	Sequential receive/send verbs
File open/read/close callbacks	OpOpenAt, OpRead, OpClose	openRoot, openFile, read
Application-owned completion count	Context with a pending count	Scoped child handles and join

The table below describes the higher-level, fiber-oriented migration specifically.

eventcore concept	Event Horizon counterpart	What changes
eventDriver.core.processEvents	LoopGroup.run	The group owns dispatch while the root scope runs.
Typed handles and callbacks	Stream, OwnedFd, direct-style verbs	A waiting fiber carries the continuation and local state.
eventDriver.timers	env.clock.sleep, Ticker	Check a returned result instead of updating state in a callback.
Handle reference management	Explicit handle cleanup and owned I/O buffers	Moving a buffer does not make TCP message-oriented.
Operation-specific result enums	IoResult!T	Failure includes errno, operation, and stage.
A higher-level task abstraction	withScope, fork, join	Child lifetime and cancellation belong to an explicit scope.

Do not mix a handle created by one driver with another runtime's I/O verbs. Move ownership of a complete connection or subsystem, including its shutdown path.

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.

The callback program explicitly rearms a relative timer after each delivery; the fiber program loops over relative sleeps. Neither proves exact timing or an absolute cadence.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_timers"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration, msecs;
import eventcore.core;
import std.stdio : writeln;

void main()
{
    int count = 0;
    auto tm = eventDriver.timers.create();
    assert(eventDriver.timers.isValid(tm));

    void onTimer(TimerID id) nothrow @safe
    {
        count++;
        try writeln("tick ", count);
        catch (Exception error) { assert(false, error.msg); }

        // eventcore's wait() only triggers once; re-arm for periodic ticks:
        if (count < 3)
        {
            eventDriver.timers.wait(id, &onTimer);
        }
        else
        {
            eventDriver.timers.stop(id);
            eventDriver.timers.releaseRef(id);
        }
    }

    eventDriver.timers.set(tm, 10.msecs, 10.msecs);
    eventDriver.timers.wait(tm, &onTimer);

    while (eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(count == 3);
}

eventcore output

tick 1
tick 2
tick 3

event-horizon fibers

#!/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);
}

event-horizon fibers output

tick 1
tick 2
tick 3

event-horizon callbacks

#!/usr/bin/env dub
/+ dub.sdl:
    name "eh_callback_timers"
    dependency "sparkles:event-horizon" path="../../../../.."
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
+/
@system:
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon.loop : DefaultLoop;
import sparkles.event_horizon.op : Completion;

struct State
{
    DefaultLoop* loop;
    int ticks;
    int[3] deliveries;
}

void tick(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.isFinal && done.res == 0);
    state.deliveries[state.ticks] = state.ticks + 1;
    if (++state.ticks < 3)
        assert(state.loop.submitAfter(10.msecs, &tick, context).hasValue);
}

void main()
{
    DefaultLoop loop;
    assert(!DefaultLoop.create(loop).hasError, "no completion backend");
    scope(exit) loop.destroy();
    State state = State(&loop);
    assert(loop.submitAfter(10.msecs, &tick, &state).hasValue);
    // State stays alive until every submitted callback has been dispatched.
    assert(!loop.run().hasError);
    assert(loop.inFlight == 0 && state.ticks == 3);
    assert(state.deliveries[] == [1, 2, 3]);
    foreach (number; state.deliveries) writeln("tick ", number);
}

event-horizon callbacks output

tick 1
tick 2
tick 3

Concurrency 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 fiber 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.

The callback program admits two operations before driving the loop, then asserts both results and a drained loop. Its counter is explicit fan-in, not a lexical child scope. The fiber program additionally proves both children have started with a ready/release handshake.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_concurrency"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration, msecs;
import eventcore.core;
import std.stdio : writeln;

struct JoinState
{
    int slowVal = 0;
    int fastVal = 0;
    int completed = 0;
    bool done = false;
}

void main()
{
    JoinState state;

    void checkDone() nothrow @safe
    {
        if (state.completed == 2)
        {
            state.done = true;
            const result = state.slowVal * 10 + state.fastVal;
            assert(result == 12);
            try writeln("joined: ", result);
            catch (Exception error) { assert(false, error.msg); }
        }
    }

    // Launch slow branch (30ms -> 1)
    auto tmSlow = eventDriver.timers.create();
    eventDriver.timers.set(tmSlow, 30.msecs, 0.msecs);
    eventDriver.timers.wait(tmSlow, (TimerID tm) nothrow @safe {
        state.slowVal = 1;
        state.completed++;
        eventDriver.timers.releaseRef(tm);
        checkDone();
    });

    // Launch fast branch (5ms -> 2)
    auto tmFast = eventDriver.timers.create();
    eventDriver.timers.set(tmFast, 5.msecs, 0.msecs);
    eventDriver.timers.wait(tmFast, (TimerID tm) nothrow @safe {
        state.fastVal = 2;
        state.completed++;
        eventDriver.timers.releaseRef(tm);
        checkDone();
    });

    while (!state.done && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(state.done);
    assert(state.slowVal == 1 && state.fastVal == 2);
}

eventcore output

joined: 12

event-horizon fibers

#!/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);
}

event-horizon fibers output

joined: 12

event-horizon callbacks

#!/usr/bin/env dub
/+ dub.sdl:
    name "eh_callback_concurrency"
    dependency "sparkles:event-horizon" path="../../../../.."
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
+/
@system:
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon.loop : DefaultLoop;
import sparkles.event_horizon.op : Completion;

struct State { uint completed; int[2] results; }
struct Child { State* owner; size_t index; }

void completed(void* context, ref Completion done) nothrow @nogc
{
    auto child = cast(Child*) context;
    assert(done.res == 0 && done.isFinal);
    assert(child.owner.results[child.index] == 0);
    child.owner.results[child.index] = cast(int) child.index + 1;
    ++child.owner.completed;
}

void main()
{
    DefaultLoop loop;
    assert(!DefaultLoop.create(loop).hasError, "no completion backend");
    scope(exit) loop.destroy();
    State state;
    Child[2] children = [Child(&state, 0), Child(&state, 1)];
    foreach (ref child; children)
        assert(loop.submitAfter(10.msecs, &completed, &child).hasValue);
    assert(state.completed == 0 && loop.inFlight == 2);
    assert(!loop.run().hasError);
    assert(loop.inFlight == 0 && state.completed == 2);
    assert(state.results[] == [1, 2]);
    // Explicit fan-in, not a structured child scope or CPU parallelism.
    writeln("joined: ", state.results[0] * 10 + state.results[1]);
}

event-horizon callbacks output

joined: 12

Cancellation, 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.

The callback deadline requests cancellation of an already-admitted long timer, observes its terminal -ECANCELED, and only then submits asynchronous cleanup. run() drains that cleanup before the context dies. This is a manually owned deadline state machine, not withDeadline or a callback-level protected scope. The fixture's long delay makes the intended cancellation path testable; real code must also handle completion winning the race.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_deadline"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration, msecs, seconds;
import eventcore.core;
import std.stdio : writeln;

void main()
{
    bool timedOut = false;
    bool cleanedUp = false;
    bool opCancelled = false;

    auto longOp = eventDriver.timers.create();
    auto deadline = eventDriver.timers.create();

    // Start a 10-second operation
    eventDriver.timers.set(longOp, 10.seconds, 0.msecs);
    eventDriver.timers.wait(longOp, (TimerID tm, bool fired) nothrow @safe {
        if (!fired)
        {
            opCancelled = true;
            try writeln("sleep returned: stopped by adapter");
            catch (Exception error) { assert(false, error.msg); }
        }
        eventDriver.timers.releaseRef(tm);
    });

    // Start a 50ms deadline watchdog timer
    eventDriver.timers.set(deadline, 50.msecs, 0.msecs);
    eventDriver.timers.wait(deadline, (TimerID tm) nothrow @safe {
        timedOut = true;
        // Explicitly cancel the long-running operation:
        eventDriver.timers.stop(longOp);
        eventDriver.timers.releaseRef(tm);

        // Manually run shielded cleanup via an auxiliary timer:
        auto cleanupTm = eventDriver.timers.create();
        eventDriver.timers.set(cleanupTm, 5.msecs, 0.msecs);
        eventDriver.timers.wait(cleanupTm, (TimerID ctm) nothrow @safe {
            cleanedUp = true;
            eventDriver.timers.releaseRef(ctm);
            try {
                writeln("timed out: ", timedOut);
                writeln("cleaned up: ", cleanedUp);
            } catch (Exception error) { assert(false, error.msg); }
        });
    });

    while (eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(opCancelled);
    assert(timedOut);
    assert(cleanedUp);
}

eventcore output

sleep returned: stopped by adapter
timed out: true
cleaned up: true

event-horizon fibers

#!/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);
}

event-horizon fibers output

sleep returned: ECANCELED
timed out: true
cleaned up: true

event-horizon callbacks

#!/usr/bin/env dub
/+ dub.sdl:
    name "eh_callback_deadline"
    dependency "sparkles:event-horizon" path="../../../../.."
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
+/
@system:
import core.time : msecs, seconds;
import core.stdc.errno : ECANCELED;
import std.stdio : writeln;
import sparkles.event_horizon.loop : DefaultLoop;
import sparkles.event_horizon.op : Completion, OpHandle;

struct State
{
    DefaultLoop* loop;
    OpHandle operation;
    bool deadlineFired, terminalSeen, cleaned;
}

void cleaned(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.res == 0 && state.terminalSeen);
    state.cleaned = true;
}

void interrupted(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.isFinal && done.res == -ECANCELED);
    assert(state.deadlineFired && !state.terminalSeen);
    state.terminalSeen = true;
    assert(state.loop.submitAfter(5.msecs, &cleaned, context).hasValue);
}

void deadline(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.res == 0 && !state.terminalSeen);
    state.deadlineFired = true;
    assert(!state.loop.cancel(state.operation).hasError);
    // cancel() is not completion: context and buffers must remain alive.
    assert(!state.terminalSeen && !state.cleaned);
}

void main()
{
    DefaultLoop loop;
    assert(!DefaultLoop.create(loop).hasError, "no completion backend");
    scope(exit) loop.destroy();
    State state = State(&loop);
    auto operation = loop.submitAfter(10.seconds, &interrupted, &state);
    assert(operation.hasValue);
    state.operation = operation.value;
    assert(loop.submitAfter(50.msecs, &deadline, &state).hasValue);
    assert(!loop.run().hasError);
    assert(loop.inFlight == 0 && state.terminalSeen && state.cleaned);
    writeln("sleep returned: ECANCELED");
    writeln("timed out: ", state.deadlineFired);
    writeln("cleaned up: ", state.cleaned);
}

event-horizon callbacks output

sleep returned: ECANCELED
timed out: true
cleaned up: true

TCP 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.

The callback implementation drives both peers through OpAccept, OpConnect, OpSend and OpRecv. Two-byte pool buffers force the five-byte frame through multiple completions. It retains unsent suffixes, accumulates receives, and asserts all four buffers returned before releasing the pool. Socket creation/bind/listen and final descriptor closure are synchronous setup/teardown.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_tcp_echo"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.sys.posix.arpa.inet : ntohs;
import core.sys.posix.netinet.in_ : sockaddr_in;
import core.sys.posix.sys.socket : sockaddr;
import core.time : Duration;
import eventcore.core;
import std.socket : InternetAddress;
import std.stdio : writeln;

void main()
{
    ushort port;
    StreamListenSocketFD listener;
    StreamSocketFD serverConn;
    StreamSocketFD clientConn;
    ubyte[64] serverBuf;
    ubyte[64] clientBuf;
    bool done = false;

    // Server accept callback
    void onAccept(StreamListenSocketFD listenSock, StreamSocketFD conn, scope RefAddress) nothrow @safe
    {
        serverConn = conn;
        eventDriver.sockets.addRef(serverConn);

        // Read into borrowed serverBuf: caller must ensure serverBuf stays alive!
        eventDriver.sockets.read(serverConn, serverBuf[0 .. 5], IOMode.all,
            (StreamSocketFD sock, IOStatus status, size_t bytesRead) nothrow @safe {
                assert(status == IOStatus.ok && bytesRead == 5);
                if (status == IOStatus.ok && bytesRead > 0)
                {
                    // Echo bytes back to client
                    eventDriver.sockets.write(sock, serverBuf[0 .. bytesRead], IOMode.all,
                        (StreamSocketFD wrSock, IOStatus wrStatus, size_t bytesWritten) nothrow @safe {
                            assert(wrStatus == IOStatus.ok && bytesWritten == 5);
                            eventDriver.sockets.shutdown(wrSock, true, true);
                            eventDriver.sockets.releaseRef(wrSock);
                        });
                }
            });
    }

    auto bindAddr = new InternetAddress("127.0.0.1", 0);
    listener = eventDriver.sockets.listenStream(bindAddr, &onAccept);
    assert(listener != StreamListenSocketFD.invalid);

    // Retrieve bound ephemeral port
    sockaddr_in sin;
    scope refAddr = new RefAddress(cast(sockaddr*)&sin, sin.sizeof);
    bool ok = eventDriver.sockets.getLocalAddress(cast(SocketFD)listener, refAddr);
    assert(ok);
    port = ntohs(sin.sin_port);

    // Client connects to server
    auto connectAddr = new InternetAddress("127.0.0.1", port);
    clientConn = eventDriver.sockets.connectStream(connectAddr, null,
        (StreamSocketFD sock, ConnectStatus status) nothrow @safe {
            assert(status == ConnectStatus.connected);
            clientConn = sock;
            eventDriver.sockets.addRef(clientConn);

            static immutable ubyte[] msg = cast(immutable(ubyte)[]) "hello";
            eventDriver.sockets.write(sock, msg, IOMode.all,
                (StreamSocketFD wrSock, IOStatus wrStatus, size_t bytesWritten) nothrow @safe {
                    assert(wrStatus == IOStatus.ok);

                    // Read echo response into clientBuf
                    assert(bytesWritten == 5);
                    eventDriver.sockets.read(wrSock, clientBuf[0 .. 5], IOMode.all,
                        (StreamSocketFD rdSock, IOStatus rdStatus, size_t bytesRead) nothrow @safe {
                            assert(rdStatus == IOStatus.ok);
                            assert(cast(const(char)[]) clientBuf[0 .. bytesRead] == "hello");
                            try writeln("echoed: ", cast(const(char)[]) clientBuf[0 .. bytesRead]);
                            catch (Exception error) { assert(false, error.msg); }

                            done = true;
                            eventDriver.sockets.shutdown(rdSock, true, true);
                            eventDriver.sockets.releaseRef(rdSock);
                            eventDriver.sockets.releaseRef(listener);
                        });
                });
        });

    while (!done && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(done);
}

eventcore output

echoed: hello

event-horizon fibers

#!/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);
}

event-horizon fibers output

echoed: hello

event-horizon callbacks

#!/usr/bin/env dub
/+ dub.sdl:
    name "eh_callback_tcp_echo"
    dependency "sparkles:event-horizon" path="../../../../.."
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
+/
@system:
import core.lifetime : move;
import core.sys.posix.sys.socket;
import core.sys.posix.netinet.in_ : sockaddr_in, INADDR_LOOPBACK;
import core.sys.posix.arpa.inet : htonl;
import core.sys.posix.unistd : close;
import std.stdio : writeln;
import sparkles.event_horizon.loop : DefaultLoop;
import sparkles.event_horizon.op;
import sparkles.event_horizon.buffer : BufferPool;

enum payload = "hello";
struct Endpoint
{
    DefaultLoop* loop;
    BufferPool!()* pool;
    int fd = -1;
    bool server;
    size_t sent, received, pendingSend;
    char[5] bytes;
}

void receive(Endpoint* endpoint) nothrow @nogc
{
    auto buffer = endpoint.pool.acquire();
    assert(buffer.hasValue);
    assert(endpoint.loop.submit(OpRecv(endpoint.fd, move(buffer.value)),
        &received, endpoint).hasValue);
}

void send(Endpoint* endpoint) nothrow @nogc
{
    auto buffer = endpoint.pool.acquire();
    assert(buffer.hasValue);
    // Two-byte buffers force the five-byte frame through several operations.
    auto remaining = payload.length - endpoint.sent;
    auto count = remaining < 2 ? remaining : 2;
    const(char)[] outgoing = endpoint.server ? endpoint.bytes[] : payload;
    buffer.value.space()[0 .. count] = cast(const(ubyte)[]) outgoing[endpoint.sent .. endpoint.sent + count];
    buffer.value.length = cast(uint) count;
    endpoint.pendingSend = count;
    assert(endpoint.loop.submit(OpSend(endpoint.fd, move(buffer.value)),
        &sent, endpoint).hasValue);
}

void sent(void* context, ref Completion done) nothrow @nogc
{
    auto endpoint = cast(Endpoint*) context;
    assert(done.res > 0 && done.res <= endpoint.pendingSend);
    endpoint.sent += done.res;
    if (endpoint.sent < payload.length) send(endpoint);
    else if (!endpoint.server) receive(endpoint);
}

void received(void* context, ref Completion done) nothrow @nogc
{
    auto endpoint = cast(Endpoint*) context;
    assert(done.res > 0 && endpoint.received + done.res <= payload.length);
    endpoint.bytes[endpoint.received .. endpoint.received + done.res] = cast(const(char)[]) done.buf[];
    endpoint.received += done.res;
    if (endpoint.received < payload.length) receive(endpoint);
    else
    {
        assert(endpoint.bytes[] == payload);
        if (endpoint.server) send(endpoint);
    }
}

void accepted(void* context, ref Completion done) nothrow @nogc
{
    auto endpoint = cast(Endpoint*) context;
    assert(done.res >= 0);
    endpoint.fd = done.res;
    receive(endpoint);
}

void connected(void* context, ref Completion done) nothrow @nogc
{
    assert(done.res == 0);
    send(cast(Endpoint*) context);
}

void main()
{
    auto listener = socket(AF_INET, SOCK_STREAM, 0);
    assert(listener >= 0);
    scope(exit) close(listener);
    sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
    assert(bind(listener, cast(sockaddr*) &address, address.sizeof) == 0);
    assert(listen(listener, 1) == 0);
    socklen_t length = address.sizeof;
    assert(getsockname(listener, cast(sockaddr*) &address, &length) == 0);
    auto clientFd = socket(AF_INET, SOCK_STREAM, 0);
    assert(clientFd >= 0);
    scope(exit) close(clientFd);
    BufferPool!() pool;
    assert(!BufferPool!().create(pool, 4, 2).hasError);
    DefaultLoop loop;
    assert(!DefaultLoop.create(loop).hasError, "no completion backend");
    scope(exit) loop.destroy();
    Endpoint server = Endpoint(&loop, &pool, -1, true);
    Endpoint client = Endpoint(&loop, &pool, clientFd);
    scope(exit) if (server.fd >= 0) close(server.fd);
    SockAddr target;
    target.storage[0 .. address.sizeof] = (cast(ubyte*) &address)[0 .. address.sizeof];
    target.len = address.sizeof;
    assert(loop.submit(OpAccept(listener), &accepted, &server).hasValue);
    assert(loop.submit(OpConnect(clientFd, target), &connected, &client).hasValue);
    assert(!loop.run().hasError);
    assert(loop.inFlight == 0 && pool.available == 4);
    assert(server.sent == 5 && client.sent == 5);
    assert(server.received == 5 && client.received == 5);
    assert(client.bytes[] == payload);
    writeln("echoed: ", client.bytes[]);
}

event-horizon callbacks output

echoed: hello

Queues, 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.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_channel"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration, msecs;
import eventcore.core;
import std.stdio : writeln;

// In eventcore, backpressure requires rolling your own ring buffer and dual EventIDs
struct BoundedQueue
{
    int[2] ring;
    size_t head;
    size_t tail;
    size_t count;
    bool closed;
    EventID hasData;
    EventID hasSpace;

    bool push(int val) @safe nothrow
    {
        if (count >= ring.length || closed) return false;
        ring[tail] = val;
        tail = (tail + 1) % ring.length;
        count++;
        eventDriver.events.trigger(hasData, false);
        return true;
    }

    bool pop(out int val) @safe nothrow
    {
        if (count == 0) return false;
        val = ring[head];
        head = (head + 1) % ring.length;
        count--;
        eventDriver.events.trigger(hasSpace, false);
        return true;
    }
}

void main()
{
    BoundedQueue q;
    q.hasData = eventDriver.events.create();
    q.hasSpace = eventDriver.events.create();

    assert(q.push(1) && q.push(2));
    assert(!q.push(3), "full adapter rejects the third item");
    int first;
    assert(q.pop(first) && first == 1);
    assert(q.push(3) && !q.push(4));
    int nextProduce = 4;
    int sum = 1;
    int expected = 2;
    bool consumerDone = false;

    // Producer callback
    void produce(EventID) nothrow @safe
    {
        while (nextProduce <= 5)
        {
            if (!q.push(nextProduce))
            {
                // Queue full: apply backpressure by waiting for space event
                eventDriver.events.wait(q.hasSpace, &produce);
                return;
            }
            nextProduce++;
        }
        q.closed = true;
        eventDriver.events.trigger(q.hasData, false);
    }

    // Consumer callback
    void consume(EventID) nothrow @safe
    {
        int item;
        while (q.pop(item))
        {
            assert(item == expected++);
            sum += item;
        }

        if (q.closed && q.count == 0)
        {
            consumerDone = true;
            try writeln("consumed: ", sum);
            catch (Exception error) { assert(false, error.msg); }
            return;
        }

        // Wait for more data
        eventDriver.events.wait(q.hasData, &consume);
    }

    // Local trigger() is synchronous and is not a stored notification. Drain
    // the prefilled queue before arming the next wait; otherwise both callbacks
    // can wait forever for an edge that happened before either was registered.
    consume(q.hasData);
    produce(q.hasSpace);

    while (!consumerDone && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    eventDriver.events.releaseRef(q.hasData);
    eventDriver.events.releaseRef(q.hasSpace);

    assert(consumerDone);
    assert(sum == 15);
    assert(expected == 6 && q.closed && q.count == 0);
}

eventcore output

consumed: 15

event-horizon fibers

#!/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);
}

event-horizon fibers output

consumed: 15

Files 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.

The callback state machine checks a missing-file -ENOENT, then opens the fixture, reads through an eight-byte buffer until zero-byte EOF, and closes through OpClose. The NUL-terminated paths stay alive until their open completions. Moving the returned buffer into the next read avoids retaining a borrowed completion slice.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_file_read"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration;
import eventcore.core;
import std.file : remove, tempDir, stdWrite = write;
import std.path : buildPath;
import std.stdio : writeln;

void main()
{
    const directory = fixtureDirectory();
    scope(exit) { import std.file : rmdir; rmdir(directory); }
    const path = buildPath(directory, "input");
    stdWrite(path, "hello from a file\n");
    scope (exit) remove(path);

    ubyte[128] buf;
    bool done = false;

    // eventcore offloads open to a worker threadpool
    eventDriver.files.open(path, FileOpenMode.read, (FileFD f, OpenStatus status) nothrow @safe {
        assert(status == OpenStatus.ok);
        assert(f != FileFD.invalid);

        // In eventcore, reading beyond file size yields readPastEOF, so clamp buffer to size:
        const sz = cast(size_t) eventDriver.files.getSize(f);
        assert(sz == 18);

        eventDriver.files.read(f, 0, buf[0 .. sz], IOMode.all, (FileFD file, IOStatus rdStatus, size_t bytesRead) nothrow @safe {
            assert(rdStatus == IOStatus.ok);
            assert(bytesRead == 18);
            assert(cast(const(char)[]) buf[0 .. bytesRead] == "hello from a file\n");

            try writeln("read ", bytesRead, " bytes: ", cast(const(char)[]) buf[0 .. bytesRead]);
            catch (Exception error) { assert(false, error.msg); }

            eventDriver.files.close(file, (FileFD clFile, CloseStatus clStatus) nothrow @safe {
                assert(clStatus == CloseStatus.ok);
                eventDriver.files.releaseRef(clFile);
                done = true;
            });
        });
    });

    while (!done && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(done);
}

// 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;
}

eventcore output

read 18 bytes: hello from a file

event-horizon fibers

#!/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);
}

event-horizon fibers output

read 18 bytes: hello from a file

event-horizon callbacks

#!/usr/bin/env dub
/+ dub.sdl:
    name "eh_callback_file_read"
    dependency "sparkles:event-horizon" path="../../../../.."
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
+/
@system:
import core.lifetime : move;
import core.stdc.errno : ENOENT;
import core.sys.posix.fcntl : AT_FDCWD, O_RDONLY;
import core.sys.posix.stdlib : mkdtemp;
import std.file : tempDir, write, remove, rmdir;
import std.path : buildPath;
import std.string : toStringz, fromStringz;
import std.stdio : writeln;
import sparkles.event_horizon.loop : DefaultLoop;
import sparkles.event_horizon.op;
import sparkles.event_horizon.buffer : BufferPool;

struct State
{
    DefaultLoop* loop;
    BufferPool!()* pool;
    const(char)* path;
    int fd = -1;
    char[18] contents;
    size_t used;
    bool missingSeen, eof, closed;
}

void closed(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.res == 0 && state.eof);
    state.fd = -1;
    state.closed = true;
}

void readNext(State* state) nothrow @nogc
{
    auto buffer = state.pool.acquire();
    assert(buffer.hasValue);
    assert(state.loop.submit(OpRead(state.fd, move(buffer.value), state.used),
        &readDone, state).hasValue);
}

void readDone(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.res >= 0, "a read error is not EOF");
    if (done.res == 0)
    {
        state.eof = true;
        assert(state.loop.submit(OpClose(state.fd), &closed, context).hasValue);
        return;
    }
    assert(state.used + done.res <= state.contents.length);
    state.contents[state.used .. state.used + done.res] = cast(const(char)[]) done.buf[];
    state.used += done.res;
    // Reuse the returned owner instead of borrowing its slice after return.
    assert(state.loop.submit(OpRead(state.fd, move(done.buf), state.used),
        &readDone, context).hasValue);
}

void opened(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.res >= 0);
    state.fd = done.res;
    readNext(state);
}

void missing(void* context, ref Completion done) nothrow @nogc
{
    auto state = cast(State*) context;
    assert(done.res == -ENOENT);
    state.missingSeen = true;
    assert(state.loop.submit(OpOpenAt(AT_FDCWD, state.path, O_RDONLY), &opened, context).hasValue);
}

void main()
{
    auto pattern = (buildPath(tempDir(), "eh-callback-file-XXXXXX") ~ '\0').dup;
    assert(mkdtemp(pattern.ptr) !is null);
    auto directory = fromStringz(pattern.ptr).idup;
    scope(exit) rmdir(directory);
    auto path = buildPath(directory, "input");
    write(path, "hello from a file\n");
    scope(exit) remove(path);
    // Both NUL-terminated paths outlive their terminal open completions.
    auto name = path.toStringz;
    auto absent = buildPath(directory, "missing").toStringz;
    BufferPool!() pool;
    assert(!BufferPool!().create(pool, 1, 8).hasError);
    DefaultLoop loop;
    assert(!DefaultLoop.create(loop).hasError, "no completion backend");
    scope(exit) loop.destroy();
    State state;
    state.loop = &loop;
    state.pool = &pool;
    state.path = name;
    assert(loop.submit(OpOpenAt(AT_FDCWD, absent, O_RDONLY), &missing, &state).hasValue);
    assert(!loop.run().hasError);
    assert(loop.inFlight == 0 && pool.available == 1);
    assert(state.missingSeen && state.eof && state.closed && state.fd == -1);
    assert(state.used == 18 && state.contents[] == "hello from a file\n");
    writeln("read ", state.used, " bytes: ", state.contents[]);
}

event-horizon callbacks output

read 18 bytes: hello from a file

Capturing 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.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_exec"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration;
import eventcore.core;
import std.stdio : writeln;

void main()
{
    ubyte[64] stdoutBuf;
    ubyte[64] stderrBuf;
    size_t stdoutLen = 0;
    size_t stderrLen = 0;
    int exitCode = -1;
    bool procDone = false;
    bool stdoutDone = false;
    bool stderrDone = false;

    auto p = eventDriver.processes.spawn(
        ["sh", "-c", "echo out; echo err >&2; exit 3"],
        ProcessStdinFile(ProcessRedirect.pipe),
        ProcessStdoutFile(ProcessRedirect.pipe),
        ProcessStderrFile(ProcessRedirect.pipe),
        null, ProcessConfig.none, null
    );
    assert(p.pid != ProcessID.invalid);

    // Close unused stdin pipe
    eventDriver.pipes.close(p.stdin, (PipeFD pfd, CloseStatus) nothrow @safe {
        eventDriver.pipes.releaseRef(pfd);
    });

    // Manually drain stdout pipe
    void readStdout(PipeFD pipe, IOStatus status, size_t bytesRead) nothrow @safe
    {
        if (status == IOStatus.ok && bytesRead > 0)
        {
            stdoutLen += bytesRead;
            eventDriver.pipes.read(pipe, stdoutBuf[stdoutLen .. $], IOMode.once, &readStdout);
        }
        else
        {
            stdoutDone = true;
            eventDriver.pipes.close(pipe, (PipeFD pfd, CloseStatus) nothrow @safe {
                eventDriver.pipes.releaseRef(pfd);
            });
        }
    }
    eventDriver.pipes.read(p.stdout, stdoutBuf[0 .. $], IOMode.once, &readStdout);

    // Manually drain stderr pipe
    void readStderr(PipeFD pipe, IOStatus status, size_t bytesRead) nothrow @safe
    {
        if (status == IOStatus.ok && bytesRead > 0)
        {
            stderrLen += bytesRead;
            eventDriver.pipes.read(pipe, stderrBuf[stderrLen .. $], IOMode.once, &readStderr);
        }
        else
        {
            stderrDone = true;
            eventDriver.pipes.close(pipe, (PipeFD pfd, CloseStatus) nothrow @safe {
                eventDriver.pipes.releaseRef(pfd);
            });
        }
    }
    eventDriver.pipes.read(p.stderr, stderrBuf[0 .. $], IOMode.once, &readStderr);

    // Wait for process termination
    eventDriver.processes.wait(p.pid, (ProcessID pid, int status) nothrow @safe {
        exitCode = status;
        procDone = true;
        eventDriver.processes.releaseRef(pid);
    });

    while ((!procDone || !stdoutDone || !stderrDone) && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(procDone && stdoutDone && stderrDone);
    assert(exitCode == 3);
    assert(cast(const(char)[]) stdoutBuf[0 .. stdoutLen] == "out\n");
    assert(cast(const(char)[]) stderrBuf[0 .. stderrLen] == "err\n");

    writeln("stdout: ", cast(const(char)[]) stdoutBuf[0 .. stdoutLen]);
    writeln("stderr: ", cast(const(char)[]) stderrBuf[0 .. stderrLen]);
    writeln("exit code: ", exitCode);
}

eventcore output

stdout: out

stderr: err

exit code: 3

event-horizon fibers

#!/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);
}

event-horizon fibers output

stdout: out

stderr: err

exit code: 3

Streaming and termination

Streaming adds a state machine: accumulate bytes into lines, observe the root, request termination, finish drains, and reap. A pipe read can split a line or contain several lines, so splitting each chunk independently is insufficient. The two source tabs expose the cleanup machinery instead of hiding it behind a printed success message.

The foreign program's manual root termination is not process-tree containment. Event Horizon reports supervision outcome and reap provenance, but its process group/cgroup tiers are not an inescapable sandbox either: descendants can race the post-spawn cgroup migration. EOF may be forced at the drain limit. See the supervision caveats.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_spawn"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.sys.posix.signal : SIGTERM;
import core.time : Duration, msecs;
import eventcore.core;
import std.stdio : writeln;

void main()
{
    ubyte[128] readBuf;
    char[128] lineBuf;
    size_t lineLen = 0;
    int linesSeen = 0;
    bool timedOut = false;
    bool procExited = false;
    bool stdoutClosed = false;
    int termSig = 0;

    auto p = eventDriver.processes.spawn(
        ["sh", "-c", "echo one; echo two; exec sleep 30"],
        ProcessStdinFile(ProcessRedirect.pipe),
        ProcessStdoutFile(ProcessRedirect.pipe),
        ProcessStderrFile(ProcessRedirect.pipe),
        null, ProcessConfig.none, null
    );
    assert(p.pid != ProcessID.invalid);

    eventDriver.pipes.close(p.stdin, (PipeFD pfd, CloseStatus) nothrow @safe {
        eventDriver.pipes.releaseRef(pfd);
    });
    eventDriver.pipes.close(p.stderr, (PipeFD pfd, CloseStatus) nothrow @safe {
        eventDriver.pipes.releaseRef(pfd);
    });

    // Schedule 100ms timeout timer
    auto timer = eventDriver.timers.create();
    // Arm only after the second complete line proves the child is ready.
    eventDriver.timers.wait(timer, (TimerID tm) nothrow @safe {
        timedOut = true;
        eventDriver.processes.kill(p.pid, SIGTERM);
        eventDriver.timers.releaseRef(tm);
    });

    // Manual line framing callback
    void readStdout(PipeFD pipe, IOStatus status, size_t bytesRead) nothrow @safe
    {
        if (status == IOStatus.ok && bytesRead > 0)
        {
            foreach (b; readBuf[0 .. bytesRead])
            {
                if (b == '\n')
                {
                    try writeln("line: ", lineBuf[0 .. lineLen]);
                    catch (Exception error) { assert(false, error.msg); }
                    linesSeen++;
                    if (linesSeen == 2) eventDriver.timers.set(timer, 100.msecs, 0.msecs);
                    lineLen = 0;
                }
                else if (lineLen < lineBuf.length)
                {
                    lineBuf[lineLen++] = cast(char) b;
                }
            }
            eventDriver.pipes.read(pipe, readBuf[], IOMode.once, &readStdout);
        }
        else
        {
            stdoutClosed = true;
            eventDriver.pipes.close(pipe, (PipeFD pfd, CloseStatus) nothrow @safe {
                eventDriver.pipes.releaseRef(pfd);
            });
        }
    }
    eventDriver.pipes.read(p.stdout, readBuf[], IOMode.once, &readStdout);

    eventDriver.processes.wait(p.pid, (ProcessID pid, int status) nothrow @safe {
        procExited = true;
        if (status < 0) termSig = -status;
        else termSig = status;
        eventDriver.processes.releaseRef(pid);
    });

    while ((!procExited || !stdoutClosed) && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(linesSeen == 2);
    assert(procExited && stdoutClosed && lineLen == 0);
    assert(timedOut);
    assert(termSig == SIGTERM);

    writeln("exited: timedOut, signaled by ", termSig);
    writeln("end: timedOut, reap: reaped");
}

eventcore output

line: one
line: two
exited: timedOut, signaled by 15
end: timedOut, reap: reaped

event-horizon fibers

#!/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);
}

event-horizon fibers output

line: one
line: two
exited: cancelled, signaled by 15
end: cancelled, reap: reaped

POSIX signals and loop notifications

Install the signal receiver before sending SIGUSR1 to this process and assert that the expected signal was observed. Linux signal masks and handler installation are process/thread integration decisions; they are not interchangeable with an ordinary callback queue. Keep signal-handler work async-signal-safe.

SignalFd receives masked POSIX signals through the event-horizon loop. A foreign notification primitive may only wake an owner loop from another thread. In particular, libasync's AsyncSignal is not a POSIX signal subscription; an OS signal bridge must supply that integration explicitly. These programs are Linux fixtures, not Windows signal portability examples.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_signal"
    dependency "eventcore" version="0.9.39"
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.sys.posix.signal : SIGUSR1, kill;
import core.sys.posix.unistd : getpid;
import core.time : Duration;
import eventcore.core;
import std.stdio : writeln;

void main()
{
    bool done = false;

    auto id = eventDriver.signals.listen(SIGUSR1, (SignalListenID lid, SignalStatus status, int sig) nothrow @safe {
        assert(status == SignalStatus.ok);
        assert(sig == SIGUSR1);
        try writeln("got signal ", sig == SIGUSR1 ? "SIGUSR1" : "other");
        catch (Exception error) { assert(false, error.msg); }
        eventDriver.signals.releaseRef(lid);
        done = true;
    });
    assert(id != SignalListenID.invalid);

    assert(kill(getpid(), SIGUSR1) == 0);

    while (!done && eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    assert(done);
}

eventcore output

got signal SIGUSR1

event-horizon fibers

#!/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);
}

event-horizon fibers output

got signal SIGUSR1

Retries 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.

The callback implementation models an operation returning EAGAIN twice, schedules 5 ms then 10 ms backoff, and asserts the third attempt produced its value. Retry state and the attempt budget are explicit. This models an ordinary dependency failure, not an assertion failure and not automatic retry of failed submit calls.

eventcore

#!/usr/bin/env dub
/+ dub.sdl:
    name "ec_retry"
    dependency "eventcore" version="0.9.39"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
    platforms "linux"
+/
import core.time : Duration, msecs;
import eventcore.core;
import std.stdio : writeln;

void main()
{
    int count = 0;
    int failures, value;
    auto tm = eventDriver.timers.create();
    assert(eventDriver.timers.isValid(tm));

    void onTimer(TimerID id) nothrow @safe
    {
        count++;
        auto error = tryOperation(count, value);
        if (error) { assert(error == 11); ++failures; }
        try if (count == 3) writeln("succeeded on attempt ", count);
        catch (Exception error) { assert(false, error.msg); }

        // eventcore's wait() only triggers once; re-arm for periodic ticks:
        if (error)
        {
            eventDriver.timers.wait(id, &onTimer);
        }
        else
        {
            eventDriver.timers.stop(id);
            eventDriver.timers.releaseRef(id);
        }
    }

    eventDriver.timers.set(tm, 10.msecs, 10.msecs);
    eventDriver.timers.wait(tm, &onTimer);

    while (eventDriver.core.waiterCount > 0)
        eventDriver.core.processEvents(Duration.max);

    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;
}

eventcore output

succeeded on attempt 3

event-horizon fibers

#!/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);
}

event-horizon fibers output

succeeded on attempt 3

event-horizon callbacks

#!/usr/bin/env dub
/+ dub.sdl:
    name "eh_callback_retry"
    dependency "sparkles:event-horizon" path="../../../../.."
    platforms "linux"
    buildType "checked" {
        buildOptions "optimize" "inline" "debugInfo"
    }
+/
@system:
import core.time : msecs;
import core.stdc.errno : EAGAIN;
import std.stdio : writeln;
import sparkles.event_horizon.loop : DefaultLoop;
import sparkles.event_horizon.op : Completion;

struct State { DefaultLoop* loop; int attempts, failures, value; }

// Deterministic dependency failure, independent of timer scheduling jitter.
int tryOperation(int attempt, out int value) @safe nothrow @nogc
{
    if (attempt < 3) return EAGAIN;
    value = 42;
    return 0;
}

void attempt(void* context, ref Completion done) nothrow @nogc
{
    assert(done.res == 0);
    auto state = cast(State*) context;
    auto error = tryOperation(++state.attempts, state.value);
    if (!error) return;
    assert(error == EAGAIN && state.attempts < 5);
    ++state.failures;
    assert(state.loop.submitAfter((5 << (state.attempts - 1)).msecs,
        &attempt, context).hasValue);
}

void main()
{
    DefaultLoop loop;
    assert(!DefaultLoop.create(loop).hasError, "no completion backend");
    scope(exit) loop.destroy();
    State state = State(&loop);
    assert(loop.submitAfter(0.msecs, &attempt, &state).hasValue);
    assert(!loop.run().hasError);
    assert(loop.inFlight == 0 && state.attempts == 3);
    assert(state.failures == 2 && state.value == 42);
    writeln("succeeded on attempt ", state.attempts);
}

event-horizon callbacks output

succeeded on attempt 3

Errors 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.