diff --git a/src/data/extractor.zig b/src/data/extractor.zig index abbf4f7..9012520 100644 --- a/src/data/extractor.zig +++ b/src/data/extractor.zig @@ -3,7 +3,7 @@ const Io = std.Io; const Decomp = @import("../decomp.zig"); const FileFinish = @import("../extract-multi.zig").FileFinish; -const ExtractError = @import("../extract-multi.zig").Error; +const Multi = @import("../extract-multi.zig"); const DataBlock = @import("../inode.zig").DataBlock; const Cache = @import("../util/cache.zig"); @@ -43,16 +43,16 @@ pub fn addCache(self: *Extractor, cache: *Cache) void { self.cache = cache; } -pub fn extractAsync(self: Extractor, alloc: std.mem.Allocator, io: Io, group: *Io.Group, err: *?ExtractError, finish: *FileFinish) 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; var read_offset: u64 = self.start; for (0.., self.blocks) |i, block| { - group.async(io, blockThread, .{ self, alloc, io, finish.file.file, read_offset, @truncate(i), err, finish }); + select.async(.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, io, finish.file.file, err, finish }); + select.async(.reg, fragThread, .{ self, io, finish.file.file, finish }); } fn blockThread( @@ -62,9 +62,8 @@ fn blockThread( file: Io.File, read_offset: u64, block_idx: u32, - err: *?ExtractError, finish: *FileFinish, -) error{Canceled}!void { +) Multi.Error!void { const size = if (self.frag_data == null and block_idx == self.blocks.len - 1) self.size % self.block_size else @@ -73,89 +72,45 @@ fn blockThread( const block = self.blocks[block_idx]; var wrt = file.writer(io, &[0]u8{}); - wrt.seekTo(block_idx * self.block_size) catch |inner_err| { - err.* = inner_err; - return; - }; + try wrt.seekTo(block_idx * self.block_size); 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; - }; + try wrt.interface.splatByteAll(0, size); + try finish.finish(io); return; } const data = self.data[read_offset..][0..block.size]; if (block.uncompressed) { - wrt.interface.writeAll(data) catch |inner_err| { - err.* = inner_err; - return; - }; - finish.finish(io) catch |inner_err| { - err.* = inner_err; - }; + try wrt.interface.writeAll(data); + try finish.finish(io); return; } if (self.cache != null) { - const decomp_block = self.cache.?.get(io, read_offset, block.size) catch |inner_err| { - switch (inner_err) { - error.Canceled => return error.Canceled, - else => |e| err.* = e, - } - return; - }; - wrt.interface.writeAll(decomp_block) catch |inner_err| { - err.* = inner_err; - }; + const decomp_block = try self.cache.?.get(io, read_offset, block.size); + try wrt.interface.writeAll(decomp_block); } else { - const tmp = alloc.alloc(u8, size) catch |inner_err| { - err.* = inner_err; - return; - }; + const tmp = try alloc.alloc(u8, size); defer alloc.free(tmp); - _ = self.decomp(alloc, data, tmp) catch |inner_err| { - err.* = inner_err; - return; - }; - wrt.interface.writeAll(tmp) catch |inner_err| { - err.* = inner_err; - }; + _ = try self.decomp(alloc, data, tmp); + try wrt.interface.writeAll(tmp); } - finish.finish(io) catch |inner_err| { - err.* = inner_err; - }; + try finish.finish(io); } fn fragThread( self: Extractor, io: Io, file: Io.File, - err: *?ExtractError, finish: *FileFinish, -) error{Canceled}!void { +) Multi.Error!void { const size = self.size % self.block_size; var wrt = file.writer(io, &[0]u8{}); - wrt.seekTo(self.blocks.len * self.block_size) catch |inner_err| { - err.* = inner_err; - return; - }; + try wrt.seekTo(self.blocks.len * self.block_size); - 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; - }; + 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 || Io.File.SeekError || Io.Writer.Error; diff --git a/src/extract-multi.zig b/src/extract-multi.zig index 54b71cf..9542dd4 100644 --- a/src/extract-multi.zig +++ b/src/extract-multi.zig @@ -16,49 +16,22 @@ const XattrTable = @import("xattr.zig"); pub fn extract(alloc: std.mem.Allocator, io: Io, super: Superblock, data: []u8, decomp: Decomp.Fn, inode: Inode, filepath: []const u8, options: ExtractionOption) !void { const path = std.mem.trim(u8, filepath, "/"); - var common: Common = .init(alloc, super, data, decomp, options); + var common: Common = try .init(alloc, io, super, data, decomp, options); defer common.deinit(); - common.group.async(io, extractAsync, .{ &common, io, inode, path, null }); + common.start(io, inode, path, null); - try common.group.await(io); + var buf: [5]SelectUnion = undefined; + + while (common.select.group.token.load(.unordered)) |_| { + const num = try common.select.awaitMany(&buf, 1); + for (buf[0..num]) |res| + try res.reg; + } if (common.err != null) return common.err.?; } - -fn extractAsync(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Parent) error{Canceled}!void { - switch (inode.hdr.type) { - .dir, .ext_dir => extractDir(common, io, inode, path, parent) catch |err| switch (err) { - error.Canceled => { - io.recancel(); - return error.Canceled; - }, - else => common.err = err, - }, - .file, .ext_file => extractReg(common, io, inode, path, parent) catch |err| switch (err) { - error.Canceled => { - io.recancel(); - return error.Canceled; - }, - else => common.err = err, - }, - .symlink, .ext_symlink => extractReg(common, io, inode, path, parent) catch |err| switch (err) { - error.Canceled => { - io.recancel(); - return error.Canceled; - }, - else => common.err = err, - }, - else => extractNod(common, io, inode, path, parent) catch |err| switch (err) { - error.Canceled => { - io.recancel(); - return error.Canceled; - }, - else => common.err = err, - }, - } -} fn extractDir(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Parent) Error!void { var xattr_idx: u32 = 0xFFFFFFFF; @@ -102,7 +75,7 @@ fn extractDir(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ? return err; }; - common.group.async(io, extractAsync, .{ common, io, new_inode, new_path, cur_dir }); + common.start(io, new_inode, new_path, cur_dir); } } fn extractReg(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Parent) Error!void { @@ -149,7 +122,7 @@ fn extractReg(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ? const fin: *FileFinish = try .init(common, io, blocks, inode.hdr, path, xattr_idx, parent); - ext.extractAsync(common.alloc, io, &common.group, @ptrCast(&common.err), fin); + ext.extractAsync(common.alloc, io, &common.select, fin); } fn extractSymlink(common: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Parent) Error!void { defer if (parent != null) { @@ -259,7 +232,7 @@ fn setMetadata(common: *Common, io: Io, hdr: Inode.Header, file: Io.File, xattr_ // Types pub const Error = error{ Mknod, SetXattr } || std.mem.Allocator.Error || Io.Cancelable || Io.Reader.Error || Io.Dir.CreateDirPathError || Cache.Error || - Io.File.SetPermissionsError || DataExtractor.Error; + Io.File.SetPermissionsError || Io.Dir.SymLinkError || Io.File.SeekError || Io.Writer.Error; pub const Parent = struct { common: *Common, @@ -342,11 +315,14 @@ pub const FileFinish = struct { } }; +pub const SelectUnion = union { reg: Error!void }; + const Common = struct { alloc: std.mem.Allocator, arena: std.heap.ArenaAllocator, - group: Io.Group = .init, + sel_buf: []SelectUnion, + select: Io.Select(SelectUnion), err: ?Error = null, data: []u8, @@ -363,11 +339,15 @@ const Common = struct { options: ExtractionOption, - fn init(alloc: std.mem.Allocator, super: Superblock, data: []u8, decomp: Decomp.Fn, options: ExtractionOption) Common { + fn init(alloc: std.mem.Allocator, io: Io, super: Superblock, data: []u8, decomp: Decomp.Fn, options: ExtractionOption) !Common { + const sel_buf = try alloc.alloc(SelectUnion, 50); return .{ .alloc = alloc, .arena = .init(alloc), + .sel_buf = sel_buf, + .select = .init(io, sel_buf), + .data = data, .decomp = decomp, @@ -384,8 +364,11 @@ const Common = struct { }; } fn deinit(self: *Common) void { + self.select.cancelDiscard(); self.arena.deinit(); + self.alloc.free(self.sel_buf); + self.id_table.deinit(); self.frag_table.deinit(); self.xattr_table.deinit(); @@ -395,4 +378,13 @@ const Common = struct { fn arenaAlloc(self: *Common) std.mem.Allocator { return self.arena.allocator(); } + + fn start(self: *Common, io: Io, inode: Inode, path: []const u8, parent: ?*Parent) void { + switch (inode.hdr.type) { + .dir, .ext_dir => self.select.async(.reg, extractDir, .{ self, io, inode, path, parent }), + .file, .ext_file => self.select.async(.reg, extractReg, .{ self, io, inode, path, parent }), + .symlink, .ext_symlink => self.select.async(.reg, extractSymlink, .{ self, io, inode, path, parent }), + else => self.select.async(.reg, extractNod, .{ self, io, inode, path, parent }), + } + } };