Coming from Photon
Photon makes concurrent D code look synchronous through fibers and intercepted I/O calls. Event Horizon also offers direct-style fibers, but its examples use explicit I/O capabilities and scopes instead of ambient syscall interception.
This guide uses Photon at 4cb737aa38bcde87ca400d1df3b092ad38655ad5. Its README explains the distinction between blocking, asynchronous, and pseudo-blocking I/O and demonstrates initPhoton, go, and runScheduler as the entry point.
NOTE
Each source tab imports a complete single-file DUB program. Run it with dub run --single <file> -b checked; assertions stay enabled. Each labelled output belongs to the immediately preceding implementation and is independently checked by ci --verify. See Running the examples for prerequisites, dependency versions and the validation command.
Map scheduling and ownership separately
| Photon idiom | Event Horizon idiom | Migration decision |
|---|---|---|
initPhoton / runScheduler | LoopGroup.start / group.run | Choose an explicit loop topology; default is single-threaded. |
go | sc.spawn or sc.fork | Give every child a scope that joins it. |
| Fiber-aware waiting | env.clock.sleep and explicit I/O verbs | Ordinary blocking calls do not automatically become cooperative. |
| Task coordination | JoinHandle!T, Outcome!T | Keep typed results and inspect cancellation/failure. |
| Channels | Channel!(T, capacity) | Event Horizon channels are scheduler-local, not M:N queues. |
| Blocking integration | An appropriate worker-pool boundary | A fiber alone does not make a blocking call safe for the loop. |
Do not carry over an assumption that Thread.sleep or an arbitrary C library will yield your Event Horizon fiber. Use the clock capability for delays and explicitly offload work that cannot run on the loop thread.
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 "photon_timers"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import photon;
void main()
{
initPhoton();
int ticks;
go({
foreach (i; 1 .. 4)
{
// delay() or standard Thread.sleep() via libc interception
delay(10.msecs);
writeln("tick ", i);
++ticks;
}
});
runScheduler();
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 "photon_concurrency"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import photon;
void main()
{
initPhoton();
int joinedValue;
go({
int slowResult;
int fastResult;
// Uncoordinated fibers: return types are discarded; results captured in closures
auto slow = go({
delay(30.msecs);
slowResult = 1;
});
auto fast = go({
delay(5.msecs);
fastResult = 2;
});
// Unstructured manual join; no enclosing lexical scope enforcement
slow.join();
fast.join();
joinedValue = slowResult * 10 + fastResult;
assert(joinedValue == 12);
writeln("joined: ", joinedValue);
});
runScheduler();
assert(joinedValue == 12);
}joined: 12#!/usr/bin/env dub
/+ dub.sdl:
name "eh_concurrency"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope root, ref Env env) {
ref Sched s = currentScheduler();
Channel!(int, 2) ready, release;
int finished;
auto outcome = withScope!((ref sc) {
JoinHandle!int slow, fast;
sc.fork(slow, () {
assert(!ready.put(s, 1).hasError);
assert(!release.take(s).hasError);
++finished;
return ioOk(1); // an IoResult: the typed success or IoError
});
sc.fork(fast, () {
assert(!ready.put(s, 2).hasError);
assert(!release.take(s).hasError);
++finished;
return ioOk(2);
});
// Both children have started, but neither can finish yet.
assert(!ready.take(s).hasError && !ready.take(s).hasError);
assert(finished == 0);
assert(!release.put(s, 0).hasError && !release.put(s, 0).hasError);
// Release both and join in either order; no scheduling-time assumption.
auto a = slow.join(s);
auto b = fast.join(s);
assert(a.hasValue && b.hasValue);
return a.value * 10 + b.value;
})(s);
assert(outcome.hasValue && outcome.value == 12);
assert(finished == 2);
writeln("joined: ", outcome.value);
return 0;
});
assert(run.hasValue);
}joined: 12Cancellation, deadlines and cleanup
The examples distinguish timeout detection from successful completion and assert that cleanup ran. Read the foreign operation's interruption mechanism carefully: cancelling a registration, signalling a flag and interrupting a fiber are not interchangeable. A timeout on a wait does not necessarily stop the operation being waited for.
Event Horizon's deadline latches cooperative cancellation. The parked sleep returns ECANCELED; protect lets cleanup cross checkpoints even with that latch set. Keep cleanup bounded. The foreign tab documents its own status representation rather than pretending its errors are event-horizon values.
#!/usr/bin/env dub
/+ dub.sdl:
name "photon_deadline"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs, seconds;
import std.stdio : writeln;
import photon;
// In Photon, Task.interrupt() is a no-op and there is no runtime cancellation tree.
// Timeouts must be manually handled using synchronization primitives like Condition.wait(dur).
void main()
{
initPhoton();
bool timedOut;
bool cleanedUp;
go({
auto mtx = mutex();
auto cond = condition();
string sleepResult;
mtx.lock();
// A manual timeout wrapper around a long wait:
// Photon's cond.wait(mtx, dur) returns false on timeout.
const success = cond.wait(mtx, 50.msecs);
if (!success)
{
sleepResult = "stopped by adapter";
timedOut = true;
}
else
{
sleepResult = "ok";
}
mtx.unlock();
writeln("sleep returned: ", sleepResult);
assert(sleepResult == "stopped by adapter");
// Photon has no protect!() primitive; cleanup must be explicitly scheduled
delay(5.msecs);
cleanedUp = true;
assert(timedOut);
assert(cleanedUp);
writeln("timed out: ", timedOut);
writeln("cleaned up: ", cleanedUp);
mtx.dispose();
});
runScheduler();
assert(timedOut && 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 "photon_tcp_echo"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.netinet.in_;
import core.sys.posix.sys.socket;
import core.sys.posix.unistd : close, read, write;
import std.stdio : writeln;
import photon;
void main()
{
initPhoton();
go({
// Standard POSIX socket: Photon hooks socket, bind, listen, accept, read, write
int serverFd = socket(AF_INET, SOCK_STREAM, 0);
assert(serverFd >= 0);
scope (exit) close(serverFd);
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_port = 0; // ephemeral port
addr.sin_addr.s_addr = 0x0100007F; // loopback on this Linux little-endian fixture
assert(bind(serverFd, cast(sockaddr*)&addr, addr.sizeof) == 0);
assert(listen(serverFd, 1) == 0);
socklen_t len = addr.sizeof;
assert(getsockname(serverFd, cast(sockaddr*)&addr, &len) == 0);
// Server fiber: accepts and echoes data using borrowed slices
go({
sockaddr_in clientAddr;
socklen_t clientLen = clientAddr.sizeof;
int connFd = accept(serverFd, cast(sockaddr*)&clientAddr, &clientLen);
if (connFd >= 0)
{
scope (exit) close(connFd);
ubyte[5] srvBuf;
// Transparent pseudo-blocking read; buffer is borrowed in-place
readExact(connFd, srvBuf[]);
writeAll(connFd, srvBuf[]);
}
});
// Client fiber: connects and sends message
int clientFd = socket(AF_INET, SOCK_STREAM, 0);
assert(clientFd >= 0);
scope (exit) close(clientFd);
sockaddr_in targetAddr;
targetAddr.sin_family = AF_INET;
targetAddr.sin_port = addr.sin_port;
targetAddr.sin_addr.s_addr = 0x0100007F; // 127.0.0.1
assert(connect(clientFd, cast(sockaddr*)&targetAddr, targetAddr.sizeof) == 0);
const char[] msg = "hello";
writeAll(clientFd, cast(const(ubyte)[]) msg);
ubyte[5] recvBuf;
readExact(clientFd, recvBuf[]);
const received = recvBuf.length;
assert(received == msg.length);
assert(cast(const(char)[]) recvBuf[0 .. received] == "hello");
writeln("echoed: ", cast(const(char)[]) recvBuf[0 .. received]);
});
runScheduler();
}
void readExact(int fd, ubyte[] buffer)
{
import core.stdc.errno : errno, EINTR;
size_t offset;
while (offset < buffer.length)
{
const n = read(fd, buffer.ptr + offset, buffer.length - offset);
if (n < 0 && errno == EINTR) continue;
assert(n > 0);
offset += n;
}
}
void writeAll(int fd, const(ubyte)[] buffer)
{
import core.stdc.errno : errno, EINTR;
size_t offset;
while (offset < buffer.length)
{
const n = write(fd, buffer.ptr + offset, buffer.length - offset);
if (n < 0 && errno == EINTR) continue;
assert(n > 0);
offset += n;
}
}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 "photon_channel"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import photon;
void main()
{
initPhoton();
int totalConsumed;
go({
// Channel with capacity 2: backpressure parks the producer on the 3rd put
auto items = channel!int(2);
int produced;
bool attemptingThird;
// Producer fiber: pushes 1..6 and closes the channel
goOnSameThread({
foreach (i; 1 .. 6)
{
if (i == 3) attemptingThird = true;
items.put(i);
produced = i;
}
items.close();
});
while (!attemptingThird) yield();
assert(produced == 2);
int sum;
int expected = 1;
// Consumer fiber: InputRange contract iterates until channel is closed & empty
foreach (val; items)
{
assert(val == expected++);
delay(2.msecs); // simulate slow consumer
sum += val;
}
totalConsumed = sum;
assert(totalConsumed == 15);
assert(produced == 5 && expected == 6);
writeln("consumed: ", totalConsumed);
});
runScheduler();
assert(totalConsumed == 15);
}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 "photon_file_read"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.fcntl : open, O_RDONLY;
import core.sys.posix.unistd : close, read;
import std.file : remove, tempDir, write;
import std.path : buildPath;
import std.stdio : writeln;
import std.string : toStringz;
import photon;
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);
initPhoton();
go({
// Photon intercepts open() and read(). Because regular files return EPERM
// on epoll_ctl, Photon marks the FD as THREADPOOL and offloads blocking
// libc read() calls to an internal background worker thread pool.
int fd = open(path.toStringz, O_RDONLY);
assert(fd >= 0);
scope (exit) close(fd);
ubyte[128] buf;
ssize_t n = read(fd, buf.ptr, buf.length);
assert(n == 18);
assert(cast(const(char)[]) buf[0 .. n] == "hello from a file\n");
writeln("read ", n, " bytes: ", cast(const(char)[]) buf[0 .. n]);
});
runScheduler();
}
// 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 fileStreaming 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.
#!/usr/bin/env dub
/+ dub.sdl:
name "photon_spawn"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.sys.posix.signal : SIGTERM, kill;
import core.time : msecs;
import std.process : Config, pipeProcess, Redirect, wait;
import std.stdio : writeln;
import photon;
void main()
{
initPhoton();
go({
// Photon lacks native process supervision. Streaming output and enforcing timeouts
// require manual pipes and auxiliary watchdog threads offloaded from the scheduler.
auto res = offload({
auto pipes = pipeProcess(["sh", "-c", "echo one; echo two; exec sleep 30"], Redirect.stdout);
import core.thread : Thread;
string[] lines;
foreach (line; pipes.stdout.byLine)
{
writeln("line: ", line);
lines ~= line.idup;
if (lines.length == 2) assert(kill(pipes.pid.osHandle, SIGTERM) == 0);
}
int code = wait(pipes.pid);
assert(code == -SIGTERM);
assert(lines == ["one", "two"]);
return lines;
});
assert(res.length == 2);
writeln("exited: SIGTERM");
writeln("end: reaped");
});
runScheduler();
}line: one
line: two
exited: SIGTERM
end: reaped#!/usr/bin/env dub
/+ dub.sdl:
name "eh_spawn"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import core.time : msecs;
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
ref Sched s = currentScheduler();
SupervisedProcessConfig cfg;
cfg.terminateGrace = 500.msecs; // cancellation: TERM, then bounded KILL grace
cfg.process.stdinSpec = StdioSpec(StdioMode.nullDev);
int linesSeen;
SupervisedProcessResult result;
bool returned;
auto scopeResult = withScope!((ref outer) {
JoinHandle!void task;
outer.fork(task, () {
auto got = supervise(s, ["sh", "-c", "echo one; echo two; exec sleep 30"], cfg,
null, (in ProcessEvent ev) {
final switch (ev.kind)
{
case ProcessEventKind.line:
assert(cast(const(char)[]) ev.line.bytes == (linesSeen == 0 ? "one" : "two"));
writeln("line: ", cast(const(char)[]) ev.line.bytes);
++linesSeen;
if (linesSeen == 2) outer.cancel(); // readiness handshake, no startup-time race
break;
case ProcessEventKind.exited:
writeln("exited: ", ev.end, ", signaled by ",
ev.status.signaled ? ev.status.code : 0);
break;
case ProcessEventKind.sample:
break; // periodic resource samples; not printed here
}
});
assert(got.hasValue);
result = got.value;
returned = true;
return ioOk();
});
task.join(s);
})(s);
assert(!scopeResult.hasError && returned);
assert(linesSeen == 2);
assert(result.end == ProcessEnd.cancelled);
assert(result.reap == ReapOutcome.reaped);
assert(result.status.signaled && result.status.code == 15);
assert(!result.eofForced);
writeln("end: ", result.end, ", reap: ", result.reap);
return 0;
});
assert(run.hasValue);
}line: one
line: two
exited: cancelled, signaled by 15
end: cancelled, reap: reapedPOSIX signals 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 "photon_signal"
dependency "photon" version="0.19.3"
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 photon;
initPhoton();
bool finished;
go({ offload({ work(); return 0; }); finished = true; });
runScheduler();
assert(finished);
}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 both child streams
Photon's offload supplies the worker boundary; Phobos supplies process and pipe management. The adapter drains stderr on a second thread while reading stdout, joins that drain, waits for the root and asserts both streams and exit code 3. This is application integration, not a Photon process-supervisor API. Event Horizon's capture owns this coordination. Real applications should add output bounds and cancellation policies rather than assuming a child will produce only this small fixture.
#!/usr/bin/env dub
/+ dub.sdl:
name "photon_exec"
dependency "photon" version="0.19.3"
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 photon;
initPhoton();
bool finished;
go({ offload({ work(); return 0; }); finished = true; });
runScheduler();
assert(finished);
}stdout: out
stderr: err
exit code: 3#!/usr/bin/env dub
/+ dub.sdl:
name "eh_exec"
dependency "sparkles:event-horizon" path="../../../../.."
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import std.stdio : writeln;
import sparkles.event_horizon;
void main()
{
LoopGroup group;
if (LoopGroup.start(group).hasError)
assert(false, "no completion backend");
scope (exit) group.shutdown();
auto run = group.run((ref RootScope sc, ref Env env) {
ref Sched s = currentScheduler();
ProcessConfig cfg;
cfg.stderrSpec = StdioSpec(StdioMode.pipe);
auto got = capture(s, ["sh", "-c", "echo out; echo err >&2; exit 3"], cfg);
assert(got.hasValue);
assert(got.value.status.code == 3);
assert(cast(const(char)[]) got.value.stdout_[] == "out\n");
assert(cast(const(char)[]) got.value.stderr_[] == "err\n");
writeln("stdout: ", cast(const(char)[]) got.value.stdout_[]);
writeln("stderr: ", cast(const(char)[]) got.value.stderr_[]);
writeln("exit code: ", got.value.status.code);
return 0;
});
assert(run.hasValue);
}stdout: out
stderr: err
exit code: 3Retries 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 "photon_retry"
dependency "photon" version="0.19.3"
buildType "checked" {
buildOptions "optimize" "inline" "debugInfo"
}
platforms "linux"
+/
import photon;
import core.time : msecs;
import std.stdio : writeln;
void main()
{
initPhoton();
int attempts, value;
go({
while (attempts < 5)
{
++attempts;
const transientFailure = attempts < 3;
if (!transientFailure) { value = attempts; break; }
delay((5 << (attempts - 1)).msecs);
}
});
runScheduler();
assert(attempts == 3 && value == 3);
writeln("succeeded on attempt ", value);
}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.