zzstoatzz.io
This repository has no description
1//! zio engine v0: a readiness loop driving per-fd handlers.
2//! no fibers, no stack switching — nothing ReleaseSafe can object to.
3//! one Engine = one loop thread; run several for thread-per-core sharding.
4const std = @import("std");
5const builtin = @import("builtin");
6const c = std.c;
7const poller_mod = @import("poller.zig");
8
9pub const Poller = poller_mod.Poller;
10pub const Event = poller_mod.Event;
11pub const Interest = poller_mod.Interest;
12
13/// registrations are rare (startup, accepts); lookups are per event batch.
14/// 0.16 has no std.Thread.Mutex (Io owns blocking sync) and the engine has
15/// no Io — a spinlock is correct and small here.
16pub const SpinLock = struct {
17 state: std.atomic.Value(bool) = .init(false),
18 pub fn lock(l: *SpinLock) void {
19 while (l.state.swap(true, .acquire)) std.atomic.spinLoopHint();
20 }
21 pub fn unlock(l: *SpinLock) void {
22 l.state.store(false, .release);
23 }
24};
25
26pub const Handler = struct {
27 ctx: ?*anyopaque,
28 /// called on the engine's loop thread. level-triggered: drain until EAGAIN.
29 cb: *const fn (ctx: ?*anyopaque, fd: c.fd_t, ev: Event) void,
30};
31
32pub const Engine = struct {
33 allocator: std.mem.Allocator,
34 poller: Poller,
35 handlers: std.AutoHashMapUnmanaged(u64, Handler),
36 handlers_lock: SpinLock,
37 stop_flag: std.atomic.Value(bool),
38 thread: ?std.Thread,
39
40 pub fn init(allocator: std.mem.Allocator) !Engine {
41 return .{
42 .allocator = allocator,
43 .poller = try Poller.init(),
44 .handlers = .empty,
45 .handlers_lock = .{},
46 .stop_flag = .init(false),
47 .thread = null,
48 };
49 }
50
51 pub fn deinit(e: *Engine) void {
52 e.stop();
53 e.handlers.deinit(e.allocator);
54 e.poller.deinit();
55 }
56
57 /// register a non-blocking fd. handler runs on the loop thread.
58 pub fn register(e: *Engine, fd: c.fd_t, interest: Interest, handler: Handler) !void {
59 {
60 e.handlers_lock.lock();
61 defer e.handlers_lock.unlock();
62 try e.handlers.put(e.allocator, @intCast(fd), handler);
63 }
64 try e.poller.add(fd, interest, @intCast(fd));
65 }
66
67 /// change interest and/or handler for an already-registered fd.
68 pub fn update(e: *Engine, fd: c.fd_t, interest: Interest, handler: Handler) !void {
69 {
70 e.handlers_lock.lock();
71 defer e.handlers_lock.unlock();
72 try e.handlers.put(e.allocator, @intCast(fd), handler);
73 }
74 try e.poller.mod(fd, interest, @intCast(fd));
75 }
76
77 /// deregister and close. safe to call from handler callbacks.
78 pub fn closeFd(e: *Engine, fd: c.fd_t) void {
79 e.poller.del(fd);
80 _ = c.close(fd);
81 e.handlers_lock.lock();
82 defer e.handlers_lock.unlock();
83 _ = e.handlers.remove(@intCast(fd));
84 }
85
86 pub fn start(e: *Engine) !void {
87 e.thread = try std.Thread.spawn(.{}, run, .{e});
88 }
89
90 pub fn stop(e: *Engine) void {
91 e.stop_flag.store(true, .release);
92 if (e.thread) |t| {
93 t.join();
94 e.thread = null;
95 }
96 }
97
98 pub fn run(e: *Engine) void {
99 var events: [poller_mod.max_batch]Event = undefined;
100 while (!e.stop_flag.load(.acquire)) {
101 const ready = e.poller.wait(&events, 50);
102 for (ready) |ev| {
103 const handler: ?Handler = blk: {
104 e.handlers_lock.lock();
105 defer e.handlers_lock.unlock();
106 break :blk e.handlers.get(ev.token);
107 };
108 if (handler) |h| h.cb(h.ctx, @intCast(ev.token), ev);
109 }
110 }
111 }
112};
113
114// --- socket helpers (ip4, non-blocking) ---------------------------------
115
116pub const wouldBlock = switch (builtin.os.tag) {
117 // EAGAIN == EWOULDBLOCK on linux and darwin
118 else => c.E.AGAIN,
119};
120
121fn sockaddrIn(octets: [4]u8, port: u16) c.sockaddr.in {
122 return .{
123 .family = c.AF.INET,
124 .port = std.mem.nativeToBig(u16, port),
125 .addr = @bitCast(octets),
126 };
127}
128
129pub fn setNonblock(fd: c.fd_t) !void {
130 const flags = c.fcntl(fd, c.F.GETFL, @as(c_int, 0));
131 if (flags < 0) return error.FcntlFailed;
132 const O_NONBLOCK: c_int = @bitCast(c.O{ .NONBLOCK = true });
133 if (c.fcntl(fd, c.F.SETFL, flags | O_NONBLOCK) < 0) return error.FcntlFailed;
134}
135
136/// blocking connect (loopback: instantaneous), then flips non-blocking.
137/// good enough for steady-state bench setup; storms use tcpConnectIp4Start.
138pub fn tcpConnectIp4(octets: [4]u8, port: u16) !c.fd_t {
139 const fd = c.socket(c.AF.INET, c.SOCK.STREAM, 0);
140 if (fd < 0) return error.SocketFailed;
141 errdefer _ = c.close(fd);
142 const sa = sockaddrIn(octets, port);
143 if (c.connect(fd, @ptrCast(&sa), @sizeOf(c.sockaddr.in)) != 0) return error.ConnectFailed;
144 try setNonblock(fd);
145 return fd;
146}
147
148/// begin a non-blocking connect. register with write interest; on the
149/// writable event call connectResult to learn the outcome.
150pub fn tcpConnectIp4Start(octets: [4]u8, port: u16) !c.fd_t {
151 const fd = c.socket(c.AF.INET, c.SOCK.STREAM, 0);
152 if (fd < 0) return error.SocketFailed;
153 errdefer _ = c.close(fd);
154 try setNonblock(fd);
155 const sa = sockaddrIn(octets, port);
156 if (c.connect(fd, @ptrCast(&sa), @sizeOf(c.sockaddr.in)) != 0) {
157 const e = c.errno(@as(isize, -1));
158 if (e != .INPROGRESS) return error.ConnectFailed;
159 }
160 return fd;
161}
162
163/// after writability on an in-progress connect: did it succeed?
164pub fn connectResult(fd: c.fd_t) !void {
165 var so_err: c_int = 0;
166 var len: c.socklen_t = @sizeOf(c_int);
167 if (c.getsockopt(fd, c.SOL.SOCKET, c.SO.ERROR, &so_err, &len) != 0) return error.GetSockOptFailed;
168 if (so_err != 0) return error.ConnectFailed;
169}
170
171pub fn tcpListenIp4(octets: [4]u8, port: u16, backlog: u31) !struct { fd: c.fd_t, port: u16 } {
172 const fd = c.socket(c.AF.INET, c.SOCK.STREAM, 0);
173 if (fd < 0) return error.SocketFailed;
174 errdefer _ = c.close(fd);
175 const one: c_int = 1;
176 _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.REUSEADDR, &one, @sizeOf(c_int));
177 const sa = sockaddrIn(octets, port);
178 if (c.bind(fd, @ptrCast(&sa), @sizeOf(c.sockaddr.in)) != 0) return error.BindFailed;
179 if (c.listen(fd, backlog) != 0) return error.ListenFailed;
180 var bound: c.sockaddr.in = undefined;
181 var len: c.socklen_t = @sizeOf(c.sockaddr.in);
182 if (c.getsockname(fd, @ptrCast(&bound), &len) != 0) return error.GetSockNameFailed;
183 try setNonblock(fd);
184 return .{ .fd = fd, .port = std.mem.bigToNative(u16, bound.port) };
185}
186
187/// accept one pending connection, non-blocking. null when drained.
188pub fn acceptNonblock(listen_fd: c.fd_t) !?c.fd_t {
189 const fd = c.accept(listen_fd, null, null);
190 if (fd < 0) {
191 if (c.errno(fd) == wouldBlock) return null;
192 return error.AcceptFailed;
193 }
194 try setNonblock(fd);
195 return fd;
196}
197
198/// read until EAGAIN; returns total bytes read. 0 with eof=false means EAGAIN
199/// hit immediately (spurious wakeup).
200pub fn drainRead(fd: c.fd_t, buf: []u8, total: *u64) enum { open, eof, err } {
201 while (true) {
202 const n = c.read(fd, buf.ptr, buf.len);
203 if (n > 0) {
204 total.* += @intCast(n);
205 continue;
206 }
207 if (n == 0) return .eof;
208 return if (c.errno(n) == wouldBlock) .open else .err;
209 }
210}
211
212/// PUMP semantics: writes `block` REPEATEDLY until the socket buffer is full
213/// (EAGAIN). this is a throughput firehose, not a send-once helper — a reader
214/// that keeps draining means this never returns. for one-shot sends use
215/// writeOnce.
216pub fn drainWrite(fd: c.fd_t, block: []const u8, total: *u64) enum { open, err } {
217 while (true) {
218 const n = c.write(fd, block.ptr, block.len);
219 if (n > 0) {
220 total.* += @intCast(n);
221 continue;
222 }
223 return if (c.errno(n) == wouldBlock) .open else .err;
224 }
225}
226
227/// single best-effort write (may be short on a full buffer).
228pub fn writeOnce(fd: c.fd_t, data: []const u8) !usize {
229 const n = c.write(fd, data.ptr, data.len);
230 if (n < 0) {
231 if (c.errno(n) == wouldBlock) return 0;
232 return error.WriteFailed;
233 }
234 return @intCast(n);
235}
236
237test "engine: loopback echo readiness round-trip" {
238 const t = std.testing;
239
240 var engine = try Engine.init(t.allocator);
241 defer engine.deinit();
242
243 const listener = try tcpListenIp4(.{ 127, 0, 0, 1 }, 0, 16);
244 defer _ = c.close(listener.fd);
245
246 const State = struct {
247 var bytes_in: std.atomic.Value(u64) = .init(0);
248 var eng: *Engine = undefined;
249
250 fn onListen(_: ?*anyopaque, fd: c.fd_t, _: Event) void {
251 while ((acceptNonblock(fd) catch null)) |conn| {
252 eng.register(conn, .{ .read = true }, .{ .ctx = null, .cb = onData }) catch unreachable;
253 }
254 }
255 fn onData(_: ?*anyopaque, fd: c.fd_t, _: Event) void {
256 var buf: [4096]u8 = undefined;
257 var total: u64 = 0;
258 _ = drainRead(fd, &buf, &total);
259 if (total > 0) _ = bytes_in.fetchAdd(total, .monotonic);
260 }
261 };
262 State.eng = &engine;
263 State.bytes_in.store(0, .monotonic);
264
265 try engine.register(listener.fd, .{ .read = true }, .{ .ctx = null, .cb = State.onListen });
266 try engine.start();
267 defer engine.stop();
268
269 const client = try tcpConnectIp4(.{ 127, 0, 0, 1 }, listener.port);
270 defer _ = c.close(client);
271 const payload = "hello from the readiness loop";
272 const written = try writeOnce(client, payload);
273 try t.expectEqual(payload.len, written);
274
275 var spins: u32 = 0;
276 while (State.bytes_in.load(.monotonic) < payload.len) : (spins += 1) {
277 if (spins > 500) return error.TimedOutWaitingForEcho;
278 const ts: c.timespec = .{ .sec = 0, .nsec = 10 * std.time.ns_per_ms };
279 _ = c.nanosleep(&ts, null);
280 }
281}