Something went wrong. Try again.
javascript for 🐜's, a tiny runtime with big ambitions antjs.org
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683const std = @import("std");const c_allocator = std.heap.c_allocator;
const debug = @import("debug.zig");const extractor = @import("extractor.zig");const uv = @import("uv.zig");const tlsuv = @import("tlsuv.zig");const nghttp2 = @import("nghttp2.zig");
const config = @import("config");const user_agent: [:0]const u8 = "ant/" ++ config.version;
pub const FetchError = error{ ConnectionFailed, TlsError, Http2Error, Timeout, InvalidUrl, ResponseError, OutOfMemory,};
pub const ParsedUrl = struct { scheme: []const u8, host: []const u8, port: u16, path: []const u8,
pub fn parse(url: []const u8) !ParsedUrl { var remaining = url; const scheme_end = std.mem.indexOf(u8, remaining, "://") orelse return error.InvalidUrl; const scheme = remaining[0..scheme_end]; remaining = remaining[scheme_end + 3 ..];
const path_start = std.mem.indexOf(u8, remaining, "/") orelse remaining.len; const host_port = remaining[0..path_start]; remaining = if (path_start < remaining.len) remaining[path_start..] else "/";
var host: []const u8 = host_port; var port: u16 = if (std.mem.eql(u8, scheme, "https")) 443 else 80;
if (std.mem.indexOf(u8, host_port, ":")) |colon| { host = host_port[0..colon]; port = std.fmt.parseInt(u16, host_port[colon + 1 ..], 10) catch return error.InvalidUrl; }
return .{ .scheme = scheme, .host = host, .port = port, .path = remaining }; }};
pub const StreamHandler = struct { on_data: *const fn ([]const u8, ?*anyopaque) void, on_complete: *const fn (u16, ?*anyopaque) void, on_error: *const fn (FetchError, ?*anyopaque) void, user_data: ?*anyopaque, pub fn init( on_data: *const fn ([]const u8, ?*anyopaque) void, on_complete: *const fn (u16, ?*anyopaque) void, on_error: *const fn (FetchError, ?*anyopaque) void, user_data: ?*anyopaque, ) StreamHandler { return .{ .on_data = on_data, .on_complete = on_complete, .on_error = on_error, .user_data = user_data }; }};
const PendingRequest = struct { url: []const u8, handler: ?StreamHandler,};
const MAX_PENDING_REQUESTS = 20;const NUM_CONNECTIONS = 6;const NUM_META_CONNECTIONS = 3;const META_SLOW_LOG_MS: u64 = 250;
const Http2Client = struct { allocator: std.mem.Allocator, loop: *uv.loop_t, tls: tlsuv.stream_t, h2_session: ?*nghttp2.session, host: [:0]const u8, use_tls: bool, connected: i32, connect_pending: bool, closing: bool, write_buf: std.ArrayListUnmanaged(u8), requests: [MAX_PENDING_REQUESTS]RequestState, request_count: usize, requests_done: usize, last_response_status_code: u16,
const RequestState = struct { stream_id: i32, path: ?[:0]const u8, on_data: ?*const fn ([]const u8, ?*anyopaque) void, on_complete: ?*const fn (u16, ?*anyopaque) void, on_error: ?*const fn (FetchError, ?*anyopaque) void, userdata: ?*anyopaque, response_body: std.ArrayListUnmanaged(u8), status_code: u16, done: bool, has_error: bool, start_ns: u64, end_ns: u64, bytes: usize, content_encoding: ContentEncoding, };
const ContentEncoding = enum { identity, gzip, };
const alpn_protocols = [_][*:0]const u8{ "h2", "http/1.1" };
pub fn init(allocator: std.mem.Allocator, host: []const u8, use_tls: bool) !*Http2Client { const client = try allocator.create(Http2Client); errdefer allocator.destroy(client);
const host_z = try allocator.dupeZ(u8, host); errdefer allocator.free(host_z);
client.* = .{ .allocator = allocator, .loop = uv.uv_default_loop(), .tls = .{}, .h2_session = null, .host = host_z, .use_tls = use_tls, .connected = 0, .connect_pending = false, .closing = false, .write_buf = .{}, .requests = undefined, .request_count = 0, .requests_done = 0, .last_response_status_code = 0, };
for (&client.requests) |*req| { req.* = .{ .stream_id = 0, .path = null, .on_data = null, .on_complete = null, .on_error = null, .userdata = null, .response_body = .{}, .status_code = 0, .done = false, .has_error = false, .start_ns = 0, .end_ns = 0, .bytes = 0, .content_encoding = .identity, }; }
if (tlsuv.tlsuv_stream_init(client.loop, &client.tls, null) != 0) { allocator.free(host_z); allocator.destroy(client); return error.ConnectionFailed; }
_ = tlsuv.tlsuv_stream_set_hostname(&client.tls, host_z.ptr); _ = tlsuv.tlsuv_stream_set_protocols(&client.tls, 2, &alpn_protocols);
return client; }
pub fn deinit(self: *Http2Client) void { self.closing = true; self.connect_pending = false;
for (&self.requests) |*req| { req.on_data = null; req.on_complete = null; req.on_error = null; req.userdata = null; }
if (self.connected > 0) { self.tls.data = self; _ = tlsuv.tlsuv_stream_close(&self.tls, onStreamClose); while (self.connected > 0) _ = uv.uv_run(self.loop, uv.RUN_ONCE); }
if (self.h2_session) |session| nghttp2.nghttp2_session_del(session);
for (&self.requests) |*req| { if (req.stream_id != -1) { if (req.path) |p| self.allocator.free(p); req.response_body.deinit(self.allocator); } } self.write_buf.deinit(self.allocator); self.allocator.free(self.host); self.allocator.destroy(self); }
pub fn resetRequests(self: *Http2Client) void { for (self.requests[0..self.request_count]) |*req| { if (req.stream_id != -1) { if (req.path) |p| self.allocator.free(p); req.response_body.deinit(self.allocator); } req.* = .{ .stream_id = 0, .path = null, .on_data = null, .on_complete = null, .on_error = null, .userdata = null, .response_body = .{}, .status_code = 0, .done = false, .has_error = false, .start_ns = 0, .end_ns = 0, .bytes = 0, .content_encoding = .identity, }; } self.request_count = 0; self.requests_done = 0; }
pub fn hasCapacity(self: *const Http2Client) bool { for (self.requests[0..self.request_count]) |req| { if (req.stream_id == -1) return true; } return self.request_count < MAX_PENDING_REQUESTS - 1; }
pub fn recycleCompletedRequests(self: *Http2Client) void { if (self.requests_done == 0) return;
for (self.requests[0..self.request_count]) |*req| { if (req.done and req.stream_id != -1) { if (req.path) |p| self.allocator.free(p); req.response_body.deinit(self.allocator); req.path = null; req.response_body = .{}; req.stream_id = -1; } } }
fn findOrAllocSlot(self: *Http2Client) ?*RequestState { for (self.requests[0..self.request_count]) |*req| { if (req.stream_id == -1) return req; } if (self.request_count < MAX_PENDING_REQUESTS) { const req = &self.requests[self.request_count]; self.request_count += 1; return req; } return null; }
fn onStreamClose(handle: *uv.handle_t) callconv(.c) void { const tls: *tlsuv.stream_t = @ptrCast(@alignCast(handle)); const client: *Http2Client = @ptrCast(@alignCast(tls.data)); client.connected = -2; client.connect_pending = false; }
fn findRequest(self: *Http2Client, stream_id: i32) ?*RequestState { for (self.requests[0..self.request_count]) |*req| if (req.stream_id == stream_id) return req; return null; }
fn h2Send(_: ?*nghttp2.session, data: [*c]const u8, len: usize, _: c_int, ud: ?*anyopaque) callconv(.c) isize { const client: *Http2Client = @ptrCast(@alignCast(ud)); client.write_buf.appendSlice(client.allocator, data[0..len]) catch return nghttp2.ERR_NOMEM; return @intCast(len); }
fn h2FrameRecv(_: ?*nghttp2.session, frame: *const nghttp2.frame, ud: ?*anyopaque) callconv(.c) c_int { const client: *Http2Client = @ptrCast(@alignCast(ud)); if (frame.hd.flags & nghttp2.FLAG_END_STREAM != 0) { if (client.findRequest(frame.hd.stream_id)) |req| { if (!req.done) { req.done = true; req.end_ns = @intCast(std.time.nanoTimestamp()); client.requests_done += 1; if (req.on_complete) |cb| cb(req.status_code, req.userdata); } } } return 0; }
fn h2DataChunk(session: ?*nghttp2.session, _: u8, stream_id: i32, data: [*c]const u8, len: usize, ud: ?*anyopaque) callconv(.c) c_int { const client: *Http2Client = @ptrCast(@alignCast(ud)); const req = client.findRequest(stream_id) orelse return 0; if (req.on_data) |cb| cb(data[0..len], req.userdata) else req.response_body.appendSlice(client.allocator, data[0..len]) catch { req.has_error = true; }; req.bytes += len; if (session) |s| _ = nghttp2.nghttp2_session_consume(s, stream_id, len);
return 0; }
fn h2Header(_: ?*nghttp2.session, frame: *const nghttp2.frame, name: [*c]const u8, namelen: usize, value: [*c]const u8, valuelen: usize, _: u8, ud: ?*anyopaque) callconv(.c) c_int { const client: *Http2Client = @ptrCast(@alignCast(ud)); if (frame.hd.type != nghttp2.HEADERS) return 0; const req = client.findRequest(frame.hd.stream_id) orelse return 0; if (namelen == 7 and std.mem.eql(u8, name[0..7], ":status")) req.status_code = std.fmt.parseInt(u16, value[0..valuelen], 10) catch 0; if (std.mem.eql(u8, name[0..namelen], "content-encoding")) { if (std.mem.startsWith(u8, value[0..valuelen], "gzip")) { req.content_encoding = .gzip; } } return 0; }
fn h2StreamClose(_: ?*nghttp2.session, stream_id: i32, error_code: u32, ud: ?*anyopaque) callconv(.c) c_int { const client: *Http2Client = @ptrCast(@alignCast(ud)); const req = client.findRequest(stream_id) orelse return 0; if (!req.done) { req.done = true; req.end_ns = @intCast(std.time.nanoTimestamp()); client.requests_done += 1; if (error_code != 0) { req.has_error = true; if (req.on_error) |cb| cb(FetchError.Http2Error, req.userdata); } else if (req.on_complete) |cb| cb(req.status_code, req.userdata); } return 0; }
fn initH2(self: *Http2Client) !void { var callbacks: *nghttp2.session_callbacks = undefined; if (nghttp2.nghttp2_session_callbacks_new(&callbacks) != 0) return error.Http2Error; defer nghttp2.nghttp2_session_callbacks_del(callbacks);
nghttp2.nghttp2_session_callbacks_set_send_callback2(callbacks, h2Send); nghttp2.nghttp2_session_callbacks_set_on_frame_recv_callback(callbacks, h2FrameRecv); nghttp2.nghttp2_session_callbacks_set_on_data_chunk_recv_callback(callbacks, h2DataChunk); nghttp2.nghttp2_session_callbacks_set_on_header_callback(callbacks, h2Header); nghttp2.nghttp2_session_callbacks_set_on_stream_close_callback(callbacks, h2StreamClose);
var session: *nghttp2.session = undefined; if (nghttp2.nghttp2_session_client_new(&session, callbacks, self) != 0) return error.Http2Error; self.h2_session = session;
var settings = [_]nghttp2.settings_entry{ .{ .settings_id = nghttp2.SETTINGS_MAX_CONCURRENT_STREAMS, .value = MAX_PENDING_REQUESTS }, .{ .settings_id = nghttp2.SETTINGS_INITIAL_WINDOW_SIZE, .value = 16 * 1024 * 1024 }, }; if (nghttp2.nghttp2_submit_settings(self.h2_session.?, nghttp2.FLAG_NONE, &settings, settings.len) != 0) return error.Http2Error;
const conn_window_increase: i32 = (16 * 1024 * 1024) - 65535; _ = nghttp2.nghttp2_submit_window_update(self.h2_session.?, nghttp2.FLAG_NONE, 0, conn_window_increase); }
fn flush(self: *Http2Client) !void { if (self.closing) return error.ConnectionFailed; if (self.h2_session) |session| while (nghttp2.nghttp2_session_want_write(session) != 0) if (nghttp2.nghttp2_session_send(session) != 0) break; if (self.write_buf.items.len > 0) { const data = try self.allocator.dupe(u8, self.write_buf.items); self.write_buf.clearRetainingCapacity(); const wr = try self.allocator.create(uv.write_t); wr.data = data.ptr; var buf = uv.buf_t{ .base = data.ptr, .len = data.len }; if (tlsuv.tlsuv_stream_write(wr, &self.tls, &buf, onWrite) != 0) { self.allocator.free(data); self.allocator.destroy(wr); return error.ConnectionFailed; } } }
fn onWrite(wr: *uv.write_t, _: c_int) callconv(.c) void { const data_ptr: [*]u8 = @ptrCast(wr.data); std.c.free(data_ptr); std.c.free(@ptrCast(wr)); }
fn allocBuf(_: *uv.handle_t, size: usize, buf: *uv.buf_t) callconv(.c) void { buf.base = @ptrCast(std.c.malloc(size) orelse return); buf.len = size; }
fn onRead(stream: *uv.stream_t, nread: isize, buf: *const uv.buf_t) callconv(.c) void { const tls: *tlsuv.stream_t = @ptrCast(@alignCast(stream)); const client: *Http2Client = @ptrCast(@alignCast(tls.data)); defer if (buf.base) |b| std.c.free(b); if (client.closing) return; if (nread < 0) { for (client.requests[0..client.request_count]) |*req| if (!req.done) { req.done = true; req.has_error = true; client.requests_done += 1; if (req.on_error) |cb| cb(FetchError.ConnectionFailed, req.userdata); }; return; } if (nread > 0 and client.h2_session != null) { _ = nghttp2.nghttp2_session_mem_recv(client.h2_session.?, @ptrCast(buf.base), @intCast(nread)); client.flush() catch {}; } }
fn onConnect(req: *uv.connect_t, status: c_int) callconv(.c) void { const ctx: *ConnectCtx = @ptrCast(@alignCast(req.data)); defer ctx.client.allocator.destroy(ctx); ctx.client.connect_pending = false; if (ctx.client.closing) { ctx.client.connected = -1; _ = tlsuv.tlsuv_stream_close(&ctx.client.tls, onStreamClose); return; } if (status < 0) { ctx.client.connected = -1; return; } ctx.client.connected = 1; ctx.client.tls.data = ctx.client; ctx.client.initH2() catch { ctx.client.connected = -1; return; }; _ = tlsuv.tlsuv_stream_read_start(&ctx.client.tls, allocBuf, onRead); ctx.client.flush() catch {}; }
const ConnectCtx = struct { client: *Http2Client, req: uv.connect_t };
fn ensureConnected(self: *Http2Client) !void { if (self.closing) return error.ConnectionFailed; if (self.connected > 0) return; if (self.connected < 0) return error.ConnectionFailed;
var conn_start: u64 = @intCast(std.time.nanoTimestamp());
if (!self.connect_pending) { const ctx = try self.allocator.create(ConnectCtx); ctx.* = .{ .client = self, .req = .{} }; ctx.req.data = ctx; if (tlsuv.tlsuv_stream_connect(&ctx.req, &self.tls, self.host.ptr, if (self.use_tls) 443 else 80, onConnect) != 0) { self.allocator.destroy(ctx); return error.ConnectionFailed; } self.connect_pending = true; }
var loop_count: u32 = 0; while (self.connected == 0) { _ = uv.uv_run(self.loop, uv.RUN_ONCE); loop_count += 1; } conn_start = debug.timer(" h2: tls connect", conn_start); debug.log(" h2: connect loop iterations={d}", .{loop_count}); if (self.connected < 0) return error.ConnectionFailed; }
pub fn initiateConnectAsync(self: *Http2Client) !void { if (self.closing) return error.ConnectionFailed; if (self.connected > 0) return; if (self.connected < 0) return error.ConnectionFailed; if (self.connect_pending) return;
const ctx = try self.allocator.create(ConnectCtx); ctx.* = .{ .client = self, .req = .{} }; ctx.req.data = ctx; if (tlsuv.tlsuv_stream_connect(&ctx.req, &self.tls, self.host.ptr, if (self.use_tls) 443 else 80, onConnect) != 0) { self.allocator.destroy(ctx); return error.ConnectionFailed; } self.connect_pending = true; }
fn makeNv(name: [:0]const u8, value: [:0]const u8) nghttp2.nv { return .{ .name = @constCast(name.ptr), .value = @constCast(value.ptr), .namelen = name.len, .valuelen = value.len, .flags = nghttp2.NV_FLAG_NONE }; }
pub fn get(self: *Http2Client, path: []const u8, allocator: std.mem.Allocator) ![]u8 { return self.getWithAccept(path, "application/json", allocator); }
pub fn getWithAccept(self: *Http2Client, path: []const u8, accept: [:0]const u8, allocator: std.mem.Allocator) ![]u8 { try self.ensureConnected(); if (self.request_count >= MAX_PENDING_REQUESTS) self.resetRequests(); const req = &self.requests[self.request_count]; self.request_count += 1; req.* = .{ .stream_id = 0, .path = try self.allocator.dupeZ(u8, path), .on_data = null, .on_complete = null, .on_error = null, .userdata = null, .response_body = .{}, .status_code = 0, .done = false, .has_error = false, .start_ns = @intCast(std.time.nanoTimestamp()), .end_ns = 0, .bytes = 0, .content_encoding = .identity, }; const session = self.h2_session orelse return error.Http2Error; var hdrs = [_]nghttp2.nv{ makeNv(":method", "GET"), makeNv(":path", req.path.?), makeNv(":scheme", "https"), makeNv(":authority", self.host), makeNv("accept", accept), makeNv("user-agent", user_agent) }; const sid = nghttp2.nghttp2_submit_request(session, null, &hdrs, hdrs.len, null, req); if (sid < 0) { self.request_count -= 1; if (req.path) |p| self.allocator.free(p); return error.Http2Error; } req.stream_id = sid; try self.flush(); while (!req.done) { _ = uv.uv_run(self.loop, uv.RUN_ONCE); try self.flush(); } self.last_response_status_code = req.status_code; if (req.has_error or req.status_code != 200) return error.ResponseError; return try allocator.dupe(u8, req.response_body.items); }
pub fn getStream(self: *Http2Client, path: []const u8, on_data: *const fn ([]const u8, ?*anyopaque) void, on_complete: *const fn (u16, ?*anyopaque) void, on_error: *const fn (FetchError, ?*anyopaque) void, userdata: ?*anyopaque) !void { try self.ensureConnected(); const req = self.findOrAllocSlot() orelse return error.OutOfMemory; req.* = .{ .stream_id = 0, .path = try self.allocator.dupeZ(u8, path), .on_data = on_data, .on_complete = on_complete, .on_error = on_error, .userdata = userdata, .response_body = .{}, .status_code = 0, .done = false, .has_error = false, .start_ns = @intCast(std.time.nanoTimestamp()), .end_ns = 0, .bytes = 0, .content_encoding = .identity };
const session = self.h2_session orelse return error.Http2Error; var hdrs = [_]nghttp2.nv{ makeNv(":method", "GET"), makeNv(":path", req.path.?), makeNv(":scheme", "https"), makeNv(":authority", self.host), makeNv("accept", "*/*"), makeNv("user-agent", user_agent) }; const sid = nghttp2.nghttp2_submit_request(session, null, &hdrs, hdrs.len, null, req); if (sid < 0) { if (req.path) |p| self.allocator.free(p); req.stream_id = -1; return error.Http2Error; } req.stream_id = sid; try self.flush(); }
pub fn run(self: *Http2Client) !void { const run_start: u64 = @intCast(std.time.nanoTimestamp()); var loop_count: u32 = 0; var last_done: usize = 0; var last_report: u64 = run_start;
while (self.requests_done < self.request_count) { if (uv.uv_run(self.loop, uv.RUN_ONCE) == 0) break; try self.flush(); loop_count += 1;
const now: u64 = @intCast(std.time.nanoTimestamp()); if (now - last_report > 1_000_000_000) { const done_delta = self.requests_done - last_done; debug.log(" h2: progress {d}/{d} (+{d} in last 1s) loops={d}", .{ self.requests_done, self.request_count, done_delta, loop_count, }); last_done = self.requests_done; last_report = now; } }
const elapsed_ns: u64 = @intCast(@as(i128, @intCast(std.time.nanoTimestamp())) - @as(i128, run_start)); const elapsed_ms = elapsed_ns / 1_000_000; debug.log(" h2: run complete in {d}ms, {d} loops, {d}/{d} done", .{ elapsed_ms, loop_count, self.requests_done, self.request_count, });
var error_count: usize = 0; for (self.requests[0..self.request_count]) |req| { if (req.has_error) error_count += 1; } if (error_count > 0) { debug.log(" h2: {d} requests had errors", .{error_count}); return error.ResponseError; } }};
pub const TarballCtx = struct { handler: StreamHandler, done: bool, has_error: bool, url: []const u8, start_ns: u64, bytes: usize,};
const TarballStats = struct { url: []const u8, bytes: usize, elapsed_ms: u64,};
const TarballCallbacks = struct { fn onData(data: []const u8, ud: ?*anyopaque) void { const ctx: *TarballCtx = @ptrCast(@alignCast(ud)); ctx.bytes += data.len; ctx.handler.on_data(data, ctx.handler.user_data); } fn onComplete(status: u16, ud: ?*anyopaque) void { const ctx: *TarballCtx = @ptrCast(@alignCast(ud)); ctx.handler.on_complete(status, ctx.handler.user_data); ctx.done = true; } fn onError(err: FetchError, ud: ?*anyopaque) void { const ctx: *TarballCtx = @ptrCast(@alignCast(ud)); ctx.handler.on_error(err, ctx.handler.user_data); ctx.done = true; ctx.has_error = true; }};
pub const Fetcher = struct { allocator: std.mem.Allocator, registry_host: []const u8, meta_clients: [NUM_META_CONNECTIONS]?*Http2Client, meta_clients_initialized: bool, pending: std.ArrayListUnmanaged(PendingRequest), tarball_clients: [NUM_CONNECTIONS]?*Http2Client, tarball_clients_initialized: bool, tarball_contexts: std.ArrayListUnmanaged(*TarballCtx), tarball_round_robin: usize, tarball_stats: std.ArrayListUnmanaged(TarballStats), last_http_error_url: ?[]u8, last_http_error_status: u16,
pub const HttpErrorInfo = struct { url: []const u8, status: u16, };
pub fn init(allocator: std.mem.Allocator, registry_host: []const u8) !*Fetcher { const f = try allocator.create(Fetcher); f.* = .{ .allocator = allocator, .registry_host = try allocator.dupe(u8, registry_host), .meta_clients = [_]?*Http2Client{null} ** NUM_META_CONNECTIONS, .meta_clients_initialized = false, .pending = .{}, .tarball_clients = [_]?*Http2Client{null} ** NUM_CONNECTIONS, .tarball_clients_initialized = false, .tarball_contexts = .{}, .tarball_round_robin = 0, .tarball_stats = .{}, .last_http_error_url = null, .last_http_error_status = 0, }; return f; }
pub fn deinit(self: *Fetcher) void { if (self.last_http_error_url) |url| self.allocator.free(url); for (&self.meta_clients) |*maybe_client| { if (maybe_client.*) |c| { c.deinit(); maybe_client.* = null; } } for (self.pending.items) |req| self.allocator.free(req.url); self.pending.deinit(self.allocator); for (&self.tarball_clients) |*maybe_client| { if (maybe_client.*) |c| { c.deinit(); maybe_client.* = null; } } for (self.tarball_contexts.items) |ctx| { self.allocator.free(ctx.url); self.allocator.destroy(ctx); } self.tarball_contexts.deinit(self.allocator); for (self.tarball_stats.items) |stat| self.allocator.free(stat.url); self.tarball_stats.deinit(self.allocator); self.allocator.free(self.registry_host); self.allocator.destroy(self); }
fn clearLastHttpError(self: *Fetcher) void { if (self.last_http_error_url) |url| self.allocator.free(url); self.last_http_error_url = null; self.last_http_error_status = 0; }
fn setLastHttpError(self: *Fetcher, url: []const u8, status: u16) void { self.clearLastHttpError(); self.last_http_error_url = self.allocator.dupe(u8, url) catch null; self.last_http_error_status = status; }
pub fn getLastHttpError(self: *const Fetcher) ?HttpErrorInfo { const url = self.last_http_error_url orelse return null; return .{ .url = url, .status = self.last_http_error_status }; }
fn ensureMetaClients(self: *Fetcher) !void { if (self.meta_clients_initialized) return;
for (&self.meta_clients, 0..) |*slot, i| { const client = Http2Client.init(self.allocator, self.registry_host, true) catch |err| { debug.log("fetcher: failed to init meta connection {d}: {}", .{ i, err }); continue; }; client.ensureConnected() catch |err| { debug.log("fetcher: failed to connect meta {d}: {}", .{ i, err }); client.deinit(); continue; }; slot.* = client; }
var any_connected = false; for (self.meta_clients) |slot| { if (slot != null) { any_connected = true; break; } } if (!any_connected) return error.ConnectionFailed; self.meta_clients_initialized = true; }
pub fn resetMetaClients(self: *Fetcher) void { self.clearLastHttpError(); for (&self.meta_clients) |*slot| { if (slot.*) |client| { client.deinit(); slot.* = null; } } self.meta_clients_initialized = false; }
fn ensureTarballClients(self: *Fetcher) !void { if (self.tarball_clients_initialized) return;
debug.log("fetcher: initializing {d} persistent connections", .{NUM_CONNECTIONS}); const init_start: u64 = @intCast(std.time.nanoTimestamp());
for (&self.tarball_clients, 0..) |*slot, i| { const client = Http2Client.init(self.allocator, self.registry_host, true) catch |err| { debug.log("fetcher: failed to init connection {d}: {}", .{ i, err }); continue; }; client.ensureConnected() catch |err| { debug.log("fetcher: failed to connect {d}: {}", .{ i, err }); client.deinit(); continue; }; slot.* = client; }
var any_connected = false; for (self.tarball_clients) |slot| { if (slot != null) { any_connected = true; break; } }
if (!any_connected) return error.ConnectionFailed; self.tarball_clients_initialized = true; _ = debug.timer("fetcher: connection pool init", init_start); } fn findAvailableClient(self: *Fetcher) ?struct { client: *Http2Client, idx: usize } { var attempts: usize = 0; while (attempts < NUM_CONNECTIONS) : (attempts += 1) { const idx = (self.tarball_round_robin + attempts) % NUM_CONNECTIONS; if (self.tarball_clients[idx]) |client| { if (client.hasCapacity()) return .{ .client = client, .idx = idx }; } } return null; }
pub fn initiateTarballConnectionsAsync(self: *Fetcher) void { if (self.tarball_clients_initialized) return; debug.log("fetcher: initiating {d} tarball connections (async)", .{NUM_CONNECTIONS});
for (&self.tarball_clients, 0..) |*slot, i| { const client = Http2Client.init(self.allocator, self.registry_host, true) catch { continue; }; client.initiateConnectAsync() catch { client.deinit(); continue; }; slot.* = client; _ = i; }
var any_connected = false; for (self.tarball_clients) |slot| { if (slot != null) { any_connected = true; break; } }
if (any_connected) self.tarball_clients_initialized = true; }
pub fn queueTarballAsync(self: *Fetcher, url: []const u8, handler: StreamHandler) !void { try self.ensureTarballClients(); const parsed = try ParsedUrl.parse(url); const available = self.findAvailableClient() orelse { try self.pending.append(self.allocator, .{ .url = try self.allocator.dupe(u8, url), .handler = handler, }); return; }; const ctx = try self.allocator.create(TarballCtx); ctx.* = .{ .handler = handler, .done = false, .has_error = false, .url = try self.allocator.dupe(u8, url), .start_ns = @intCast(std.time.nanoTimestamp()), .bytes = 0, }; try self.tarball_contexts.append( self.allocator, ctx );
try available.client.getStream( parsed.path, TarballCallbacks.onData, TarballCallbacks.onComplete, TarballCallbacks.onError, ctx, ); self.tarball_round_robin = (available.idx + 1) % NUM_CONNECTIONS; }
pub fn tick(self: *Fetcher) usize { self.ensureTarballClients() catch return 0; const loop = uv.uv_default_loop(); for (&self.tarball_clients) |maybe_client| { if (maybe_client) |c| c.flush() catch {}; } _ = uv.uv_run(loop, uv.RUN_NOWAIT); for (&self.tarball_clients) |maybe_client| { if (maybe_client) |c| c.recycleCompletedRequests(); } const completed = self.cleanupCompletedContexts(); self.dispatchPending(); return completed; } fn cleanupCompletedContexts(self: *Fetcher) usize { var completed: usize = 0; var i: usize = 0; while (i < self.tarball_contexts.items.len) { const ctx = self.tarball_contexts.items[i]; if (ctx.done) { completed += 1; self.allocator.free(ctx.url); self.allocator.destroy(ctx); _ = self.tarball_contexts.swapRemove(i); } else i += 1; } return completed; } fn dispatchPending(self: *Fetcher) void { while (self.pending.items.len > 0) { const available = self.findAvailableClient() orelse break; const req = self.pending.pop() orelse break; const handler = req.handler orelse { self.allocator.free(req.url); continue; }; self.dispatchRequest(available.client, req.url, handler) catch |err| { handler.on_error(errToFetchError(err), handler.user_data); self.allocator.free(req.url); continue; }; } } fn dispatchRequest(self: *Fetcher, client: *Http2Client, url: []const u8, handler: StreamHandler) !void { const parsed = try ParsedUrl.parse(url); const ctx = try self.allocator.create(TarballCtx); ctx.* = .{ .handler = handler, .done = false, .has_error = false, .url = url, .start_ns = @intCast(std.time.nanoTimestamp()), .bytes = 0, }; errdefer self.allocator.destroy(ctx); try self.tarball_contexts.append(self.allocator, ctx); try client.getStream( parsed.path, TarballCallbacks.onData, TarballCallbacks.onComplete, TarballCallbacks.onError, ctx, ); } fn errToFetchError(err: anyerror) FetchError { return switch (err) { error.InvalidUrl => FetchError.InvalidUrl, error.OutOfMemory => FetchError.OutOfMemory, else => FetchError.Http2Error, }; } pub fn pendingTarballCount(self: *Fetcher) usize { return self.tarball_contexts.items.len; }
pub fn finishTarballs(self: *Fetcher) void { const loop = uv.uv_default_loop(); var last_report: u64 = @intCast(std.time.nanoTimestamp()); var loops: usize = 0; var completed: usize = 0; const start = last_report;
while (self.tarball_contexts.items.len > 0 or self.pending.items.len > 0) { for (&self.tarball_clients) |maybe_client| { if (maybe_client) |c| c.flush() catch {}; }
if (uv.uv_run(loop, uv.RUN_ONCE) == 0 and self.pending.items.len == 0 and self.tarball_contexts.items.len == 0) break; loops += 1;
for (&self.tarball_clients) |maybe_client| { if (maybe_client) |c| c.recycleCompletedRequests(); }
var i: usize = 0; while (i < self.tarball_contexts.items.len) { const ctx = self.tarball_contexts.items[i]; if (ctx.done) { if (!ctx.has_error) { const elapsed_ms: u64 = @intCast((@as(u64, @intCast(std.time.nanoTimestamp())) - ctx.start_ns) / 1_000_000); const url_copy = self.allocator.dupe(u8, ctx.url) catch null; if (url_copy) |url| { self.tarball_stats.append(self.allocator, .{ .url = url, .bytes = ctx.bytes, .elapsed_ms = elapsed_ms }) catch {}; } } self.allocator.free(ctx.url); self.allocator.destroy(ctx); _ = self.tarball_contexts.swapRemove(i); completed += 1; } else { i += 1; } }
while (self.pending.items.len > 0) { var queued = false; for (&self.tarball_clients, 0..) |maybe_client, conn_idx| { if (maybe_client) |client| { if (client.hasCapacity()) { const maybe_req = self.pending.pop(); const req = maybe_req orelse break; if (req.handler) |handler| { const parsed = ParsedUrl.parse(req.url) catch { handler.on_error(FetchError.InvalidUrl, handler.user_data); self.allocator.free(req.url); continue; };
const ctx = self.allocator.create(TarballCtx) catch { handler.on_error(FetchError.OutOfMemory, handler.user_data); self.allocator.free(req.url); continue; }; ctx.* = .{ .handler = handler, .done = false, .has_error = false, .url = req.url, .start_ns = @intCast(std.time.nanoTimestamp()), .bytes = 0, }; self.tarball_contexts.append(self.allocator, ctx) catch { self.allocator.destroy(ctx); self.allocator.free(req.url); continue; };
client.getStream( parsed.path, struct { fn onData(data: []const u8, ud: ?*anyopaque) void { const c: *TarballCtx = @ptrCast(@alignCast(ud)); c.bytes += data.len; c.handler.on_data(data, c.handler.user_data); } }.onData, struct { fn onComplete(status: u16, ud: ?*anyopaque) void { const c: *TarballCtx = @ptrCast(@alignCast(ud)); c.handler.on_complete(status, c.handler.user_data); if (debug.enabled) { const elapsed_ms: u64 = @intCast((@as(u64, @intCast(std.time.nanoTimestamp())) - c.start_ns) / 1_000_000); debug.log(" tarball: done {s} {d}ms {d} bytes status={d}", .{ c.url, elapsed_ms, c.bytes, status }); } c.done = true; } }.onComplete, struct { fn onError(err: FetchError, ud: ?*anyopaque) void { const c: *TarballCtx = @ptrCast(@alignCast(ud)); c.handler.on_error(err, c.handler.user_data); if (debug.enabled) { const elapsed_ms: u64 = @intCast((@as(u64, @intCast(std.time.nanoTimestamp())) - c.start_ns) / 1_000_000); debug.log(" tarball: error {s} {d}ms {d} bytes", .{ c.url, elapsed_ms, c.bytes }); } c.done = true; c.has_error = true; } }.onError, ctx, ) catch { handler.on_error(FetchError.Http2Error, handler.user_data); ctx.done = true; }; queued = true; _ = conn_idx; } else { self.allocator.free(req.url); } break; } } } if (!queued) break; }
const now: u64 = @intCast(std.time.nanoTimestamp()); if (now - last_report > 1_000_000_000) { var total_bytes: usize = 0; for (self.tarball_contexts.items) |ctx| { total_bytes += ctx.bytes; } debug.log(" h2: {d} in-flight, {d} pending, {d} completed, {d} loops", .{ self.tarball_contexts.items.len, self.pending.items.len, completed, loops, }); debug.log(" h2: tarball progress in-flight bytes={d}", .{ total_bytes }); last_report = now; } }
const elapsed_ns: u64 = @intCast(@as(i128, @intCast(std.time.nanoTimestamp())) - @as(i128, start)); debug.log("fetcher: finishTarballs completed in {d}ms, {d} loops, {d} completed", .{ elapsed_ns / 1_000_000, loops, completed, }); if (debug.enabled and self.tarball_stats.items.len > 0) { var top_time: [5]?TarballStats = .{null} ** 5; var top_size: [5]?TarballStats = .{null} ** 5;
for (self.tarball_stats.items) |stat| { var idx_time: usize = top_time.len; for (top_time, 0..) |slot, i| { if (slot == null or stat.elapsed_ms > slot.?.elapsed_ms) { idx_time = i; break; } } if (idx_time < top_time.len) { var carry = stat; var j = idx_time; while (j < top_time.len) : (j += 1) { const next = top_time[j]; top_time[j] = carry; if (next) |n| { carry = n; } else { break; } } }
var idx_size: usize = top_size.len; for (top_size, 0..) |slot, i| { if (slot == null or stat.bytes > slot.?.bytes) { idx_size = i; break; } } if (idx_size < top_size.len) { var carry_size = stat; var k = idx_size; while (k < top_size.len) : (k += 1) { const next_size = top_size[k]; top_size[k] = carry_size; if (next_size) |n| { carry_size = n; } else { break; } } } }
debug.log("fetcher: top tarballs by time", .{}); for (top_time, 0..) |maybe_stat, i| { if (maybe_stat) |stat| { debug.log(" {d}. {s} {d}ms {d} bytes", .{ i + 1, stat.url, stat.elapsed_ms, stat.bytes }); } } debug.log("fetcher: top tarballs by size", .{}); for (top_size, 0..) |maybe_stat, i| { if (maybe_stat) |stat| { debug.log(" {d}. {s} {d} bytes {d}ms", .{ i + 1, stat.url, stat.bytes, stat.elapsed_ms }); } } } }
pub fn fetchMetadata(self: *Fetcher, package_name: []const u8, allocator: std.mem.Allocator) ![]u8 { return self.fetchMetadataFull(package_name, false, allocator); }
const DecodedMetadata = struct { data: []u8, compressed: bool, };
fn metaClientCanQueue(c: *const Http2Client) bool { return c.h2_session != null and c.connected == 1 and c.request_count < MAX_PENDING_REQUESTS - 1; }
fn nextMetaClient(self: *Fetcher, conn_idx: *usize) ?*Http2Client { var attempts: usize = 0; while (attempts < NUM_META_CONNECTIONS) : (attempts += 1) { if (self.meta_clients[conn_idx.*]) |c| { if (metaClientCanQueue(c)) return c; } conn_idx.* = (conn_idx.* + 1) % NUM_META_CONNECTIONS; } return null; }
fn flushMetaClients(self: *Fetcher) void { for (self.meta_clients) |maybe_client| { if (maybe_client) |c| c.flush() catch {}; } }
fn metaRequestsComplete(self: *Fetcher) bool { for (self.meta_clients) |maybe_client| { if (maybe_client) |c| { for (c.requests[0..c.request_count]) |*req| { if (!req.done and !req.has_error) return false; } } } return true; }
fn decodeMetadataOwned( req: *Http2Client.RequestState, allocator: std.mem.Allocator, decompress_buf: *std.ArrayListUnmanaged(u8), ) ?DecodedMetadata { if (req.has_error or req.status_code != 200) return null;
if (req.content_encoding != .gzip) { const data = allocator.dupe(u8, req.response_body.items) catch return null; return .{ .data = data, .compressed = false }; }
decompress_buf.clearRetainingCapacity(); const decomp = extractor.GzipDecompressor.init(allocator) catch return null; defer decomp.deinit();
_ = decomp.decompress(req.response_body.items, struct { fn onChunk(data: []const u8, ctx: ?*anyopaque) anyerror!void { const buf: *std.ArrayListUnmanaged(u8) = @ptrCast(@alignCast(ctx)); try buf.appendSlice(c_allocator, data); } }.onChunk, decompress_buf) catch return null;
const data = allocator.dupe(u8, decompress_buf.items) catch return null; return .{ .data = data, .compressed = true }; }
pub fn fetchMetadataFull(self: *Fetcher, package_name: []const u8, full: bool, allocator: std.mem.Allocator) ![]u8 { try self.ensureMetaClients(); self.clearLastHttpError(); for (self.meta_clients) |maybe_client| { if (maybe_client) |client| { var path_buf: [512]u8 = undefined; const path_slice = std.fmt.bufPrint(&path_buf, "/{s}", .{package_name}) catch return error.OutOfMemory; const accept: [:0]const u8 = if (full) "application/json" else "application/vnd.npm.install-v1+json"; return client.getWithAccept(path_slice, accept, allocator) catch |err| { if (err == error.ResponseError) { var url_buf: [1024]u8 = undefined; const url = std.fmt.bufPrint(&url_buf, "https://{s}/{s}", .{ self.registry_host, package_name }) catch ""; self.setLastHttpError(url, client.last_response_status_code); } return err; }; } } return error.ConnectionFailed; }
pub const MetadataResult = struct { name: []const u8, data: ?[]u8, compressed: bool, has_error: bool, };
fn storeMetadataBatchResult( req: *Http2Client.RequestState, result: *MetadataResult, allocator: std.mem.Allocator, decompress_buf: *std.ArrayListUnmanaged(u8), ) bool { const decoded = decodeMetadataOwned(req, allocator, decompress_buf) orelse { result.has_error = true; return false; }; result.data = decoded.data; result.compressed = decoded.compressed; return true; }
pub fn fetchMetadataBatch(self: *Fetcher, names: []const []const u8, allocator: std.mem.Allocator) ![]MetadataResult { if (names.len == 0) return &[_]MetadataResult{};
var total_start: u64 = @intCast(std.time.nanoTimestamp()); try self.ensureMetaClients(); total_start = debug.timer(" meta: get clients", total_start);
var active_connections: usize = 0; for (self.meta_clients) |maybe_client| { if (maybe_client != null) active_connections += 1; } if (active_connections == 0) return error.ConnectionFailed;
debug.log(" meta: batch {d} packages across {d} connections", .{ names.len, active_connections });
var results = try allocator.alloc(MetadataResult, names.len); for (results, 0..) |*r, i| { r.* = .{ .name = names[i], .data = null, .compressed = false, .has_error = false }; }
const total_capacity = active_connections * (MAX_PENDING_REQUESTS - 1); var offset: usize = 0; var batch_num: usize = 0;
var decompress_buf = std.ArrayListUnmanaged(u8){}; defer decompress_buf.deinit(c_allocator);
while (offset < names.len) { const end = @min(offset + total_capacity, names.len); var batch_start: u64 = @intCast(std.time.nanoTimestamp()); debug.log(" meta: batch {d} ({d}-{d})", .{ batch_num, offset, end });
var queued: usize = 0; var conn_idx: usize = 0; for (offset..end) |i| { const result = &results[i]; const name = names[i];
const c = self.nextMetaClient(&conn_idx) orelse { result.has_error = true; continue; }; const session = c.h2_session orelse { result.has_error = true; continue; }; var path_buf: [512]u8 = undefined; const path = std.fmt.bufPrint(&path_buf, "/{s}", .{name}) catch { result.has_error = true; continue; };
var hdrs = [_]nghttp2.nv{ Http2Client.makeNv(":method", "GET"), Http2Client.makeNv(":path", c.allocator.dupeZ(u8, path) catch { result.has_error = true; continue; }), Http2Client.makeNv(":scheme", "https"), Http2Client.makeNv(":authority", c.host), Http2Client.makeNv("accept", "application/vnd.npm.install-v1+json"), Http2Client.makeNv("accept-encoding", "gzip"), Http2Client.makeNv("user-agent", user_agent), };
const req = &c.requests[c.request_count]; c.request_count += 1; req.* = .{ .stream_id = 0, .path = hdrs[1].value[0..hdrs[1].valuelen :0], .on_data = null, .on_complete = null, .on_error = null, .userdata = result, .response_body = .{}, .status_code = 0, .done = false, .has_error = false, .start_ns = @intCast(std.time.nanoTimestamp()), .end_ns = 0, .bytes = 0, .content_encoding = .identity, };
const sid = nghttp2.nghttp2_submit_request(session, null, &hdrs, hdrs.len, null, req); if (sid < 0) { c.request_count -= 1; if (req.path) |p| c.allocator.free(p); result.has_error = true; continue; } req.stream_id = sid; queued += 1; conn_idx = (conn_idx + 1) % NUM_META_CONNECTIONS; } batch_start = debug.timer(" meta: queue requests", batch_start); self.flushMetaClients();
const loop = uv.uv_default_loop(); var all_done = false; var loops: usize = 0; const run_start: u64 = @intCast(std.time.nanoTimestamp());
while (!all_done) { _ = uv.uv_run(loop, uv.RUN_ONCE); loops += 1; all_done = self.metaRequestsComplete(); }
const elapsed_ns: u64 = @intCast(@as(i128, @intCast(std.time.nanoTimestamp())) - @as(i128, run_start)); debug.log(" h2: run complete in {d}ms, {d} loops", .{ elapsed_ns / 1_000_000, loops }); batch_start = debug.timer(" meta: run h2 loop", batch_start);
var slow_count: usize = 0; var max_req_ms: u64 = 0; var max_req_name: []const u8 = ""; var total_bytes: usize = 0; for (self.meta_clients) |maybe_client| { if (maybe_client) |c| { for (c.requests[0..c.request_count]) |*req| { const result: *MetadataResult = @ptrCast(@alignCast(req.userdata)); const end_ns = if (req.end_ns != 0) req.end_ns else @as(u64, @intCast(std.time.nanoTimestamp())); const duration_ms: u64 = @intCast((end_ns - req.start_ns) / 1_000_000); total_bytes += req.response_body.items.len; if (duration_ms > max_req_ms) { max_req_ms = duration_ms; max_req_name = result.name; } if (duration_ms >= META_SLOW_LOG_MS) { slow_count += 1; debug.log(" meta: slow {s} {d}ms {d} bytes status={d}", .{ result.name, duration_ms, req.response_body.items.len, req.status_code, }); } } } } debug.log(" meta: summary slow={d} max={s} {d}ms total_bytes={d}", .{ slow_count, max_req_name, max_req_ms, total_bytes });
var success: usize = 0; for (self.meta_clients) |maybe_client| { if (maybe_client) |c| { for (c.requests[0..c.request_count]) |*req| { const result: *MetadataResult = @ptrCast(@alignCast(req.userdata)); if (storeMetadataBatchResult(req, result, allocator, &decompress_buf)) success += 1; } c.resetRequests(); } } _ = debug.timer(" meta: copy results", batch_start); debug.log(" meta: queued={d} success={d}", .{ queued, success });
offset = end; batch_num += 1; }
return results; }
pub const MetadataCallback = *const fn ( name: []const u8, data: ?[]const u8, has_error: bool, userdata: ?*anyopaque ) void;
const MetadataStreamTracker = struct { name: []const u8, index: usize, };
fn emitMetadataStreamingResult( req: *Http2Client.RequestState, name: []const u8, allocator: std.mem.Allocator, decompress_buf: *std.ArrayListUnmanaged(u8), callback: MetadataCallback, userdata: ?*anyopaque, ) void { if (req.has_error or req.status_code != 200) { callback(name, null, true, userdata); return; }
if (req.content_encoding != .gzip) { callback(name, req.response_body.items, false, userdata); return; }
decompress_buf.clearRetainingCapacity(); const decomp = extractor.GzipDecompressor.init(allocator) catch { callback(name, null, true, userdata); return; }; defer decomp.deinit();
_ = decomp.decompress(req.response_body.items, struct { fn onChunk(data: []const u8, ctx: ?*anyopaque) anyerror!void { const buf: *std.ArrayListUnmanaged(u8) = @ptrCast(@alignCast(ctx)); try buf.appendSlice(c_allocator, data); } }.onChunk, decompress_buf) catch { callback(name, null, true, userdata); return; };
callback(name, decompress_buf.items, false, userdata); }
fn emitCompletedStreamingMetadataCallbacks( self: *Fetcher, processed: []bool, allocator: std.mem.Allocator, decompress_buf: *std.ArrayListUnmanaged(u8), callback: MetadataCallback, userdata: ?*anyopaque, ) void { for (self.meta_clients) |maybe_client| { const c = maybe_client orelse continue; for (c.requests[0..c.request_count]) |*req| { if (!req.done and !req.has_error) continue;
const tracker: *MetadataStreamTracker = @ptrCast(@alignCast(req.userdata)); if (processed[tracker.index]) continue;
processed[tracker.index] = true; emitMetadataStreamingResult(req, tracker.name, allocator, decompress_buf, callback, userdata); } } }
pub fn fetchMetadataStreaming( self: *Fetcher, names: []const []const u8, allocator: std.mem.Allocator, callback: MetadataCallback, userdata: ?*anyopaque, ) !void { if (names.len == 0) return;
var total_start: u64 = @intCast(std.time.nanoTimestamp()); try self.ensureMetaClients(); total_start = debug.timer(" meta: get clients", total_start);
var active_connections: usize = 0; for (self.meta_clients) |maybe_client| { if (maybe_client != null) active_connections += 1; } if (active_connections == 0) return error.ConnectionFailed; debug.log(" meta: streaming {d} packages across {d} connections", .{ names.len, active_connections });
const processed = try allocator.alloc(bool, names.len); defer allocator.free(processed); @memset(processed, false);
var trackers = try allocator.alloc(MetadataStreamTracker, names.len); defer allocator.free(trackers); for (names, 0..) |name, i| { trackers[i] = .{ .name = name, .index = i }; }
const total_capacity = active_connections * (MAX_PENDING_REQUESTS - 1); var offset: usize = 0; var batch_num: usize = 0;
var decompress_buf = std.ArrayListUnmanaged(u8){}; defer decompress_buf.deinit(c_allocator);
while (offset < names.len) { const end = @min(offset + total_capacity, names.len); var batch_start: u64 = @intCast(std.time.nanoTimestamp());
debug.log(" meta: batch {d} ({d}-{d})", .{ batch_num, offset, end });
var queued: usize = 0; var conn_idx: usize = 0; for (offset..end) |i| { const tracker = &trackers[i]; const name = names[i];
const c = self.nextMetaClient(&conn_idx) orelse continue; const session = c.h2_session orelse continue; var path_buf: [512]u8 = undefined; const path = std.fmt.bufPrint(&path_buf, "/{s}", .{name}) catch continue;
var hdrs = [_]nghttp2.nv{ Http2Client.makeNv(":method", "GET"), Http2Client.makeNv(":path", c.allocator.dupeZ(u8, path) catch continue), Http2Client.makeNv(":scheme", "https"), Http2Client.makeNv(":authority", c.host), Http2Client.makeNv("accept", "application/vnd.npm.install-v1+json"), Http2Client.makeNv("accept-encoding", "gzip"), Http2Client.makeNv("user-agent", user_agent), };
const req = &c.requests[c.request_count]; c.request_count += 1; req.* = .{ .stream_id = 0, .path = hdrs[1].value[0..hdrs[1].valuelen :0], .on_data = null, .on_complete = null, .on_error = null, .userdata = tracker, .response_body = .{}, .status_code = 0, .done = false, .has_error = false, .start_ns = 0, .end_ns = 0, .bytes = 0, .content_encoding = .identity, }; req.start_ns = @intCast(std.time.nanoTimestamp());
const sid = nghttp2.nghttp2_submit_request(session, null, &hdrs, hdrs.len, null, req); if (sid < 0) { c.request_count -= 1; if (req.path) |p| c.allocator.free(p); continue; } req.stream_id = sid; queued += 1; conn_idx = (conn_idx + 1) % NUM_META_CONNECTIONS; } batch_start = debug.timer(" meta: queue requests", batch_start); self.flushMetaClients();
const loop = uv.uv_default_loop(); var all_done = false; var loops: usize = 0; const run_start: u64 = @intCast(std.time.nanoTimestamp());
while (!all_done) { _ = uv.uv_run(loop, uv.RUN_ONCE); loops += 1; self.emitCompletedStreamingMetadataCallbacks(processed, allocator, &decompress_buf, callback, userdata);
all_done = self.metaRequestsComplete(); }
const elapsed_ns: u64 = @intCast(@as(i128, @intCast(std.time.nanoTimestamp())) - @as(i128, run_start)); debug.log(" h2: run complete in {d}ms, {d} loops", .{ elapsed_ns / 1_000_000, loops });
var slow_count: usize = 0; var max_req_ms: u64 = 0; var max_req_name: []const u8 = ""; var total_bytes: usize = 0; for (self.meta_clients) |maybe_client| { if (maybe_client) |c| { for (c.requests[0..c.request_count]) |*req| { const tracker: *MetadataStreamTracker = @ptrCast(@alignCast(req.userdata)); const end_ns = if (req.end_ns != 0) req.end_ns else @as(u64, @intCast(std.time.nanoTimestamp())); const duration_ms: u64 = @intCast((end_ns - req.start_ns) / 1_000_000); total_bytes += req.response_body.items.len; if (duration_ms > max_req_ms) { max_req_ms = duration_ms; max_req_name = tracker.name; } if (duration_ms >= META_SLOW_LOG_MS) { slow_count += 1; debug.log(" meta: slow {s} {d}ms {d} bytes status={d}", .{ tracker.name, duration_ms, req.response_body.items.len, req.status_code, }); } } } } debug.log(" meta: summary slow={d} max={s} {d}ms total_bytes={d}", .{ slow_count, max_req_name, max_req_ms, total_bytes });
for (self.meta_clients) |maybe_client| { if (maybe_client) |c| c.resetRequests(); }
offset = end; batch_num += 1; } }
pub fn fetchTarball(self: *Fetcher, url: []const u8, handler: StreamHandler) !void { try self.pending.append(self.allocator, .{ .url = try self.allocator.dupe(u8, url), .handler = handler }); }
pub fn run(self: *Fetcher) !void { if (self.pending.items.len == 0 and self.tarball_contexts.items.len == 0) return;
const run_start: u64 = @intCast(std.time.nanoTimestamp()); const total_requests = self.pending.items.len + self.tarball_contexts.items.len; debug.log("fetcher: {d} tarballs to download (pending={d}, in-flight={d})", .{ total_requests, self.pending.items.len, self.tarball_contexts.items.len, });
try self.ensureTarballClients(); self.finishTarballs();
const elapsed_ns: u64 = @intCast(@as(i128, @intCast(std.time.nanoTimestamp())) - @as(i128, run_start)); debug.log("fetcher: {d} tarballs complete in {d}ms", .{ total_requests, elapsed_ns / 1_000_000 }); }};