Trying to fix race conditions...

src/data/extractor.zig

@@ -44,54 +44,50 @@ 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;
 
-    // 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);
-
     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 });
+        group.async(io, blockThread, .{ self, alloc, io, file, read_offset, @truncate(i), &err });
         read_offset += block.size;
     }
     if (self.frag_data != null)
-        group.async(io, fragThread, .{ self, map.memory });
+        group.async(io, fragThread, .{ self, io, file, &err });
 
     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(self: Extractor, alloc: std.mem.Allocator, io: Io, file: Io.File, read_offset: u64, block_idx: u32, err: *?Error) error{Canceled}!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{});
+    wrt.seekTo(block_idx * self.block_size) catch |inner_err| {
+        err.* = inner_err;
+        return;
+    };
+
     if (block.size == 0) {
-        @memset(map_data[offset..][0..size], 0);
+        wrt.interface.splatByteAll(0, size) catch |inner_err| {
+            err.* = inner_err;
+        };
         return;
     }
 
     const data = self.data[read_offset..][0..block.size];
 
     if (block.uncompressed) {
-        @memcpy(map_data[offset..][0..block.size], data);
+        wrt.interface.writeAll(data) catch |inner_err| {
+            err.* = inner_err;
+        };
         return;
     }
 
@@ -103,20 +99,39 @@ fn blockThread(self: Extractor, alloc: std.mem.Allocator, io: Io, map_data: []u8
             }
             return;
         };
-        @memcpy(map_data[offset..][0..size], decomp_block[0..size]);
+        wrt.interface.writeAll(decomp_block) catch |inner_err| {
+            err.* = inner_err;
+        };
     } else {
-        _ = self.decomp(alloc, data, map_data[offset..][0..size]) catch |inner_err| {
+        const tmp = alloc.alloc(u8, size) catch |inner_err| {
+            err.* = inner_err;
+            return;
+        };
+        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;
         };
     }
 }
-fn fragThread(self: Extractor, map_data: []u8) error{Canceled}!void {
+fn fragThread(self: Extractor, io: Io, file: Io.File, err: *?Error) error{Canceled}!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{});
+    wrt.seekTo(self.blocks.len * self.block_size) catch |inner_err| {
+        err.* = inner_err;
+        return;
+    };
+
+    wrt.interface.writeAll(self.frag_data.?[self.frag_offset..][0..size]) catch |inner_err| {
+        err.* = inner_err;
+    };
 }
 
 // Types
 
-pub const Error = Io.File.WritePositionalError || Io.File.MemoryMap.CreateError || Decomp.Error || Cache.Error;
+pub const Error = Io.File.WritePositionalError || Io.File.MemoryMap.CreateError || Decomp.Error || Cache.Error || Io.File.SeekError || Io.Writer.Error;

src/extract-multi.zig

@@ -37,463 +37,11 @@ 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 dirs: std.PriorityDequeue(DirReturn, Io.Mutex, compareDir) = .initContext(.init);
-    defer dirs.deinit(alloc);
-
-    errdefer while (dirs.popMax()) |d| {
-        if (d.hdr.num != inode.hdr.num)
-            alloc.free(d.path);
-    };
-
-    var atomic_err: ?Error = null;
-
-    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(
-    alloc: std.mem.Allocator,
-    io: Io,
-    super: Superblock,
-    data: []u8,
-    decomp: Decomp.Fn,
-    cache: *Cache,
-    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: *?Error,
-) error{Canceled}!void {
-    errdefer if (!origin) alloc.free(path);
-
-    var xattr_idx: u32 = 0xFFFFFFFF;
-
-    Io.Dir.cwd().createDirPath(io, path) catch |err| switch (err) {
-        error.Canceled => {
-            io.recancel();
-            return error.Canceled;
-        },
-        else => |e| {
-            atomic_err.* = e;
-            return;
-        },
-    };
-
-    _ = 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 Directory.init(alloc, &meta.interface, d.size) catch |err| break :blk err;
-            },
-            .ext_dir => |d| d_blk: {
-                xattr_idx = d.xattr_idx;
-
-                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 Directory.init(alloc, &meta.interface, d.size) catch |err| break :blk err;
-            },
-            else => unreachable,
-        };
-        defer dir.deinit(alloc);
-
-        for (dir.entries) |entry| {
-            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);
-                break :blk err;
-            };
-
-            switch (entry.type) {
-                .dir => group.async(io, extractDir, .{
-                    alloc,
-                    io,
-                    super,
-                    data,
-                    decomp,
-                    cache,
-                    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,
-                }),
-            }
-        }
-    } 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.* = e;
-            return;
-        },
-    };
-
-    extractor.extractAsync(alloc, io, atomic.file) catch |err| switch (err) {
-        error.Canceled => {
-            io.recancel();
-            return error.Canceled;
-        },
-        else => |e| {
-            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 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),
-    xattr_table: *XattrTable,
-    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,
-) 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);
-        defer xattr.deinit(alloc);
-
-        for (xattr.kvs) |kv| {
-            const res = std.os.linux.fsetxattr(fil.handle, kv.key, kv.value.ptr, kv.value.len, 0);
-            if (res != 0)
-                return error.SetXattrError;
-        }
-    }
-    if (!options.ignore_permissions) {
-        try fil.setTimestamps(io, .{
-            .modify_timestamp = .init(Io.Timestamp.fromNanoseconds(@as(i96, @intCast(hdr.mod_time)) * std.time.ns_per_s)),
-        });
-        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 Error = error{ MknodError, SetXattrError } || Decomp.Error || Directory.Error || DataExtractor.Error || Io.Dir.CreateDirPathError ||
-    Io.Dir.SymLinkError || Io.File.Atomic.LinkError || Io.Reader.StreamRemainingError;
+const Error = error{ MknodError, SetXattrError };
 
 const DirReturn = struct {
     hdr: Inode.Header,

src/util/protected_map.zig

@@ -33,43 +33,40 @@ 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())
+                    if (!value.?.filled.isSet()) {
+                        self.mut.unlockShared(io);
+                        defer self.mut.lockShared(io);
+
                         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| {
-                self.mut.unlock(io);
-                return err;
+                const res = try self.map.getOrPut(key);
+                if (res.found_existing)
+                    return self.getOrPut(io, key, create_fn_args);
+                res.value_ptr.* = .{};
+                break :blk res.value_ptr;
             };
-            if (res.found_existing) {
-                self.mut.unlock(io);
-                return self.getOrPut(io, key, create_fn_args);
-            }
-            res.value_ptr.* = .{};
-            defer res.value_ptr.filled.set(io);
+            defer value.filled.set(io);
 
-            self.mut.unlock(io);
-
-            self.mut.lockSharedUncancelable(io);
-            defer self.mut.unlockShared(io);
-
-            res.value_ptr.value = if (@TypeOf(CreateError) == void)
+            value.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