authorgravatar for me@paperdave.netdave caruso <me@paperdave.net> 2024-09-01 22:40:22-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2024-09-01 22:40:22-07:00
log4a2887aab0628d3b5b834049ea398d98f2d5549c
tree60d59a1c9c10468ed0f7b686a8ec96957e6e3fe0
parent516355984fb4d9adeb15cca20b91da46bb4d263a
signature Signed by PGP key B5690EEEBB952194

chore: implement initial `loader` module (#87)

The `loader` module uses a dedicated thread to load URLs for code and assets. `Loader` uses its own thread, and can concurrently manage downloads, which the main compiler can coordinate completions using a callback as well as a "Module Index" (not to be serialized). Loader features a cache mechanism, file URL support, as well as proper error propogation.

3 files changed, 1149 insertions(+), 2 deletions(-)

build.zig+18-2
......@@ -66,10 +66,11 @@ pub fn build(b: *Build) !void {
6666 const native_debug = buildWithTarget(b, b.graph.host, .Debug, .none, version, no_llvm);
6767 const main = buildWithTarget(b, target, optimize, tracing, version, no_llvm);
6868
69 b.installArtifact(main.cli(b, BuildConfig.all));
69 b.installArtifact(main.cli(b, .all));
7070
7171 inline for (.{
72 .{ main.cli(b, BuildConfig.minimal), "compiler" },
72 .{ main.cli(b, .minimal), "compiler" },
73 .{ main.loader_bin, "loader" },
7374 }) |entry| {
7475 const artifact, const name = entry;
7576
......@@ -79,6 +80,7 @@ pub fn build(b: *Build) !void {
7980 }
8081 const run_step = b.step(name, "Run " ++ name ++ " binary");
8182 run_step.dependOn(&run_cmd.step);
83 run_step.dependOn(&b.addInstallArtifact(artifact, .{}).step);
8284 }
8385
8486 const test_all = b.step("test", "Run all tests");
......@@ -195,6 +197,7 @@ const Artifacts = struct {
195197 core_tests: *Build.Step.Compile,
196198 opts_mod: *Build.Module,
197199 behavior_tests: *Build.Step.Compile,
200 loader_bin: *Build.Step.Compile,
198201
199202 /// Construct a CLI with certain features.
200203 /// When developing on a small component, such as the compiler core, you
......@@ -253,6 +256,7 @@ fn buildWithTarget(
253256 // Options
254257 const opts = b.addOptions();
255258 opts.addOption(std.SemanticVersion, "version", version);
259 opts.addOption([]const u8, "version_string", b.fmt("{}", .{version}));
256260 opts.addOption([]const u8, "tracing", @tagName(tracing));
257261 const opts_mod = opts.createModule();
258262
......@@ -280,6 +284,17 @@ fn buildWithTarget(
280284 core_tests.root_module.addImport("tracer", tracer.module("tracer"));
281285 core_tests.root_module.addImport("zg:PropsData", zg.module("PropsData"));
282286
287 // Loader
288 const loader = b.addExecutable(.{
289 .name = "loader",
290 .root_source_file = b.path("src/loader/main.zig"),
291 .target = target,
292 .optimize = optimize,
293 .use_lld = use_lld,
294 .use_llvm = use_llvm,
295 .strip = false,
296 });
297
283298 const behavior_tests = b.addExecutable(.{
284299 .name = "run_behavior_tests",
285300 .root_source_file = b.path("test/behavior.zig"),
......@@ -302,6 +317,7 @@ fn buildWithTarget(
302317 .core_tests = core_tests,
303318 .opts_mod = opts_mod,
304319 .behavior_tests = behavior_tests,
320 .loader_bin = loader,
305321 };
306322}
307323
src/loader/Loader.zig created+1096
......@@ -0,0 +1,1096 @@
1//! The `Loader` is responsible for resolving, fetching and caching files from
2//! URL sources. `Loader` is designed to run on it's own thread (the "Loader
3//! Thread"), and communicate with the compiler core via the `enqueue` function
4//! and the `completion_cb` callback.
5//!
6//! Note that this subsystem is entirely self-contained in the `loader` module,
7//! and is an optional dependency of the compiler. This allows testing the
8//! compiler without network determinism, and stress+fuzz testing `Loader`.
9//!
10//! The loader's thread-safe allocator owns source file contents.
11
12/// Shared http client. This also stores the Loader's allocator
13http: std.http.Client,
14
15/// This map is accessed by multiple threads and is covered by `entries_mutex`
16entries: std.StringArrayHashMapUnmanaged(Module),
17entries_lock: Thread.RwLock,
18
19/// Pre-allocated up to the max concurrent requests. This is
20/// allocated at thread start up, and kept alive forever.
21ongoing: OngoingLoad.List.Slice,
22
23/// Stores the highest index in `ongoing` that is currently active.
24/// If this is `0`, then there are no ongoing downloads.
25last_ongoing: u16,
26
27/// Thread-safe queue for pending tasks. They are moved
28/// to `ongoing_downloads` when they initiate.
29queue: Task.Queue,
30
31/// External callback when a fetch completes. Called on the Loader thread.
32completion_ctx: ?*anyopaque,
33completion_cb: ?*const anyopaque, // cannot use CompletionCallback because dependency loop
34
35/// This is the parent progress node (or the program root node). This is
36/// distinct from `progress` because the latter may be stopped and started as
37/// the thread sleeps (when there are no pending downloads)
38root_progress: std.Progress.Node,
39progress: std.Progress.Node,
40
41cache: ?std.fs.Dir,
42
43/// To ensure run is not being used on more than one thread.
44run_lock: std.debug.SafetyLock,
45
46redirect_behavior: std.http.Client.Request.RedirectBehavior,
47
48pub const LoaderOptions = struct {
49 allocator: Allocator,
50 max_concurrent_downloads: u16 = 32,
51 cache: ?std.fs.Dir = null,
52 /// Pass the parent progress node
53 progress: std.Progress.Node = .{ .index = .none },
54 completion_cb: ?*const CompletionCallback = null,
55 completion_ctx: ?*anyopaque = null,
56 redirect_behavior: std.http.Client.Request.RedirectBehavior = @enumFromInt(3),
57};
58
59pub fn init(options: LoaderOptions) !Loader {
60 var ongoing = OngoingLoad.List{};
61 try ongoing.setCapacity(options.allocator, options.max_concurrent_downloads);
62 ongoing.len = options.max_concurrent_downloads;
63 errdefer ongoing.deinit(options.allocator);
64 const ongoing_slice = ongoing.slice();
65
66 @memset(ongoing_slice.items(.state), .unused);
67 @memset(ongoing_slice.items(.buffer), .{});
68
69 return .{
70 .http = .{ .allocator = options.allocator },
71 .root_progress = options.progress,
72 .ongoing = ongoing_slice,
73 .completion_ctx = options.completion_ctx,
74 .completion_cb = @ptrCast(options.completion_cb),
75 .redirect_behavior = options.redirect_behavior,
76 .queue = .{},
77 .progress = .{ .index = .none },
78 .entries = .{},
79 .entries_lock = .{},
80 .run_lock = .{},
81 .last_ongoing = 0,
82 .cache = options.cache,
83 };
84}
85
86pub fn deinit(loader: *Loader) void {
87 loader.entries_lock.lock();
88
89 const entries = loader.entries.entries.slice();
90 for (entries.items(.key), entries.items(.value)) |k, v| {
91 loader.allocator().free(k);
92 v.deinit(loader.allocator());
93 }
94 loader.entries.deinit(loader.allocator());
95
96 for (
97 0..,
98 loader.ongoing.items(.state)[0..loader.last_ongoing],
99 ) |i, state| {
100 if (state == .unused) continue;
101 const slot: OngoingLoad.Index = @enumFromInt(i);
102 slot.request(loader).deinit();
103 slot.buffer(loader).deinit(loader.allocator());
104 slot.progress(loader).end();
105 }
106
107 loader.progress.end();
108 loader.ongoing.deinit(loader.allocator());
109
110 // get around assertion in http client
111 // TODO: this might be wrong
112 loader.http.connection_pool.deinit(loader.allocator());
113 loader.http.connection_pool = .{};
114
115 loader.http.deinit();
116
117 loader.* = undefined;
118}
119
120pub const Scheme = enum {
121 file,
122 http,
123 https,
124
125 pub fn fromSlice(scheme: []const u8) ?Scheme {
126 return std.meta.stringToEnum(Scheme, scheme);
127 }
128};
129
130pub const EnqueueOptions = struct {
131 /// This data is duplicated if stored into `Loader`
132 uri_text: []const u8,
133};
134
135/// Retrieve a module from the Loader, returning immediately if it is already
136/// available. If a module load is pending, an index is returned, which can be
137/// looked up later when needed.
138pub fn enqueue(loader: *Loader, opts: EnqueueOptions) Allocator.Error!EnqueueResult {
139 // give a assertion fail on the thread that enqueued it
140 // safe builds will still panic, but instead on the loader's thread
141 if (@import("builtin").mode == .Debug) {
142 _ = Uri.parse(opts.uri_text) catch
143 unreachable; // invalid URI
144 }
145
146 const index = index: {
147 loader.entries_lock.lock();
148 defer loader.entries_lock.unlock();
149
150 // TODO: consider overwriting space in .dead modules
151
152 const gop = try loader.entries.getOrPut(loader.allocator(), opts.uri_text);
153 errdefer gop.value_ptr.* = .dead;
154
155 if (gop.found_existing) {
156 return switch (gop.value_ptr.*) {
157 .available => |data| .{ .available = data },
158 .fetching => .{ .fetching = @enumFromInt(gop.index) },
159 .failed => |fail| .{ .failed = fail },
160 .dead => {
161 gop.value_ptr.* = .fetching;
162 break :index gop.index;
163 },
164 };
165 }
166
167 gop.value_ptr.* = .fetching;
168 gop.key_ptr.* = try loader.allocator().dupe(u8, opts.uri_text);
169
170 break :index gop.index;
171 };
172 const module: Module.Index = @enumFromInt(index);
173
174 errdefer {
175 // Returning an error from enqueue means that the fetch was
176 // not actually enqueued (due to OOM), however since we
177 // already allocated the slot in the hashmap, we'll replace
178 // it with a tombstone.
179 loader.entries_lock.lockShared();
180 defer loader.entries_lock.unlockShared();
181
182 loader.entries.values()[index] = .{ .dead = {} };
183 }
184
185 // Prepare the node
186 const node = try loader.allocator().create(Task.Node);
187 node.* = .{ .data = .{
188 .uri_text = opts.uri_text,
189 .module = module,
190 } };
191 loader.queue.add(node);
192
193 return .{ .fetching = module };
194}
195
196/// This function is only used for testing basic behaviors.
197/// It is useful if you want to test stuff simply.
198pub fn loadModuleSync(loader: *Loader, opts: EnqueueOptions) !CompletionResult {
199 loader.run_lock.assertUnlocked();
200 const id = switch (try loader.enqueue(opts)) {
201 .available => |available| return .{ .available = available },
202 .failed => |failed| return .{ .failed = failed },
203 .fetching => |id| id,
204 };
205 loader.run(.until_empty);
206 return switch (loader.getModule(id)) {
207 .dead => unreachable,
208 .fetching => unreachable, // .until_empty returned before it was empty
209 .available => |available| return .{ .available = available },
210 .failed => |failed| return .{ .failed = failed },
211 };
212}
213
214pub const EnqueueResult = union(enum) {
215 available: Module.Data,
216 failed: Failure,
217 fetching: Module.Index,
218};
219
220pub const CompletionResult = union(enum) {
221 available: Module.Data,
222 failed: Failure,
223
224 /// This function is used for quick tests, where
225 /// it is okay to lose potentially relevant error data.
226 pub fn lossyUnwrap(result: CompletionResult) !Module.Loaded {
227 return switch (result) {
228 .available => |data| data,
229 .failed => |fail| fail.err,
230 };
231 }
232};
233
234pub const Failure = struct {
235 err: Err,
236 /// Which phase of the fetch had a failure.
237 phase: Phase,
238 /// For some error codes, additional information is helpful.
239 detail: union {
240 none: void,
241 http_status: std.http.Status,
242 } = .{ .none = {} },
243
244 pub const Err = Allocator.Error ||
245 std.http.Client.RequestError ||
246 std.http.Client.Request.SendError ||
247 std.http.Client.Request.FinishError ||
248 std.http.Client.Request.WaitError ||
249 std.http.Client.Request.ReadError ||
250 std.fs.File.OpenError ||
251 std.fs.File.ReadError ||
252 std.posix.PollError ||
253 error{
254 SchemeNotSupported,
255 /// `stat` lied
256 UnexpectedEndOfFile,
257 /// non-200 status codes are errors
258 HttpStatus,
259 };
260
261 pub const Phase = enum {
262 prepare,
263 http_open,
264 http_send,
265 http_finish,
266 http_wait,
267 socket_poll,
268 socket_read,
269 /// an error interacting with cache is a user configuration issue
270 cache_open,
271 cache_read,
272 fs_open,
273 fs_stat,
274 fs_read,
275 };
276
277 pub fn format(fail: Failure, comptime empty: []const u8, _: std.fmt.FormatOptions, writer: std.io.AnyWriter) !void {
278 comptime assert(empty.len == 0);
279
280 switch (fail.err) {
281 error.HttpStatus => {
282 try writer.print("HTTP {d}", .{@intFromEnum(fail.detail.http_status)});
283 if (fail.detail.http_status.phrase()) |phrase| {
284 try writer.print(": {s}", .{phrase});
285 }
286 },
287 inline else => |err| try writer.writeAll(switch (err) {
288 error.HttpStatus => unreachable, // handled
289
290 error.OutOfMemory => "Ran out of memory",
291 error.Unexpected => "Unexpected OS error",
292 error.InputOutput => "I/O Error",
293 error.IsDir => "Unexpected Directory",
294
295 else => comptime camel_case: {
296 // TODO: write a more robust version of this
297 const name = @errorName(err);
298 var buf: [name.len * 2]u8 = undefined;
299 var i = 1;
300 assert(std.ascii.isUpper(name[0]));
301 buf[0] = std.ascii.toLower(name[0]);
302 for (name[1..]) |char| {
303 assert(char < 0x80);
304 if (std.ascii.isUpper(char)) {
305 buf[i] = ' ';
306 buf[i + 1] = std.ascii.toLower(char);
307 i += 2;
308 } else {
309 buf[i] = char;
310 i += 1;
311 }
312 }
313
314 for (.{ "uri", "tls ", "http ", "fd " }) |acronym| {
315 if (std.mem.indexOf(u8, buf[0..i], acronym)) |j| {
316 for (j..j + acronym.len - 1) |k| {
317 buf[k] = std.ascii.toUpper(buf[k]);
318 }
319 }
320 }
321
322 buf[0] = std.ascii.toUpper(name[0]);
323
324 const final = buf[0..i].*;
325 break :camel_case &final;
326 },
327 }),
328 }
329
330 try writer.writeAll(switch (fail.phase) {
331 .prepare,
332 .http_open,
333 .http_send,
334 .http_finish,
335 .http_wait,
336 .socket_poll,
337 .socket_read,
338 => " while fetching module",
339 .cache_open,
340 .cache_read,
341 => " while reading from module cache",
342 .fs_open,
343 .fs_stat,
344 .fs_read,
345 => " while reading file",
346 });
347 }
348};
349
350pub const Module = union(enum) {
351 /// The module is not yet ready.
352 fetching,
353 /// The module data is valid.
354 available: Data,
355 /// An error occurred.
356 failed: Failure,
357 /// The module was unloaded.
358 /// Using this module is a use-after-free error.
359 dead,
360
361 pub const Data = struct {
362 contents: [:0]const u8,
363 path_digest: BinDigest,
364 stat: Stat,
365 };
366
367 pub const Meta = struct {
368 path_digest: BinDigest,
369 stat: Stat,
370 };
371
372 pub const Index = enum(u32) {
373 _,
374 };
375
376 pub const Pointer = struct {
377 hash_table_lock: *Thread.RwLock,
378 module: *Module,
379
380 pub fn unlock(ptr: Pointer) void {
381 ptr.hash_table_lock.unlockShared();
382 }
383 };
384
385 fn deinit(module: Module, alloc: Allocator) void {
386 switch (module) {
387 .available => |data| {
388 alloc.free(data.contents);
389 },
390 else => {},
391 }
392 }
393};
394
395/// Holds a reading lock to the hash table entries.
396/// The returned `*Module` may be mutated.
397pub fn getModulePtr(loader: *Loader, index: Module.Index) Module.Pointer {
398 loader.entries_lock.lockShared();
399 return .{
400 .hash_table_lock = &loader.entries_lock,
401 .module = &loader.entries.values()[@intFromEnum(index)],
402 };
403}
404
405pub fn getModule(loader: *Loader, index: Module.Index) Module {
406 loader.entries_lock.lockShared();
407 defer loader.entries_lock.unlockShared();
408 return loader.entries.values()[@intFromEnum(index)];
409}
410
411pub fn unloadModule(loader: *Loader, index: Module.Index) void {
412 const locked_module = loader.getModulePtr(index);
413 defer locked_module.unlock();
414 locked_module.module.deinit(loader.allocator());
415 locked_module.module = .dead;
416}
417
418const OngoingLoad = struct {
419 /// Initialized to `.unused` at startup
420 state: State,
421 /// Status of the last poll
422 poll_state: PollState,
423 /// The resulting module index
424 module: Module.Index,
425 /// Progress node associated with this fetch
426 progress: std.Progress.Node,
427 /// Stored here to reduce contention
428 meta: Module.Meta,
429 /// The in-progress HTTP request.
430 request: std.http.Client.Request,
431
432 /// This field is used for polling of all ongoing downloads, however the
433 /// indices of these do not correspond to the actual file descriptors used
434 /// for the sockets (stored in request.connection.stream.handle), but
435 /// it is instead known that `loader.ongoing.items(.pollfd)` will be an
436 /// array with the maximum possible of ongoing downloads.
437 pollfd: std.posix.pollfd,
438 pollfd_n: Index,
439
440 /// Static bytes for storing request header information
441 server_header_buffer: ServerHeaderBuffer,
442 /// Initialized to `.{}` at startup
443 buffer: std.ArrayListUnmanaged(u8),
444
445 const ServerHeaderBuffer = [8 * 1024]u8;
446 const PollState = enum { ready, wait };
447
448 /// An index into `loader.ongoing`
449 /// Can be "freed" with `.release(loader)`
450 const Index = enum(u16) {
451 _,
452
453 pub fn release(i: Index, loader: *Loader) void {
454 const states = loader.ongoing.items(.state);
455 const index = @intFromEnum(i);
456
457 states[index] = .unused;
458
459 const node = &loader.ongoing.items(.progress)[index];
460 node.end();
461 node.* = .{ .index = .none };
462
463 // update `last_ongoing` to the latest node
464 if (index + 1 == loader.last_ongoing) {
465 loader.last_ongoing -= 1;
466 while (loader.last_ongoing > 0) {
467 loader.last_ongoing -= 1;
468 if (states[loader.last_ongoing] != .unused) {
469 return;
470 }
471 }
472
473 // Complete the loader-wide progress node.
474 loader.progress.end();
475 loader.progress = .{ .index = .none };
476 }
477 }
478
479 pub fn state(i: Index, loader: *Loader) *State {
480 return &loader.ongoing.items(.state)[@intFromEnum(i)];
481 }
482
483 pub fn module(i: Index, loader: *Loader) *Module.Index {
484 return &loader.ongoing.items(.module)[@intFromEnum(i)];
485 }
486
487 pub fn request(i: Index, loader: *Loader) *std.http.Client.Request {
488 return &loader.ongoing.items(.request)[@intFromEnum(i)];
489 }
490
491 pub fn serverHeaderBuffer(i: Index, loader: *Loader) *ServerHeaderBuffer {
492 return &loader.ongoing.items(.server_header_buffer)[@intFromEnum(i)];
493 }
494
495 pub fn progress(i: Index, loader: *Loader) *std.Progress.Node {
496 return &loader.ongoing.items(.progress)[@intFromEnum(i)];
497 }
498
499 pub fn buffer(i: Index, loader: *Loader) *std.ArrayListUnmanaged(u8) {
500 return &loader.ongoing.items(.buffer)[@intFromEnum(i)];
501 }
502
503 pub fn pollState(i: Index, loader: *Loader) *PollState {
504 return &loader.ongoing.items(.poll_state)[@intFromEnum(i)];
505 }
506
507 pub fn meta(i: Index, loader: *Loader) *Module.Meta {
508 return &loader.ongoing.items(.meta)[@intFromEnum(i)];
509 }
510 };
511
512 const List = std.MultiArrayList(OngoingLoad);
513 const State = enum {
514 /// This slot is not being used
515 unused,
516 wait_for_headers,
517 collecting_body,
518 };
519};
520
521const Task = struct {
522 uri_text: []const u8,
523 module: Module.Index,
524
525 const Node = std.DoublyLinkedList(Task).Node;
526
527 /// This first-in-first-out queue is threadsafe in that any thread may call
528 /// `add`, but only the Loader thread may use `takeAll` and `giveBack`.
529 const Queue = struct {
530 head: atomic.Value(?*Node) = .{ .raw = null },
531 tail: atomic.Value(?*Node) = .{ .raw = null },
532 event: Thread.ResetEvent = .{},
533
534 pub fn add(queue: *Queue, node: *Node) void {
535 assert(node.next == null);
536 assert(node.prev == null);
537
538 if (queue.tail.swap(node, .seq_cst)) |last| {
539 last.next = node;
540 node.prev = last;
541 } else {
542 queue.head.store(node, .seq_cst);
543 }
544 queue.event.set();
545 }
546
547 fn takeAll(queue: *Queue) Take {
548 const head = queue.head.swap(null, .monotonic) orelse
549 return .{ .head = null, .tail = undefined };
550 const tail = queue.tail.swap(null, .seq_cst) orelse
551 unreachable; // to get non-null from tail, head must also be null
552 return .{ .head = head, .tail = tail };
553 }
554
555 const Take = struct {
556 head: ?*Node,
557 tail: *Node,
558 };
559
560 fn giveBack(queue: *Queue, take: Take) void {
561 if (take.head == null) return;
562
563 if (queue.tail.cmpxchgStrong(null, take.tail, .seq_cst, .seq_cst) != null) {
564 const old_head = queue.head.load(.seq_cst) orelse head: {
565 // Wait for head.store() to finish
566 queue.event.wait();
567 if (queue.head.load(.seq_cst)) |head|
568 break :head head;
569 // In the unliekly event that two `add`s happen during
570 // `giveBack`, and the first one calls swap, but the second
571 // add hits `queue.event.set`, we have to wait for the first
572 // add to finish assigning `queue.head`.
573 while (true) {
574 std.atomic.spinLoopHint();
575 break :head queue.head.load(.seq_cst) orelse continue;
576 }
577 };
578
579 assert(take.tail.next == null);
580 take.tail.next = old_head;
581 assert(old_head.prev == null);
582 old_head.prev = take.tail;
583 }
584
585 queue.head.store(take.head, .monotonic);
586 }
587 };
588};
589
590pub const CompletionCallback = fn (
591 ctx: ?*anyopaque,
592 loader: *Loader,
593 module: Module.Index,
594 result: CompletionResult,
595) void;
596
597pub fn run(
598 loader: *Loader,
599 comptime lifetime: enum { until_empty, no_block, forever },
600) if (lifetime == .forever) noreturn else void {
601 loader.run_lock.lock();
602 defer loader.run_lock.unlock();
603
604 while (true) {
605 loader.queue.event.reset();
606 var enqueued = loader.queue.takeAll();
607 while (enqueued.head) |node| : ({
608 enqueued.head = node.next;
609 loader.allocator().destroy(node);
610 }) {
611 const slot = loader.reserveDownloadSlot() orelse {
612 loader.queue.giveBack(enqueued);
613 break;
614 };
615
616 var phase: Failure.Phase = .prepare;
617 const data = loader.startRequest(slot, node.data.uri_text, &phase) catch |err| {
618 const fail: Failure = .{
619 .err = err,
620 .phase = phase,
621 };
622
623 {
624 const module_lock = loader.getModulePtr(node.data.module);
625 defer module_lock.unlock();
626 module_lock.module.* = .{ .failed = fail };
627 }
628
629 if (loader.completion_cb) |uncast_cb| {
630 const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
631 cb(loader.completion_ctx, loader, node.data.module, .{ .failed = fail });
632 }
633
634 slot.release(loader);
635
636 continue;
637 };
638
639 if (data) |result| {
640 // Save the module contents
641 {
642 const locked_module = loader.getModulePtr(node.data.module);
643 defer locked_module.unlock();
644 locked_module.module.* = .{ .available = result };
645 }
646
647 if (loader.completion_cb) |uncast_cb| {
648 const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
649 cb(loader.completion_ctx, loader, node.data.module, .{ .available = result });
650 }
651
652 slot.release(loader);
653 } else {
654 slot.module(loader).* = node.data.module;
655 }
656 }
657
658 if (loader.last_ongoing == 0) {
659 switch (lifetime) {
660 .forever => {
661 loader.queue.event.wait();
662 continue;
663 },
664 .until_empty,
665 .no_block,
666 => return,
667 }
668 }
669
670 const pollfd = loader.ongoing.items(.pollfd);
671 const pollfd_n = loader.ongoing.items(.pollfd_n);
672 var pollfd_count: u16 = 0;
673
674 for (
675 0..,
676 loader.ongoing.items(.state)[0..loader.last_ongoing],
677 ) |i, *state| {
678 if (state.* == .unused) continue;
679
680 const slot: OngoingLoad.Index = @enumFromInt(i);
681 var potential_failure: Failure = .{ .err = undefined, .phase = .prepare };
682 const tick_result = loader.tickDownload(slot, state, &potential_failure) catch |err| {
683 potential_failure.err = err;
684
685 // Save the module contents
686 {
687 const locked_module = loader.getModulePtr(slot.module(loader).*);
688 defer locked_module.unlock();
689 locked_module.module.* = .{ .failed = potential_failure };
690 }
691
692 if (loader.completion_cb) |uncast_cb| {
693 const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
694 cb(loader.completion_ctx, loader, slot.module(loader).*, .{ .failed = potential_failure });
695 }
696
697 slot.release(loader);
698 continue;
699 };
700 switch (tick_result) {
701 .pending => |fd| {
702 pollfd[pollfd_count] = .{
703 .fd = fd,
704 .events = std.posix.POLL.IN | std.posix.POLL.HUP,
705 .revents = 0,
706 };
707 pollfd_n[pollfd_count] = slot;
708 pollfd_count += 1;
709 },
710 .result => |data| {
711 // Save the module contents
712 {
713 const locked_module = loader.getModulePtr(slot.module(loader).*);
714 defer locked_module.unlock();
715 locked_module.module.* = .{ .available = data };
716 }
717
718 if (loader.completion_cb) |uncast_cb| {
719 const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
720 cb(loader.completion_ctx, loader, slot.module(loader).*, .{ .available = data });
721 }
722
723 slot.release(loader);
724 },
725 }
726 }
727
728 if (pollfd_count > 0) {
729 _ = std.posix.poll(pollfd[0..pollfd_count], 500) catch {};
730 for (pollfd_n[0..pollfd_count], pollfd[0..pollfd_count]) |slot, poll| {
731 if (poll.revents != 0)
732 slot.pollState(loader).* = .ready;
733 }
734 }
735
736 if (loader.last_ongoing == 0) {
737 switch (lifetime) {
738 .forever => {
739 loader.queue.event.wait();
740 continue;
741 },
742 .until_empty,
743 .no_block,
744 => return,
745 }
746 }
747 }
748}
749
750fn reserveDownloadSlot(loader: *Loader) ?OngoingLoad.Index {
751 const states = loader.ongoing.items(.state);
752 const index = std.mem.indexOfScalar(OngoingLoad.State, states, .unused) orelse
753 return null;
754 states[index] = .wait_for_headers;
755 loader.last_ongoing = @max(@as(u16, @intCast(index + 1)), loader.last_ongoing);
756 loader.ongoing.items(.progress)[index] = .{ .index = .none };
757 return @enumFromInt(index);
758}
759
760fn startProgress(loader: *Loader, name: []const u8, estimated_total_items: usize) std.Progress.Node {
761 if (loader.root_progress.index == .none) {
762 return .{ .index = .none };
763 }
764 if (loader.progress.index == .none) {
765 loader.progress = loader.root_progress.start("Module Loader", 0);
766 }
767 return loader.progress.start(name, estimated_total_items);
768}
769
770// returns `null` if a request was initialized and needs to be read from
771// returns `Module.Data` if a cache entry was retrieved.
772fn startRequest(
773 loader: *Loader,
774 slot: OngoingLoad.Index,
775 uri_text: []const u8,
776 phase_ptr: *Failure.Phase,
777) !?Module.Data {
778 const uri = Uri.parse(uri_text) catch
779 unreachable; // pre-validate these in Sema
780
781 const path_digest = hash: {
782 var path_hash: std.Build.Cache.HashHelper = .{};
783 path_hash.addBytes("0.1.6");
784 path_hash.addBytes(uri_text);
785 var bin: BinDigest = undefined;
786 path_hash.hasher.final(&bin);
787 break :hash bin;
788 };
789
790 const scheme = Scheme.fromSlice(uri.scheme) orelse
791 return error.SchemeNotSupported;
792
793 return switch (scheme) {
794 inline else => |s| switch (s) {
795 .file => startRequestFile,
796 .http, .https => startRequestHttp,
797 }(loader, slot, uri_text, phase_ptr, uri, path_digest),
798 };
799}
800
801fn startRequestFile(
802 loader: *Loader,
803 _: OngoingLoad.Index,
804 _: []const u8,
805 phase_ptr: *Failure.Phase,
806 uri: Uri,
807 path_digest: BinDigest,
808) !?Module.Data {
809 var bytes: [std.fs.max_path_bytes]u8 = undefined;
810 var fba = std.heap.FixedBufferAllocator.init(&bytes);
811 const path = uri.path.toRawMaybeAlloc(fba.allocator()) catch |err| switch (err) {
812 error.OutOfMemory => return error.NameTooLong,
813 };
814
815 phase_ptr.* = .fs_open;
816 const f = try std.fs.openFileAbsolute(path, .{});
817 defer f.close();
818
819 phase_ptr.* = .fs_stat;
820 const stat = try f.stat();
821
822 if (stat.size > std.math.maxInt(u32))
823 return error.FileTooBig;
824
825 phase_ptr.* = .fs_read;
826 const contents = try loader.allocator().allocSentinel(u8, @as(usize, @intCast(stat.size)), 0);
827 errdefer loader.allocator().free(contents);
828 const amt = try f.readAll(contents);
829 if (amt != stat.size)
830 return error.UnexpectedEndOfFile;
831
832 return .{
833 .contents = contents,
834 .path_digest = path_digest,
835 .stat = .{
836 .size = stat.size,
837 .inode = stat.inode,
838 .mtime = stat.mtime,
839 },
840 };
841}
842
843fn startRequestHttp(
844 loader: *Loader,
845 slot: OngoingLoad.Index,
846 uri_text: []const u8,
847 phase_ptr: *Failure.Phase,
848 uri: Uri,
849 path_digest: BinDigest,
850) !?Module.Data {
851 if (loader.cache) |cache| try_cache: {
852 const hex_digest = std.fmt.bytesToHex(path_digest, .lower);
853
854 phase_ptr.* = .cache_open;
855 const f = cache.openFile(&hex_digest, .{}) catch |err| switch (err) {
856 else => break :try_cache,
857 error.NetworkNotFound,
858 error.AccessDenied,
859 error.AntivirusInterference,
860 error.Unexpected,
861 error.ProcessFdQuotaExceeded,
862 error.SystemFdQuotaExceeded,
863 => return err,
864 };
865 defer f.close();
866
867 const stat = try f.stat();
868 const payload_size = stat.size - @sizeOf(u32) - @sizeOf(i128);
869 if (payload_size > std.math.maxInt(u32))
870 return error.FileTooBig;
871
872 phase_ptr.* = .cache_read;
873
874 const r = f.reader();
875
876 const version = r.readInt(u32, .little) catch break :try_cache;
877 if (version != 1) break :try_cache;
878
879 const mtime = r.readInt(i128, .little) catch break :try_cache;
880
881 const contents = try loader.allocator().allocSentinel(u8, @as(usize, @intCast(stat.size)), 0);
882 errdefer loader.allocator().free(contents);
883
884 {
885 const amt = try f.readAll(contents);
886 if (amt != payload_size)
887 return error.UnexpectedEndOfFile;
888 }
889
890 return .{
891 .contents = contents,
892 .path_digest = path_digest,
893 .stat = .{
894 .size = payload_size,
895 .inode = 0,
896 .mtime = mtime,
897 },
898 };
899 }
900
901 phase_ptr.* = .prepare;
902 try slot.buffer(loader).ensureTotalCapacity(loader.allocator(), 4096);
903
904 slot.progress(loader).* = loader.startProgress(uri_text, 0);
905 errdefer slot.progress(loader).end();
906
907 const req = slot.request(loader);
908 phase_ptr.* = .http_open;
909 req.* = try loader.http.open(.GET, uri, .{
910 .server_header_buffer = slot.serverHeaderBuffer(loader),
911 .keep_alive = true,
912 });
913 errdefer req.deinit();
914
915 phase_ptr.* = .http_send;
916 try req.send();
917
918 phase_ptr.* = .http_finish;
919 try req.finish();
920
921 slot.meta(loader).* = .{
922 .path_digest = path_digest,
923 .stat = undefined,
924 };
925
926 return null;
927}
928
929const TickResult = union(enum) {
930 result: Module.Data,
931 /// There is more work to wait on, call again later
932 pending: std.posix.fd_t,
933};
934
935fn tickDownload(
936 loader: *Loader,
937 slot: OngoingLoad.Index,
938 state: *OngoingLoad.State,
939 failure_ptr: *Failure,
940) !TickResult {
941 const req = slot.request(loader);
942 switch (state.*) {
943 .unused => unreachable,
944
945 .wait_for_headers => {
946 failure_ptr.phase = .http_wait;
947 try req.wait();
948
949 if (req.response.status.class() != .success) {
950 failure_ptr.detail = .{ .http_status = req.response.status };
951 return error.HttpStatus;
952 }
953
954 state.* = .collecting_body;
955 if (req.response.content_length) |len| {
956 slot.progress(loader).setEstimatedTotalItems(len);
957 // TODO: pre-allocate
958 failure_ptr.phase = .prepare;
959 }
960
961 var it = req.response.iterateHeaders();
962 var mtime = std.time.nanoTimestamp();
963 while (it.next()) |header| {
964 std.debug.print("header {s} = {s}\n", .{ header.name, header.value });
965
966 // TODO: get modified time, get cache control rules.
967 _ = &mtime;
968 }
969 slot.meta(loader).stat.mtime = mtime;
970 },
971
972 else => {},
973 }
974
975 const buffer = slot.buffer(loader);
976 const original_len = buffer.items.len;
977
978 var start_index: usize = original_len;
979 buffer.expandToCapacity();
980 const dest_slice = buffer.items[start_index..];
981
982 const handle = req.connection.?.stream.handle;
983 const poll_state = slot.pollState(loader);
984 if (poll_state.* == .wait) {
985 failure_ptr.phase = .socket_poll;
986 var polls = [_]std.posix.pollfd{.{
987 .fd = handle,
988 .events = std.posix.POLL.IN | std.posix.POLL.HUP,
989 .revents = 0,
990 }};
991 const polled = try std.posix.poll(&polls, 0);
992 if (polled == 0) return .{ .pending = handle };
993 poll_state.* = .ready;
994 }
995
996 failure_ptr.phase = .socket_read;
997
998 // TODO: This is going to block until the dest is full, right?
999 const bytes_read = try req.reader().readAll(dest_slice);
1000 start_index += bytes_read;
1001
1002 slot.progress(loader).setCompletedItems(start_index);
1003
1004 if (bytes_read != 0 and bytes_read == dest_slice.len) {
1005 // This will trigger ArrayList to expand superlinearly at whatever its growth rate is.
1006 try buffer.ensureTotalCapacity(loader.allocator(), start_index + 1);
1007 poll_state.* = .ready;
1008 return .{ .pending = handle };
1009 }
1010
1011 req.deinit();
1012
1013 const meta = slot.meta(loader);
1014
1015 const data: Module.Data = .{
1016 .contents = try buffer.toOwnedSliceSentinel(loader.allocator(), 0),
1017 .path_digest = meta.path_digest,
1018 .stat = .{
1019 .inode = 0,
1020 .size = buffer.items.len,
1021 .mtime = meta.stat.mtime,
1022 },
1023 };
1024
1025 if (loader.cache) |cache|
1026 writeToCache(cache, data) catch {
1027 // TODO: consider a warning for this
1028 };
1029
1030 return .{ .result = data };
1031}
1032
1033fn writeToCache(cache: std.fs.Dir, data: Module.Data) !void {
1034 var rng: [16]u8 = undefined;
1035 std.posix.getrandom(&rng) catch {
1036 rng = std.mem.asBytes(&std.time.nanoTimestamp()).*;
1037 };
1038
1039 // TODO: i do not trust the atomicy of this on windows
1040 // i think renameat will not use directory handles, when that is totally possible
1041 // this "temp name handle -> rename to final name" should be abstracted later
1042
1043 var temp_name_buf: [32 + ".tmp\x00".len]u8 = undefined;
1044 const temp_name = std.fmt.bufPrintZ(
1045 &temp_name_buf,
1046 "{s}.tmp",
1047 .{&std.fmt.bytesToHex(rng, .lower)},
1048 ) catch unreachable; // enough space is allocated
1049
1050 var final_name_buf: [std.Build.Cache.hex_digest_len + "\x00".len]u8 = undefined;
1051 const final_name = std.fmt.bufPrintZ(
1052 &final_name_buf,
1053 "{s}",
1054 .{&std.fmt.bytesToHex(data.path_digest, .lower)},
1055 ) catch unreachable; // enough space is allocated
1056
1057 {
1058 const file = try cache.createFileZ(temp_name, .{});
1059
1060 errdefer cache.deleteFile(temp_name) catch {}; // unhandlable
1061
1062 defer file.close();
1063
1064 var bufw = std.io.bufferedWriter(file.writer());
1065 const writer = bufw.writer();
1066
1067 try writer.writeInt(u32, 1, .little);
1068 try writer.writeInt(i128, data.stat.mtime, .little);
1069 try writer.writeAll(data.contents);
1070
1071 try bufw.flush();
1072 }
1073
1074 errdefer cache.deleteFile(temp_name) catch {}; // unhandlable
1075
1076 try cache.renameZ(temp_name, final_name);
1077}
1078
1079/// The allocator is stored on the http client struct
1080/// (There is no cost to the extra dot access)
1081pub inline fn allocator(loader: *const Loader) Allocator {
1082 return loader.http.allocator;
1083}
1084
1085const Loader = @This();
1086
1087const std = @import("std");
1088const assert = std.debug.assert;
1089const atomic = std.atomic;
1090const Allocator = std.mem.Allocator;
1091const Uri = std.Uri;
1092const Thread = std.Thread;
1093
1094const BinDigest = std.Build.Cache.BinDigest;
1095const HexDigest = std.Build.Cache.HexDigest;
1096const Stat = std.Build.Cache.File.Stat;
src/loader/main.zig created+35
......@@ -0,0 +1,35 @@
1const Loader = @import("Loader.zig");
2const std = @import("std");
3
4pub fn main() !void {
5 var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init;
6 const alloc = gpa.allocator();
7
8 const node = std.Progress.start(.{});
9 // defer node.end();
10
11 var cache = try std.fs.cwd().makeOpenPath(".cache", .{});
12 defer cache.close();
13
14 var loader = try Loader.init(.{
15 .allocator = alloc,
16 .progress = node,
17 .max_concurrent_downloads = 32,
18 .cache = cache,
19 });
20 defer loader.deinit();
21
22 const uri_text = "http://localhost:3000/src/loader/main.zig";
23 const sync = try loader.loadModuleSync(.{ .uri_text = uri_text });
24 switch (sync) {
25 .failed => |fail| {
26 std.debug.print("error: {}\n", .{fail});
27 std.debug.print("in {s}\n", .{uri_text});
28 },
29 .available => |data| {
30 std.debug.print("path_digest: 0x{s}\n", .{&std.fmt.bytesToHex(data.path_digest, .lower)});
31 std.debug.print("stat: {}\n", .{data.stat});
32 std.debug.print("{s}\n", .{data.contents});
33 },
34 }
35}