Some minor work with mutli-threaded extraction
This commit is contained in:
+20
-65
@@ -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;
|
||||
|
||||
+33
-41
@@ -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 }),
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user