tiny.http.Connection

1pub const Connection = struct {2    id: usize,3    socket: sys.Socket,4    state: std.atomic.Value(State),5    transport: std.atomic.Value(TransportState),6    boot_clock: time.BootClock,7    last_activity_ns: std.atomic.Value(u64),8    input_storage: []u8,9    input_length: usize,10    request_scratch: message.RequestScratch,11    response_scratch: response.Scratch,12    input_capacity_rejections: std.atomic.Value(u64),13    write_timeout_ms: ?u32,14    protocol: ?Protocol,15    turn_disposition: TurnDisposition,1617    pub const State = enum(u8) {18        http,19        websocket,20        closing,21        closed,22    };2324    const TurnDisposition = enum {25        close,26        wait_for_read,27        consume_buffered_input,28    };2930    const TransportState = enum(u8) {31        open,32        interrupting,33        interrupted,34        closed,35    };3637    const Protocol = struct {38        context: *anyopaque,39        drive: *const fn (*anyopaque) anyerror!void,40        deinit: *const fn (*anyopaque) void,41    };4243    pub const InputLimits: type = StorageLimits;44    pub const InputCapacity: type = StorageCapacity;45    pub const InputStorage: type = ConnectionInputStorage;4647    pub fn init(48        id: usize,49        socket: sys.Socket,50        input_storage: []u8,51        request_scratch: message.RequestScratch,52        response_scratch: response.Scratch,53        boot_clock: time.BootClock,54        started_at: time.BootInstant,55    ) Connection {56        return .{57            .id = id,58            .socket = socket,59            .state = std.atomic.Value(State).init(.http),60            .transport = std.atomic.Value(TransportState).init(.open),61            .boot_clock = boot_clock,62            .last_activity_ns = std.atomic.Value(u64).init(started_at.asNanoseconds()),63            .input_storage = input_storage,64            .input_length = 0,65            .request_scratch = request_scratch,66            .response_scratch = response_scratch,67            .input_capacity_rejections = std.atomic.Value(u64).init(0),68            .write_timeout_ms = null,69            .protocol = null,70            .turn_disposition = .close,71        };72    }7374    pub fn deinit(self: *Connection) void {75        self.close();76        if (self.protocol) |protocol| protocol.deinit(protocol.context);77        self.protocol = null;78        self.input_length = 0;79        self.input_storage = &.{};80        self.request_scratch = .{81            .headers = &.{},82            .body = &.{},83            .header_line_bytes = 0,84        };85        self.response_scratch = .{ .headers = &.{}, .head = &.{} };86    }8788    pub fn close(self: *Connection) void {89        while (true) {90            const current = self.transport.load(.acquire);91            switch (current) {92                .closed => return,93                .interrupting => {94                    std.atomic.spinLoopHint();95                    continue;96                },97                .open, .interrupted => {},98            }99            if (self.transport.cmpxchgWeak(current, .closed, .acq_rel, .acquire) == null) break;100        }101        sys.close(self.socket);102        self.state.store(.closed, .release);103    }104105    pub fn takeSocket(self: *Connection) sys.Socket {106        std.debug.assert(self.protocol == null);107        std.debug.assert(self.transport.cmpxchgStrong(108            .open,109            .closed,110            .acq_rel,111            .acquire,112        ) == null);113        self.state.store(.closed, .release);114        return self.socket;115    }116117    pub fn interrupt(self: *Connection) void {118        if (self.transport.cmpxchgStrong(.open, .interrupting, .acq_rel, .acquire) != null) return;119        self.markClosing();120        sys.shutdownReadWrite(self.socket);121        self.transport.store(.interrupted, .release);122    }123124    pub fn transportClosed(self: *Connection) void {125        self.transport.store(.closed, .release);126        self.state.store(.closed, .release);127    }128129    pub fn currentState(self: *const Connection) State {130        return self.state.load(.acquire);131    }132133    pub fn markClosing(self: *Connection) void {134        if (self.transport.load(.acquire) != .closed) self.state.store(.closing, .release);135    }136137    pub fn markWebSocket(self: *Connection) void {138        std.debug.assert(self.currentState() == .http);139        self.state.store(.websocket, .release);140        self.waitForInput();141    }142143    pub fn beginTurn(self: *Connection) void {144        self.turn_disposition = .close;145    }146147    pub fn waitForRead(self: *Connection) void {148        self.turn_disposition = .wait_for_read;149    }150151    pub fn waitForInput(self: *Connection) void {152        self.turn_disposition = .consume_buffered_input;153    }154155    pub fn shouldWaitForRead(self: *const Connection) bool {156        return self.turn_disposition != .close;157    }158159    pub fn shouldConsumeBufferedInput(self: *const Connection) bool {160        return self.turn_disposition == .consume_buffered_input;161    }162163    pub fn lastActivity(self: *const Connection) time.BootInstant {164        return .fromNanoseconds(self.last_activity_ns.load(.acquire));165    }166167    pub fn hasBufferedInput(self: *const Connection) bool {168        return self.input_length != 0;169    }170171    pub fn bufferedInput(self: *const Connection) []const u8 {172        return self.input_storage[0..self.input_length];173    }174175    pub fn inputCapacity(self: *const Connection) usize {176        return self.input_storage.len;177    }178179    pub fn requestScratch(self: *Connection) message.RequestScratch {180        return self.request_scratch;181    }182183    pub fn responseScratch(self: *Connection) response.Scratch {184        return self.response_scratch;185    }186187    pub fn inputStatus(self: *const Connection) InputStatus {188        return .{189            .capacity_rejections = self.input_capacity_rejections.load(.acquire),190        };191    }192193    pub fn installProtocol(194        self: *Connection,195        context: anytype,196        comptime drive: anytype,197        comptime deinit_protocol: anytype,198    ) void {199        const Context = @TypeOf(context);200        const info = @typeInfo(Context);201        if (info != .pointer or info.pointer.size != .one) {202            @compileError("connection protocol context must be a single-item pointer");203        }204        const Callbacks = struct {205            fn run(erased: *anyopaque) anyerror!void {206                const typed: Context = @ptrCast(@alignCast(erased));207                return drive(typed);208            }209210            fn destroy(erased: *anyopaque) void {211                const typed: Context = @ptrCast(@alignCast(erased));212                deinit_protocol(typed);213            }214        };215        std.debug.assert(self.protocol == null);216        self.protocol = .{217            .context = @ptrCast(@constCast(context)),218            .drive = Callbacks.run,219            .deinit = Callbacks.destroy,220        };221        self.state.store(.websocket, .release);222        self.waitForInput();223    }224225    pub fn driveProtocol(self: *Connection) !void {226        const protocol = self.protocol orelse return error.ProtocolNotInstalled;227        try protocol.drive(protocol.context);228    }229230    pub fn read(self: *Connection, buf: []u8) !usize {231        if (self.transport.load(.acquire) == .closed) {232            return error.ConnectionClosed;233        }234        if (buf.len == 0) return 0;235        if (self.input_length > 0) {236            const n = @min(buf.len, self.input_length);237            @memcpy(buf[0..n], self.input_storage[0..n]);238            self.consumeBufferedInput(n);239            try self.noteActivity();240            return n;241        }242        return self.readSocket(buf);243    }244245    pub fn bufferInput(self: *Connection) !usize {246        if (self.input_length == self.input_storage.len) {247            self.recordInputCapacityRejection();248            return error.InputCapacityExceeded;249        }250        const n = try self.readSocket(self.input_storage[self.input_length..]);251        self.input_length += n;252        return n;253    }254255    pub fn retainInput(self: *Connection, data: []const u8) InputExhaustion!void {256        if (data.len > self.input_storage.len - self.input_length) {257            self.recordInputCapacityRejection();258            return error.InputCapacityExceeded;259        }260        @memcpy(self.input_storage[self.input_length..][0..data.len], data);261        self.input_length += data.len;262    }263264    pub fn consumeBufferedInput(self: *Connection, count: usize) void {265        std.debug.assert(count <= self.input_length);266        const remaining = self.input_length - count;267        std.mem.copyForwards(268            u8,269            self.input_storage[0..remaining],270            self.input_storage[count..self.input_length],271        );272        self.input_length = remaining;273    }274275    fn recordInputCapacityRejection(self: *Connection) void {276        var current = self.input_capacity_rejections.load(.acquire);277        while (current != std.math.maxInt(u64)) {278            if (self.input_capacity_rejections.cmpxchgWeak(279                current,280                current + 1,281                .acq_rel,282                .acquire,283            )) |observed| {284                current = observed;285            } else {286                return;287            }288        }289    }290291    fn readSocket(self: *Connection, buf: []u8) !usize {292        std.debug.assert(buf.len != 0);293        if (self.transport.load(.acquire) == .closed) {294            return error.ConnectionClosed;295        }296        const n = sys.recv(self.socket, buf, 0) catch |err| switch (err) {297            error.BadFileDescriptor => {298                self.markClosing();299                return error.ConnectionClosed;300            },301            error.ConnectionResetByPeer,302            error.ConnectionTimedOut,303            => {304                self.markClosing();305                return error.ConnectionClosed;306            },307            error.WouldBlock => return error.WouldBlock,308            else => return err,309        };310        if (n == 0) {311            self.markClosing();312            return error.ConnectionClosed;313        }314        try self.noteActivity();315        return n;316    }317318    pub fn write(self: *Connection, data: []const u8) !void {319        if (self.transport.load(.acquire) == .closed) {320            return error.ConnectionClosed;321        }322        var sent: usize = 0;323        while (sent < data.len) {324            const n = sys.sendNoSignal(self.socket, data[sent..]) catch |err| switch (err) {325                error.ConnectionResetByPeer,326                error.BrokenPipe,327                => {328                    self.markClosing();329                    return error.ConnectionClosed;330                },331                error.WouldBlock => {332                    try self.waitWritable();333                    continue;334                },335                else => return err,336            };337            if (n == 0) {338                self.markClosing();339                return error.ConnectionClosed;340            }341            sent += n;342            try self.noteActivity();343        }344    }345346    fn waitWritable(self: *Connection) !void {347        const ready = try sys.pollWritable(self.socket, self.writePollTimeoutMs());348        if (!ready) {349            self.markClosing();350            return error.ConnectionTimedOut;351        }352    }353354    fn noteActivity(self: *Connection) time.ClockError!void {355        const now = try self.boot_clock.now();356        self.last_activity_ns.store(now.asNanoseconds(), .release);357    }358359    fn writePollTimeoutMs(self: *const Connection) i32 {360        const timeout = self.write_timeout_ms orelse return -1;361        const max_timeout: u32 = @intCast(std.math.maxInt(i32));362        return @intCast(@min(timeout, max_timeout));363    }364365    pub fn setReadTimeout(self: *Connection, timeout_ms: u32) !void {366        try sys.setReadTimeout(self.socket, timeout_ms);367    }368369    pub fn setWriteTimeout(self: *Connection, timeout_ms: u32) !void {370        try sys.setWriteTimeout(self.socket, timeout_ms);371        self.write_timeout_ms = if (timeout_ms == 0) null else timeout_ms;372    }373};

1pub const Connection = connection.Connection;