Refactor how updates work

Author: lalinsky

src/Index2.zig

@@ -266,54 +266,49 @@ fn loadSegments(self: *Self) !void {
 }
 
 fn doCheckpoint(self: *Self) !bool {
-    const start_time = std.time.milliTimestamp();
+    var segments = self.acquireSegments();
+    defer self.releaseSegments(&segments);
 
-    var src = self.readyForCheckpoint() orelse return false;
+    const source = segments.memory_segments.value.getFirst() orelse return false;
+    if (source.value.getSize() < self.options.min_segment_size) {
+        return false;
+    }
 
-    var src_reader = src.data.reader();
-    defer src_reader.close();
+    // build new file segment
 
-    var dest = try self.file_segments.createSegment(.{self.allocator});
-    errdefer self.file_segments.destroySegment(dest);
+    var target = try FileSegmentList.createSegment(self.allocator, .{ .dir = self.data_dir });
+    errdefer FileSegmentList.destroySegment(self.allocator, &target);
 
-    try dest.data.build(self.data_dir, &src_reader);
+    var reader = source.value.reader();
+    defer reader.close();
 
-    errdefer dest.data.delete(self.data_dir);
+    try target.value.build(&reader);
+    errdefer target.value.cleanup();
 
-    self.file_segments_lock.lock();
-    defer self.file_segments_lock.unlock();
+    // update memory segments list
 
-    var ids = try self.file_segments.getIdsAfterAppend(dest, self.allocator);
-    defer ids.deinit();
+    var memory_segments_update = try self.memory_segments.beginUpdate();
+    defer self.memory_segments.cleanupAfterUpdate(&memory_segments_update);
 
-    try filefmt.writeIndexFile(self.data_dir, ids.items);
+    memory_segments_update.removeSegment(source);
 
-    // we are about to remove segment from the memory_segments list
-    self.memory_segments_lock.lock();
-    defer self.memory_segments_lock.unlock();
+    // update file segments list
 
-    self.segments_lock.lock();
-    defer self.segments_lock.unlock();
+    var file_segments_update = try self.file_segments.beginUpdate();
+    defer self.file_segments.cleanupAfterUpdate(&file_segments_update);
 
-    log.info("stage stats size={}, len={}", .{ self.memory_segments.getTotalSize(), self.memory_segments.segments.len });
+    file_segments_update.appendSegment(target);
 
-    if (src != self.memory_segments.segments.first) {
-        std.debug.panic("checkpoint node is not first in list", .{});
-    }
+    try self.writeIndexFile(file_segments_update.segments.value);
 
-    if (self.file_segments.segments.last) |last_file_segment| {
-        if (last_file_segment.data.id.version >= dest.data.id.version) {
-            std.debug.panic("inconsistent versions between memory and file segments", .{});
-        }
-    }
+    // commit updated lists
 
-    self.file_segments.appendSegment(dest);
-    self.memory_segments.removeAndDestroySegment(src);
+    self.segments_lock.lock();
+    defer self.segments_lock.unlock();
 
-    log.info("saved changes up to commit {} to disk", .{dest.data.max_commit_id});
+    self.memory_segments.commitUpdate(&memory_segments_update);
+    self.file_segments.commitUpdate(&file_segments_update);
 
-    const end_time = std.time.milliTimestamp();
-    log.info("checkpoint took {} ms", .{end_time - start_time});
     return true;
 }
 
@@ -336,7 +331,7 @@ fn startCheckpointThread(self: *Self) !void {
     if (self.checkpoint_thread != null) return;
 
     log.info("starting checkpoint thread", .{});
-    // self.checkpoint_thread = try std.Thread.spawn(.{}, checkpointThreadFn, .{self});
+    self.checkpoint_thread = try std.Thread.spawn(.{}, checkpointThreadFn, .{self});
 }
 
 fn stopCheckpointThread(self: *Self) void {

src/segment_list2.zig

@@ -64,35 +64,33 @@ pub fn SegmentList(Segment: type) type {
         }
 
         pub fn appendSegmentInto(self: Self, copy: *Self, node: Node) void {
+            copy.nodes.clearRetainingCapacity();
             for (self.nodes.items) |n| {
                 copy.nodes.appendAssumeCapacity(n.acquire());
             }
             copy.nodes.appendAssumeCapacity(node.acquire());
         }
 
-        pub fn appendSegment(self: *Self, allocator: Allocator, node: Node) Allocator.Error!Self {
-            var copy = try Self.init(allocator, self.nodes.items.len + 1);
-            self.appendSegmentInto(&copy, node);
-            return copy;
-        }
-
-        pub fn removeSegment(self: *Self, allocator: Allocator, idx: usize) Allocator.Error!Self {
-            var copy = try Self.init(allocator, self.nodes.items.len - 1);
-            for (self.nodes.items, 0..) |n, i| {
-                if (i != idx) {
+        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());
                 }
             }
-            return copy;
         }
 
-        pub fn replaceSegments(self: *Self, allocator: Allocator, node: Node, start_idx: usize, end_idx: usize) Allocator.Error!Self {
-            var copy = try Self.init(allocator, self.nodes.items.len + 1 - (end_idx - start_idx));
-            for (self.nodes.items, 0..) |n, i| {
-                if (i < start_idx or i >= end_idx) {
+        pub fn replaceMergedSegmentInto(self: *Self, copy: *Self, node: Node) Self {
+            copy.nodes.clearRetainingCapacity();
+            var inserted_merged = false;
+            for (self.nodes.items) |n| {
+                if (node.value.id.contains(n.value.id)) {
+                    if (!inserted_merged) {
+                        copy.nodes.appendAssumeCapacity(node);
+                        inserted_merged = true;
+                    }
+                } else {
                     copy.nodes.appendAssumeCapacity(n.acquire());
-                } else if (i == start_idx) {
-                    copy.nodes.appendAssumeCapacity(node.acquire());
                 }
             }
             return copy;
@@ -232,8 +230,8 @@ 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, self.options);
-            errdefer List.destroySegment(self.allocator, &target);
+            var new_segment = try List.createSegment(self.allocator, self.options);
+            defer List.destroySegment(self.allocator, &new_segment);
 
             var merger = SegmentMerger(Segment).init(self.allocator, segments.value);
             defer merger.deinit();
@@ -243,64 +241,73 @@ 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();
-
-            var new_segments = try SharedPtr(List).create(self.allocator, undefined);
-            defer new_segments.release(self.allocator, .{self.allocator});
-
-            new_segments.value.* = try List.init(self.allocator, self.segments.value.nodes.items.len);
-            defer new_segments.value.deinit(self.allocator);
+            try new_segment.value.merge(&merger);
+            errdefer new_segment.value.cleanup();
 
-            var inserted_merged = false;
-            for (self.segments.value.nodes.items) |node| {
-                if (target.value.id.contains(node.value.id)) {
-                    if (!inserted_merged) {
-                        new_segments.value.nodes.appendAssumeCapacity(target);
-                        inserted_merged = true;
-                    }
-                } else {
-                    new_segments.value.nodes.appendAssumeCapacity(node.acquire());
-                }
-            }
+            var update = try self.beginUpdate();
+            defer self.cleanupAfterUpdate(&update);
 
-            try @call(.auto, preCommitFn, .{ ctx, new_segments.value });
+            try @call(.auto, preCommitFn, .{ ctx, update.segments.value });
 
             lock.lock();
             defer lock.unlock();
 
-            self.segments.swap(&new_segments);
+            self.commitUpdate(&update);
 
             return true;
         }
 
-        pub fn update(self: *Self, changes: []const Change, lock: *std.Thread.RwLock, preCommitFn: anytype, ctx: anytype) !bool {
-            var new_segment = try List.createSegment(self.allocator, self.options);
-            errdefer List.destroySegment(self.allocator, &new_segment);
+        pub const Update = struct {
+            manager: *Self,
+            segments: SharedPtr(List),
+            committed: bool = false,
 
-            try new_segment.value.build(changes);
-            errdefer new_segment.value.cleanup();
+            pub fn removeSegment(self: *@This(), node: List.Node) void {
+                self.manager.segments.value.removeSegmentInto(self.segments.value, node);
+            }
 
+            pub fn appendSegment(self: *@This(), node: List.Node) void {
+                self.manager.segments.value.appendSegmentInto(self.segments.value, node);
+            }
+
+            pub fn replaceMergedSegment(self: *@This(), node: List.Node) void {
+                self.manager.segments.value.replaceMergedSegmentInto(self.segments.value, node);
+            }
+        };
+
+        pub fn beginUpdate(self: *Self) !Update {
             self.update_lock.lock();
-            defer self.update_lock.unlock();
+            errdefer self.update_lock.unlock();
 
-            var new_segments_list = try self.segments.value.appendSegment(self.allocator, new_segment);
-            errdefer new_segments_list.deinit(self.allocator);
+            var segments = try SharedPtr(List).create(self.allocator, List.initEmpty());
+            errdefer self.releaseSegments(&segments);
 
-            var new_segments = try SharedPtr(List).create(self.allocator, new_segments_list);
-            defer new_segments.release(self.allocator, .{self.allocator});
+            try segments.value.nodes.ensureTotalCapacity(self.allocator, self.segments.value.nodes.items.len + 1);
 
-            try @call(.auto, preCommitFn, .{ ctx, new_segments.value });
+            return .{
+                .manager = self,
+                .segments = segments,
+            };
+        }
 
-            lock.lock();
-            defer lock.unlock();
+        pub fn commitUpdate(self: *Self, update: *Update) void {
+            self.segments.swap(&update.segments);
+            self.update_lock.unlock();
+            update.committed = true;
+        }
 
-            self.segments.swap(&new_segments);
+        pub fn cleanupAfterUpdate(self: *Self, update: *Update) void {
+            if (!update.committed) {
+                self.update_lock.unlock();
+            }
+            update.segments.releaseWithCleanup(self.allocator, destroySegmentList, .{self});
+        }
 
-            return true;
+        fn destroySegmentList(segments: *List, self: *Self) void {
+            for (segments.nodes.items) |node| {
+                node.value.cleanup();
+            }
+            segments.deinit(self.allocator);
         }
     };
 }

src/utils/smartptr.zig

@@ -71,7 +71,7 @@ pub fn SharedPtr(comptime T: type) type {
             }
         }
 
-        pub fn releaseWithCleanup(self: *Self, allocator: Allocator, cleanupFn: fn (*T) void, extra_args: anytype) void {
+        pub fn releaseWithCleanup(self: *Self, allocator: Allocator, cleanupFn: anytype, extra_args: anytype) void {
             const inner_ptr = self.getInnerPtr();
             if (inner_ptr.refs.decr()) {
                 @call(.auto, cleanupFn, .{&inner_ptr.value} ++ extra_args);