From 2de9a6a664e8898496980794bf415e5202cf8ea3 Mon Sep 17 00:00:00 2001 From: "Caleb J. Gardner" Date: Mon, 22 Jun 2026 07:04:41 -0500 Subject: [PATCH] Finished re-writing multi-threaded extraction --- src/data/extractor.zig | 11 +- src/data/reader.zig | 2 - src/extract-multi.zig | 378 ++++++++++++++++++++++++++++++++++++----- 3 files changed, 338 insertions(+), 53 deletions(-) diff --git a/src/data/extractor.zig b/src/data/extractor.zig index 82fd54d..5d7ba20 100644 --- a/src/data/extractor.zig +++ b/src/data/extractor.zig @@ -44,11 +44,13 @@ pub fn addCache(self: *Extractor, cache: *Cache) void { pub fn extractAsync(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File) Error!void { if (self.size == 0) return; - try file.writePositionalAll(io, &[0]u8{}, self.size); + // We write to the last byte to make sure the file has the correct size. + try file.writePositionalAll(io, &[1]u8{0}, self.size - 1); var map = try file.createMemoryMap(io, .{ .len = self.size, .protection = .{ .write = true }, + .populate = false, }); defer map.destroy(io); @@ -72,7 +74,7 @@ pub fn extractAsync(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io. } fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, map_data: []u8, read_offset: u64, block_idx: u32, err: *?Error) error{Canceled}!void { - const size = if (self.frag_data == null and block_idx == (self.size - 1 / self.block_size)) + const size = if (self.frag_data == null and block_idx == self.blocks.len - 1) self.size % self.block_size else self.block_size; @@ -93,22 +95,17 @@ fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, map_data: []u8 return; } - std.debug.print("offset: {} start: {} block: {any}\n", .{ read_offset, self.start, block }); - if (self.cache != null) { const decomp_block = self.cache.?.get(io, read_offset, block.size) catch |inner_err| { - std.debug.print("cache extractor err: {}\n", .{inner_err}); switch (inner_err) { error.Canceled => return error.Canceled, else => |e| err.* = e, } return; }; - std.debug.print("cached block size: {} should be {}\n", .{ decomp_block.len, size }); @memcpy(map_data[offset..][0..size], decomp_block[0..size]); } else { _ = self.decomp(alloc, data, map_data[offset..][0..size]) catch |inner_err| { - std.debug.print("data decomp err: {}\n", .{inner_err}); err.* = inner_err; }; } diff --git a/src/data/reader.zig b/src/data/reader.zig index 8608adf..92a776d 100644 --- a/src/data/reader.zig +++ b/src/data/reader.zig @@ -98,8 +98,6 @@ fn advance(self: *Reader) Io.Reader.Error!void { const block = self.blocks[self.block_idx]; defer self.offset += block.size; - std.debug.print("offset: {} block: {any}\n", .{ self.offset, block }); - if (block.size == 0) { self.sparse_block = true; self.interface.end = size; diff --git a/src/extract-multi.zig b/src/extract-multi.zig index 5a4a874..5bc3a31 100644 --- a/src/extract-multi.zig +++ b/src/extract-multi.zig @@ -1,5 +1,6 @@ const std = @import("std"); const Io = std.Io; +const Atomic = std.atomic.Value; const DataExtractor = @import("data/extractor.zig"); const DataReader = @import("data/reader.zig"); @@ -12,7 +13,6 @@ const MetadataReader = @import("meta_rdr.zig"); const Superblock = @import("archive.zig").Superblock; const Cache = @import("util/cache.zig"); const XattrTable = @import("xattr.zig"); -const Atomic = std.atomic.Value; pub fn extract( alloc: std.mem.Allocator, @@ -38,16 +38,68 @@ pub fn extract( var frag_table: Lookup.Table(Lookup.FragEntry) = .init(alloc, data, decomp, super.frag_start, super.frag_count); defer frag_table.deinit(); - var atomic_err: Atomic(?Error) = .init(null); + var dirs: std.PriorityDequeue(DirReturn, Io.Mutex, compareDir) = .initContext(.init); + defer dirs.deinit(alloc); - var future = switch (inode.hdr.type) {}; + errdefer while (dirs.popMax()) |d| { + if (d.hdr.num != inode.hdr.num) + alloc.free(d.path); + }; - _ = io; - _ = inode; - _ = options; - _ = path; + var atomic_err: ?Error = null; - return error.TODO; + var group: Io.Group = .init; + + switch (inode.hdr.type) { + .dir, .ext_dir => group.async(io, extractDir, .{ + alloc, + io, + super, + data, + decomp, + &cache, + &frag_table, + &id_table, + &xattr_table, + &group, + &dirs, + inode, + path, + options, + true, + &atomic_err, + }), + .file, .ext_file => group.async(io, extractFile, .{ + alloc, io, super.block_size, data, decomp, &cache, &frag_table, &id_table, &xattr_table, inode, path, options, true, &atomic_err, + }), + .symlink, .ext_symlink => group.async(io, extractSymlink, .{ + alloc, + io, + inode, + path, + true, + &atomic_err, + }), + else => group.async(io, extractNode, .{ + alloc, + io, + &id_table, + &xattr_table, + inode, + path, + options, + true, + &atomic_err, + }), + } + + try group.await(io); + + while (dirs.popMax()) |d| { + defer if (d.hdr.num != inode.hdr.num) + alloc.free(d.path); + try setMetadata(alloc, io, &id_table, &xattr_table, d.hdr, path, options, d.xattr_idx); + } } fn extractDir( @@ -60,34 +112,36 @@ fn extractDir( frag_table: *Lookup.Table(Lookup.FragEntry), id_table: *Lookup.Table(u16), xattr_table: *XattrTable, + group: *Io.Group, + dirs: *std.PriorityDequeue(DirReturn, Io.Mutex, compareDir), inode: Inode, path: []const u8, options: ExtractionOptions, origin: bool, - atomic_err: *Atomic(?Error), + atomic_err: *?Error, ) error{Canceled}!void { - defer if (!origin) alloc.free(path); + errdefer if (!origin) alloc.free(path); var xattr_idx: u32 = 0xFFFFFFFF; - Io.Dir.cwd().createDirPath(io, path, .{}) catch |err| switch (err) { + Io.Dir.cwd().createDirPath(io, path) catch |err| switch (err) { error.Canceled => { io.recancel(); return error.Canceled; }, else => |e| { - atomic_err.store(e, .unordered); + atomic_err.* = e; return; }, }; - var group: Io.Group = blk: { + _ = blk: { var dir: Directory = switch (inode.data) { .dir => |d| d_blk: { var meta: MetadataReader = .init(alloc, data, decomp, super.dir_start + d.block_start); meta.interface.discardAll(d.block_offset) catch |err| break :blk err; - break :d_blk .init(alloc, &meta.interface, d.size) catch |err| break :blk err; + break :d_blk Directory.init(alloc, &meta.interface, d.size) catch |err| break :blk err; }, .ext_dir => |d| d_blk: { xattr_idx = d.xattr_idx; @@ -95,23 +149,18 @@ fn extractDir( var meta: MetadataReader = .init(alloc, data, decomp, super.dir_start + d.block_start); meta.interface.discardAll(d.block_offset) catch |err| break :blk err; - break :d_blk .init(alloc, &meta.interface, d.size) catch |err| break :blk err; + break :d_blk Directory.init(alloc, &meta.interface, d.size) catch |err| break :blk err; }, else => unreachable, }; defer dir.deinit(alloc); - var group: Io.Group = .init; - for (dir.entries) |entry| { - var new_inode: Inode = .initEntry(alloc, data, decomp, super.inode_start, super.block_size, entry) catch |err| { - group.cancel(io); + var new_inode: Inode = Inode.initEntry(alloc, data, decomp, super.inode_start, super.block_size, entry) catch |err| break :blk err; - }; const new_path = std.mem.concat(alloc, u8, &.{ path, "/", entry.name }) catch |err| { new_inode.deinit(alloc); - group.cancel(io); break :blk err; }; @@ -126,42 +175,208 @@ fn extractDir( frag_table, id_table, xattr_table, + group, + dirs, + new_inode, + new_path, + options, + false, + atomic_err, + }), + .file => group.async(io, extractFile, .{ + alloc, + io, + super.block_size, + data, + decomp, + cache, + frag_table, + id_table, + xattr_table, + new_inode, + new_path, + options, + false, + atomic_err, + }), + .symlink => group.async(io, extractSymlink, .{ + alloc, + io, + new_inode, + new_path, + false, + atomic_err, + }), + else => group.async(io, extractNode, .{ + alloc, + io, + id_table, + xattr_table, new_inode, new_path, options, false, atomic_err, }), - .file => {}, - .symlink => {}, - else => {}, } } + } catch |err| { + atomic_err.* = err; + return; + }; + + try dirs.context.lock(io); + defer dirs.context.unlock(io); + + dirs.push(alloc, .{ .hdr = inode.hdr, .path = path, .xattr_idx = xattr_idx }) catch |err| { + atomic_err.* = err; + }; +} +fn extractFile( + alloc: std.mem.Allocator, + io: Io, + block_size: u32, + data: []u8, + decomp: Decomp.Fn, + cache: *Cache, + frag_table: *Lookup.Table(Lookup.FragEntry), + id_table: *Lookup.Table(u16), + xattr_table: *XattrTable, + inode: Inode, + path: []const u8, + options: ExtractionOptions, + origin: bool, + atomic_err: *?Error, +) error{Canceled}!void { + defer if (!origin) { + inode.deinit(alloc); + alloc.free(path); + }; + + var atomic = Io.Dir.cwd().createFileAtomic(io, path, .{}) catch |err| switch (err) { + error.Canceled => { + io.recancel(); + return error.Canceled; + }, + else => |e| { + atomic_err.* = e; + return; + }, + }; + defer atomic.deinit(io); + + var xattr_idx: u32 = 0xFFFFFFFF; + + var extractor: DataExtractor = switch (inode.data) { + .file => |f| blk: { + var ext: DataExtractor = .init(data, decomp, block_size, f.blocks, f.block_start, f.size); + ext.addCache(cache); + + if (f.frag_idx != 0xFFFFFFFF) { + const entry: Lookup.FragEntry = frag_table.get(io, f.frag_idx) catch |err| break :blk err; + + if (entry.size.uncompressed) { + ext.addFrag(data[entry.block_start..][0..entry.size.size], f.frag_offset); + } else { + const frag_blk = cache.get(io, entry.block_start, entry.size.size) catch |err| break :blk err; + ext.addFrag(frag_blk, f.frag_offset); + } + } + break :blk ext; + }, + .ext_file => |f| blk: { + xattr_idx = f.xattr_idx; + + var ext: DataExtractor = .init(data, decomp, block_size, f.blocks, f.block_start, f.size); + ext.addCache(cache); + + if (f.frag_idx != 0xFFFFFFFF) { + const entry: Lookup.FragEntry = frag_table.get(io, f.frag_idx) catch |err| break :blk err; + + if (entry.size.uncompressed) { + ext.addFrag(data[entry.block_start..][0..entry.size.size], f.frag_offset); + } else { + const frag_blk = cache.get(io, entry.block_start, entry.size.size) catch |err| break :blk err; + ext.addFrag(frag_blk, f.frag_offset); + } + } + break :blk ext; + }, + else => unreachable, } catch |err| switch (err) { error.Canceled => { io.recancel(); return error.Canceled; }, else => |e| { - atomic_err.store(e, .unordered); + atomic_err.* = e; return; }, }; - try group.await(io); - - setMetadata(alloc, io, id_table, xattr_table, inode, path, options, xattr_idx) catch |err| switch (err) { + extractor.extractAsync(alloc, io, atomic.file) catch |err| switch (err) { error.Canceled => { io.recancel(); return error.Canceled; }, else => |e| { - atomic_err.store(e, .unordered); + atomic_err.* = e; + return; + }, + }; + + atomic.link(io) catch |err| switch (err) { + error.Canceled => { + io.recancel(); + return error.Canceled; + }, + else => |e| { + atomic_err.* = e; + return; + }, + }; + + setMetadata(alloc, io, id_table, xattr_table, inode.hdr, path, options, xattr_idx) catch |err| switch (err) { + error.Canceled => { + io.recancel(); + return error.Canceled; + }, + else => |e| { + atomic_err.* = e; return; }, }; } -fn setMetadata( +fn extractSymlink( + alloc: std.mem.Allocator, + io: Io, + inode: Inode, + path: []const u8, + origin: bool, + atomic_err: *?Error, +) error{Canceled}!void { + defer if (!origin) { + inode.deinit(alloc); + alloc.free(path); + }; + + const target = switch (inode.data) { + .symlink => |s| s.target, + .ext_symlink => |s| s.target, + else => unreachable, + }; + Io.Dir.cwd().symLink(io, path, target, .{}) catch |err| switch (err) { + error.Canceled => { + io.recancel(); + return error.Canceled; + }, + else => |e| { + atomic_err.* = e; + return; + }, + }; +} +fn extractNode( alloc: std.mem.Allocator, io: Io, id_table: *Lookup.Table(u16), @@ -169,15 +384,91 @@ fn setMetadata( inode: Inode, path: []const u8, options: ExtractionOptions, + origin: bool, + atomic_err: *?Error, +) error{Canceled}!void { + defer if (!origin) alloc.free(path); + + var xattr_idx: u32 = 0xFFFFFFFF; + + var mode: u32 = undefined; + var dev: u32 = 0; + + const DT = std.os.linux.DT; + + switch (inode.data) { + .block_dev => |d| { + mode = DT.BLK; + dev = d.device; + }, + .ext_block_dev => |d| { + xattr_idx = d.xattr_idx; + + mode = DT.BLK; + dev = d.device; + }, + .char_dev => |d| { + mode = DT.CHR; + dev = d.device; + }, + .ext_char_dev => |d| { + xattr_idx = d.xattr_idx; + + mode = DT.CHR; + dev = d.device; + }, + .fifo => mode = DT.FIFO, + .ext_fifo => |f| { + xattr_idx = f.xattr_idx; + + mode = DT.FIFO; + }, + .socket => mode = DT.SOCK, + .ext_socket => |s| { + xattr_idx = s.xattr_idx; + + mode = DT.SOCK; + }, + else => unreachable, + } + + const sentinel_path = alloc.dupeSentinel(u8, path, 0) catch |err| { + atomic_err.* = err; + return; + }; + + const res = std.os.linux.mknod(sentinel_path, mode, dev); + alloc.free(sentinel_path); + if (res != 0) { + atomic_err.* = Error.MknodError; + return; + } + + setMetadata(alloc, io, id_table, xattr_table, inode.hdr, path, options, xattr_idx) catch |err| switch (err) { + error.Canceled => { + io.recancel(); + return error.Canceled; + }, + else => |e| atomic_err.* = e, + }; +} +fn setMetadata( + alloc: std.mem.Allocator, + io: Io, + id_table: *Lookup.Table(u16), + xattr_table: *XattrTable, + hdr: Inode.Header, + path: []const u8, + options: ExtractionOptions, xattr_idx: u32, -) !void { - if (options.ignore_permissions and (options.ignore_xattr or xattr_idx == null)) return; +) Error!void { + if (options.ignore_permissions and (options.ignore_xattr or xattr_idx == 0xFFFFFFFF)) return; var fil: Io.File = try Io.Dir.cwd().openFile(io, path, .{}); defer fil.close(io); if (!options.ignore_xattr and xattr_idx != 0xFFFFFFFF) { - const xattr = try xattr_table.get(alloc, io, xattr_idx.?); + const xattr = try xattr_table.get(alloc, io, xattr_idx); defer xattr.deinit(alloc); for (xattr.kvs) |kv| { @@ -188,25 +479,24 @@ fn setMetadata( } if (!options.ignore_permissions) { try fil.setTimestamps(io, .{ - .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(@as(i96, @intCast(inode.hdr.mod_time)) * std.time.ns_per_s)), + .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(@as(i96, @intCast(hdr.mod_time)) * std.time.ns_per_s)), }); - try fil.setPermissions(io, @enumFromInt(inode.hdr.permissions)); - try fil.setOwner(io, try id_table.get(io, inode.hdr.uid_idx), try id_table.get(io, inode.hdr.gid_idx)); + try fil.setPermissions(io, @enumFromInt(hdr.permissions)); + try fil.setOwner(io, try id_table.get(io, hdr.uid_idx), try id_table.get(io, hdr.gid_idx)); } } +fn compareDir(_: Io.Mutex, a: DirReturn, b: DirReturn) std.math.Order { + return std.math.order(std.mem.count(u8, a.path, "/"), std.mem.count(u8, b.path, "/")); +} + // Types -const ReturnUnion = union(enum) { - path: Error!PathReturn, - void: Error!void, -}; - -const Error = error{MknodError} || Decomp.Error || Directory.Error || DataExtractor.Error || Io.Dir.CreateDirPathError || +const Error = error{ MknodError, SetXattrError } || Decomp.Error || Directory.Error || DataExtractor.Error || Io.Dir.CreateDirPathError || Io.Dir.SymLinkError || Io.File.Atomic.LinkError || Io.Reader.StreamRemainingError; -const PathReturn = struct { +const DirReturn = struct { hdr: Inode.Header, path: []const u8, - xattr_idx: ?u32 = null, + xattr_idx: u32, };