Unify segment cleanup

Author: lalinsky

src/FileSegment.zig

@@ -13,7 +13,12 @@ const filefmt = @import("filefmt.zig");
 
 const Self = @This();
 
+pub const Options = struct {
+    dir: std.fs.Dir,
+};
+
 allocator: std.mem.Allocator,
+options: Options,
 id: SegmentId = .{ .version = 0, .included_merges = 0 },
 max_commit_id: u64 = 0,
 docs: std.AutoHashMap(u32, bool),
@@ -25,9 +30,10 @@ num_items: usize = 0,
 
 raw_data: ?[]align(std.mem.page_size) u8 = null,
 
-pub fn init(allocator: std.mem.Allocator) Self {
+pub fn init(allocator: std.mem.Allocator, options: Options) Self {
     return Self{
         .allocator = allocator,
+        .options = options,
         .docs = std.AutoHashMap(u32, bool).init(allocator),
         .index = std.ArrayList(u32).init(allocator),
         .blocks = undefined,
@@ -76,23 +82,23 @@ pub fn search(self: Self, sorted_hashes: []const u32, results: *SearchResults) !
     }
 }
 
-pub fn open(self: *Self, dir: std.fs.Dir, id: SegmentId) !void {
-    try filefmt.readSegmentFile(dir, id, self);
+pub fn open(self: *Self, id: SegmentId) !void {
+    try filefmt.readSegmentFile(self.options.dir, id, self);
 }
 
-pub fn delete(self: *Self, dir: std.fs.Dir) void {
+pub fn delete(self: *Self) void {
     var file_name_buf: [filefmt.max_file_name_size]u8 = undefined;
     const file_name = filefmt.buildSegmentFileName(&file_name_buf, self.id);
 
     log.info("deleting segment file {s}", .{file_name});
 
-    dir.deleteFile(file_name) catch |err| {
+    self.options.deleteFile(file_name) catch |err| {
         log.err("failed to clean up segment file {s}: {}", .{ file_name, err });
     };
 }
 
-pub fn cleanup(self: *Self, dir: std.fs.Dir) void {
-    self.delete(dir);
+pub fn cleanup(self: *Self) void {
+    self.delete();
 }
 
 pub fn build(self: *Self, dir: std.fs.Dir, source: anytype) !void {

src/Index2.zig

@@ -91,20 +91,30 @@ pub fn init(allocator: std.mem.Allocator, dir: std.fs.Dir, options: Options) !Se
     var oplog_dir = try dir.makeOpenPath("oplog", .{ .iterate = true });
     errdefer oplog_dir.close();
 
-    const memory_segments = try SegmentListManager(MemorySegment).init(allocator, .{
-        .min_segment_size = 100,
-        .max_segment_size = options.min_segment_size,
-        .segments_per_level = 5,
-        .segments_per_merge = 10,
-        .max_segments = 16,
-    });
-
-    const file_segments = try SegmentListManager(FileSegment).init(allocator, .{
-        .min_segment_size = options.min_segment_size,
-        .max_segment_size = options.max_segment_size,
-        .segments_per_level = 10,
-        .segments_per_merge = 10,
-    });
+    const memory_segments = try SegmentListManager(MemorySegment).init(
+        allocator,
+        .{},
+        .{
+            .min_segment_size = 100,
+            .max_segment_size = options.min_segment_size,
+            .segments_per_level = 5,
+            .segments_per_merge = 10,
+            .max_segments = 16,
+        },
+    );
+
+    const file_segments = try SegmentListManager(FileSegment).init(
+        allocator,
+        .{
+            .dir = data_dir,
+        },
+        .{
+            .min_segment_size = options.min_segment_size,
+            .max_segment_size = options.max_segment_size,
+            .segments_per_level = 10,
+            .segments_per_merge = 10,
+        },
+    );
 
     return .{
         .options = options,
@@ -220,10 +230,10 @@ const Updater = struct {
 };
 
 fn loadSegment(self: *Self, segment_id: SegmentId) !FileSegmentNode {
-    var node = try FileSegmentList.createSegment(self.allocator, .{});
+    var node = try FileSegmentList.createSegment(self.allocator, .{ .dir = self.data_dir });
     errdefer FileSegmentList.destroySegment(self.allocator, &node);
 
-    try node.value.open(self.data_dir, segment_id);
+    try node.value.open(segment_id);
 
     return node;
 }

src/MemorySegment.zig

@@ -14,14 +14,17 @@ const SegmentMerger = @import("segment_merger.zig").SegmentMerger;
 
 const Self = @This();
 
+pub const Options = struct {};
+
 allocator: std.mem.Allocator,
 id: SegmentId = .{ .version = 0, .included_merges = 0 },
 max_commit_id: u64 = 0,
 docs: std.AutoHashMap(u32, bool),
 items: std.ArrayList(Item),
 frozen: bool = false,
 
-pub fn init(allocator: std.mem.Allocator) Self {
+pub fn init(allocator: std.mem.Allocator, opts: Options) Self {
+    _ = opts;
     return .{
         .allocator = allocator,
         .docs = std.AutoHashMap(u32, bool).init(allocator),

src/segment_list2.zig

@@ -54,8 +54,8 @@ pub fn SegmentList(Segment: type) type {
             self.nodes.deinit(allocator);
         }
 
-        pub fn createSegment(allocator: Allocator, extra_init_args: anytype) Allocator.Error!Node {
-            return Node.create(allocator, @call(.auto, Segment.init, .{allocator} ++ extra_init_args));
+        pub fn createSegment(allocator: Allocator, options: Segment.Options) Allocator.Error!Node {
+            return Node.create(allocator, @call(.auto, Segment.init, .{ allocator, options }));
         }
 
         pub fn destroySegment(allocator: Allocator, node: *Node) void {
@@ -172,15 +172,17 @@ pub fn SegmentListManager(Segment: type) type {
         pub const MergePolicy = TieredMergePolicy(List.Node, getSegmentSize(Segment));
 
         allocator: Allocator,
+        options: Segment.Options,
         segments: SharedPtr(List),
         merge_policy: MergePolicy,
         num_allowed_segments: std.atomic.Value(usize),
         update_lock: std.Thread.Mutex,
 
-        pub fn init(allocator: Allocator, merge_policy: MergePolicy) !Self {
+        pub fn init(allocator: Allocator, options: Segment.Options, merge_policy: MergePolicy) !Self {
             const segments = try SharedPtr(List).create(allocator, List.initEmpty());
             return Self{
                 .allocator = allocator,
+                .options = options,
                 .segments = segments,
                 .merge_policy = merge_policy,
                 .num_allowed_segments = std.atomic.Value(usize).init(0),
@@ -229,7 +231,7 @@ pub fn SegmentListManager(Segment: type) type {
 
             const candidate = self.merge_policy.findSegmentsToMerge(segments.value.nodes.items) orelse return false;
 
-            var target = try List.createSegment(self.allocator, .{});
+            var target = try List.createSegment(self.allocator, self.options);
             errdefer List.destroySegment(self.allocator, &target);
 
             var merger = SegmentMerger(Segment).init(self.allocator, segments.value);
@@ -241,6 +243,7 @@ pub fn SegmentListManager(Segment: type) type {
             try merger.prepare();
 
             try target.value.merge(&merger);
+            errdefer target.value.cleanup();
 
             self.update_lock.lock();
             defer self.update_lock.unlock();