tiny.http.Server

1pub const Server = struct {2    allocator: std.mem.Allocator,3    listener: ?sys.Socket,4    pool: ThreadPool,5    event_loop: event.Loop,6    wake: event.Async,7    wake_completion: event.Completion,8    accept_completion: event.Completion,9    listener_close_completion: event.Completion,10    idle_timer: event.Timer,11    idle_completion: event.Completion,12    connection_slots: []ServerConnectionSlot,13    connection_input_storage: Connection.InputStorage,14    request_storage: http.RequestStorage,15    response_storage: http.ResponseStorage,16    free_connection_slot: ?usize,17    completed_connection_head: ?usize,18    completed_connection_tail: ?usize,19    active_connections: usize,20    connections_mutex: std.atomic.Mutex,21    input_capacity_rejections: u64,22    shutdown: std.atomic.Value(bool),23    listen_active: std.atomic.Value(bool),24    config: Config,25    next_conn_id: std.atomic.Value(usize),26    handler: ?*const fn (?*anyopaque, *Connection) void,27    handler_context: ?*anyopaque,2829    pub const Config = struct {30        address: []const u8 = "127.0.0.1",31        port: u16 = 8080,32        num_workers: usize = 4,33        max_connections: usize = 1024,34        read_timeout_ms: u32 = 30_000,35        write_timeout_ms: u32 = 30_000,36        boot_clock: time.BootClock = .system(),37        tcp_no_delay: bool = false,38        connection_input_bytes_per_connection: usize =39            http.default_connection_input_bytes_per_connection,40        request_header_count_per_connection: usize = http.default_request_header_count,41        request_header_line_bytes: usize = http.default_request_header_line_bytes,42        request_body_bytes_per_connection: usize = http.default_request_body_bytes,43        response_header_count_per_connection: usize = http.default_response_header_count,44        response_head_bytes_per_connection: usize = http.default_response_head_bytes,45    };4647    pub fn init(allocator: std.mem.Allocator, config: Config) !*Server {48        if (config.max_connections > std.math.maxInt(u32) - 4) return error.ConnectionLimitTooLarge;49        const self = try allocator.create(Server);50        errdefer allocator.destroy(self);5152        self.pool = try ThreadPool.init(allocator, .{53            .workers = config.num_workers,54            .tasks = config.max_connections,55        });56        errdefer self.pool.deinit(allocator);5758        self.event_loop = try event.Loop.init(.{59            .allocator = allocator,60            .entries = @intCast(config.max_connections + 4),61        });62        errdefer self.event_loop.deinit();6364        self.wake = try event.Async.init();65        errdefer self.wake.deinit();6667        self.idle_timer = try event.Timer.init();68        errdefer self.idle_timer.deinit();6970        self.connection_slots = try allocator.alloc(ServerConnectionSlot, config.max_connections);71        errdefer allocator.free(self.connection_slots);7273        self.connection_input_storage = try Connection.InputStorage.init(allocator, .{74            .connection_count = config.max_connections,75            .bytes_per_connection = config.connection_input_bytes_per_connection,76        });77        errdefer self.connection_input_storage.deinit(allocator);7879        self.request_storage = try http.RequestStorage.init(allocator, .{80            .request_count = config.max_connections,81            .header_count_per_request = config.request_header_count_per_connection,82            .header_line_bytes = config.request_header_line_bytes,83            .body_bytes_per_request = config.request_body_bytes_per_connection,84        });85        errdefer self.request_storage.deinit(allocator);8687        self.response_storage = try http.ResponseStorage.init(allocator, .{88            .response_count = config.max_connections,89            .header_count_per_response = config.response_header_count_per_connection,90            .head_bytes_per_response = config.response_head_bytes_per_connection,91        });92        errdefer self.response_storage.deinit(allocator);9394        const listener = try createListenerSocket(config.address, config.port);95        errdefer sys.close(listener);9697        self.allocator = allocator;98        self.listener = listener;99        self.wake_completion = .{};100        self.accept_completion = .{};101        self.listener_close_completion = .{};102        self.idle_completion = .{};103        self.free_connection_slot = if (self.connection_slots.len == 0) null else 0;104        self.completed_connection_head = null;105        self.completed_connection_tail = null;106        self.active_connections = 0;107        self.connections_mutex = .unlocked;108        self.input_capacity_rejections = 0;109        self.shutdown = std.atomic.Value(bool).init(false);110        self.listen_active = std.atomic.Value(bool).init(false);111        self.config = config;112        self.next_conn_id = std.atomic.Value(usize).init(1);113        self.handler = null;114        self.handler_context = null;115116        self.initializeSlots();117118        try self.pool.activate();119        self.connection_input_storage.activate();120        self.request_storage.activate();121        self.response_storage.activate();122123        return self;124    }125126    fn initializeSlots(self: *Server) void {127        for (self.connection_slots, 0..) |*slot, index| {128            slot.* = .{129                .server = self,130                .index = index,131                .next_free = if (index + 1 < self.connection_slots.len) index + 1 else null,132                .next_completed = null,133                .occupied = false,134                .phase = .free,135                .connection = undefined,136                .read_completion = .{},137            };138        }139    }140141    fn createListenerSocket(address: []const u8, port: u16) !sys.Socket {142        const addr = try sys.ip4AddressForHost(address, port);143        const sock = try sys.tcpStreamSocket(.{});144        errdefer sys.close(sock);145146        try sys.setReuseAddress(sock);147        try sys.bindIp4(sock, addr);148        try sys.listen(sock, 128);149        try sys.setNonBlocking(sock);150        return sock;151    }152153    pub fn deinit(self: *Server) void {154        self.stop();155        var wait_rounds: usize = 0;156        while (self.listen_active.load(.acquire) and wait_rounds < 5000) : (wait_rounds += 1) {157            sleepMillis(1);158        }159        std.debug.assert(!self.listen_active.load(.acquire));160161        if (self.listener) |sock| {162            sys.close(sock);163            self.listener = null;164        }165166        lock(&self.connections_mutex);167        for (self.connection_slots) |*slot| {168            if (slot.occupied) slot.connection.interrupt();169        }170        self.connections_mutex.unlock();171172        _ = self.pool.drain(173            .system(),174            .fromNanoseconds(5 * std.time.ns_per_s),175        ) catch false;176        self.pool.deinit(self.allocator);177178        lock(&self.connections_mutex);179        for (self.connection_slots) |*slot| {180            if (slot.occupied) slot.connection.deinit();181        }182        self.connections_mutex.unlock();183184        self.idle_timer.deinit();185        self.wake.deinit();186        self.event_loop.deinit();187        self.response_storage.deinit(self.allocator);188        self.request_storage.deinit(self.allocator);189        self.connection_input_storage.deinit(self.allocator);190        self.allocator.free(self.connection_slots);191        self.allocator.destroy(self);192    }193194    pub fn setHandler(self: *Server, context: anytype, comptime handler: fn (@TypeOf(context), *Connection) void) void {195        const Context = @TypeOf(context);196        comptime std.debug.assert(@typeInfo(Context) == .pointer);197        const Erased = struct {198            fn call(erased: ?*anyopaque, conn: *Connection) void {199                handler(@ptrCast(@alignCast(erased)), conn);200            }201        };202        self.handler_context = @ptrCast(@constCast(context));203        self.handler = Erased.call;204    }205206    pub fn listen(self: *Server) !void {207        const listener = self.listener orelse return error.NotListening;208        if (self.listen_active.swap(true, .acq_rel)) return error.AlreadyListening;209        defer self.listen_active.store(false, .release);210211        self.wake.wait(212            &self.event_loop,213            &self.wake_completion,214            Server,215            self,216            wakeReady,217        );218        event.TCP.initFd(listener).accept(219            &self.event_loop,220            &self.accept_completion,221            Server,222            self,223            acceptReady,224        );225        if (self.config.read_timeout_ms != 0) {226            const interval = time.Duration.fromMilliseconds(227                idleSweepInterval(self.config.read_timeout_ms),228            );229            try self.idle_timer.run(230                &self.event_loop,231                &self.idle_completion,232                .{ .after = interval, .repeat = .{ .fixed_delay = interval } },233                Server,234                self,235                idleTimerReady,236            );237        }238239        try self.event_loop.run(.until_done);240    }241242    fn acquireConnection(243        self: *Server,244        socket: sys.Socket,245        boot_clock: time.BootClock,246        accepted_at: time.BootInstant,247    ) ?*ServerConnectionSlot {248        lock(&self.connections_mutex);249        defer self.connections_mutex.unlock();250251        const index = self.free_connection_slot orelse return null;252        const slot = &self.connection_slots[index];253        self.free_connection_slot = slot.next_free;254        slot.next_free = null;255        slot.next_completed = null;256        slot.occupied = true;257        slot.phase = .waiting;258        slot.connection = Connection.init(259            self.next_conn_id.fetchAdd(1, .monotonic),260            socket,261            self.connection_input_storage.connection(index) catch unreachable,262            self.request_storage.request(index) catch unreachable,263            self.response_storage.response(index) catch unreachable,264            boot_clock,265            accepted_at,266        );267        self.active_connections += 1;268        return slot;269    }270271    fn releaseConnection(self: *Server, slot: *ServerConnectionSlot) void {272        lock(&self.connections_mutex);273        defer self.connections_mutex.unlock();274275        std.debug.assert(slot.occupied);276        self.input_capacity_rejections +|=277            slot.connection.inputStatus().capacity_rejections;278        slot.occupied = false;279        slot.phase = .free;280        slot.next_free = self.free_connection_slot;281        slot.next_completed = null;282        self.free_connection_slot = slot.index;283        self.active_connections -= 1;284    }285286    fn handleConnection(slot: *ServerConnectionSlot) void {287        if (slot.server.handler) |handler| {288            handler(slot.server.handler_context, &slot.connection);289        } else {290            slot.connection.markClosing();291        }292    }293294    fn finishConnection(slot: *ServerConnectionSlot) void {295        const server = slot.server;296        lock(&server.connections_mutex);297        std.debug.assert(slot.occupied);298        std.debug.assert(slot.phase == .running);299        slot.phase = .completed;300        slot.next_completed = null;301        if (server.completed_connection_tail) |tail| {302            server.connection_slots[tail].next_completed = slot.index;303        } else {304            server.completed_connection_head = slot.index;305        }306        server.completed_connection_tail = slot.index;307        server.connections_mutex.unlock();308        server.wake.notify() catch {};309    }310311    pub fn stop(self: *Server) void {312        self.shutdown.store(true, .release);313        self.wake.notify() catch {};314    }315316    pub fn connectionCount(self: *Server) usize {317        lock(&self.connections_mutex);318        defer self.connections_mutex.unlock();319        return self.active_connections;320    }321322    pub fn connectionInputCapacity(self: *const Server) Connection.InputCapacity {323        return self.connection_input_storage.capacity;324    }325326    pub fn requestCapacity(self: *const Server) http.RequestCapacity {327        return self.request_storage.capacity;328    }329330    pub fn responseCapacity(self: *const Server) http.ResponseCapacity {331        return self.response_storage.capacity;332    }333334    pub fn connectionInputStatus(self: *Server) http.ConnectionInputStatus {335        lock(&self.connections_mutex);336        defer self.connections_mutex.unlock();337        var status = http.ConnectionInputStatus{338            .capacity_rejections = self.input_capacity_rejections,339        };340        for (self.connection_slots) |*slot| {341            if (!slot.occupied) continue;342            status.capacity_rejections +|= slot.connection.inputStatus().capacity_rejections;343        }344        return status;345    }346347    fn acceptReady(348        maybe_server: ?*Server,349        loop: *event.Loop,350        _: *event.Completion,351        result: event.AcceptError!event.TCP,352    ) event.CallbackAction {353        const self = maybe_server.?;354        const client = result catch return if (self.shutdown.load(.acquire)) .disarm else .rearm;355        if (self.shutdown.load(.acquire)) {356            sys.close(client.fd);357            return .disarm;358        }359        if (self.config.tcp_no_delay) {360            sys.setTcpNoDelay(client.fd) catch |err| {361                log.warn(362                    "failed to disable TCP delay: {s}",363                    .{@errorName(err)},364                );365                sys.close(client.fd);366                return .rearm;367            };368        }369370        const accepted_at = self.config.boot_clock.now() catch {371            sys.close(client.fd);372            return .rearm;373        };374        const slot = self.acquireConnection(client.fd, self.config.boot_clock, accepted_at) orelse {375            sys.close(client.fd);376            return .rearm;377        };378        slot.connection.setWriteTimeout(self.config.write_timeout_ms) catch |err| {379            log.warn("conn {d}: failed to set write timeout: {s}", .{ slot.connection.id, @errorName(err) });380        };381        self.armWaiting(loop, slot);382        return .rearm;383    }384385    fn readableReady(386        maybe_slot: ?*ServerConnectionSlot,387        _: *event.Loop,388        _: *event.Completion,389        _: event.File,390        result: event.PollError!event.PollEvent,391    ) event.CallbackAction {392        const slot = maybe_slot.?;393        _ = result catch {394            slot.connection.deinit();395            slot.server.releaseConnection(slot);396            return .disarm;397        };398        slot.server.dispatch(slot) catch {399            slot.connection.deinit();400            slot.server.releaseConnection(slot);401        };402        return .disarm;403    }404405    fn wakeReady(406        maybe_server: ?*Server,407        loop: *event.Loop,408        _: *event.Completion,409        result: event.Async.WaitError!void,410    ) event.CallbackAction {411        const self = maybe_server.?;412        _ = result catch return .disarm;413        self.drainCompleted(loop);414        if (!self.shutdown.load(.acquire)) return .rearm;415        self.closeListener(loop);416        self.closeConnections(loop);417        return .disarm;418    }419420    fn idleTimerReady(421        maybe_server: ?*Server,422        loop: *event.Loop,423        _: *event.Completion,424        result: event.Timer.RunError!void,425    ) event.CallbackAction {426        const self = maybe_server.?;427        _ = result catch return .disarm;428        if (self.shutdown.load(.acquire)) return .disarm;429430        const now = self.config.boot_clock.now() catch return .disarm;431        for (self.connection_slots) |*slot| {432            lock(&self.connections_mutex);433            const expired = slot.occupied and slot.phase == .waiting and434                idleExpired(now, slot.connection.lastActivity(), self.config.read_timeout_ms);435            if (expired) slot.phase = .closing;436            self.connections_mutex.unlock();437            if (expired) self.closeWaiting(loop, slot);438        }439        return .rearm;440    }441442    fn listenerClosed(443        _: ?*Server,444        _: *event.Loop,445        _: *event.Completion,446        _: event.TCP,447        result: event.CloseError!void,448    ) event.CallbackAction {449        _ = result catch {};450        return .disarm;451    }452453    fn connectionClosed(454        maybe_slot: ?*ServerConnectionSlot,455        _: *event.Loop,456        _: *event.Completion,457        _: event.TCP,458        result: event.CloseError!void,459    ) event.CallbackAction {460        const slot = maybe_slot.?;461        _ = result catch {};462        slot.connection.transportClosed();463        slot.connection.deinit();464        slot.server.releaseConnection(slot);465        return .disarm;466    }467468    fn dispatch(self: *Server, slot: *ServerConnectionSlot) !void {469        lock(&self.connections_mutex);470        std.debug.assert(slot.occupied);471        std.debug.assert(canDispatch(slot.phase));472        slot.phase = .running;473        self.connections_mutex.unlock();474        slot.connection.beginTurn();475        self.pool.submit(.{ .slot = slot }) catch |err| {476            lock(&self.connections_mutex);477            slot.phase = .waiting;478            self.connections_mutex.unlock();479            return err;480        };481    }482483    fn armWaiting(self: *Server, loop: *event.Loop, slot: *ServerConnectionSlot) void {484        lock(&self.connections_mutex);485        std.debug.assert(slot.occupied);486        std.debug.assert(canDispatch(slot.phase));487        slot.phase = .waiting;488        self.connections_mutex.unlock();489        event.File.initFd(slot.connection.socket).poll(490            loop,491            &slot.read_completion,492            .read,493            ServerConnectionSlot,494            slot,495            readableReady,496        );497    }498499    fn drainCompleted(self: *Server, loop: *event.Loop) void {500        while (self.popCompleted()) |slot| {501            const state = slot.connection.currentState();502            if (self.shutdown.load(.acquire) or state == .closing or state == .closed or !slot.connection.shouldWaitForRead()) {503                lock(&self.connections_mutex);504                slot.phase = .closing;505                self.connections_mutex.unlock();506                slot.connection.deinit();507                self.releaseConnection(slot);508            } else if (slot.connection.shouldConsumeBufferedInput() and slot.connection.hasBufferedInput()) {509                self.dispatch(slot) catch {510                    slot.connection.deinit();511                    self.releaseConnection(slot);512                };513            } else {514                self.armWaiting(loop, slot);515            }516        }517    }518519    fn popCompleted(self: *Server) ?*ServerConnectionSlot {520        lock(&self.connections_mutex);521        defer self.connections_mutex.unlock();522        const index = self.completed_connection_head orelse return null;523        const slot = &self.connection_slots[index];524        self.completed_connection_head = slot.next_completed;525        if (self.completed_connection_head == null) self.completed_connection_tail = null;526        slot.next_completed = null;527        return slot;528    }529530    fn closeListener(self: *Server, loop: *event.Loop) void {531        const socket = self.listener orelse return;532        self.listener = null;533        event.TCP.initFd(socket).close(534            loop,535            &self.listener_close_completion,536            Server,537            self,538            listenerClosed,539        );540    }541542    fn closeConnections(self: *Server, loop: *event.Loop) void {543        for (self.connection_slots) |*slot| {544            lock(&self.connections_mutex);545            if (!slot.occupied) {546                self.connections_mutex.unlock();547                continue;548            }549            const phase = slot.phase;550            if (phase == .waiting) slot.phase = .closing;551            self.connections_mutex.unlock();552            switch (phase) {553                .waiting => self.closeWaiting(loop, slot),554                .running => slot.connection.interrupt(),555                .completed => {},556                .free, .closing => {},557            }558        }559        self.drainCompleted(loop);560    }561562    fn closeWaiting(self: *Server, loop: *event.Loop, slot: *ServerConnectionSlot) void {563        _ = self;564        event.TCP.initFd(slot.connection.socket).close(565            loop,566            &slot.read_completion,567            ServerConnectionSlot,568            slot,569            connectionClosed,570        );571    }572};

1pub const Server = server.Server;