tickets

148.3 KB

1. // hl:http1 plugin — HTTP/1.1 server with epoll + I/O thread pool
2. // Compiled to libhttp1.so, loaded by runtime via dlopen
3. //
4. // Architecture:
5. // Acceptor thread (epoll on server fd) → PARK (epoll) → I/O thread pool (TLS + parse)
6. // → request queue → Hybriel main thread
7. //
8. // An IDLE connection never occupies an I/O worker (mission 084). A worker's read() is
9. // blocking, so a connection handed straight to the pool pins a thread until bytes arrive;
10. // with the default `threads = 4`, four idle keep-alive sockets (one open browser tab is
11. // already several) starved every later request. Instead every connection that is not
12. // known to have readable bytes is PARKED in the parker thread's epoll, and only enters
13. // conn_queue when it is actually readable (or hung up). See ParkedConns below.
14. //
15. // Exports:
16. // hl_http1_create_server(port, host, cert_path, key_path, threads) → iterator of request objects
17. // hl_http1_listen(port) → same with defaults (backward compat)
18. //
19. // Each request object: method, path, query (object), headers (object), body, respond (handle),
20. // remoteAddress, bytes (the body as a Bytes, ticket #88)
21. // Respond handle: call("send", status_code, body [, content_type]) → writes response
23. const std = @import("std");
24. const api = @import("plugin_api");
25. const http = @import("http_common");
26. const ws = @import("ws_common");
27. // The outgoing WebSocket client's TLS (wss://, ticket #109) — the SAME client
28. // half hl:fetch uses (system trust store, hostname verification on, an
29. // optional extra `caFile` for a test fixture). Aliased away from `TlsContext`:
30. // the name is taken below by http1's own pre-existing SERVER-only TLS type,
31. // which this does not touch.
32. const tls_client = @import("tls_common");
34. const HlValue = api.HlValue;
35. const HlObject = api.HlObject;
36. const HlField = api.HlField;
37. const HlIterator = api.HlIterator;
38. const HlHandle = api.HlHandle;
39. const HlString = api.HlString;
41. const linux = std.os.linux;
42. const posix = std.posix;
43. const c = std.c;
44. const PthreadMutex = c.pthread_mutex_t;
45. const PthreadCond = c.pthread_cond_t;
47. fn mutexInit(m: *PthreadMutex) void {
48. // Static initializer is sufficient; explicit init not exposed in this std.c.
49. _ = m;
50. }
51. fn mutexLock(m: *PthreadMutex) void {
52. _ = c.pthread_mutex_lock(m);
53. }
54. fn mutexUnlock(m: *PthreadMutex) void {
55. _ = c.pthread_mutex_unlock(m);
56. }
57. fn condInit(cnd: *PthreadCond) void {
58. _ = cnd;
59. }
60. fn condSignal(cnd: *PthreadCond) void {
61. _ = c.pthread_cond_signal(cnd);
62. }
63. fn condBroadcast(cnd: *PthreadCond) void {
64. _ = c.pthread_cond_broadcast(cnd);
65. }
66. fn condWait(cnd: *PthreadCond, m: *PthreadMutex) void {
67. _ = c.pthread_cond_wait(cnd, m);
68. }
70. // allocations cross the acceptor, I/O-worker and interpreter threads: the plugins'
71. // allocator (plugin_api.zig), never the debug one, whose bookkeeping segfaulted in its
72. // own free() under request churn (mission 027)
73. const allocator = api.allocator;
75. // Use direct syscall for stderr writes — std.debug.print uses std.Progress
76. // which has ABI-incompatible global state when loaded as a plugin into a
77. // binary compiled with a different Zig version.
78. fn logMsg(msg: []const u8) void {
79. _ = linux.write(2, msg.ptr, msg.len);
80. }
82. fn logFmt(comptime fmt: []const u8, args: anytype) void {
83. var buf: [512]u8 = undefined;
84. const s = std.fmt.bufPrint(&buf, fmt, args) catch return;
85. logMsg(s);
86. }
88. // =========================================================================
89. // SSE (Server-Sent Events) push channel — mission 031, ADDRESSED in 080
90. // A registry of connected event-stream clients. `sse_start` (on the respond
91. // handle) upgrades a GET request into a long-lived text/event-stream, assigns
92. // the connection a stable id and registers it; `hl_http1_sse_broadcast` writes
93. // a `data:` frame to every registered client and `hl_http1_sse_send` to ONE of
94. // them, dropping any that error (peer closed). Touched only from the
95. // interpreter (main) thread, but guarded by a mutex for safety.
96. //
97. // Mission 080 (D26): the id is a MONOTONIC counter, never the fd — a closed
98. // connection's fd is recycled by the kernel within milliseconds, so an fd-keyed
99. // push would eventually land on a stranger's socket. `sse_start` returns the id
100. // (it used to return the raw fd, which no caller used), and the framework sends
101. // it to the browser as the `__hlHello` event so the client can name itself.
102. // =========================================================================
103. // Mission 084: a subscription is reaped PROACTIVELY, not on the first failed push.
104. // A closed tab used to stay in this registry until something happened to be pushed to
105. // it — and because a scoped push (§9.5) sends nothing to a client whose needs did not
106. // change, "something" could be never. Two mechanisms, both in the reaper thread:
107. //
108. // • EPOLLRDHUP on every registered fd — a closed tab sends FIN, which fires
109. // immediately and costs no traffic at all. This is the primary detector.
110. // • a periodic SSE COMMENT heartbeat (`:\n\n`, which EventSource ignores) — catches
111. // a peer that vanished WITHOUT a FIN (killed machine, dropped NAT entry), which no
112. // amount of epolling can see, and keeps intermediaries from timing the stream out.
113. //
114. // A heartbeat is also why one failed write is enough to declare death here: the probe
115. // runs repeatedly, so the "first write after FIN succeeds, second gets EPIPE" TCP
116. // behaviour just means the peer is reaped one tick later.
117. const MSG_NOSIGNAL: u32 = 0x4000;
118. const SseConn = struct { id: u32, fd: i32 };
119. var sse_mutex: PthreadMutex = c.PTHREAD_MUTEX_INITIALIZER;
120. var sse_subs: std.ArrayListUnmanaged(SseConn) = .empty;
121. var sse_next_id: u32 = 1;
123. /// how often the reaper writes its comment heartbeat / probes for silent death
124. const SSE_HEARTBEAT_MS: i64 = 5_000;
126. var sse_epoll_fd: i32 = -1;
127. var sse_wake_fd: i32 = -1;
128. var sse_reaper_thread: ?std.Thread = null;
129. var sse_reaper_running = std.atomic.Value(bool).init(false);
131. /// Start the reaper thread + its epoll. Idempotent; called under sse_mutex from the
132. /// first sseRegister(), so a server that never opens a stream never spawns it.
133. fn sseReaperEnsureLocked() void {
134. if (sse_reaper_running.load(.acquire)) return;
135. const ep_rc = linux.epoll_create1(linux.EPOLL.CLOEXEC);
136. const ep: i32 = @bitCast(@as(u32, @truncate(ep_rc)));
137. if (ep < 0) return;
138. const ef_rc = linux.eventfd(0, linux.EFD.CLOEXEC | linux.EFD.NONBLOCK);
139. const ef: i32 = @bitCast(@as(u32, @truncate(ef_rc)));
140. if (ef < 0) {
141. _ = linux.close(ep);
142. return;
143. }
144. var wake_ev = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .fd = ef } };
145. _ = linux.epoll_ctl(ep, linux.EPOLL.CTL_ADD, ef, &wake_ev);
146. sse_epoll_fd = ep;
147. sse_wake_fd = ef;
148. sse_reaper_running.store(true, .release);
149. sse_reaper_thread = std.Thread.spawn(.{}, sseReaperLoop, .{}) catch {
150. sse_reaper_running.store(false, .release);
151. _ = linux.close(ep);
152. _ = linux.close(ef);
153. sse_epoll_fd = -1;
154. sse_wake_fd = -1;
155. return;
156. };
157. }
159. /// Drop subscription at index `i`: epoll DEL, close, remove. Caller holds sse_mutex.
160. fn sseDropLocked(i: usize) void {
161. const fd = sse_subs.items[i].fd;
162. if (sse_epoll_fd >= 0) _ = linux.epoll_ctl(sse_epoll_fd, linux.EPOLL.CTL_DEL, fd, null);
163. _ = linux.close(fd);
164. _ = sse_subs.swapRemove(i);
165. }
167. /// The reaper: EPOLLRDHUP wakes it the instant a tab closes; the 1s timeout paces the
168. /// heartbeat. Nothing here pushes application data, so a reap needs no traffic from the
169. /// app at all — which is the whole point (mission 080's gap).
170. fn sseReaperLoop() void {
171. var events: [64]linux.epoll_event = undefined;
172. var last_beat: i64 = monotonicMs();
173. while (sse_reaper_running.load(.acquire)) {
174. const n_rc = linux.epoll_wait(sse_epoll_fd, &events, events.len, 1000);
175. const n: i32 = @bitCast(@as(u32, @truncate(n_rc)));
176. if (n > 0) {
177. for (events[0..@intCast(n)]) |ev| {
178. if (ev.data.fd == sse_wake_fd) {
179. var drain: u64 = 0;
180. _ = linux.read(sse_wake_fd, @ptrCast(&drain), @sizeOf(u64));
181. continue;
182. }
183. // A subscriber never SENDS on its stream, so any readable/hangup event
184. // means the peer went away (or is misbehaving) — either way it is dead.
185. mutexLock(&sse_mutex);
186. for (sse_subs.items, 0..) |conn, i| {
187. if (conn.fd == ev.data.fd) {
188. sseDropLocked(i);
189. break;
190. }
191. }
192. mutexUnlock(&sse_mutex);
193. }
194. }
195. const now = monotonicMs();
196. if (now - last_beat >= SSE_HEARTBEAT_MS) {
197. last_beat = now;
198. mutexLock(&sse_mutex);
199. var i: usize = 0;
200. while (i < sse_subs.items.len) {
201. // ":\n\n" is an SSE comment — EventSource ignores it, so this is a pure
202. // liveness probe that never reaches an `onmessage` handler.
203. if (!sseRawWrite(sse_subs.items[i].fd, ":\n\n")) {
204. sseDropLocked(i);
205. continue;
206. }
207. i += 1;
208. }
209. mutexUnlock(&sse_mutex);
210. }
211. }
212. }
214. // Write via sendto with MSG_NOSIGNAL so a dead peer yields EPIPE instead of
215. // killing the process with SIGPIPE. Returns false on any short/failed write.
216. fn sseRawWrite(fd: i32, data: []const u8) bool {
217. var written: usize = 0;
218. while (written < data.len) {
219. const rc = linux.sendto(fd, data[written..].ptr, data.len - written, MSG_NOSIGNAL, null, 0);
220. const n: isize = @bitCast(rc);
221. if (n <= 0) return false;
222. written += @intCast(n);
223. }
224. return true;
225. }
227. fn sseRegister(fd: i32) u32 {
228. mutexLock(&sse_mutex);
229. defer mutexUnlock(&sse_mutex);
230. const id = sse_next_id;
231. sse_next_id += 1;
232. sse_subs.append(allocator, .{ .id = id, .fd = fd }) catch return 0;
233. // Watch it for hangup from now on — see the reaper comment above.
234. sseReaperEnsureLocked();
235. if (sse_epoll_fd >= 0) {
236. var ev = linux.epoll_event{
237. .events = linux.EPOLL.RDHUP | linux.EPOLL.IN,
238. .data = .{ .fd = fd },
239. };
240. _ = linux.epoll_ctl(sse_epoll_fd, linux.EPOLL.CTL_ADD, fd, &ev);
241. }
242. return id;
243. }
245. // One `data:` frame on one socket. Caller holds sse_mutex.
246. fn sseWriteFrameLocked(fd: i32, msg: []const u8) bool {
247. var ok = sseRawWrite(fd, "data: ");
248. if (ok) ok = sseRawWrite(fd, msg);
249. if (ok) ok = sseRawWrite(fd, "\n\n");
250. return ok;
251. }
253. // Broadcast one SSE message to all subscribers. `msg` should be a single line
254. // (callers send single-line JSON). Dead sockets are closed and removed.
255. // Returns the number of clients successfully written to.
256. fn sseBroadcast(msg: []const u8) usize {
257. mutexLock(&sse_mutex);
258. defer mutexUnlock(&sse_mutex);
259. var count: usize = 0;
260. var i: usize = 0;
261. while (i < sse_subs.items.len) {
262. if (!sseWriteFrameLocked(sse_subs.items[i].fd, msg)) {
263. sseDropLocked(i);
264. continue;
265. }
266. count += 1;
267. i += 1;
268. }
269. return count;
270. }
272. // __native("http1.sse_count") → how many SSE subscriptions are currently LIVE.
273. // The observable the proactive reaper exists to keep honest (mission 084): it must fall
274. // when a client goes away, with nothing being pushed.
275. export fn hl_http1_sse_count(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
276. _ = argc;
277. _ = argv;
278. mutexLock(&sse_mutex);
279. defer mutexUnlock(&sse_mutex);
280. return api.makeNumber(@floatFromInt(sse_subs.items.len));
281. }
283. // __native("http1.sse_alive", connId) → 1 if that subscription is still registered.
284. // Lets the framework prune its own per-connection bookkeeping without pushing anything.
285. export fn hl_http1_sse_alive(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
286. if (argc < 1 or argv[0].type != .hl_number) return api.makeNumber(0);
287. const id: u32 = @intFromFloat(argv[0].data.number);
288. mutexLock(&sse_mutex);
289. defer mutexUnlock(&sse_mutex);
290. for (sse_subs.items) |conn| {
291. if (conn.id == id) return api.makeNumber(1);
292. }
293. return api.makeNumber(0);
294. }
296. // __native("http1.sse_broadcast", jsonString) → number of clients pushed to.
297. export fn hl_http1_sse_broadcast(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
298. if (argc < 1 or argv[0].type != .hl_string) return api.makeNumber(0);
299. const msg = argv[0].data.string.ptr[0..argv[0].data.string.len];
300. const n = sseBroadcast(msg);
301. return api.makeNumber(@floatFromInt(n));
302. }
304. // __native("http1.sse_send", connId, jsonString) → 1 pushed / 0 gone.
305. // Mission 080 (D26): the ADDRESSED half of the channel — the dependency-scoped
306. // push sends a different payload to each client, so it cannot use broadcast.
307. export fn hl_http1_sse_send(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
308. if (argc < 2 or argv[0].type != .hl_number or argv[1].type != .hl_string) return api.makeNumber(0);
309. const id: u32 = @intFromFloat(argv[0].data.number);
310. const msg = argv[1].data.string.ptr[0..argv[1].data.string.len];
311. mutexLock(&sse_mutex);
312. defer mutexUnlock(&sse_mutex);
313. for (sse_subs.items, 0..) |conn, i| {
314. if (conn.id != id) continue;
315. if (!sseWriteFrameLocked(conn.fd, msg)) {
316. sseDropLocked(i);
317. return api.makeNumber(0);
318. }
319. return api.makeNumber(1);
320. }
321. return api.makeNumber(0);
322. }
324. // =========================================================================
325. // WebSocket engine (mission 064, decisions D8/D9)
326. //
327. // Framing lives in the shared core (plugins/http/ws_common.zig); this engine
328. // owns the h1-specific parts: the Upgrade/101 handshake (done inline in the
329. // I/O worker, see readAndParseRequest) and the socket lifecycle after it.
330. //
331. // One GLOBAL engine per plugin (mirrors the SSE registry): a single reader
332. // thread epolls all upgraded sockets with level-triggered EPOLLIN only —
333. // EPOLLOUT is never armed, so an idle connection never wakes the loop (the
334. // busy-spin trap found in the http2 TLS path). Complete messages become
335. // WsEvent entries that the interpreter drains via the hl_http1_ws_events
336. // iterator (registered on the event loop like the request iterator).
337. //
338. // Outbound writes (send/broadcast/ping/close + pong replies) happen under
339. // ws_mutex from either the interpreter thread or the reader thread. Sockets
340. // are non-blocking; a slow consumer gets bounded EAGAIN retries (~100ms)
341. // and is dropped rather than buffered (no outbound queue, no EPOLLOUT).
342. // WS upgrades are plain-HTTP only for now (TLS WS = future work, like SSE).
343. // =========================================================================
345. const WsEventKind = enum(u8) { connect, message, close, pong };
347. const WsEvent = struct {
348. kind: WsEventKind,
349. id: u32,
350. data: ?[]u8 = null, // allocated payload (message data / close reason)
351. is_binary: bool = false,
352. code: u16 = 0, // close code
353. // Mission 093: the UPGRADE REQUEST's `Cookie` header, verbatim, carried on the
354. // `connect` event only. A WebSocket handshake is an ordinary HTTP request, so
355. // the browser sends the session cookie with it automatically — this is the one
356. // moment the socket can be attributed to whoever loaded the page, and after the
357. // 101 the request (and its headers) is freed. Allocated; freed with the event.
358. cookie: ?[]u8 = null,
359. // Ticket #74: the upgrade request's `Host` header, verbatim, on `connect` only —
360. // the address the browser dialled, which a page served per subdomain is rendered
361. // for. Allocated; freed with the event.
362. host: ?[]u8 = null,
363. // Ticket #105: EVERY header of the upgrade request, `name: value` lines joined by
364. // `\n` (names already lowercase), on `connect` only — what a page constructed for
365. // a navigation over this socket reads as `headers`. Allocated; freed with the event.
366. headers: ?[]u8 = null,
367. stamp: u64 = 0, // api.stamp() when queued (wsQueue)
368. };
370. /// Queue one server-side event, stamped (plugin_api.zig `head_stamp_fn`). Caller holds `ws_mutex`.
371. fn wsQueue(ev: WsEvent) !void {
372. var e = ev;
373. e.stamp = api.stamp();
374. try ws_event_queue.append(allocator, e);
375. }
377. /// `head_stamp_fn` of the server's event source: when its next event was queued.
378. fn wsEventsHeadStamp(_: ?*anyopaque) callconv(.c) u64 {
379. mutexLock(&ws_mutex);
380. defer mutexUnlock(&ws_mutex);
381. return if (ws_event_queue.items.len == 0) api.STAMP_NONE else ws_event_queue.items[0].stamp;
382. }
384. const WsClient = struct {
385. id: u32,
386. fd: i32,
387. decoder: ws.Decoder,
388. closing: bool = false, // server sent close, awaiting peer echo
389. // liveness (mission 091): when this socket last produced a frame, and when
390. // the sweep's ping went out (0 = none outstanding). A TCP connection whose
391. // peer vanished without a FIN stays writable indefinitely, so "still open"
392. // is not evidence of a live peer — the pong is.
393. last_seen_ms: i64 = 0,
394. ping_at_ms: i64 = 0,
395. };
397. var ws_mutex: PthreadMutex = c.PTHREAD_MUTEX_INITIALIZER;
398. var ws_clients: std.ArrayListUnmanaged(*WsClient) = .empty;
399. var ws_event_queue: std.ArrayListUnmanaged(WsEvent) = .empty;
400. var ws_epoll_fd: i32 = -1;
401. var ws_wake_fd: i32 = -1;
402. /// THE EVENT LOOP'S BELL for the WS event queue (mission 256), and a different fd
403. /// from `ws_wake_fd` above — that one wakes the reader THREAD's own epoll, this
404. /// one wakes the interpreter loop. Without it an app that turns sockets on has a
405. /// source with no fd, and ONE such source puts the whole loop back on the 1ms
406. /// poll (loop_wait.Waiter.observe) — so the server would busy-poll for as long as
407. /// WebSockets were enabled.
408. var ws_loop_wake_fd: i32 = -1;
409. var ws_thread: ?std.Thread = null;
410. var ws_running = std.atomic.Value(bool).init(false);
411. var ws_enabled = std.atomic.Value(bool).init(false);
412. var ws_next_id: u32 = 1;
414. // --- liveness sweep (mission 091) ----------------------------------------
415. // The reader thread already wakes every 500ms (the epoll timeout), so the sweep
416. // costs no timer and no new thread: on every tick it pings sockets that have
417. // gone quiet and drops the ones whose pong is overdue. The drop enqueues the
418. // ordinary `close` event, so every consumer above (hl:web's subscription
419. // registry included) prunes through the path it already had — nothing upstream
420. // learns a new concept, and nothing at the hl level needs a timer.
421. //
422. // Operator knobs, read once at engine start; the defaults are production
423. // values, the tests shrink them.
424. const WS_PING_MS_DEFAULT: i64 = 15000; // quiet this long -> ask
425. const WS_PONG_MS_DEFAULT: i64 = 10000; // no answer in this long -> gone
426. var ws_ping_ms: i64 = WS_PING_MS_DEFAULT;
427. var ws_pong_ms: i64 = WS_PONG_MS_DEFAULT;
429. /// MONOTONIC milliseconds — a timeout measured against the wall clock would fire
430. /// early or never after an NTP step.
431. fn wsNowMs() i64 {
432. var ts: linux.timespec = undefined;
433. _ = linux.clock_gettime(linux.CLOCK.MONOTONIC, &ts);
434. return @as(i64, ts.sec) * 1000 + @divTrunc(@as(i64, ts.nsec), 1_000_000);
435. }
437. fn wsEnvMs(name: [:0]const u8, fallback: i64) i64 {
438. const raw = std.mem.span(std.c.getenv(name) orelse return fallback);
439. const n = std.fmt.parseInt(i64, std.mem.trim(u8, raw, " \t"), 10) catch return fallback;
440. if (n <= 0) return fallback;
441. return n;
442. }
444. const EAGAIN_ERR: isize = 11;
445. const EINTR_ERR: isize = 4;
447. /// WriteFn callback over a raw fd (context = fd stuffed into the pointer).
448. /// Non-blocking socket: bounded EAGAIN retries, then give up (caller drops).
449. fn wsFdWrite(ctx: ?*anyopaque, data: []const u8) bool {
450. const fd: i32 = @intCast(@intFromPtr(ctx));
451. var written: usize = 0;
452. var retries: u32 = 0;
453. while (written < data.len) {
454. const rc = linux.sendto(fd, data[written..].ptr, data.len - written, MSG_NOSIGNAL, null, 0);
455. const n: isize = @bitCast(rc);
456. if (n > 0) {
457. written += @intCast(n);
458. retries = 0;
459. continue;
460. }
461. const e = -n;
462. if (e == EAGAIN_ERR) {
463. retries += 1;
464. if (retries > 100) return false; // ~100ms of backpressure → drop
465. const req = linux.timespec{ .sec = 0, .nsec = 1_000_000 }; // 1ms
466. _ = linux.nanosleep(&req, null);
467. continue;
468. }
469. if (e == EINTR_ERR) continue;
470. return false;
471. }
472. return true;
473. }
475. fn wsFdCtx(fd: i32) ?*anyopaque {
476. return @ptrFromInt(@as(usize, @intCast(fd)));
477. }
479. /// Start the reader thread + epoll instance. Idempotent. Called from the
480. /// interpreter thread (hl_http1_ws_events); also flips ws_enabled so the I/O
481. /// workers start honoring Upgrade requests.
482. fn wsEnsureStarted() bool {
483. mutexLock(&ws_mutex);
484. defer mutexUnlock(&ws_mutex);
485. if (ws_running.load(.acquire)) return true;
487. ws_ping_ms = wsEnvMs("HL_WS_PING_MS", WS_PING_MS_DEFAULT);
488. ws_pong_ms = wsEnvMs("HL_WS_PONG_TIMEOUT_MS", WS_PONG_MS_DEFAULT);
490. const ep_rc = linux.epoll_create1(linux.EPOLL.CLOEXEC);
491. const ep: i32 = @bitCast(@as(u32, @truncate(ep_rc)));
492. if (ep < 0) return false;
494. const ef_rc = linux.eventfd(0, linux.EFD.CLOEXEC | linux.EFD.NONBLOCK);
495. const ef: i32 = @bitCast(@as(u32, @truncate(ef_rc)));
496. if (ef < 0) {
497. _ = linux.close(ep);
498. return false;
499. }
500. var wake_ev = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .fd = ef } };
501. _ = linux.epoll_ctl(ep, linux.EPOLL.CTL_ADD, ef, &wake_ev);
503. ws_epoll_fd = ep;
504. ws_wake_fd = ef;
505. if (ws_loop_wake_fd < 0) ws_loop_wake_fd = http.makeWakeFd(); // mission 256
506. ws_running.store(true, .release);
507. ws_thread = std.Thread.spawn(.{}, wsReaderLoop, .{}) catch {
508. ws_running.store(false, .release);
509. _ = linux.close(ep);
510. _ = linux.close(ef);
511. ws_epoll_fd = -1;
512. ws_wake_fd = -1;
513. return false;
514. };
515. ws_enabled.store(true, .release);
516. return true;
517. }
519. fn wsWake() void {
520. if (ws_wake_fd >= 0) {
521. const one: u64 = 1;
522. _ = linux.write(ws_wake_fd, @ptrCast(&one), @sizeOf(u64));
523. }
524. }
526. /// The upgrade's headers as `name: value` lines joined by `\n` (ticket #105) — one
527. /// allocation the connect event carries; a parsed header holds no `\n`. Null when
528. /// there are none or the allocation fails (the connection still opens).
529. fn joinHeaders(items: []const HeaderPair) ?[]u8 {
530. var len: usize = 0;
531. for (items) |hdr| len += hdr.key.len + 2 + hdr.value.len + 1;
532. if (len == 0) return null;
533. const out = allocator.alloc(u8, len - 1) catch return null;
534. var at: usize = 0;
535. for (items, 0..) |hdr, i| {
536. if (i > 0) {
537. out[at] = '\n';
538. at += 1;
539. }
540. @memcpy(out[at .. at + hdr.key.len], hdr.key);
541. at += hdr.key.len;
542. @memcpy(out[at .. at + 2], ": ");
543. at += 2;
544. @memcpy(out[at .. at + hdr.value.len], hdr.value);
545. at += hdr.value.len;
546. }
547. return out;
548. }
550. /// Hand an upgraded socket to the engine. Called from an I/O worker thread
551. /// right after the 101 was written. `leftover` = bytes the client sent after
552. /// the handshake that were already consumed into the header buffer. `cookie` is
553. /// the handshake's `Cookie` header (mission 093) — copied here, because the
554. /// parsed request is freed the moment this returns. `host` is its `Host` header
555. /// (ticket #74), copied for the same reason. `headers` is all of them as
556. /// `name: value` lines (ticket #105), already an allocated copy: owned from here.
557. /// `handshake` is the 101 response. It is written HERE, under `ws_mutex` and after the
558. /// `connect` event is queued (ticket: 039_http1_ws_client, 2026-10-02): written before
559. /// registering, an in-process client could read it and queue its `open` ahead of the
560. /// server's `connect`, and a `peer.send` from `on connect` could not overtake it either.
561. fn wsRegisterClient(fd: i32, handshake: []const u8, leftover: []const u8, cookie: []const u8, host: []const u8, headers: ?[]u8) void {
562. // Non-blocking for the reader loop
563. const flags_rc = linux.fcntl(fd, linux.F.GETFL, @as(usize, 0));
564. const flags_i: isize = @bitCast(flags_rc);
565. if (flags_i >= 0) {
566. var oflags: linux.O = @bitCast(@as(u32, @truncate(flags_rc)));
567. oflags.NONBLOCK = true;
568. _ = linux.fcntl(fd, linux.F.SETFL, @as(usize, @as(u32, @bitCast(oflags))));
569. }
571. mutexLock(&ws_mutex);
572. const client = allocator.create(WsClient) catch {
573. mutexUnlock(&ws_mutex);
574. if (headers) |hd| allocator.free(hd);
575. _ = linux.close(fd);
576. return;
577. };
578. client.* = .{
579. .id = ws_next_id,
580. .fd = fd,
581. .decoder = ws.Decoder.init(allocator),
582. .last_seen_ms = wsNowMs(),
583. };
584. ws_next_id += 1;
585. ws_clients.append(allocator, client) catch {
586. allocator.destroy(client);
587. mutexUnlock(&ws_mutex);
588. if (headers) |hd| allocator.free(hd);
589. _ = linux.close(fd);
590. return;
591. };
592. const cookie_copy: ?[]u8 = if (cookie.len > 0) (allocator.dupe(u8, cookie) catch null) else null;
593. const host_copy: ?[]u8 = if (host.len > 0) (allocator.dupe(u8, host) catch null) else null;
594. wsQueue(.{ .kind = .connect, .id = client.id, .cookie = cookie_copy, .host = host_copy, .headers = headers }) catch {
595. if (cookie_copy) |cc| allocator.free(cc);
596. if (host_copy) |hc| allocator.free(hc);
597. if (headers) |hd| allocator.free(hd);
598. };
599. if (leftover.len > 0) client.decoder.feed(leftover) catch {};
600. if (!wsFdWrite(wsFdCtx(fd), handshake)) {
601. // the same answer a failed send gets: `close` 1006 after the `connect`
602. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
603. wsRemoveLocked(ws_clients.items.len - 1);
604. mutexUnlock(&ws_mutex);
605. return;
606. }
607. mutexUnlock(&ws_mutex);
609. // Level-triggered EPOLLIN only (never EPOLLOUT): pending socket data fires
610. // immediately, idle connections cost nothing.
611. var ev = linux.epoll_event{ .events = linux.EPOLL.IN | linux.EPOLL.RDHUP, .data = .{ .fd = fd } };
612. _ = linux.epoll_ctl(ws_epoll_fd, linux.EPOLL.CTL_ADD, fd, &ev);
613. if (leftover.len > 0) wsWake(); // decoder-buffered bytes won't fire EPOLLIN
614. // This runs on an I/O WORKER thread, not the reader thread, and it queued a
615. // `connect` — so it rings the event loop itself (mission 256).
616. http.ringWake(ws_loop_wake_fd);
617. }
619. fn wsFindByFdLocked(fd: i32) ?usize {
620. for (ws_clients.items, 0..) |cl, i| {
621. if (cl.fd == fd) return i;
622. }
623. return null;
624. }
626. fn wsFindByIdLocked(id: u32) ?usize {
627. for (ws_clients.items, 0..) |cl, i| {
628. if (cl.id == id) return i;
629. }
630. return null;
631. }
633. /// Remove client at index: epoll DEL, close fd, free state. Mutex held.
634. fn wsRemoveLocked(idx: usize) void {
635. const client = ws_clients.items[idx];
636. _ = linux.epoll_ctl(ws_epoll_fd, linux.EPOLL.CTL_DEL, client.fd, null);
637. _ = linux.close(client.fd);
638. client.decoder.deinit();
639. _ = ws_clients.swapRemove(idx);
640. allocator.destroy(client);
641. }
643. /// Drain the client's decoder; enqueue events, auto-reply pings, run the
644. /// close handshake. Returns true if the client was removed. Mutex held.
645. fn wsDrainDecoderLocked(idx: usize) bool {
646. const client = ws_clients.items[idx];
647. while (true) {
648. const maybe_ev = client.decoder.next() catch {
649. // OOM mid-decode — drop the connection
650. wsQueue(.{ .kind = .close, .id = client.id, .code = 1011 }) catch {};
651. wsRemoveLocked(idx);
652. return true;
653. };
654. const ev = maybe_ev orelse return false;
655. // Any complete frame is proof of life, and it answers an outstanding
656. // sweep ping whatever its opcode — a peer that is talking is not dead.
657. client.last_seen_ms = wsNowMs();
658. client.ping_at_ms = 0;
659. switch (ev) {
660. .text => |p| wsQueue(.{ .kind = .message, .id = client.id, .data = p }) catch allocator.free(p),
661. .binary => |p| wsQueue(.{ .kind = .message, .id = client.id, .data = p, .is_binary = true }) catch allocator.free(p),
662. .ping => |p| {
663. _ = ws.writeFrame(&wsFdWrite, wsFdCtx(client.fd), .pong, p);
664. allocator.free(p);
665. },
666. .pong => |p| {
667. allocator.free(p);
668. wsQueue(.{ .kind = .pong, .id = client.id }) catch {};
669. },
670. // QUEUED BEFORE THE ANSWER IS WRITTEN, as the 101 is: the peer may be a
671. // client in this process, whose `close` the echo causes, and the loop
672. // delivers in stamp order (plugin_api.zig `head_stamp_fn`)
673. .close => |cl| {
674. const code = cl.code;
675. wsQueue(.{ .kind = .close, .id = client.id, .code = code, .data = cl.reason }) catch allocator.free(cl.reason);
676. if (!client.closing) {
677. const echo_code = if (code == ws.CLOSE_NO_STATUS) ws.CLOSE_NORMAL else code;
678. _ = ws.writeClose(&wsFdWrite, wsFdCtx(client.fd), echo_code, "");
679. }
680. wsRemoveLocked(idx);
681. return true;
682. },
683. .protocol_error => |code| {
684. wsQueue(.{ .kind = .close, .id = client.id, .code = code }) catch {};
685. _ = ws.writeClose(&wsFdWrite, wsFdCtx(client.fd), code, "");
686. wsRemoveLocked(idx);
687. return true;
688. },
689. }
690. }
691. }
693. /// Read all available bytes from the client socket into its decoder, then
694. /// drain. Returns true if the client was removed. Mutex held.
695. fn wsServiceClientLocked(idx: usize) bool {
696. const client = ws_clients.items[idx];
697. var buf: [16384]u8 = undefined;
698. while (true) {
699. const rc = linux.read(client.fd, &buf, buf.len);
700. const n: isize = @bitCast(rc);
701. if (n > 0) {
702. client.decoder.feed(buf[0..@intCast(n)]) catch {};
703. continue;
704. }
705. if (n == 0) {
706. // Peer closed without a close frame → 1006 abnormal closure
707. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
708. wsRemoveLocked(idx);
709. return true;
710. }
711. const e = -n;
712. if (e == EINTR_ERR) continue;
713. if (e == EAGAIN_ERR) break; // all available data consumed
714. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
715. wsRemoveLocked(idx);
716. return true;
717. }
718. return wsDrainDecoderLocked(idx);
719. }
721. /// One liveness pass over every upgraded socket. Quiet for longer than
722. /// ws_ping_ms → send a protocol ping (RFC 6455 §5.5.2; every browser answers it
723. /// at the protocol layer, so nothing application-side is involved). A ping that
724. /// stands unanswered for ws_pong_ms → the peer is gone however open the socket
725. /// looks: enqueue the SAME `close` event a real FIN would have produced and drop
726. /// the connection. That is the whole mechanism — no separate timer, no new
727. /// thread, no concept added above this file.
728. fn wsSweep() void {
729. const now = wsNowMs();
730. mutexLock(&ws_mutex);
731. defer mutexUnlock(&ws_mutex);
732. var i: usize = 0;
733. while (i < ws_clients.items.len) {
734. const client = ws_clients.items[i];
735. if (client.ping_at_ms != 0) {
736. if (now - client.ping_at_ms > ws_pong_ms) {
737. wsQueue(.{
738. .kind = .close,
739. .id = client.id,
740. .code = ws.CLOSE_GOING_AWAY,
741. }) catch {};
742. wsRemoveLocked(i);
743. continue;
744. }
745. } else if (now - client.last_seen_ms > ws_ping_ms) {
746. if (ws.writeFrame(&wsFdWrite, wsFdCtx(client.fd), .ping, "")) {
747. client.ping_at_ms = now;
748. } else {
749. // the write itself failed — this one needs no grace period
750. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
751. wsRemoveLocked(i);
752. continue;
753. }
754. }
755. i += 1;
756. }
757. }
759. fn wsReaderLoop() void {
760. var events: [64]linux.epoll_event = undefined;
761. while (ws_running.load(.acquire)) {
762. // RING THE EVENT LOOP'S BELL AT THE END OF EVERY ITERATION (mission 256),
763. // unconditionally and without taking `ws_mutex`. Fifteen places in this
764. // file push onto `ws_event_queue`; a bell is coalesced and a spurious one
765. // is harmless (loop_wait.zig: "only a signal that is never sent at all
766. // could ever be a bug"), so one ring per pass covers all of them and can
767. // never deadlock against a path that still holds the lock. The idle cost
768. // is two empty loop rounds a second — this epoll has a 500ms timeout.
769. defer http.ringWake(ws_loop_wake_fd);
770. const n_rc = linux.epoll_wait(ws_epoll_fd, &events, events.len, 500);
771. const n: i32 = @bitCast(@as(u32, @truncate(n_rc)));
772. // The 500ms epoll timeout IS the sweep's clock: an idle loop still ticks,
773. // and a busy one sweeps just as often because the check is on wall time.
774. if (n <= 0) {
775. wsSweep();
776. continue;
777. }
778. for (events[0..@intCast(n)]) |ev| {
779. if (ev.data.fd == ws_wake_fd) {
780. var drain: u64 = 0;
781. _ = linux.read(ws_wake_fd, @ptrCast(&drain), @sizeOf(u64));
782. if (!ws_running.load(.acquire)) return;
783. // Drain decoder-buffered data (e.g. handshake leftover)
784. mutexLock(&ws_mutex);
785. var i: usize = 0;
786. while (i < ws_clients.items.len) {
787. if (!wsDrainDecoderLocked(i)) i += 1;
788. }
789. mutexUnlock(&ws_mutex);
790. continue;
791. }
792. mutexLock(&ws_mutex);
793. if (wsFindByFdLocked(ev.data.fd)) |idx| {
794. _ = wsServiceClientLocked(idx);
795. }
796. mutexUnlock(&ws_mutex);
797. }
798. wsSweep();
799. }
800. }
802. /// Stop the engine: join the reader thread, close all sockets, free queues.
803. /// Runs at interpreter teardown via the events-iterator deinit — BEFORE the
804. /// runtime dlcloses this .so, so the thread never outlives its code.
805. fn wsShutdown() void {
806. if (!ws_running.swap(false, .acq_rel)) return;
807. wsWake();
808. if (ws_thread) |t| {
809. t.join();
810. ws_thread = null;
811. }
812. mutexLock(&ws_mutex);
813. for (ws_clients.items) |client| {
814. _ = ws.writeClose(&wsFdWrite, wsFdCtx(client.fd), ws.CLOSE_GOING_AWAY, "");
815. _ = linux.close(client.fd);
816. client.decoder.deinit();
817. allocator.destroy(client);
818. }
819. ws_clients.deinit(allocator);
820. ws_clients = .empty;
821. for (ws_event_queue.items) |*ev| {
822. if (ev.data) |d| allocator.free(d);
823. if (ev.cookie) |ck| allocator.free(ck);
824. if (ev.host) |hc| allocator.free(hc);
825. if (ev.headers) |hd| allocator.free(hd);
826. }
827. ws_event_queue.deinit(allocator);
828. ws_event_queue = .empty;
829. mutexUnlock(&ws_mutex);
830. if (ws_epoll_fd >= 0) _ = linux.close(ws_epoll_fd);
831. if (ws_wake_fd >= 0) _ = linux.close(ws_wake_fd);
832. ws_epoll_fd = -1;
833. ws_wake_fd = -1;
834. // `ws_loop_wake_fd` is deliberately NOT closed: the event loop may still hold
835. // it in its epoll set, and a closed fd number gets REUSED — the loop would
836. // then be watching whatever opened next. One eventfd per process, kept for
837. // the process's life and reused if the engine restarts (mission 256).
838. ws_enabled.store(false, .release);
839. }
841. // --- Interpreter-facing exports ------------------------------------------
843. const ws_kind_names = [_][]const u8{ "connect", "message", "close", "pong" };
845. fn wsEventObjDeinit(obj: *HlObject) callconv(.c) void {
846. // fields[2] is "data", fields[5] "cookie", fields[6] "host" and fields[7]
847. // "headers" — allocated iff non-empty (empty = the static "").
848. const s = obj.fields[2].value.data.string;
849. if (s.len > 0) allocator.free(s.ptr[0..s.len]);
850. const ck = obj.fields[5].value.data.string;
851. if (ck.len > 0) allocator.free(ck.ptr[0..ck.len]);
852. const hs = obj.fields[6].value.data.string;
853. if (hs.len > 0) allocator.free(hs.ptr[0..hs.len]);
854. const hd = obj.fields[7].value.data.string;
855. if (hd.len > 0) allocator.free(hd.ptr[0..hd.len]);
856. allocator.free(obj.fields[0..obj.field_count]);
857. allocator.destroy(obj);
858. }
860. /// try_next over the WS event queue → { kind, id, data, binary, code, cookie, host, headers }
861. /// or null. `cookie` is the handshake's Cookie header, `host` its Host header and
862. /// `headers` all of its headers as `name: value` lines, each non-empty only on
863. /// `connect` (mission 093, tickets #74 and #105).
864. fn wsEventsTryNext(ctx: ?*anyopaque) callconv(.c) HlValue {
865. _ = ctx;
866. mutexLock(&ws_mutex);
867. if (ws_event_queue.items.len == 0) {
868. mutexUnlock(&ws_mutex);
869. return api.makeNull();
870. }
871. const ev = ws_event_queue.orderedRemove(0);
872. mutexUnlock(&ws_mutex);
874. const fields = allocator.alloc(HlField, 8) catch {
875. if (ev.data) |d| allocator.free(d);
876. if (ev.cookie) |ck| allocator.free(ck);
877. if (ev.host) |hc| allocator.free(hc);
878. if (ev.headers) |hd| allocator.free(hd);
879. return api.makeNull();
880. };
881. const data_slice: []const u8 = if (ev.data) |d| d else "";
882. const cookie_slice: []const u8 = if (ev.cookie) |ck| ck else "";
883. const host_slice: []const u8 = if (ev.host) |hc| hc else "";
884. const headers_slice: []const u8 = if (ev.headers) |hd| hd else "";
885. fields[0] = .{ .key = http.hlStr("kind"), .value = api.makeString(ws_kind_names[@intFromEnum(ev.kind)]) };
886. fields[1] = .{ .key = http.hlStr("id"), .value = api.makeNumber(@floatFromInt(ev.id)) };
887. fields[2] = .{ .key = http.hlStr("data"), .value = api.makeString(data_slice) };
888. fields[3] = .{ .key = http.hlStr("binary"), .value = api.makeBool(ev.is_binary) };
889. fields[4] = .{ .key = http.hlStr("code"), .value = api.makeNumber(@floatFromInt(ev.code)) };
890. fields[5] = .{ .key = http.hlStr("cookie"), .value = api.makeString(cookie_slice) };
891. fields[6] = .{ .key = http.hlStr("host"), .value = api.makeString(host_slice) };
892. fields[7] = .{ .key = http.hlStr("headers"), .value = api.makeString(headers_slice) };
894. const obj = allocator.create(HlObject) catch {
895. if (ev.data) |d| allocator.free(d);
896. if (ev.cookie) |ck| allocator.free(ck);
897. if (ev.host) |hc| allocator.free(hc);
898. if (ev.headers) |hd| allocator.free(hd);
899. allocator.free(fields);
900. return api.makeNull();
901. };
902. obj.* = .{ .fields = fields.ptr, .field_count = 8, .deinit_fn = &wsEventObjDeinit };
903. return api.makeObject(obj);
904. }
906. fn wsEventsIterDeinit(ctx: ?*anyopaque) callconv(.c) void {
907. _ = ctx;
908. wsShutdown();
909. }
911. /// __native("http1.ws_events") → event iterator; starting it enables WS
912. /// upgrades on all plain-HTTP hl:http1 servers in this process.
913. export fn hl_http1_ws_events(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
914. _ = argc;
915. _ = argv;
916. if (!wsEnsureStarted()) return api.makeNull();
917. const iter = allocator.create(HlIterator) catch return api.makeNull();
918. iter.* = .{
919. .context = null,
920. .next_fn = &wsEventsTryNext, // non-blocking either way — event loop only
921. .deinit_fn = &wsEventsIterDeinit,
922. .try_next_fn = &wsEventsTryNext,
923. .wake_fd = ws_loop_wake_fd, // mission 256 — set by wsEnsureStarted above
924. .head_stamp_fn = &wsEventsHeadStamp,
925. };
926. return api.makeIterator(iter);
927. }
929. // --- cookie-grade random (mission 093) ------------------------------------
930. // A session id is the ONLY thing standing between a stranger and someone else's
931. // session, so it may not come from a seeded PRNG: `hl:math`'s random() is a
932. // clock-seeded xoshiro, and a few of its outputs reveal its state, which would
933. // make every other session's id derivable from one's own. This reads the
934. // kernel's CSPRNG directly. It lives in hl:http1 because the session cookie is
935. // an HTTP artifact and this plugin is the one that parses and sets it; there is
936. // no other CSPRNG at the hl: level yet (named as a gap in mission 093's report).
937. var token_buf: [128]u8 = undefined;
939. /// __native("http1.random_token", n) → n lowercase hex chars (default 32, max 128)
940. export fn hl_http1_random_token(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
941. var want: usize = 32;
942. if (argc >= 1 and argv[0].type == .hl_number) {
943. const n = argv[0].data.number;
944. if (n >= 1 and n <= 128) want = @intFromFloat(n);
945. }
946. var raw: [64]u8 = undefined;
947. const need = (want + 1) / 2;
948. if (linux.getrandom(&raw, need, 0) != need) return api.makeNull();
949. const hex = "0123456789abcdef";
950. var i: usize = 0;
951. while (i < want) : (i += 1) {
952. const byte = raw[i / 2];
953. const nib: u8 = if (i % 2 == 0) (byte >> 4) else (byte & 0x0f);
954. token_buf[i] = hex[nib];
955. }
956. return api.makeString(token_buf[0..want]);
957. }
959. /// __native("http1.ws_send", id, text[, binaryFlag]) → 1 sent / 0 gone
960. export fn hl_http1_ws_send(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
961. if (argc < 2 or argv[0].type != .hl_number or argv[1].type != .hl_string) return api.makeNumber(0);
962. const id: u32 = @intFromFloat(argv[0].data.number);
963. const text = argv[1].data.string.ptr[0..argv[1].data.string.len];
964. const opcode: ws.Opcode = if (argc >= 3 and argv[2].type == .hl_bool and argv[2].data.boolean) .binary else .text;
965. mutexLock(&ws_mutex);
966. defer mutexUnlock(&ws_mutex);
967. const idx = wsFindByIdLocked(id) orelse return api.makeNumber(0);
968. const client = ws_clients.items[idx];
969. if (!ws.writeFrame(&wsFdWrite, wsFdCtx(client.fd), opcode, text)) {
970. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
971. wsRemoveLocked(idx);
972. return api.makeNumber(0);
973. }
974. return api.makeNumber(1);
975. }
977. /// __native("http1.ws_broadcast", text) → number of clients written
978. export fn hl_http1_ws_broadcast(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
979. if (argc < 1 or argv[0].type != .hl_string) return api.makeNumber(0);
980. const text = argv[0].data.string.ptr[0..argv[0].data.string.len];
981. mutexLock(&ws_mutex);
982. defer mutexUnlock(&ws_mutex);
983. var count: usize = 0;
984. var i: usize = 0;
985. while (i < ws_clients.items.len) {
986. const client = ws_clients.items[i];
987. if (ws.writeFrame(&wsFdWrite, wsFdCtx(client.fd), .text, text)) {
988. count += 1;
989. i += 1;
990. } else {
991. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
992. wsRemoveLocked(i);
993. }
994. }
995. return api.makeNumber(@floatFromInt(count));
996. }
998. /// __native("http1.ws_ping", id) → 1 sent / 0 gone
999. export fn hl_http1_ws_ping(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
1000. if (argc < 1 or argv[0].type != .hl_number) return api.makeNumber(0);
1001. const id: u32 = @intFromFloat(argv[0].data.number);
1002. mutexLock(&ws_mutex);
1003. defer mutexUnlock(&ws_mutex);
1004. const idx = wsFindByIdLocked(id) orelse return api.makeNumber(0);
1005. const client = ws_clients.items[idx];
1006. if (!ws.writeFrame(&wsFdWrite, wsFdCtx(client.fd), .ping, "")) {
1007. wsRemoveLocked(idx);
1008. return api.makeNumber(0);
1009. }
1010. return api.makeNumber(1);
1011. }
1013. /// __native("http1.ws_close", id[, code]) → 1 initiated / 0 gone.
1014. /// Sends the close frame and waits for the peer echo (reader completes it).
1015. export fn hl_http1_ws_close(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
1016. if (argc < 1 or argv[0].type != .hl_number) return api.makeNumber(0);
1017. const id: u32 = @intFromFloat(argv[0].data.number);
1018. const code: u16 = if (argc >= 2 and argv[1].type == .hl_number) @intFromFloat(argv[1].data.number) else ws.CLOSE_NORMAL;
1019. mutexLock(&ws_mutex);
1020. defer mutexUnlock(&ws_mutex);
1021. const idx = wsFindByIdLocked(id) orelse return api.makeNumber(0);
1022. const client = ws_clients.items[idx];
1023. if (!ws.writeClose(&wsFdWrite, wsFdCtx(client.fd), code, "")) {
1024. wsQueue(.{ .kind = .close, .id = client.id, .code = 1006 }) catch {};
1025. wsRemoveLocked(idx);
1026. return api.makeNumber(0);
1027. }
1028. client.closing = true;
1029. return api.makeNumber(1);
1030. }
1032. // =========================================================================
1033. // TLS context — OpenSSL via dlopen (optional, no compile-time dep)
1034. // =========================================================================
1036. const c_dlfcn = @cImport({
1037. @cInclude("dlfcn.h");
1038. });
1040. const SSL_CTX = opaque {};
1041. const SSL = opaque {};
1042. const SSL_METHOD = opaque {};
1044. // OpenSSL function pointer types
1045. const SSL_library_init_fn = *const fn () callconv(.c) c_int;
1046. const SSL_load_error_strings_fn = *const fn () callconv(.c) void;
1047. const TLS_server_method_fn = *const fn () callconv(.c) ?*const SSL_METHOD;
1048. const SSL_CTX_new_fn = *const fn (?*const SSL_METHOD) callconv(.c) ?*SSL_CTX;
1049. const SSL_CTX_free_fn = *const fn (?*SSL_CTX) callconv(.c) void;
1050. const SSL_CTX_use_certificate_chain_file_fn = *const fn (?*SSL_CTX, [*:0]const u8) callconv(.c) c_int;
1051. const SSL_CTX_use_PrivateKey_file_fn = *const fn (?*SSL_CTX, [*:0]const u8, c_int) callconv(.c) c_int;
1052. const SSL_new_fn = *const fn (?*SSL_CTX) callconv(.c) ?*SSL;
1053. const SSL_free_fn = *const fn (?*SSL) callconv(.c) void;
1054. const SSL_set_fd_fn = *const fn (?*SSL, c_int) callconv(.c) c_int;
1055. const SSL_accept_fn = *const fn (?*SSL) callconv(.c) c_int;
1056. const SSL_read_fn = *const fn (?*SSL, [*]u8, c_int) callconv(.c) c_int;
1057. const SSL_write_fn = *const fn (?*SSL, [*]const u8, c_int) callconv(.c) c_int;
1058. const SSL_shutdown_fn = *const fn (?*SSL) callconv(.c) c_int;
1059. const SSL_get_error_fn = *const fn (?*SSL, c_int) callconv(.c) c_int;
1060. const OPENSSL_init_ssl_fn = *const fn (u64, ?*anyopaque) callconv(.c) c_int;
1062. const SSL_FILETYPE_PEM: c_int = 1;
1064. const TlsContext = struct {
1065. ssl_ctx: ?*SSL_CTX = null,
1066. lib_ssl: ?*anyopaque = null,
1067. lib_crypto: ?*anyopaque = null,
1069. // Function pointers
1070. fn_ssl_ctx_new: ?SSL_CTX_new_fn = null,
1071. fn_ssl_ctx_free: ?SSL_CTX_free_fn = null,
1072. fn_ssl_ctx_use_cert: ?SSL_CTX_use_certificate_chain_file_fn = null,
1073. fn_ssl_ctx_use_key: ?SSL_CTX_use_PrivateKey_file_fn = null,
1074. fn_ssl_new: ?SSL_new_fn = null,
1075. fn_ssl_free: ?SSL_free_fn = null,
1076. fn_ssl_set_fd: ?SSL_set_fd_fn = null,
1077. fn_ssl_accept: ?SSL_accept_fn = null,
1078. fn_ssl_read: ?SSL_read_fn = null,
1079. fn_ssl_write: ?SSL_write_fn = null,
1080. fn_ssl_shutdown: ?SSL_shutdown_fn = null,
1081. fn_ssl_get_error: ?SSL_get_error_fn = null,
1083. fn loadSym(lib: ?*anyopaque, comptime T: type, name: [*:0]const u8) ?T {
1084. const sym = c_dlfcn.dlsym(lib, name) orelse return null;
1085. return @ptrCast(sym);
1086. }
1088. fn init(cert_path: []const u8, key_path: []const u8) ?TlsContext {
1089. var ctx = TlsContext{};
1091. // Try loading libssl and libcrypto
1092. const ssl_paths = [_][*:0]const u8{ "libssl.so.3", "libssl.so.1.1", "libssl.so" };
1093. const crypto_paths = [_][*:0]const u8{ "libcrypto.so.3", "libcrypto.so.1.1", "libcrypto.so" };
1095. for (ssl_paths) |path| {
1096. ctx.lib_ssl = c_dlfcn.dlopen(path, c_dlfcn.RTLD_NOW | c_dlfcn.RTLD_LOCAL);
1097. if (ctx.lib_ssl != null) break;
1098. }
1099. if (ctx.lib_ssl == null) {
1100. logMsg("http1: TLS: failed to load libssl.so\n");
1101. return null;
1102. }
1104. for (crypto_paths) |path| {
1105. ctx.lib_crypto = c_dlfcn.dlopen(path, c_dlfcn.RTLD_NOW | c_dlfcn.RTLD_LOCAL);
1106. if (ctx.lib_crypto != null) break;
1107. }
1108. if (ctx.lib_crypto == null) {
1109. logMsg("http1: TLS: failed to load libcrypto.so\n");
1110. _ = c_dlfcn.dlclose(ctx.lib_ssl);
1111. return null;
1112. }
1114. // Load function pointers
1115. // Try OPENSSL_init_ssl first (OpenSSL 1.1+), fall back to SSL_library_init
1116. if (loadSym(ctx.lib_ssl, OPENSSL_init_ssl_fn, "OPENSSL_init_ssl")) |init_fn| {
1117. _ = init_fn(0, null);
1118. } else if (loadSym(ctx.lib_ssl, SSL_library_init_fn, "SSL_library_init")) |lib_init| {
1119. _ = lib_init();
1120. if (loadSym(ctx.lib_ssl, SSL_load_error_strings_fn, "SSL_load_error_strings")) |load_err| {
1121. load_err();
1122. }
1123. }
1125. const method_fn = loadSym(ctx.lib_ssl, TLS_server_method_fn, "TLS_server_method") orelse {
1126. logMsg("http1: TLS: TLS_server_method not found\n");
1127. ctx.deinit();
1128. return null;
1129. };
1131. ctx.fn_ssl_ctx_new = loadSym(ctx.lib_ssl, SSL_CTX_new_fn, "SSL_CTX_new");
1132. ctx.fn_ssl_ctx_free = loadSym(ctx.lib_ssl, SSL_CTX_free_fn, "SSL_CTX_free");
1133. ctx.fn_ssl_ctx_use_cert = loadSym(ctx.lib_ssl, SSL_CTX_use_certificate_chain_file_fn, "SSL_CTX_use_certificate_chain_file");
1134. ctx.fn_ssl_ctx_use_key = loadSym(ctx.lib_ssl, SSL_CTX_use_PrivateKey_file_fn, "SSL_CTX_use_PrivateKey_file");
1135. ctx.fn_ssl_new = loadSym(ctx.lib_ssl, SSL_new_fn, "SSL_new");
1136. ctx.fn_ssl_free = loadSym(ctx.lib_ssl, SSL_free_fn, "SSL_free");
1137. ctx.fn_ssl_set_fd = loadSym(ctx.lib_ssl, SSL_set_fd_fn, "SSL_set_fd");
1138. ctx.fn_ssl_accept = loadSym(ctx.lib_ssl, SSL_accept_fn, "SSL_accept");
1139. ctx.fn_ssl_read = loadSym(ctx.lib_ssl, SSL_read_fn, "SSL_read");
1140. ctx.fn_ssl_write = loadSym(ctx.lib_ssl, SSL_write_fn, "SSL_write");
1141. ctx.fn_ssl_shutdown = loadSym(ctx.lib_ssl, SSL_shutdown_fn, "SSL_shutdown");
1142. ctx.fn_ssl_get_error = loadSym(ctx.lib_ssl, SSL_get_error_fn, "SSL_get_error");
1144. if (ctx.fn_ssl_ctx_new == null or ctx.fn_ssl_new == null or
1145. ctx.fn_ssl_set_fd == null or ctx.fn_ssl_accept == null or
1146. ctx.fn_ssl_read == null or ctx.fn_ssl_write == null)
1147. {
1148. logMsg("http1: TLS: missing required SSL symbols\n");
1149. ctx.deinit();
1150. return null;
1151. }
1153. // Create SSL_CTX
1154. const method = method_fn();
1155. ctx.ssl_ctx = ctx.fn_ssl_ctx_new.?(method);
1156. if (ctx.ssl_ctx == null) {
1157. logMsg("http1: TLS: SSL_CTX_new failed\n");
1158. ctx.deinit();
1159. return null;
1160. }
1162. // Load cert and key
1163. const cert_z = allocator.dupeZ(u8, cert_path) catch {
1164. ctx.deinit();
1165. return null;
1166. };
1167. defer allocator.free(cert_z);
1169. const key_z = allocator.dupeZ(u8, key_path) catch {
1170. ctx.deinit();
1171. return null;
1172. };
1173. defer allocator.free(key_z);
1175. if (ctx.fn_ssl_ctx_use_cert) |use_cert| {
1176. if (use_cert(ctx.ssl_ctx, cert_z.ptr) != 1) {
1177. logFmt("http1: TLS: failed to load certificate: {s}\n", .{cert_path});
1178. ctx.deinit();
1179. return null;
1180. }
1181. }
1183. if (ctx.fn_ssl_ctx_use_key) |use_key| {
1184. if (use_key(ctx.ssl_ctx, key_z.ptr, SSL_FILETYPE_PEM) != 1) {
1185. logFmt("http1: TLS: failed to load private key: {s}\n", .{key_path});
1186. ctx.deinit();
1187. return null;
1188. }
1189. }
1191. logMsg("http1: TLS initialized\n");
1192. return ctx;
1193. }
1195. fn wrapConnection(self: *const TlsContext, fd: i32) ?*SSL {
1196. const ssl = self.fn_ssl_new.?(self.ssl_ctx);
1197. if (ssl == null) return null;
1198. _ = self.fn_ssl_set_fd.?(ssl, fd);
1199. const ret = self.fn_ssl_accept.?(ssl);
1200. if (ret != 1) {
1201. self.fn_ssl_free.?(ssl);
1202. return null;
1203. }
1204. return ssl;
1205. }
1207. fn sslRead(self: *const TlsContext, ssl: *SSL, buf: []u8) isize {
1208. const ret = self.fn_ssl_read.?(ssl, buf.ptr, @intCast(@min(buf.len, std.math.maxInt(c_int))));
1209. if (ret <= 0) return 0;
1210. return @intCast(ret);
1211. }
1213. fn sslWrite(self: *const TlsContext, ssl: *SSL, data: []const u8) isize {
1214. var written: usize = 0;
1215. while (written < data.len) {
1216. const chunk_len: c_int = @intCast(@min(data.len - written, std.math.maxInt(c_int)));
1217. const ret = self.fn_ssl_write.?(ssl, data[written..].ptr, chunk_len);
1218. if (ret <= 0) return @intCast(written);
1219. written += @intCast(ret);
1220. }
1221. return @intCast(written);
1222. }
1224. fn sslShutdown(self: *const TlsContext, ssl: *SSL) void {
1225. _ = self.fn_ssl_shutdown.?(ssl);
1226. self.fn_ssl_free.?(ssl);
1227. }
1229. fn deinit(self: *TlsContext) void {
1230. if (self.ssl_ctx != null) {
1231. if (self.fn_ssl_ctx_free) |free_fn| {
1232. free_fn(self.ssl_ctx);
1233. }
1234. self.ssl_ctx = null;
1235. }
1236. if (self.lib_ssl != null) {
1237. _ = c_dlfcn.dlclose(self.lib_ssl);
1238. self.lib_ssl = null;
1239. }
1240. if (self.lib_crypto != null) {
1241. _ = c_dlfcn.dlclose(self.lib_crypto);
1242. self.lib_crypto = null;
1243. }
1244. }
1245. };
1247. // =========================================================================
1248. // MPSC Request Queue — thread-safe, blocks on dequeue
1249. // =========================================================================
1251. const ParsedRequest = struct {
1252. method: []const u8,
1253. path: []const u8, // URL-decoded
1254. query_params: []http.QueryParam, // parsed key-value pairs
1255. query_raw: []const u8, // raw query string
1256. headers: std.ArrayListUnmanaged(HeaderPair),
1257. body: []const u8,
1258. client_fd: i32,
1259. ssl: ?*SSL, // null if plain HTTP
1260. keep_alive: bool,
1261. };
1263. const HeaderPair = struct {
1264. key: []const u8,
1265. value: []const u8,
1266. };
1268. const RequestQueue = struct {
1269. queue: std.ArrayListUnmanaged(ParsedRequest),
1270. mutex: PthreadMutex,
1271. condvar: PthreadCond,
1272. shutdown: bool,
1273. /// THE EVENT LOOP'S BELL (mission 256). The condvar above wakes a `for (req of
1274. /// server)` consumer blocked in `dequeue`; the interpreter's event loop is
1275. /// NOT that consumer — it polls `tryDequeue` from a loop that also watches
1276. /// every other source, so it cannot block on a condvar belonging to one of
1277. /// them. An eventfd it can: `native/src/loop_wait.zig` epolls this fd, and an
1278. /// idle server costs nothing instead of 1000 empty poll rounds a second.
1279. loop_wake_fd: i32,
1281. fn init() RequestQueue {
1282. var self: RequestQueue = .{
1283. .queue = .empty,
1284. .mutex = c.PTHREAD_MUTEX_INITIALIZER,
1285. .condvar = c.PTHREAD_COND_INITIALIZER,
1286. .shutdown = false,
1287. .loop_wake_fd = http.makeWakeFd(),
1288. };
1289. mutexInit(&self.mutex);
1290. condInit(&self.condvar);
1291. return self;
1292. }
1294. fn enqueue(self: *RequestQueue, req: ParsedRequest) void {
1295. mutexLock(&self.mutex);
1296. defer mutexUnlock(&self.mutex);
1297. self.queue.append(allocator, req) catch return;
1298. condSignal(&self.condvar);
1299. // Rung INSIDE the lock, so the fd's counter is already up by the time the
1300. // request is visible to `tryDequeue`. A loop that is between its last poll
1301. // and its next `epoll_wait` therefore finds the level raised and returns
1302. // at once instead of sleeping on a queue that has work in it.
1303. http.ringWake(self.loop_wake_fd);
1304. }
1306. fn dequeue(self: *RequestQueue) ?ParsedRequest {
1307. mutexLock(&self.mutex);
1308. defer mutexUnlock(&self.mutex);
1309. while (self.queue.items.len == 0 and !self.shutdown) {
1310. condWait(&self.condvar, &self.mutex);
1311. }
1312. if (self.shutdown and self.queue.items.len == 0) return null;
1313. return self.queue.orderedRemove(0);
1314. }
1316. /// Non-blocking: returns null immediately if queue is empty
1317. fn tryDequeue(self: *RequestQueue) ?ParsedRequest {
1318. mutexLock(&self.mutex);
1319. defer mutexUnlock(&self.mutex);
1320. if (self.queue.items.len == 0) return null;
1321. return self.queue.orderedRemove(0);
1322. }
1324. fn signalShutdown(self: *RequestQueue) void {
1325. mutexLock(&self.mutex);
1326. defer mutexUnlock(&self.mutex);
1327. self.shutdown = true;
1328. condBroadcast(&self.condvar);
1329. // The event loop is told too — it is blocked on this fd and its `while`
1330. // condition (has every source retired?) can only be re-read on the way out.
1331. http.ringWake(self.loop_wake_fd);
1332. }
1334. fn deinit(self: *RequestQueue) void {
1335. // Free any remaining queued requests
1336. for (self.queue.items) |*req| {
1337. freeRequest(req);
1338. }
1339. self.queue.deinit(allocator);
1340. if (self.loop_wake_fd >= 0) {
1341. _ = linux.close(self.loop_wake_fd);
1342. self.loop_wake_fd = -1;
1343. }
1344. _ = c.pthread_cond_destroy(&self.condvar);
1345. _ = c.pthread_mutex_destroy(&self.mutex);
1346. }
1347. };
1349. fn freeRequest(req: *ParsedRequest) void {
1350. allocator.free(req.method);
1351. allocator.free(req.path);
1352. allocator.free(req.query_raw);
1353. for (req.query_params) |param| {
1354. allocator.free(param.key);
1355. allocator.free(param.value);
1356. }
1357. allocator.free(req.query_params);
1358. for (req.headers.items) |hdr| {
1359. allocator.free(hdr.key);
1360. allocator.free(hdr.value);
1361. }
1362. req.headers.deinit(allocator);
1363. if (req.body.len > 0) allocator.free(req.body);
1364. }
1366. // =========================================================================
1367. // Connection — per-connection state tracked by I/O workers
1368. // =========================================================================
1370. const Connection = struct {
1371. fd: i32,
1372. ssl: ?*SSL,
1373. keep_alive: bool,
1374. request_count: u32,
1375. max_requests: u32,
1377. fn read(self: *const Connection, tls: ?*const TlsContext, buf: []u8) usize {
1378. if (self.ssl) |ssl_ptr| {
1379. if (tls) |t| {
1380. const n = t.sslRead(ssl_ptr, buf);
1381. if (n <= 0) return 0;
1382. return @intCast(n);
1383. }
1384. return 0;
1385. }
1386. return sysRead(self.fd, buf);
1387. }
1389. fn write(self: *const Connection, tls: ?*const TlsContext, data: []const u8) void {
1390. if (self.ssl) |ssl_ptr| {
1391. if (tls) |t| {
1392. _ = t.sslWrite(ssl_ptr, data);
1393. return;
1394. }
1395. }
1396. sysWriteAll(self.fd, data);
1397. }
1399. fn close(self: *Connection, tls: ?*const TlsContext) void {
1400. if (self.ssl) |ssl_ptr| {
1401. if (tls) |t| {
1402. t.sslShutdown(ssl_ptr);
1403. }
1404. self.ssl = null;
1405. }
1406. _ = linux.close(self.fd);
1407. }
1408. };
1410. // =========================================================================
1411. // ServerCore — owns server socket, epoll, threads, queue, optional TLS
1412. // =========================================================================
1414. const MAX_KEEPALIVE_REQUESTS: u32 = 100;
1415. const MAX_IO_THREADS: u32 = 16;
1417. const ServerCore = struct {
1418. port: u16 = 0,
1419. server_fd: i32,
1420. epoll_fd: i32,
1421. shutdown_fd: i32, // eventfd for shutdown signal
1422. tls: ?TlsContext,
1423. request_queue: RequestQueue,
1424. running: std.atomic.Value(bool),
1425. acceptor_thread: ?std.Thread,
1426. io_threads: []std.Thread,
1427. num_threads: u32,
1429. // Connection tracking for I/O threads
1430. conn_queue: ConnectionQueue,
1431. /// idle connections wait here instead of inside a blocking worker read (mission 084)
1432. parked: ParkedConns,
1434. fn create(port: u16, host_str: []const u8, cert_path: []const u8, key_path: []const u8, num_threads: u32) ?*ServerCore {
1435. const core = allocator.create(ServerCore) catch return null;
1436. core.* = .{
1437. .server_fd = -1,
1438. .epoll_fd = -1,
1439. .shutdown_fd = -1,
1440. .tls = null,
1441. .request_queue = RequestQueue.init(),
1442. .running = std.atomic.Value(bool).init(false),
1443. .acceptor_thread = null,
1444. .io_threads = &[_]std.Thread{},
1445. .num_threads = @min(num_threads, MAX_IO_THREADS),
1446. .conn_queue = ConnectionQueue.init(),
1447. .parked = ParkedConns.init(),
1448. };
1450. // Create TCP socket. NONBLOCK IS NOT DECORATION (mission 260): the acceptor
1451. // drains with `while (true) accept4(...)` until the call fails, and on a
1452. // BLOCKING listener that last call does not fail — it sleeps in the kernel
1453. // (`inet_csk_accept`) until the next client arrives. The thread then never
1454. // re-reads `core.running`, so `shutdown()`'s `join(acceptor)` waits forever
1455. // and `close()` never returns. Measured: after ONE connection the process was
1456. // down to two threads, main in `__futex_wait` (the join) and the acceptor in
1457. // `inet_csk_accept`; each extra TCP connect added exactly one fd and put the
1458. // acceptor straight back into `inet_csk_accept`. hl:http2's listener
1459. // (`http2.zig:926`) has always carried NONBLOCK — http1 was the outlier.
1460. // The accept flags below apply to the ACCEPTED socket, never to this one.
1461. const sock_rc = linux.socket(linux.AF.INET, linux.SOCK.STREAM | linux.SOCK.CLOEXEC | linux.SOCK.NONBLOCK, 0);
1462. const server_fd: i32 = @bitCast(@as(u32, @truncate(sock_rc)));
1463. if (server_fd < 0) {
1464. logMsg("http1: socket() failed\n");
1465. allocator.destroy(core);
1466. return null;
1467. }
1468. core.server_fd = server_fd;
1470. // SO_REUSEADDR
1471. const one: i32 = 1;
1472. _ = linux.setsockopt(server_fd, linux.SOL.SOCKET, linux.SO.REUSEADDR, @ptrCast(&one), @sizeOf(i32));
1474. // Parse host address
1475. var addr_val: u32 = 0; // INADDR_ANY
1476. if (host_str.len > 0 and !std.mem.eql(u8, host_str, "0.0.0.0")) {
1477. addr_val = parseIPv4(host_str) orelse 0;
1478. }
1480. // Bind
1481. const addr = linux.sockaddr.in{
1482. .port = std.mem.nativeToBig(u16, port),
1483. .addr = addr_val,
1484. };
1486. const bind_rc = linux.bind(server_fd, @ptrCast(&addr), @sizeOf(linux.sockaddr.in));
1487. const bind_err: i32 = @bitCast(@as(u32, @truncate(bind_rc)));
1488. if (bind_err < 0) {
1489. // The prefix is LOAD-BEARING: `tests/browser/fixtures.mjs` breaks its
1490. // readiness wait on `bind() failed on port N` (mission 285). The errno
1491. // and the holder are appended to it, never in front of it.
1492. var why: [320]u8 = undefined;
1493. logFmt("http1: bind() failed on port {d}: {s}\n", .{ port, http.bindFailureDetail(&why, bind_rc, port, false) });
1494. core.destroy();
1495. return null;
1496. }
1498. const listen_rc = linux.listen(server_fd, 128);
1499. const listen_err: i32 = @bitCast(@as(u32, @truncate(listen_rc)));
1500. if (listen_err < 0) {
1501. logMsg("http1: listen() failed\n");
1502. core.destroy();
1503. return null;
1504. }
1506. // Create eventfd for shutdown signaling
1507. const efd_rc = linux.eventfd(0, linux.EFD.CLOEXEC | linux.EFD.NONBLOCK);
1508. const shutdown_fd: i32 = @bitCast(@as(u32, @truncate(efd_rc)));
1509. if (shutdown_fd < 0) {
1510. logMsg("http1: eventfd() failed\n");
1511. core.destroy();
1512. return null;
1513. }
1514. core.shutdown_fd = shutdown_fd;
1516. // Create epoll instance
1517. const epoll_rc = linux.epoll_create1(linux.EPOLL.CLOEXEC);
1518. const epoll_fd: i32 = @bitCast(@as(u32, @truncate(epoll_rc)));
1519. if (epoll_fd < 0) {
1520. logMsg("http1: epoll_create1() failed\n");
1521. core.destroy();
1522. return null;
1523. }
1524. core.epoll_fd = epoll_fd;
1526. // Add server_fd to epoll
1527. var ev = linux.epoll_event{
1528. .events = linux.EPOLL.IN,
1529. .data = .{ .fd = server_fd },
1530. };
1531. _ = linux.epoll_ctl(epoll_fd, linux.EPOLL.CTL_ADD, server_fd, &ev);
1533. // Add shutdown_fd to epoll
1534. var shutdown_ev = linux.epoll_event{
1535. .events = linux.EPOLL.IN,
1536. .data = .{ .fd = shutdown_fd },
1537. };
1538. _ = linux.epoll_ctl(epoll_fd, linux.EPOLL.CTL_ADD, shutdown_fd, &shutdown_ev);
1540. // Initialize TLS if cert+key provided
1541. if (cert_path.len > 0 and key_path.len > 0) {
1542. core.tls = TlsContext.init(cert_path, key_path);
1543. if (core.tls == null) {
1544. logMsg("http1: TLS initialization failed, falling back to plain HTTP\n");
1545. }
1546. }
1548. logFmt("http1: listening on :{d}{s}\n", .{ port, if (core.tls != null) " (TLS)" else "" });
1550. // Start threads
1551. core.running.store(true, .release);
1553. // Parker thread first: the acceptor parks into it from its very first accept.
1554. // If it cannot start we log and keep going — connections then go straight to the
1555. // pool, which is the pre-084 (starvable) behaviour rather than an outage.
1556. if (!core.parked.start(core)) {
1557. logMsg("http1: parker thread unavailable — idle keep-alive connections will hold I/O workers\n");
1558. }
1560. // Allocate I/O threads
1561. const threads = allocator.alloc(std.Thread, core.num_threads) catch {
1562. core.destroy();
1563. return null;
1564. };
1565. core.io_threads = threads;
1567. for (0..core.num_threads) |i| {
1568. core.io_threads[i] = std.Thread.spawn(.{}, ioWorker, .{core}) catch {
1569. logFmt("http1: failed to spawn I/O thread {d}\n", .{i});
1570. core.num_threads = @intCast(i);
1571. core.io_threads = core.io_threads[0..i];
1572. break;
1573. };
1574. }
1576. // Start acceptor thread
1577. core.acceptor_thread = std.Thread.spawn(.{}, acceptorLoop, .{core}) catch {
1578. logMsg("http1: failed to spawn acceptor thread\n");
1579. core.shutdown();
1580. core.destroy();
1581. return null;
1582. };
1584. core.port = port;
1585. registerCore(core);
1586. return core;
1587. }
1589. fn shutdown(self: *ServerCore) void {
1590. unregisterCore(self);
1591. if (!self.running.swap(false, .acq_rel)) return;
1593. // Signal shutdown via eventfd
1594. if (self.shutdown_fd >= 0) {
1595. const val: u64 = 1;
1596. _ = linux.write(self.shutdown_fd, @ptrCast(&val), @sizeOf(u64));
1597. }
1599. // Wake up the request queue so dequeue() unblocks
1600. self.request_queue.signalShutdown();
1602. // Stop parking before the workers: the parker must not push new work into a
1603. // queue whose consumers are being torn down.
1604. self.parked.stop();
1606. // Signal connection queue to wake I/O workers
1607. self.conn_queue.signalShutdown();
1609. // Join acceptor thread
1610. if (self.acceptor_thread) |t| {
1611. t.join();
1612. self.acceptor_thread = null;
1613. }
1615. // Join I/O threads
1616. for (self.io_threads) |t| {
1617. t.join();
1618. }
1619. }
1621. fn destroy(self: *ServerCore) void {
1622. self.shutdown();
1624. if (self.epoll_fd >= 0) _ = linux.close(self.epoll_fd);
1625. if (self.shutdown_fd >= 0) _ = linux.close(self.shutdown_fd);
1626. if (self.server_fd >= 0) _ = linux.close(self.server_fd);
1628. if (self.tls) |*tls| tls.deinit();
1630. // Close every still-parked idle connection, then drain conn_queue
1631. self.parked.deinit(if (self.tls) |*t| t else null);
1632. self.conn_queue.deinit(if (self.tls) |*t| t else null);
1634. self.request_queue.deinit();
1636. if (self.io_threads.len > 0) allocator.free(self.io_threads);
1637. allocator.destroy(self);
1638. }
1639. };
1641. // =========================================================================
1642. // Connection Queue — MPSC queue for acceptor → I/O workers
1643. // =========================================================================
1645. const ConnectionQueue = struct {
1646. queue: std.ArrayListUnmanaged(Connection),
1647. mutex: PthreadMutex,
1648. condvar: PthreadCond,
1649. shutdown: bool,
1651. fn init() ConnectionQueue {
1652. var self: ConnectionQueue = .{
1653. .queue = .empty,
1654. .mutex = c.PTHREAD_MUTEX_INITIALIZER,
1655. .condvar = c.PTHREAD_COND_INITIALIZER,
1656. .shutdown = false,
1657. };
1658. mutexInit(&self.mutex);
1659. condInit(&self.condvar);
1660. return self;
1661. }
1663. fn enqueue(self: *ConnectionQueue, conn: Connection) void {
1664. mutexLock(&self.mutex);
1665. defer mutexUnlock(&self.mutex);
1666. self.queue.append(allocator, conn) catch return;
1667. condSignal(&self.condvar);
1668. }
1670. fn dequeue(self: *ConnectionQueue) ?Connection {
1671. mutexLock(&self.mutex);
1672. defer mutexUnlock(&self.mutex);
1673. while (self.queue.items.len == 0 and !self.shutdown) {
1674. condWait(&self.condvar, &self.mutex);
1675. }
1676. if (self.shutdown and self.queue.items.len == 0) return null;
1677. return self.queue.orderedRemove(0);
1678. }
1680. fn signalShutdown(self: *ConnectionQueue) void {
1681. mutexLock(&self.mutex);
1682. defer mutexUnlock(&self.mutex);
1683. self.shutdown = true;
1684. condBroadcast(&self.condvar);
1685. }
1687. /// Close everything still queued and stay a VALID, EMPTY queue — `deinit`
1688. /// leaves the list `undefined`, which is only safe on a core that is being
1689. /// freed, and a closed core deliberately is not (see `hl_http1_close`).
1690. fn closeAll(self: *ConnectionQueue, tls: ?*TlsContext) void {
1691. mutexLock(&self.mutex);
1692. defer mutexUnlock(&self.mutex);
1693. for (self.queue.items) |*conn| conn.close(tls);
1694. self.queue.clearRetainingCapacity();
1695. }
1697. fn deinit(self: *ConnectionQueue, tls: ?*TlsContext) void {
1698. for (self.queue.items) |*conn| {
1699. conn.close(tls);
1700. }
1701. self.queue.deinit(allocator);
1702. }
1703. };
1705. // =========================================================================
1706. // ParkedConns — idle connections wait HERE, not in an I/O worker (mission 084)
1707. // =========================================================================
1708. //
1709. // The starvation this fixes, measured: a `threads = 4` server, four keep-alive
1710. // connections that have gone quiet, and every subsequent request times out — each idle
1711. // socket sat inside a worker's blocking read(). Parking inverts that: an idle connection
1712. // costs one epoll registration and ZERO threads, and a worker only ever picks up a
1713. // connection that already has bytes waiting (or has hung up, which it reads as EOF and
1714. // closes). Connections are parked from the acceptor (a fresh socket may be silent — a
1715. // pre-connected browser socket routinely is) and after every keep-alive response.
1716. //
1717. // TLS caveat: an ESTABLISHED TLS connection is never parked. OpenSSL may hold already-
1718. // decrypted plaintext in its own buffer, which epoll on the raw fd cannot see, so parking
1719. // it could hang a live request. `ssl == null` covers all plain HTTP plus the pre-handshake
1720. // TLS socket (the ClientHello does arrive on the raw fd), which is what the park path takes.
1722. /// How long a parked, silent connection is kept before it is closed. It costs no thread,
1723. /// only an fd, so this is generous compared to the old in-worker 30s SO_RCVTIMEO.
1724. const PARK_IDLE_TIMEOUT_MS: i64 = 60_000;
1726. /// Milliseconds on CLOCK_MONOTONIC. This zig's `std.time` exposes no timestamp function,
1727. /// and monotonic is the right clock anyway — a wall-clock step must not expire a live
1728. /// connection early or keep a dead one parked.
1729. fn monotonicMs() i64 {
1730. var ts: linux.timespec = undefined;
1731. if (linux.clock_gettime(linux.CLOCK.MONOTONIC, &ts) != 0) return 0;
1732. return @as(i64, ts.sec) * 1000 + @divTrunc(@as(i64, ts.nsec), 1_000_000);
1733. }
1735. const ParkedConn = struct {
1736. conn: Connection,
1737. /// monotonic ms after which this silent connection is closed
1738. deadline_ms: i64,
1739. };
1741. const ParkedConns = struct {
1742. epoll_fd: i32,
1743. wake_fd: i32,
1744. thread: ?std.Thread,
1745. running: std.atomic.Value(bool),
1746. mutex: PthreadMutex,
1747. /// fd → parked connection. Guarded by `mutex`; the parker thread is the only
1748. /// consumer, park() the only producer, so a plain map is enough.
1749. map: std.AutoHashMapUnmanaged(i32, ParkedConn),
1751. fn init() ParkedConns {
1752. var self: ParkedConns = .{
1753. .epoll_fd = -1,
1754. .wake_fd = -1,
1755. .thread = null,
1756. .running = std.atomic.Value(bool).init(false),
1757. .mutex = c.PTHREAD_MUTEX_INITIALIZER,
1758. .map = .empty,
1759. };
1760. mutexInit(&self.mutex);
1761. return self;
1762. }
1764. /// Create the epoll instance + wake eventfd and spawn the parker thread.
1765. /// Returns false if the kernel objects could not be made — the caller then falls
1766. /// back to handing connections straight to the pool (old behaviour, still correct,
1767. /// just starvable).
1768. fn start(self: *ParkedConns, core: *ServerCore) bool {
1769. const ep_rc = linux.epoll_create1(linux.EPOLL.CLOEXEC);
1770. const ep: i32 = @bitCast(@as(u32, @truncate(ep_rc)));
1771. if (ep < 0) return false;
1772. const ef_rc = linux.eventfd(0, linux.EFD.CLOEXEC | linux.EFD.NONBLOCK);
1773. const ef: i32 = @bitCast(@as(u32, @truncate(ef_rc)));
1774. if (ef < 0) {
1775. _ = linux.close(ep);
1776. return false;
1777. }
1778. var wake_ev = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .fd = ef } };
1779. _ = linux.epoll_ctl(ep, linux.EPOLL.CTL_ADD, ef, &wake_ev);
1780. self.epoll_fd = ep;
1781. self.wake_fd = ef;
1782. self.running.store(true, .release);
1783. self.thread = std.Thread.spawn(.{}, parkerLoop, .{core}) catch {
1784. self.running.store(false, .release);
1785. _ = linux.close(ep);
1786. _ = linux.close(ef);
1787. self.epoll_fd = -1;
1788. self.wake_fd = -1;
1789. return false;
1790. };
1791. return true;
1792. }
1794. /// Hand a connection to the parker. The map insert happens BEFORE the epoll ADD so
1795. /// the parker can never see a readable fd it has no entry for.
1796. fn park(self: *ParkedConns, conn: Connection) bool {
1797. if (self.epoll_fd < 0 or !self.running.load(.acquire)) return false;
1798. mutexLock(&self.mutex);
1799. self.map.put(allocator, conn.fd, .{
1800. .conn = conn,
1801. .deadline_ms = monotonicMs() + PARK_IDLE_TIMEOUT_MS,
1802. }) catch {
1803. mutexUnlock(&self.mutex);
1804. return false;
1805. };
1806. var ev = linux.epoll_event{
1807. .events = linux.EPOLL.IN | linux.EPOLL.RDHUP,
1808. .data = .{ .fd = conn.fd },
1809. };
1810. const rc = linux.epoll_ctl(self.epoll_fd, linux.EPOLL.CTL_ADD, conn.fd, &ev);
1811. const err: i32 = @bitCast(@as(u32, @truncate(rc)));
1812. if (err < 0) {
1813. _ = self.map.remove(conn.fd);
1814. mutexUnlock(&self.mutex);
1815. return false;
1816. }
1817. mutexUnlock(&self.mutex);
1818. return true;
1819. }
1821. /// Take a parked connection off the epoll set. Returns it if we still owned it.
1822. fn take(self: *ParkedConns, fd: i32) ?Connection {
1823. mutexLock(&self.mutex);
1824. defer mutexUnlock(&self.mutex);
1825. const entry = self.map.fetchRemove(fd) orelse return null;
1826. _ = linux.epoll_ctl(self.epoll_fd, linux.EPOLL.CTL_DEL, fd, null);
1827. return entry.value.conn;
1828. }
1830. /// Close every parked connection whose silence outlived PARK_IDLE_TIMEOUT_MS.
1831. fn sweepExpired(self: *ParkedConns, tls: ?*TlsContext) void {
1832. const now = monotonicMs();
1833. mutexLock(&self.mutex);
1834. defer mutexUnlock(&self.mutex);
1835. var expired: [64]i32 = undefined;
1836. var n: usize = 0;
1837. var it = self.map.iterator();
1838. while (it.next()) |kv| {
1839. if (kv.value_ptr.deadline_ms <= now) {
1840. if (n == expired.len) break;
1841. expired[n] = kv.key_ptr.*;
1842. n += 1;
1843. }
1844. }
1845. for (expired[0..n]) |fd| {
1846. if (self.map.fetchRemove(fd)) |e| {
1847. _ = linux.epoll_ctl(self.epoll_fd, linux.EPOLL.CTL_DEL, fd, null);
1848. var conn = e.value.conn;
1849. conn.close(tls);
1850. }
1851. }
1852. }
1854. fn stop(self: *ParkedConns) void {
1855. if (!self.running.swap(false, .acq_rel)) return;
1856. if (self.wake_fd >= 0) {
1857. const val: u64 = 1;
1858. _ = linux.write(self.wake_fd, @ptrCast(&val), @sizeOf(u64));
1859. }
1860. if (self.thread) |t| {
1861. t.join();
1862. self.thread = null;
1863. }
1864. }
1866. /// Close every parked connection and stay a VALID, EMPTY map. Unlike `deinit`
1867. /// this keeps the epoll/wake fds and the struct usable — a CLOSED core is not
1868. /// a freed one (`hl_http1_close`), and `park()` already refuses once `running`
1869. /// is false, so the emptied map simply stays empty.
1870. fn closeAll(self: *ParkedConns, tls: ?*TlsContext) void {
1871. mutexLock(&self.mutex);
1872. defer mutexUnlock(&self.mutex);
1873. var it = self.map.iterator();
1874. while (it.next()) |kv| {
1875. var conn = kv.value_ptr.conn;
1876. conn.close(tls);
1877. }
1878. self.map.clearRetainingCapacity();
1879. }
1881. fn deinit(self: *ParkedConns, tls: ?*TlsContext) void {
1882. self.stop();
1883. mutexLock(&self.mutex);
1884. var it = self.map.iterator();
1885. while (it.next()) |kv| {
1886. var conn = kv.value_ptr.conn;
1887. conn.close(tls);
1888. }
1889. self.map.deinit(allocator);
1890. self.map = .empty;
1891. mutexUnlock(&self.mutex);
1892. if (self.epoll_fd >= 0) _ = linux.close(self.epoll_fd);
1893. if (self.wake_fd >= 0) _ = linux.close(self.wake_fd);
1894. self.epoll_fd = -1;
1895. self.wake_fd = -1;
1896. }
1897. };
1899. /// The parker thread: waits for a parked connection to become READABLE and only then
1900. /// hands it to an I/O worker. The 1s epoll timeout doubles as the idle-sweep tick.
1901. fn parkerLoop(core: *ServerCore) void {
1902. var events: [64]linux.epoll_event = undefined;
1903. while (core.parked.running.load(.acquire)) {
1904. const n_rc = linux.epoll_wait(core.parked.epoll_fd, &events, events.len, 1000);
1905. const n: i32 = @bitCast(@as(u32, @truncate(n_rc)));
1906. if (n > 0) {
1907. for (events[0..@intCast(n)]) |ev| {
1908. if (ev.data.fd == core.parked.wake_fd) {
1909. var drain: u64 = 0;
1910. _ = linux.read(core.parked.wake_fd, @ptrCast(&drain), @sizeOf(u64));
1911. continue;
1912. }
1913. // Readable, hung up or errored — all three are a worker's job: it either
1914. // parses the request or reads EOF and closes.
1915. if (core.parked.take(ev.data.fd)) |conn| {
1916. if (core.running.load(.acquire)) {
1917. core.conn_queue.enqueue(conn);
1918. } else {
1919. var dead = conn;
1920. dead.close(if (core.tls) |*t| t else null);
1921. }
1922. }
1923. }
1924. }
1925. core.parked.sweepExpired(if (core.tls) |*t| t else null);
1926. }
1927. }
1929. /// Park `conn` if we can, otherwise hand it straight to the pool. Every enqueue site
1930. /// that is NOT known to have bytes waiting goes through here.
1931. fn parkOrEnqueue(core: *ServerCore, conn: Connection) void {
1932. // An established TLS connection may hold decrypted bytes epoll cannot see — see the
1933. // ParkedConns header comment. Those go straight to a worker, as before.
1934. if (conn.ssl == null and core.parked.park(conn)) return;
1935. core.conn_queue.enqueue(conn);
1936. }
1938. // =========================================================================
1939. // Acceptor thread — epoll loop accepting new connections
1940. // =========================================================================
1942. fn acceptorLoop(core: *ServerCore) void {
1943. var events: [64]linux.epoll_event = undefined;
1945. while (core.running.load(.acquire)) {
1946. const n_rc = linux.epoll_wait(core.epoll_fd, &events, events.len, 1000);
1947. const n: i32 = @bitCast(@as(u32, @truncate(n_rc)));
1948. if (n < 0) continue;
1949. if (n == 0) continue;
1951. for (events[0..@intCast(n)]) |ev| {
1952. if (ev.data.fd == core.shutdown_fd) {
1953. return; // shutdown signaled
1954. }
1955. if (ev.data.fd == core.server_fd) {
1956. // Accept all pending connections — the listener is NONBLOCK, so the
1957. // drain ends on EAGAIN instead of sleeping inside accept4().
1958. while (true) {
1959. var client_addr: linux.sockaddr.in = undefined;
1960. var addr_len: u32 = @sizeOf(linux.sockaddr.in);
1961. const accept_rc = linux.accept4(core.server_fd, @ptrCast(&client_addr), &addr_len, linux.SOCK.CLOEXEC | linux.SOCK.NONBLOCK);
1962. const client_fd: i32 = @bitCast(@as(u32, @truncate(accept_rc)));
1963. if (client_fd < 0) break;
1964. // A shutdown that landed mid-drain: the parker is stopped and the
1965. // conn_queue has no consumers left, so hand this socket to nobody —
1966. // close it and leave, rather than leaking the fd into a dead queue.
1967. if (!core.running.load(.acquire)) {
1968. _ = linux.close(client_fd);
1969. return;
1970. }
1972. // Set back to blocking for I/O workers (simpler read/write)
1973. const flags_rc = linux.fcntl(client_fd, linux.F.GETFL, @as(usize, 0));
1974. const flags_i: isize = @bitCast(flags_rc);
1975. if (flags_i >= 0) {
1976. var oflags: linux.O = @bitCast(@as(u32, @truncate(flags_rc)));
1977. oflags.NONBLOCK = false;
1978. _ = linux.fcntl(client_fd, linux.F.SETFL, @as(usize, @as(u32, @bitCast(oflags))));
1979. }
1981. // PARK, don't hand to a worker: a just-accepted socket has no bytes
1982. // yet (browsers routinely pre-open connections and send nothing), and
1983. // a worker blocking on it is exactly the starvation this replaces.
1984. parkOrEnqueue(core, .{
1985. .fd = client_fd,
1986. .ssl = null,
1987. .keep_alive = true,
1988. .request_count = 0,
1989. .max_requests = MAX_KEEPALIVE_REQUESTS,
1990. });
1991. }
1992. }
1993. }
1994. }
1995. }
1997. // =========================================================================
1998. // I/O Worker thread — TLS handshake + read/parse HTTP → enqueue request
1999. // =========================================================================

Only the first lines are shown.