gitoriaLog in with ident

tickets

All repositories: gitoria

ReadmeCodePull requestsReleasesTicketsSettingsCommit38bdd5e438bdd5e4deploy.sh: never send .git or .gitignore to Byrodinmre38bdd5e4/plugins/http1/http1.zig

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

Only the first lines are shown.

Branches

Latest commits

  • 38bdd5e4deploy.sh: never send .git or .gitignore to Byrodinmre
  • f12fa1bcState of 2026-09-27, before the move to gitoriamre