Re-writing much of multi-threaded extraction

src/data/extractor.zig

@@ -2,6 +2,7 @@ const std = @import("std");
 const Io = std.Io;
 
 const Decomp = @import("../decomp.zig");
+const Finish = @import("../extract-multi.zig").Finish;
 const DataBlock = @import("../inode.zig").DataBlock;
 const Cache = @import("../util/cache.zig");
 
@@ -41,27 +42,28 @@ pub fn addCache(self: *Extractor, cache: *Cache) void {
     self.cache = cache;
 }
 
-pub fn extractAsync(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File) Error!void {
+pub fn extractAsync(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File, group: *Io.Group, err: *?Error, finish: *Finish) Error!void {
     if (self.size == 0) return;
 
-    var err: ?Error = null;
-    var group: Io.Group = .init;
-
     var read_offset: u64 = self.start;
     for (0.., self.blocks) |i, block| {
-        group.async(io, blockThread, .{ self, alloc, io, file, read_offset, @truncate(i), &err });
+        group.async(io, blockThread, .{ self, alloc, io, file, read_offset, @truncate(i), err, finish });
         read_offset += block.size;
     }
     if (self.frag_data != null)
-        group.async(io, fragThread, .{ self, io, file, &err });
+        group.async(io, fragThread, .{ self, io, file, err, finish });
-
-    try group.await(io);
-
-    if (err != null)
-        return err.?;
 }
 
-fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File, read_offset: u64, block_idx: u32, err: *?Error) error{Canceled}!void {
+fn blockThread(
+    self: Extractor,
+    alloc: std.mem.Allocator,
+    io: Io,
+    file: Io.File,
+    read_offset: u64,
+    block_idx: u32,
+    err: *?Error,
+    finish: *Finish,
+) error{Canceled}!void {
     const size = if (self.frag_data == null and block_idx == self.blocks.len - 1)
         self.size % self.block_size
     else
@@ -78,6 +80,10 @@ fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File,
     if (block.size == 0) {
         wrt.interface.splatByteAll(0, size) catch |inner_err| {
             err.* = inner_err;
+            return;
+        };
+        finish.finish(io) catch |inner_err| {
+            err.* = inner_err;
         };
         return;
     }
@@ -87,6 +93,10 @@ fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File,
     if (block.uncompressed) {
         wrt.interface.writeAll(data) catch |inner_err| {
             err.* = inner_err;
+            return;
+        };
+        finish.finish(io) catch |inner_err| {
+            err.* = inner_err;
         };
         return;
     }
@@ -117,8 +127,17 @@ fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File,
             err.* = inner_err;
         };
     }
+    finish.finish(io) catch |inner_err| {
+        err.* = inner_err;
+    };
 }
-fn fragThread(self: Extractor, io: Io, file: Io.File, err: *?Error) error{Canceled}!void {
+fn fragThread(
+    self: Extractor,
+    io: Io,
+    file: Io.File,
+    err: *?Error,
+    finish: *Finish,
+) error{Canceled}!void {
     const size = self.size % self.block_size;
 
     var wrt = file.writer(io, &[0]u8{});
@@ -129,6 +148,10 @@ fn fragThread(self: Extractor, io: Io, file: Io.File, err: *?Error) error{Cancel
 
     wrt.interface.writeAll(self.frag_data.?[self.frag_offset..][0..size]) catch |inner_err| {
         err.* = inner_err;
+        return;
+    };
+    finish.finish(io) catch |inner_err| {
+        err.* = inner_err;
     };
 }
 

src/extract-multi.zig

@@ -2,223 +2,241 @@ const std = @import("std");
 const Io = std.Io;
 const Atomic = std.atomic.Value;
 
+const Archive = @import("archive.zig");
 const DataExtractor = @import("data/extractor.zig");
-const DataReader = @import("data/reader.zig");
 const Decomp = @import("decomp.zig");
 const Directory = @import("directory.zig");
-const ExtractionOptions = @import("options.zig");
+const ExtractionOption = @import("options.zig");
 const Inode = @import("inode.zig");
 const Lookup = @import("lookup.zig");
 const MetadataReader = @import("meta_rdr.zig");
-const Superblock = @import("archive.zig").Superblock;
-const Cache = @import("util/cache.zig");
 const XattrTable = @import("xattr.zig");
 
-pub fn extract(
+pub fn extract(alloc: std.mem.Allocator, io: Io, arc: Archive, options: ExtractionOption) !void {
-    alloc: std.mem.Allocator,
+    var common: Common = .init(alloc, arc, options);
-    io: Io,
+    defer common.deinit();
-    super: Superblock,
-    data: []u8,
-    decomp: Decomp.Fn,
-    inode: Inode,
-    ext_loc: []const u8,
-    options: ExtractionOptions,
-) !void {
-    const path = std.mem.trim(u8, ext_loc, "/");
-
-    var common: ExtractCommon = .init(alloc, super, data, decomp, options);
-    defer common.deinit(io);
-
-    common.start(alloc, io, inode, path, null);
-
-    try common.group.await(io);
 }
 
-fn extractDir(
+fn extractAsync(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Finish) error{Canceled}!void {
-    alloc: std.mem.Allocator,
+    switch (inode.hdr.type) {
-    io: Io,
+        .dir, .ext_dir => extractDir(common, io, inode, path, parent) catch |err| switch (err) {
-    common: *ExtractCommon,
+            error.Canceled => {
-    inode: Inode,
+                io.recancel(io);
-    path: []const u8,
+                return error.Canceled;
-    parent: ?*DirReturn,
+            },
-) Error!void {
+            else => common.err = err,
+        },
+        .file, .ext_file => extractReg(common, io, inode, path, parent) catch |err| switch (err) {
+            error.Canceled => {
+                io.recancel(io);
+                return error.Canceled;
+            },
+            else => common.err = err,
+        },
+        .symlink, .ext_symlink => extractReg(common, io, inode, path, parent) catch |err| switch (err) {
+            error.Canceled => {
+                io.recancel(io);
+                return error.Canceled;
+            },
+            else => common.err = err,
+        },
+        else => extractNod(common, io, inode, path, parent) catch |err| switch (err) {
+            error.Canceled => {
+                io.recancel(io);
+                return error.Canceled;
+            },
+            else => common.err = err,
+        },
+    }
+}
+fn extractDir(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Finish) Error!void {
     var xattr_idx: u32 = 0xFFFFFFFF;
 
-    try Io.Dir.cwd().createDirPath(io, path);
+    const dir: Directory = switch (inode.data) {
-
-    var dir: Directory = switch (inode.data) {
         .dir => |d| blk: {
-            var meta: MetadataReader = .init(alloc, common.data, common.decomp, d.block_start + common.dir_start);
+            var meta: MetadataReader = .init(common.alloc, common.data, common.decomp, d.block_start);
             try meta.interface.discardAll(d.block_offset);
 
-            break :blk try Directory.init(alloc, &meta.interface, d.size);
+            break :blk try .init(common.alloc, &meta.interface, d.size);
         },
         .ext_dir => |d| blk: {
             xattr_idx = d.xattr_idx;
 
-            var meta: MetadataReader = .init(alloc, common.data, common.decomp, d.block_start + common.dir_start);
+            var meta: MetadataReader = .init(common.alloc, common.data, common.decomp, d.block_start);
             try meta.interface.discardAll(d.block_offset);
 
-            break :blk try Directory.init(alloc, &meta.interface, d.size);
+            break :blk try .init(common.alloc, &meta.interface, d.size);
         },
         else => unreachable,
     };
-    defer dir.deinit(alloc);
+    defer dir.deinit(common.alloc);
+
+    try Io.Dir.cwd().createDirPath(io, path);
 
     if (dir.entries.len == 0) {
-        defer if (parent != null) {
+        try setMetadataPath(io, inode.hdr, path, xattr_idx);
-            alloc.free(path);
+        if (parent != null) {
-            parent.?.finish(alloc);
+            try parent.?.finish();
-        };
+            common.alloc.free(path);
-        try setMetadata(alloc, io, common, inode, path, xattr_idx);
+        }
         return;
     }
 
-    var dir_ret: *DirReturn = try .init(common.dir_alloc.allocator(), inode.hdr, path, xattr_idx, dir.entries.len, parent);
+    const cur_dir: *Finish = try .init(common, dir.entries.len, path, xattr_idx, parent);
-    errdefer dir_ret.deinit(common.dir_alloc.allocator());
 
     for (dir.entries) |entry| {
-        const new_inode: Inode = try .initEntry(alloc, common.data, common.decomp, common.inode_start, common.block_size, entry);
+        var new_inode: Inode = try .initEntry(common.alloc, common.data, common.decomp, common.inode_start, common.block_size, entry);
 
-        const new_path = std.mem.concat(u8, &.{ path, "/", entry.name }) catch |err| {
+        const new_path = std.mem.concat(common.alloc, u8, &.{ path, "/", entry.name }) catch |err| {
-            new_inode.deinit(alloc);
+            new_inode.deinit(common.alloc);
             return err;
         };
 
-        common.start(alloc, io, new_inode, new_path, dir_ret);
+        common.group.async(io, extractAsync, .{ common, io, new_inode, new_path, cur_dir });
     }
 }
+fn extractReg(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Finish) Error!void {
+    var blocks: usize = 0;
+    var xattr_idx: u32 = 0xFFFFFFFF;
 
-fn setMetadata(
+    var ext: DataExtractor = switch (inode.data) {
-    alloc: std.mem.Allocator,
+        .file => |f| blk: {
-    io: Io,
+            var ext: DataExtractor = .init(common.data, common.decomp, common.block_size, f.blocks, f.block_start, f.size);
-    common: *ExtractCommon,
+            blocks = f.blocks.len;
-    inode: Inode,
-    path: []const u8,
-    xattr_idx: u32,
-) Error!void {
-    if (common.options.ignore_permissions and (common.options.ignore_xattr or xattr_idx == null)) return;
 
-    var fil: Io.File = try Io.Dir.cwd().openFile(io, path, .{});
+            if (f.frag_idx != 0xFFFFFFFF) {}
-    defer fil.close(io);
+        },
+        .ext_file => |f| blk: {
+            xattr_idx = f.xattr_idx;
+        },
+        else => unreachable,
+    };
 
-    if (!common.options.ignore_xattr and xattr_idx != null) {
+    var fin: *Finish = try .init(common, blocks, inode.hdr, path, xattr_idx, parent);
-        const xattr = try common.xattr_table.get(alloc, io, xattr_idx.?);
+}
-        defer xattr.deinit(alloc);
+
+fn setMetadataPath(common: *Common, io: Io, hdr: Inode.Header, path: []const u8, xattr_idx: u32) Error!void {
+    if (common.options.ignore_permissions and (common.options.ignore_xattr or xattr_idx == 0xFFFFFFFF)) return;
+
+    var file = try Io.Dir.cwd().openFile(io, path, .{});
+    defer file.deinit();
+
+    return setMetadata(common, io, hdr, file, xattr_idx);
+}
+fn setMetadata(common: *Common, io: Io, hdr: Inode.Header, file: Io.File, xattr_idx: u32) Error!void {
+    if (common.options.ignore_permissions and (common.options.ignore_xattr or xattr_idx == 0xFFFFFFFF)) return;
+
+    if (!common.options.ignore_xattr and xattr_idx != 0xFFFFFFFF) {
+        var xattr = try common.xattr_table.get(common.alloc, io, xattr_idx);
+        defer xattr.deinit(common.alloc);
 
         for (xattr.kvs) |kv| {
-            const res = std.os.linux.fsetxattr(fil.handle, kv.key, kv.value.ptr, kv.value.len, 0);
+            const res = std.os.linux.fsetxattr(file.handle, kv.key, kv.value, kv.value.len, 0);
             if (res != 0)
-                return error.SetXattrError;
+                return Error.SetXattr;
         }
     }
-    if (!common.options.ignore_permissions) {
+
-        try fil.setTimestamps(io, .{
+    if (common.options.ignore_permissions) return;
-            .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(@as(i96, @intCast(inode.hdr.mod_time)) * std.time.ns_per_s)),
+
+    try file.setTimestamps(io, .{
+        .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(hdr.mod_time * std.time.ns_per_s)),
     });
-        try fil.setPermissions(io, @enumFromInt(inode.hdr.permissions));
+    try file.setPermissions(io, @enumFromInt(hdr.permissions));
-        try fil.setOwner(
+    try file.setOwner(io, try common.id_table.get(hdr.uid_idx), try common.id_table.get(hdr.gid_idx));
-            io,
-            try common.id_table.get(io, inode.hdr.uid_idx),
-            try common.id_table.get(io, inode.hdr.gid_idx),
-        );
-    }
 }
 
 // Types
 
-const Error = error{ MknodError, SetXattrError } || Io.Dir.CreateDirPathError || Io.Reader.StreamError || Directory.Error;
+const Error = error{ Mknod, SetXattr } || std.mem.Allocator.Error;
 
-const DirReturn = struct {
+pub const Finish = struct {
-    common: *ExtractCommon,
+    common: *Common,
 
+    atomic: Atomic(usize),
     hdr: Inode.Header,
     path: []const u8,
-    xattr_idx: u32 = 0xFFFFFFFF,
+    xattr_idx: u32,
 
-    atomic: Atomic(u32),
+    parent: ?*Finish,
 
-    parent: ?*DirReturn,
+    fn init(common: *Common, len: usize, hdr: Inode.Header, path: []const u8, xattr_idx: u32, parent: ?*Finish) !*Finish {
-
+        const out = try common.dirAlloc().create(Finish);
-    fn init(alloc: std.mem.Allocator, hdr: Inode.Header, path: []const u8, xattr_idx: u32, sub_files: u32, parent: ?*DirReturn) !*DirReturn {
-        const out = try alloc.create(DirReturn);
 
         out.* = .{
+            .common = common,
+
+            .atomic = .init(len),
             .hdr = hdr,
             .path = path,
             .xattr_idx = xattr_idx,
 
-            .atomic = .init(sub_files),
-
             .parent = parent,
         };
-
         return out;
     }
-    fn deinit(self: *DirReturn, alloc: std.mem.Allocator) void {
+
-        alloc.destroy(self);
+    pub fn finish(self: *Finish, io: Io) Error!void {
+        if (self.atomic.fetchSub(1, .unordered) >= 0) return;
+
+        try setMetadataPath(self.common, io, self.hdr, self.path, self.xattr_idx);
+
+        if (self.parent != null) {
+            if (self.hdr.type != .dir and self.hdr.type != .ext_dir)
+                self.common.alloc.free(self.path);
+            try self.parent.?.finish(io);
         }
-
-    fn finish(self: *DirReturn, alloc: std.mem.Allocator) !void {
-        if (self.atomic.fetchSub(1, .release) != 0) return;
-
-        defer self.deinit(alloc);
     }
 };
 
-const ExtractCommon = struct {
+const Common = struct {
-    group: Io.Group,
+    alloc: std.mem.Allocator,
     dir_alloc: std.heap.ArenaAllocator,
 
+    group: Io.Group = .init,
+    err: ?Error = null,
+
     data: []u8,
     decomp: Decomp.Fn,
 
-    dir_start: u64,
     inode_start: u64,
+    dir_start: u64,
     block_size: u32,
 
     id_table: Lookup.Table(u16),
-    xattr_table: XattrTable,
     frag_table: Lookup.Table(Lookup.FragEntry),
-    cache: Cache,
+    xattr_table: XattrTable,
 
-    options: ExtractionOptions,
+    options: ExtractionOption,
 
-    fn init(alloc: std.mem.Allocator, super: Superblock, data: []u8, decomp: Decomp.Fn, options: ExtractionOptions) ExtractCommon {
+    fn init(alloc: std.mem.Allocator, arc: Archive, options: ExtractionOption) Common {
         return .{
-            .group = .init,
+            .alloc = alloc,
             .dir_alloc = .init(alloc),
 
-            .data = data,
+            .data = arc.map.memory,
-            .decomp = decomp,
+            .decomp = arc.decomp,
 
-            .dir_start = super.dir_start,
+            .inode_start = arc.super.inode_start,
-            .inode_start = super.inode_start,
+            .dir_start = arc.super.dir_start,
-            .block_size = super.block_size,
+            .block_size = arc.super.block_size,
 
-            .id_table = .init(alloc, data, decomp, super.id_start, super.id_count),
+            .id_table = .init(alloc, arc.map.memory, arc.decomp, arc.super.id_start, arc.super.id_count),
-            .xattr_table = .init(alloc, data, decomp, super.xattr_start),
+            .frag_table = .init(alloc, arc.map.memory, arc.decomp, arc.super.frag_start, arc.super.frag_count),
-            .frag_table = .init(alloc, data, decomp, super.frag_start, super.frag_count),
+            .xattr_table = .init(alloc, arc.map.memory, arc.decomp, arc.super.xattr_start),
-            .cache = .init(alloc, data, decomp),
 
             .options = options,
         };
     }
-    fn deinit(self: *ExtractCommon, io: Io) void {
+    fn deinit(self: *Common) void {
-        self.group.cancel(io);
         self.dir_alloc.deinit();
+
         self.id_table.deinit();
-        self.xattr_table.deinit();
         self.frag_table.deinit();
-        self.cache.deinit();
+        self.xattr_table.deinit();
     }
 
-    fn start(self: *ExtractCommon, alloc: std.mem.Allocator, io: Io, inode: Inode, path: []const u8, parent: ?*DirReturn) void {
+    fn dirAlloc(self: *Common) std.mem.Allocator {
-        switch (inode.hdr.type) {
+        return self.dir_alloc.allocator();
-            .dir, .ext_dir => self.group.async(io, extractDir, .{ alloc, io, self, inode, path, parent }),
-            .file, .ext_file => {},
-            .symlink, .ext_symlink => {},
-            else => {},
-        }
     }
 };