Single threaded accountsdb manager & misc changes

Author: InKryption

src/accountsdb/db.zig

@@ -1650,8 +1650,9 @@ pub const AccountsDB = struct {
                     self.unrooted_accounts.readWithLock();
                 defer unrooted_accounts_lg.unlock();
 
-                const accounts = (unrooted_accounts.get(account_ref.slot) orelse
-                    return error.SlotNotFound).items(.account);
+                const slots_and_accounts = unrooted_accounts.get(account_ref.slot) orelse
+                    return error.SlotNotFound;
+                const accounts: []Account = slots_and_accounts.items(.account);
                 const account = accounts[ref_info.index];
 
                 return try account.cloneOwned(self.allocator);

src/accountsdb/fuzz.zig

@@ -62,6 +62,8 @@ pub const RunCmd = struct {
     });
 };
 
+const Logger = sig.trace.Logger("accountsdb.fuzz");
+
 pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
     var prng_state: std.Random.DefaultPrng = .init(seed);
     const random = prng_state.random();
@@ -83,7 +85,7 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
         .max_buffer = 1 << 20,
     }, null);
     defer std_logger.deinit();
-    const logger = std_logger.logger("accountsdb.fuzz");
+    const logger: Logger = std_logger.logger("accountsdb.fuzz");
 
     const run_cmd: RunCmd = cmd: {
         var argv_list: std.ArrayListUnmanaged([]const u8) = .empty;
@@ -146,18 +148,13 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
     try accounts_db.account_index.expandRefCapacity(1_000_000);
 
     var manager_exit: std.atomic.Value(bool) = .init(false);
-    const manager_handle: std.Thread = try .spawn(.{}, sig.accounts_db.manager.runLoop, .{
-        &accounts_db, sig.accounts_db.manager.ManagerLoopConfig{
-            .exit = &manager_exit,
-            .slots_per_full_snapshot = 50_000,
-            .slots_per_incremental_snapshot = 5_000,
-            .zstd_nb_workers = @intCast(std.Thread.getCpuCount() catch 0),
-        },
+    var manager: sig.accounts_db.manager.Manager = try .init(allocator, &accounts_db, .{
+        .exit = &manager_exit,
+        .slots_per_full_snapshot = 50_000,
+        .slots_per_incremental_snapshot = 5_000,
+        .zstd_nb_workers = @intCast(std.Thread.getCpuCount() catch 0),
     });
-    defer {
-        manager_exit.store(true, .release);
-        manager_handle.join();
-    }
+    defer manager.deinit(allocator);
 
     var tracked_accounts_rw: sig.sync.RwMux(TrackedAccountsMap) = .init(.empty);
     defer {
@@ -185,13 +182,13 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
         // any validation (in the .get block of the main fuzzer
         // loop, we perform validation)
         threads.appendAssumeCapacity(try .spawn(.{}, readRandomAccounts, .{
+            logger,
             &accounts_db,
             &tracked_accounts_rw,
             seed + thread_i,
             &reader_exit,
             thread_i,
         }));
-        logger.debug().logf("started readRandomAccounts thread: {}", .{thread_i});
     }
 
     var largest_rooted_slot: Slot = 0;
@@ -206,9 +203,12 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
             logger.info().logf("reached max slots: {}", .{max_slots});
             break;
         };
-        defer switch (random.int(u2)) {
-            0 => slot += random.intRangeAtMost(Slot, 1, 2),
-            1, 2, 3 => {},
+        const will_inc_slot = switch (random.int(u2)) {
+            0 => true,
+            1, 2, 3 => false,
+        };
+        defer if (will_inc_slot) {
+            slot += random.intRangeAtMost(Slot, 1, 2);
         };
         try ancestors.addSlot(allocator, slot);
 
@@ -281,7 +281,10 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
 
                 const account =
                     try accounts_db.getAccountWithAncestors(&pubkey, &ancestors_sub) orelse {
-                        logger.err().logf("accounts_db missing tracked account '{}': {}", .{ pubkey, tracked_account });
+                        logger.err().logf(
+                            "accounts_db missing tracked account '{}': {}",
+                            .{ pubkey, tracked_account },
+                        );
                         return error.MissingAccount;
                     };
                 defer account.deinit(allocator);
@@ -297,13 +300,12 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
             },
         }
 
-        const create_new_root = random.boolean();
-        if (create_new_root) {
+        const create_new_root = will_inc_slot and random.int(u8) == 0;
+        if (create_new_root) snapshot_validation: {
             largest_rooted_slot = @min(slot, largest_rooted_slot + 2);
             accounts_db.largest_rooted_slot.store(largest_rooted_slot, .monotonic);
-        }
+            try manager.manage(allocator);
 
-        snapshot_validation: {
             // holding the lock here means that the snapshot archive(s) wont be deleted
             // since deletion requires a write lock
             const maybe_latest_snapshot_info, //
@@ -449,20 +451,21 @@ pub fn run(seed: u64, args: *std.process.ArgIterator) !void {
 }
 
 fn readRandomAccounts(
+    logger: Logger,
     db: *AccountsDB,
     tracked_accounts_rw: *sig.sync.RwMux(TrackedAccountsMap),
     seed: u64,
     exit: *std.atomic.Value(bool),
     thread_id: usize,
 ) void {
-    var prng = std.Random.DefaultPrng.init(seed);
+    logger.debug().logf("started readRandomAccounts thread: {}", .{thread_id});
+    defer logger.debug().logf("finishing readRandomAccounts thread: {}", .{thread_id});
+
+    var prng: std.Random.DefaultPrng = .init(seed);
     const random = prng.random();
 
     while (true) {
-        if (exit.load(.seq_cst)) {
-            std.debug.print("finishing readRandomAccounts thread: {}\n", .{thread_id});
-            return;
-        }
+        if (exit.load(.seq_cst)) return;
 
         var pubkeys: [50]Pubkey = undefined;
         {

src/accountsdb/manager.zig

@@ -301,8 +301,7 @@ fn flushSlot(db: *AccountsDB, slot: Slot) !FileId {
         const unrooted_accounts, var unrooted_accounts_lg = db.unrooted_accounts.readWithLock();
         defer unrooted_accounts_lg.unlock();
 
-        const pubkeys_and_accounts = unrooted_accounts.get(slot) orelse
-            return error.SlotNotFound;
+        const pubkeys_and_accounts = unrooted_accounts.get(slot) orelse return error.SlotNotFound;
         break :blk pubkeys_and_accounts;
     };