Fix file segments not being deleted on merge

Author: lalinsky

src/FileSegment.zig

@@ -6,6 +6,7 @@ const common = @import("common.zig");
 const Item = common.Item;
 const SearchResults = common.SearchResults;
 const SegmentId = common.SegmentId;
+const KeepOrDelete = common.KeepOrDelete;
 
 const MemorySegment = @import("MemorySegment.zig");
 
@@ -27,6 +28,7 @@ block_size: usize = 0,
 blocks: []const u8,
 merged: u32 = 0,
 num_items: usize = 0,
+delete_in_deinit: bool = false,
 
 raw_data: ?[]align(std.mem.page_size) u8 = null,
 
@@ -40,14 +42,18 @@ pub fn init(allocator: std.mem.Allocator, options: Options) Self {
     };
 }
 
-pub fn deinit(self: *Self) void {
+pub fn deinit(self: *Self, delete_file: KeepOrDelete) void {
     self.docs.deinit();
     self.index.deinit();
 
     if (self.raw_data) |data| {
         std.posix.munmap(data);
         self.raw_data = null;
     }
+
+    if (delete_file == .delete) {
+        self.delete();
+    }
 }
 
 pub fn getBlockData(self: Self, block: usize) []const u8 {
@@ -126,7 +132,7 @@ test "build" {
     defer data_dir.close();
 
     var source = MemorySegment.init(std.testing.allocator, .{});
-    defer source.deinit();
+    defer source.deinit(.delete);
 
     source.id.version = 1;
     source.frozen = true;
@@ -138,7 +144,7 @@ test "build" {
     defer source_reader.close();
 
     var segment = Self.init(std.testing.allocator, .{ .dir = tmp_dir.dir });
-    defer segment.deinit();
+    defer segment.deinit(.delete);
 
     try segment.build(&source_reader);
 

src/Index.zig

@@ -134,8 +134,8 @@ pub fn deinit(self: *Self) void {
     self.stopMemorySegmentMergeThread();
     self.stopFileSegmentMergeThread();
 
-    self.memory_segments.deinit();
-    self.file_segments.deinit();
+    self.memory_segments.deinit(.keep);
+    self.file_segments.deinit(.keep);
 
     self.oplog.deinit();
     self.dir.close();
@@ -255,7 +255,7 @@ fn doCheckpoint(self: *Self) !bool {
     // build new file segment
 
     var target = try FileSegmentList.createSegment(self.allocator, .{ .dir = self.dir });
-    errdefer FileSegmentList.destroySegment(self.allocator, &target);
+    defer FileSegmentList.destroySegment(self.allocator, &target);
 
     var reader = source.value.reader();
     defer reader.close();
@@ -341,6 +341,7 @@ fn maybeMergeFileSegments(self: *Self) !bool {
     defer self.segments_lock.unlock();
 
     self.file_segments.commitUpdate(&upd);
+    log.debug("committed file segments merge", .{});
 
     return true;
 }
@@ -383,6 +384,7 @@ fn maybeMergeMemorySegments(self: *Self) !bool {
     defer self.segments_lock.unlock();
 
     self.memory_segments.commitUpdate(&upd);
+    log.debug("committed memory segments merge", .{});
 
     self.maybeScheduleCheckpoint();
 

src/MemorySegment.zig

@@ -5,6 +5,7 @@ const common = @import("common.zig");
 const Item = common.Item;
 const SearchResults = common.SearchResults;
 const SegmentId = common.SegmentId;
+const KeepOrDelete = common.KeepOrDelete;
 
 const Change = @import("change.zig").Change;
 
@@ -32,9 +33,11 @@ pub fn init(allocator: std.mem.Allocator, opts: Options) Self {
     };
 }
 
-pub fn deinit(self: *Self) void {
+pub fn deinit(self: *Self, delete_file: KeepOrDelete) void {
+    _ = delete_file;
     self.docs.deinit();
     self.items.deinit();
+    log.debug("deinit memory segment {}:{}", .{ self.id.version, self.id.included_merges });
 }
 
 pub fn search(self: Self, sorted_hashes: []const u32, results: *SearchResults) !void {

src/common.zig

@@ -3,6 +3,11 @@ const testing = std.testing;
 
 const msgpack = @import("msgpack");
 
+pub const KeepOrDelete = enum {
+    keep,
+    delete,
+};
+
 pub const Item = packed struct(u64) {
     id: u32,
     hash: u32,

src/filefmt.zig

@@ -199,7 +199,7 @@ pub fn encodeBlock(data: []u8, reader: anytype) !u16 {
 
 test "writeBlock/readBlock/readFirstItemFromBlock" {
     var segment = MemorySegment.init(std.testing.allocator, .{});
-    defer segment.deinit();
+    defer segment.deinit(.delete);
 
     try segment.items.append(.{ .hash = 1, .id = 1 });
     try segment.items.append(.{ .hash = 2, .id = 1 });
@@ -457,7 +457,7 @@ test "writeFile/readFile" {
 
     {
         var in_memory_segment = MemorySegment.init(testing.allocator, .{});
-        defer in_memory_segment.deinit();
+        defer in_memory_segment.deinit(.delete);
 
         in_memory_segment.id = version;
 
@@ -473,7 +473,7 @@ test "writeFile/readFile" {
 
     {
         var segment = FileSegment.init(testing.allocator, .{ .dir = tmp.dir });
-        defer segment.deinit();
+        defer segment.deinit(.delete);
 
         try readSegmentFile(tmp.dir, version, &segment);
 

src/segment_list.zig

@@ -5,6 +5,7 @@ const SearchResults = @import("common.zig").SearchResults;
 
 const Change = @import("change.zig").Change;
 const SegmentId = @import("common.zig").SegmentId;
+const KeepOrDelete = @import("common.zig").KeepOrDelete;
 
 const Deadline = @import("utils/Deadline.zig");
 
@@ -48,9 +49,9 @@ pub fn SegmentList(Segment: type) type {
             return try SharedPtr(Self).create(allocator, self);
         }
 
-        pub fn deinit(self: *Self, allocator: Allocator) void {
+        pub fn deinit(self: *Self, allocator: Allocator, delete_files: KeepOrDelete) void {
             for (self.nodes.items) |*node| {
-                node.release(allocator, .{});
+                node.release(allocator, .{delete_files});
             }
             self.nodes.deinit(allocator);
         }
@@ -60,25 +61,11 @@ pub fn SegmentList(Segment: type) type {
         }
 
         pub fn destroySegment(allocator: Allocator, segment: *Node) void {
-            segment.releaseWithCleanup(allocator, destroySegmentCallback, .{});
+            segment.release(allocator, .{.delete});
         }
 
         pub fn destroySegments(allocator: Allocator, segments: *SharedPtr(Self)) void {
-            // we also call cleanup on these segments, to ensure that unused segments will get deleted from disk
-            segments.releaseWithCleanup(allocator, destroySegmentListCallback, .{allocator});
-        }
-
-        fn destroySegmentListCallback(segments: *Self, allocator: Allocator) void {
-            while (segments.nodes.items.len > 0) {
-                var node = segments.nodes.pop();
-                destroySegment(allocator, &node);
-            }
-            segments.deinit(allocator);
-        }
-
-        fn destroySegmentCallback(segment: *Segment) void {
-            segment.cleanup();
-            segment.deinit();
+            segments.release(allocator, .{ allocator, .delete });
         }
 
         pub fn appendSegmentInto(self: Self, copy: *Self, node: Node) void {
@@ -87,13 +74,16 @@ pub fn SegmentList(Segment: type) type {
                 copy.nodes.appendAssumeCapacity(n.acquire());
             }
             copy.nodes.appendAssumeCapacity(node.acquire());
+            std.log.debug("adding {s} {}:{}", .{ @typeName(Segment), node.value.id.version, node.value.id.included_merges });
         }
 
         pub fn removeSegmentInto(self: Self, copy: *Self, node: Node) void {
             copy.nodes.clearRetainingCapacity();
             for (self.nodes.items) |n| {
                 if (n.value != node.value) {
                     copy.nodes.appendAssumeCapacity(n.acquire());
+                } else {
+                    std.log.debug("removing {s} {}:{}", .{ @typeName(Segment), n.value.id.version, n.value.id.included_merges });
                 }
             }
         }
@@ -103,9 +93,11 @@ pub fn SegmentList(Segment: type) type {
             var inserted_merged = false;
             for (self.nodes.items) |n| {
                 if (node.value.id.contains(n.value.id)) {
+                    std.log.debug("removing {s} {}:{}", .{ @typeName(Segment), n.value.id.version, n.value.id.included_merges });
                     if (!inserted_merged) {
                         copy.nodes.appendAssumeCapacity(node.acquire());
                         inserted_merged = true;
+                        std.log.debug("adding {s} {}:{}", .{ @typeName(Segment), node.value.id.version, node.value.id.included_merges });
                     }
                 } else {
                     copy.nodes.appendAssumeCapacity(n.acquire());
@@ -207,8 +199,8 @@ pub fn SegmentListManager(Segment: type) type {
             };
         }
 
-        pub fn deinit(self: *Self) void {
-            self.segments.release(self.allocator, .{self.allocator});
+        pub fn deinit(self: *Self, delete_files: KeepOrDelete) void {
+            self.segments.release(self.allocator, .{ self.allocator, delete_files });
         }
 
         pub fn count(self: Self) usize {
@@ -284,7 +276,7 @@ pub fn SegmentListManager(Segment: type) type {
             errdefer self.update_lock.unlock();
 
             var segments = try SharedPtr(List).create(self.allocator, List.initEmpty());
-            errdefer self.releaseSegments(&segments);
+            errdefer self.destroySegments(&segments);
 
             // allocate memory for one extra segment, if it's going to be unused, it's going to be unused, but we need to have it ready
             try segments.value.nodes.ensureTotalCapacity(self.allocator, self.count() + 1);

src/segment_merger.zig

@@ -149,7 +149,7 @@ test "merge segments" {
     const MemorySegment = @import("MemorySegment.zig");
 
     var collection = try SegmentList(MemorySegment).init(std.testing.allocator, 3);
-    defer collection.deinit(std.testing.allocator);
+    defer collection.deinit(std.testing.allocator, .delete);
 
     var merger = SegmentMerger(MemorySegment).init(std.testing.allocator, &collection);
     defer merger.deinit();