Something went wrong. Try again.
websocket forked from zzstoatzz.io/websocket.zig
Something went wrong. Try again.
142 kB · 3674 lines
Zig
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675const std = @import("std");const builtin = @import("builtin");const proto = @import("../proto.zig");const buffer = @import("../buffer.zig");const deflate = @import("../deflate.zig");
const libc = std.c;const Io = std.Io;const net = Io.net;const posix = std.posix;const Thread = std.Thread;const Allocator = std.mem.Allocator;const FixedBufferAllocator = std.heap.FixedBufferAllocator;
const log = std.log.scoped(.websocket);
// Cross-platform O_NONBLOCK via zig's typed packed struct — avoids hardcoded platform constantsconst O_NONBLOCK: c_int = @bitCast(libc.O{ .NONBLOCK = true });
// 0.16: Address removed, create compatibility shimconst Address = struct { // sockaddr.storage so unix (110B) and in6 (28B) addresses actually fit — // a bare sockaddr is 16 bytes. any: posix.sockaddr.storage,
const Self = @This();
pub fn initUnix(path: []const u8) !Self { var addr: posix.sockaddr.storage = std.mem.zeroes(posix.sockaddr.storage); const sun = @as(*posix.sockaddr.un, @ptrCast(@alignCast(&addr))); sun.* = .{ .path = @splat(0) }; sun.family = posix.AF.UNIX; if (path.len >= sun.path.len) return error.NameTooLong; @memcpy(sun.path[0..path.len], path); return .{ .any = addr }; }
pub fn parseIp(address: []const u8, port: u16) !Self { var addr: posix.sockaddr.storage = std.mem.zeroes(posix.sockaddr.storage); // IPv6 any ("::") — required for IPv6-only networks (e.g. fly 6PN). // On linux V6ONLY defaults off, so this accepts IPv4 connections too. if (std.mem.eql(u8, address, "::")) { const in6 = @as(*posix.sockaddr.in6, @ptrCast(@alignCast(&addr))); in6.family = posix.AF.INET6; in6.port = @byteSwap(port); if (@hasField(posix.sockaddr.in6, "len")) in6.len = @sizeOf(posix.sockaddr.in6); return .{ .any = addr }; } const in = @as(*posix.sockaddr.in, @ptrCast(@alignCast(&addr))); in.family = posix.AF.INET; in.port = @byteSwap(port); if (@hasField(posix.sockaddr.in, "len")) in.len = @sizeOf(posix.sockaddr.in); // Parse simple IPv4 addresses if (std.mem.eql(u8, address, "0.0.0.0")) { in.addr = 0; } else if (std.mem.eql(u8, address, "127.0.0.1")) { in.addr = 0x0100007f; } else { // Generic parse var parts: [4]u8 = undefined; var part_idx: usize = 0; var num: u8 = 0; for (address) |byte| { if (byte == '.') { if (part_idx >= 4) return error.InvalidAddress; parts[part_idx] = num; part_idx += 1; num = 0; } else if (byte >= '0' and byte <= '9') { num = num * 10 + (byte - '0'); } else { return error.InvalidAddress; } } if (part_idx != 3) return error.InvalidAddress; parts[part_idx] = num; in.addr = @as(u32, parts[0]) | (@as(u32, parts[1]) << 8) | (@as(u32, parts[2]) << 16) | (@as(u32, parts[3]) << 24); } return .{ .any = addr }; }
// 0.16: format signature changed - no more FormatOptions, Writer is std.Io.Writer pub fn format(self: Self, writer: *std.Io.Writer) std.Io.Writer.Error!void { if (self.any.family == posix.AF.UNIX) { const sun = @as(*const posix.sockaddr.un, @ptrCast(@alignCast(&self.any))); try writer.print("unix:{s}", .{std.mem.sliceTo(&sun.path, 0)}); } else if (self.any.family == posix.AF.INET) { const in = @as(*const posix.sockaddr.in, @ptrCast(@alignCast(&self.any))); const addr = @as([4]u8, @bitCast(in.addr)); try writer.print("{}.{}.{}.{}:{}", .{ addr[0], addr[1], addr[2], addr[3], @byteSwap(in.port) }); } else if (self.any.family == posix.AF.INET6) { const in6 = @as(*const posix.sockaddr.in6, @ptrCast(@alignCast(&self.any))); try writer.print("[ipv6]:{}", .{@byteSwap(in6.port)}); } else { try writer.print("unknown", .{}); } }
/// Write only the peer IP, excluding its ephemeral source port. IPv6 is /// emitted in valid uncompressed hexadecimal form; callers that need a /// stable per-host key (rate limiting, abuse accounting) must not use the /// ordinary `format`, whose port changes on every connection. fn formatIp(self: Self, writer: *std.Io.Writer) std.Io.Writer.Error!void { if (self.any.family == posix.AF.INET) { const in = @as(*const posix.sockaddr.in, @ptrCast(@alignCast(&self.any))); const addr = @as([4]u8, @bitCast(in.addr)); try writer.print("{d}.{d}.{d}.{d}", .{ addr[0], addr[1], addr[2], addr[3] }); } else if (self.any.family == posix.AF.INET6) { const in6 = @as(*const posix.sockaddr.in6, @ptrCast(@alignCast(&self.any))); for (0..8) |index| { if (index > 0) try writer.writeByte(':'); const at = index * 2; try writer.print("{x}", .{std.mem.readInt(u16, in6.addr[at..][0..2], .big)}); } } else if (self.any.family == posix.AF.UNIX) { try writer.writeAll("unix"); } else { try writer.writeAll("unknown"); } }
pub fn getPort(self: Self) u16 { const in = @as(*const posix.sockaddr.in, @ptrCast(@alignCast(&self.any))); return @byteSwap(in.port); }
pub fn getOsSockLen(self: Self) posix.socklen_t { if (self.any.family == posix.AF.UNIX) { return @sizeOf(posix.sockaddr.un); } if (self.any.family == posix.AF.INET6) { return @sizeOf(posix.sockaddr.in6); } return @sizeOf(posix.sockaddr.in); }};
// Wraps a socket handle for proto.Reader.fill. This path uses posix.read// directly because socket-level SO_RCVTIMEO reports EAGAIN, which std.Io's// blocking net_read operation intentionally does not expose.const SocketReader = struct { socket: posix.socket_t,
pub fn read(self: SocketReader, buf: []u8) !usize { return posix.read(self.socket, buf) catch |err| { return switch (err) { error.ConnectionResetByPeer => error.ConnectionResetByPeer, error.WouldBlock => error.WouldBlock, else => error.Unexpected, }; }; }};
fn socketWriteAll(io: Io, socket: posix.socket_t, data: []const u8) !void { var remaining = data; while (remaining.len > 0) { // netWrite: header is sent first, data array's last element is the splat pattern. // Pass remaining as header, empty pattern with splat=0. const empty = [_][]const u8{""}; const n = io.vtable.netWrite(io.userdata, socket, remaining, &empty, 0) catch |err| { return switch (err) { error.ConnectionResetByPeer => error.ConnectionResetByPeer, else => error.Unexpected, }; }; if (n == 0) return error.Unexpected; remaining = remaining[n..]; }}
const OpCode = proto.OpCode;const Reader = proto.Reader;const Message = proto.Message;pub const Handshake = @import("handshake.zig").Handshake;const Compression = @import("../websocket.zig").Compression;const FallbackAllocator = @import("fallback_allocator.zig").FallbackAllocator;
const DEFAULT_MAX_CONN = 16_384;const DEFAULT_BUFFER_SIZE = 2048;const DEFAULT_MAX_MESSAGE_SIZE = 65_536;
const EMPTY_PONG = ([2]u8{ @backingInt(OpCode.pong), 0 })[0..];// CLOSE, 2 length, codeconst CLOSE_NORMAL = ([_]u8{ @backingInt(OpCode.close), 2, 3, 232 })[0..]; // code: 1000const CLOSE_PROTOCOL_ERROR = ([_]u8{ @backingInt(OpCode.close), 2, 3, 234 })[0..]; //code: 1002
const force_blocking: bool = blk: { const build = @import("build"); if (@hasDecl(build, "websocket_blocking")) { break :blk build.websocket_blocking; } break :blk false;};
pub fn blockingMode() bool { if (force_blocking) { return true; } return switch (builtin.os.tag) { .linux, .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => false, else => true, };}
pub const Config = struct { port: u16 = 9882, address: []const u8 = "127.0.0.1", unix_path: ?[]const u8 = null,
worker_count: ?u8 = null,
max_conn: ?usize = null, max_message_size: ?usize = null,
handshake: Config.Handshake = .{}, thread_pool: ThreadPool = .{}, buffers: Config.Buffers = .{}, compression: ?Compression = null,
/// Time allowed for upgraded WebSocket peers to answer an application /// `serverClose` hook before their transports are forced down. Ordinary /// HTTP fallback requests have an independent budget below. Both clocks /// are measured from the same `runIo` or blocking-mode shutdown instant, /// so the drain waits use the larger value rather than their sum. /// The legacy nonblocking event-loop listener still stops immediately. websocket_shutdown_grace: Io.Duration = .fromSeconds(1), http_shutdown_grace: Io.Duration = .fromSeconds(1),
pub const ThreadPool = struct { count: ?u16 = null, backlog: ?u32 = null, buffer_size: ?usize = null, };
pub const Handshake = struct { timeout: u32 = 10, max_size: ?u16 = null, max_headers: ?u16 = null, max_res_headers: ?u16 = null, count: ?u16 = null, };
pub const Buffers = struct { small_size: ?usize = null, small_pool: ?usize = null, large_size: ?usize = null, large_pool: ?u16 = null, };
fn workerCount(self: *const Config) usize { if (comptime blockingMode()) { return 1; } return self.worker_count orelse 1; }};
pub fn Server(comptime H: type) type { return struct { config: Config, allocator: Allocator, io: Io,
_state: WorkerState, _signals: []posix.fd_t, _mut: Io.Mutex, _cond: Io.Condition,
const Self = @This();
pub fn init(allocator: Allocator, io: Io, config: Config) !Self { if (blockingMode()) { if (config.buffers.small_pool) |p| { if (p > 1) { log.warn("blockingMode() cannot utilize a small buffer pool, using per-connection buffer instead", .{}); } } }
const signals = try allocator.alloc(posix.fd_t, config.workerCount()); errdefer allocator.free(signals);
var state = try WorkerState.init(allocator, io, config); errdefer state.deinit();
return .{ .io = io, ._mut = .init, ._cond = .init, ._state = state, ._signals = signals, .config = config, .allocator = allocator, }; }
pub fn deinit(self: *Self) void { self._state.deinit(); self.allocator.free(self._signals); }
pub fn listenInNewThread(self: *Self, ctx: anytype) !Thread { const io = self.io; self._mut.lockUncancelable(io); defer self._mut.unlock(io); const thrd = try Thread.spawn(.{}, Self.listen, .{ self, ctx });
// we don't return until listen() signals us that the server is up self._cond.waitUncancelable(io, &self._mut); return thrd; }
pub fn listen(self: *Self, ctx: anytype) !void { const io = self.io; self._mut.lockUncancelable(io); errdefer { self._cond.signal(io); self._mut.unlock(io); }
const config = &self.config;
var no_delay = true; const address = blk: { if (config.unix_path) |unix_path| { if (comptime net.has_unix_sockets == false) { return error.UnixPathNotSupported; } no_delay = false; // 0.16: deleteFileAbsolute moved to Io.Dir Io.Dir.deleteFileAbsolute(io, unix_path) catch {}; break :blk try Address.initUnix(unix_path); } else { const listen_port = config.port; const listen_address = config.address; break :blk try Address.parseIp(listen_address, listen_port); } };
// 0.16: posix.socket removed, use libc // Note: SOCK.CLOEXEC/NONBLOCK not supported by darwin libc, set via fcntl const socket = blk: { const sock_flags: c_uint = libc.SOCK.STREAM; const socket_proto: c_uint = if (address.any.family == posix.AF.UNIX) 0 else libc.IPPROTO.TCP; const fd = libc.socket(@intCast(address.any.family), sock_flags, socket_proto); if (fd == -1) return error.SocketError; _ = libc.fcntl(fd, libc.F.SETFD, @as(c_int, libc.FD_CLOEXEC)); if (blockingMode() == false) { const flags = libc.fcntl(fd, libc.F.GETFL); _ = libc.fcntl(fd, libc.F.SETFL, flags | O_NONBLOCK); } break :blk fd; };
if (no_delay) { // TODO: Broken on darwin: // https://github.com/ziglang/zig/issues/17260 // if (@hasDecl(os.TCP, "NODELAY")) { // try os.setsockopt(socket.sockfd.?, os.IPPROTO.TCP, os.TCP.NODELAY, &std.mem.toBytes(@as(c_int, 1))); // } try posix.setsockopt(socket, posix.IPPROTO.TCP, 1, &std.mem.toBytes(@as(c_int, 1))); }
if (@hasDecl(posix.SO, "REUSEPORT_LB")) { try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.REUSEPORT_LB, &std.mem.toBytes(@as(c_int, 1))); } else if (@hasDecl(posix.SO, "REUSEPORT")) { try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.REUSEPORT, &std.mem.toBytes(@as(c_int, 1))); } else { try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.REUSEADDR, &std.mem.toBytes(@as(c_int, 1))); }
{ // 0.16: posix.bind/listen removed, use libc const socklen = address.getOsSockLen(); if (libc.bind(socket, @ptrCast(&address.any), socklen) == -1) return error.BindError; if (libc.listen(socket, 1024) == -1) return error.ListenError; }
const C = @TypeOf(ctx);
if (comptime blockingMode()) { errdefer _ = libc.close(socket); var w = try Blocking(H).init(self.allocator, &self._state); defer w.deinit();
const thrd = try std.Thread.spawn(.{}, Blocking(H).run, .{ &w, socket, ctx }); log.info("starting blocking worker to listen on {f}", .{address});
// incase listenInNewThread was used and is waiting for us to start self._cond.signal(io);
// this is what we'll shutdown when stop() is called self._signals[0] = socket; self._mut.unlock(io); thrd.join(); } else { defer _ = libc.close(socket); const W = NonBlocking(H, C);
const allocator = self.allocator;
var signals = self._signals; const worker_count = signals.len; const threads = try allocator.alloc(Thread, worker_count); const workers = try allocator.alloc(W, worker_count);
var started: usize = 0;
errdefer for (0..started) |i| { // on success, these will be closed by a call to stop(); _ = libc.close(signals[i]); };
defer { for (0..started) |i| { workers[i].deinit(); } allocator.free(threads); allocator.free(workers); }
for (0..worker_count) |i| { // 0.16: posix.pipe2 removed, use libc var pipe: [2]c_int = undefined; if (libc.pipe(&pipe) == -1) return error.PipeError; errdefer _ = libc.close(pipe[1]); const flags0 = libc.fcntl(pipe[0], libc.F.GETFL); _ = libc.fcntl(pipe[0], libc.F.SETFL, flags0 | O_NONBLOCK); const flags1 = libc.fcntl(pipe[1], libc.F.GETFL); _ = libc.fcntl(pipe[1], libc.F.SETFL, flags1 | O_NONBLOCK);
workers[i] = try W.init(self.allocator, &self._state, ctx); errdefer workers[i].deinit();
threads[i] = try Thread.spawn(.{}, W.run, .{ &workers[i], socket, pipe[0] });
signals[i] = pipe[1]; started += 1; }
log.info("starting nonblocking worker to listen on {f}", .{address});
// in case startInNewThread is waiting self._cond.signal(io);
self._mut.unlock(io);
for (threads) |thrd| { thrd.join(); } } self._cond.signal(io); }
pub fn stop(self: *Self) void { const io = self.io; self._mut.lockUncancelable(io); defer self._mut.unlock(io); for (self._signals) |s| { if (blockingMode()) { _ = libc.shutdown(s, 0); // SHUT_RD = 0 } _ = libc.close(s); } self._cond.waitUncancelable(io, &self._mut); }
/// Io-native accept loop. Replaces listen()/listenInNewThread() for /// callers that already have an Io context. /// /// The caller creates and owns the listener. Shutdown by closing the /// listener (listener.deinit(io)), which unblocks accept and causes /// this function to return. /// /// Under Evented: accept yields the fiber, io.concurrent spawns a /// fiber per connection. Under Threaded: accept blocks, io.concurrent /// spawns a thread per connection. /// /// Usage: /// const addr = net.Ip4Address.unspecified(port); /// var listener = (net.IpAddress{ .ip4 = addr }).listen(io, .{ .reuse_address = true }); /// var future = try io.concurrent(Server(H).runIo, .{ &server, &listener, &ctx }); /// // shutdown: /// listener.deinit(io); /// future.cancel(io); pub fn runIo(self: *Self, listener: *net.Server, ctx: anytype) void { const Ctx = @TypeOf(ctx); const io = self.io; const config = &self.config;
var handler = Blocking(H).init(self.allocator, &self._state) catch |err| { log.err("failed to init connection handler: {}", .{err}); return; }; defer handler.deinit();
const max_conn = config.max_conn orelse DEFAULT_MAX_CONN;
// connection tasks are owned by this group — canceled on shutdown var connections: Io.Group = .init;
// concrete wrapper: Group.concurrent needs ArgsTuple, which can't handle anytype const Handle = struct { fn connection(h: *Blocking(H), socket: posix.socket_t, address: Address, c: Ctx) void { h.handleConnection(socket, address, c); } };
log.info("Io-native accept loop started", .{});
while (true) { const stream = listener.accept(io) catch |err| { switch (err) { error.SocketNotListening => { log.info("listener closed, shutting down", .{}); handler.shutdown(); connections.cancel(io); return; }, error.Canceled => { handler.shutdown(); connections.cancel(io); return; }, else => { log.err("accept error: {s}", .{@errorName(err)}); continue; }, } };
const socket = stream.socket.handle;
// enforce connection limit if (handler.conn_manager.count() >= max_conn) { stream.close(io); continue; }
// TCP_NODELAY for TCP connections if (config.unix_path == null) { setSockOptBestEffort(socket, posix.IPPROTO.TCP, 1, &std.mem.toBytes(@as(c_int, 1))); }
// CLOEXEC _ = libc.fcntl(socket, libc.F.SETFD, @as(c_int, libc.FD_CLOEXEC));
// peer address for logging var address: Address = undefined; var address_len: posix.socklen_t = @sizeOf(posix.sockaddr.storage); _ = libc.getpeername(socket, @ptrCast(&address.any), &address_len);
log.debug("({f}) connected", .{address});
// spawn fiber (Evented) or thread (Threaded) per connection connections.concurrent(io, Handle.connection, .{ &handler, socket, address, ctx }) catch { stream.close(io); continue; }; } } };}
// This is our Blocking worker. It's very different than NonBlocking and much simpler.pub fn Blocking(comptime H: type) type { return struct { allocator: Allocator, handshake_timeout: Timeout, connection_buffer_size: usize, conn_manager: ConnManager(H, false), handshake_pool: *Handshake.Pool, buffer_provider: *buffer.Provider, compression: ?Compression, websocket_shutdown_grace: Io.Duration, http_shutdown_grace: Io.Duration,
const Timeout = struct { sec: u32, timeval: [@sizeOf(std.posix.timeval)]u8,
// if sec is null, it means we want to cancel the timeout. fn init(sec: u32) Timeout { return .{ .sec = sec, .timeval = std.mem.toBytes(posix.timeval{ .sec = @intCast(sec), .usec = 0 }), }; }
pub const none = std.mem.toBytes(posix.timeval{ .sec = 0, .usec = 0 }); };
const Self = @This();
pub fn init(allocator: Allocator, state: *WorkerState) !Self { const config = &state.config;
var conn_manager = try ConnManager(H, false).init(allocator, state.io, config.compression); errdefer conn_manager.deinit();
return .{ .conn_manager = conn_manager, .allocator = allocator, .compression = config.compression, .handshake_pool = state.handshake_pool, .buffer_provider = &state.buffer_provider, .handshake_timeout = Timeout.init(config.handshake.timeout), .connection_buffer_size = config.buffers.small_size orelse DEFAULT_BUFFER_SIZE, .websocket_shutdown_grace = config.websocket_shutdown_grace, .http_shutdown_grace = config.http_shutdown_grace, }; }
pub fn deinit(self: *Self) void { self.conn_manager.deinit(); }
pub fn run(self: *Self, listener: posix.socket_t, ctx: anytype) void { defer self.shutdown(); while (true) { var address: Address = undefined; var address_len: posix.socklen_t = @sizeOf(posix.sockaddr.storage); // 0.16: posix.accept removed, use libc.accept const socket = libc.accept(listener, @ptrCast(&address.any), &address_len); if (socket == -1) { const err = posix.errno(-1); switch (err) { .INTR, .CONNABORTED => continue, // stop() shuts down and closes the listening socket. // Linux can report either EINVAL or EBADF depending on // which syscall wins the race with accept(). .INVAL, .BADF, .NOTSOCK => { log.info("received shutdown signal", .{}); return; }, else => { log.err("failed to accept socket: {}", .{err}); // Resource errors can persist. Bound the retry rate // so a degraded listener cannot become a CPU/log // amplification loop. self.conn_manager.io.sleep(.fromMilliseconds(10), .awake) catch {}; continue; }, } } // Set CLOEXEC on accepted socket _ = libc.fcntl(socket, libc.F.SETFD, @as(c_int, libc.FD_CLOEXEC)); log.debug("({f}) connected", .{address});
const thread = std.Thread.spawn(.{}, Self.handleConnection, .{ self, socket, address, ctx }) catch |err| { _ = libc.close(socket); log.err("({f}) failed to spawn connection thread: {}", .{ address, err }); continue; }; thread.detach(); } }
// Called in a thread started above in listen. // Wrapper around _handleConnection so that we can handle erros fn handleConnection(self: *Self, socket: posix.socket_t, address: Address, ctx: anytype) void { self._handleConnection(socket, address, ctx) catch |err| { log.err("({f}) uncaught error in connection handler: {}", .{ address, err }); }; }
fn _handleConnection(self: *Self, socket: posix.socket_t, address: Address, ctx: anytype) !void { const conn_manager = &self.conn_manager; const hc = try conn_manager.create(socket, address, timestamp());
{ // Do our handshake errdefer self.cleanupConn(hc); const timeout = self.handshake_timeout; const deadline = timestamp() + timeout.sec; setSockOptBestEffort(socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &timeout.timeval);
while (true) { const compression, const ok = handleHandshake(H, self, hc, ctx); if (ok == false) { self.cleanupConn(hc); return; } if (hc.handler != null) { // if we have a handler, the our handshake completed if (compression) { try conn_manager.setupCompression(hc); } try afterInit(H, hc, ctx); break; } if (timestamp() > deadline) { self.cleanupConn(hc); return; } } }
return self.readLoop(hc); }
// The readloop is extracted from _handleConnection so that it can be called // directly when integrating with an http server pub fn readLoop(self: *Self, hc: *HandlerConn(H)) !void { defer self.cleanupConn(hc); setSockOptBestEffort(hc.socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &Timeout.none);
// In BlockingMode, we always assign a reader for the duration of the connection // In scenarios where client rarely send data, this is going to use up an unecessary amount // of memory, but unlike nonblocking mode, we can't initialize the buffer just-in-time. const reader_buf = try self.allocator.alloc(u8, self.connection_buffer_size); defer self.allocator.free(reader_buf);
hc.reader = Reader.init(reader_buf, self.buffer_provider, hc.compression); while (true) { if (handleClientData(H, hc, self.allocator, null) == false) { break; } } }
fn cleanupConn(self: *Self, hc: *HandlerConn(H)) void { hc.conn.closeSocket(); self.conn_manager.cleanup(hc); }
fn discardConn(self: *Self, hc: *HandlerConn(H)) void { self.conn_manager.cleanup(hc); }
fn shutdown(self: *Self) void { const conn_manager = &self.conn_manager; const started = Io.Timestamp.now(conn_manager.io, .awake).toNanoseconds(); var notifications: Io.Group = .init; conn_manager.beginShutdownConcurrent(¬ifications);
self.waitForConnections(started, self.websocket_shutdown_grace, true); conn_manager.interruptWebSockets(self); conn_manager.forceWebSockets(self); self.waitForConnections(started, self.http_shutdown_grace, false); conn_manager.forceAll(self); notifications.await(conn_manager.io) catch {}; }
fn interruptShutdown(_: *Self, hc: *HandlerConn(H)) void { hc.conn.shutdownSocket(.both); }
fn waitForConnections(self: *Self, started_ns: i96, grace: Io.Duration, websocket_only: bool) void { const conn_manager = &self.conn_manager; const deadline = started_ns + grace.nanoseconds; while (if (websocket_only) conn_manager.websocketCount() > 0 else conn_manager.count() > 0) { const now = Io.Timestamp.now(conn_manager.io, .awake).toNanoseconds(); if (now >= deadline) return; const remaining = Io.Duration.fromNanoseconds(@min(deadline - now, 10 * std.time.ns_per_ms)); conn_manager.io.sleep(remaining, .awake) catch {}; } }
// called for each hc when shutting down fn shutdownCleanup(_: *Self, hc: *HandlerConn(H)) bool { // Wake both readers and writers without racing their next syscall // against a closed descriptor. The connection task owns the final // close after it observes EOF/error or cancellation. hc.conn.shutdownSocket(.both); return false; } };}
fn NonBlocking(comptime H: type, comptime C: type) type { return struct { ctx: C,
max_conn: usize,
// KQueue or Epoll, depending on the platform loop: Loop, thread_pool: *ThreadPool,
handshake_timeout: u32, handshake_pool: *Handshake.Pool, base: NonBlockingBase(H, true), compression: ?Compression,
const Self = @This(); const ThreadPool = @import("thread_pool.zig").ThreadPool(Self.dataAvailable);
pub fn init(allocator: Allocator, state: *WorkerState, ctx: C) !Self { var base = try NonBlockingBase(H, true).init(allocator, state); errdefer base.deinit();
const loop = try Loop.init(); errdefer loop.deinit();
const config = &state.config; var thread_pool = try ThreadPool.init(allocator, .{ .count = config.thread_pool.count orelse 4, .backlog = config.thread_pool.backlog orelse 500, .buffer_size = config.thread_pool.buffer_size orelse if (needsAllocator(H)) 32_768 else 0, }); errdefer thread_pool.deinit();
return .{ .ctx = ctx, .loop = loop, .base = base, .thread_pool = thread_pool, .compression = config.compression, .handshake_pool = state.handshake_pool, .handshake_timeout = state.config.handshake.timeout, .max_conn = config.max_conn orelse DEFAULT_MAX_CONN, }; }
pub fn deinit(self: *Self) void { self.loop.deinit(); self.base.deinit(); self.thread_pool.deinit(); }
fn run(self: *Self, listener: posix.socket_t, signal: posix.fd_t) void { self.loop.monitorAccept(listener) catch |err| { log.err("failed to add monitor to listening socket: {}", .{err}); return; };
self.loop.monitorSignal(signal) catch |err| { log.err("failed to add monitor to signal pipe: {}", .{err}); return; };
const thread_pool = self.thread_pool; const conn_manager = &self.base.conn_manager; const handshake_timeout = self.handshake_timeout;
var now = timestamp(); var oldest_pending_start_time: ?u32 = null; while (true) { const handshake_cutoff = now - handshake_timeout; const timeout = self.prepareToWait(handshake_cutoff) orelse blk: { if (oldest_pending_start_time) |started| { break :blk if (started < handshake_cutoff) 1 else @as(i32, @intCast(started - handshake_cutoff)); } break :blk null; };
var it = self.loop.wait(timeout) catch |err| { log.err("failed to wait on events: {}", .{err}); // 0.16: Thread.sleep removed, use libc.nanosleep _ = libc.nanosleep(&.{ .sec = 1, .nsec = 0 }, null); continue; };
now = timestamp(); oldest_pending_start_time = null; while (it.next()) |data| { if (data == 0) { self.accept(listener, now) catch |err| { log.err("accept error: {}", .{err}); // 0.16: Thread.sleep removed, use libc.nanosleep _ = libc.nanosleep(&.{ .sec = 0, .nsec = 1_000_000 }, null); }; continue; }
if (data == 1) { self.base.shutdown(); return; }
const hc: *HandlerConn(H) = @ptrFromInt(data); if (hc.state == .handshake) { // we need to get this out of the pending list, so that it doesn't // cause a timeout while we're processing it. // But we stll need to care about its timeout. if (oldest_pending_start_time) |s| { oldest_pending_start_time = @min(hc.conn.started, s); } else { oldest_pending_start_time = hc.conn.started; } conn_manager.activate(hc); } thread_pool.spawn(.{ self, hc }); } } }
// Enforces timeouts, and returns when the next timeout should be checked. fn prepareToWait(self: *Self, cutoff: u32) ?i32 { const cm = &self.base.conn_manager;
// always ordered from oldest to newest, so once we find a conneciton // that isn't timed out, we can stop cm.lock.lockUncancelable(cm.io); defer cm.lock.unlock(cm.io);
var next_conn = cm.pending.head; while (next_conn) |hc| { const conn = &hc.conn; const started = conn.started; if (started > cutoff) { // This is the first connection which hasn't timed out // return the time until it times out. return @intCast(started - cutoff); }
next_conn = hc.next;
// this connection has timed out. Don't use self.cleanup since there's // a bunch of stuff we can assume here..like there's no handler or reader conn.closeSocket(); log.debug("({f}) handshake timeout", .{conn.address}); if (hc.handshake) |h| { h.release(); } cm.pending.remove(hc); cm.pool.destroy(hc); } return null; }
fn accept(self: *Self, listener: posix.fd_t, now: u32) !void { const max_conn = self.max_conn; const conn_manager = &self.base.conn_manager;
while (conn_manager.count() < max_conn) { // 0.16: use local Address type instead of Address var address: Address = undefined; var address_len: posix.socklen_t = @sizeOf(posix.sockaddr.storage);
// 0.16: posix.accept removed, use libc.accept const socket = libc.accept(listener, @ptrCast(&address.any), &address_len); if (socket == -1) { const err = posix.errno(-1); // When available, we use SO_REUSEPORT_LB or SO_REUSEPORT, so WouldBlock // should not be possible in those cases, but if it isn't available // this error should be ignored as it means another thread picked it up. // darwin: AGAIN is the same as WOULDBLOCK return if (err == .AGAIN) {} else error.AcceptError; } // Set CLOEXEC on accepted socket _ = libc.fcntl(socket, libc.F.SETFD, @as(c_int, libc.FD_CLOEXEC));
log.debug("({f}) connected", .{address});
{ errdefer _ = libc.close(socket); // socket is _probably_ in NONBLOCKING mode (it inherits // the flag from the listening socket). const flags = libc.fcntl(socket, libc.F.GETFL); if (flags & O_NONBLOCK == O_NONBLOCK) { // Yup, it's in nonblocking mode. Disable that flag to // put it in blocking mode. _ = libc.fcntl(socket, libc.F.SETFL, flags & ~O_NONBLOCK); } } const hc = try self.base.newConn(socket, address, now); self.loop.monitorRead(hc, false) catch |err| { self.base.cleanupConn(hc); return err; }; } }
// Called in a thread-pool thread/ // !! Access to self has to be synchronized !! // There can only be 1 dataAvailable executing per HC at any given time. // Access to HC *should not* need to be synchronized. Access to hc.conn // needs to be synchronized once &hc.conn is passed to the handler during // handshake (Conn is self-synchronized). // Shutdown throws a wrench in our synchronization model, since we could // be shutting down while we're processing data...but hopefully the way we // shutdown (by waiting for all the thread pools threads to end) solves this. // Else, we'll need to throw a bunch of locking around HC just to handle shutdown. fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) void { var success = false; { hc.cleanup.lockUncancelable(hc.conn.io); defer hc.cleanup.unlock(hc.conn.io); if (hc.handler == null) { success = self.dataForHandshake(hc) catch |err| blk: { log.err("({f}) error processing handshake: {}", .{ hc.conn.address, err }); break :blk false; }; } else { success = self.base.dataAvailable(hc, thread_buf); } }
var conn = &hc.conn; var closed: bool = undefined; if (success == false) { conn.closeSocket(); closed = true; } else { closed = conn.isClosed(); }
if (closed) { self.base.cleanupConn(hc); } else { self.loop.monitorRead(hc, true) catch |err| { log.debug("({f}) failed to add read event monitor: {}", .{ conn.address, err }); conn.closeSocket(); self.base.cleanupConn(hc); }; } }
fn dataForHandshake(self: *Self, hc: *HandlerConn(H)) !bool { var conn_manager = &self.base.conn_manager; const compression, const ok = handleHandshake(H, self, hc, self.ctx); if (ok == false) { return false; }
if (hc.handler == null) { // still don't have a handshake conn_manager.inactive(hc); }
if (compression) { try conn_manager.setupCompression(hc); } if (hc.handler != null) try afterInit(H, hc, self.ctx); return true; } };}
fn NonBlockingBase(comptime H: type, comptime MANAGE_HS: bool) type { return struct { allocator: Allocator,
buffer_provider: *buffer.Provider,
// App can configure the use of a "small" buffer pool. This is the difference // between assigning a buffer to the connection just-in-time per message, or always small_buffer_pool: ?buffer.Pool, connection_buffer_size: usize,
conn_manager: ConnManager(H, MANAGE_HS),
const Self = @This();
fn init(allocator: Allocator, state: *WorkerState) !Self { const config = &state.config;
var conn_manager = try ConnManager(H, MANAGE_HS).init(allocator, state.io, config.compression); errdefer conn_manager.deinit();
const connection_buffer_size = config.buffers.small_size orelse DEFAULT_BUFFER_SIZE; var small_buffer_pool: ?buffer.Pool = null; if (config.buffers.small_pool) |pool_count| { small_buffer_pool = try buffer.Pool.init(allocator, pool_count, connection_buffer_size); }
errdefer if (small_buffer_pool) |sbp| { sbp.deinit(); };
return .{ .allocator = allocator, .conn_manager = conn_manager, .small_buffer_pool = small_buffer_pool, .buffer_provider = &state.buffer_provider, .connection_buffer_size = connection_buffer_size, }; }
fn deinit(self: *Self) void { self.conn_manager.deinit(); if (self.small_buffer_pool) |*sbp| { sbp.deinit(); } }
fn newConn(self: *Self, socket: posix.socket_t, address: Address, time: u32) !*HandlerConn(H) { return self.conn_manager.create(socket, address, time); }
pub fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) bool { return self._dataAvailable(hc, thread_buf) catch |err| { log.err("({f}) error processing client message: {}", .{ hc.conn.address, err }); return false; }; }
fn _dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) !bool { if (hc.reader == null) { const reader_buf = if (self.small_buffer_pool) |*sbp| try sbp.acquireOrCreate() else try self.allocator.alloc(u8, self.connection_buffer_size); hc.reader = Reader.init(reader_buf, self.buffer_provider, hc.compression); } const reader = &hc.reader.?;
var fba = FixedBufferAllocator.init(thread_buf); const ok = handleClientData(H, hc, self.allocator, &fba);
if (self.small_buffer_pool) |*sbp| { if (reader.isEmpty()) { sbp.release(reader.static); hc.reader = null; } }
return ok; }
fn cleanupConn(self: *Self, hc: *HandlerConn(H)) void { self.releaseConn(hc, true); }
fn discardConn(self: *Self, hc: *HandlerConn(H)) void { self.releaseConn(hc, false); }
fn releaseConn(self: *Self, hc: *HandlerConn(H), close_socket: bool) void { { hc.cleanup.lockUncancelable(hc.conn.io); defer hc.cleanup.unlock(hc.conn.io); if (hc.reader) |*reader| { if (self.small_buffer_pool) |*sbp| { sbp.release(reader.static); } else { self.allocator.free(reader.static); } hc.reader = null; } } // The blocking-mode counterpart closes the socket before destroying // the HandlerConn. The nonblocking path was missing it, leaking a // file descriptor per accepted connection. Symptom: CLOSE_WAIT / // CLOSED sockets accumulate on the server until ulimit exhaustion. if (close_socket) hc.conn.closeSocket(); self.conn_manager.cleanup(hc); }
fn shutdown(self: *Self) void { log.info("received shutdown signal", .{}); self.conn_manager.beginShutdown(); self.conn_manager.forceAll(self); }
// called for each hc when shutting down fn shutdownCleanup(self: *Self, hc: *HandlerConn(H)) bool { hc.cleanup.lockUncancelable(hc.conn.io); defer hc.cleanup.unlock(hc.conn.io); if (hc.reader) |*reader| { if (self.small_buffer_pool == null) { self.allocator.free(reader.static); } hc.reader = null; } return true; } };}
const Loop = switch (@import("builtin").os.tag) { .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KQueue, .linux => EPoll, else => unreachable,};
const KQueue = struct { q: i32, change_count: usize, change_buffer: [16]Kevent, event_list: [64]Kevent,
const Kevent = posix.Kevent;
// 0.16: posix.kevent removed, use libc.kevent directly fn keventSyscall(q: i32, changelist: []const Kevent, eventlist: []Kevent, timeout: ?*const libc.timespec) !usize { const rc = libc.kevent( q, changelist.ptr, @intCast(changelist.len), eventlist.ptr, @intCast(eventlist.len), timeout, ); if (rc == -1) return error.KqueueError; return @intCast(rc); }
fn init() !KQueue { // 0.16: posix.kqueue removed, use libc.kqueue const q = libc.kqueue(); if (q == -1) return error.KqueueError; return .{ .q = q, .change_count = 0, .change_buffer = undefined, .event_list = undefined, }; }
fn deinit(self: KQueue) void { _ = libc.close(self.q); }
fn monitorAccept(self: *KQueue, fd: c_int) !void { try self.change(fd, 0, posix.system.EVFILT.READ, posix.system.EV.ADD); }
fn monitorSignal(self: *KQueue, fd: c_int) !void { try self.change(fd, 1, posix.system.EVFILT.READ, posix.system.EV.ADD); }
// Normally, we add the socket in the worker thread with rearm == false. // Because this is a DISPATCH, it'll only fire once until we rearm it which // we do by re-enabling it with rearm == true. // However, notice that the rearm path also has the EV.ADD flag. From the above // description, this should not be necessary. // But monitorRead where rearm == true is also used by our generic ServerLoop when // taking over a connection (say from httpz). Hence, we need the EV.ADD flag too. fn monitorRead(self: *KQueue, hc: anytype, comptime rearm: bool) !void { if (rearm == false) { return self.change(hc.socket, @intFromPtr(hc), posix.system.EVFILT.READ, posix.system.EV.ADD | posix.system.EV.ENABLE | posix.system.EV.DISPATCH); } const event = Kevent{ .ident = @intCast(hc.socket), .filter = posix.system.EVFILT.READ, .flags = posix.system.EV.ADD | posix.system.EV.ENABLE | posix.system.EV.DISPATCH, .fflags = 0, .data = 0, .udata = @intFromPtr(hc), }; _ = try keventSyscall(self.q, &.{event}, &[_]Kevent{}, null); }
fn change(self: *KQueue, fd: posix.fd_t, data: usize, filter: i16, flags: u16) !void { var change_count = self.change_count; var change_buffer = &self.change_buffer;
if (change_count == change_buffer.len) { // calling this with an empty event_list will return immediate _ = try keventSyscall(self.q, change_buffer, &[_]Kevent{}, null); change_count = 0; } change_buffer[change_count] = .{ .ident = @intCast(fd), .filter = filter, .flags = flags, .fflags = 0, .data = 0, .udata = data, }; self.change_count = change_count + 1; }
fn wait(self: *KQueue, timeout_sec: ?i32) !Iterator { const event_list = &self.event_list; const timeout: ?libc.timespec = if (timeout_sec) |ts| libc.timespec{ .sec = ts, .nsec = 0 } else null; const event_count = try keventSyscall(self.q, self.change_buffer[0..self.change_count], event_list, if (timeout) |ts| &ts else null); self.change_count = 0;
return .{ .index = 0, .events = event_list[0..event_count], }; }
const Iterator = struct { index: usize, events: []Kevent,
fn next(self: *Iterator) ?usize { const index = self.index; const events = self.events; if (index == events.len) { return null; } self.index = index + 1; return self.events[index].udata; } };};
const EPoll = struct { q: i32, event_list: [64]EpollEvent,
const linux = std.os.linux; const EpollEvent = linux.epoll_event;
fn init() !EPoll { const q = linux.epoll_create1(0); const fd = std.math.cast(i32, q) orelse return error.EpollError; return .{ .event_list = undefined, .q = fd, }; }
fn deinit(self: EPoll) void { _ = libc.close(self.q); }
fn epollCtl(self: *EPoll, op: u32, fd: i32, event: *linux.epoll_event) !void { const rc = linux.epoll_ctl(self.q, op, fd, event); if (linux.errno(rc) != .SUCCESS) return error.EpollError; }
fn monitorAccept(self: *EPoll, fd: c_int) !void { var event = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .ptr = 0 } }; return self.epollCtl(linux.EPOLL.CTL_ADD, fd, &event); }
fn monitorSignal(self: *EPoll, fd: c_int) !void { var event = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .ptr = 1 } }; return self.epollCtl(linux.EPOLL.CTL_ADD, fd, &event); }
fn monitorRead(self: *EPoll, hc: anytype, comptime rearm: bool) !void { const op = if (rearm) linux.EPOLL.CTL_MOD else linux.EPOLL.CTL_ADD; var event = linux.epoll_event{ .events = linux.EPOLL.IN | linux.EPOLL.ONESHOT, .data = .{ .ptr = @intFromPtr(hc) } }; return self.epollCtl(op, hc.socket, &event); }
fn wait(self: *EPoll, timeout_sec: ?i32) !Iterator { const event_list = &self.event_list; var timeout: i32 = -1; if (timeout_sec) |sec| { if (sec > 2147483) { // max supported timeout by epoll_wait. timeout = 2147483647; } else { timeout = sec * 1000; } }
const rc = linux.epoll_wait(self.q, event_list, event_list.len, timeout); if (linux.errno(rc) != .SUCCESS) return error.EpollError; return .{ .index = 0, .events = event_list[0..rc], }; }
const Iterator = struct { index: usize, events: []EpollEvent,
fn next(self: *Iterator) ?usize { const index = self.index; const events = self.events; if (index == events.len) { return null; } self.index = index + 1; return self.events[index].data.ptr; } };};
// We want to extract as much common logic as possible from the Blocking and// NonBlockign workers. The code from this point on is meant to be used with// both workers, independently from the blocking/nonblocking nonsense.
// Abstraction ontop of NonBlocking and Blocking. Exists solely for integration// with httpz (or any other http library I guess.). Serves a similar purpose// as Server, but doesn't accept/listen.pub fn Worker(comptime H: type) type { return struct { worker: W,
const Self = @This(); const W = if (blockingMode()) Blocking(H) else NonBlockingBase(H, false);
pub fn init(allocator: Allocator, state: *WorkerState) !Self { return .{ .worker = try W.init(allocator, state), }; }
pub fn deinit(self: *Self) void { self.worker.deinit(); }
pub fn createConn(self: *Self, socket: posix.socket_t, address: Address, now: u32) !*HandlerConn(H) { return self.worker.conn_manager.create(socket, address, now); }
pub fn cleanupConn(self: *Self, hc: *HandlerConn(H)) void { self.worker.cleanupConn(hc); }
/// Destroy websocket state without closing its transport. HTTP server /// integrations use this when an upgrade fails before socket ownership /// transfers from the HTTP connection to websocket.zig. pub fn discardConn(self: *Self, hc: *HandlerConn(H)) void { self.worker.discardConn(hc); }
pub fn canCompress(self: *const Self) bool { return self.worker.conn_manager.compression != null; }
pub fn setupConnection( self: *Self, hc: *HandlerConn(H), ) !void { return self.worker.conn_manager.setupCompression(hc); }
pub fn shutdown(self: *Self) void { self.worker.shutdown(); } };}
// These are things that both the Blocking and NonBlocking workers need. Just cleaner// to have a single place for it. This could used to be directly in Server(H), with// Server(H) having handshake_pool and buffer_provider fields, but it was extracted// into its own struct for integration with webservers (i.e. httpz). The goal is// that a webserver can have websocket support without starting a full Server.pub const WorkerState = struct { config: Config, io: Io, handshake_pool: *Handshake.Pool, buffer_provider: buffer.Provider,
pub fn init(allocator: Allocator, io: Io, config: Config) !WorkerState { const handshake_pool_count = config.handshake.count orelse 32; const handshake_max_size = config.handshake.max_size orelse 1024; const handshake_max_headers = config.handshake.max_headers orelse 10; const handshake_max_res_headers = config.handshake.max_res_headers orelse 2;
var handshake_pool = try Handshake.Pool.init(allocator, handshake_pool_count, handshake_max_size, handshake_max_headers, handshake_max_res_headers); errdefer handshake_pool.deinit();
const max_message_size = config.max_message_size orelse DEFAULT_MAX_MESSAGE_SIZE; const large_buffer_pool = config.buffers.large_pool orelse 8; const large_buffer_size = config.buffers.large_size orelse @min((config.buffers.small_size orelse DEFAULT_BUFFER_SIZE) * 2, max_message_size);
var buffer_provider = try buffer.Provider.init(allocator, .{ .max = max_message_size, .size = large_buffer_size, .count = large_buffer_pool, }); errdefer buffer_provider.deinit();
return .{ .config = config, .io = io, .handshake_pool = handshake_pool, .buffer_provider = buffer_provider, }; }
pub fn deinit(self: *WorkerState) void { self.handshake_pool.deinit(); self.buffer_provider.deinit(); }};
// In the Blocking worker, all the state could be stored on the spawn'd threads// stack. The only reason we use a HandlerConn(H) in there is to be able to re-use// code with the NonBlocking worker.//// For the NonBlocking worker, HandlerConn(H) is critical as it contains all the// state for a connection. It lives on the heap, a pointer is registered into// the event loop, and passed back when data is ready to read.// * If handler is null, it means we haven't done our handshake yet.// * If handler is null AND handshake is null, it means we havent' received// any data yet (we delay creating the handshake state until we at least have// some data ready).pub fn HandlerConn(comptime H: type) type { return struct { state: State, conn: Conn, handler: ?H, reader: ?Reader, socket: posix.socket_t, // denormalization from conn.stream.socket.handle handshake: ?*Handshake.State, cleanup: Io.Mutex = .init, compression: ?Compression = null, shutdown_notified: bool = false, // Shutdown must still be able to count and interrupt this socket while // its notification task is blocked inside the handler's write path. upgraded: std.atomic.Value(bool) = .init(false), // A queued notification borrows this HandlerConn. Cleanup waits for // the borrow to end before returning the allocation to the pool. shutdown_notification_refs: std.atomic.Value(usize) = .init(0), next: ?*HandlerConn(H) = null, prev: ?*HandlerConn(H) = null,
const State = enum { handshake, active, }; };}
pub fn ConnManager(comptime H: type, comptime MANAGE_HS: bool) type { return struct { lock: Io.Mutex, io: Io, allocator: Allocator, active: List(HandlerConn(H)), pending: List(HandlerConn(H)), pool: std.heap.MemoryPool(HandlerConn(H)), compression: ?Compression, compression_pool: std.heap.MemoryPool(Conn.Compression),
const Self = @This();
pub fn init(allocator: Allocator, io: Io, compression: ?Compression) !Self { // 0.16: MemoryPool uses .empty and create/deinit take allocator var pool: std.heap.MemoryPool(HandlerConn(H)) = .empty; errdefer pool.deinit(allocator);
var compression_pool: std.heap.MemoryPool(Conn.Compression) = .empty; errdefer compression_pool.deinit(allocator);
return .{ .lock = .init, .io = io, .pool = pool, .active = .{}, .pending = .{}, .allocator = allocator, .compression = compression, .compression_pool = compression_pool, }; }
pub fn deinit(self: *Self) void { // 0.16: MemoryPool.deinit requires allocator self.pool.deinit(self.allocator); self.compression_pool.deinit(self.allocator); }
pub fn count(self: *Self) usize { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io); if (MANAGE_HS == false) { return self.active.len; } return self.active.len + self.pending.len; }
pub fn create(self: *Self, socket: posix.socket_t, address: Address, now: u32) !*HandlerConn(H) { errdefer _ = libc.close(socket);
self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
const hc = try self.pool.create(self.allocator); hc.* = .{ .state = if (MANAGE_HS) .handshake else .active, .socket = socket, .handler = null, .handshake = null, .reader = null, .compression = null, .shutdown_notified = false, .upgraded = .init(false), .shutdown_notification_refs = .init(0), .conn = .{ ._closed = false, .accept_compression = true, .started = now, .address = address, .io = self.io, .stream = .{ .socket = .{ .handle = socket, .address = .{ .ip4 = net.Ip4Address.loopback(0) } } }, .compression = null, }, };
if (comptime MANAGE_HS) { // Still waiting for a handshake. Only care about this with the full // NonBlocking worker self.pending.insert(hc); } else { self.active.insert(hc); } return hc; }
pub fn activate(self: *Self, hc: *HandlerConn(H)) void { std.debug.assert(MANAGE_HS == true);
// our caller made sute this was the case std.debug.assert(hc.state == .handshake); self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io); self.pending.remove(hc); self.active.insert(hc); hc.state = .active; }
pub fn inactive(self: *Self, hc: *HandlerConn(H)) void { std.debug.assert(MANAGE_HS == true);
// this should only be called when we need more data to complete the handshake // which should only happen on an active connection std.debug.assert(hc.state == .active); hc.state = .handshake;
self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io); self.active.remove(hc); self.pending.insert(hc); }
pub fn cleanup(self: *Self, hc: *HandlerConn(H)) void { // Shutdown walks the lists under the manager lock and then takes // this per-connection lock. Teardown must serialize with that walk, // but release this lock before taking the manager lock below to // preserve one lock ordering and avoid manager <-> connection // deadlocks. hc.cleanup.lockUncancelable(hc.conn.io); if (hc.handshake) |h| { h.release(); hc.handshake = null; }
if (hc.reader) |*r| { r.deinit(); hc.reader = null; }
if (hc.handler) |*h| { if (comptime std.meta.hasFn(H, "close")) { h.close(); } hc.handler = null; hc.upgraded.store(false, .release); }
if (hc.conn.compression) |c| { c.output.deinit(self.allocator); c.compressor.deinit(); } hc.cleanup.unlock(hc.conn.io);
self.lock.lockUncancelable(self.io); if (hc.state == .active) { self.active.remove(hc); } else { self.pending.remove(hc); }
if (hc.conn.compression) |c| { self.compression_pool.destroy(c); }
while (hc.shutdown_notification_refs.load(.acquire) != 0) self.io.sleep(.fromMilliseconds(1), .awake) catch {};
self.pool.destroy(hc); self.lock.unlock(self.io); }
fn setupCompression(self: *Self, hc: *HandlerConn(H)) !void { const config = hc.compression orelse return;
if (config.write_threshold == null) { // if write_threshold is null, then we never want to compress // outgoing messages. We don't need to set the conn.compression // field. // We'll still [potentially] decompress incoming messages, but // that's set on the proto. return; }
const compression = try self.compression_pool.create(self.allocator); errdefer self.compression_pool.destroy(compression);
compression.* = .{ .allocator = self.allocator, .write_threshold = config.write_threshold.?, .retain_writer = config.retain_write_buffer, .output = .empty, .server_no_context_takeover = config.server_no_context_takeover, .compressor = undefined, }; try compression.compressor.init(); hc.conn.compression = compression; }
pub fn websocketCount(self: *Self) usize { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
return websocketCountList(self.active.head) + websocketCountList(self.pending.head); }
fn websocketCountList(head: ?*HandlerConn(H)) usize { var total: usize = 0; var next_node = head; while (next_node) |hc| : (next_node = hc.next) { if (hc.upgraded.load(.acquire)) total += 1; } return total; }
pub fn beginShutdownConcurrent(self: *Self, notifications: *Io.Group) void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
notifyListConcurrent(self.active.head, notifications, self.io); notifyListConcurrent(self.pending.head, notifications, self.io); }
fn notifyListConcurrent(head: ?*HandlerConn(H), notifications: *Io.Group, io: Io) void { var next_node = head; while (next_node) |hc| : (next_node = hc.next) { hc.cleanup.lockUncancelable(hc.conn.io); const should_notify = hc.handler != null and !hc.shutdown_notified; if (should_notify) { hc.shutdown_notified = true; _ = hc.shutdown_notification_refs.fetchAdd(1, .release); } hc.cleanup.unlock(hc.conn.io);
if (!should_notify) continue; notifications.concurrent(io, notifyOne, .{hc}) catch { hc.cleanup.lockUncancelable(hc.conn.io); hc.shutdown_notified = false; hc.cleanup.unlock(hc.conn.io); _ = hc.shutdown_notification_refs.fetchSub(1, .release); }; } }
fn notifyOne(hc: *HandlerConn(H)) void { hc.cleanup.lockUncancelable(hc.conn.io); if (hc.handler) |*handler| { if (comptime std.meta.hasFn(H, "serverClose")) handler.serverClose(); } hc.cleanup.unlock(hc.conn.io); _ = hc.shutdown_notification_refs.fetchSub(1, .release); }
pub fn beginShutdown(self: *Self) void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
notifyList(self.active.head); notifyList(self.pending.head); }
fn notifyList(head: ?*HandlerConn(H)) void { var next_node = head; while (next_node) |hc| : (next_node = hc.next) { hc.cleanup.lockUncancelable(hc.conn.io); if (hc.handler) |*handler| { if (!hc.shutdown_notified) { if (comptime std.meta.hasFn(H, "serverClose")) handler.serverClose(); hc.shutdown_notified = true; } } hc.cleanup.unlock(hc.conn.io); } }
pub fn forceWebSockets(self: *Self, worker: anytype) void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
forceList(self.active.head, worker, true); forceList(self.pending.head, worker, true); }
pub fn interruptWebSockets(self: *Self, worker: anytype) void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
interruptList(self.active.head, worker); interruptList(self.pending.head, worker); }
fn interruptList(head: ?*HandlerConn(H), worker: anytype) void { var next_node = head; while (next_node) |hc| : (next_node = hc.next) { if (hc.upgraded.load(.acquire)) worker.interruptShutdown(hc); } }
pub fn forceAll(self: *Self, worker: anytype) void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
forceList(self.active.head, worker, false); forceList(self.pending.head, worker, false); }
// This is sloppy and leaves things in an unrecoverable state. To keep // things clean, we should call self.cleanup(hc) on each entry in the list // but that does a bunch of things we don't need if we know that we're // shutting down - like returning data to the pools, and popping items // out of the list. fn forceList(head: ?*HandlerConn(H), worker: anytype, websocket_only: bool) void { var next_node = head; while (next_node) |hc| { hc.cleanup.lockUncancelable(hc.conn.io); if (websocket_only and hc.handler == null) { hc.cleanup.unlock(hc.conn.io); next_node = hc.next; continue; } if (comptime std.meta.hasFn(H, "close")) { if (hc.handler) |*h| { if (!hc.shutdown_notified and comptime std.meta.hasFn(H, "serverClose")) { h.serverClose(); } h.close(); hc.handler = null; hc.upgraded.store(false, .release); } } hc.cleanup.unlock(hc.conn.io);
if (worker.shutdownCleanup(hc)) hc.conn.closeSocket(); next_node = hc.next; } } };}
// This is what actually gets exposed to the apppub const Conn = struct { _closed: bool, started: u32, stream: net.Stream, address: Address, io: Io, lock: Io.Mutex = .init, transport_lock: Io.Mutex = .init, compression: ?*Conn.Compression = null, accept_compression: bool = true,
const Compression = struct { allocator: Allocator, retain_writer: bool, write_threshold: usize, output: std.ArrayList(u8), server_no_context_takeover: bool, compressor: deflate.Compressor, };
/// Decline an offered extension for this request before the handshake /// response is written. Servers with endpoint-specific compression use /// this from Handler.init. pub fn disableCompression(self: *Conn) void { self.accept_compression = false; }
pub fn isClosed(self: *Conn) bool { // don't use lock to protect _closed. `isClosed` is called from // the worker thread and we don't want that potentially blocked while // a write is going on. return @atomicLoad(bool, &self._closed, .monotonic); }
/// Return a stable, port-free peer IP suitable for per-source accounting. /// The returned slice borrows `buffer`; 64 bytes accommodates every IPv6 /// representation. Unix-domain and unknown peers return literal keys. pub fn peerIp(self: *const Conn, out_buffer: []u8) std.Io.Writer.Error![]const u8 { var writer: std.Io.Writer = .fixed(out_buffer); try self.address.formatIp(&writer); return writer.buffered(); }
pub fn writeBin(self: *Conn, data: []const u8) !void { return self.writeFrame(.binary, data); }
pub fn writeText(self: *Conn, data: []const u8) !void { return self.writeFrame(.text, data); }
pub fn write(self: *Conn, data: []const u8) !void { return self.writeFrame(.text, data); }
pub fn writePing(self: *Conn, data: []u8) !void { return self.writeFrame(.ping, data); }
pub fn writePong(self: *Conn, data: []u8) !void { return self.writeFrame(.pong, data); }
/// Bound every subsequent socket write. A zero duration restores the /// platform default (no send timeout). pub fn writeTimeout(self: *const Conn, ms: u32) !void { const timeout = std.mem.toBytes(posix.timeval{ .sec = @intCast(@divTrunc(ms, 1000)), .usec = @intCast(@mod(ms, 1000) * 1000), }); try posix.setsockopt(self.stream.socket.handle, posix.SOL.SOCKET, posix.SO.SNDTIMEO, &timeout); }
/// Wake the server's blocked reader without closing its descriptor out /// from underneath it. The read loop owns final socket cleanup. pub fn interruptRead(self: *Conn) void { self.shutdownSocket(.recv); }
const CloseOpts = struct { code: u16 = 1000, reason: []const u8 = "", };
pub fn close(self: *Conn, opts: CloseOpts) !void { if (self.isClosed()) { return; } defer self.closeSocket();
try self.writeClose(opts); }
/// Send a close frame without closing the descriptor. Server shutdown /// uses this before half-closing the read side, so the blocked reader can /// unwind through its ordinary cleanup instead of racing a raw close. pub fn writeClose(self: *Conn, opts: CloseOpts) !void { if (self.isClosed()) return;
const reason = opts.reason; if (reason.len == 0) { var buf: [2]u8 = undefined; std.mem.writeInt(u16, &buf, opts.code, .big); return self.writeFrame(.close, &buf); }
if (reason.len > 123) { return error.ReasonTooLong; }
var buf: [4]u8 = undefined; buf[0] = @backingInt(OpCode.close); buf[1] = @intCast(reason.len + 2); std.mem.writeInt(u16, buf[2..], opts.code, .big);
var vec = [2]std.posix.iovec_const{ .{ .len = buf.len, .base = &buf }, .{ .len = reason.len, .base = reason.ptr }, };
try writeAllIOVec(self, &vec); }
pub fn writeFrame(self: *Conn, op_code: OpCode, data: []const u8) !void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io);
var payload = data; var compressed = false; if (op_code == .text or op_code == .binary) if (self.compression) |c| { if (data.len >= c.write_threshold) { compressed = true; payload = try c.compressor.compress(c.allocator, data, &c.output); } }; defer if (compressed) resetCompressionOutput(self.compression.?);
// maximum possible prefix length. op_code + length_type + 8byte length var buf: [10]u8 = undefined; const header = proto.writeFrameHeader(&buf, op_code, payload.len, compressed);
const stream = self.stream;
if (payload.len == 0) { // no body, just write the header return socketWriteAll(self.io, stream.socket.handle, header); }
var vec = [2]std.posix.iovec_const{ .{ .len = header.len, .base = header.ptr }, .{ .len = payload.len, .base = payload.ptr }, };
return self.writeAllIOVecUnlocked(&vec); }
pub fn writeFramed(self: *Conn, data: []const u8) !void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io); try socketWriteAll(self.io, self.stream.socket.handle, data); }
fn writeAllIOVec(self: *Conn, vec: []std.posix.iovec_const) !void { self.lock.lockUncancelable(self.io); defer self.lock.unlock(self.io); return self.writeAllIOVecUnlocked(vec); }
fn writeAllIOVecUnlocked(self: *Conn, vec: []std.posix.iovec_const) !void { const socket = self.stream.socket.handle; const io = self.io;
// write each iovec segment via Io vtable const empty = [_][]const u8{""}; for (vec) |*v| { var remaining = v.base[0..v.len]; while (remaining.len > 0) { const n = io.vtable.netWrite(io.userdata, socket, remaining, &empty, 0) catch |err| { return switch (err) { error.ConnectionResetByPeer => error.ConnectionResetByPeer, else => error.Unexpected, }; }; if (n == 0) return error.Unexpected; remaining = remaining[n..]; } } }
fn resetCompressionOutput(c: *Conn.Compression) void { if (c.retain_writer) { c.output.clearRetainingCapacity(); } else { c.output.clearAndFree(c.allocator); } if (c.server_no_context_takeover) { c.compressor.reset() catch unreachable; } }
pub fn writeBuffer(self: *Conn, allocator: Allocator, op_code: OpCode) Writer { return .{ .conn = self, .buf = .empty, .op_code = op_code, .allocator = allocator, .interface = .{ .vtable = &.{ .drain = Writer.drain }, .buffer = &.{}, }, }; }
fn closeSocket(self: *Conn) void { self.transport_lock.lockUncancelable(self.io); defer self.transport_lock.unlock(self.io);
if (@atomicRmw(bool, &self._closed, .Xchg, true, .monotonic) == false) { self.io.vtable.netClose(self.io.userdata, (&self.stream.socket.handle)[0..1]); } }
fn shutdownSocket(self: *Conn, how: net.ShutdownHow) void { self.transport_lock.lockUncancelable(self.io); defer self.transport_lock.unlock(self.io);
if (!self.isClosed()) { self.io.vtable.netShutdown(self.io.userdata, self.stream.socket.handle, how) catch {}; } }
pub const Writer = struct { conn: *Conn, op_code: OpCode, allocator: Allocator, buf: std.ArrayList(u8), interface: std.Io.Writer,
pub const Error = Allocator.Error;
pub fn deinit(self: *Writer) void { self.buf.deinit(self.allocator); }
pub fn drain(io_w: *std.Io.Writer, data: []const []const u8, splat: usize) error{WriteFailed}!usize { _ = splat; const self: *Writer = @alignCast(@fieldParentPtr("interface", io_w)); self.buf.appendSlice(self.allocator, data[0]) catch return error.WriteFailed; return data[0].len; }
pub fn send(self: *Writer) !void { return self.conn.writeFrame(self.op_code, self.buf.items) catch error.WriteFailed; } };};
fn handleHandshake(comptime H: type, worker: anytype, hc: *HandlerConn(H), ctx: anytype) struct { bool, bool } { return _handleHandshake(H, worker, hc, ctx) catch |err| { log.warn("({f}) uncaugh error processing handshake: {}", .{ hc.conn.address, err }); return .{ false, false }; };}
fn _handleHandshake(comptime H: type, worker: anytype, hc: *HandlerConn(H), ctx: anytype) !struct { bool, bool } { std.debug.assert(hc.handler == null);
var state = hc.handshake orelse blk: { const s = try worker.handshake_pool.acquire(); hc.handshake = s; break :blk s; };
var buf = state.buf; var conn = &hc.conn; const len = state.len;
if (len == buf.len) { log.warn("({f}) handshake request exceeded maximum configured size ({d})", .{ conn.address, buf.len }); return .{ false, false }; }
const n = (SocketReader{ .socket = hc.socket }).read(buf[len..]) catch |err| { switch (err) { error.ConnectionResetByPeer => log.debug("({f}) handshake connection closed: {}", .{ conn.address, err }), error.WouldBlock => { std.debug.assert(blockingMode()); log.debug("({f}) handshake timeout", .{conn.address}); }, else => log.warn("({f}) handshake error reading from socket: {}", .{ conn.address, err }), } return .{ false, false }; };
if (n == 0) { log.debug("({f}) handshake connection closed", .{conn.address}); return .{ false, false }; }
state.len = len + n; var handshake = Handshake.parse(state) catch |err| { // These errors mean "valid HTTP request, but not a websocket upgrade": // - MissingHeaders: no websocket-specific headers (plain GET/POST) // - InvalidConnection: Connection header present without "upgrade" (e.g. "keep-alive") // - InvalidUpgrade: Upgrade header present but not "websocket" if (comptime std.meta.hasFn(H, "httpFallback")) { switch (err) { error.MissingHeaders, error.InvalidConnection, error.InvalidUpgrade => { if (dispatchHttpFallback(H, state, conn, ctx)) { return .{ false, false }; } }, else => {}, } } log.debug("({f}) error parsing handshake: {}", .{ conn.address, err }); respondToHandshakeError(conn, err); return .{ false, false }; } orelse { // we need more data return .{ false, true }; };
defer state.release(); hc.handshake = null;
// After this, the app has access to &hc.conn, so any access to the // conn has to be synchronized (which the conn does internally).
const handler = H.init(&handshake, conn, ctx) catch |err| { if (comptime std.meta.hasFn(H, "handshakeErrorResponse")) { preHandOffWrite(H.handshakeErrorResponse(err)); } else { respondToHandshakeError(conn, err); } log.debug("({f}) " ++ @typeName(H) ++ ".init rejected request {}", .{ conn.address, err }); return .{ false, false }; };
hc.handler = handler; hc.upgraded.store(true, .release);
var negotiated: ?Handshake.Compression = null; if (conn.accept_compression) if (handshake.compression) |offer| if (worker.compression) |configured| { var effective = configured; effective.client_no_context_takeover = configured.client_no_context_takeover or offer.client_no_context_takeover; effective.server_no_context_takeover = configured.server_no_context_takeover or offer.server_no_context_takeover; hc.compression = effective; negotiated = .{ .client_no_context_takeover = effective.client_no_context_takeover, .server_no_context_takeover = effective.server_no_context_takeover, }; }; const compression = negotiated != null;
var reply_buf: [2048]u8 = undefined; const handshake_reply = try Handshake.createReplyNegotiated(handshake.key, handshake.res_headers, negotiated, &reply_buf); try conn.writeFramed(handshake_reply);
log.debug("({f}) connection successfully upgraded", .{conn.address}); return .{ compression, true };}
fn afterInit(comptime H: type, hc: *HandlerConn(H), ctx: anytype) !void { if (comptime std.meta.hasFn(H, "afterInit")) { const params = @typeInfo(@TypeOf(H.afterInit)).@"fn".param_types; const result = if (params.len == 1) hc.handler.?.afterInit() else hc.handler.?.afterInit(ctx); result catch |err| { log.debug("({f}) " ++ @typeName(H) ++ ".afterInit error: {}", .{ hc.conn.address, err }); return err; }; }}
fn handleClientData(comptime H: type, hc: *HandlerConn(H), allocator: Allocator, fba: ?*FixedBufferAllocator) bool { std.debug.assert(hc.handshake == null); return _handleClientData(H, hc, allocator, fba) catch |err| { log.warn("({f}) uncaugh error handling incoming data: {}", .{ hc.conn.address, err }); return false; };}
fn _handleClientData(comptime H: type, hc: *HandlerConn(H), allocator: Allocator, fba: ?*FixedBufferAllocator) !bool { var conn = &hc.conn; var reader = &hc.reader.?; reader.fill(SocketReader{ .socket = conn.stream.socket.handle }) catch |err| { switch (err) { error.Closed, error.ConnectionResetByPeer => log.debug("({f}) connection closed: {}", .{ conn.address, err }), else => log.warn("({f}) error reading from connection: {}", .{ conn.address, err }), } return false; };
const handler = &hc.handler.?; while (true) { const has_more, const message = reader.read() catch |err| { if (comptime @hasDecl(H, "clientError")) handler.clientError(err); switch (err) { error.LargeControl => conn.writeFramed(CLOSE_PROTOCOL_ERROR) catch {}, error.ReservedFlags => conn.writeFramed(CLOSE_PROTOCOL_ERROR) catch {}, error.CompressionDisabled => conn.writeFramed(CLOSE_PROTOCOL_ERROR) catch {}, error.CompressionError => conn.writeFramed(CLOSE_PROTOCOL_ERROR) catch {}, else => {}, } log.debug("({f}) invalid websocket packet: {}", .{ conn.address, err }); return false; } orelse { // everything is fine, we just need more data return true; };
const message_type = message.type; defer reader.done(message_type);
log.debug("({f}) received {s} message", .{ hc.conn.address, @tagName(message_type) }); switch (message_type) { .text, .binary => { const params = @typeInfo(@TypeOf(H.clientMessage)).@"fn".param_types; const needs_allocator = comptime needsAllocator(H);
var arena: std.heap.ArenaAllocator = undefined; var fallback_allocator: FallbackAllocator = undefined; var aa: Allocator = undefined;
if (comptime needs_allocator) { arena = std.heap.ArenaAllocator.init(allocator); // the per-thread fba only exists on the worker path; // blocking-style read loops (blockingMode, runIo) pass // null and get the arena directly if (fba) |f| { fallback_allocator = FallbackAllocator{ .fba = f, .fallback = arena.allocator(), .fixed = f.allocator(), }; aa = fallback_allocator.allocator(); } else { aa = arena.allocator(); } }
defer if (comptime needs_allocator) { arena.deinit(); };
switch (comptime params.len) { 2 => handler.clientMessage(message.data) catch return false, 3 => if (needs_allocator) { handler.clientMessage(aa, message.data) catch return false; } else { handler.clientMessage(message.data, if (message_type == .text) .text else .binary) catch return false; }, 4 => handler.clientMessage(aa, message.data, if (message_type == .text) .text else .binary) catch return false, else => @compileError(@typeName(H) ++ ".clientMessage has invalid parameter count"), } }, .pong => if (comptime std.meta.hasFn(H, "clientPong")) { try handler.clientPong(message.data); }, .ping => { const data = message.data; if (comptime std.meta.hasFn(H, "clientPing")) { try handler.clientPing(data); } else if (data.len == 0) { try hc.conn.writeFramed(EMPTY_PONG); } else { try hc.conn.writeFrame(.pong, data); } }, .close => { const data = message.data; if (comptime std.meta.hasFn(H, "clientClose")) { try handler.clientClose(data); return false; }
const l = data.len; if (l == 0) { try conn.close(.{}); return false; }
if (l == 1) { // close with a payload always has to have at least a 2-byte payload, // since a 2-byte code is required try conn.writeFramed(CLOSE_PROTOCOL_ERROR); return false; }
const code = @as(u16, @intCast(data[1])) | (@as(u16, @intCast(data[0])) << 8); if (code < 1000 or code == 1004 or code == 1005 or code == 1006 or (code > 1013 and code < 3000)) { try conn.writeFramed(CLOSE_PROTOCOL_ERROR); return false; }
if (l == 2) { try conn.writeFramed(CLOSE_NORMAL); return false; }
const payload = data[2..]; if (!std.unicode.utf8ValidateSlice(payload)) { // if we have a payload, it must be UTF8 (why?!) try conn.writeFramed(CLOSE_PROTOCOL_ERROR); } else { try conn.close(.{}); } return false; }, }
if (conn.isClosed()) { return false; }
if (has_more == false) { // we don't have more data ready to be processed in our buffer // back to our caller for more data return true; } }}
fn needsAllocator(comptime H: type) bool { const params = @typeInfo(@TypeOf(H.clientMessage)).@"fn".param_types; return comptime params[1] == Allocator;}
/// Result of parsing a plain HTTP request from a raw buffer.pub const HttpRequest = struct { method: []const u8, url: []const u8, body: []const u8,};
/// Parse a complete HTTP/1.1 request from a raw buffer, populating headers./// Returns null if the buffer doesn't contain a parseable HTTP request./// Lowercases all header names in-place for consistency with Handshake.parse.pub fn parseHttpRequest(buf: []u8, len: usize, headers: *Handshake.KeyValue) ?HttpRequest { const request = buf[0..len];
// Find header/body separator. This is the canonical end of headers. // For GET requests, \r\n\r\n is at the end. For POST, the body follows it. const header_end = std.mem.indexOf(u8, request, "\r\n\r\n") orelse return null;
// Parse request line: "METHOD /url HTTP/1.1\r\n..." const request_line_end = std.mem.indexOfScalar(u8, request, '\r') orelse return null; const request_line = request[0..request_line_end];
const method_end = std.mem.indexOfScalar(u8, request_line, ' ') orelse return null; const method = request_line[0..method_end];
// URL is between first space and last space (protocol) const rest = request_line[method_end + 1 ..]; const proto_start = std.mem.lastIndexOfScalar(u8, rest, ' ') orelse return null; const url = rest[0..proto_start];
// Reset and re-parse all headers from scratch. // Handshake.parse may have partially populated headers before returning an error, // and may have lowercased only some header names. Re-parsing gives us a clean, // complete, consistently-lowercased header set. headers.len = 0; // Only parse up to header_end (don't scan into the body) var hdr_buf = request[request_line_end + 2 .. header_end + 2]; while (hdr_buf.len > 2) { if (hdr_buf[0] == '\r' and hdr_buf[1] == '\n') break; const line_end = std.mem.indexOfScalar(u8, hdr_buf, '\r') orelse break; const separator = std.mem.indexOfScalar(u8, hdr_buf[0..line_end], ':') orelse { hdr_buf = hdr_buf[line_end + 2 ..]; continue; }; // Lowercase header name in-place (idempotent for already-lowered names) for (hdr_buf[0..separator]) |*c| { c.* = std.ascii.toLower(c.*); } const name = std.mem.trim(u8, hdr_buf[0..separator], &std.ascii.whitespace); const value = std.mem.trim(u8, hdr_buf[separator + 1 .. line_end], &std.ascii.whitespace); headers.add(name, value); hdr_buf = hdr_buf[line_end + 2 ..]; }
return .{ .method = method, .url = url, .body = request[header_end + 4 ..], };}
/// Dispatch a non-upgrade HTTP request to H.httpFallback./// Returns true if fallback was successfully dispatched.fn dispatchHttpFallback(comptime H: type, state: *Handshake.State, conn: *Conn, ctx: anytype) bool { const http = parseHttpRequest(state.buf, state.len, &state.req_headers) orelse return false; H.httpFallback(conn, http.method, http.url, http.body, &state.req_headers, ctx); return true;}
fn respondToHandshakeError(conn: *Conn, err: anyerror) void { const response = switch (err) { error.Close => return, error.RequestTooLarge => buildError(400, "too large"), error.Timeout, error.WouldBlock => buildError(400, "timeout"), error.InvalidProtocol => buildError(400, "invalid protocol"), error.InvalidRequestLine => buildError(400, "invalid requestline"), error.InvalidHeader => buildError(400, "invalid header"), error.InvalidUpgrade => buildError(400, "invalid upgrade"), error.InvalidVersion => buildError(400, "invalid version"), error.InvalidConnection => buildError(400, "invalid connection"), error.MissingHeaders => buildError(400, "missingheaders"), error.Empty => buildError(400, "invalid request"), error.WhitespaceBeforeColon => buildError(400, "whitespace before colon"), error.AmbiguousBodyLength => buildError(400, "ambiguous body length"), else => buildError(400, "unknown"), }; preHandOffWrite(conn, response);}
fn buildError(comptime status: u16, comptime err: []const u8) []const u8 { return std.fmt.comptimePrint("HTTP/1.1 {d} \r\nConnection: Close\r\nError: {s}\r\nContent-Length: 0\r\n\r\n", .{ status, err });}
fn preHandOffWrite(conn: *Conn, response: []const u8) void { // "preHandOff" means we haven't given the application handler a reference // to *Conn yet. In theory, this means we don't need to worry about thread-safety // However, it is possible for the worker to be stopped while we're doing this // which causes issues unless we lock conn.lock.lockUncancelable(conn.io); defer conn.lock.unlock(conn.io);
if (conn.isClosed()) { return; }
const socket = conn.stream.socket.handle; const timeout = std.mem.toBytes(posix.timeval{ .sec = 5, .usec = 0 }); setSockOptBestEffort(socket, posix.SOL.SOCKET, posix.SO.SNDTIMEO, &timeout);
socketWriteAll(conn.io, socket, response) catch return;}
// best-effort setsockopt for connection sockets. std.posix.setsockopt maps// EBADF/ENOTSOCK to `unreachable` ("always a race condition"), so a socket the// peer reset or another thread closed mid-flight panics the worker rather than// returning an error `catch` could swallow. timeouts / TCP_NODELAY are// non-essential, so issue the raw syscall and ignore the result.fn setSockOptBestEffort(fd: posix.socket_t, level: i32, optname: u32, opt: []const u8) void { _ = posix.system.setsockopt(fd, level, optname, opt.ptr, @intCast(opt.len));}
fn timestamp() u32 { const io = std.Options.debug_io; const ts = Io.Timestamp.now(io, .real); return @intCast(@divTrunc(ts.nanoseconds, std.time.ns_per_s));}
// intrusive doubly-linked list with count, not thread safefn List(comptime T: type) type { return struct { len: usize = 0, head: ?*T = null, tail: ?*T = null,
const Self = @This();
pub fn insert(self: *Self, node: *T) void { if (self.tail) |tail| { tail.next = node; node.prev = tail; self.tail = node; } else { self.head = node; self.tail = node; } self.len += 1; node.next = null; }
pub fn remove(self: *Self, node: *T) void { if (node.prev) |prev| { prev.next = node.next; } else { self.head = node.next; }
if (node.next) |next| { next.prev = node.prev; } else { self.tail = node.prev; } node.prev = null; node.next = null; self.len -= 1; } };}
const t = @import("../t.zig");
var test_thread: Thread = undefined;var test_server: Server(TestHandler) = undefined;var global_test_allocator: std.heap.DebugAllocator(.{}) = .init;
test "tests:beforeAll" { test_server = try Server(TestHandler).init(global_test_allocator.allocator(), std.Options.debug_io, .{ .port = 9292, .address = "127.0.0.1", }); test_thread = try test_server.listenInNewThread({});}
test "tests:afterAll" { test_server.stop(); test_thread.join(); test_server.deinit(); // 0.16: detectLeaks returns usize (count of leaks), not bool try t.expectEqual(@as(usize, 0), global_test_allocator.detectLeaks());}
test "Server: runIo supports allocator-taking clientMessage" { // debug_io can't spawn concurrent tasks; runIo needs a real backend var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var server = try Server(TestHandler).init(t.allocator, io, .{ .port = 9293, .address = "127.0.0.1", }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9293) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); // concrete wrapper: io.concurrent needs ArgsTuple, which can't handle // runIo's anytype ctx const Run = struct { fn go(srv: *Server(TestHandler), l: *net.Server) void { srv.runIo(l, {}); } }; var future = try io.concurrent(Run.go, .{ &server, &listener });
var stream = try testStreamPort(true, 9293); defer stream.close();
// "dyn" makes TestHandler reply via the message allocator — the path // that dereferenced an undefined FixedBufferAllocator before the fix try stream.writeAll(&proto.frame(.text, "dyn")); var buf: [12]u8 = undefined; _ = try stream.readAtLeast(&buf, 12); try t.expectSlice(u8, &.{ 129, 10, 'o', 'v', 'e', 'r', ' ', '9', '0', '0', '0', '!' }, buf[0..12]);
// cancel first: accept handles error.Canceled; closing the listener out // from under a blocked accept panics (BADF) on Io.Threaded/macos _ = future.cancel(io); listener.deinit(io);}
test "Server: parser errors are reported to the handler" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var observed: std.atomic.Value(u8) = .init(0); var server = try Server(ErrorHandler).init(t.allocator, io, .{ .port = 9294, .address = "127.0.0.1", .max_message_size = 4096, }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9294) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(ErrorHandler), l: *net.Server, result: *std.atomic.Value(u8)) void { srv.runIo(l, result); } }; var future = try io.concurrent(Run.go, .{ &server, &listener, &observed });
var stream = try testStreamPort(true, 9294); defer stream.close(); const oversized_payload: [4097]u8 = @splat('x'); try stream.writeAll(&proto.frame(.text, &oversized_payload));
var attempts: usize = 0; while (observed.load(.acquire) == 0 and attempts < 1000) : (attempts += 1) { _ = try io.sleep(.fromMilliseconds(1), .awake); } try t.expectEqual(@as(u8, 1), observed.load(.acquire));
_ = future.cancel(io); listener.deinit(io);}
test "Server: exposes write deadlines and intentional shutdown" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var lifecycle: std.atomic.Value(u8) = .init(0); var server = try Server(LifecycleHandler).init(t.allocator, io, .{ .port = 9295, .address = "127.0.0.1", .websocket_shutdown_grace = .fromSeconds(5), .http_shutdown_grace = .fromSeconds(5), }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9295) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(LifecycleHandler), l: *net.Server, state: *std.atomic.Value(u8)) void { srv.runIo(l, state); } }; var future = try io.concurrent(Run.go, .{ &server, &listener, &lifecycle });
var stream = try testStreamPort(true, 9295); defer stream.close(); var attempts: usize = 0; while (lifecycle.load(.acquire) < 1 and attempts < 1000) : (attempts += 1) _ = try io.sleep(.fromMilliseconds(1), .awake); try t.expectEqual(@as(u8, 1), lifecycle.load(.acquire));
var close_frame: [24]u8 = undefined; const Stop = struct { fn go(server_future: *Io.Future(void), task_io: Io) void { _ = server_future.cancel(task_io); } }; const started = Io.Timestamp.now(io, .awake).toNanoseconds(); var stopping = try io.concurrent(Stop.go, .{ &future, io }); _ = try stream.readAtLeast(&close_frame, close_frame.len); try t.expectSlice(u8, &.{ 0x88, 22, 0x03, 0xe9 }, close_frame[0..4]); try t.expectString("server shutting down", close_frame[4..]); try stream.writeAll(&proto.frame(.close, "\x03\xe9")); stopping.await(io); listener.deinit(io); const elapsed = Io.Timestamp.now(io, .awake).toNanoseconds() - started; try std.testing.expect(elapsed < std.time.ns_per_s); try t.expectEqual(@as(u8, 2), lifecycle.load(.acquire));}
test "Server: websocket shutdown grace forces a silent peer at its own deadline" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var lifecycle: std.atomic.Value(u8) = .init(0); var server = try Server(LifecycleHandler).init(t.allocator, io, .{ .port = 9296, .address = "127.0.0.1", .websocket_shutdown_grace = .fromMilliseconds(80), .http_shutdown_grace = .fromMilliseconds(500), }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9296) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(LifecycleHandler), l: *net.Server, state: *std.atomic.Value(u8)) void { srv.runIo(l, state); } }; var future = try io.concurrent(Run.go, .{ &server, &listener, &lifecycle });
var stream = try testStreamPort(true, 9296); defer stream.close(); var attempts: usize = 0; while (lifecycle.load(.acquire) < 1 and attempts < 1000) : (attempts += 1) _ = try io.sleep(.fromMilliseconds(1), .awake); try t.expectEqual(@as(u8, 1), lifecycle.load(.acquire));
const started = Io.Timestamp.now(io, .awake).toNanoseconds(); _ = future.cancel(io); listener.deinit(io); const elapsed = Io.Timestamp.now(io, .awake).toNanoseconds() - started; try std.testing.expect(elapsed >= 60 * std.time.ns_per_ms); try std.testing.expect(elapsed < 400 * std.time.ns_per_ms); try t.expectEqual(@as(u8, 2), lifecycle.load(.acquire));}
test "Server: shutdown fan-out is concurrent when an earlier writer is wedged" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var context: FanoutShutdownContext = .{}; var server = try Server(FanoutShutdownHandler).init(t.allocator, io, .{ .port = 9299, .address = "127.0.0.1", .websocket_shutdown_grace = .fromMilliseconds(150), .http_shutdown_grace = .fromMilliseconds(150), }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9299) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(FanoutShutdownHandler), l: *net.Server, ctx: *FanoutShutdownContext) void { srv.runIo(l, ctx); } }; var future = try io.concurrent(Run.go, .{ &server, &listener, &context });
// List traversal is connection order. The blocked writer comes first so // this test would time out under the old serial notification loop. var blocked = try testStreamPortPath(9299, "/blocked"); defer blocked.close(); var healthy = try testStreamPortPath(9299, "/healthy"); defer healthy.close(); const receive_timeout = std.mem.toBytes(posix.timeval{ .sec = 1, .usec = 0 }); try posix.setsockopt(healthy.socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &receive_timeout);
var attempts: usize = 0; while ((!context.blocked_ready.load(.acquire) or !context.healthy_ready.load(.acquire)) and attempts < 1000) : (attempts += 1) _ = try io.sleep(.fromMilliseconds(1), .awake); try std.testing.expect(context.blocked_ready.load(.acquire)); try std.testing.expect(context.healthy_ready.load(.acquire));
const Stop = struct { fn go(server_future: *Io.Future(void), task_io: Io) void { _ = server_future.cancel(task_io); } }; const started = Io.Timestamp.now(io, .awake).toNanoseconds(); var stopping = try io.concurrent(Stop.go, .{ &future, io });
var close_frame: [4]u8 = undefined; _ = try healthy.readAtLeast(&close_frame, close_frame.len); const healthy_elapsed = Io.Timestamp.now(io, .awake).toNanoseconds() - started; try t.expectSlice(u8, &.{ 0x88, 0x02, 0x03, 0xe9 }, &close_frame); try std.testing.expect(healthy_elapsed < 100 * std.time.ns_per_ms); try healthy.writeAll(&proto.frame(.close, "\x03\xe9"));
stopping.await(io); listener.deinit(io); const total_elapsed = Io.Timestamp.now(io, .awake).toNanoseconds() - started; try std.testing.expect(context.blocked_close_started.load(.acquire)); try std.testing.expect(context.blocked_close_unblocked.load(.acquire)); try std.testing.expect(context.healthy_notified.load(.acquire)); try std.testing.expect(total_elapsed >= 110 * std.time.ns_per_ms); try std.testing.expect(total_elapsed < 400 * std.time.ns_per_ms);}
test "Server: HTTP shutdown grace forces an active fallback at its own deadline" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var context = SlowHttpContext{ .io = io }; var server = try Server(SlowHttpHandler).init(t.allocator, io, .{ .port = 9297, .address = "127.0.0.1", .websocket_shutdown_grace = .fromMilliseconds(500), .http_shutdown_grace = .fromMilliseconds(80), }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9297) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(SlowHttpHandler), l: *net.Server, ctx: *SlowHttpContext) void { srv.runIo(l, ctx); } }; var future = try io.concurrent(Run.go, .{ &server, &listener, &context });
var stream = try testStreamPort(false, 9297); defer stream.close(); try stream.writeAll("GET /slow HTTP/1.1\r\nHost: localhost\r\n\r\n");
var attempts: usize = 0; while (!context.entered.load(.acquire) and attempts < 1000) : (attempts += 1) _ = try io.sleep(.fromMilliseconds(1), .awake); try std.testing.expect(context.entered.load(.acquire));
const started = Io.Timestamp.now(io, .awake).toNanoseconds(); _ = future.cancel(io); listener.deinit(io); const elapsed = Io.Timestamp.now(io, .awake).toNanoseconds() - started; try std.testing.expect(elapsed >= 60 * std.time.ns_per_ms); try std.testing.expect(elapsed < 400 * std.time.ns_per_ms); try std.testing.expect(context.unblocked.load(.acquire));}
test "Server: WebSocket and HTTP shutdown clocks run concurrently" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var context = MixedShutdownContext{ .io = io }; var server = try Server(MixedShutdownHandler).init(t.allocator, io, .{ .port = 9298, .address = "127.0.0.1", .websocket_shutdown_grace = .fromMilliseconds(150), .http_shutdown_grace = .fromMilliseconds(150), }); defer server.deinit();
var addr = net.IpAddress.parse("127.0.0.1", 9298) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(MixedShutdownHandler), l: *net.Server, ctx: *MixedShutdownContext) void { srv.runIo(l, ctx); } }; var future = try io.concurrent(Run.go, .{ &server, &listener, &context });
var websocket_stream = try testStreamPort(true, 9298); defer websocket_stream.close(); var http_stream = try testStreamPort(false, 9298); defer http_stream.close(); try http_stream.writeAll("GET /slow HTTP/1.1\r\nHost: localhost\r\n\r\n");
var attempts: usize = 0; while ((!context.websocket_entered.load(.acquire) or !context.http_entered.load(.acquire)) and attempts < 1000) : (attempts += 1) _ = try io.sleep(.fromMilliseconds(1), .awake); try std.testing.expect(context.websocket_entered.load(.acquire)); try std.testing.expect(context.http_entered.load(.acquire));
const started = Io.Timestamp.now(io, .awake).toNanoseconds(); _ = future.cancel(io); listener.deinit(io); const elapsed = Io.Timestamp.now(io, .awake).toNanoseconds() - started; try std.testing.expect(elapsed >= 110 * std.time.ns_per_ms); try std.testing.expect(elapsed < 260 * std.time.ns_per_ms); try std.testing.expect(context.http_canceled.load(.acquire));}
test "Server: permessage-deflate context takeover is real on the wire" { var threaded: std.Io.Threaded = .init(t.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var server = try Server(CompressionHandler).init(t.allocator, io, .{ .port = 9296, .address = "127.0.0.1", .compression = .{ .write_threshold = 128, .client_no_context_takeover = false, .server_no_context_takeover = false, }, }); defer server.deinit(); var addr = net.IpAddress.parse("127.0.0.1", 9296) catch unreachable; var listener = try addr.listen(io, .{ .reuse_address = true }); const Run = struct { fn go(srv: *Server(CompressionHandler), l: *net.Server) void { srv.runIo(l, {}); } }; var future = try io.concurrent(Run.go, .{ &server, &listener }); defer listener.deinit(io); defer _ = future.cancel(io);
const stream = try testStreamPort(false, 9296); defer stream.close(); try stream.writeAll( "GET / HTTP/1.1\r\n" ++ "content-length: 0\r\n" ++ "upgrade: websocket\r\n" ++ "sec-websocket-version: 13\r\n" ++ "connection: upgrade\r\n" ++ "sec-websocket-key: my-key\r\n" ++ "sec-websocket-extensions: permessage-deflate\r\n\r\n", ); var response: [512]u8 = undefined; var response_len: usize = 0; while (response_len < response.len) { _ = try stream.readAtLeast(response[response_len..][0..1], 1); response_len += 1; if (std.mem.endsWith(u8, response[0..response_len], "\r\n\r\n")) break; } try std.testing.expect(std.mem.indexOf(u8, response[0..response_len], "Sec-WebSocket-Extensions: permessage-deflate\r\n") != null); try std.testing.expect(std.mem.indexOf(u8, response[0..response_len], "no_context_takeover") == null);
var frames: [4096]u8 = undefined; var frames_len: usize = 0; var payload_lengths: [compression_test_messages.len]usize = undefined; for (compression_test_messages, 0..) |expected, message_index| { const start = frames_len; _ = try stream.readAtLeast(frames[frames_len..][0..2], 2); frames_len += 2; try t.expectEqual(@as(u8, if (expected.len >= 128) 0xc1 else 0x81), frames[start]); const short_len = frames[start + 1] & 0x7f; const payload_len: usize = switch (short_len) { 0...125 => short_len, 126 => blk: { _ = try stream.readAtLeast(frames[frames_len..][0..2], 2); const n = std.mem.readInt(u16, frames[frames_len..][0..2], .big); frames_len += 2; break :blk n; }, else => return error.TestUnexpectedResult, }; payload_lengths[message_index] = payload_len; _ = try stream.readAtLeast(frames[frames_len..][0..payload_len], payload_len); frames_len += payload_len; } // The third message repeats the first after a different middle message. // Its smaller frame proves the compressor retained cross-message history; // successful decoding below proves the inflater retained the same history. try std.testing.expect(payload_lengths[2] < payload_lengths[0]);
const SliceStream = struct { bytes: []const u8, pos: usize = 0, pub fn read(self: *@This(), out: []u8) !usize { if (self.pos == self.bytes.len) return error.Closed; const n = @min(out.len, self.bytes.len - self.pos); @memcpy(out[0..n], self.bytes[self.pos..][0..n]); self.pos += n; return n; } }; var source: SliceStream = .{ .bytes = frames[0..frames_len] }; var provider = try buffer.Provider.init(t.allocator, .{ .max = 65_536, .count = 0, .size = 0 }); defer provider.deinit(); var reader_buf: [2048]u8 = undefined; var reader = Reader.init(&reader_buf, &provider, .{ .client_no_context_takeover = false, .server_no_context_takeover = false, }); defer reader.deinit(); var more = false; for (compression_test_messages) |expected| { if (!more) try reader.fill(&source); more, const message = (try reader.read()).?; try t.expectEqual(Message.Type.text, message.type); try t.expectString(expected, message.data); reader.done(message.type); }
const disabled = try testStreamPort(false, 9296); defer disabled.close(); try disabled.writeAll( "GET /disabled HTTP/1.1\r\n" ++ "content-length: 0\r\n" ++ "upgrade: websocket\r\n" ++ "sec-websocket-version: 13\r\n" ++ "connection: upgrade\r\n" ++ "sec-websocket-key: my-key\r\n" ++ "sec-websocket-extensions: permessage-deflate\r\n\r\n", ); var disabled_response: [512]u8 = undefined; var disabled_len: usize = 0; while (disabled_len < disabled_response.len) { _ = try disabled.readAtLeast(disabled_response[disabled_len..][0..1], 1); disabled_len += 1; if (std.mem.endsWith(u8, disabled_response[0..disabled_len], "\r\n\r\n")) break; } try std.testing.expect(std.mem.indexOf( u8, disabled_response[0..disabled_len], "Sec-WebSocket-Extensions", ) == null);}
test "Server: invalid handshake" { const stream = try testStream(false); defer stream.close();
try stream.writeAll("GET / HTTP/1.1\r\n\r\n"); var buf: [1024]u8 = undefined; var pos: usize = 0; while (pos < buf.len) { const n = try stream.read(buf[0..]); if (n == 0) { break; } pos += n; } else { unreachable; }
try t.expectString("HTTP/1.1 400 \r\nConnection: Close\r\nError: missingheaders\r\nContent-Length: 0\r\n\r\n", buf[0..pos]);}
test "Server: read and write" { const stream = try testStream(true); defer stream.close();
try stream.writeAll(&proto.frame(.text, "over")); var buf: [12]u8 = undefined; _ = try stream.readAtLeast(&buf, 6); try t.expectSlice(u8, &.{ 129, 4, '9', '0', '0', '0' }, buf[0..6]);}
test "Server: handler can interrupt its blocked reader safely" { const stream = try testStream(true); defer stream.close();
try stream.writeAll(&proto.frame(.text, "interrupt")); var byte: [1]u8 = undefined; try t.expectEqual(@as(usize, 0), try stream.read(&byte));}
test "Server: clientMessage allocator" { const stream = try testStream(true); defer stream.close();
try stream.writeAll(&proto.frame(.text, "dyn")); var buf: [12]u8 = undefined; _ = try stream.readAtLeast(&buf, 12); try t.expectSlice(u8, &.{ 129, 10, 'o', 'v', 'e', 'r', ' ', '9', '0', '0', '0', '!' }, buf[0..12]);}
test "Server: clientMessage writer" { const stream = try testStream(true); defer stream.close();
try stream.writeAll(&proto.frame(.text, "writer")); var buf: [9]u8 = undefined; _ = try stream.readAtLeast(&buf, 9); try t.expectSlice(u8, &.{ 129, 7, '9', '0', '0', '0', '!', '!', '!' }, buf[0..9]);}
// Same as above, but client doesn't shutdown the connection// When afterAll is runs and things are shutdown, this should still be properly cleaned uptest "Server: dirty clientMessage allocator" { const stream = try testStream(true);
try stream.writeAll(&proto.frame(.text, "dyn")); var buf: [12]u8 = undefined; _ = try stream.readAtLeast(&buf, 12); try t.expectSlice(u8, &.{ 129, 10, 'o', 'v', 'e', 'r', ' ', '9', '0', '0', '0', '!' }, buf[0..12]);}
test "Conn: close" { { // plain close const stream = try testStream(true); defer stream.close(); try stream.writeAll(&proto.frame(.text, "close1")); var buf: [4]u8 = undefined; _ = try stream.readAtLeast(&buf, 4); try t.expectSlice(u8, &.{ 136, 2, 3, 232 }, buf[0..4]); }
{ // close with code const stream = try testStream(true); defer stream.close(); try stream.writeAll(&proto.frame(.text, "close2")); var buf: [4]u8 = undefined; _ = try stream.readAtLeast(&buf, 4); try t.expectSlice(u8, &.{ 136, 2, 0, 0x7b }, buf[0..4]); }
{ // close with reason const stream = try testStream(true); defer stream.close(); try stream.writeAll(&proto.frame(.text, "close3")); var buf: [7]u8 = undefined; _ = try stream.readAtLeast(&buf, 7); try t.expectSlice(u8, &.{ 136, 5, 0, 0xea, 'b', 'y', 'e' }, buf[0..7]); }}
const TestStream = struct { socket: posix.socket_t, io: Io,
fn close(self: *const TestStream) void { self.io.vtable.netClose(self.io.userdata, (&self.socket)[0..1]); }
fn writeAll(self: *const TestStream, data: []const u8) !void { try socketWriteAll(self.io, self.socket, data); }
fn readAtLeast(self: *const TestStream, buf: []u8, min: usize) !usize { var total: usize = 0; while (total < min) { const n = try self.read(buf[total..]); if (n == 0) break; total += n; } return total; }
fn read(self: *const TestStream, buf: []u8) !usize { return (SocketReader{ .socket = self.socket }).read(buf); }};
fn testStream(handshake: bool) !TestStream { return testStreamPort(handshake, 9292);}
fn testStreamPort(handshake: bool, port: u16) !TestStream { const io = std.Options.debug_io;
// Connect via Io.net var addr = net.IpAddress.parse("127.0.0.1", port) catch unreachable; const stream = try net.IpAddress.connect(&addr, io, .{ .mode = .stream });
const socket = stream.socket.handle;
// Socket options (timeouts) — no Io equivalent, use posix.setsockopt const timeout = std.mem.toBytes(posix.timeval{ .sec = 0, .usec = 20_000 }); try posix.setsockopt(socket, posix.IPPROTO.TCP, 1, &std.mem.toBytes(@as(c_int, 1))); try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &timeout); try posix.setsockopt(socket, posix.SOL.SOCKET, posix.SO.SNDTIMEO, &timeout);
const result = TestStream{ .socket = socket, .io = io };
if (handshake == false) { return result; }
return finishTestHandshake(result, "/");}
fn testStreamPortPath(port: u16, path: []const u8) !TestStream { const result = try testStreamPort(false, port); return finishTestHandshake(result, path);}
fn finishTestHandshake(result: TestStream, path: []const u8) !TestStream { try result.writeAll("GET "); try result.writeAll(path); try result.writeAll(" HTTP/1.1\r\ncontent-length: 0\r\nupgrade: websocket\r\nsec-websocket-version: 13\r\nconnection: upgrade\r\nsec-websocket-key: my-key\r\n\r\n"); var buf: [1024]u8 = undefined; var pos: usize = 0; while (pos < buf.len) { const n = try result.read(buf[pos..]); if (n == 0) break;
pos += n; if (std.mem.endsWith(u8, buf[0..pos], "\r\n\r\n")) { break; } } else { unreachable; }
try t.expectString("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: upgrade\r\nSec-Websocket-Accept: L8KGBs4w2MNLLzhfzlVoM0scCIE=\r\n\r\n", buf[0..pos]);
return result;}
const TestHandler = struct { conn: *Conn,
pub fn init(h: *const Handshake, conn: *Conn, _: void) !TestHandler { try t.expectString("upgrade", h.headers.get("connection").?); return .{ .conn = conn, }; } pub fn clientMessage( self: *TestHandler, allocator: Allocator, data: []const u8, ) !void { if (std.mem.eql(u8, data, "over")) { return self.conn.writeText("9000"); } if (std.mem.eql(u8, data, "dyn")) { return self.conn.writeText(try std.fmt.allocPrint(allocator, "over {d}!", .{9000})); } if (std.mem.eql(u8, data, "writer")) { var wb = self.conn.writeBuffer(allocator, .text); try wb.interface.print("{d}!!!", .{9000}); return wb.send(); } if (std.mem.eql(u8, data, "ping")) { var buf = [_]u8{ 'a', '-', 'p', 'i', 'n', 'g' }; return self.conn.writePing(&buf); } if (std.mem.eql(u8, data, "pong")) { var buf = [_]u8{ 'a', '-', 'p', 'o', 'n', 'g' }; return self.conn.writePong(&buf); } if (std.mem.eql(u8, data, "interrupt")) { self.conn.interruptRead(); return; } if (std.mem.eql(u8, data, "close1")) { return self.conn.close(.{}); } if (std.mem.eql(u8, data, "close2")) { return self.conn.close(.{ .code = 123 }); } if (std.mem.eql(u8, data, "close3")) { return self.conn.close(.{ .code = 234, .reason = "bye" }); } }};
const compression_history_a_chunk = "alpha-window::the first dictionary must survive a different middle message::0123456789abcdef::";const compression_history_a_repeated: [120][compression_history_a_chunk.len]u8 = @splat(compression_history_a_chunk.*);const compression_history_a = std.mem.asBytes(&compression_history_a_repeated);
const compression_history_b_chunk = "beta-window::this second dictionary must not erase the earlier stream history::fedcba9876543210::";const compression_history_b_repeated: [120][compression_history_b_chunk.len]u8 = @splat(compression_history_b_chunk.*);const compression_history_b = std.mem.asBytes(&compression_history_b_repeated);const compression_test_messages = [_][]const u8{ compression_history_a, compression_history_b, compression_history_a, "", compression_history_a,};
const CompressionHandler = struct { conn: *Conn, enabled: bool,
pub fn init(handshake: *const Handshake, conn: *Conn, _: void) !CompressionHandler { const enabled = !std.mem.eql(u8, handshake.url, "/disabled"); if (!enabled) conn.disableCompression(); return .{ .conn = conn, .enabled = enabled }; }
pub fn afterInit(self: *CompressionHandler) !void { if (!self.enabled) return; for (compression_test_messages) |message| try self.conn.writeText(message); }
pub fn clientMessage(_: *CompressionHandler, _: []const u8) !void {}};
const ErrorHandler = struct { observed: *std.atomic.Value(u8),
pub fn init(_: *const Handshake, _: *Conn, observed: *std.atomic.Value(u8)) !ErrorHandler { return .{ .observed = observed }; }
pub fn clientMessage(_: *ErrorHandler, _: []const u8) !void {}
pub fn clientError(self: *ErrorHandler, err: anyerror) void { self.observed.store(if (err == error.TooLarge) 1 else 2, .release); }};
const LifecycleHandler = struct { lifecycle: *std.atomic.Value(u8), conn: *Conn,
pub fn init(_: *const Handshake, conn: *Conn, lifecycle: *std.atomic.Value(u8)) !LifecycleHandler { try conn.writeTimeout(20); lifecycle.store(1, .release); return .{ .lifecycle = lifecycle, .conn = conn }; }
pub fn clientMessage(_: *LifecycleHandler, _: []const u8) !void {}
pub fn serverClose(self: *LifecycleHandler) void { self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch unreachable; self.lifecycle.store(2, .release); }
pub fn close(_: *LifecycleHandler) void {}};
const FanoutShutdownContext = struct { blocked_ready: std.atomic.Value(bool) = .init(false), healthy_ready: std.atomic.Value(bool) = .init(false), blocked_close_started: std.atomic.Value(bool) = .init(false), blocked_close_unblocked: std.atomic.Value(bool) = .init(false), healthy_notified: std.atomic.Value(bool) = .init(false),};
const FanoutShutdownHandler = struct { context: *FanoutShutdownContext, conn: *Conn, blocked: bool,
pub fn init(handshake: *const Handshake, conn: *Conn, context: *FanoutShutdownContext) !FanoutShutdownHandler { const blocked = std.mem.eql(u8, handshake.url, "/blocked"); if (blocked) { const send_buffer = std.mem.toBytes(@as(c_int, 4096)); setSockOptBestEffort(conn.stream.socket.handle, posix.SOL.SOCKET, posix.SO.SNDBUF, &send_buffer); try conn.writeTimeout(5000); context.blocked_ready.store(true, .release); } else { context.healthy_ready.store(true, .release); } return .{ .context = context, .conn = conn, .blocked = blocked }; }
pub fn clientMessage(_: *FanoutShutdownHandler, _: []const u8) !void {}
pub fn serverClose(self: *FanoutShutdownHandler) void { if (self.blocked) { self.context.blocked_close_started.store(true, .release); const payload: [16 * 1024]u8 = @splat('x'); while (true) self.conn.writeBin(&payload) catch { self.context.blocked_close_unblocked.store(true, .release); return; }; }
self.conn.writeClose(.{ .code = 1001 }) catch return; self.context.healthy_notified.store(true, .release); }
pub fn close(_: *FanoutShutdownHandler) void {}};
const SlowHttpContext = struct { io: Io, entered: std.atomic.Value(bool) = .init(false), unblocked: std.atomic.Value(bool) = .init(false),};
const SlowHttpHandler = struct { pub fn init(_: *const Handshake, _: *Conn, _: *SlowHttpContext) !SlowHttpHandler { return .{}; }
pub fn clientMessage(_: *SlowHttpHandler, _: []const u8) !void {}
pub fn httpFallback( conn: *Conn, _: []const u8, _: []const u8, _: []const u8, _: anytype, context: *SlowHttpContext, ) void { context.entered.store(true, .release); const send_buffer = std.mem.toBytes(@as(c_int, 4096)); const socket = conn.stream.socket.handle; setSockOptBestEffort(socket, posix.SOL.SOCKET, posix.SO.SNDBUF, &send_buffer); socketWriteAll(context.io, socket, "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 1073741824\r\n\r\n") catch { context.unblocked.store(true, .release); return; }; const body: [16 * 1024]u8 = @splat('x'); while (true) socketWriteAll(context.io, socket, &body) catch { context.unblocked.store(true, .release); return; }; }};
const MixedShutdownContext = struct { io: Io, websocket_entered: std.atomic.Value(bool) = .init(false), http_entered: std.atomic.Value(bool) = .init(false), http_canceled: std.atomic.Value(bool) = .init(false),};
const MixedShutdownHandler = struct { context: *MixedShutdownContext, conn: *Conn,
pub fn init(_: *const Handshake, conn: *Conn, context: *MixedShutdownContext) !MixedShutdownHandler { context.websocket_entered.store(true, .release); return .{ .context = context, .conn = conn }; }
pub fn clientMessage(_: *MixedShutdownHandler, _: []const u8) !void {}
pub fn serverClose(self: *MixedShutdownHandler) void { self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch {}; }
pub fn close(_: *MixedShutdownHandler) void {}
pub fn httpFallback( _: *Conn, _: []const u8, _: []const u8, _: []const u8, _: anytype, context: *MixedShutdownContext, ) void { context.http_entered.store(true, .release); context.io.sleep(.fromSeconds(5), .awake) catch |err| { if (err == error.Canceled) context.http_canceled.store(true, .release); }; }};
test "List" { var list = List(TestNode){}; try expectList(&.{}, list);
var n1 = TestNode{ .id = 1 }; list.insert(&n1); try expectList(&.{1}, list);
list.remove(&n1); try expectList(&.{}, list);
var n2 = TestNode{ .id = 2 }; list.insert(&n2); list.insert(&n1); try expectList(&.{ 2, 1 }, list);
var n3 = TestNode{ .id = 3 }; list.insert(&n3); try expectList(&.{ 2, 1, 3 }, list);
list.remove(&n1); try expectList(&.{ 2, 3 }, list);
list.insert(&n1); try expectList(&.{ 2, 3, 1 }, list);
list.remove(&n2); try expectList(&.{ 3, 1 }, list);
list.remove(&n1); try expectList(&.{3}, list);
list.remove(&n3); try expectList(&.{}, list);}
const TestNode = struct { id: i32, next: ?*TestNode = null, prev: ?*TestNode = null,};
fn expectList(expected: []const i32, list: List(TestNode)) !void { if (expected.len == 0) { try t.expectEqual(null, list.head); try t.expectEqual(null, list.tail); return; }
var i: usize = 0; var next = list.head; while (next) |node| { try t.expectEqual(expected[i], node.id); i += 1; next = node.next; } try t.expectEqual(expected.len, i);
i = expected.len; var prev = list.tail; while (prev) |node| { i -= 1; try t.expectEqual(expected[i], node.id); prev = node.prev; } try t.expectEqual(0, i);}
// -- parseHttpRequest tests --// All tests use Handshake.Pool to get properly-initialized State (with pre-allocated KeyValue headers).
fn testParseHttpWithState(raw: []const u8, state: *Handshake.State) ?HttpRequest { @memcpy(state.buf[0..raw.len], raw); return parseHttpRequest(state.buf, raw.len, &state.req_headers);}
test "Conn peerIp excludes ephemeral ports for IPv4 and IPv6" { var conn: Conn = undefined; conn.address = try Address.parseIp("203.0.113.10", 5555); var buf: [64]u8 = undefined; try t.expectString("203.0.113.10", try conn.peerIp(&buf));
var storage = std.mem.zeroes(posix.sockaddr.storage); const in6 = @as(*posix.sockaddr.in6, @ptrCast(@alignCast(&storage))); in6.family = posix.AF.INET6; in6.port = @byteSwap(@as(u16, 5556)); in6.addr = .{ 0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 }; conn.address = .{ .any = storage }; try t.expectString("2001:db8:0:0:0:0:0:1", try conn.peerIp(&buf));}
test "parseHttpRequest: plain GET health probe" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const result = testParseHttpWithState( "GET /_healthz HTTP/1.1\r\nHost: localhost:3000\r\n\r\n", state, ) orelse return error.ExpectedResult;
try t.expectString("GET", result.method); try t.expectString("/_healthz", result.url); try t.expectEqual(@as(usize, 0), result.body.len); try t.expectEqual(@as(usize, 1), state.req_headers.len); try t.expectString("localhost:3000", state.req_headers.get("host").?);}
test "parseHttpRequest: GET with Connection: keep-alive" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const result = testParseHttpWithState( "GET /_healthz HTTP/1.1\r\nHost: localhost:3000\r\nConnection: keep-alive\r\nAccept: */*\r\n\r\n", state, ) orelse return error.ExpectedResult;
try t.expectString("GET", result.method); try t.expectString("/_healthz", result.url); try t.expectEqual(@as(usize, 0), result.body.len); try t.expectEqual(@as(usize, 3), state.req_headers.len); try t.expectString("localhost:3000", state.req_headers.get("host").?); try t.expectString("keep-alive", state.req_headers.get("connection").?); try t.expectString("*/*", state.req_headers.get("accept").?);}
test "parseHttpRequest: POST with body" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const result = testParseHttpWithState( "POST /xrpc/com.atproto.sync.requestCrawl HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: 29\r\n\r\n{\"hostname\":\"example.com\"}", state, ) orelse return error.ExpectedResult;
try t.expectString("POST", result.method); try t.expectString("/xrpc/com.atproto.sync.requestCrawl", result.url); try t.expectString("{\"hostname\":\"example.com\"}", result.body); try t.expectEqual(@as(usize, 3), state.req_headers.len); try t.expectString("application/json", state.req_headers.get("content-type").?);}
test "parseHttpRequest: header names are lowercased" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const result = testParseHttpWithState( "GET / HTTP/1.1\r\nHost: localhost\r\nX-Custom-Header: SomeValue\r\nACCEPT: text/html\r\n\r\n", state, ) orelse return error.ExpectedResult;
try t.expectString("GET", result.method); try t.expectString("/", result.url); try t.expectString("localhost", state.req_headers.get("host").?); try t.expectString("SomeValue", state.req_headers.get("x-custom-header").?); try t.expectString("text/html", state.req_headers.get("accept").?);}
test "parseHttpRequest: incomplete request returns null" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
try t.expectEqual(null, testParseHttpWithState( "GET /_healthz HTTP/1.1\r\nHost: localhost\r\n", state, ));}
test "parseHttpRequest: empty buffer returns null" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
try t.expectEqual(null, parseHttpRequest(state.buf, 0, &state.req_headers));}
test "parseHttpRequest: malformed request line returns null" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
try t.expectEqual(null, testParseHttpWithState("GARBAGE\r\n\r\n", state));}
test "parseHttpRequest: URL with query string" { var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const result = testParseHttpWithState( "GET /xrpc/com.atproto.sync.getLatestCommit?did=did:plc:abc HTTP/1.1\r\nHost: localhost\r\n\r\n", state, ) orelse return error.ExpectedResult;
try t.expectString("GET", result.method); try t.expectString("/xrpc/com.atproto.sync.getLatestCommit?did=did:plc:abc", result.url);}
test "parseHttpRequest: clean re-parse after failed Handshake.parse" { // Simulate the real scenario: Handshake.parse fails with InvalidConnection, // leaving partial header state and partially-lowercased buffer. // parseHttpRequest must re-parse all headers cleanly. var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const raw = "GET /_healthz HTTP/1.1\r\nHost: localhost\r\nConnection: keep-alive\r\nX-Request-Id: abc123\r\n\r\n"; @memcpy(state.buf[0..raw.len], raw); state.len = raw.len;
// Handshake.parse fails at Connection: keep-alive (no "upgrade") try t.expectError(error.InvalidConnection, Handshake.parse(state)); // state.req_headers now has partial data (host, connection but NOT x-request-id) // buffer has "host" and "connection" lowered, "X-Request-Id" untouched
// parseHttpRequest re-parses cleanly: all 3 headers, all names lowercase const result = parseHttpRequest(state.buf, state.len, &state.req_headers) orelse { return error.ExpectedResult; };
try t.expectString("GET", result.method); try t.expectString("/_healthz", result.url); try t.expectEqual(@as(usize, 3), state.req_headers.len); try t.expectString("localhost", state.req_headers.get("host").?); try t.expectString("keep-alive", state.req_headers.get("connection").?); try t.expectString("abc123", state.req_headers.get("x-request-id").?);}
test "parseHttpRequest: MissingHeaders error also recoverable" { // Plain HTTP with no websocket headers at all → MissingHeaders var pool = try Handshake.Pool.init(t.allocator, 1, 4096, 32, 1); defer pool.deinit(); var state = try pool.acquire(); defer state.release();
const raw = "GET /_readyz HTTP/1.1\r\nHost: localhost\r\nUser-Agent: kube-probe/1.28\r\n\r\n"; @memcpy(state.buf[0..raw.len], raw); state.len = raw.len;
// Handshake.parse fails with MissingHeaders (no websocket headers) try t.expectError(error.MissingHeaders, Handshake.parse(state));
// parseHttpRequest recovers the request const result = parseHttpRequest(state.buf, state.len, &state.req_headers) orelse { return error.ExpectedResult; };
try t.expectString("GET", result.method); try t.expectString("/_readyz", result.url); try t.expectEqual(@as(usize, 2), state.req_headers.len); try t.expectString("localhost", state.req_headers.get("host").?); try t.expectString("kube-probe/1.28", state.req_headers.get("user-agent").?);}