From 4a2887aab0628d3b5b834049ea398d98f2d5549c Mon Sep 17 00:00:00 2001 From: dave caruso Date: Sun, 1 Sep 2024 22:40:22 -0700 Subject: [PATCH] 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. --- build.zig | 20 +- src/loader/Loader.zig | 1096 +++++++++++++++++++++++++++++++++++++++++ src/loader/main.zig | 35 ++ 3 files changed, 1149 insertions(+), 2 deletions(-) create mode 100644 src/loader/Loader.zig create mode 100644 src/loader/main.zig diff --git a/build.zig b/build.zig index 46d942a1a0373a43e657e0ea48c4ae53f8dfe4cf..e03ca50e5f09eeb19ca07336dd0286d4a3a35d38 100644 --- a/build.zig +++ b/build.zig @@ -66,10 +66,11 @@ pub fn build(b: *Build) !void { const native_debug = buildWithTarget(b, b.graph.host, .Debug, .none, version, no_llvm); const main = buildWithTarget(b, target, optimize, tracing, version, no_llvm); - b.installArtifact(main.cli(b, BuildConfig.all)); + b.installArtifact(main.cli(b, .all)); inline for (.{ - .{ main.cli(b, BuildConfig.minimal), "compiler" }, + .{ main.cli(b, .minimal), "compiler" }, + .{ main.loader_bin, "loader" }, }) |entry| { const artifact, const name = entry; @@ -79,6 +80,7 @@ pub fn build(b: *Build) !void { } const run_step = b.step(name, "Run " ++ name ++ " binary"); run_step.dependOn(&run_cmd.step); + run_step.dependOn(&b.addInstallArtifact(artifact, .{}).step); } const test_all = b.step("test", "Run all tests"); @@ -195,6 +197,7 @@ const Artifacts = struct { core_tests: *Build.Step.Compile, opts_mod: *Build.Module, behavior_tests: *Build.Step.Compile, + loader_bin: *Build.Step.Compile, /// Construct a CLI with certain features. /// When developing on a small component, such as the compiler core, you @@ -253,6 +256,7 @@ fn buildWithTarget( // Options const opts = b.addOptions(); opts.addOption(std.SemanticVersion, "version", version); + opts.addOption([]const u8, "version_string", b.fmt("{}", .{version})); opts.addOption([]const u8, "tracing", @tagName(tracing)); const opts_mod = opts.createModule(); @@ -280,6 +284,17 @@ fn buildWithTarget( core_tests.root_module.addImport("tracer", tracer.module("tracer")); core_tests.root_module.addImport("zg:PropsData", zg.module("PropsData")); + // Loader + const loader = b.addExecutable(.{ + .name = "loader", + .root_source_file = b.path("src/loader/main.zig"), + .target = target, + .optimize = optimize, + .use_lld = use_lld, + .use_llvm = use_llvm, + .strip = false, + }); + const behavior_tests = b.addExecutable(.{ .name = "run_behavior_tests", .root_source_file = b.path("test/behavior.zig"), @@ -302,6 +317,7 @@ fn buildWithTarget( .core_tests = core_tests, .opts_mod = opts_mod, .behavior_tests = behavior_tests, + .loader_bin = loader, }; } diff --git a/src/loader/Loader.zig b/src/loader/Loader.zig new file mode 100644 index 0000000000000000000000000000000000000000..2394ea25bbec1b72ef89b3315867801044ca33c6 --- /dev/null +++ b/src/loader/Loader.zig @@ -0,0 +1,1096 @@ +//! The `Loader` is responsible for resolving, fetching and caching files from +//! URL sources. `Loader` is designed to run on it's own thread (the "Loader +//! Thread"), and communicate with the compiler core via the `enqueue` function +//! and the `completion_cb` callback. +//! +//! Note that this subsystem is entirely self-contained in the `loader` module, +//! and is an optional dependency of the compiler. This allows testing the +//! compiler without network determinism, and stress+fuzz testing `Loader`. +//! +//! The loader's thread-safe allocator owns source file contents. + +/// Shared http client. This also stores the Loader's allocator +http: std.http.Client, + +/// This map is accessed by multiple threads and is covered by `entries_mutex` +entries: std.StringArrayHashMapUnmanaged(Module), +entries_lock: Thread.RwLock, + +/// Pre-allocated up to the max concurrent requests. This is +/// allocated at thread start up, and kept alive forever. +ongoing: OngoingLoad.List.Slice, + +/// Stores the highest index in `ongoing` that is currently active. +/// If this is `0`, then there are no ongoing downloads. +last_ongoing: u16, + +/// Thread-safe queue for pending tasks. They are moved +/// to `ongoing_downloads` when they initiate. +queue: Task.Queue, + +/// External callback when a fetch completes. Called on the Loader thread. +completion_ctx: ?*anyopaque, +completion_cb: ?*const anyopaque, // cannot use CompletionCallback because dependency loop + +/// This is the parent progress node (or the program root node). This is +/// distinct from `progress` because the latter may be stopped and started as +/// the thread sleeps (when there are no pending downloads) +root_progress: std.Progress.Node, +progress: std.Progress.Node, + +cache: ?std.fs.Dir, + +/// To ensure run is not being used on more than one thread. +run_lock: std.debug.SafetyLock, + +redirect_behavior: std.http.Client.Request.RedirectBehavior, + +pub const LoaderOptions = struct { + allocator: Allocator, + max_concurrent_downloads: u16 = 32, + cache: ?std.fs.Dir = null, + /// Pass the parent progress node + progress: std.Progress.Node = .{ .index = .none }, + completion_cb: ?*const CompletionCallback = null, + completion_ctx: ?*anyopaque = null, + redirect_behavior: std.http.Client.Request.RedirectBehavior = @enumFromInt(3), +}; + +pub fn init(options: LoaderOptions) !Loader { + var ongoing = OngoingLoad.List{}; + try ongoing.setCapacity(options.allocator, options.max_concurrent_downloads); + ongoing.len = options.max_concurrent_downloads; + errdefer ongoing.deinit(options.allocator); + const ongoing_slice = ongoing.slice(); + + @memset(ongoing_slice.items(.state), .unused); + @memset(ongoing_slice.items(.buffer), .{}); + + return .{ + .http = .{ .allocator = options.allocator }, + .root_progress = options.progress, + .ongoing = ongoing_slice, + .completion_ctx = options.completion_ctx, + .completion_cb = @ptrCast(options.completion_cb), + .redirect_behavior = options.redirect_behavior, + .queue = .{}, + .progress = .{ .index = .none }, + .entries = .{}, + .entries_lock = .{}, + .run_lock = .{}, + .last_ongoing = 0, + .cache = options.cache, + }; +} + +pub fn deinit(loader: *Loader) void { + loader.entries_lock.lock(); + + const entries = loader.entries.entries.slice(); + for (entries.items(.key), entries.items(.value)) |k, v| { + loader.allocator().free(k); + v.deinit(loader.allocator()); + } + loader.entries.deinit(loader.allocator()); + + for ( + 0.., + loader.ongoing.items(.state)[0..loader.last_ongoing], + ) |i, state| { + if (state == .unused) continue; + const slot: OngoingLoad.Index = @enumFromInt(i); + slot.request(loader).deinit(); + slot.buffer(loader).deinit(loader.allocator()); + slot.progress(loader).end(); + } + + loader.progress.end(); + loader.ongoing.deinit(loader.allocator()); + + // get around assertion in http client + // TODO: this might be wrong + loader.http.connection_pool.deinit(loader.allocator()); + loader.http.connection_pool = .{}; + + loader.http.deinit(); + + loader.* = undefined; +} + +pub const Scheme = enum { + file, + http, + https, + + pub fn fromSlice(scheme: []const u8) ?Scheme { + return std.meta.stringToEnum(Scheme, scheme); + } +}; + +pub const EnqueueOptions = struct { + /// This data is duplicated if stored into `Loader` + uri_text: []const u8, +}; + +/// Retrieve a module from the Loader, returning immediately if it is already +/// available. If a module load is pending, an index is returned, which can be +/// looked up later when needed. +pub fn enqueue(loader: *Loader, opts: EnqueueOptions) Allocator.Error!EnqueueResult { + // give a assertion fail on the thread that enqueued it + // safe builds will still panic, but instead on the loader's thread + if (@import("builtin").mode == .Debug) { + _ = Uri.parse(opts.uri_text) catch + unreachable; // invalid URI + } + + const index = index: { + loader.entries_lock.lock(); + defer loader.entries_lock.unlock(); + + // TODO: consider overwriting space in .dead modules + + const gop = try loader.entries.getOrPut(loader.allocator(), opts.uri_text); + errdefer gop.value_ptr.* = .dead; + + if (gop.found_existing) { + return switch (gop.value_ptr.*) { + .available => |data| .{ .available = data }, + .fetching => .{ .fetching = @enumFromInt(gop.index) }, + .failed => |fail| .{ .failed = fail }, + .dead => { + gop.value_ptr.* = .fetching; + break :index gop.index; + }, + }; + } + + gop.value_ptr.* = .fetching; + gop.key_ptr.* = try loader.allocator().dupe(u8, opts.uri_text); + + break :index gop.index; + }; + const module: Module.Index = @enumFromInt(index); + + errdefer { + // Returning an error from enqueue means that the fetch was + // not actually enqueued (due to OOM), however since we + // already allocated the slot in the hashmap, we'll replace + // it with a tombstone. + loader.entries_lock.lockShared(); + defer loader.entries_lock.unlockShared(); + + loader.entries.values()[index] = .{ .dead = {} }; + } + + // Prepare the node + const node = try loader.allocator().create(Task.Node); + node.* = .{ .data = .{ + .uri_text = opts.uri_text, + .module = module, + } }; + loader.queue.add(node); + + return .{ .fetching = module }; +} + +/// This function is only used for testing basic behaviors. +/// It is useful if you want to test stuff simply. +pub fn loadModuleSync(loader: *Loader, opts: EnqueueOptions) !CompletionResult { + loader.run_lock.assertUnlocked(); + const id = switch (try loader.enqueue(opts)) { + .available => |available| return .{ .available = available }, + .failed => |failed| return .{ .failed = failed }, + .fetching => |id| id, + }; + loader.run(.until_empty); + return switch (loader.getModule(id)) { + .dead => unreachable, + .fetching => unreachable, // .until_empty returned before it was empty + .available => |available| return .{ .available = available }, + .failed => |failed| return .{ .failed = failed }, + }; +} + +pub const EnqueueResult = union(enum) { + available: Module.Data, + failed: Failure, + fetching: Module.Index, +}; + +pub const CompletionResult = union(enum) { + available: Module.Data, + failed: Failure, + + /// This function is used for quick tests, where + /// it is okay to lose potentially relevant error data. + pub fn lossyUnwrap(result: CompletionResult) !Module.Loaded { + return switch (result) { + .available => |data| data, + .failed => |fail| fail.err, + }; + } +}; + +pub const Failure = struct { + err: Err, + /// Which phase of the fetch had a failure. + phase: Phase, + /// For some error codes, additional information is helpful. + detail: union { + none: void, + http_status: std.http.Status, + } = .{ .none = {} }, + + pub const Err = Allocator.Error || + std.http.Client.RequestError || + std.http.Client.Request.SendError || + std.http.Client.Request.FinishError || + std.http.Client.Request.WaitError || + std.http.Client.Request.ReadError || + std.fs.File.OpenError || + std.fs.File.ReadError || + std.posix.PollError || + error{ + SchemeNotSupported, + /// `stat` lied + UnexpectedEndOfFile, + /// non-200 status codes are errors + HttpStatus, + }; + + pub const Phase = enum { + prepare, + http_open, + http_send, + http_finish, + http_wait, + socket_poll, + socket_read, + /// an error interacting with cache is a user configuration issue + cache_open, + cache_read, + fs_open, + fs_stat, + fs_read, + }; + + pub fn format(fail: Failure, comptime empty: []const u8, _: std.fmt.FormatOptions, writer: std.io.AnyWriter) !void { + comptime assert(empty.len == 0); + + switch (fail.err) { + error.HttpStatus => { + try writer.print("HTTP {d}", .{@intFromEnum(fail.detail.http_status)}); + if (fail.detail.http_status.phrase()) |phrase| { + try writer.print(": {s}", .{phrase}); + } + }, + inline else => |err| try writer.writeAll(switch (err) { + error.HttpStatus => unreachable, // handled + + error.OutOfMemory => "Ran out of memory", + error.Unexpected => "Unexpected OS error", + error.InputOutput => "I/O Error", + error.IsDir => "Unexpected Directory", + + else => comptime camel_case: { + // TODO: write a more robust version of this + const name = @errorName(err); + var buf: [name.len * 2]u8 = undefined; + var i = 1; + assert(std.ascii.isUpper(name[0])); + buf[0] = std.ascii.toLower(name[0]); + for (name[1..]) |char| { + assert(char < 0x80); + if (std.ascii.isUpper(char)) { + buf[i] = ' '; + buf[i + 1] = std.ascii.toLower(char); + i += 2; + } else { + buf[i] = char; + i += 1; + } + } + + for (.{ "uri", "tls ", "http ", "fd " }) |acronym| { + if (std.mem.indexOf(u8, buf[0..i], acronym)) |j| { + for (j..j + acronym.len - 1) |k| { + buf[k] = std.ascii.toUpper(buf[k]); + } + } + } + + buf[0] = std.ascii.toUpper(name[0]); + + const final = buf[0..i].*; + break :camel_case &final; + }, + }), + } + + try writer.writeAll(switch (fail.phase) { + .prepare, + .http_open, + .http_send, + .http_finish, + .http_wait, + .socket_poll, + .socket_read, + => " while fetching module", + .cache_open, + .cache_read, + => " while reading from module cache", + .fs_open, + .fs_stat, + .fs_read, + => " while reading file", + }); + } +}; + +pub const Module = union(enum) { + /// The module is not yet ready. + fetching, + /// The module data is valid. + available: Data, + /// An error occurred. + failed: Failure, + /// The module was unloaded. + /// Using this module is a use-after-free error. + dead, + + pub const Data = struct { + contents: [:0]const u8, + path_digest: BinDigest, + stat: Stat, + }; + + pub const Meta = struct { + path_digest: BinDigest, + stat: Stat, + }; + + pub const Index = enum(u32) { + _, + }; + + pub const Pointer = struct { + hash_table_lock: *Thread.RwLock, + module: *Module, + + pub fn unlock(ptr: Pointer) void { + ptr.hash_table_lock.unlockShared(); + } + }; + + fn deinit(module: Module, alloc: Allocator) void { + switch (module) { + .available => |data| { + alloc.free(data.contents); + }, + else => {}, + } + } +}; + +/// Holds a reading lock to the hash table entries. +/// The returned `*Module` may be mutated. +pub fn getModulePtr(loader: *Loader, index: Module.Index) Module.Pointer { + loader.entries_lock.lockShared(); + return .{ + .hash_table_lock = &loader.entries_lock, + .module = &loader.entries.values()[@intFromEnum(index)], + }; +} + +pub fn getModule(loader: *Loader, index: Module.Index) Module { + loader.entries_lock.lockShared(); + defer loader.entries_lock.unlockShared(); + return loader.entries.values()[@intFromEnum(index)]; +} + +pub fn unloadModule(loader: *Loader, index: Module.Index) void { + const locked_module = loader.getModulePtr(index); + defer locked_module.unlock(); + locked_module.module.deinit(loader.allocator()); + locked_module.module = .dead; +} + +const OngoingLoad = struct { + /// Initialized to `.unused` at startup + state: State, + /// Status of the last poll + poll_state: PollState, + /// The resulting module index + module: Module.Index, + /// Progress node associated with this fetch + progress: std.Progress.Node, + /// Stored here to reduce contention + meta: Module.Meta, + /// The in-progress HTTP request. + request: std.http.Client.Request, + + /// This field is used for polling of all ongoing downloads, however the + /// indices of these do not correspond to the actual file descriptors used + /// for the sockets (stored in request.connection.stream.handle), but + /// it is instead known that `loader.ongoing.items(.pollfd)` will be an + /// array with the maximum possible of ongoing downloads. + pollfd: std.posix.pollfd, + pollfd_n: Index, + + /// Static bytes for storing request header information + server_header_buffer: ServerHeaderBuffer, + /// Initialized to `.{}` at startup + buffer: std.ArrayListUnmanaged(u8), + + const ServerHeaderBuffer = [8 * 1024]u8; + const PollState = enum { ready, wait }; + + /// An index into `loader.ongoing` + /// Can be "freed" with `.release(loader)` + const Index = enum(u16) { + _, + + pub fn release(i: Index, loader: *Loader) void { + const states = loader.ongoing.items(.state); + const index = @intFromEnum(i); + + states[index] = .unused; + + const node = &loader.ongoing.items(.progress)[index]; + node.end(); + node.* = .{ .index = .none }; + + // update `last_ongoing` to the latest node + if (index + 1 == loader.last_ongoing) { + loader.last_ongoing -= 1; + while (loader.last_ongoing > 0) { + loader.last_ongoing -= 1; + if (states[loader.last_ongoing] != .unused) { + return; + } + } + + // Complete the loader-wide progress node. + loader.progress.end(); + loader.progress = .{ .index = .none }; + } + } + + pub fn state(i: Index, loader: *Loader) *State { + return &loader.ongoing.items(.state)[@intFromEnum(i)]; + } + + pub fn module(i: Index, loader: *Loader) *Module.Index { + return &loader.ongoing.items(.module)[@intFromEnum(i)]; + } + + pub fn request(i: Index, loader: *Loader) *std.http.Client.Request { + return &loader.ongoing.items(.request)[@intFromEnum(i)]; + } + + pub fn serverHeaderBuffer(i: Index, loader: *Loader) *ServerHeaderBuffer { + return &loader.ongoing.items(.server_header_buffer)[@intFromEnum(i)]; + } + + pub fn progress(i: Index, loader: *Loader) *std.Progress.Node { + return &loader.ongoing.items(.progress)[@intFromEnum(i)]; + } + + pub fn buffer(i: Index, loader: *Loader) *std.ArrayListUnmanaged(u8) { + return &loader.ongoing.items(.buffer)[@intFromEnum(i)]; + } + + pub fn pollState(i: Index, loader: *Loader) *PollState { + return &loader.ongoing.items(.poll_state)[@intFromEnum(i)]; + } + + pub fn meta(i: Index, loader: *Loader) *Module.Meta { + return &loader.ongoing.items(.meta)[@intFromEnum(i)]; + } + }; + + const List = std.MultiArrayList(OngoingLoad); + const State = enum { + /// This slot is not being used + unused, + wait_for_headers, + collecting_body, + }; +}; + +const Task = struct { + uri_text: []const u8, + module: Module.Index, + + const Node = std.DoublyLinkedList(Task).Node; + + /// This first-in-first-out queue is threadsafe in that any thread may call + /// `add`, but only the Loader thread may use `takeAll` and `giveBack`. + const Queue = struct { + head: atomic.Value(?*Node) = .{ .raw = null }, + tail: atomic.Value(?*Node) = .{ .raw = null }, + event: Thread.ResetEvent = .{}, + + pub fn add(queue: *Queue, node: *Node) void { + assert(node.next == null); + assert(node.prev == null); + + if (queue.tail.swap(node, .seq_cst)) |last| { + last.next = node; + node.prev = last; + } else { + queue.head.store(node, .seq_cst); + } + queue.event.set(); + } + + fn takeAll(queue: *Queue) Take { + const head = queue.head.swap(null, .monotonic) orelse + return .{ .head = null, .tail = undefined }; + const tail = queue.tail.swap(null, .seq_cst) orelse + unreachable; // to get non-null from tail, head must also be null + return .{ .head = head, .tail = tail }; + } + + const Take = struct { + head: ?*Node, + tail: *Node, + }; + + fn giveBack(queue: *Queue, take: Take) void { + if (take.head == null) return; + + if (queue.tail.cmpxchgStrong(null, take.tail, .seq_cst, .seq_cst) != null) { + const old_head = queue.head.load(.seq_cst) orelse head: { + // Wait for head.store() to finish + queue.event.wait(); + if (queue.head.load(.seq_cst)) |head| + break :head head; + // In the unliekly event that two `add`s happen during + // `giveBack`, and the first one calls swap, but the second + // add hits `queue.event.set`, we have to wait for the first + // add to finish assigning `queue.head`. + while (true) { + std.atomic.spinLoopHint(); + break :head queue.head.load(.seq_cst) orelse continue; + } + }; + + assert(take.tail.next == null); + take.tail.next = old_head; + assert(old_head.prev == null); + old_head.prev = take.tail; + } + + queue.head.store(take.head, .monotonic); + } + }; +}; + +pub const CompletionCallback = fn ( + ctx: ?*anyopaque, + loader: *Loader, + module: Module.Index, + result: CompletionResult, +) void; + +pub fn run( + loader: *Loader, + comptime lifetime: enum { until_empty, no_block, forever }, +) if (lifetime == .forever) noreturn else void { + loader.run_lock.lock(); + defer loader.run_lock.unlock(); + + while (true) { + loader.queue.event.reset(); + var enqueued = loader.queue.takeAll(); + while (enqueued.head) |node| : ({ + enqueued.head = node.next; + loader.allocator().destroy(node); + }) { + const slot = loader.reserveDownloadSlot() orelse { + loader.queue.giveBack(enqueued); + break; + }; + + var phase: Failure.Phase = .prepare; + const data = loader.startRequest(slot, node.data.uri_text, &phase) catch |err| { + const fail: Failure = .{ + .err = err, + .phase = phase, + }; + + { + const module_lock = loader.getModulePtr(node.data.module); + defer module_lock.unlock(); + module_lock.module.* = .{ .failed = fail }; + } + + if (loader.completion_cb) |uncast_cb| { + const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb)); + cb(loader.completion_ctx, loader, node.data.module, .{ .failed = fail }); + } + + slot.release(loader); + + continue; + }; + + if (data) |result| { + // Save the module contents + { + const locked_module = loader.getModulePtr(node.data.module); + defer locked_module.unlock(); + locked_module.module.* = .{ .available = result }; + } + + if (loader.completion_cb) |uncast_cb| { + const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb)); + cb(loader.completion_ctx, loader, node.data.module, .{ .available = result }); + } + + slot.release(loader); + } else { + slot.module(loader).* = node.data.module; + } + } + + if (loader.last_ongoing == 0) { + switch (lifetime) { + .forever => { + loader.queue.event.wait(); + continue; + }, + .until_empty, + .no_block, + => return, + } + } + + const pollfd = loader.ongoing.items(.pollfd); + const pollfd_n = loader.ongoing.items(.pollfd_n); + var pollfd_count: u16 = 0; + + for ( + 0.., + loader.ongoing.items(.state)[0..loader.last_ongoing], + ) |i, *state| { + if (state.* == .unused) continue; + + const slot: OngoingLoad.Index = @enumFromInt(i); + var potential_failure: Failure = .{ .err = undefined, .phase = .prepare }; + const tick_result = loader.tickDownload(slot, state, &potential_failure) catch |err| { + potential_failure.err = err; + + // Save the module contents + { + const locked_module = loader.getModulePtr(slot.module(loader).*); + defer locked_module.unlock(); + locked_module.module.* = .{ .failed = potential_failure }; + } + + if (loader.completion_cb) |uncast_cb| { + const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb)); + cb(loader.completion_ctx, loader, slot.module(loader).*, .{ .failed = potential_failure }); + } + + slot.release(loader); + continue; + }; + switch (tick_result) { + .pending => |fd| { + pollfd[pollfd_count] = .{ + .fd = fd, + .events = std.posix.POLL.IN | std.posix.POLL.HUP, + .revents = 0, + }; + pollfd_n[pollfd_count] = slot; + pollfd_count += 1; + }, + .result => |data| { + // Save the module contents + { + const locked_module = loader.getModulePtr(slot.module(loader).*); + defer locked_module.unlock(); + locked_module.module.* = .{ .available = data }; + } + + if (loader.completion_cb) |uncast_cb| { + const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb)); + cb(loader.completion_ctx, loader, slot.module(loader).*, .{ .available = data }); + } + + slot.release(loader); + }, + } + } + + if (pollfd_count > 0) { + _ = std.posix.poll(pollfd[0..pollfd_count], 500) catch {}; + for (pollfd_n[0..pollfd_count], pollfd[0..pollfd_count]) |slot, poll| { + if (poll.revents != 0) + slot.pollState(loader).* = .ready; + } + } + + if (loader.last_ongoing == 0) { + switch (lifetime) { + .forever => { + loader.queue.event.wait(); + continue; + }, + .until_empty, + .no_block, + => return, + } + } + } +} + +fn reserveDownloadSlot(loader: *Loader) ?OngoingLoad.Index { + const states = loader.ongoing.items(.state); + const index = std.mem.indexOfScalar(OngoingLoad.State, states, .unused) orelse + return null; + states[index] = .wait_for_headers; + loader.last_ongoing = @max(@as(u16, @intCast(index + 1)), loader.last_ongoing); + loader.ongoing.items(.progress)[index] = .{ .index = .none }; + return @enumFromInt(index); +} + +fn startProgress(loader: *Loader, name: []const u8, estimated_total_items: usize) std.Progress.Node { + if (loader.root_progress.index == .none) { + return .{ .index = .none }; + } + if (loader.progress.index == .none) { + loader.progress = loader.root_progress.start("Module Loader", 0); + } + return loader.progress.start(name, estimated_total_items); +} + +// returns `null` if a request was initialized and needs to be read from +// returns `Module.Data` if a cache entry was retrieved. +fn startRequest( + loader: *Loader, + slot: OngoingLoad.Index, + uri_text: []const u8, + phase_ptr: *Failure.Phase, +) !?Module.Data { + const uri = Uri.parse(uri_text) catch + unreachable; // pre-validate these in Sema + + const path_digest = hash: { + var path_hash: std.Build.Cache.HashHelper = .{}; + path_hash.addBytes("0.1.6"); + path_hash.addBytes(uri_text); + var bin: BinDigest = undefined; + path_hash.hasher.final(&bin); + break :hash bin; + }; + + const scheme = Scheme.fromSlice(uri.scheme) orelse + return error.SchemeNotSupported; + + return switch (scheme) { + inline else => |s| switch (s) { + .file => startRequestFile, + .http, .https => startRequestHttp, + }(loader, slot, uri_text, phase_ptr, uri, path_digest), + }; +} + +fn startRequestFile( + loader: *Loader, + _: OngoingLoad.Index, + _: []const u8, + phase_ptr: *Failure.Phase, + uri: Uri, + path_digest: BinDigest, +) !?Module.Data { + var bytes: [std.fs.max_path_bytes]u8 = undefined; + var fba = std.heap.FixedBufferAllocator.init(&bytes); + const path = uri.path.toRawMaybeAlloc(fba.allocator()) catch |err| switch (err) { + error.OutOfMemory => return error.NameTooLong, + }; + + phase_ptr.* = .fs_open; + const f = try std.fs.openFileAbsolute(path, .{}); + defer f.close(); + + phase_ptr.* = .fs_stat; + const stat = try f.stat(); + + if (stat.size > std.math.maxInt(u32)) + return error.FileTooBig; + + phase_ptr.* = .fs_read; + const contents = try loader.allocator().allocSentinel(u8, @as(usize, @intCast(stat.size)), 0); + errdefer loader.allocator().free(contents); + const amt = try f.readAll(contents); + if (amt != stat.size) + return error.UnexpectedEndOfFile; + + return .{ + .contents = contents, + .path_digest = path_digest, + .stat = .{ + .size = stat.size, + .inode = stat.inode, + .mtime = stat.mtime, + }, + }; +} + +fn startRequestHttp( + loader: *Loader, + slot: OngoingLoad.Index, + uri_text: []const u8, + phase_ptr: *Failure.Phase, + uri: Uri, + path_digest: BinDigest, +) !?Module.Data { + if (loader.cache) |cache| try_cache: { + const hex_digest = std.fmt.bytesToHex(path_digest, .lower); + + phase_ptr.* = .cache_open; + const f = cache.openFile(&hex_digest, .{}) catch |err| switch (err) { + else => break :try_cache, + error.NetworkNotFound, + error.AccessDenied, + error.AntivirusInterference, + error.Unexpected, + error.ProcessFdQuotaExceeded, + error.SystemFdQuotaExceeded, + => return err, + }; + defer f.close(); + + const stat = try f.stat(); + const payload_size = stat.size - @sizeOf(u32) - @sizeOf(i128); + if (payload_size > std.math.maxInt(u32)) + return error.FileTooBig; + + phase_ptr.* = .cache_read; + + const r = f.reader(); + + const version = r.readInt(u32, .little) catch break :try_cache; + if (version != 1) break :try_cache; + + const mtime = r.readInt(i128, .little) catch break :try_cache; + + const contents = try loader.allocator().allocSentinel(u8, @as(usize, @intCast(stat.size)), 0); + errdefer loader.allocator().free(contents); + + { + const amt = try f.readAll(contents); + if (amt != payload_size) + return error.UnexpectedEndOfFile; + } + + return .{ + .contents = contents, + .path_digest = path_digest, + .stat = .{ + .size = payload_size, + .inode = 0, + .mtime = mtime, + }, + }; + } + + phase_ptr.* = .prepare; + try slot.buffer(loader).ensureTotalCapacity(loader.allocator(), 4096); + + slot.progress(loader).* = loader.startProgress(uri_text, 0); + errdefer slot.progress(loader).end(); + + const req = slot.request(loader); + phase_ptr.* = .http_open; + req.* = try loader.http.open(.GET, uri, .{ + .server_header_buffer = slot.serverHeaderBuffer(loader), + .keep_alive = true, + }); + errdefer req.deinit(); + + phase_ptr.* = .http_send; + try req.send(); + + phase_ptr.* = .http_finish; + try req.finish(); + + slot.meta(loader).* = .{ + .path_digest = path_digest, + .stat = undefined, + }; + + return null; +} + +const TickResult = union(enum) { + result: Module.Data, + /// There is more work to wait on, call again later + pending: std.posix.fd_t, +}; + +fn tickDownload( + loader: *Loader, + slot: OngoingLoad.Index, + state: *OngoingLoad.State, + failure_ptr: *Failure, +) !TickResult { + const req = slot.request(loader); + switch (state.*) { + .unused => unreachable, + + .wait_for_headers => { + failure_ptr.phase = .http_wait; + try req.wait(); + + if (req.response.status.class() != .success) { + failure_ptr.detail = .{ .http_status = req.response.status }; + return error.HttpStatus; + } + + state.* = .collecting_body; + if (req.response.content_length) |len| { + slot.progress(loader).setEstimatedTotalItems(len); + // TODO: pre-allocate + failure_ptr.phase = .prepare; + } + + var it = req.response.iterateHeaders(); + var mtime = std.time.nanoTimestamp(); + while (it.next()) |header| { + std.debug.print("header {s} = {s}\n", .{ header.name, header.value }); + + // TODO: get modified time, get cache control rules. + _ = &mtime; + } + slot.meta(loader).stat.mtime = mtime; + }, + + else => {}, + } + + const buffer = slot.buffer(loader); + const original_len = buffer.items.len; + + var start_index: usize = original_len; + buffer.expandToCapacity(); + const dest_slice = buffer.items[start_index..]; + + const handle = req.connection.?.stream.handle; + const poll_state = slot.pollState(loader); + if (poll_state.* == .wait) { + failure_ptr.phase = .socket_poll; + var polls = [_]std.posix.pollfd{.{ + .fd = handle, + .events = std.posix.POLL.IN | std.posix.POLL.HUP, + .revents = 0, + }}; + const polled = try std.posix.poll(&polls, 0); + if (polled == 0) return .{ .pending = handle }; + poll_state.* = .ready; + } + + failure_ptr.phase = .socket_read; + + // TODO: This is going to block until the dest is full, right? + const bytes_read = try req.reader().readAll(dest_slice); + start_index += bytes_read; + + slot.progress(loader).setCompletedItems(start_index); + + if (bytes_read != 0 and bytes_read == dest_slice.len) { + // This will trigger ArrayList to expand superlinearly at whatever its growth rate is. + try buffer.ensureTotalCapacity(loader.allocator(), start_index + 1); + poll_state.* = .ready; + return .{ .pending = handle }; + } + + req.deinit(); + + const meta = slot.meta(loader); + + const data: Module.Data = .{ + .contents = try buffer.toOwnedSliceSentinel(loader.allocator(), 0), + .path_digest = meta.path_digest, + .stat = .{ + .inode = 0, + .size = buffer.items.len, + .mtime = meta.stat.mtime, + }, + }; + + if (loader.cache) |cache| + writeToCache(cache, data) catch { + // TODO: consider a warning for this + }; + + return .{ .result = data }; +} + +fn writeToCache(cache: std.fs.Dir, data: Module.Data) !void { + var rng: [16]u8 = undefined; + std.posix.getrandom(&rng) catch { + rng = std.mem.asBytes(&std.time.nanoTimestamp()).*; + }; + + // TODO: i do not trust the atomicy of this on windows + // i think renameat will not use directory handles, when that is totally possible + // this "temp name handle -> rename to final name" should be abstracted later + + var temp_name_buf: [32 + ".tmp\x00".len]u8 = undefined; + const temp_name = std.fmt.bufPrintZ( + &temp_name_buf, + "{s}.tmp", + .{&std.fmt.bytesToHex(rng, .lower)}, + ) catch unreachable; // enough space is allocated + + var final_name_buf: [std.Build.Cache.hex_digest_len + "\x00".len]u8 = undefined; + const final_name = std.fmt.bufPrintZ( + &final_name_buf, + "{s}", + .{&std.fmt.bytesToHex(data.path_digest, .lower)}, + ) catch unreachable; // enough space is allocated + + { + const file = try cache.createFileZ(temp_name, .{}); + + errdefer cache.deleteFile(temp_name) catch {}; // unhandlable + + defer file.close(); + + var bufw = std.io.bufferedWriter(file.writer()); + const writer = bufw.writer(); + + try writer.writeInt(u32, 1, .little); + try writer.writeInt(i128, data.stat.mtime, .little); + try writer.writeAll(data.contents); + + try bufw.flush(); + } + + errdefer cache.deleteFile(temp_name) catch {}; // unhandlable + + try cache.renameZ(temp_name, final_name); +} + +/// The allocator is stored on the http client struct +/// (There is no cost to the extra dot access) +pub inline fn allocator(loader: *const Loader) Allocator { + return loader.http.allocator; +} + +const Loader = @This(); + +const std = @import("std"); +const assert = std.debug.assert; +const atomic = std.atomic; +const Allocator = std.mem.Allocator; +const Uri = std.Uri; +const Thread = std.Thread; + +const BinDigest = std.Build.Cache.BinDigest; +const HexDigest = std.Build.Cache.HexDigest; +const Stat = std.Build.Cache.File.Stat; diff --git a/src/loader/main.zig b/src/loader/main.zig new file mode 100644 index 0000000000000000000000000000000000000000..19f338641a6612c8911f0bcde6be00ab84d5246d --- /dev/null +++ b/src/loader/main.zig @@ -0,0 +1,35 @@ +const Loader = @import("Loader.zig"); +const std = @import("std"); + +pub fn main() !void { + var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; + const alloc = gpa.allocator(); + + const node = std.Progress.start(.{}); + // defer node.end(); + + var cache = try std.fs.cwd().makeOpenPath(".cache", .{}); + defer cache.close(); + + var loader = try Loader.init(.{ + .allocator = alloc, + .progress = node, + .max_concurrent_downloads = 32, + .cache = cache, + }); + defer loader.deinit(); + + const uri_text = "http://localhost:3000/src/loader/main.zig"; + const sync = try loader.loadModuleSync(.{ .uri_text = uri_text }); + switch (sync) { + .failed => |fail| { + std.debug.print("error: {}\n", .{fail}); + std.debug.print("in {s}\n", .{uri_text}); + }, + .available => |data| { + std.debug.print("path_digest: 0x{s}\n", .{&std.fmt.bytesToHex(data.path_digest, .lower)}); + std.debug.print("stat: {}\n", .{data.stat}); + std.debug.print("{s}\n", .{data.contents}); + }, + } +} -- 2.54.0