Work on mutli-threaded extraction

src/data/extractor.zig

@@ -2,6 +2,8 @@ const std = @import("std");
 const Io = std.Io;
 
 const Decomp = @import("../decomp.zig");
+const FileFinish = @import("../extract-multi.zig").FileFinish;
+const Multi = @import("../extract-multi.zig");
 const DataBlock = @import("../inode.zig").DataBlock;
 const Cache = @import("../util/cache.zig");
 
@@ -41,85 +43,74 @@ 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, select: *Io.Select(Multi.SelectUnion), finish: *FileFinish) !void {
     if (self.size == 0) return;
 
-    try file.writePositionalAll(io, &[0]u8{}, self.size);
-
-    var map = try file.createMemoryMap(io, .{
-        .len = self.size,
-        .protection = .{ .write = true },
-    });
-    defer map.destroy(io);
-
-    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, map.memory, read_offset, @truncate(i), &err });
+        try select.concurrent(.reg, blockThread, .{ self, alloc, io, finish.file.file, read_offset, @truncate(i), finish });
         read_offset += block.size;
     }
     if (self.frag_data != null)
-        group.async(io, fragThread, .{ self, map.memory });
+        try select.concurrent(.reg, fragThread, .{ self, io, finish.file.file, finish });
-
-    try group.await(io);
-
-    if (err != null)
-        return err.?;
-
-    try map.write(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 {
+fn blockThread(
-    const size = if (self.frag_data == null and block_idx == (self.size - 1 / self.block_size))
+    self: Extractor,
+    alloc: std.mem.Allocator,
+    io: Io,
+    file: Io.File,
+    read_offset: u64,
+    block_idx: u32,
+    finish: *FileFinish,
+) Multi.Error!void {
+    const size = if (self.frag_data == null and block_idx == self.blocks.len - 1)
         self.size % self.block_size
     else
         self.block_size;
 
-    const offset = block_idx * self.block_size;
-
     const block = self.blocks[block_idx];
 
+    var wrt = file.writer(io, &[0]u8{});
+    try wrt.seekTo(block_idx * self.block_size);
+
     if (block.size == 0) {
-        @memset(map_data[offset..][0..size], 0);
+        try wrt.interface.splatByteAll(0, size);
+        try finish.finish(io);
         return;
     }
 
     const data = self.data[read_offset..][0..block.size];
 
     if (block.uncompressed) {
-        @memcpy(map_data[offset..][0..block.size], data);
+        try wrt.interface.writeAll(data);
+        try finish.finish(io);
         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| {
+        const decomp_block = try self.cache.?.get(io, read_offset, block.size);
-            std.debug.print("cache extractor err: {}\n", .{inner_err});
+        try wrt.interface.writeAll(decomp_block);
-            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| {
+        const tmp = try alloc.alloc(u8, size);
-            std.debug.print("data decomp err: {}\n", .{inner_err});
+        defer alloc.free(tmp);
-            err.* = inner_err;
+
-        };
+        _ = try self.decomp(alloc, data, tmp);
+        try wrt.interface.writeAll(tmp);
     }
+    try finish.finish(io);
 }
-fn fragThread(self: Extractor, map_data: []u8) error{Canceled}!void {
+fn fragThread(
+    self: Extractor,
+    io: Io,
+    file: Io.File,
+    finish: *FileFinish,
+) Multi.Error!void {
     const size = self.size % self.block_size;
-    const offset = self.blocks.len * self.block_size;
 
-    @memcpy(map_data[offset..][0..size], self.frag_data.?[self.frag_offset..][0..size]);
+    var wrt = file.writer(io, &[0]u8{});
+    try wrt.seekTo(self.blocks.len * self.block_size);
+
+    try wrt.interface.writeAll(self.frag_data.?[self.frag_offset..][0..size]);
+    try finish.finish(io);
 }
-
-// Types
-
-pub const Error = Io.File.WritePositionalError || Io.File.MemoryMap.CreateError || Decomp.Error || Cache.Error;

src/data/reader.zig

@@ -27,7 +27,7 @@ frag_offset: u32 = 0,
 io: ?Io = null,
 cache: ?*Cache = null,
 
-block: [1024 * 1024]u8 = undefined,
+block_alloc: bool = false,
 
 interface: Io.Reader = .{
     .buffer = &[0]u8{},
@@ -55,6 +55,10 @@ pub fn init(alloc: std.mem.Allocator, data: []u8, decomp: Decomp.Fn, block_size:
         .offset = data_start,
     };
 }
+pub fn deinit(self: *Reader) void {
+    if (self.block_alloc)
+        self.alloc.free(self.interface.buffer);
+}
 pub fn addFrag(self: *Reader, frag_data: []u8, frag_offset: u32) void {
     self.frag_data = frag_data;
     self.frag_offset = frag_offset;
@@ -65,6 +69,8 @@ pub fn addCache(self: *Reader, io: Io, cache: *Cache) void {
 }
 
 fn advance(self: *Reader) Io.Reader.Error!void {
+    if (self.block_alloc) self.alloc.free(self.interface.buffer);
+
     if (self.block_idx > self.blocks.len) return error.EndOfStream;
     defer self.block_idx += 1;
 
@@ -107,8 +113,14 @@ fn advance(self: *Reader) Io.Reader.Error!void {
     }
 
     if (self.cache == null) {
-        _ = self.decomp(self.alloc, self.data[self.offset..][0..block.size], self.block[0..size]) catch return error.ReadFailed;
+        self.block_alloc = true;
-        self.interface.buffer = self.block[0..size];
+        self.interface.buffer = self.alloc.alloc(u8, size) catch return error.ReadFailed;
+        errdefer {
+            self.alloc.free(self.interface.buffer);
+            self.interface.buffer = &[0]u8{};
+        }
+
+        _ = self.decomp(self.alloc, self.data[self.offset..][0..block.size], self.interface.buffer) catch return error.ReadFailed;
         self.interface.end = size;
     } else {
         self.interface.buffer = self.cache.?.get(self.io.?, self.offset, block.size) catch return error.ReadFailed;

src/extract-multi.zig

@@ -1,11 +1,11 @@
 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");
 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");
@@ -13,17 +13,11 @@ 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, super: Superblock, data: []u8, decomp: Decomp.Fn, inode: Inode, filepath: []const u8, options: ExtractionOption) !void {
-    alloc: std.mem.Allocator,
+    const path = std.mem.trim(u8, filepath, "/");
-    io: Io,
+
-    super: Superblock,
+    var frag_table: Lookup.Table(Lookup.FragEntry) = .init(alloc, data, decomp, super.frag_start, super.frag_count);
-    data: []u8,
+    defer frag_table.deinit();
-    decomp: Decomp.Fn,
-    inode: Inode,
-    ext_loc: []const u8,
-    options: ExtractionOptions,
-) !void {
-    const path = std.mem.trim(u8, ext_loc, "/");
 
     var id_table: Lookup.Table(u16) = .init(alloc, data, decomp, super.id_start, super.id_count);
     defer id_table.deinit();
@@ -31,314 +25,181 @@ pub fn extract(
     var xattr_table: XattrTable = .init(alloc, data, decomp, super.xattr_start);
     defer xattr_table.deinit();
 
-    var buf: [150]ReturnUnion = undefined;
-    var sel: Io.Select(ReturnUnion) = .init(io, &buf);
-    defer while (sel.cancel()) |res|
-        switch (res) {
-            .path => |p| {
-                const path_return = p catch continue;
-                if (path_return.path.len != path.len)
-                    alloc.free(path_return.path);
-            },
-            else => {},
-        };
-
-    var loop = io.async(finishLoop, .{ alloc, io, &sel, &id_table, &xattr_table, inode.hdr.num, options });
-
     var cache: Cache = .init(alloc, data, decomp);
     defer cache.deinit();
 
-    var frag_table: Lookup.Table(Lookup.FragEntry) = .init(alloc, data, decomp, super.frag_start, super.frag_count);
+    var err: ?Error = null;
-    defer frag_table.deinit();
+    var group: Io.Group = .init;
 
+    group.async(
+        io,
+        groupExtract,
+        .{ alloc, io, data, decomp, super, &group, &frag_table, &id_table, &xattr_table, &cache, &err, inode, path, options, null },
+    );
+}
+
+fn groupExtract(
+    alloc: std.mem.Allocator,
+    io: Io,
+    data: []u8,
+    decomp: Decomp.Fn,
+    super: Superblock,
+    group: *Io.Group,
+    frag_table: *Lookup.Table(Lookup.FragEntry),
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    cache: *Cache,
+    err: *?Error,
+    inode: Inode,
+    path: []const u8,
+    options: ExtractionOption,
+    parent: ?*Parent,
+) error{Canceled}!void {
+    if (err != null) {
+        if (parent != null) {
+            alloc.free(path);
+            inode.deinit(alloc);
+            parent.?.finish(alloc, io);
+        }
+        return;
+    }
     switch (inode.hdr.type) {
-        .file, .ext_file => sel.async(
+        .dir, .ext_dir => group.async(io, extractDir, .{ alloc, io, data, decomp, super, group, frag_table, id_table, xattr_table, cache, inode, path, options, parent }),
-            .path,
+        .file, .ext_file => group.async(io, extractReg, .{ alloc, io, data, decomp, super.block_size, group, frag_table, cache, inode, path, options, parent }),
-            extractFile,
+        .symlink, .ext_symlink => group.async(io, extractSymlink, .{ alloc, io, inode, path, parent }),
-            .{ alloc, io, data, decomp, super.block_size, &cache, &frag_table, inode, path, true },
+        else => group.async(io, extractNod, .{ alloc, io, id_table, xattr_table, inode, path, options, parent }),
-        ),
-        .dir, .ext_dir => sel.async(
-            .path,
-            extractDir,
-            .{ alloc, io, super, data, decomp, &sel, &cache, &frag_table, inode, path, true },
-        ),
-        .symlink, .ext_symlink => sel.async(
-            .void,
-            extractSymlink,
-            .{ alloc, io, inode, path, true },
-        ),
-        else => sel.async(
-            .path,
-            extractNod,
-            .{ alloc, inode, path, true },
-        ),
-    }
-
-    try loop.await(io);
-}
-
-fn dirOrder(_: void, a: PathReturn, b: PathReturn) std.math.Order {
-    return std.math.order(std.mem.count(u8, a.path, "/"), std.mem.count(u8, b.path, "/"));
-}
-fn finishLoop(alloc: std.mem.Allocator, io: Io, sel: *Io.Select(ReturnUnion), id_table: *Lookup.Table(u16), xattr_table: *XattrTable, start_num: u32, options: ExtractionOptions) !void {
-    var dirs: std.PriorityDequeue(PathReturn, void, dirOrder) = .empty;
-    defer dirs.deinit(alloc);
-    errdefer while (dirs.popMax()) |d|
-        if (d.hdr.num != start_num) alloc.free(d.path);
-
-    while (true) {
-        const value: ReturnUnion = try sel.await();
-
-        const path_ret = switch (value) {
-            .void => {
-                _ = sel.group.token.load(.unordered) orelse break;
-                continue;
-            },
-            .path => |p| try p,
-        };
-
-        if (options.ignore_permissions and (options.ignore_xattr or path_ret.xattr_idx == null)) {
-            if (path_ret.hdr.num != start_num)
-                alloc.free(path_ret.path);
-            continue;
-        }
-
-        if (path_ret.hdr.type == .dir or path_ret.hdr.type == .ext_dir) {
-            dirs.push(alloc, path_ret) catch |err| {
-                if (path_ret.hdr.num != start_num)
-                    alloc.free(path_ret.path);
-                return err;
-            };
-            continue;
-        }
-        defer if (path_ret.hdr.num != start_num)
-            alloc.free(path_ret.path);
-
-        var file = try Io.Dir.cwd().openFile(io, path_ret.path, .{});
-        defer file.close(io);
-
-        if (!options.ignore_xattr and path_ret.xattr_idx != null) {
-            const xattr = try xattr_table.get(alloc, io, path_ret.xattr_idx.?);
-            defer xattr.deinit(alloc);
-
-            for (xattr.kvs) |kv| {
-                const res = std.os.linux.fsetxattr(file.handle, kv.key, kv.value.ptr, kv.value.len, 0);
-                if (res != 0)
-                    return error.SetXattrError;
-            }
-        }
-        if (!options.ignore_permissions) {
-            try file.setTimestamps(io, .{
-                .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(@as(i96, @intCast(path_ret.hdr.mod_time)) * std.time.ns_per_s)),
-            });
-            try file.setPermissions(io, @enumFromInt(path_ret.hdr.permissions));
-            try file.setOwner(io, try id_table.get(io, path_ret.hdr.uid_idx), try id_table.get(io, path_ret.hdr.gid_idx));
-        }
-
-        _ = sel.group.token.load(.unordered) orelse break;
-    }
-
-    while (dirs.popMax()) |path_ret| {
-        defer if (path_ret.hdr.num != start_num)
-            alloc.free(path_ret.path);
-
-        var file = try Io.Dir.cwd().openFile(io, path_ret.path, .{});
-        defer file.close(io);
-
-        if (!options.ignore_xattr and path_ret.xattr_idx != null) {
-            const xattr = try xattr_table.get(alloc, io, path_ret.xattr_idx.?);
-            defer xattr.deinit(alloc);
-
-            for (xattr.kvs) |kv| {
-                const res = std.os.linux.fsetxattr(file.handle, kv.key, kv.value.ptr, kv.value.len, 0);
-                if (res != 0)
-                    return error.SetXattrError;
-            }
-        }
-        if (!options.ignore_permissions) {
-            try file.setTimestamps(io, .{
-                .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(@as(i96, @intCast(path_ret.hdr.mod_time)) * std.time.ns_per_s)),
-            });
-            try file.setPermissions(io, @enumFromInt(path_ret.hdr.permissions));
-            try file.setOwner(io, try id_table.get(io, path_ret.hdr.uid_idx), try id_table.get(io, path_ret.hdr.gid_idx));
-        }
     }
 }
 
 fn extractDir(
     alloc: std.mem.Allocator,
     io: Io,
-    super: Superblock,
     data: []u8,
     decomp: Decomp.Fn,
-    sel: *Io.Select(ReturnUnion),
+    super: Superblock,
-    cache: *Cache,
+    group: *Io.Group,
     frag_table: *Lookup.Table(Lookup.FragEntry),
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    cache: *Cache,
+    err: *?Error,
     inode: Inode,
     path: []const u8,
-    origin: bool,
+    options: ExtractionOption,
-) Error!PathReturn {
+    parent: ?*Parent,
-    defer if (!origin) inode.deinit(alloc);
+) Error!void {
-    errdefer if (!origin) alloc.free(path);
+    var xattr_idx: u32 = 0xFFFFFFFF;
 
-    var ret: PathReturn = .{
+    const dir: Directory = switch (inode.data) {
-        .hdr = inode.hdr,
-        .path = path,
-    };
-
-    try Io.Dir.cwd().createDirPath(io, path);
-
-    var dir: Directory = switch (inode.data) {
         .dir => |d| blk: {
-            var meta: MetadataReader = .init(alloc, data, decomp, d.block_start + super.dir_start);
+            var meta: MetadataReader = .init(alloc, data, decomp, super.dir_start + d.block_start);
             try meta.interface.discardAll(d.block_offset);
 
-            break :blk try Directory.init(alloc, &meta.interface, d.size);
+            break :blk try .init(alloc, &meta.interface, d.size);
         },
         .ext_dir => |d| blk: {
-            if (d.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = d.xattr_idx;
+            xattr_idx = d.xattr_idx;
 
-            var meta: MetadataReader = .init(alloc, data, decomp, d.block_start + super.dir_start);
+            var meta: MetadataReader = .init(alloc, data, decomp, super.dir_start + d.block_start);
             try meta.interface.discardAll(d.block_offset);
 
-            break :blk try Directory.init(alloc, &meta.interface, d.size);
+            break :blk try .init(alloc, &meta.interface, d.size);
         },
         else => unreachable,
     };
     defer dir.deinit(alloc);
 
-    for (dir.entries) |entry| {
+    try Io.Dir.cwd().createDirPath(io, path);
-        var new_inode: Inode = try .initEntry(alloc, data, decomp, super.inode_start, super.block_size, entry);
 
-        const new_path = std.mem.concat(alloc, u8, &.{ path, "/", entry.name }) catch |err| {
+    if (dir.entries.len == 0) {
-            new_inode.deinit(alloc);
+        try setMetadataPath(alloc, io, inode.hdr, id_table, xattr_table, path, xattr_idx, options);
-            return err;
+        if (parent != null) {
-        };
+            try parent.?.finish(io);
-
-        switch (entry.type) {
-            .dir => sel.async(.path, extractDir, .{ alloc, io, super, data, decomp, sel, cache, frag_table, new_inode, new_path, false }),
-            .file => sel.async(.path, extractFile, .{ alloc, io, data, decomp, super.block_size, cache, frag_table, new_inode, new_path, false }),
-            .symlink => sel.async(.void, extractSymlink, .{ alloc, io, new_inode, new_path, false }),
-            else => sel.async(.path, extractNod, .{ alloc, new_inode, new_path, false }),
         }
+        return;
     }
 
-    return ret;
+    const cur_dir = try Parent.create(alloc, id_table, xattr_table, err, inode.hdr, path, options, parent, xattr_idx, dir.entries);
+
+    for (dir.entries) |entry| {
+        const new_inode: Inode = try .initEntry(alloc, data, decomp, super.inode_start, super.block_size, entry);
+
+        const new_path = std.mem.concat(alloc, u8, &.{ path, "/", entry.name }) catch |e| {
+            new_inode.deinit(alloc);
+            return e;
+        };
+
+        group.async(
+            io,
+            groupExtract,
+            .{ alloc, io, data, decomp, super, group, frag_table, id_table, xattr_table, cache, err, new_inode, new_path, options, cur_dir },
+        );
+    }
 }
-fn extractFile(
+fn extractReg(
     alloc: std.mem.Allocator,
     io: Io,
     data: []u8,
     decomp: Decomp.Fn,
     block_size: u32,
-    cache: *Cache,
+    group: *Io.Group,
     frag_table: *Lookup.Table(Lookup.FragEntry),
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    cache: *Cache,
+    err: *?Error,
     inode: Inode,
     path: []const u8,
-    origin: bool,
+    options: ExtractionOption,
-) Error!PathReturn {
+    parent: ?*Parent,
-    defer if (!origin) inode.deinit(alloc);
+) Error!void {
-    errdefer if (!origin) alloc.free(path);
+    var blocks: usize = 0;
+    var xattr_idx: u32 = 0xFFFFFFFF;
 
-    try io.checkCancel();
+    var ext: DataExtractor = switch (inode.data) {
-
-    var ret: PathReturn = .{
-        .hdr = inode.hdr,
-        .path = path,
-    };
-
-    // var ext: DataExtractor = switch (inode.data) {
-    //     .file => |f| blk: {
-    //         var rdr: DataExtractor = .init(data, decomp, block_size, f.blocks, f.size, f.block_start);
-    //         rdr.addCache(cache);
-
-    //         if (f.frag_idx == 0xFFFFFFFF) break :blk rdr;
-
-    //         const entry: Lookup.FragEntry = try frag_table.get(io, f.frag_idx);
-    //         if (entry.size.uncompressed) {
-    //             rdr.addFrag(data[entry.block_start..][0..entry.size.size], f.frag_offset);
-    //         } else {
-    //             rdr.addFrag(try cache.get(io, entry.block_start, entry.size.size), f.frag_offset);
-    //         }
-
-    //         break :blk rdr;
-    //     },
-    //     .ext_file => |f| blk: {
-    //         if (f.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = f.xattr_idx;
-
-    //         var rdr: DataExtractor = .init(data, decomp, block_size, f.blocks, f.size, f.block_start);
-    //         rdr.addCache(cache);
-
-    //         if (f.frag_idx == 0xFFFFFFFF) break :blk rdr;
-
-    //         const entry: Lookup.FragEntry = try frag_table.get(io, f.frag_idx);
-    //         if (entry.size.uncompressed) {
-    //             rdr.addFrag(data[entry.block_start..][0..entry.size.size], f.frag_offset);
-    //         } else {
-    //             rdr.addFrag(try cache.get(io, entry.block_start, entry.size.size), f.frag_offset);
-    //         }
-
-    //         break :blk rdr;
-    //     },
-    //     else => unreachable,
-    // };
-
-    // var atomic = try Io.Dir.cwd().createFileAtomic(io, path, .{});
-    // defer atomic.deinit(io);
-
-    // try ext.extractAsync(alloc, io, atomic.file);
-
-    // try atomic.link(io);
-
-    var rdr: DataReader = switch (inode.data) {
         .file => |f| blk: {
-            var rdr: DataReader = .init(alloc, data, decomp, block_size, f.blocks, f.size, f.block_start);
+            var ext: DataExtractor = .init(data, decomp, block_size, f.blocks, f.block_start, f.size);
-            rdr.addCache(io, cache);
+            blocks = f.blocks.len;
 
-            if (f.frag_idx == 0xFFFFFFFF) break :blk rdr;
+            if (f.frag_idx != 0xFFFFFFFF) {
-
+                blocks += 1;
-            const entry: Lookup.FragEntry = try frag_table.get(io, f.frag_idx);
+                const frag: Lookup.FragEntry = try frag_table.get(io, f.frag_idx);
-            if (entry.size.uncompressed) {
+                if (frag.size.uncompressed) {
-                rdr.addFrag(data[entry.block_start..][0..entry.size.size], f.frag_offset);
+                    ext.addFrag(data[frag.block_start..][0..frag.size.size], f.frag_offset);
-            } else {
+                } else {
-                rdr.addFrag(try cache.get(io, entry.block_start, entry.size.size), f.frag_offset);
+                    const frag_block = try cache.get(io, frag.block_start, frag.size.size);
+                    ext.addFrag(frag_block, f.frag_offset);
+                }
             }
-
+            break :blk ext;
-            break :blk rdr;
         },
         .ext_file => |f| blk: {
-            if (f.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = f.xattr_idx;
+            xattr_idx = f.xattr_idx;
 
-            var rdr: DataReader = .init(alloc, data, decomp, block_size, f.blocks, f.size, f.block_start);
+            var ext: DataExtractor = .init(data, decomp, block_size, f.blocks, f.block_start, f.size);
-            rdr.addCache(io, cache);
+            blocks = f.blocks.len;
 
-            if (f.frag_idx == 0xFFFFFFFF) break :blk rdr;
+            if (f.frag_idx != 0xFFFFFFFF) {
-
+                blocks += 1;
-            const entry: Lookup.FragEntry = try frag_table.get(io, f.frag_idx);
+                const frag: Lookup.FragEntry = try frag_table.get(io, f.frag_idx);
-            if (entry.size.uncompressed) {
+                if (frag.size.uncompressed) {
-                rdr.addFrag(data[entry.block_start..][0..entry.size.size], f.frag_offset);
+                    ext.addFrag(data[frag.block_start..][0..frag.size.size], f.frag_offset);
-            } else {
+                } else {
-                rdr.addFrag(try cache.get(io, entry.block_start, entry.size.size), f.frag_offset);
+                    const frag_block = try cache.get(io, frag.block_start, frag.size.size);
+                    ext.addFrag(frag_block, f.frag_offset);
+                }
             }
-
+            break :blk ext;
-            break :blk rdr;
         },
         else => unreachable,
     };
 
-    var atomic = try Io.Dir.cwd().createFileAtomic(io, path, .{});
+    const fin = try FileFinish.create(alloc, io, id_table, xattr_table, err, inode.hdr, path, options, parent, xattr_idx, blocks);
-    defer atomic.deinit(io);
 
-    var writer = atomic.file.writer(io, &[0]u8{});
+    try ext.extractAsync(alloc, io, group, fin);
-    _ = try rdr.interface.streamRemaining(&writer.interface);
-    try writer.flush();
-
-    try atomic.link(io);
-
-    return ret;
 }
-fn extractSymlink(alloc: std.mem.Allocator, io: Io, inode: Inode, path: []const u8, origin: bool) Error!void {
+fn extractSymlink(alloc: std.mem.Allocator, io: Io, inode: Inode, path: []const u8, parent: ?*Parent) Error!void {
-    defer if (!origin) {
+    defer if (parent != null) {
         inode.deinit(alloc);
         alloc.free(path);
     };
@@ -348,80 +209,295 @@ fn extractSymlink(alloc: std.mem.Allocator, io: Io, inode: Inode, path: []const
         .ext_symlink => |s| s.target,
         else => unreachable,
     };
+
     try Io.Dir.cwd().symLink(io, target, path, .{});
+
+    if (parent != null)
+        parent.?.finish(io);
 }
-fn extractNod(alloc: std.mem.Allocator, inode: Inode, path: []const u8, origin: bool) Error!PathReturn {
+fn extractNod(
-    errdefer if (!origin)
+    alloc: std.mem.Allocator,
-        alloc.free(path);
+    io: Io,
-
+    id_table: *Lookup.Table(u16),
-    var ret: PathReturn = .{
+    xattr_table: *XattrTable,
-        .hdr = inode.hdr,
+    inode: Inode,
-        .path = path,
+    path: []const u8,
-    };
+    parent: ?*Parent,
-
+    options: ExtractionOption,
-    var dev: u32 = 0;
+) Error!void {
+    var xattr_idx: u32 = 0xFFFFFFFF;
+    var device: u32 = 0;
     var mode: u32 = undefined;
 
-    const DT = std.posix.DT;
+    const DT = std.os.linux.DT;
 
     switch (inode.data) {
-        .char_dev => |d| {
-            dev = d.device;
-            mode = DT.CHR;
-        },
         .block_dev => |d| {
-            dev = d.device;
             mode = DT.BLK;
-        },
+            device = d.device;
-        .ext_char_dev => |d| {
-            if (d.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = d.xattr_idx;
-
-            dev = d.device;
-            mode = DT.CHR;
         },
         .ext_block_dev => |d| {
-            if (d.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = d.xattr_idx;
-
-            dev = d.device;
             mode = DT.BLK;
+            device = d.device;
+            xattr_idx = d.xattr_idx;
         },
-        .fifo => mode = DT.FIFO,
+        .char_dev => |d| {
-        .socket => mode = DT.SOCK,
+            mode = DT.CHR;
-        .ext_fifo => |i| {
+            device = d.device;
-            if (i.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = i.xattr_idx;
+        },
-
+        .ext_char_dev => |d| {
+            mode = DT.CHR;
+            device = d.device;
+            xattr_idx = d.xattr_idx;
+        },
+        .fifo => {
             mode = DT.FIFO;
         },
-        .ext_socket => |i| {
+        .ext_fifo => |f| {
-            if (i.xattr_idx != 0xFFFFFFFF) ret.xattr_idx = i.xattr_idx;
+            mode = DT.FIFO;
-
+            xattr_idx = f.xattr_idx;
+        },
+        .socket => {
             mode = DT.SOCK;
         },
+        .ext_socket => |s| {
+            mode = DT.SOCK;
+            xattr_idx = s.xattr_idx;
+        },
         else => unreachable,
     }
 
     const sentinel_path = try alloc.dupeSentinel(u8, path, 0);
     defer alloc.free(sentinel_path);
 
-    const res = std.os.linux.mknod(sentinel_path, mode, dev);
+    const res = std.os.linux.mknod(sentinel_path, mode, device);
     if (res != 0)
-        return error.MknodError;
+        return Error.Mknod;
 
-    return ret;
+    try setMetadataPath(alloc, io, inode.hdr, id_table, xattr_table, path, xattr_idx, options);
+
+    if (parent != null)
+        try parent.?.finish(io);
+}
+
+fn setMetadataPath(
+    alloc: std.mem.Allocator,
+    io: Io,
+    hdr: Inode.Header,
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    path: []const u8,
+    xattr_idx: u32,
+    options: ExtractionOption,
+) Error!void {
+    if (options.ignore_permissions and (options.ignore_xattr or xattr_idx == 0xFFFFFFFF)) return;
+
+    var file = try Io.Dir.cwd().openFile(io, path, .{});
+    defer file.close(io);
+
+    return setMetadata(alloc, io, hdr, id_table, xattr_table, file, xattr_idx, options);
+}
+fn setMetadata(
+    alloc: std.mem.Allocator,
+    io: Io,
+    hdr: Inode.Header,
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    file: Io.File,
+    xattr_idx: u32,
+    options: ExtractionOption,
+) Error!void {
+    if (options.ignore_permissions and (options.ignore_xattr or xattr_idx == 0xFFFFFFFF)) return;
+
+    if (!options.ignore_xattr and xattr_idx != 0xFFFFFFFF) {
+        var xattr = try xattr_table.get(alloc, io, xattr_idx);
+        defer xattr.deinit(alloc);
+
+        for (xattr.kvs) |kv| {
+            const res = std.os.linux.fsetxattr(file.handle, kv.key, kv.value.ptr, kv.value.len, 0);
+            if (res != 0)
+                return Error.SetXattr;
+        }
+    }
+
+    if (options.ignore_permissions) return;
+
+    try file.setTimestamps(io, .{
+        .modify_timestamp = .init(
+            Io.Timestamp.fromNanoseconds(@as(i96, @intCast(hdr.mod_time)) * std.time.ns_per_s),
+        ),
+    });
+    try file.setPermissions(io, @enumFromInt(hdr.permissions));
+    try file.setOwner(io, try id_table.get(io, hdr.uid_idx), try id_table.get(io, hdr.gid_idx));
 }
 
 // Types
 
-const ReturnUnion = union(enum) {
+pub const Error = error{ Mknod, SetXattr } || std.mem.Allocator.Error || Io.Cancelable || Io.Reader.Error || Io.Dir.CreateDirPathError || Cache.Error ||
-    path: Error!PathReturn,
+    Io.File.SetPermissionsError || Io.Dir.SymLinkError || Io.File.SeekError || Io.Writer.Error || Io.File.Atomic.LinkError || Io.ConcurrentError;
-    void: Error!void,
+
+const FileFinish = struct {
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    err: *?Error,
+
+    inode: Inode,
+    options: ExtractionOption,
+    parent: ?*Parent,
+    xattr_idx: u32,
+
+    file: Io.File.Atomic,
+
+    atomic: Atomic(usize),
+    has_err: bool = false,
+
+    fn create(
+        alloc: std.mem.Allocator,
+        io: Io,
+        id_table: *Lookup.Table(u16),
+        xattr_table: *XattrTable,
+        err: *?Error,
+        hdr: Inode.Header,
+        path: []const u8,
+        options: ExtractionOption,
+        parent: ?*Parent,
+        xattr_idx: u32,
+        blocks: usize,
+    ) Error!*FileFinish {
+        const out = try alloc.create(FileFinish);
+        out.* = .{
+            .id_table = id_table,
+            .xattr_table = xattr_table,
+            .err = err,
+
+            .hdr = hdr,
+            .path = path,
+            .options = options,
+            .parent = parent,
+            .xattr_idx = xattr_idx,
+
+            .file = Io.Dir.cwd().createFileAtomic(io, path, .{}),
+
+            .atomic = .init(blocks),
+        };
+        return out;
+    }
+
+    fn finish(self: *FileFinish, alloc: std.mem.Allocator, io: Io) void {
+        if (self.atomic.fetchSub(1, .release) > 0) return;
+
+        defer {
+            if (self.parent != null) {
+                alloc.free(self.path);
+                if (self.has_err)
+                    self.parent.?.errFinish(alloc, io)
+                else
+                    self.parent.?.finish(alloc, io);
+            }
+            alloc.destroy(self);
+        }
+
+        if (self.has_err) return;
+
+        setMetadata(alloc, io, self.hdr, self.id_table, self.xattr_table, self.file, self.xattr_idx, self.options) catch |err| {
+            self.has_err = true;
+            self.err.* = err;
+            return;
+        };
+        self.file.link(io) catch |err| {
+            self.has_err = true;
+            self.err.* = err;
+            return;
+        };
+
+        self.file.deinit(io);
+    }
+    fn errFinish(self: *FileFinish, alloc: std.mem.Allocator, io: Io) void {
+        self.has_err = true;
+
+        if (self.atomic.fetchSub(1, .release) > 0) return;
+
+        if (self.parent != null) {
+            alloc.free(self.path);
+            self.parent.?.errFinish(alloc, io);
+        }
+        self.file.deinit(io);
+        alloc.destroy(self);
+    }
 };
+const Parent = struct {
+    id_table: *Lookup.Table(u16),
+    xattr_table: *XattrTable,
+    err: *?Error,
 
-const Error = error{MknodError} || Decomp.Error || Directory.Error || DataExtractor.Error || Io.Dir.CreateDirPathError ||
-    Io.Dir.SymLinkError || Io.File.Atomic.LinkError || Io.Reader.StreamRemainingError;
-
-const PathReturn = struct {
     hdr: Inode.Header,
     path: []const u8,
-    xattr_idx: ?u32 = null,
+    options: ExtractionOption,
+    parent: ?*Parent,
+    xattr_idx: u32,
+
+    atomic: Atomic(usize),
+    has_err: bool = false,
+
+    fn create(
+        alloc: std.mem.Allocator,
+        id_table: *Lookup.Table(u16),
+        xattr_table: *XattrTable,
+        err: *?Error,
+        hdr: Inode.Header,
+        path: []const u8,
+        options: ExtractionOption,
+        parent: ?*Parent,
+        xattr_idx: u32,
+        num: usize,
+    ) Error!*Parent {
+        const out = try alloc.create(Parent);
+        out.* = .{
+            .id_table = id_table,
+            .xattr_table = xattr_table,
+            .err = err,
+
+            .hdr = hdr,
+            .path = path,
+            .options = options,
+            .parent = parent,
+            .xattr_idx = xattr_idx,
+
+            .atomic = .init(num),
+        };
+        return out;
+    }
+
+    fn finish(self: *Parent, alloc: std.mem.Allocator, io: Io) void {
+        if (self.atomic.fetchSub(1, .release) > 0) return;
+
+        defer {
+            if (self.parent != null) {
+                alloc.free(self.path);
+                if (self.has_err)
+                    self.parent.?.errFinish(alloc, io)
+                else
+                    self.parent.?.finish(alloc, io);
+            }
+            alloc.destroy(self);
+        }
+
+        if (self.has_err) return;
+
+        setMetadataPath(alloc, io, self.hdr, self.id_table, self.xattr_table, self.path, self.xattr_idx, self.options) catch |err| {
+            self.has_err = true;
+            self.err.* = err;
+        };
+    }
+    fn errFinish(self: *Parent, alloc: std.mem.Allocator, io: Io) void {
+        self.has_err = true;
+
+        if (self.atomic.fetchSub(1, .release) > 0) return;
+
+        if (self.parent != null) {
+            alloc.free(self.path);
+            self.parent.?.errFinish(alloc, io);
+        }
+        alloc.destroy(self);
+    }
 };

src/extract-single.zig

@@ -1,6 +1,8 @@
 const std = @import("std");
 const Io = std.Io;
 
+const Archive = @import("archive.zig");
+const Superblock = Archive.Superblock;
 const DataReader = @import("data/reader.zig");
 const Decomp = @import("decomp.zig");
 const Directory = @import("directory.zig");
@@ -8,7 +10,6 @@ const ExtractionOptions = @import("options.zig");
 const Inode = @import("inode.zig");
 const Lookup = @import("lookup.zig");
 const MetadaReader = @import("meta_rdr.zig");
-const Superblock = @import("archive.zig").Superblock;
 const Cache = @import("util/cache.zig");
 const XattrTable = @import("xattr.zig");
 
@@ -183,6 +184,7 @@ fn extractFile(
         },
         else => unreachable,
     };
+    defer rdr.deinit();
 
     var atomic = try Io.Dir.cwd().createFileAtomic(io, path, .{});
     defer atomic.deinit(io);

src/extract.zig

@@ -1,12 +1,12 @@
 const std = @import("std");
 const Io = std.Io;
 
-const Superblock = @import("archive.zig").Superblock;
 const Decomp = @import("decomp.zig");
-const Inode = @import("inode.zig");
 const ExtractionOptions = @import("options.zig");
-const Single = @import("extract-single.zig");
+const Inode = @import("inode.zig");
 const Multi = @import("extract-multi.zig");
+const Single = @import("extract-single.zig");
+const Superblock = @import("archive.zig").Superblock;
 
 pub fn extract(
     alloc: std.mem.Allocator,

src/util/protected_map.zig

@@ -33,43 +33,37 @@ pub fn ProtectedMap(comptime K: anytype, comptime T: anytype, comptime create_fn
         pub fn getOrPut(self: *Map, io: Io, key: K, create_fn_args: std.meta.ArgsTuple(@TypeOf(create_fn))) Error!*T {
             {
                 try self.mut.lockShared(io);
-                defer self.mut.unlockShared(io);
-
                 const value = self.map.getPtr(key);
+                self.mut.unlockShared(io);
                 if (value != null) {
                     if (!value.?.filled.isSet())
                         try value.?.filled.wait(io);
+
                     if (value.?.err != null) return value.?.err.?;
                     return &value.?.value;
                 }
             }
-            try self.mut.lock(io);
+            var value: *ProtectedValue = blk: {
+                try self.mut.lock(io);
+                defer self.mut.unlock(io);
 
-            const res = self.map.getOrPut(key) catch |err| {
+                const res = try self.map.getOrPut(key);
-                self.mut.unlock(io);
+                if (res.found_existing)
-                return err;
+                    return self.getOrPut(io, key, create_fn_args);
+                res.value_ptr.* = .{};
+                break :blk res.value_ptr;
             };
-            if (res.found_existing) {
+            defer value.filled.set(io);
-                self.mut.unlock(io);
-                return self.getOrPut(io, key, create_fn_args);
-            }
-            res.value_ptr.* = .{};
-            defer res.value_ptr.filled.set(io);
 
-            self.mut.unlock(io);
+            value.value = if (@TypeOf(CreateError) == void)
-
-            self.mut.lockSharedUncancelable(io);
-            defer self.mut.unlockShared(io);
-
-            res.value_ptr.value = if (@TypeOf(CreateError) == void)
                 @call(.auto, create_fn, create_fn_args)
             else
                 @call(.auto, create_fn, create_fn_args) catch |err| {
-                    res.value_ptr.err = err;
+                    value.err = err;
                     return err;
                 };
 
-            return &res.value_ptr.value;
+            return &value.value;
         }
 
         // Map Types