zzstoatzz.io
forked from
karlseguin.tngl.sh/http.zig
An HTTP/1.1 server for zig
1const std = @import("std");
2const builtin = @import("builtin");
3
4const posix = @import("posix.zig");
5const httpz = @import("httpz.zig");
6const metrics = @import("metrics.zig");
7const ws = @import("websocket").server;
8
9const Config = httpz.Config;
10const Request = httpz.Request;
11const Response = httpz.Response;
12
13const BufferPool = @import("buffer.zig").Pool;
14const ThreadPool = @import("thread_pool.zig").ThreadPool;
15
16const Io = std.Io;
17const Stream = Io.Stream;
18const Allocator = std.mem.Allocator;
19const ArenaAllocator = std.heap.ArenaAllocator;
20
21const log = std.log.scoped(.httpz);
22
23const MAX_TIMEOUT = 2_147_483_647;
24// Small test batches force lifecycle tests to cross wait() boundaries; the
25// production size remains tuned for throughput.
26const EVENT_BATCH_CAPACITY = if (builtin.is_test) 8 else 128;
27
28// This is our Blocking worker. It's very different than NonBlocking and much
29// simpler. (WSH is our websocket handler, and can be void)
30pub fn Blocking(comptime S: type, comptime WSH: type) type {
31 return struct {
32 io: Io,
33 server: S,
34 mut: Io.Mutex,
35 config: *const Config,
36 allocator: Allocator,
37 buffer_pool: *BufferPool,
38 http_conn_pool: HTTPConnPool,
39 websocket: *ws.Worker(WSH),
40 timeout_request: ?Timeout,
41 timeout_keepalive: ?Timeout,
42 timeout_write_error: Timeout,
43 retain_allocated_bytes_keepalive: usize,
44 connections: List(ConnNode),
45 conn_node_pool: std.heap.MemoryPool(ConnNode),
46 thread_pool: ThreadPool(Self.handleConnection),
47
48 const ConnNode = struct {
49 next: ?*ConnNode,
50 prev: ?*ConnNode,
51 socket: posix.fd_t,
52 };
53
54 const Timeout = struct {
55 sec: u32,
56 timeval: [@sizeOf(posix.timeval)]u8,
57
58 // if sec is null, it means we want to cancel the timeout.
59 fn init(sec: ?u32) Timeout {
60 return .{
61 .sec = if (sec) |s| s else MAX_TIMEOUT,
62 .timeval = std.mem.toBytes(posix.timeval{ .sec = @intCast(sec orelse 0), .usec = 0 }),
63 };
64 }
65 };
66
67 const Self = @This();
68
69 pub fn init(io: Io, allocator: Allocator, server: S, config: *const Config) !Self {
70 const buffer_pool = try initializeBufferPool(io, allocator, config);
71 errdefer allocator.destroy(buffer_pool);
72
73 errdefer buffer_pool.deinit();
74
75 var timeout_request: ?Timeout = null;
76 if (config.timeout.request) |sec| {
77 timeout_request = Timeout.init(sec);
78 } else if (config.timeout.keepalive != null) {
79 // We have to set this up, so that when we transition from
80 // keepalive state to a request parsing state, we remove the timeout
81 timeout_request = Timeout.init(0);
82 }
83
84 var timeout_keepalive: ?Timeout = null;
85 if (config.timeout.keepalive) |sec| {
86 timeout_keepalive = Timeout.init(sec);
87 } else if (timeout_request != null) {
88 // We have to set this up, so that when we transition from
89 // request processing state to a kepealive state, we remove the timeout
90 timeout_keepalive = Timeout.init(0);
91 }
92
93 const websocket = try allocator.create(ws.Worker(WSH));
94 errdefer allocator.destroy(websocket);
95 websocket.* = try ws.Worker(WSH).init(allocator, &server._websocket_state);
96 errdefer websocket.deinit();
97
98 var http_conn_pool = try HTTPConnPool.init(io, allocator, buffer_pool, websocket, 0, config);
99 errdefer http_conn_pool.deinit();
100
101 const retain_allocated_bytes_keepalive = config.workers.retain_allocated_bytes orelse 8192;
102
103 var thread_pool = try ThreadPool(Self.handleConnection).init(io, allocator, .{
104 .count = config.threadPoolCount(),
105 .backlog = config.thread_pool.backlog orelse 500,
106 .buffer_size = config.thread_pool.buffer_size orelse 32_768,
107 });
108 errdefer {
109 thread_pool.stop();
110 thread_pool.deinit();
111 }
112
113 return .{
114 .io = io,
115 .mut = .init,
116 .server = server,
117 .config = config,
118 .connections = .{},
119 .allocator = allocator,
120 .websocket = websocket,
121 .buffer_pool = buffer_pool,
122 .thread_pool = thread_pool,
123 .http_conn_pool = http_conn_pool,
124 .timeout_request = timeout_request,
125 .timeout_keepalive = timeout_keepalive,
126 .timeout_write_error = Timeout.init(5),
127 .conn_node_pool = .empty,
128 .retain_allocated_bytes_keepalive = retain_allocated_bytes_keepalive,
129 };
130 }
131
132 pub fn deinit(self: *Self) void {
133 const allocator = self.allocator;
134
135 self.websocket.deinit();
136 self.thread_pool.deinit();
137 allocator.destroy(self.websocket);
138
139 self.http_conn_pool.deinit();
140 self.conn_node_pool.deinit(allocator);
141
142 self.buffer_pool.deinit();
143 allocator.destroy(self.buffer_pool);
144 }
145
146 pub fn listen(self: *Self, listener: posix.socket_t) void {
147 const io = self.io;
148 var thread_pool = &self.thread_pool;
149 while (true) {
150 var address: posix.Address = undefined;
151 var address_len: posix.socklen_t = @sizeOf(posix.Address);
152 const socket = posix.accept(listener, &address.any, &address_len, posix.CLOEXEC) catch |err| {
153 if (err == error.ConnectionAborted or err == error.SocketNotListening) {
154 self.websocket.shutdown();
155 break;
156 }
157 log.err("Failed to accept socket: {}", .{err});
158 continue;
159 };
160 metrics.connection();
161 // calls handleConnection through the server's thread_pool
162 thread_pool.spawnOne(.{ self, socket, address });
163 }
164
165 {
166 self.mut.lockUncancelable(io);
167 defer self.mut.unlock(io);
168 var node = self.connections.head;
169 while (node) |n| {
170 node = n.next;
171 posix.close(n.socket);
172 }
173 }
174 thread_pool.stop();
175 }
176
177 pub fn stop(self: *const Self) void {
178 // The HTTP server will stop when the http.Server shutdown the listening socket.
179 self.websocket.shutdown();
180 }
181
182 // Called in a worker thread. `thread_buf` is a thread-specific buffer that
183 // we are free to use as needed.
184 pub fn handleConnection(self: *Self, socket: posix.socket_t, address: posix.Address, thread_buf: []u8) void {
185 const io = self.io;
186 const connection_node = blk: {
187 self.mut.lockUncancelable(io);
188 defer self.mut.unlock(io);
189 const node = self.conn_node_pool.create(self.allocator) catch |err| {
190 log.err("Failed to initialize connection node: {}", .{err});
191 return;
192 };
193 node.* = .{
194 .next = null,
195 .prev = null,
196 .socket = socket,
197 };
198 self.connections.insert(node);
199 break :blk node;
200 };
201
202 defer {
203 self.mut.lockUncancelable(io);
204 defer self.mut.unlock(io);
205 self.connections.remove(connection_node);
206 self.conn_node_pool.destroy(connection_node);
207 }
208
209 var conn = self.http_conn_pool.acquire() catch |err| {
210 log.err("Failed to initialize connection: {}", .{err});
211 return;
212 };
213
214 const ip_address = address.toIOAddress();
215 conn.address = ip_address;
216 conn.handover = .unknown;
217 conn.stream = .{ .socket = .{ .handle = socket, .address = ip_address } };
218
219 var is_keepalive = false;
220 while (true) {
221 switch (self.handleRequest(conn, is_keepalive, thread_buf) catch .close) {
222 .keepalive => {
223 is_keepalive = true;
224 conn.requestDone(self.retain_allocated_bytes_keepalive, false) catch unreachable;
225 },
226 .close, .unknown => {
227 posix.close(socket);
228 // impossible for this to fail in blocking mode
229 conn.requestDone(self.retain_allocated_bytes_keepalive, false) catch unreachable;
230 self.http_conn_pool.release(conn);
231 return;
232 },
233 .websocket => |ptr| {
234 const hc: *ws.HandlerConn(WSH) = @ptrCast(@alignCast(ptr));
235 // impossible for this to fail in blocking mode
236 conn.requestDone(self.retain_allocated_bytes_keepalive, false) catch unreachable;
237 self.http_conn_pool.release(conn);
238 // blocking read loop
239 // will close the connection
240 self.handleWebSocket(hc) catch |err| {
241 log.err("({f} websocket connection error: {}", .{ address, err });
242 };
243 return;
244 },
245 .disown => {
246 // impossible for this to fail in blocking mode
247 conn.requestDone(self.retain_allocated_bytes_keepalive, false) catch unreachable;
248 self.http_conn_pool.release(conn);
249 return;
250 },
251 }
252 }
253 }
254
255 fn handleRequest(self: *const Self, conn: *HTTPConn, is_keepalive: bool, thread_buf: []u8) !HTTPConn.Handover {
256 const io = self.io;
257 const socket = conn.stream.socket.handle;
258 const timeout: ?Timeout = if (is_keepalive) self.timeout_keepalive else self.timeout_request;
259
260 var deadline: ?i64 = null;
261
262 if (timeout) |to| {
263 if (is_keepalive == false) {
264 deadline = timestamp(io) + to.sec;
265 }
266 try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &to.timeval);
267 }
268
269 var is_first = true;
270 while (true) {
271 const done = conn.req_state.parse(conn, socket) catch |err| {
272 switch (err) {
273 error.WouldBlock => {
274 if (is_keepalive and is_first) {
275 metrics.timeoutKeepalive(1);
276 } else {
277 metrics.timeoutRequest(1);
278 }
279 return .close;
280 },
281 error.NotOpenForReading => {
282 // This can only happen when we're shutting down and our
283 // listener has called posix.close(socket) to unblock
284 // this thread. Using `.disown` is a bit of a hack, but
285 // disown is handled in handleConnection the way we want
286 // WE DO NOT WANT to return .close, else that would result
287 // in posix.close(socket) being called on an already-closed
288 // socket, which would panic.
289 return .disown;
290 },
291 else => {},
292 }
293 requestError(conn, err) catch {};
294 posix.close(socket);
295 return .disown;
296 };
297
298 if (done) {
299 // we have a complete request, time to process it
300 break;
301 }
302
303 if (is_keepalive) {
304 if (is_first) {
305 if (self.timeout_request) |to| {
306 // This was the first data from a keepalive request.
307 // We would have been on a keepalive timeout (if there was one),
308 // and now need to switch to a request timeout. This might be
309 // an actual timeout, or it could just be removing the keepalive timeout
310 // either way, it's the same code (timeval will just be set to 0 for
311 // the second case)
312 deadline = timestamp(io) + to.sec;
313 try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &to.timeval);
314 }
315 is_first = false;
316 }
317 } else if (deadline) |dl| {
318 if (timestamp(io) > dl) {
319 metrics.timeoutRequest(1);
320 return .close;
321 }
322 }
323 }
324
325 metrics.request();
326 self.server.handleRequest(conn, thread_buf);
327 return conn.handover;
328 }
329
330 fn handleWebSocket(self: *const Self, hc: *ws.HandlerConn(WSH)) !void {
331 posix.setsockopt(hc.socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &std.mem.toBytes(posix.timeval{ .sec = 0, .usec = 0 })) catch |err| {
332 self.websocket.cleanupConn(hc);
333 return err;
334 };
335 // closes the connection before returning
336 return self.websocket.worker.readLoop(hc);
337 }
338 };
339}
340
341// This is a NonBlocking worker. We have N workers, each accepting connections
342// and largely working in isolation from each other (the only thing they share
343// is the *const config, a reference to the Server and to the Websocket server).
344// The bulk of the code in this file exists to support the NonBlocking Worker.
345// Our listening socket is nonblocking
346// Request sockets are blocking. WHAT? We'll use epoll/kqueue to know when
347// a read wouldn't block. But we want writes to block because we want a nice
348// and easy API for the app by giving their response a predictable/controllable
349// lifetime.
350// A previous version had the sockets in nonblocking and would switch to blocking
351// as necessary, but I don't see what value that adds. It's just more syscalls
352// and complexity.
353pub fn NonBlocking(comptime S: type, comptime WSH: type) type {
354 return struct {
355 io: Io,
356
357 // Reference to the httpz.Server. After we've parsed the request we
358 // call its handleRequest method.
359 server: S,
360
361 // The allocator passed to the server
362 allocator: Allocator,
363
364 // KQueue or Epoll, depending on the platform
365 loop: Loop,
366
367 // The maximum number of connection we should have at any given time
368 // This is specific to this worker (it's a per-worker config).
369 max_conn: usize,
370
371 // # of connections this worker is handling.
372 len: usize,
373
374 // whether or not the worker is full (len == max_conn)
375 full: bool,
376
377 config: *const Config,
378
379 websocket: *ws.Worker(WSH),
380
381 // how many bytes should we retain in a connection's arena allocator
382 retain_allocated_bytes: usize,
383
384 // The thread pool that'll actually handle any incoming data. This is what
385 // will call server.handleRequest which will eventually call the application
386 // action.
387 thread_pool: ThreadPool(Self.processData),
388
389 // List of connections waiting on for more data to complete a request
390 // (we've gotten some data, but not enough to service the request).
391 // This is only ever manipulated directly from the worker thread
392 // so doesn't need thread safety.
393 request_list: List(Conn(WSH)),
394
395 // Connections which have had at least 1 sucessful request and response
396 // and which we are now waiting for another request.
397 // This is only ever manipulated directly from the worker thread
398 // so doesn't need thread safety.
399 keepalive_list: ConcurrentList(Conn(WSH)),
400
401 // Requests currently being processed.
402 // The worker thread moves the connnection here, but the thread pool
403 // will remove it and put it into handover_list, so this has to be
404 // thread safe.
405 active_list: ConcurrentList(Conn(WSH)),
406
407 // List of connections which have had a request->response fully handled
408 // and are now being handed back to the worker (from the thread pool) to
409 // process, essentially telling the worker to handle the conn.handover case.
410 // The thread pool inserts into this (so it has to be thread safe for
411 // that alone) and the worker thread will process it.
412 handover_list: ConcurrentList(Conn(WSH)),
413
414 // Conn nodes whose websocket connection died in the thread pool. The
415 // pool thread can't release them itself — conn_mem_pool and len are
416 // reactor-owned — so it parks them here and signals the worker, which
417 // drains this after consuming the current event batch. Without this,
418 // every websocket
419 // connection permanently consumed a conn slot and the worker stopped
420 // accepting once len reached max_conn.
421 ws_graveyard: ConcurrentList(Conn(WSH)),
422
423 // A pool of HTTPConn objects. The pool maintains a configured min # of these.
424 // An HTTPConn is relatively expensive, since we pre-allocate all types
425 // of things (a Request.State and Response.State).
426 http_conn_pool: HTTPConnPool,
427
428 // The bulk of this code works with an *HTTPConn, but we wrap it in a
429 // Conn(WSH) because an actual connection can either be an HTTPConn or
430 // a ws.HandlerConn. So Conn(WSH) is essentially a union between these two.
431 // It also has a next/prev, allowing it to be inserted in the above 4
432 // lists (requets/keepalive/active/handover).
433 // It's lightweight enough that we can use a MemoryPool and don't need
434 // the fancier management of something like HTTPConnPool for the HTTPConns
435 conn_mem_pool: std.heap.MemoryPool(Conn(WSH)),
436
437 // Request and response processing may require larger buffers than the static
438 // buffered of our req/res states. The BufferPool has larger pre-allocated
439 // buffers that can be used and, when empty or when a larger buffer is needed,
440 // will do dynamic allocation
441 buffer_pool: *BufferPool,
442
443 // The timeout, in seconds, that connections in the request phase (in the
444 // request_list) will timeout. Entries in request_list are sorted with
445 // connections that are closest to timeing out at the head.
446 timeout_request: u32,
447
448 // The timeout, in seconds, that connections in the keepalive phase (in the
449 // keepalive_list will timeout. Entries in keepalive_list are sorted with
450 // connections that are closest to timeing out at the head.
451 timeout_keepalive: u32,
452
453 const Self = @This();
454
455 const Loop = switch (loopType()) {
456 .kqueue => KQueue(WSH),
457 .epoll => EPoll(WSH),
458 };
459
460 pub fn init(io: Io, allocator: Allocator, server: S, config: *const Config) !Self {
461 const loop = try Loop.init();
462 errdefer loop.deinit();
463
464 const websocket = try allocator.create(ws.Worker(WSH));
465 errdefer allocator.destroy(websocket);
466 websocket.* = try ws.Worker(WSH).init(allocator, &server._websocket_state);
467 errdefer websocket.deinit();
468
469 var buffer_pool = try initializeBufferPool(io, allocator, config);
470 errdefer buffer_pool.deinit();
471
472 var conn_mem_pool: std.heap.MemoryPool(Conn(WSH)) = .empty;
473 errdefer conn_mem_pool.deinit(allocator);
474
475 var http_conn_pool = try HTTPConnPool.init(io, allocator, buffer_pool, websocket, loop.fd, config);
476 errdefer http_conn_pool.deinit();
477
478 const thread_pool = try ThreadPool(Self.processData).init(io, allocator, .{
479 .count = config.threadPoolCount(),
480 .backlog = config.thread_pool.backlog orelse 500,
481 .buffer_size = config.thread_pool.buffer_size orelse 32_768,
482 });
483
484 errdefer {
485 thread_pool.stop();
486 thread_pool.deinit();
487 }
488
489 return .{
490 .io = io,
491 .len = 0,
492 .full = false,
493 .loop = loop,
494 .config = config,
495 .server = server,
496 .allocator = allocator,
497 .websocket = websocket,
498 .thread_pool = thread_pool,
499 .active_list = .{},
500 .request_list = .{},
501 .handover_list = .{},
502 .ws_graveyard = .{},
503 .keepalive_list = .{},
504 .buffer_pool = buffer_pool,
505 .conn_mem_pool = conn_mem_pool,
506 .http_conn_pool = http_conn_pool,
507 .max_conn = config.workers.max_conn orelse 8_192,
508 .timeout_request = config.timeout.request orelse MAX_TIMEOUT,
509 .timeout_keepalive = config.timeout.keepalive orelse MAX_TIMEOUT,
510 .retain_allocated_bytes = config.workers.retain_allocated_bytes orelse 8192,
511 };
512 }
513
514 pub fn deinit(self: *Self) void {
515 const allocator = self.allocator;
516
517 // Join in-flight handlers before freeing ANY connection state they
518 // may still be reading. run() only stops the event loop, so a handler
519 // can still be mid-callback here. This must come before
520 // websocket.deinit(), which frees the per-connection read buffers via
521 // buffer_provider.deinit(): a handler still in dataAvailable() can be
522 // reading a parsed message that holds zero-copy slices into those
523 // buffers, so joining afterward would be too late -> use-after-free.
524 self.thread_pool.stop();
525
526 self.websocket.deinit();
527 allocator.destroy(self.websocket);
528
529 self.thread_pool.deinit();
530
531 self.shutdownList(&self.request_list);
532 self.shutdownConcurrentList(&self.active_list);
533 self.shutdownConcurrentList(&self.handover_list);
534 self.shutdownConcurrentList(&self.keepalive_list);
535
536 self.buffer_pool.deinit();
537 self.conn_mem_pool.deinit(allocator);
538 self.http_conn_pool.deinit();
539 allocator.destroy(self.buffer_pool);
540
541 self.loop.deinit();
542 }
543
544 pub fn stop(self: *Self) void {
545 // causes run to break out of its loop
546 self.loop.stop();
547 }
548
549 pub fn run(self: *Self, listener: posix.fd_t, ready_sem: *Io.Semaphore) void {
550 const io = self.io;
551 var thread_pool = &self.thread_pool;
552
553 self.loop.start() catch |err| {
554 log.err("Failed to start event loop: {}", .{err});
555 return;
556 };
557
558 // Whether this fails or succeeds we can confidently signal upstream
559 // that we're ready enough to be stopped in necessary.
560 self.loop.monitorAccept(listener) catch |err| {
561 log.err("Failed to add monitor to listening socket: {}", .{err});
562 ready_sem.post(io);
563 return;
564 };
565 ready_sem.post(io);
566 defer self.websocket.shutdown();
567
568 var now = timestamp(io);
569 var last_timeout = now;
570 while (true) {
571 var timeout: ?i32 = 1;
572 if (now - last_timeout > 1) {
573
574 // we don't all prepareToWait more than once per second
575 const had_timeouts, timeout = self.prepareToWait(now);
576 if (had_timeouts and self.full) {
577 self.enableListener(listener);
578 }
579 last_timeout = now;
580 }
581
582 var it = self.loop.wait(timeout) catch |err| {
583 log.err("Failed to wait on events: {}", .{err});
584 io.sleep(.fromMilliseconds(100), .awake) catch |err2| {
585 log.err("Failed to do a mini recovery sleep: {}", .{err2});
586 };
587 continue;
588 };
589
590 now = timestamp(io);
591 var closed_conn = false;
592 var retired_http: List(Conn(WSH)) = .{};
593
594 // Defer signal handling (which disowns/frees handed-over
595 // connections) until the whole event batch is drained. epoll can
596 // return a .signal and a .recv for the same connection in one
597 // batch; processing the signal first would free the connection,
598 // and the later .recv would then dereference freed memory in
599 // getState(). By deferring, every .recv runs while its connection
600 // is still alive, sees the .handover state set before the signal,
601 // and skips it; the free happens safely afterward.
602 var has_signal = false;
603
604 while (it.next()) |event| {
605 switch (event) {
606 .accept => self.accept(listener, now) catch |err| {
607 log.err("Failed to accept connection: {}", .{err});
608 io.sleep(.fromMilliseconds(5), .awake) catch |err2| {
609 log.err("Failed to do a mini recovery sleep: {}", .{err2});
610 };
611 },
612 // Signal handling can recycle Conn storage. Defer it
613 // until every recv pointer in this batch is consumed.
614 .signal => has_signal = true,
615 .recv => |conn| switch (conn.protocol) {
616 .http => |http_conn| {
617 switch (http_conn.getState()) {
618 .request, .keepalive => {},
619 .active, .handover => {
620 // we need to finish whatever we're doing
621 // before we can process more data (i.e. if
622 // the connection is being upgrade to websocket,
623 // this will be a websocket message.)
624 continue;
625 },
626 }
627
628 // At this point, the connection is either in
629 // keepalive or request. Either way, we know no
630 // other thread is accessing the connection, so we
631 // can access _state directly.
632
633 const stream = http_conn.stream;
634 var reader = stream.reader(io, &.{}); // Request.State does its own buffering
635 const done = http_conn.req_state.parse(http_conn, &reader.interface) catch |err| {
636 // maybe a write fail or something, doesn't matter, we're closing the connection
637 requestError(http_conn, reader.err orelse err) catch {};
638
639 // impossible to fail when false is passed
640 http_conn.requestDone(self.retain_allocated_bytes, false) catch unreachable;
641 self.loop.remove(conn);
642 conn.close();
643 self.retireHttp(conn, &retired_http);
644 closed_conn = true;
645 continue;
646 };
647
648 if (done == false) {
649 // A keepalive deadline only covers the idle
650 // gap between requests. As soon as bytes for
651 // the next request arrive, give that request
652 // the full request timeout. Keeping the old
653 // deadline can expire a split header/body in
654 // the middle of parsing.
655 if (http_conn.getState() == .keepalive) {
656 http_conn.timeout = now + self.timeout_request;
657 }
658 self.swapList(conn, .request);
659 continue;
660 }
661
662 self.swapList(conn, .active);
663 thread_pool.spawn(.{ self, now, conn });
664 },
665 .websocket => {
666 if (conn.acquireProcessing() == false) {
667 // Connection is already being processed. We need
668 // to wait for the current processing to complete.
669 // See the processing field in Conn
670 continue;
671 }
672 thread_pool.spawn(.{ self, now, conn });
673 },
674 // Stable tombstone for WebSocket userdata whose fd
675 // was closed on a pool thread. A queued kernel event
676 // may still carry this pointer; it must be ignored.
677 .retired => continue,
678 },
679 .shutdown => return,
680 }
681 }
682
683 // Reclamation follows signal handling only after every event in
684 // this batch has stopped carrying a Conn pointer.
685 if (has_signal) self.processSignal(&closed_conn);
686 self.releaseRetiredHttp(&retired_http);
687 if (self.drainWebsocketGraveyard()) closed_conn = true;
688 const batch_size = thread_pool.batch_size;
689 if (batch_size > 0) {
690 thread_pool.flush(batch_size);
691 }
692
693 if (self.full and closed_conn) {
694 self.enableListener(listener);
695 }
696 }
697 }
698
699 fn swapList(self: *Self, conn: *Conn(WSH), new_state: HTTPConn.State) void {
700 const io = self.io;
701 const http_conn = conn.protocol.http;
702 http_conn._mut.lockUncancelable(io);
703 defer http_conn._mut.unlock(io);
704
705 switch (http_conn._state) {
706 .active => self.active_list.remove(io, conn),
707 .keepalive => self.keepalive_list.remove(io, conn),
708 .request => self.request_list.remove(conn),
709 .handover => self.handover_list.remove(io, conn),
710 }
711
712 http_conn.setState(new_state);
713
714 switch (new_state) {
715 .active => self.active_list.insert(io, conn),
716 .keepalive => self.keepalive_list.insert(io, conn),
717 .request => self.request_list.insert(conn),
718 .handover => self.handover_list.insert(io, conn),
719 }
720 }
721
722 fn accept(self: *Self, listener: posix.fd_t, now: u32) !void {
723 var len = self.len;
724 const max_conn = self.max_conn;
725
726 // Don't try to disconnect keepalive connections when len == max_conn.
727 // If you do, you'll get hard-to-reproduce segfaults. Why?
728 // This accept function is being called our event loop. If we call
729 // self.closeClose(keepalive_list.head) here, there's a chance that
730 // the connection which we're closing is in our poll event.
731 // And if it is, we'll segfault when we try to do anything with that
732 // connection after freeing it here.
733 while (true) {
734 if (len == max_conn) {
735 self.full = true;
736 self.loop.pauseAccept(listener) catch {};
737 return;
738 }
739 var address: posix.Address = undefined;
740 var address_len: posix.socklen_t = @sizeOf(posix.Address);
741
742 const socket = posix.accept(listener, &address.any, &address_len, posix.CLOEXEC | posix.NONBLOCK) catch |err| {
743 // On BSD, REUSEPORT_LB means that only 1 worker should get notified
744 // of a connetion. On Linux, however, we only have REUSEPORT, which will
745 // notify all workers. However, we monitor the listener using EPOLLEXCLUSIVE.
746 // This makes it so that "one or more" workers receive it.
747 // In other words, no guarantee that there'll just be 1, but Linux will try
748 // to minimze the count.
749 // In the end, when we call accept, we might get a WouldBlock because
750 // Linux can wake up multiple epoll fds for a single connection.
751 return if (err == error.WouldBlock) {} else err;
752 };
753 errdefer posix.close(socket);
754 metrics.connection();
755
756 const socket_flags = try posix.fcntl(socket, posix.F.GETFL, 0);
757 const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
758 std.debug.assert(socket_flags & nonblocking == nonblocking);
759
760 const conn = try self.conn_mem_pool.create(self.allocator);
761 errdefer self.conn_mem_pool.destroy(conn);
762
763 const ip_address = address.toIOAddress();
764
765 const http_conn = try self.http_conn_pool.acquire();
766 http_conn.request_count = 1;
767 http_conn._state = .request;
768 http_conn.handover = .unknown;
769 http_conn._io_mode = .nonblocking;
770 http_conn.address = ip_address;
771 http_conn.socket_flags = socket_flags;
772 http_conn.stream = .{ .socket = .{ .handle = socket, .address = ip_address } };
773 http_conn.timeout = now + self.timeout_request;
774
775 self.len += 1;
776 conn.* = .{
777 .next = null,
778 .prev = null,
779 .protocol = .{ .http = http_conn },
780 };
781 self.request_list.insert(conn);
782 errdefer {
783 conn.close();
784 self.disown(conn);
785 }
786
787 try self.loop.monitorRead(conn);
788 len += 1;
789 }
790 }
791
792 fn processSignal(self: *Self, closed_bool: *bool) void {
793 const io = self.io;
794 const loop = &self.loop;
795
796 var hl = &self.handover_list;
797
798 // We take the handover list, and then re-initialize it. We do this
799 // under lock, so that any thread-pool waiting to handover will either
800 // make it into this snapshot, or into the new list.
801 // The benefit of this approach is that we don't ahve to hold the
802 // handover_list mutex for very long, just long enough to grab the
803 // head and re-initialize the inner list.
804 // This is ok, because _every_ connetion in the handover list is
805 // going to end up in another list (or closed) by the time we're
806 // done.
807 var c = blk: {
808 hl.mut.lockUncancelable(io);
809 defer hl.mut.unlock(io);
810 const head = hl.inner.head;
811 hl.inner = .{};
812 break :blk head;
813 };
814
815 // Every node in this snapshot was detached from handover_list when
816 // the inner list was re-initialized above. They must be released
817 // with releaseDetached, never disown: disown runs List.remove
818 // using the node's stale snapshot links, which grafts the rest of
819 // the snapshot back onto the live list. Those connections then get
820 // processed a second time on the next signal — reading a websocket
821 // union as http, double-releasing pooled HTTPConns, and ultimately
822 // interleaving responses across unrelated sockets.
823 while (c) |conn| {
824 c = conn.next;
825 const http_conn = conn.protocol.http;
826 switch (http_conn.handover) {
827 .close, .unknown => {
828 closed_bool.* = true;
829 // Keep close, event-loop removal, and storage recycling
830 // on the reactor thread, in that order. Closing on the
831 // handler thread can leave epoll armed with this Conn
832 // pointer after the allocator reuses its storage.
833 loop.remove(conn);
834 conn.close();
835 self.releaseDetached(conn);
836 },
837 .disown => {
838 // When res.disown() was called, we immediately removed
839 // the socket from the event loop. This is necessary
840 // because the new owner of the socket might close it
841 // before we get back here.
842 // https://github.com/karlseguin/http.zig/issues/129#issuecomment-3031411404
843 closed_bool.* = true;
844 self.releaseDetached(conn);
845 },
846 .websocket => |ptr| {
847 if (comptime WSH == httpz.DummyWebsocketHandler) {
848 std.debug.print("Your httpz handler must have a `WebsocketHandler` declaration. This must be the same type passed to `httpz.upgradeWebsocket`. Closing the connection.\n", .{});
849 closed_bool.* = true;
850 loop.remove(conn);
851 conn.close();
852 self.releaseDetached(conn);
853 continue;
854 }
855
856 const hc: *ws.HandlerConn(WSH) = @ptrCast(@alignCast(ptr));
857 loop.switchToOneShot(conn) catch {
858 metrics.internalError();
859 closed_bool.* = true;
860 loop.remove(conn);
861 conn.close();
862 self.releaseDetached(conn);
863 continue;
864 };
865
866 self.http_conn_pool.release(http_conn);
867 conn.protocol = .{ .websocket = hc };
868 conn.next = null;
869 conn.prev = null;
870 },
871 .keepalive => unreachable,
872 }
873 }
874 }
875
876 fn releaseRetiredHttp(self: *Self, retired: *List(Conn(WSH))) void {
877 var conn = retired.head;
878 while (conn) |c| {
879 conn = c.next;
880 self.releaseDetached(c);
881 }
882 retired.* = .{};
883 }
884
885 fn drainWebsocketGraveyard(self: *Self) bool {
886 var g = blk: {
887 const gl = &self.ws_graveyard;
888 gl.mut.lockUncancelable(self.io);
889 defer gl.mut.unlock(self.io);
890 const head = gl.inner.head;
891 gl.inner = .{};
892 break :blk head;
893 };
894 const released = g != null;
895 while (g) |conn| {
896 g = conn.next;
897 const hc = conn.protocol.websocket;
898 // The websocket worker has already closed the socket. Closing
899 // removes its kqueue/epoll registration; deleting by numeric fd
900 // here is unsafe because accept may already have reused that fd
901 // for another connection. In that race, this DEL removed the new
902 // connection's read monitor and left it stuck in request_list.
903 // More importantly, close happens on another thread, so the
904 // reactor cannot prove that a readiness notification carrying
905 // this userdata pointer is not still queued beyond the current
906 // epoll_wait batch. Never recycle websocket Conn userdata: an
907 // old notification must continue to point at retired websocket
908 // state, never at a newly accepted HTTP connection (the ABA
909 // that cross-wired responses in production). MemoryPool.deinit
910 // releases these tiny stable-lifetime nodes at server shutdown.
911 self.websocket.cleanupConn(hc);
912 conn.protocol = .{ .retired = {} };
913 self.len -= 1;
914 }
915 return released;
916 }
917
918 // Entry-point of our thread pool. `thread_buf` is a thread-specific buffer
919 // that we are free to use as needed.
920 // Currently, for an HTTP connection, this is only called when we have a
921 // full HTTP request ready to process. For an WebSocket packet, it's called
922 // when we have data available - there may or may not be a full message ready
923 pub fn processData(self: *Self, now: u32, conn: *Conn(WSH), thread_buf: []u8) void {
924 switch (conn.protocol) {
925 .http => |http_conn| self.processHTTPData(now, conn, thread_buf, http_conn),
926 .websocket => |hc| self.processWebsocketData(conn, thread_buf, hc),
927 .retired => {},
928 }
929 }
930
931 pub fn processHTTPData(self: *Self, now: u32, conn: *Conn(WSH), thread_buf: []u8, http_conn: *HTTPConn) void {
932 metrics.request();
933 http_conn.request_count += 1;
934 self.server.handleRequest(http_conn, thread_buf);
935
936 var handover = http_conn.handover;
937 http_conn.requestDone(self.retain_allocated_bytes, handover == .keepalive or handover == .websocket) catch {
938 // This means we failed to put the connection into
939 // nonblocking mode. Rare, but safer to close the connection
940 // at this point.
941 handover = .close;
942 };
943
944 switch (handover) {
945 .keepalive => {
946 http_conn.timeout = now + self.timeout_keepalive;
947 self.swapList(conn, .keepalive);
948 return;
949 },
950 .close, .unknown => {
951 // Closing the socket (which removes it from epoll/kqueue) is
952 // deferred to processSignal on the event-loop thread. Closing
953 // here, on a worker thread, raced epoll_wait: the fd could
954 // stay armed past the point where disown() recycled the Conn,
955 // so a later batch delivered a .recv carrying a pointer to
956 // freed memory (getState on a null/recycled HTTPConn). The
957 // signal-vs-recv-in-one-batch case this used to guard against
958 // is now handled by deferring signal processing to the end of
959 // the event batch plus the getState() .handover check in recv.
960 },
961 .websocket, .disown => {},
962 }
963 self.swapList(conn, .handover);
964 self.loop.signal() catch |err| log.err("failed to signal worker: {}", .{err});
965 }
966
967 pub fn processWebsocketData(self: *Self, conn: *Conn(WSH), thread_buf: []u8, hc: *ws.HandlerConn(WSH)) void {
968 var ws_conn = &hc.conn;
969 const success = self.websocket.worker.dataAvailable(hc, thread_buf);
970 if (success == false) {
971 ws_conn.close(.{ .code = 4997, .reason = "wsz" }) catch {};
972 return self.wsConnDied(conn, hc);
973 }
974 if (ws_conn.isClosed()) {
975 return self.wsConnDied(conn, hc);
976 }
977
978 // EPOLLONESHOT must not be re-armed while `processing` is still
979 // held. A readable event in that window would consume the one-shot,
980 // fail acquireProcessing(), and leave the websocket permanently
981 // unarmed. Release first so the event can dispatch another worker.
982 conn.releaseProcessing();
983 self.loop.rearmRead(conn) catch |err| {
984 log.debug("({f}) failed to add read event monitor: {}", .{ ws_conn.address, err });
985 ws_conn.close(.{ .code = 4998, .reason = "wsz" }) catch {};
986 return self.wsConnDied(conn, hc);
987 };
988 }
989
990 // The websocket connection is dead and its socket has been closed. Hand
991 // both the websocket state and Conn node to the reactor for reclamation
992 // after the current kernel-event batch no longer carries its pointer.
993 // Callers must not touch the node after parking it here. A terminal read
994 // keeps `processing` held; a failed re-arm has already released it but
995 // cannot produce another event because the one-shot remains unarmed.
996 fn wsConnDied(self: *Self, conn: *Conn(WSH), hc: *ws.HandlerConn(WSH)) void {
997 _ = hc;
998 self.ws_graveyard.insert(self.io, conn);
999 self.loop.signal() catch |err| log.err("failed to signal worker: {}", .{err});
1000 }
1001
1002 fn disown(self: *Self, conn: *Conn(WSH)) void {
1003 const http_conn = conn.protocol.http;
1004 self.detachHttp(conn);
1005 self.len -= 1;
1006 self.http_conn_pool.release(http_conn);
1007 self.conn_mem_pool.destroy(conn);
1008 }
1009
1010 // Remove an HTTP connection from its state list without recycling its
1011 // Conn/HTTPConn storage. Event-loop callers use this when the current
1012 // wait() batch may still contain another event carrying the Conn pointer.
1013 fn retireHttp(self: *Self, conn: *Conn(WSH), retired: *List(Conn(WSH))) void {
1014 self.detachHttp(conn);
1015 conn.protocol.http.setState(.handover);
1016 retired.insert(conn);
1017 }
1018
1019 fn detachHttp(self: *Self, conn: *Conn(WSH)) void {
1020 const io = self.io;
1021 const http_conn = conn.protocol.http;
1022 switch (http_conn._state) {
1023 .request => self.request_list.remove(conn),
1024 .handover => self.handover_list.remove(io, conn),
1025 .keepalive => self.keepalive_list.remove(io, conn),
1026 .active => unreachable,
1027 }
1028 }
1029
1030 // Enforces timeouts, and returns when the next timeout should be checked.
1031 fn prepareToWait(self: *Self, now: u32) struct { bool, ?i32 } {
1032 const request_timed_out, const request_count, const request_timeout = collectTimedOut(&self.request_list, now);
1033
1034 const io = self.io;
1035 const keepalive_timed_out, const keepalive_count, const keepalive_timeout = blk: {
1036 const list = &self.keepalive_list;
1037 list.mut.lockUncancelable(io);
1038 defer list.mut.unlock(io);
1039 break :blk collectTimedOut(&list.inner, now);
1040 };
1041
1042 var closed = false;
1043 if (request_count > 0) {
1044 closed = true;
1045 self.closeList(request_timed_out);
1046 metrics.timeoutRequest(request_count);
1047 }
1048 if (keepalive_count > 0) {
1049 closed = true;
1050 self.closeList(keepalive_timed_out);
1051 metrics.timeoutKeepalive(keepalive_count);
1052 }
1053
1054 if (request_timeout == null and keepalive_timeout == null) {
1055 return .{ closed, null };
1056 }
1057
1058 const next = @min(request_timeout orelse MAX_TIMEOUT, keepalive_timeout orelse MAX_TIMEOUT);
1059 if (next < now) {
1060 // can happen if a socket was just about to timeout when prepareToWait
1061 // was called
1062 return .{ closed, 1 };
1063 }
1064
1065 return .{ closed, @intCast(next - now) };
1066 }
1067
1068 // lists are ordered from soonest to timeout to last. Detach and return
1069 // the timed-out prefix; callers own the returned nodes and must not try
1070 // to remove them from the original list again.
1071 fn collectTimedOut(list: *List(Conn(WSH)), now: u32) struct { List(Conn(WSH)), usize, ?u32 } {
1072 var count: usize = 0;
1073 var conn = list.head;
1074
1075 while (conn) |c| {
1076 const timeout = c.protocol.http.timeout;
1077 if (timeout > now) {
1078 return .{ detachPrefix(Conn(WSH), list, count), count, timeout };
1079 }
1080 count += 1;
1081 conn = c.next;
1082 }
1083 return .{ detachPrefix(Conn(WSH), list, count), count, null };
1084 }
1085
1086 fn closeList(self: *Self, list: List(Conn(WSH))) void {
1087 var conn = list.head;
1088 while (conn) |c| {
1089 conn = c.next;
1090 // Timeout reclamation follows the same ownership rule as a
1091 // handler-requested close: the reactor must synchronously
1092 // remove the still-open descriptor before either the fd or
1093 // its Conn storage can be reused. Relying on close(2)'s
1094 // implicit epoll/kqueue cleanup leaves a queued event carrying
1095 // this pointer racing pool reuse under keepalive expiry.
1096 self.loop.remove(c);
1097 c.close();
1098 // request_list may contain a partially parsed request. Never
1099 // return that HTTPConn to the pool with its parser still
1100 // carrying the old method/path/body expectation: the next
1101 // accepted socket would otherwise be parsed as the remainder
1102 // of the timed-out request.
1103 c.protocol.http.requestDone(self.retain_allocated_bytes, false) catch unreachable;
1104 self.releaseDetached(c);
1105 }
1106 }
1107
1108 fn releaseDetached(self: *Self, conn: *Conn(WSH)) void {
1109 const http_conn = conn.protocol.http;
1110 self.len -= 1;
1111 self.http_conn_pool.release(http_conn);
1112 self.conn_mem_pool.destroy(conn);
1113 }
1114
1115 fn shutdownList(self: *Self, list: *List(Conn(WSH))) void {
1116 const allocator = self.allocator;
1117 var conn = list.head;
1118 while (conn) |c| {
1119 conn = c.next;
1120 const http_conn = c.protocol.http;
1121 posix.close(http_conn.stream.socket.handle);
1122 http_conn.deinit(allocator);
1123 }
1124 }
1125
1126 fn shutdownConcurrentList(self: *Self, list: *ConcurrentList(Conn(WSH))) void {
1127 const io = self.io;
1128 list.mut.lockUncancelable(io);
1129 defer list.mut.unlock(io);
1130 self.shutdownList(&list.inner);
1131 }
1132
1133 inline fn enableListener(self: *Self, listener: posix.fd_t) void {
1134 self.full = false;
1135 self.loop.monitorAccept(listener) catch |err| log.err("Failed to enable monitor to listening socket: {}", .{err});
1136 }
1137 };
1138}
1139
1140pub fn List(comptime T: type) type {
1141 return struct {
1142 head: ?*T = null,
1143 tail: ?*T = null,
1144
1145 const Self = @This();
1146
1147 pub fn insert(self: *Self, node: *T) void {
1148 if (self.tail) |tail| {
1149 tail.next = node;
1150 node.prev = tail;
1151 self.tail = node;
1152 } else {
1153 node.prev = null;
1154 self.head = node;
1155 self.tail = node;
1156 }
1157 node.next = null;
1158 }
1159
1160 pub fn remove(self: *Self, node: *T) void {
1161 if (node.prev) |prev| {
1162 prev.next = node.next;
1163 } else {
1164 self.head = node.next;
1165 }
1166
1167 if (node.next) |next| {
1168 next.prev = node.prev;
1169 } else {
1170 self.tail = node.prev;
1171 }
1172
1173 node.prev = null;
1174 node.next = null;
1175 }
1176 };
1177}
1178
1179pub fn ConcurrentList(comptime T: type) type {
1180 return struct {
1181 inner: List(T) = .{},
1182 mut: Io.Mutex = .init,
1183
1184 const Self = @This();
1185
1186 pub fn insert(self: *Self, io: Io, node: *T) void {
1187 self.mut.lockUncancelable(io);
1188 defer self.mut.unlock(io);
1189 self.inner.insert(node);
1190 }
1191
1192 pub fn remove(self: *Self, io: Io, node: *T) void {
1193 self.mut.lockUncancelable(io);
1194 defer self.mut.unlock(io);
1195 self.inner.remove(node);
1196 }
1197 };
1198}
1199
1200fn detachPrefix(comptime T: type, list: *List(T), count: usize) List(T) {
1201 if (count == 0) {
1202 return .{};
1203 }
1204
1205 const head = list.head.?;
1206 var tail = head;
1207 for (1..count) |_| {
1208 tail = tail.next.?;
1209 }
1210
1211 const next = tail.next;
1212 tail.next = null;
1213 head.prev = null;
1214
1215 if (next) |n| {
1216 n.prev = null;
1217 list.head = n;
1218 } else {
1219 list.head = null;
1220 list.tail = null;
1221 }
1222
1223 return .{ .head = head, .tail = tail };
1224}
1225
1226fn KQueue(comptime WSH: type) type {
1227 return struct {
1228 fd: i32,
1229 change_count: usize,
1230 change_buffer: [32]Kevent,
1231 event_list: [EVENT_BATCH_CAPACITY]Kevent,
1232
1233 const Self = @This();
1234 const Kevent = posix.Kevent;
1235
1236 fn init() !Self {
1237 return .{
1238 .fd = try posix.kqueue(),
1239 .change_count = 0,
1240 .change_buffer = undefined,
1241 .event_list = undefined,
1242 };
1243 }
1244
1245 fn deinit(self: *const Self) void {
1246 posix.close(self.fd);
1247 }
1248
1249 fn start(self: *Self) !void {
1250 try self.change(1, 1, posix.system.EVFILT.USER, posix.system.EV.ADD | posix.system.EV.CLEAR, posix.system.NOTE.FFNOP);
1251 return self.change(2, 2, posix.system.EVFILT.USER, posix.system.EV.ADD | posix.system.EV.ONESHOT, posix.system.NOTE.FFNOP);
1252 }
1253
1254 fn stop(self: *Self) void {
1255 // called from an arbitrary thread, can't use change
1256 _ = posix.kevent(self.fd, &.{
1257 .{
1258 .ident = 2,
1259 .filter = posix.system.EVFILT.USER,
1260 .flags = 0,
1261 .fflags = posix.system.NOTE.TRIGGER,
1262 .data = 0,
1263 .udata = 2,
1264 },
1265 }, &.{}, null) catch |err| {
1266 log.err("Failed to send stop signal: {}", .{err});
1267 };
1268 }
1269
1270 fn signal(self: *Self) !void {
1271 // called from thread pool thread, cant queue these in self.changes
1272 _ = try posix.kevent(self.fd, &.{.{
1273 .ident = 1,
1274 .filter = posix.system.EVFILT.USER,
1275 .flags = 0,
1276 .fflags = posix.system.NOTE.TRIGGER,
1277 .data = 0,
1278 .udata = 1,
1279 }}, &.{}, null);
1280 }
1281
1282 fn monitorAccept(self: *Self, fd: posix.fd_t) !void {
1283 try self.change(fd, 0, posix.system.EVFILT.READ, posix.system.EV.ENABLE | posix.system.EV.ADD, 0);
1284 }
1285
1286 fn pauseAccept(self: *Self, fd: posix.fd_t) !void {
1287 try self.change(fd, 0, posix.system.EVFILT.READ, posix.system.EV.DISABLE, 0);
1288 }
1289
1290 fn monitorRead(self: *Self, conn: *Conn(WSH)) !void {
1291 try self.change(conn.getSocket(), @intFromPtr(conn), posix.system.EVFILT.READ, posix.system.EV.ADD | posix.system.EV.ENABLE, 0);
1292 }
1293
1294 fn remove(self: *Self, conn: *Conn(WSH)) void {
1295 _ = posix.kevent(self.fd, &.{.{
1296 .ident = @intCast(conn.getSocket()),
1297 .filter = posix.system.EVFILT.READ,
1298 .flags = posix.system.EV.DELETE,
1299 .fflags = 0,
1300 .data = 0,
1301 .udata = 0,
1302 }}, &.{}, null) catch {};
1303 }
1304
1305 fn rearmRead(self: *Self, conn: *Conn(WSH)) !void {
1306 if (conn.isProcessing()) return error.ConnectionStillProcessing;
1307 // called from the worker thread, can't use change_buffer
1308 // must remove then re-add for macOS
1309 const socket = conn.getSocket();
1310 _ = try posix.kevent(self.fd, &.{ .{
1311 .ident = @intCast(socket),
1312 .filter = posix.system.EVFILT.READ,
1313 .flags = posix.system.EV.DELETE,
1314 .fflags = 0,
1315 .data = 0,
1316 .udata = 0,
1317 }, .{
1318 .ident = @intCast(socket),
1319 .filter = posix.system.EVFILT.READ,
1320 .flags = posix.system.EV.ADD | posix.system.EV.DISPATCH,
1321 .fflags = 0,
1322 .data = 0,
1323 .udata = @intFromPtr(conn),
1324 } }, &.{}, null);
1325 }
1326
1327 fn switchToOneShot(self: *Self, conn: *Conn(WSH)) !void {
1328 // From the Kqueue docs, you'd think you can just re-add the socket with EV.DISPATCH to enable
1329 // dispatching. But that _does not_ work on MacOS. Removing and Re-adding does.
1330 const socket = conn.getSocket();
1331 try self.change(socket, 0, posix.system.EVFILT.READ, posix.system.EV.DELETE, 0);
1332 try self.change(socket, @intFromPtr(conn), posix.system.EVFILT.READ, posix.system.EV.ADD | posix.system.EV.DISPATCH, 0);
1333 }
1334
1335 fn change(self: *Self, fd: posix.fd_t, data: usize, filter: i16, flags: u16, fflags: u16) !void {
1336 var change_count = self.change_count;
1337 var change_buffer = &self.change_buffer;
1338
1339 if (change_count == change_buffer.len) {
1340 // calling this with an empty event_list will return immediate
1341 _ = try posix.kevent(self.fd, change_buffer, &.{}, null);
1342 change_count = 0;
1343 }
1344 change_buffer[change_count] = .{
1345 .ident = @intCast(fd),
1346 .filter = filter,
1347 .flags = flags,
1348 .fflags = fflags,
1349 .data = 0,
1350 .udata = data,
1351 };
1352 self.change_count = change_count + 1;
1353 }
1354
1355 fn wait(self: *Self, timeout_sec: ?i32) !Iterator {
1356 const event_list = &self.event_list;
1357 const timeout: ?posix.timespec = if (timeout_sec) |ts| posix.timespec{ .sec = ts, .nsec = 0 } else null;
1358 const event_count = try posix.kevent(self.fd, self.change_buffer[0..self.change_count], event_list, if (timeout) |ts| &ts else null);
1359 self.change_count = 0;
1360
1361 return .{
1362 .index = 0,
1363 .loop = self,
1364 .events = event_list[0..event_count],
1365 };
1366 }
1367
1368 const Iterator = struct {
1369 loop: *Self,
1370 index: usize,
1371 events: []Kevent,
1372
1373 fn next(self: *Iterator) ?Event(WSH) {
1374 const index = self.index;
1375 const events = self.events;
1376 if (index == events.len) {
1377 return null;
1378 }
1379
1380 const event = &self.events[index];
1381 self.index = index + 1;
1382
1383 switch (event.udata) {
1384 0 => return .{ .accept = {} },
1385 1 => return .{ .signal = {} },
1386 2 => return .{ .shutdown = {} },
1387 else => |nptr| return .{ .recv = @ptrFromInt(nptr) },
1388 }
1389 }
1390 };
1391 };
1392}
1393
1394fn EPoll(comptime WSH: type) type {
1395 return struct {
1396 fd: i32,
1397 close_fd: i32,
1398 event_fd: i32,
1399 event_list: [EVENT_BATCH_CAPACITY]EpollEvent,
1400
1401 const Self = @This();
1402 const linux = std.os.linux;
1403 const EpollEvent = linux.epoll_event;
1404
1405 fn init() !Self {
1406 const close_fd = try posix.eventfd(0, std.os.linux.EFD.CLOEXEC | std.os.linux.EFD.NONBLOCK);
1407 errdefer posix.close(close_fd);
1408
1409 const event_fd = try posix.eventfd(0, std.os.linux.EFD.CLOEXEC | std.os.linux.EFD.NONBLOCK);
1410 errdefer posix.close(event_fd);
1411
1412 return .{
1413 .close_fd = close_fd,
1414 .event_fd = event_fd,
1415 .event_list = undefined,
1416 .fd = try posix.epoll_create1(0),
1417 };
1418 }
1419
1420 fn deinit(self: *const Self) void {
1421 posix.close(self.close_fd);
1422 posix.close(self.event_fd);
1423 posix.close(self.fd);
1424 }
1425
1426 fn start(self: *Self) !void {
1427 {
1428 var event = linux.epoll_event{
1429 .data = .{ .ptr = 2 },
1430 .events = linux.EPOLL.IN,
1431 };
1432 try posix.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, self.close_fd, &event);
1433 }
1434
1435 {
1436 var event = linux.epoll_event{
1437 .data = .{ .ptr = 1 },
1438 .events = linux.EPOLL.IN | linux.EPOLL.ET,
1439 };
1440 try posix.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, self.event_fd, &event);
1441 }
1442 }
1443
1444 fn stop(self: *Self) void {
1445 // eventfd requires exactly 8 bytes; use u64 so 32-bit (e.g. Linux ARM) does not get EINVAL.
1446 const increment: u64 = 1;
1447 _ = posix.write(self.close_fd, std.mem.asBytes(&increment)) catch |err| {
1448 log.err("Failed to write to closefd: {}", .{err});
1449 };
1450 }
1451
1452 fn signal(self: *const Self) !void {
1453 // eventfd requires exactly 8 bytes; use u64 so 32-bit (e.g. Linux ARM) does not get EINVAL.
1454 const increment: u64 = 1;
1455 _ = try posix.write(self.event_fd, std.mem.asBytes(&increment));
1456 }
1457
1458 fn monitorAccept(self: *Self, fd: posix.fd_t) !void {
1459 var event = linux.epoll_event{ .events = linux.EPOLL.IN | linux.EPOLL.EXCLUSIVE, .data = .{ .ptr = 0 } };
1460 return posix.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, fd, &event);
1461 }
1462
1463 fn pauseAccept(self: *Self, fd: posix.fd_t) !void {
1464 return posix.epoll_ctl(self.fd, linux.EPOLL.CTL_DEL, fd, null);
1465 }
1466
1467 fn monitorRead(self: *Self, conn: *Conn(WSH)) !void {
1468 var event = linux.epoll_event{
1469 .data = .{ .ptr = @intFromPtr(conn) },
1470 .events = linux.EPOLL.IN | linux.EPOLL.RDHUP,
1471 };
1472 return posix.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, conn.getSocket(), &event);
1473 }
1474
1475 fn remove(self: *Self, conn: *Conn(WSH)) void {
1476 posix.epoll_ctl(self.fd, linux.EPOLL.CTL_DEL, conn.getSocket(), null) catch {};
1477 }
1478
1479 fn rearmRead(self: *Self, conn: *Conn(WSH)) !void {
1480 if (conn.isProcessing()) return error.ConnectionStillProcessing;
1481 var event = linux.epoll_event{
1482 .data = .{ .ptr = @intFromPtr(conn) },
1483 .events = linux.EPOLL.IN | linux.EPOLL.RDHUP | linux.EPOLL.ONESHOT,
1484 };
1485 return posix.epoll_ctl(self.fd, linux.EPOLL.CTL_MOD, conn.getSocket(), &event);
1486 }
1487
1488 fn switchToOneShot(self: *Self, conn: *Conn(WSH)) !void {
1489 return self.rearmRead(conn);
1490 }
1491
1492 fn wait(self: *Self, timeout_sec: ?i32) !Iterator {
1493 const event_list = &self.event_list;
1494 var timeout: i32 = -1;
1495 if (timeout_sec) |sec| {
1496 if (sec > 2147483) {
1497 // max supported timeout by epoll_wait.
1498 timeout = 2147483647;
1499 } else {
1500 timeout = sec * 1000;
1501 }
1502 }
1503
1504 const event_count = posix.epoll_wait(self.fd, event_list, timeout);
1505 return .{
1506 .index = 0,
1507 .event_fd = self.event_fd,
1508 .events = event_list[0..event_count],
1509 };
1510 }
1511
1512 const Iterator = struct {
1513 index: usize,
1514 event_fd: i32,
1515 events: []EpollEvent,
1516
1517 fn next(self: *Iterator) ?Event(WSH) {
1518 const index = self.index;
1519 const events = self.events;
1520 if (index == events.len) {
1521 return null;
1522 }
1523 self.index = index + 1;
1524
1525 switch (events[index].data.ptr) {
1526 0 => return .{ .accept = {} },
1527 1 => return .{ .signal = {} },
1528 2 => return .{ .shutdown = {} },
1529 else => |nptr| return .{ .recv = @ptrFromInt(nptr) },
1530 }
1531 }
1532 };
1533 };
1534}
1535
1536fn Event(comptime WSH: type) type {
1537 return union(enum) {
1538 accept: void,
1539 signal: void,
1540 shutdown: void,
1541 recv: *Conn(WSH),
1542 };
1543}
1544
1545// There's some shared logic between the NonBlocking and Blocking workers.
1546// Whatever we can de-duplicate, goes here.
1547const HTTPConnPool = struct {
1548 io: Io,
1549 mut: Io.Mutex,
1550 conns: []*HTTPConn,
1551 available: usize,
1552 allocator: Allocator,
1553 config: *const Config,
1554 buffer_pool: *BufferPool,
1555 retain_allocated_bytes: usize,
1556 http_mem_pool_mut: Io.Mutex,
1557 http_mem_pool: std.heap.MemoryPool(HTTPConn),
1558
1559 // we erase the type because we don't want Conn, and therefore Request and
1560 // Response to have to carry this type around. This is all about making the
1561 // API cleaner.
1562 websocket: *anyopaque,
1563
1564 // Unfortunately, in some special cases, we need to interact with the loop
1565 // directly from a worker thread. Rather than store a *Loop, which could be
1566 // misused (not all of the methods are safe to call from the non-worker
1567 // thread), we store the loop's FD which is more opaque.
1568 loop: i32,
1569
1570 fn init(io: Io, allocator: Allocator, buffer_pool: *BufferPool, websocket: *anyopaque, loop: i32, config: *const Config) !HTTPConnPool {
1571 const min = config.workers.min_conn orelse @min(config.workers.max_conn orelse 64, 64);
1572
1573 var conns = try allocator.alloc(*HTTPConn, min);
1574 errdefer allocator.free(conns);
1575
1576 var http_mem_pool: std.heap.MemoryPool(HTTPConn) = .empty;
1577 errdefer http_mem_pool.deinit(allocator);
1578
1579 var initialized: usize = 0;
1580 errdefer {
1581 for (0..initialized) |i| {
1582 conns[i].deinit(allocator);
1583 }
1584 }
1585
1586 for (0..min) |i| {
1587 const conn = try http_mem_pool.create(allocator);
1588 conn.* = try HTTPConn.init(io, allocator, buffer_pool, websocket, loop, config);
1589
1590 conns[i] = conn;
1591 initialized += 1;
1592 }
1593
1594 return .{
1595 .io = io,
1596 .mut = .init,
1597 .loop = loop,
1598 .conns = conns,
1599 .config = config,
1600 .available = min,
1601 .websocket = websocket,
1602 .allocator = allocator,
1603 .buffer_pool = buffer_pool,
1604 .http_mem_pool = http_mem_pool,
1605 .http_mem_pool_mut = .init,
1606 .retain_allocated_bytes = config.workers.retain_allocated_bytes orelse 4096,
1607 };
1608 }
1609
1610 fn deinit(self: *HTTPConnPool) void {
1611 const allocator = self.allocator;
1612 // the rest of the conns are "checked out" and owned by the Manager
1613 // whichi will free them.
1614 for (self.conns[0..self.available]) |conn| {
1615 conn.deinit(allocator);
1616 }
1617 allocator.free(self.conns);
1618 self.http_mem_pool.deinit(allocator);
1619 }
1620
1621 fn acquire(self: *HTTPConnPool) !*HTTPConn {
1622 const io = self.io;
1623 const conns = self.conns;
1624
1625 self.lock();
1626 const available = self.available;
1627 if (available == 0) {
1628 self.unlock();
1629
1630 try self.http_mem_pool_mut.lock(io);
1631 const conn = try self.http_mem_pool.create(self.allocator);
1632 self.http_mem_pool_mut.unlock(io);
1633 errdefer {
1634 self.http_mem_pool_mut.lockUncancelable(io);
1635 self.http_mem_pool.destroy(conn);
1636 self.http_mem_pool_mut.unlock(io);
1637 }
1638
1639 conn.* = try HTTPConn.init(io, self.allocator, self.buffer_pool, self.websocket, self.loop, self.config);
1640 return conn;
1641 }
1642
1643 const index = available - 1;
1644 const conn = conns[index];
1645 self.available = index;
1646 self.unlock();
1647 return conn;
1648 }
1649
1650 fn release(self: *HTTPConnPool, conn: *HTTPConn) void {
1651 const conns = self.conns;
1652 self.lock();
1653 const available = self.available;
1654 if (available == conns.len) {
1655 self.unlock();
1656 conn.deinit(self.allocator);
1657
1658 const io = self.io;
1659 self.http_mem_pool_mut.lockUncancelable(io);
1660 self.http_mem_pool.destroy(conn);
1661 self.http_mem_pool_mut.unlock(io);
1662 return;
1663 }
1664
1665 conns[available] = conn;
1666 self.available = available + 1;
1667 self.unlock();
1668 }
1669
1670 // don't need thread safety in nonblocking
1671 fn lock(self: *HTTPConnPool) void {
1672 if (comptime httpz.blockingMode()) {
1673 self.mut.lockUncancelable(self.io);
1674 }
1675 }
1676
1677 // don't need thread safety in nonblocking
1678 fn unlock(self: *HTTPConnPool) void {
1679 if (comptime httpz.blockingMode()) {
1680 self.mut.unlock(self.io);
1681 }
1682 }
1683};
1684
1685pub fn Conn(comptime WSH: type) type {
1686 return struct {
1687 protocol: union(enum) {
1688 http: *HTTPConn,
1689 websocket: *ws.HandlerConn(WSH),
1690 retired: void,
1691 },
1692
1693 // Node in a List(WSH). List is [obviously] intrusive.
1694 next: ?*Conn(WSH),
1695 prev: ?*Conn(WSH),
1696
1697 // Used exclusively in the websocket path. In the HTTP Path, the
1698 // HTTPConn's state is used to prevent 2 requests on the same socket
1699 // from being processed at the same time. This is 100% necessary because
1700 // our event loop will keep emitting events for new data. For HTTP, this
1701 // is OK, because a "health" flow is REQ->RES - there shouldn't be
1702 // multiple inflight requests. NOT using ONESHOT/EV_DISPATCH is more
1703 // efficient because it avoids a bunch of system calls.
1704 // But for WebSockets, it's normal to have multiple incoming messages.
1705 // So we switch the event notifier to be ONESHOT/DISPATCH. And, that
1706 // SHOULD be enough to stop 2 concurrent messages from being processed
1707 // at the same time. And on Linux, it seems to work fine. But on MacOS,
1708 // there's a window where we might get 2 messages across 2 different
1709 // wait calls as we're switching to DISPATCH. So this guard is there just
1710 // for thar narrow window.
1711 processing: bool = false,
1712
1713 const Self = @This();
1714
1715 fn close(self: *Self) void {
1716 switch (self.protocol) {
1717 .http => |http_conn| posix.close(http_conn.stream.socket.handle),
1718 .websocket => |hc| hc.conn.close(.{}) catch {},
1719 .retired => {},
1720 }
1721 }
1722
1723 pub fn getSocket(self: Self) posix.fd_t {
1724 return switch (self.protocol) {
1725 .http => |hc| hc.stream.socket.handle,
1726 .websocket => |hc| hc.socket,
1727 .retired => unreachable,
1728 };
1729 }
1730
1731 pub fn acquireProcessing(self: *Self) bool {
1732 // returns true if it was previously false
1733 return @atomicRmw(bool, &self.processing, .Xchg, true, .monotonic) == false;
1734 }
1735
1736 pub fn releaseProcessing(self: *Self) void {
1737 return @atomicStore(bool, &self.processing, false, .release);
1738 }
1739
1740 pub fn isProcessing(self: *const Self) bool {
1741 return @atomicLoad(bool, &self.processing, .acquire);
1742 }
1743 };
1744}
1745
1746// Wraps a socket with a application-specific details, such as information needed
1747// to manage the connection's lifecycle (e.g. timeouts). Conns are placed in a
1748// linked list, hence the next/prev.
1749//
1750// The Conn can be re-used (as parf of a pool), either for keepalive, or for
1751// completely different tcp connections. From the Conn's point of view, there's
1752// no difference, just need to `reset` between each request.
1753//
1754// The Conn contains request and response state information necessary to operate
1755// in nonblocking mode. A pointer to the conn is the userdata passed to epoll/kqueue.
1756// Should only be created through the worker's HTTPConnPool
1757pub const HTTPConn = struct {
1758 // A connection can be in one of four states. This mostly corresponds to which
1759 // of the worker's lists it is in.
1760 // - active: There's a thread in the thread pool serving a request
1761 //
1762 // - handover: The threadpool is done with the request and wants to transfer
1763 // control back to the worker.
1764 //
1765 // - request: We have some data, but not enough to service the request. i.e.
1766 // we don't have a complete header or missing [part of] the body.
1767 // Connections here are subject to the request timeout.
1768 //
1769 // - keepalive: Conenction is between requests. We're waiting for the start
1770 // of the new request. connections here are suject to the keepalive timeout.
1771 const State = enum {
1772 active,
1773 request,
1774 handover,
1775 keepalive,
1776 };
1777
1778 pub const Handover = union(enum) {
1779 disown,
1780 close,
1781 unknown,
1782 keepalive,
1783 websocket: *anyopaque,
1784 };
1785
1786 pub const IOMode = enum {
1787 blocking,
1788 nonblocking,
1789 };
1790
1791 io: Io,
1792
1793 // can be concurrently accessed, use getState
1794 _state: State,
1795
1796 _mut: Io.Mutex,
1797
1798 _io_mode: IOMode,
1799
1800 handover: Handover,
1801
1802 // unix timestamp (seconds) where this connection should timeout
1803 timeout: u32,
1804
1805 // number of requests made on this connection (within a keepalive session)
1806 request_count: u64,
1807
1808 stream: Io.net.Stream,
1809 address: Io.net.IpAddress,
1810 socket_flags: usize,
1811
1812 // Data needed to parse a request. This contains pre-allocated memory, e.g.
1813 // as a read buffer and to store parsed headers. It also contains the state
1814 // necessary to parse the request over successive nonblocking read calls.
1815 req_state: Request.State,
1816
1817 // Data needed to create the response. This contains pre-allocate memory, .e.
1818 // header buffer to write the buffer. It also contains the state necessary
1819 // to write the response over successive nonblocking write calls.
1820 res_state: Response.State,
1821
1822 // Memory that is needed for the lifetime of a request, specifically from the
1823 // point where the request is parsed to after the response is sent, can be
1824 // allocated in this arena. An allocator for this arena is available to the
1825 // application as req.arena and res.arena.
1826 req_arena: std.heap.ArenaAllocator,
1827
1828 // Memory that is needed for the lifetime of the connection. In most cases
1829 // the connection outlives the request for two reasons: keepalive and the
1830 // fact that we keepa pool of connections.
1831 conn_arena: *std.heap.ArenaAllocator,
1832
1833 // This is our ws.Worker(WSH) but the type is erased so that Conn isn't
1834 // a generic. We don't want Conn to be a generic, because we don't want Response
1835 // to be generics since that would make using the library unecessarily complex
1836 // (especially since not everyone cares about websockets).
1837 ws_worker: *anyopaque,
1838
1839 // Unfortunately, in some special cases, we need to interact with the loop
1840 // directly from a worker thread. Rather than store a *Loop, which could be
1841 // misused (not all of the methods are safe to call from the non-worker
1842 // thread), we store the loop's FD which is more opaque.
1843 loop: i32,
1844
1845 fn init(io: Io, allocator: Allocator, buffer_pool: *BufferPool, ws_worker: *anyopaque, loop: i32, config: *const Config) !HTTPConn {
1846 const conn_arena = try allocator.create(std.heap.ArenaAllocator);
1847 errdefer allocator.destroy(conn_arena);
1848
1849 conn_arena.* = std.heap.ArenaAllocator.init(allocator);
1850 errdefer conn_arena.deinit();
1851
1852 const req_state = try Request.State.init(conn_arena.allocator(), buffer_pool, &config.request);
1853 const res_state = try Response.State.init(conn_arena.allocator(), &config.response);
1854
1855 var req_arena = std.heap.ArenaAllocator.init(allocator);
1856 errdefer req_arena.deinit();
1857
1858 return .{
1859 .io = io,
1860 .timeout = 0,
1861 ._mut = .init,
1862 ._state = .request,
1863 .handover = .unknown,
1864 .stream = undefined,
1865 .address = undefined,
1866 .loop = loop,
1867 .request_count = 0,
1868 .socket_flags = 0,
1869 .ws_worker = ws_worker,
1870 .req_state = req_state,
1871 .res_state = res_state,
1872 .req_arena = req_arena,
1873 .conn_arena = conn_arena,
1874 ._io_mode = if (comptime httpz.blockingMode()) .blocking else .nonblocking,
1875 };
1876 }
1877
1878 pub fn getState(self: *const HTTPConn) State {
1879 return @atomicLoad(State, &self._state, .acquire);
1880 }
1881
1882 pub fn setState(self: *HTTPConn, state: State) void {
1883 return @atomicStore(State, &self._state, state, .release);
1884 }
1885
1886 pub fn deinit(self: *HTTPConn, allocator: Allocator) void {
1887 self.req_state.deinit();
1888 self.req_arena.deinit();
1889 self.conn_arena.deinit();
1890 allocator.destroy(self.conn_arena);
1891 }
1892
1893 // This method is being called from a worker thread. Be careful what you
1894 // do here.
1895 // When a connection is disowned, we need to remove it from the
1896 // loop, since we aren't responsible for it anymore. However, we
1897 // can't wait until the normal signal flow to do that, because, for
1898 // all we know, the new owner of the socket will have closed the
1899 // connection by then. We need to remove the socket from the loop
1900 // synchronously when we're asked to disown it (as part of
1901 // res.disown()).
1902 pub fn disown(self: *HTTPConn) !void {
1903 if (comptime httpz.blockingMode()) {
1904 self.handover = .disown;
1905 return;
1906 }
1907
1908 const loop = self.loop;
1909 const socket = self.stream.socket.handle;
1910 switch (comptime loopType()) {
1911 .kqueue => {
1912 _ = try posix.kevent(loop, &.{
1913 .{
1914 .ident = @intCast(socket),
1915 .filter = posix.system.EVFILT.READ,
1916 .flags = posix.system.EV.DELETE,
1917 .fflags = 0,
1918 .data = 0,
1919 .udata = 0,
1920 },
1921 }, &.{}, null);
1922 },
1923 .epoll => {
1924 try posix.epoll_ctl(loop, std.os.linux.EPOLL.CTL_DEL, socket, null);
1925 },
1926 }
1927
1928 // Transfer ownership only after the event loop has definitely stopped
1929 // retaining the Conn pointer. If deregistration fails, leaving the
1930 // handover unchanged makes the normal reactor close path own cleanup.
1931 self.handover = .disown;
1932 }
1933
1934 pub fn requestDone(self: *HTTPConn, retain_allocated_bytes: usize, revert_blocking: bool) !void {
1935 self.req_state.reset();
1936 self.res_state.reset();
1937 _ = self.req_arena.reset(.{ .retain_with_limit = retain_allocated_bytes });
1938 if (revert_blocking) {
1939 try self.nonblockingMode();
1940 }
1941 }
1942
1943 pub fn writeAll(self: *HTTPConn, data: []const u8) !void {
1944 var i: usize = 0;
1945 var blocking = false;
1946
1947 const socket = self.stream.socket.handle;
1948
1949 while (i < data.len) {
1950 const n = posix.write(socket, data[i..]) catch |err| switch (err) {
1951 error.WouldBlock => {
1952 try self.blockingMode();
1953 blocking = true;
1954 continue;
1955 },
1956 else => return err,
1957 };
1958
1959 // shouldn't be posssible on a correct posix implementation
1960 // but let's assert to make sure
1961 std.debug.assert(n != 0);
1962 i += n;
1963 }
1964 }
1965
1966 pub fn writeAllIOVec(self: *HTTPConn, vec: []posix.iovec_const) !void {
1967 const socket = self.stream.socket.handle;
1968
1969 var i: usize = 0;
1970 while (true) {
1971 var n = posix.writev(socket, vec[i..]) catch |err| switch (err) {
1972 error.WouldBlock => {
1973 try self.blockingMode();
1974 continue;
1975 },
1976 else => return err,
1977 };
1978
1979 while (n >= vec[i].len) {
1980 n -= vec[i].len;
1981 i += 1;
1982 if (i >= vec.len) {
1983 return;
1984 }
1985 }
1986 vec[i].base += n;
1987 vec[i].len -= n;
1988 }
1989 }
1990
1991 pub fn blockingMode(self: *HTTPConn) !void {
1992 if (comptime httpz.blockingMode() == true) {
1993 // When httpz is in blocking mode, than we always keep the socket in
1994 // blocking mode
1995 return;
1996 }
1997 if (self._io_mode == .blocking) {
1998 return;
1999 }
2000 _ = try posix.fcntl(self.stream.socket.handle, posix.F.SETFL, self.socket_flags & ~@as(u32, @bitCast(posix.O{ .NONBLOCK = true })));
2001 self._io_mode = .blocking;
2002 }
2003
2004 pub fn nonblockingMode(self: *HTTPConn) !void {
2005 if (comptime httpz.blockingMode() == true) {
2006 // When httpz is in blocking mode, than we always keep the socket in
2007 // blocking mode
2008 return;
2009 }
2010 if (self._io_mode == .nonblocking) {
2011 return;
2012 }
2013 _ = try posix.fcntl(self.stream.socket.handle, posix.F.SETFL, self.socket_flags);
2014 self._io_mode = .nonblocking;
2015 }
2016};
2017
2018pub fn timestamp(io: Io) u32 {
2019 return @intCast(Io.Timestamp.now(io, .awake).toSeconds());
2020}
2021
2022fn initializeBufferPool(io: Io, allocator: Allocator, config: *const Config) !*BufferPool {
2023 const large_buffer_count = config.workers.large_buffer_count orelse blk: {
2024 if (comptime httpz.blockingMode()) {
2025 break :blk config.threadPoolCount();
2026 } else {
2027 break :blk 16;
2028 }
2029 };
2030 const large_buffer_size = config.workers.large_buffer_size orelse config.request.max_body_size orelse 65536;
2031
2032 const buffer_pool = try allocator.create(BufferPool);
2033 errdefer allocator.destroy(buffer_pool);
2034
2035 buffer_pool.* = try BufferPool.init(io, allocator, large_buffer_count, large_buffer_size);
2036 return buffer_pool;
2037}
2038
2039// Handles parsing errors. If this returns true, it means Conn has a body ready
2040// to write. In this case the worker (blocking or nonblocking) will want to send
2041// the response. If it returns false, the worker probably wants to close the connection.
2042// This function ensures that both Blocking and NonBlocking workers handle these
2043// errors with the same response
2044fn requestError(conn: *HTTPConn, err: anyerror) !void {
2045 const handle = conn.stream.socket.handle;
2046 switch (err) {
2047 error.HeaderTooBig => {
2048 metrics.invalidRequest();
2049 return writeError(handle, 431, "Request header is too big");
2050 },
2051 error.UnknownMethod, error.InvalidRequestTarget, error.UnknownProtocol, error.UnsupportedProtocol, error.InvalidHeaderLine, error.InvalidHost, error.InvalidContentLength, error.InvalidTransferEncoding, error.InvalidChunkSize, error.InvalidChunkLine => {
2052 metrics.invalidRequest();
2053 return writeError(handle, 400, "Invalid Request");
2054 },
2055 error.BodyTooBig => {
2056 metrics.bodyTooBig();
2057 return writeError(handle, 413, "Request body is too big");
2058 },
2059 error.BrokenPipe, error.ConnectionClosed, error.ConnectionResetByPeer, error.EndOfStream => return,
2060 else => {
2061 log.err("server error: {}", .{err});
2062 metrics.internalError();
2063 return writeError(handle, 500, "Internal Server Error");
2064 },
2065 }
2066 return err;
2067}
2068
2069fn writeError(socket: posix.socket_t, comptime status: u16, comptime msg: []const u8) !void {
2070 const response = std.fmt.comptimePrint("HTTP/1.1 {d} \r\nConnection: Close\r\nContent-Length: {d}\r\n\r\n{s}", .{ status, msg.len, msg });
2071
2072 var i: usize = 0;
2073 while (i < msg.len) {
2074 i += try posix.write(socket, response[i..]);
2075 }
2076}
2077
2078const LoopType = enum {
2079 kqueue,
2080 epoll,
2081};
2082
2083fn loopType() LoopType {
2084 return switch (builtin.os.tag) {
2085 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => .kqueue,
2086 .linux => .epoll,
2087 else => unreachable,
2088 };
2089}
2090
2091const t = @import("t.zig");
2092test "HTTPConnPool" {
2093 var bp = try BufferPool.init(t.io, t.allocator, 2, 64);
2094 defer bp.deinit();
2095
2096 var p = try HTTPConnPool.init(t.io, t.allocator, &bp, undefined, 0, &.{
2097 .workers = .{ .min_conn = 2 },
2098 .request = .{ .buffer_size = 64 },
2099 });
2100 defer p.deinit();
2101
2102 const s1 = try p.acquire();
2103 const s2 = try p.acquire();
2104 const s3 = try p.acquire();
2105
2106 try t.expectEqual(true, s1.req_state.buf.ptr != s2.req_state.buf.ptr);
2107 try t.expectEqual(true, s1.req_state.buf.ptr != s3.req_state.buf.ptr);
2108 try t.expectEqual(true, s2.req_state.buf.ptr != s3.req_state.buf.ptr);
2109
2110 p.release(s2);
2111 const s4 = try p.acquire();
2112 try t.expectEqual(true, s4.req_state.buf.ptr == s2.req_state.buf.ptr);
2113
2114 p.release(s1);
2115 p.release(s3);
2116 p.release(s4);
2117}
2118
2119test "HTTPConn.disown preserves reactor ownership when deregistration fails" {
2120 if (httpz.blockingMode()) return;
2121
2122 const loop = switch (comptime loopType()) {
2123 .kqueue => try posix.kqueue(),
2124 .epoll => try posix.epoll_create1(0),
2125 };
2126 defer posix.close(loop);
2127
2128 var bp = try BufferPool.init(t.io, t.allocator, 1, 64);
2129 defer bp.deinit();
2130
2131 var p = try HTTPConnPool.init(t.io, t.allocator, &bp, undefined, loop, &.{
2132 .workers = .{ .min_conn = 1 },
2133 .request = .{ .buffer_size = 64 },
2134 });
2135 defer p.deinit();
2136
2137 const conn = try p.acquire();
2138 defer p.release(conn);
2139 conn.handover = .unknown;
2140 // This fresh event loop has no registration for stdin, so deleting it
2141 // deterministically exercises the deregistration failure path.
2142 conn.stream.socket.handle = 0;
2143
2144 if (conn.disown()) {
2145 return error.ExpectedDeregistrationFailure;
2146 } else |_| {}
2147
2148 switch (conn.handover) {
2149 .unknown => {},
2150 else => return error.OwnershipTransferredAfterFailedDeregistration,
2151 }
2152}
2153
2154test "List: insert & remove" {
2155 var list = List(TestNode){};
2156 try expectList(&.{}, list);
2157
2158 var n1 = TestNode{ .id = 1 };
2159 list.insert(&n1);
2160 try expectList(&.{1}, list);
2161
2162 list.remove(&n1);
2163 try expectList(&.{}, list);
2164
2165 var n2 = TestNode{ .id = 2 };
2166 list.insert(&n2);
2167 list.insert(&n1);
2168 try expectList(&.{ 2, 1 }, list);
2169
2170 var n3 = TestNode{ .id = 3 };
2171 list.insert(&n3);
2172 try expectList(&.{ 2, 1, 3 }, list);
2173
2174 list.remove(&n1);
2175 try expectList(&.{ 2, 3 }, list);
2176
2177 list.insert(&n1);
2178 try expectList(&.{ 2, 3, 1 }, list);
2179
2180 list.remove(&n2);
2181 try expectList(&.{ 3, 1 }, list);
2182
2183 list.remove(&n1);
2184 try expectList(&.{3}, list);
2185
2186 list.remove(&n3);
2187 try expectList(&.{}, list);
2188}
2189
2190test "List: detach prefix clears intrusive links" {
2191 var list = List(TestNode){};
2192 var n1 = TestNode{ .id = 1 };
2193 var n2 = TestNode{ .id = 2 };
2194 var n3 = TestNode{ .id = 3 };
2195 var n4 = TestNode{ .id = 4 };
2196 list.insert(&n1);
2197 list.insert(&n2);
2198 list.insert(&n3);
2199 list.insert(&n4);
2200
2201 const empty = detachPrefix(TestNode, &list, 0);
2202 try expectList(&.{}, empty);
2203 try expectList(&.{ 1, 2, 3, 4 }, list);
2204
2205 const prefix = detachPrefix(TestNode, &list, 2);
2206 try expectList(&.{ 1, 2 }, prefix);
2207 try expectList(&.{ 3, 4 }, list);
2208 try t.expectEqual(null, prefix.tail.?.next);
2209 try t.expectEqual(null, list.head.?.prev);
2210
2211 const rest = detachPrefix(TestNode, &list, 2);
2212 try expectList(&.{ 3, 4 }, rest);
2213 try expectList(&.{}, list);
2214}
2215
2216const TestNode = struct {
2217 id: i32,
2218 next: ?*TestNode = null,
2219 prev: ?*TestNode = null,
2220
2221 fn alloc(id: i32) *TestNode {
2222 const tn = t.allocator.create(TestNode) catch unreachable;
2223 tn.* = .{ .id = id };
2224 return tn;
2225 }
2226};
2227
2228fn expectList(expected: []const i32, list: List(TestNode)) !void {
2229 if (expected.len == 0) {
2230 try t.expectEqual(null, list.head);
2231 try t.expectEqual(null, list.tail);
2232 return;
2233 }
2234
2235 var i: usize = 0;
2236 var next = list.head;
2237 while (next) |node| {
2238 try t.expectEqual(expected[i], node.id);
2239 i += 1;
2240 next = node.next;
2241 }
2242 try t.expectEqual(expected.len, i);
2243
2244 i = expected.len;
2245 var prev = list.tail;
2246 while (prev) |node| {
2247 i -= 1;
2248 try t.expectEqual(expected[i], node.id);
2249 prev = node.prev;
2250 }
2251 try t.expectEqual(0, i);
2252}