lib/http/src/server/runtime.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sys_root = @import("sys");
3 const event = sys_root.event;
4 const sys = sys_root.net;
5 const thread = sys_root.thread;
6 const time = sys_root.time;
7 const http = @import("../root.zig");
8 const Connection = http.Connection;
9
10 const ConnectionTask = struct {
11 slot: *ServerConnectionSlot,
12
13 pub fn run(task: ConnectionTask) void {
14 Server.handleConnection(task.slot);
15 }
16
17 pub fn complete(task: ConnectionTask) void {
18 Server.finishConnection(task.slot);
19 }
20 };
21
22 const ThreadPool = http.ThreadPool(ConnectionTask);
23
24 const log = std.log.scoped(.http_server);
25
26 const SlotPhase = enum {
27 free,
28 waiting,
29 running,
30 completed,
31 closing,
32 };
33
34 const ServerConnectionSlot = struct {
35 server: *Server,
36 index: usize,
37 next_free: ?usize,
38 next_completed: ?usize,
39 occupied: bool,
40 phase: SlotPhase,
41 connection: Connection,
42 read_completion: event.Completion,
43 };
44
45 pub const Server = struct {
46 allocator: std.mem.Allocator,
47 listener: ?sys.Socket,
48 pool: ThreadPool,
49 event_loop: event.Loop,
50 wake: event.Async,
51 wake_completion: event.Completion,
52 accept_completion: event.Completion,
53 listener_close_completion: event.Completion,
54 idle_timer: event.Timer,
55 idle_completion: event.Completion,
56 connection_slots: []ServerConnectionSlot,
57 connection_input_storage: Connection.InputStorage,
58 request_storage: http.RequestStorage,
59 response_storage: http.ResponseStorage,
60 free_connection_slot: ?usize,
61 completed_connection_head: ?usize,
62 completed_connection_tail: ?usize,
63 active_connections: usize,
64 connections_mutex: std.atomic.Mutex,
65 input_capacity_rejections: u64,
66 shutdown: std.atomic.Value(bool),
67 listen_active: std.atomic.Value(bool),
68 config: Config,
69 next_conn_id: std.atomic.Value(usize),
70 handler: ?*const fn (?*anyopaque, *Connection) void,
71 handler_context: ?*anyopaque,
72
73 pub const Config = struct {
74 address: []const u8 = "127.0.0.1",
75 port: u16 = 8080,
76 num_workers: usize = 4,
77 max_connections: usize = 1024,
78 read_timeout_ms: u32 = 30_000,
79 write_timeout_ms: u32 = 30_000,
80 boot_clock: time.BootClock = .system(),
81 tcp_no_delay: bool = false,
82 connection_input_bytes_per_connection: usize =
83 http.default_connection_input_bytes_per_connection,
84 request_header_count_per_connection: usize = http.default_request_header_count,
85 request_header_line_bytes: usize = http.default_request_header_line_bytes,
86 request_body_bytes_per_connection: usize = http.default_request_body_bytes,
87 response_header_count_per_connection: usize = http.default_response_header_count,
88 response_head_bytes_per_connection: usize = http.default_response_head_bytes,
89 };
90
91 pub fn init(allocator: std.mem.Allocator, config: Config) !*Server {
92 if (config.max_connections > std.math.maxInt(u32) - 4) return error.ConnectionLimitTooLarge;
93 const self = try allocator.create(Server);
94 errdefer allocator.destroy(self);
95
96 self.pool = try ThreadPool.init(allocator, .{
97 .workers = config.num_workers,
98 .tasks = config.max_connections,
99 });
100 errdefer self.pool.deinit(allocator);
101
102 self.event_loop = try event.Loop.init(.{
103 .allocator = allocator,
104 .entries = @intCast(config.max_connections + 4),
105 });
106 errdefer self.event_loop.deinit();
107
108 self.wake = try event.Async.init();
109 errdefer self.wake.deinit();
110
111 self.idle_timer = try event.Timer.init();
112 errdefer self.idle_timer.deinit();
113
114 self.connection_slots = try allocator.alloc(ServerConnectionSlot, config.max_connections);
115 errdefer allocator.free(self.connection_slots);
116
117 self.connection_input_storage = try Connection.InputStorage.init(allocator, .{
118 .connection_count = config.max_connections,
119 .bytes_per_connection = config.connection_input_bytes_per_connection,
120 });
121 errdefer self.connection_input_storage.deinit(allocator);
122
123 self.request_storage = try http.RequestStorage.init(allocator, .{
124 .request_count = config.max_connections,
125 .header_count_per_request = config.request_header_count_per_connection,
126 .header_line_bytes = config.request_header_line_bytes,
127 .body_bytes_per_request = config.request_body_bytes_per_connection,
128 });
129 errdefer self.request_storage.deinit(allocator);
130
131 self.response_storage = try http.ResponseStorage.init(allocator, .{
132 .response_count = config.max_connections,
133 .header_count_per_response = config.response_header_count_per_connection,
134 .head_bytes_per_response = config.response_head_bytes_per_connection,
135 });
136 errdefer self.response_storage.deinit(allocator);
137
138 const listener = try createListenerSocket(config.address, config.port);
139 errdefer sys.close(listener);
140
141 self.allocator = allocator;
142 self.listener = listener;
143 self.wake_completion = .{};
144 self.accept_completion = .{};
145 self.listener_close_completion = .{};
146 self.idle_completion = .{};
147 self.free_connection_slot = if (self.connection_slots.len == 0) null else 0;
148 self.completed_connection_head = null;
149 self.completed_connection_tail = null;
150 self.active_connections = 0;
151 self.connections_mutex = .unlocked;
152 self.input_capacity_rejections = 0;
153 self.shutdown = std.atomic.Value(bool).init(false);
154 self.listen_active = std.atomic.Value(bool).init(false);
155 self.config = config;
156 self.next_conn_id = std.atomic.Value(usize).init(1);
157 self.handler = null;
158 self.handler_context = null;
159
160 self.initializeSlots();
161
162 try self.pool.activate();
163 self.connection_input_storage.activate();
164 self.request_storage.activate();
165 self.response_storage.activate();
166
167 return self;
168 }
169
170 fn initializeSlots(self: *Server) void {
171 for (self.connection_slots, 0..) |*slot, index| {
172 slot.* = .{
173 .server = self,
174 .index = index,
175 .next_free = if (index + 1 < self.connection_slots.len) index + 1 else null,
176 .next_completed = null,
177 .occupied = false,
178 .phase = .free,
179 .connection = undefined,
180 .read_completion = .{},
181 };
182 }
183 }
184
185 fn createListenerSocket(address: []const u8, port: u16) !sys.Socket {
186 const addr = try sys.ip4AddressForHost(address, port);
187 const sock = try sys.tcpStreamSocket(.{});
188 errdefer sys.close(sock);
189
190 try sys.setReuseAddress(sock);
191 try sys.bindIp4(sock, addr);
192 try sys.listen(sock, 128);
193 try sys.setNonBlocking(sock);
194 return sock;
195 }
196
197 pub fn deinit(self: *Server) void {
198 self.stop();
199 var wait_rounds: usize = 0;
200 while (self.listen_active.load(.acquire) and wait_rounds < 5000) : (wait_rounds += 1) {
201 sleepMillis(1);
202 }
203 std.debug.assert(!self.listen_active.load(.acquire));
204
205 if (self.listener) |sock| {
206 sys.close(sock);
207 self.listener = null;
208 }
209
210 lock(&self.connections_mutex);
211 for (self.connection_slots) |*slot| {
212 if (slot.occupied) slot.connection.interrupt();
213 }
214 self.connections_mutex.unlock();
215
216 _ = self.pool.drain(
217 .system(),
218 .fromNanoseconds(5 * std.time.ns_per_s),
219 ) catch false;
220 self.pool.deinit(self.allocator);
221
222 lock(&self.connections_mutex);
223 for (self.connection_slots) |*slot| {
224 if (slot.occupied) slot.connection.deinit();
225 }
226 self.connections_mutex.unlock();
227
228 self.idle_timer.deinit();
229 self.wake.deinit();
230 self.event_loop.deinit();
231 self.response_storage.deinit(self.allocator);
232 self.request_storage.deinit(self.allocator);
233 self.connection_input_storage.deinit(self.allocator);
234 self.allocator.free(self.connection_slots);
235 self.allocator.destroy(self);
236 }
237
238 pub fn setHandler(self: *Server, context: anytype, comptime handler: fn (@TypeOf(context), *Connection) void) void {
239 const Context = @TypeOf(context);
240 comptime std.debug.assert(@typeInfo(Context) == .pointer);
241 const Erased = struct {
242 fn call(erased: ?*anyopaque, conn: *Connection) void {
243 handler(@ptrCast(@alignCast(erased)), conn);
244 }
245 };
246 self.handler_context = @ptrCast(@constCast(context));
247 self.handler = Erased.call;
248 }
249
250 pub fn listen(self: *Server) !void {
251 const listener = self.listener orelse return error.NotListening;
252 if (self.listen_active.swap(true, .acq_rel)) return error.AlreadyListening;
253 defer self.listen_active.store(false, .release);
254
255 self.wake.wait(
256 &self.event_loop,
257 &self.wake_completion,
258 Server,
259 self,
260 wakeReady,
261 );
262 event.TCP.initFd(listener).accept(
263 &self.event_loop,
264 &self.accept_completion,
265 Server,
266 self,
267 acceptReady,
268 );
269 if (self.config.read_timeout_ms != 0) {
270 const interval = time.Duration.fromMilliseconds(
271 idleSweepInterval(self.config.read_timeout_ms),
272 );
273 try self.idle_timer.run(
274 &self.event_loop,
275 &self.idle_completion,
276 .{ .after = interval, .repeat = .{ .fixed_delay = interval } },
277 Server,
278 self,
279 idleTimerReady,
280 );
281 }
282
283 try self.event_loop.run(.until_done);
284 }
285
286 fn acquireConnection(
287 self: *Server,
288 socket: sys.Socket,
289 boot_clock: time.BootClock,
290 accepted_at: time.BootInstant,
291 ) ?*ServerConnectionSlot {
292 lock(&self.connections_mutex);
293 defer self.connections_mutex.unlock();
294
295 const index = self.free_connection_slot orelse return null;
296 const slot = &self.connection_slots[index];
297 self.free_connection_slot = slot.next_free;
298 slot.next_free = null;
299 slot.next_completed = null;
300 slot.occupied = true;
301 slot.phase = .waiting;
302 slot.connection = Connection.init(
303 self.next_conn_id.fetchAdd(1, .monotonic),
304 socket,
305 self.connection_input_storage.connection(index) catch unreachable,
306 self.request_storage.request(index) catch unreachable,
307 self.response_storage.response(index) catch unreachable,
308 boot_clock,
309 accepted_at,
310 );
311 self.active_connections += 1;
312 return slot;
313 }
314
315 fn releaseConnection(self: *Server, slot: *ServerConnectionSlot) void {
316 lock(&self.connections_mutex);
317 defer self.connections_mutex.unlock();
318
319 std.debug.assert(slot.occupied);
320 self.input_capacity_rejections +|=
321 slot.connection.inputStatus().capacity_rejections;
322 slot.occupied = false;
323 slot.phase = .free;
324 slot.next_free = self.free_connection_slot;
325 slot.next_completed = null;
326 self.free_connection_slot = slot.index;
327 self.active_connections -= 1;
328 }
329
330 fn handleConnection(slot: *ServerConnectionSlot) void {
331 if (slot.server.handler) |handler| {
332 handler(slot.server.handler_context, &slot.connection);
333 } else {
334 slot.connection.markClosing();
335 }
336 }
337
338 fn finishConnection(slot: *ServerConnectionSlot) void {
339 const server = slot.server;
340 lock(&server.connections_mutex);
341 std.debug.assert(slot.occupied);
342 std.debug.assert(slot.phase == .running);
343 slot.phase = .completed;
344 slot.next_completed = null;
345 if (server.completed_connection_tail) |tail| {
346 server.connection_slots[tail].next_completed = slot.index;
347 } else {
348 server.completed_connection_head = slot.index;
349 }
350 server.completed_connection_tail = slot.index;
351 server.connections_mutex.unlock();
352 server.wake.notify() catch {};
353 }
354
355 pub fn stop(self: *Server) void {
356 self.shutdown.store(true, .release);
357 self.wake.notify() catch {};
358 }
359
360 pub fn connectionCount(self: *Server) usize {
361 lock(&self.connections_mutex);
362 defer self.connections_mutex.unlock();
363 return self.active_connections;
364 }
365
366 pub fn connectionInputCapacity(self: *const Server) Connection.InputCapacity {
367 return self.connection_input_storage.capacity;
368 }
369
370 pub fn requestCapacity(self: *const Server) http.RequestCapacity {
371 return self.request_storage.capacity;
372 }
373
374 pub fn responseCapacity(self: *const Server) http.ResponseCapacity {
375 return self.response_storage.capacity;
376 }
377
378 pub fn connectionInputStatus(self: *Server) http.ConnectionInputStatus {
379 lock(&self.connections_mutex);
380 defer self.connections_mutex.unlock();
381 var status = http.ConnectionInputStatus{
382 .capacity_rejections = self.input_capacity_rejections,
383 };
384 for (self.connection_slots) |*slot| {
385 if (!slot.occupied) continue;
386 status.capacity_rejections +|= slot.connection.inputStatus().capacity_rejections;
387 }
388 return status;
389 }
390
391 fn acceptReady(
392 maybe_server: ?*Server,
393 loop: *event.Loop,
394 _: *event.Completion,
395 result: event.AcceptError!event.TCP,
396 ) event.CallbackAction {
397 const self = maybe_server.?;
398 const client = result catch return if (self.shutdown.load(.acquire)) .disarm else .rearm;
399 if (self.shutdown.load(.acquire)) {
400 sys.close(client.fd);
401 return .disarm;
402 }
403 if (self.config.tcp_no_delay) {
404 sys.setTcpNoDelay(client.fd) catch |err| {
405 log.warn(
406 "failed to disable TCP delay: {s}",
407 .{@errorName(err)},
408 );
409 sys.close(client.fd);
410 return .rearm;
411 };
412 }
413
414 const accepted_at = self.config.boot_clock.now() catch {
415 sys.close(client.fd);
416 return .rearm;
417 };
418 const slot = self.acquireConnection(client.fd, self.config.boot_clock, accepted_at) orelse {
419 sys.close(client.fd);
420 return .rearm;
421 };
422 slot.connection.setWriteTimeout(self.config.write_timeout_ms) catch |err| {
423 log.warn("conn {d}: failed to set write timeout: {s}", .{ slot.connection.id, @errorName(err) });
424 };
425 self.armWaiting(loop, slot);
426 return .rearm;
427 }
428
429 fn readableReady(
430 maybe_slot: ?*ServerConnectionSlot,
431 _: *event.Loop,
432 _: *event.Completion,
433 _: event.File,
434 result: event.PollError!event.PollEvent,
435 ) event.CallbackAction {
436 const slot = maybe_slot.?;
437 _ = result catch {
438 slot.connection.deinit();
439 slot.server.releaseConnection(slot);
440 return .disarm;
441 };
442 slot.server.dispatch(slot) catch {
443 slot.connection.deinit();
444 slot.server.releaseConnection(slot);
445 };
446 return .disarm;
447 }
448
449 fn wakeReady(
450 maybe_server: ?*Server,
451 loop: *event.Loop,
452 _: *event.Completion,
453 result: event.Async.WaitError!void,
454 ) event.CallbackAction {
455 const self = maybe_server.?;
456 _ = result catch return .disarm;
457 self.drainCompleted(loop);
458 if (!self.shutdown.load(.acquire)) return .rearm;
459 self.closeListener(loop);
460 self.closeConnections(loop);
461 return .disarm;
462 }
463
464 fn idleTimerReady(
465 maybe_server: ?*Server,
466 loop: *event.Loop,
467 _: *event.Completion,
468 result: event.Timer.RunError!void,
469 ) event.CallbackAction {
470 const self = maybe_server.?;
471 _ = result catch return .disarm;
472 if (self.shutdown.load(.acquire)) return .disarm;
473
474 const now = self.config.boot_clock.now() catch return .disarm;
475 for (self.connection_slots) |*slot| {
476 lock(&self.connections_mutex);
477 const expired = slot.occupied and slot.phase == .waiting and
478 idleExpired(now, slot.connection.lastActivity(), self.config.read_timeout_ms);
479 if (expired) slot.phase = .closing;
480 self.connections_mutex.unlock();
481 if (expired) self.closeWaiting(loop, slot);
482 }
483 return .rearm;
484 }
485
486 fn listenerClosed(
487 _: ?*Server,
488 _: *event.Loop,
489 _: *event.Completion,
490 _: event.TCP,
491 result: event.CloseError!void,
492 ) event.CallbackAction {
493 _ = result catch {};
494 return .disarm;
495 }
496
497 fn connectionClosed(
498 maybe_slot: ?*ServerConnectionSlot,
499 _: *event.Loop,
500 _: *event.Completion,
501 _: event.TCP,
502 result: event.CloseError!void,
503 ) event.CallbackAction {
504 const slot = maybe_slot.?;
505 _ = result catch {};
506 slot.connection.transportClosed();
507 slot.connection.deinit();
508 slot.server.releaseConnection(slot);
509 return .disarm;
510 }
511
512 fn dispatch(self: *Server, slot: *ServerConnectionSlot) !void {
513 lock(&self.connections_mutex);
514 std.debug.assert(slot.occupied);
515 std.debug.assert(canDispatch(slot.phase));
516 slot.phase = .running;
517 self.connections_mutex.unlock();
518 slot.connection.beginTurn();
519 self.pool.submit(.{ .slot = slot }) catch |err| {
520 lock(&self.connections_mutex);
521 slot.phase = .waiting;
522 self.connections_mutex.unlock();
523 return err;
524 };
525 }
526
527 fn armWaiting(self: *Server, loop: *event.Loop, slot: *ServerConnectionSlot) void {
528 lock(&self.connections_mutex);
529 std.debug.assert(slot.occupied);
530 std.debug.assert(canDispatch(slot.phase));
531 slot.phase = .waiting;
532 self.connections_mutex.unlock();
533 event.File.initFd(slot.connection.socket).poll(
534 loop,
535 &slot.read_completion,
536 .read,
537 ServerConnectionSlot,
538 slot,
539 readableReady,
540 );
541 }
542
543 fn drainCompleted(self: *Server, loop: *event.Loop) void {
544 while (self.popCompleted()) |slot| {
545 const state = slot.connection.currentState();
546 if (self.shutdown.load(.acquire) or state == .closing or state == .closed or !slot.connection.shouldWaitForRead()) {
547 lock(&self.connections_mutex);
548 slot.phase = .closing;
549 self.connections_mutex.unlock();
550 slot.connection.deinit();
551 self.releaseConnection(slot);
552 } else if (slot.connection.shouldConsumeBufferedInput() and slot.connection.hasBufferedInput()) {
553 self.dispatch(slot) catch {
554 slot.connection.deinit();
555 self.releaseConnection(slot);
556 };
557 } else {
558 self.armWaiting(loop, slot);
559 }
560 }
561 }
562
563 fn popCompleted(self: *Server) ?*ServerConnectionSlot {
564 lock(&self.connections_mutex);
565 defer self.connections_mutex.unlock();
566 const index = self.completed_connection_head orelse return null;
567 const slot = &self.connection_slots[index];
568 self.completed_connection_head = slot.next_completed;
569 if (self.completed_connection_head == null) self.completed_connection_tail = null;
570 slot.next_completed = null;
571 return slot;
572 }
573
574 fn closeListener(self: *Server, loop: *event.Loop) void {
575 const socket = self.listener orelse return;
576 self.listener = null;
577 event.TCP.initFd(socket).close(
578 loop,
579 &self.listener_close_completion,
580 Server,
581 self,
582 listenerClosed,
583 );
584 }
585
586 fn closeConnections(self: *Server, loop: *event.Loop) void {
587 for (self.connection_slots) |*slot| {
588 lock(&self.connections_mutex);
589 if (!slot.occupied) {
590 self.connections_mutex.unlock();
591 continue;
592 }
593 const phase = slot.phase;
594 if (phase == .waiting) slot.phase = .closing;
595 self.connections_mutex.unlock();
596 switch (phase) {
597 .waiting => self.closeWaiting(loop, slot),
598 .running => slot.connection.interrupt(),
599 .completed => {},
600 .free, .closing => {},
601 }
602 }
603 self.drainCompleted(loop);
604 }
605
606 fn closeWaiting(self: *Server, loop: *event.Loop, slot: *ServerConnectionSlot) void {
607 _ = self;
608 event.TCP.initFd(slot.connection.socket).close(
609 loop,
610 &slot.read_completion,
611 ServerConnectionSlot,
612 slot,
613 connectionClosed,
614 );
615 }
616 };
617
618 fn idleSweepInterval(read_timeout_ms: u32) u64 {
619 return @max(@min(read_timeout_ms, 100), 1);
620 }
621
622 fn canDispatch(phase: SlotPhase) bool {
623 return switch (phase) {
624 .waiting, .completed => true,
625 .free, .running, .closing => false,
626 };
627 }
628
629 fn lock(mutex: *std.atomic.Mutex) void {
630 while (!mutex.tryLock()) thread.yield();
631 }
632
633 fn idleExpired(
634 now: time.BootInstant,
635 last_activity: time.BootInstant,
636 timeout_ms: u32,
637 ) bool {
638 const timeout = time.Duration.fromMilliseconds(timeout_ms);
639 return now.reached(last_activity.deadlineAfter(timeout));
640 }
641
642 test "suspend gaps expire an HTTP idle deadline without wall participation" {
643 var clock = time.FakeClock.zero();
644 const last_activity = clock.boot;
645 clock.setWall(.fromNanoseconds(std.math.maxInt(u64)));
646 try std.testing.expect(!idleExpired(clock.boot, last_activity, 10));
647 clock.suspendGap(.fromMilliseconds(10));
648 try std.testing.expect(idleExpired(clock.boot, last_activity, 10));
649 }
650
651 fn sleepMillis(ms: u64) void {
652 time.sleepMilliseconds(ms);
653 }