Contains shared state among all Fetch tasks.
pub const JobQueue = structpub const JobQueue = struct {
io: Io,
mutex: Io.Mutex = .init,
/// It's an array hash map so that it can be sorted before rendering the
/// dependencies.zig source file.
/// Protected by `mutex`.
table: Table = .{},
/// `table` may be missing some tasks such as ones that failed, so this
/// field contains references to all of them.
/// Protected by `mutex`.
all_fetches: std.ArrayList(*Fetch) = .empty,
prog_node: std.Progress.Node,
http_client: *std.http.Client,
/// This tracks `Fetch` tasks as well as recompression tasks.
group: Io.Group = .init,
global_cache: Directory,
/// If `null`, indicates fetch globally only.
local_storage: ?*const LocalStorage,
/// If true then, no fetching occurs, and:
/// * The `global_cache` directory is assumed to be the direct parent
/// directory of on-disk packages rather than having the "p/" directory
/// prefix inside of it.
/// * An error occurs if any non-lazy packages are not already present in
/// the package cache directory.
/// * Missing hash field causes an error, and no fetching occurs so it does
/// not print the correct hash like usual.
read_only: bool,
recursive: bool,
/// Dumps hash information to stdout which can be used to troubleshoot why
/// two hashes of the same package do not match.
/// If this is true, `recursive` must be false.
debug_hash: bool,
mode: Mode,
/// Set of hashes that will be additionally fetched even if they are marked
/// as lazy.
unlazy_set: UnlazySet = .{},
/// Identifies paths that override all packages in the tree with matching
/// project ids.
fork_set: ForkSet = .{},
pub const Mode = enum {
/// Non-lazy dependencies are always fetched.
/// Lazy dependencies are fetched only when needed.
needed,
/// Both non-lazy and lazy dependencies are always fetched.
all,
};
pub const Table = std.array_hash_map.Auto(Package.Hash, *Fetch);
pub const UnlazySet = std.array_hash_map.Auto(Package.Hash, void);
pub const ForkSet = std.array_hash_map.Custom(Fork, void, Fork.Context, false);
pub const Fork = struct {
path: Path,
manifest_ast: std.zig.Ast,
manifest: Package.Manifest,
uses: usize,
pub const Context = struct {
pub fn hash(_: @This(), a: Fork) u32 {
const project_id: Package.ProjectId = .init(a.manifest.name, a.manifest.id);
return @truncate(project_id.hash());
}
pub fn eql(_: @This(), a: Fork, b: Fork, _: usize) bool {
const a_project_id: Package.ProjectId = .init(a.manifest.name, a.manifest.id);
const b_project_id: Package.ProjectId = .init(b.manifest.name, b.manifest.id);
return a_project_id.eql(&b_project_id);
}
};
pub const Adapter = struct {
pub fn hash(_: @This(), a: Package.ProjectId) u32 {
return @truncate(a.hash());
}
pub fn eql(_: @This(), a_project_id: Package.ProjectId, b: Fork, _: usize) bool {
const b_project_id: Package.ProjectId = .init(b.manifest.name, b.manifest.id);
return a_project_id.eql(&b_project_id);
}
};
};
pub fn deinit(jq: *JobQueue) void {
const io = jq.io;
jq.group.cancel(io);
if (jq.all_fetches.items.len == 0) return;
const gpa = jq.all_fetches.items[0].arena.child_allocator;
jq.table.deinit(gpa);
// These must be deinitialized in reverse order because subsequent
// `Fetch` instances are allocated in prior ones' arenas.
// Sorry, I know it's a bit weird, but it slightly simplifies the
// critical section.
while (jq.all_fetches.pop()) |f| f.deinit();
jq.all_fetches.deinit(gpa);
jq.* = undefined;
}
/// Dumps all subsequent error bundles into the first one.
pub fn consolidateErrors(jq: *JobQueue) !void {
const root = &jq.all_fetches.items[0].error_bundle;
const gpa = root.gpa;
for (jq.all_fetches.items[1..]) |fetch| {
if (fetch.error_bundle.root_list.items.len > 0) {
var bundle = try fetch.error_bundle.toOwnedBundle("");
defer bundle.deinit(gpa);
try root.addBundleAsRoots(bundle);
}
}
}
/// Creates the dependencies.zig source code for the build runner to obtain
/// via `@import("@dependencies")`.
pub fn createDependenciesSource(jq: *JobQueue, w: *Io.Writer) Io.Writer.Error!void {
const keys = jq.table.keys();
assert(keys.len != 0); // caller should have added the first one
if (keys.len == 1) {
// This is the first one. It must have no dependencies.
return createEmptyDependenciesSource(w);
}
try w.writeAll("pub const packages = struct {\n");
// Ensure the generated .zig file is deterministic.
jq.table.sortUnstable(@as(struct {
keys: []const Package.Hash,
pub fn lessThan(ctx: @This(), a_index: usize, b_index: usize) bool {
return std.mem.lessThan(u8, &ctx.keys[a_index].bytes, &ctx.keys[b_index].bytes);
}
}, .{ .keys = keys }));
for (keys, jq.table.values()) |*hash, fetch| {
if (fetch == jq.all_fetches.items[0]) {
// The first one is a dummy package for the current project.
continue;
}
const hash_slice = hash.toSlice();
try w.print(
\\ pub const {f} = struct {{
\\
, .{std.zig.fmtId(hash_slice)});
lazy: {
switch (fetch.lazy_status) {
.eager => break :lazy,
.available => {
try w.writeAll(
\\ pub const available = true;
\\
);
break :lazy;
},
.unavailable => {
try w.writeAll(
\\ pub const available = false;
\\ };
\\
);
continue;
},
}
}
try w.print(
\\ pub const build_root = "{f}";
\\
, .{std.fmt.alt(fetch.package_root, .formatEscapeString)});
if (fetch.has_build_zig) {
try w.print(
\\ pub const build_zig = @import("{f}");
\\
, .{std.zig.fmtString(hash_slice)});
}
if (fetch.have_manifest) {
const manifest = &fetch.manifest;
try w.writeAll(
\\ pub const deps: []const struct { []const u8, []const u8 } = &.{
\\
);
for (manifest.dependencies.keys(), manifest.dependencies.values()) |name, dep| {
const h = depDigest(fetch.package_root, jq.global_cache, dep) orelse continue;
try w.print(
" .{{ \"{f}\", \"{f}\" }},\n",
.{ std.zig.fmtString(name), std.zig.fmtString(h.toSlice()) },
);
}
try w.writeAll(
\\ };
\\ };
\\
);
} else {
try w.writeAll(
\\ pub const deps: []const struct { []const u8, []const u8 } = &.{};
\\ };
\\
);
}
}
try w.writeAll(
\\};
\\
\\pub const root_deps: []const struct { []const u8, []const u8 } = &.{
\\
);
const root_fetch = jq.all_fetches.items[0];
assert(root_fetch.have_manifest);
const root_manifest = &root_fetch.manifest;
for (root_manifest.dependencies.keys(), root_manifest.dependencies.values()) |name, dep| {
const h = depDigest(root_fetch.package_root, jq.global_cache, dep) orelse continue;
try w.print(
" .{{ \"{f}\", \"{f}\" }},\n",
.{ std.zig.fmtString(name), std.zig.fmtString(h.toSlice()) },
);
}
try w.writeAll("};\n");
}
pub fn createEmptyDependenciesSource(w: *Io.Writer) Io.Writer.Error!void {
try w.writeAll(
\\pub const packages = struct {};
\\pub const root_deps: []const struct { []const u8, []const u8 } = &.{};
\\
);
}
fn recompress(jq: *JobQueue, package_hash: Package.Hash, package_root: Path) Io.Cancelable!void {
const pkg_hash_slice = package_hash.toSlice();
const prog_node = jq.prog_node.startFmt(0, "recompress {s}", .{pkg_hash_slice});
defer prog_node.end();
var dest_sub_path_buf: ["p/".len + Package.Hash.max_len + ".tar.gz".len]u8 = undefined;
const dest_path: Path = .{
.root_dir = jq.global_cache,
.sub_path = std.fmt.bufPrint(&dest_sub_path_buf, "p/{s}.tar.gz", .{pkg_hash_slice}) catch unreachable,
};
const gpa = jq.http_client.allocator;
var arena_instance = std.heap.ArenaAllocator.init(gpa);
defer arena_instance.deinit();
const arena = arena_instance.allocator();
recompressFallible(jq, arena, dest_path, pkg_hash_slice, package_root, prog_node) catch |err| switch (err) {
error.Canceled => |e| return e,
error.ReadFailed => comptime unreachable,
error.WriteFailed => comptime unreachable,
else => |e| log.warn("failed caching recompressed tarball to {f}: {t}", .{ dest_path, e }),
};
}
fn recompressFallible(
jq: *JobQueue,
arena: Allocator,
dest_path: Path,
pkg_hash_slice: []const u8,
package_root: Path,
prog_node: std.Progress.Node,
) !void {
const gpa = jq.http_client.allocator;
const io = jq.io;
// We have to walk the file system up front in order to sort the file
// list for determinism purposes. The hash of the recompressed file is
// not critical because the true hash is based on the content alone.
// However, if we want Zig users to be able to share cached package
// data with each other via peer-to-peer protocols, we benefit greatly
// from the data being identical on everyone's computers.
var scanned_files: std.ArrayList(ScannedFile) = .empty;
defer scanned_files.deinit(gpa);
var pkg_dir = try package_root.root_dir.handle.openDir(io, package_root.sub_path, .{ .iterate = true });
defer pkg_dir.close(io);
{
var walker = try pkg_dir.walk(gpa);
defer walker.deinit();
while (try walker.next(io)) |entry| {
const symlink = switch (entry.kind) {
.directory => continue,
.file => false,
.sym_link => true,
else => return error.IllegalFileType,
};
const entry_path = try arena.dupe(u8, entry.path);
// If necessary, normalize path separators to POSIX-style since the tar format requires that.
if (comptime (std.fs.path.sep != std.fs.path.sep_posix)) {
std.mem.replaceScalar(u8, entry_path, std.fs.path.sep, std.fs.path.sep_posix);
}
try scanned_files.append(gpa, .{
.ptr = entry_path.ptr,
.len = @intCast(entry_path.len),
.symlink = symlink,
});
}
std.mem.sortUnstable(ScannedFile, scanned_files.items, {}, stringCmp);
}
prog_node.setEstimatedTotalItems(scanned_files.items.len);
var atomic_file = try dest_path.root_dir.handle.createFileAtomic(io, dest_path.sub_path, .{
.make_path = true,
.replace = true,
});
defer atomic_file.deinit(io);
var file_write_buffer: [4096]u8 = undefined;
var file_writer = atomic_file.file.writer(io, &file_write_buffer);
var compress_buffer: [std.compress.flate.max_window_len]u8 = undefined;
var compress = std.compress.flate.Compress.init(&file_writer.interface, &compress_buffer, .gzip, .level_9) catch |err| switch (err) {
error.WriteFailed => return file_writer.err.?,
};
var archiver: std.tar.Writer = .{ .underlying_writer = &compress.writer };
archiver.prefix = pkg_hash_slice;
var file_read_buffer: [4096]u8 = undefined;
var link_buf: [fs.max_path_bytes]u8 = undefined;
for (scanned_files.items) |scanned_file| {
const entry_path = scanned_file.ptr[0..scanned_file.len];
if (scanned_file.symlink) {
const link_name = link_buf[0..try pkg_dir.readLink(io, entry_path, &link_buf)];
archiver.writeLink(entry_path, link_name, .{}) catch |err| switch (err) {
error.WriteFailed => return file_writer.err.?,
else => |e| return e,
};
} else {
var file = try pkg_dir.openFile(io, entry_path, .{});
defer file.close(io);
var file_reader: Io.File.Reader = .init(file, io, &file_read_buffer);
archiver.writeFile(entry_path, &file_reader, 0) catch |err| switch (err) {
error.ReadFailed => return file_reader.err.?,
error.WriteFailed => return file_writer.err.?,
else => |e| return e,
};
}
prog_node.completeOne();
}
// intentionally omitting the pointless trailer
//try archiver.finish();
compress.finish() catch |err| switch (err) {
error.WriteFailed => return file_writer.err.?,
};
try file_writer.flush();
try atomic_file.replace(io);
}
}