diff --git a/v2/build.zig b/v2/build.zig index 725b748c1a..e5c202b2df 100644 --- a/v2/build.zig +++ b/v2/build.zig @@ -333,6 +333,13 @@ const Sig = struct { .{ .name = "services", .module = services_mod }, }, }); + + // accounts_db creates a lot of large globals. + // convert globals from [rip + 32-bit offset] to 64bit offsets + if (std.mem.eql(u8, service.name, "accounts_db")) { + service_mod.code_model = .medium; + } + unit_tests.add(service.name, service_mod); const service_lib = b.addLibrary(.{ diff --git a/v2/config/example.zon b/v2/config/example.zon index 1019996373..a667520275 100644 --- a/v2/config/example.zon +++ b/v2/config/example.zon @@ -23,5 +23,6 @@ .file = "./validator/accounts.db", .rooted = .{ .gb = 8 }, // mainnet needs 64, testnet only 8 or so .unrooted = .{ .gb = 8 }, // TODO + .metadata = .{ .gb = 1 }, }, } diff --git a/v2/init/main.zig b/v2/init/main.zig index 5aa833b072..8786760217 100644 --- a/v2/init/main.zig +++ b/v2/init/main.zig @@ -66,6 +66,7 @@ const Config = struct { file: []const u8, rooted: MemorySize, unrooted: MemorySize, + metadata: MemorySize, // For nicer initialization of constants (instead of x * 1024 * 1024 * 1024) const MemorySize = union(enum) { @@ -258,8 +259,10 @@ pub fn main() !void { var snapshot_ready_to_accounts_db: Region(lib.snapshot.SnapshotData) = try .simple(); snapshot_ready_to_accounts_db.ptr().init(); - var snapshot_metadata: Region(lib.accounts_db.RuntimeMetadata) = try .simple(); - snapshot_metadata.ptr().init(); + var snapshot_metadata: Region(lib.snapshot.SnapshotMetadata) = try .sized( + @sizeOf(lib.snapshot.SnapshotMetadata) + config.accounts_db.metadata.toBytes(), + ); + snapshot_metadata.ptr().init(config.accounts_db.metadata.toBytes()); const unrooted_memory = config.accounts_db.unrooted.toBytes(); var account_pool: Region(lib.accounts_db.AccountPool) = diff --git a/v2/init/services.zig b/v2/init/services.zig index 10b8905759..15fbf60539 100644 --- a/v2/init/services.zig +++ b/v2/init/services.zig @@ -15,7 +15,7 @@ pub const accounts_db = struct { pub const ReadWrite = struct { config: *lib.accounts_db.RootedConfig, ready_snapshot_in: *lib.snapshot.SnapshotData, - snapshot_metadata_out: *lib.accounts_db.RuntimeMetadata, + snapshot_metadata_out: *lib.snapshot.SnapshotMetadata, account_pool: *lib.accounts_db.AccountPool, replay_lookups: *lib.accounts_db.AccountLookups, tel: *lib.telemetry.Region, @@ -60,7 +60,7 @@ pub const replay = struct { pub const ReadWrite = struct { scratch_memory: *[lib.replay.scratch_buffer_size]u8, - snapshot_metadata_in: *lib.accounts_db.RuntimeMetadata, + snapshot_metadata_in: *lib.snapshot.SnapshotMetadata, deshredded_in: *lib.shred.DeshredRing, replay_transaction_pool: *lib.replay.TransactionPool, block_pool: *lib.replay.BlockPool, @@ -78,7 +78,7 @@ pub const shred_receiver = struct { pub const ReadWrite = struct { /// Gets slot (& soon leader-schedule info) from replay / runtime init. - snapshot_metadata: *lib.accounts_db.RuntimeMetadata, + snapshot_metadata: *lib.snapshot.SnapshotMetadata, /// Transaction Validation Unit (TVU) UDP socket, i.e. where we receive /// shreds. This is typically port 8002. While we've obtained a net diff --git a/v2/lib/accounts_db.zig b/v2/lib/accounts_db.zig index 5e337070d0..722cca7c72 100644 --- a/v2/lib/accounts_db.zig +++ b/v2/lib/accounts_db.zig @@ -10,8 +10,6 @@ comptime { } const Pubkey = lib.solana.Pubkey; -const Hash = lib.solana.Hash; -const Slot = lib.solana.Slot; pub const AccountPool = @import("accounts_db/pool.zig").AccountPool; pub const Rooted = @import("accounts_db/rooted.zig").Rooted; @@ -40,52 +38,3 @@ pub const AccountLookups = extern struct { self.out.init(); } }; - -/// How to consume this struct: -/// 1. read all ring buffers in the correct order (specify the correct order -/// here if more are added, currently there is only one: the blockhash queue) -/// 2. call getSlotBlocking -/// 3. read other fields -pub const RuntimeMetadata = extern struct { - slot: std.atomic.Value(u64), - /// The merkle root of the last fec set (tower) or the root of roots (alpenglow) - block_id: Hash, - blockhash_queue: extern struct { - /// read after consuming all of hashes - max_age: u64, - /// Accountsdb blocks until enough hashes are read from here to make - /// room for accountsdb to write all of its block hashes here. - hashes: lib.ipc.Ring(256, Hash), - }, - - // 0 may be a valid slot, so use something that will never be reached. - const invalid_slot = std.math.maxInt(Slot); - - pub fn init(self: *RuntimeMetadata) void { - self.slot = .init(invalid_slot); - - self.blockhash_queue.max_age = 0; - self.blockhash_queue.hashes.init(); - } - - /// Unblocks all getSlotBlocking() callers with the given slot value. - /// Can be called only once. - /// Should also only call after all other RuntimeMetadata fields are populated. - pub fn populateSlot(self: *RuntimeMetadata, slot: Slot) void { - std.debug.assert(slot != invalid_slot); - std.debug.assert(self.slot.swap(slot, .release) == invalid_slot); - } - - /// Accountsdb writes the slot last, so you need to empty all the ring - /// buffers before calling this. - pub fn getSlotBlocking(self: *RuntimeMetadata, runner: lib.runner.Connection) !Slot { - while (true) { - const slot = self.slot.load(.acquire); - if (slot != invalid_slot) { - try runner.activity.signalActive(); - return slot; - } - try runner.activity.signalIdleSpinning(); - } - } -}; diff --git a/v2/lib/accounts_db/rooted.zig b/v2/lib/accounts_db/rooted.zig index 12976d41db..dcca31cefc 100644 --- a/v2/lib/accounts_db/rooted.zig +++ b/v2/lib/accounts_db/rooted.zig @@ -10,14 +10,19 @@ const FileReader = lib.fio.FileReader; const Table = lib.accounts_db.Table; const AccountPool = lib.accounts_db.AccountPool; const AccountLookups = lib.accounts_db.AccountLookups; -const RuntimeMetadata = lib.accounts_db.RuntimeMetadata; + +const SnapshotMetadata = lib.snapshot.SnapshotMetadata; const Pubkey = lib.solana.Pubkey; const Slot = lib.solana.Slot; const Epoch = lib.solana.Epoch; -const Hash = lib.solana.Hash; -/// The rooted database stores data on disk in the form of [journal][ring of sector data]. +const Manifest = lib.solana.snapshot.Manifest; +const StatusCache = lib.solana.snapshot.StatusCache; +const SnapshotIter = lib.solana.snapshot.SnapshotIter; + +/// The rooted database stores data on disk in the form of +/// [journal][manifest+status_cache+FBA blob][ring of account sectors]. /// TODO: The ring aspect is not implemented, so for now it grows indefinitely. pub const Rooted = struct { table: Table, @@ -65,6 +70,8 @@ pub const Rooted = struct { committed_slot: u32 align(1), committed_offset: u64 align(1), blockhash_max_age: u32 align(1), + /// Padded on-disk size of the SnapshotMetadata blob. 0 means "no metadata yet". + metadata_bytes: u32 align(1), const empty: Journal = .{ .magic = .valid, @@ -73,6 +80,7 @@ pub const Rooted = struct { .committed_slot = 0, .committed_offset = 0, .blockhash_max_age = 300, + .metadata_bytes = 0, }; }; comptime { @@ -83,12 +91,11 @@ pub const Rooted = struct { pub fn init( self: *Rooted, logger: tel.Logger("Rooted.init"), - runner: lib.runner.Connection, dir: std.fs.Dir, path: []const u8, table_memory: []u8, account_pool: *AccountPool, - runtime_metadata: *RuntimeMetadata, + snapshot_metadata: *SnapshotMetadata, ) !void { const seed: u64 = 0; // TODO: maybe in RootedConfig? self.put_batch = .empty; @@ -105,9 +112,8 @@ pub const Rooted = struct { logger.info().log("loading from existing rooted db"); self.loadExisting( .from(logger), - runner, read_file, - runtime_metadata, + snapshot_metadata, ) catch |err| switch (err) { error.InvalidJournal => { self.journal = .empty; // reset any modifications from loadExisting() @@ -168,9 +174,6 @@ pub const Rooted = struct { padding, /// holds an actual account account, - /// holds the block_id followed by a serialized addition to the blockhash queue. - /// body layout: `{ block_id: Hash, [info.count]Hash }` - block_metadata, _, // TODO: add other types of sections }, info: packed union { @@ -197,9 +200,8 @@ pub const Rooted = struct { fn loadExisting( self: *Rooted, logger: tel.Logger("Rooted.loadExisting"), - runner: lib.runner.Connection, file: std.fs.File, - runtime_metadata: *RuntimeMetadata, + snapshot_metadata: *SnapshotMetadata, ) !void { const zone = tracy.Zone.init(@src(), .{ .name = "Rooted.loadExisting" }); defer zone.deinit(); @@ -253,18 +255,32 @@ pub const Rooted = struct { logger.info().logf("read journal: {any}", .{self.journal}); } - var blockhash_writer = runtime_metadata.blockhash_queue.hashes.getView(.writer); - defer { - runtime_metadata.blockhash_queue.max_age = self.journal.blockhash_max_age; - blockhash_writer.close(); // close when done. + // read the persisted Manifest + StatusCache + FBA blob back into place. + // Layout on disk: + // [Journal (block_size)] + // [manifest + status_cache + memory (metadata_bytes)] + // [account sectors...] + { + const disk_bytes = self.journal.metadata_bytes; + const metadata = snapshot_metadata.getSerializable(); + const padded = std.mem.alignForward(u64, metadata.len, block_size); + if (disk_bytes == 0 or disk_bytes != padded) { + logger.err().logf( + "invalid metadata_bytes: disk={}, expected={}", + .{ disk_bytes, padded }, + ); + return error.InvalidJournal; + } + + try self.readExisting(.from(logger), metadata.ptr, metadata.len); // read metadata + try self.readExisting(.from(logger), null, padded - metadata.len); // skip padding } var timer = try std.time.Timer.start(); var n_puts: usize = 0; var n_bytes_read: usize = 0; - var last_block_id: ?Hash = null; - // read sectors until EOF + // read account sectors until EOF while ((try self.io.reader.getBuffer(.from(logger))).len > 0) { const file_offset = self.io.reader.getOffset(); if (file_offset >= self.journal.committed_offset) break; // ignore overrun file data @@ -305,38 +321,6 @@ pub const Rooted = struct { }); n_puts += 1; }, - .block_metadata => { - // read block_id - var block_id: Hash = undefined; - try self.readExisting(.from(logger), (&block_id.data).ptr, @sizeOf(Hash)); - last_block_id = block_id; - - // read blockhashes - var num_hashes = header.info.count; - while (num_hashes > 0) { - // get a buffer to write hashes into - const hash_buf: []Hash = try blockhash_writer.getBufferBlocking(runner); - if (hash_buf.len == 0) { - // blockhash_reader closed their end. just skip the hashes then. - try self.readExisting( - .from(logger), - null, - @sizeOf(Hash) * num_hashes, - ); - break; - } - - // read hashes into blockhash_writer to send over - const take = @min(num_hashes, hash_buf.len); - try self.readExisting( - .from(logger), - @ptrCast(hash_buf.ptr), - @sizeOf(Hash) * take, - ); - blockhash_writer.advance(take); - num_hashes -= take; - } - }, _ => return error.InvalidSector, } @@ -359,11 +343,9 @@ pub const Rooted = struct { self.io.reader.io_stalled = 0; } } - runtime_metadata.block_id = last_block_id orelse return error.NoBlockIDs; - // write the slot to commit the RuntimeMetadata stuff - const slot = self.journal.committed_slot; - runtime_metadata.populateSlot(slot); + // release the readiness barrier now that Manifest + accounts are loaded. + snapshot_metadata.populateSlot(self.journal.committed_slot); self.table.flushPuts(&self.put_batch); logger.info().logf("loaded rooted db: {} accounts", .{self.table.count()}); @@ -391,16 +373,42 @@ pub const Rooted = struct { pub fn loadSnapshot( self: *Rooted, logger: tel.Logger("Rooted.loadSnapshot"), - runner: lib.runner.Connection, - snapshot_iter: anytype, // lib.solana.snapshot.SnapshotIter(anytype), - runtime_metadata: *RuntimeMetadata, + snapshot_metadata: *SnapshotMetadata, + buf_reader: anytype, ) !void { const zone = tracy.Zone.init(@src(), .{ .name = "loadSnapshot" }); defer zone.deinit(); - const slot = snapshot_iter.manifest.bank_fields.slot; + const BufReader = @TypeOf(buf_reader); + var snapshot_iter = try SnapshotIter(BufReader).init(snapshot_metadata, buf_reader); + + const slot = snapshot_metadata.manifest.bank_fields.slot; try self.beginTransaction(.from(logger), slot); + // Persist the Manifest + StatusCache + max FBA bytes + // contiguous blob right after the journal block. Padded to block_size + // so account sectors start block-aligned. + { + const metadata = snapshot_metadata.getSerializable(); + const padded = std.mem.alignForward(u64, metadata.len, block_size); + if (padded > std.math.maxInt(u32)) return error.ManifestTooLarge; + self.journal.metadata_bytes = @intCast(padded); + + // write unpadded + var r = std.Io.Reader.fixed(metadata); + try self.queueWrite(.from(logger), metadata.len, &r); + + // write padding if any + const pad_len = padded - metadata.len; + if (pad_len > 0) try self.queueWrite(.from(logger), pad_len, struct { + pub fn readSliceAll(_: @This(), b: []u8) !void { + @memset(b, 0); + } + }{}); + } + + logger.info().logf("reading snapshot accounts", .{}); + var timer = try std.time.Timer.start(); var n_puts: usize = 0; var n_transfer: usize = 0; @@ -409,7 +417,7 @@ pub const Rooted = struct { n_transfer += @sizeOf(AccountMeta) + @sizeOf(SectorHeader) + acc.data.len; try self.put( .from(logger), - snapshot_iter, // data reader + &snapshot_iter, // data reader .{ .slot = acc.slot, .pubkey = acc.pubkey, @@ -441,54 +449,8 @@ pub const Rooted = struct { } } - { // write the block_id + current blockhash queue - const blockhash_queue = &snapshot_iter.manifest.bank_fields.blockhash_queue; - self.journal.blockhash_max_age = std.math.lossyCast(u32, blockhash_queue.max_age); - - const block_id = snapshot_iter.manifest.extra_fields.block_id; - const num_hashes = blockhash_queue.hashes.count; - const hashes = blockhash_queue.hashes.array[0..num_hashes]; - - // write block_metadata header - const header: SectorHeader = .{ - .type = .block_metadata, - .info = .{ .count = @intCast(num_hashes) }, - }; - - var r = std.Io.Reader.fixed(std.mem.asBytes(&header)); - try self.queueWrite(.from(logger), @sizeOf(SectorHeader), &r); - - // write block_id - r = std.Io.Reader.fixed(std.mem.asBytes(&block_id)); - try self.queueWrite(.from(logger), @sizeOf(Hash), &r); - - // write blockhashes - r = std.Io.Reader.fixed(std.mem.sliceAsBytes(hashes)); - try self.queueWrite(.from(logger), num_hashes * @sizeOf(Hash), &r); - - runtime_metadata.block_id = block_id; - - // send blockhashes over as metadata - var blockhash_writer = runtime_metadata.blockhash_queue.hashes.getView(.writer); - defer { - runtime_metadata.blockhash_queue.max_age = self.journal.blockhash_max_age; - blockhash_writer.close(); // close when done. - } - - var i: usize = 0; - while (i < hashes.len) { - const buf = try blockhash_writer.getBufferBlocking(runner); - if (buf.len == 0) break; // reader closed somehow - - const take = @min(buf.len, hashes.len - i); - @memcpy(buf[0..take], hashes[i..][0..take]); - blockhash_writer.advance(take); - i += take; - } - } - - // write the slot to commit the RuntimeMetadata stuff - runtime_metadata.populateSlot(slot); + // release the readiness barrier now that Manifest + accounts are loaded. + snapshot_metadata.populateSlot(slot); try self.commitTransaction(.from(logger)); logger.info().logf("populated from snapshot: {} accounts", .{self.table.count()}); diff --git a/v2/lib/ipc/ring.zig b/v2/lib/ipc/ring.zig index 20cc6d03a0..f39b882c9c 100644 --- a/v2/lib/ipc/ring.zig +++ b/v2/lib/ipc/ring.zig @@ -159,6 +159,15 @@ pub fn Ring(N: comptime_int, T: type) type { return ptr; } + /// Same as next(), but blocks until an element is available. + /// Returns `error.Closed` instead of null, if the other side is closed. + pub fn nextBlocking(self: *Self, runner: lib.runner.Connection) !Ptr { + const buf = try self.view.getBufferBlocking(runner); + if (buf.len == 0) return error.Closed; + self.view.bump(1); + return &buf[0]; + } + // Using the increments done via `next()` since either 1) the .get() creating this Iterator or // 2) the last markUsed() call, update the position of this side on the ring buffer, making any // changes to the memory visible to the other side. diff --git a/v2/lib/snapshot.zig b/v2/lib/snapshot.zig index 11178b18e5..5bd0c8e4a9 100644 --- a/v2/lib/snapshot.zig +++ b/v2/lib/snapshot.zig @@ -76,3 +76,63 @@ pub const ReadySnapshot = extern struct { return try std.fmt.bufPrint(buf, "{f}", .{self}); } }; + +/// A deserialized snapshot Manifest + StatusCache. +/// +/// All variable-sized data (blockhash queue is fixed at 300 entries inline; +/// pubkey maps, vote-account chains, etc.) points into the trailing `memory` (manifestBase) +/// VLA via `snapshot.RelativeSlice` / `snapshot.RelativeOffset`. +/// +/// Before a consumer reads the fields, it must call `getSlotBlocking()`. +/// The producer that sets the fields will call `populateSlot()` to mark them as consumable. +pub const SnapshotMetadata = extern struct { + slot: std.atomic.Value(u64), + + manifest: lib.solana.snapshot.Manifest, + status_cache: lib.solana.snapshot.StatusCache, + + memory_len: usize, + memory: [0]u8 align(16), // VLA for [0..memory_len] + + // 0 may be a valid slot, so use something that will never be reached. + const invalid_slot = std.math.maxInt(Slot); + + pub fn init(self: *SnapshotMetadata, memory_len: usize) void { + self.slot = .init(invalid_slot); + self.memory_len = memory_len; + } + + /// Gets a slice of the serializable memory of the SnapshotMetadata + pub fn getSerializable(self: *SnapshotMetadata) []u8 { + // skip the slot & return all bytes from manifest onwards + comptime std.debug.assert(@offsetOf(SnapshotMetadata, "slot") == 0); + comptime std.debug.assert(@offsetOf(SnapshotMetadata, "manifest") == 8); + const header_size = @sizeOf(SnapshotMetadata) - 8; + return @as([*]u8, @ptrCast(&self.manifest))[0 .. header_size + self.memory_len]; + } + + /// Returns the base pointer used to resolve `RelativeSlice`/`RelativeOffset` + /// values inside `manifest` / `status_cache`. + pub fn getMemory(self: *SnapshotMetadata) []u8 { + return self.memory[0..].ptr[0..self.memory_len]; + } + + /// Unblocks all getSlotBlocking() callers with the given slot value. + /// Can be called only once. + /// Should also only call after all other SnapshotMetadata fields are populated. + pub fn populateSlot(self: *SnapshotMetadata, slot: Slot) void { + std.debug.assert(slot != invalid_slot); + std.debug.assert(self.slot.swap(slot, .release) == invalid_slot); + } + + pub fn getSlotBlocking(self: *SnapshotMetadata, runner: lib.runner.Connection) !Slot { + while (true) { + const slot = self.slot.load(.acquire); + if (slot != invalid_slot) { + try runner.activity.signalActive(); + return slot; + } + try runner.activity.signalIdleSpinning(); + } + } +}; diff --git a/v2/lib/solana.zig b/v2/lib/solana.zig index 4dff682180..0b2ca833fd 100644 --- a/v2/lib/solana.zig +++ b/v2/lib/solana.zig @@ -6,6 +6,7 @@ comptime { _ = @import("solana/features.zig"); _ = @import("solana/hash.zig"); _ = @import("solana/ids.zig"); + _ = @import("solana/inflation.zig"); _ = @import("solana/leader_schedule.zig"); _ = @import("solana/pubkey.zig"); _ = @import("solana/signature.zig"); @@ -29,6 +30,7 @@ pub const Signature = @import("solana/signature.zig").Signature; pub const Cluster = @import("solana/cluster.zig").Cluster; pub const LeaderSchedule = @import("solana/leader_schedule.zig").LeaderSchedule; pub const EpochSchedule = @import("solana/epoch_schedule.zig").EpochSchedule; +pub const Inflation = @import("solana/inflation.zig").Inflation; pub const Lamports = u64; pub const Nonce = u32; diff --git a/v2/lib/solana/epoch_schedule.zig b/v2/lib/solana/epoch_schedule.zig index 2271d6e5f5..691016eb0c 100644 --- a/v2/lib/solana/epoch_schedule.zig +++ b/v2/lib/solana/epoch_schedule.zig @@ -1,3 +1,5 @@ +// TODO: move this into core + const std = @import("std"); const lib = @import("../lib.zig"); @@ -18,24 +20,24 @@ pub const MINIMUM_SLOTS_PER_EPOCH: u64 = 32; /// Analogous to [EpochSchedule](https://github.com/anza-xyz/agave/blob/5a9906ebf4f24cd2a2b15aca638d609ceed87797/sdk/program/src/epoch_schedule.rs#L35) pub const EpochSchedule = extern struct { /// The maximum number of slots in each epoch. - slots_per_epoch: u64, + slots_per_epoch: u64 align(1), /// A number of slots before beginning of an epoch to calculate /// a leader schedule for that epoch. - leader_schedule_slot_offset: u64, + leader_schedule_slot_offset: u64 align(1), /// Whether epochs start short and grow. - warmup: bool, + warmup: bool align(1), /// The first epoch after the warmup period. /// /// Basically: `log2(slots_per_epoch) - log2(MINIMUM_SLOTS_PER_EPOCH)`. - first_normal_epoch: Epoch, + first_normal_epoch: Epoch align(1), /// The first slot after the warmup period. /// /// Basically: `MINIMUM_SLOTS_PER_EPOCH * (2.pow(first_normal_epoch) - 1)`. - first_normal_slot: Slot, + first_normal_slot: Slot align(1), pub const ID: Pubkey = .parse("SysvarEpochSchedu1e111111111111111111111111"); pub const STORAGE_SIZE: u64 = 33; diff --git a/v2/lib/solana/inflation.zig b/v2/lib/solana/inflation.zig new file mode 100644 index 0000000000..4c56e54885 --- /dev/null +++ b/v2/lib/solana/inflation.zig @@ -0,0 +1,108 @@ +// TODO: move this into core + +const std = @import("std"); + +/// Zig's `std.math.pow` may return a result that is off by up to one ULP, when comparing to glibc or musl's `pow()`. +/// As these calculations affect consensus, that is an unacceptable difference for us, so we import libc's pow and +/// use that. For reference: +/// - `std.math.pow`: pow(0.85, 4.019250798563942) -> 7.805634650110366e-2 +/// - glibc/musl: pow(0.85, 4.019250798563942) -> 7.805634650110367e-2 +extern fn pow(f64, f64) f64; + +/// Analogous to [Inflation](https://github.com/anza-xyz/agave/blob/55aff7288e596e93d1184ba827048b1e3dc98061/sdk/src/inflation.rs#L6) +pub const Inflation = extern struct { + /// Initial inflation percentage, from time=0 + initial: f64 align(1), + + /// Terminal inflation percentage, to time=INF + terminal: f64 align(1), + + /// Rate per year, at which inflation is lowered until reaching terminal + /// i.e. inflation(year) == MAX(terminal, initial*((1-taper)^year)) + taper: f64 align(1), + + /// Percentage of total inflation allocated to the foundation + foundation: f64 align(1), + + /// Duration of foundation pool inflation, in years + foundation_term: f64 align(1), + + /// DEPRECATED, this field is currently unused + __unused: f64 align(1), + + pub const DEFAULT = Inflation{ + .initial = 0.08, + .terminal = 0.015, + .taper = 0.15, + .foundation = 0.05, + .foundation_term = 7.0, + .__unused = 0.0, + }; + + pub const FULL: Inflation = .{ + .initial = DEFAULT.initial, + .terminal = DEFAULT.terminal, + .taper = DEFAULT.taper, + .foundation = 0.0, + .foundation_term = 0.0, + .__unused = 0.0, + }; + + pub const PICO = fixed(0.0001); // 0.01% inflation + + pub fn fixed(validator: f64) Inflation { + return .{ + .initial = validator, + .terminal = validator, + .taper = 1.0, + .foundation = 0.0, + .foundation_term = 0.0, + .__unused = 0.0, + }; + } + + pub fn initRandom(random: std.Random) Inflation { + return .{ + .initial = random.float(f64), + .terminal = random.float(f64), + .taper = random.float(f64), + .foundation = random.float(f64), + .foundation_term = random.float(f64), + .__unused = random.float(f64), + }; + } + + pub fn total(self: *const Inflation, slot_in_years: f64) f64 { + std.debug.assert(slot_in_years >= 0.0); + return @max( + self.terminal, + self.initial * pow(1.0 - self.taper, slot_in_years), + ); + } + + pub fn validatorRate(self: *const Inflation, slot_in_years: f64) f64 { + std.debug.assert(slot_in_years >= 0.0); + return self.total(slot_in_years) - self.foundationRate(slot_in_years); + } + + pub fn foundationRate(self: *const Inflation, slot_in_years: f64) f64 { + return if (slot_in_years < self.foundation_term) + self.total(slot_in_years) * self.foundation + else + 0.0; + } +}; + +test "inflation" { + const inflation = Inflation{ + .initial = 0.15, + .terminal = 0.015, + .taper = 0.15, + .foundation = 0.0, + .foundation_term = 0.0, + .__unused = 0.0, + }; + + try std.testing.expectEqual(7.805634650110367e-2, inflation.total(4.019250798563942)); + std.debug.assert(4602862346652160054 == @as(u64, @bitCast(pow(0.85, 4.019250798563942)))); +} diff --git a/v2/lib/solana/snapshot.zig b/v2/lib/solana/snapshot.zig index efc492feaf..37db7c92bc 100644 --- a/v2/lib/solana/snapshot.zig +++ b/v2/lib/solana/snapshot.zig @@ -8,6 +8,77 @@ const Pubkey = lib.solana.Pubkey; const Slot = lib.solana.Slot; const Epoch = lib.solana.Epoch; const Hash = lib.solana.Hash; +const EpochSchedule = lib.solana.EpochSchedule; +const Inflation = lib.solana.Inflation; + +/// A slice into a shared-memory region, expressed as an offset from a base pointer +/// plus a length. Position-independent so it can live inside an extern struct that is +/// mmap'd at different virtual addresses across processes. +pub fn RelativeSlice(comptime T: type) type { + return extern struct { + offset: u32 = 0, + len: u32 = 0, + + const Self = @This(); + + pub fn fromSlice(base: [*]const u8, s: []const T) Self { + const start = @intFromPtr(s.ptr) - @intFromPtr(base); + std.debug.assert(start <= std.math.maxInt(u32)); + std.debug.assert(s.len <= std.math.maxInt(u32)); + return .{ .offset = @intCast(start), .len = @intCast(s.len) }; + } + + pub fn slice(self: Self, base: [*]u8) []T { + const raw = base + self.offset; + std.debug.assert(@intFromPtr(raw) % @alignOf(T) == 0); + const ptr: [*]T = @ptrCast(@alignCast(raw)); + return ptr[0..self.len]; + } + + pub fn sliceConst(self: Self, base: [*]const u8) []const T { + const raw = base + self.offset; + std.debug.assert(@intFromPtr(raw) % @alignOf(T) == 0); + const ptr: [*]const T = @ptrCast(@alignCast(raw)); + return ptr[0..self.len]; + } + }; +} + +/// A single-pointer variant of `RelativeSlice`. `offset == 0` is the null sentinel; +/// callers must guarantee no valid allocation lands at offset 0 (see `Rooted.loadSnapshot` +/// which reserves the first byte of its FBA for exactly this purpose). +pub fn RelativeOffset(comptime T: type) type { + return extern struct { + offset: u32 = 0, + + const Self = @This(); + + pub fn from(base: [*]const u8, ptr: *const T) Self { + const o = @intFromPtr(ptr) - @intFromPtr(base); + std.debug.assert(o != 0); + std.debug.assert(o <= std.math.maxInt(u32)); + return .{ .offset = @intCast(o) }; + } + + pub fn isNull(self: Self) bool { + return self.offset == 0; + } + + pub fn pointer(self: Self, base: [*]u8) *T { + std.debug.assert(self.offset != 0); + const raw = base + self.offset; + std.debug.assert(@intFromPtr(raw) % @alignOf(T) == 0); + return @ptrCast(@alignCast(raw)); + } + + pub fn pointerConst(self: Self, base: [*]const u8) *const T { + std.debug.assert(self.offset != 0); + const raw = base + self.offset; + std.debug.assert(@intFromPtr(raw) % @alignOf(T) == 0); + return @ptrCast(@alignCast(raw)); + } + }; +} fn readInt(Int: type, r: anytype) !u64 { var buf: [@sizeOf(Int)]u8 = undefined; @@ -22,7 +93,12 @@ fn readBool(r: anytype) !bool { return buf[0] > 0; } -pub const StatusCache = struct { +pub const StatusCache = extern struct { + /// Placeholder — the deserialized contents are currently discarded. Kept as an + /// extern struct so it can live inline in `SnapshotMetadata` and be persisted + /// alongside the `Manifest` in the rooted DB blob. + _reserved: u8 = 0, + pub fn read(fba: *std.heap.FixedBufferAllocator, r: anytype) !StatusCache { const zone = tracy.Zone.init(@src(), .{ .name = "StatusCache.read" }); defer zone.deinit(); @@ -89,7 +165,7 @@ pub const StatusCache = struct { } }; -pub const Manifest = struct { +pub const Manifest = extern struct { bank_fields: BankFields, accounts_db_fields: AccountsDbFields, extra_fields: ExtraFields, @@ -106,23 +182,37 @@ pub const Manifest = struct { } }; -pub const BankFields = struct { +pub const BankFields = extern struct { slot: Slot, blockhash_queue: BlockHashQueue, - - pub const BlockHashQueue = struct { - last_hash: ?Hash, + epoch_schedule: EpochSchedule, + inflation: Inflation, + stakes_cache: extern struct { + epoch: Epoch, + vote_accounts: RelativeSlice(Pubkey), // acc.data contains stake for the voter + stake_accounts: RelativeSlice(Pubkey), // acc.data contains voter pubkey + Delegation + }, + + pub const BlockHashQueue = extern struct { + /// Agave's MAX_RECENT_BLOCKHASHES and the current `Rooted.Journal.blockhash_max_age`. + /// It's plus-one given cutoff is `<= max_age` instead of `< max_age`. + /// [agave] https://github.com/anza-xyz/solana-sdk/blob/clock%40v3.1.1/clock/src/lib.rs#L95 + pub const MAX_RECENT_BLOCKHASHES: u32 = 300 + 1; + + last_hash: Hash, // .ZEROES if null in snapshot max_age: u64, - hashes: struct { - array: []Hash, - count: usize, - }, + hashes: [MAX_RECENT_BLOCKHASHES]Hash, + hashes_count: u32, pub const Entry = extern struct { hash: Hash, hash_index: u64, }; + pub fn getHashes(self: *const BlockHashQueue) []const Hash { + return self.hashes[0..self.hashes_count]; + } + pub fn read(fba: *std.heap.FixedBufferAllocator, r: anytype) !BlockHashQueue { const last_hash_index = try readInt(u64, r); const maybe_last_hash: ?Hash = if (!(try readBool(r))) null else blk: { @@ -132,7 +222,6 @@ pub const BankFields = struct { }; const n_hash_infos = try readInt(u64, r); - const hashes = try fba.allocator().alloc(Hash, n_hash_infos); const BlockhashEntry = extern struct { hash: Hash, @@ -154,21 +243,24 @@ pub const BankFields = struct { } }.lessThan); - // then add only the live ones by max_age to the hashes - var count: usize = 0; + var out: BlockHashQueue = .{ + .last_hash = maybe_last_hash orelse .ZEROES, + .max_age = max_age, + .hashes = @splat(.ZEROES), + .hashes_count = 0, + }; + + // then add only the live ones by max_age for (entries) |*e| { const age = last_hash_index - e.hash_index; if (age <= max_age) { - hashes[count] = e.hash; - count += 1; + if (out.hashes_count >= MAX_RECENT_BLOCKHASHES) return error.TooManyBlockhashes; + out.hashes[out.hashes_count] = e.hash; + out.hashes_count += 1; } } - return .{ - .last_hash = maybe_last_hash, - .max_age = max_age, - .hashes = .{ .array = hashes, .count = count }, - }; + return out; } }; @@ -211,68 +303,102 @@ pub const BankFields = struct { 8 // accounts_data_len: u64 ); const slot = try readInt(Slot, r); - try r.discardAll( - 8 + // _unused_epoch: Epoch - 8 + // block_height: u64 - 32 + // leader_id: Pubkey - 8 + // _unused_collector_fees: u64 - 8 + // _unused_fee_calculator: u64 - // fee_rate_governor: - 8 + // target_lamports_per_signature: u64 - 8 + // target_signatures_per_slot: u64 - 8 + // min_lamports_per_signature: u64 - 8 + // max_lamports_per_signature: u64 - 1 + // burn_percent: u8 - 8 + // _unused_collected_rent: u64 - // _unused_rent_collector: - 8 + // epoch: Epoch - // epoch_schedule: EpochSchedule: - 8 + // slots_per_epoch: u64 - 8 + // leader_schedule_slot_offset: u64 - 1 + // warmup: bool - 8 + // first_normal_epoch: u64 - 8 + // first_normal_slot: u64 - 8 + // slots_per_year: f64 - // rent: - 8 + // lamports_per_byte: u64 - 8 + // exemption_threshold: [8]u8 - 1 + // burn_percent: u8 - // epoch_schedule: EpochSchedule: - 8 + // slots_per_epoch: u64 - 8 + // leader_schedule_slot_offset: u64 - 1 + // warmup: bool - 8 + // first_normal_epoch: u64 - 8 + // first_normal_slot: u64 - // inflation: - 8 + // initial: f64 - 8 + // terminal: f64 - 8 + // taper: f64 - 8 + // foundation: f64 - 8 + // foundation_term: f64 - 8, // __unused: f64 + try r.discardAll(8 + // _unused_epoch: Epoch + 8 + // block_height: u64 + 32 + // leader_id: Pubkey + 8 + // _unused_collector_fees: u64 + 8 + // _unused_fee_calculator: u64 + // fee_rate_governor: + 8 + // target_lamports_per_signature: u64 + 8 + // target_signatures_per_slot: u64 + 8 + // min_lamports_per_signature: u64 + 8 + // max_lamports_per_signature: u64 + 1 + // burn_percent: u8 + 8 + // _unused_collected_rent: u64 + // _unused_rent_collector: + 8 + // epoch: Epoch + // epoch_schedule: EpochSchedule: + 8 + // slots_per_epoch: u64 + 8 + // leader_schedule_slot_offset: u64 + 1 + // warmup: bool + 8 + // first_normal_epoch: u64 + 8 + // first_normal_slot: u64 + 8 + // slots_per_year: f64 + // rent: + 8 + // lamports_per_byte: u64 + 8 + // exemption_threshold: [8]u8 + 1 // burn_percent: u8 ); - // stakes: Stakes(Delegation) - // vote_accounts: VoteAccounts - try discardVoteAccounts(r); + var epoch_schedule: EpochSchedule = undefined; + try r.readSliceAll(std.mem.asBytes(&epoch_schedule)); + + var inflation: Inflation = undefined; + try r.readSliceAll(std.mem.asBytes(&inflation)); + + // stakes: Stakes(.Delegation) + // vote_accounts: HashMap(Pubkey, {stake, AccountData}) + // + // NOTE: Only stores pubkeys, as we should verify their data against the accounts instead. + const vote_len = try readInt(u64, r); + const vote_accounts = try fba.allocator().alloc(Pubkey, vote_len); + for (vote_accounts) |*vote_pubkey| { + var header: extern struct { + pubkey: Pubkey, // key: Pubkey + stake: u64, // value.stake: u64 + lamports: u64, // value.account.lamports: u64 + data_len: u64, // value.account.data: Vec(u8) + } = undefined; + try r.readSliceAll(std.mem.asBytes(&header)); + + vote_pubkey.* = header.pubkey; + try r.discardAll( + header.data_len + // account data bytes (TODO: validate this against account?) + 32 + // value.account.owner: Pubkey + 1 + // value.account.executable: bool + 8, // value.account.rent_epoch: Epoch(u64) + ); + } // stake_delegations: HashMap(Pubkey, Delegation) + // Delegation = { voter_pubkey, stake, activation_epoch, deactivation_epoch, warmup } + // + // NOTE: only read the pubkeys. The stake data should be fetched from the accounts instead. const stake_del_len = try readInt(u64, r); - try r.discardAll(stake_del_len * (32 + // key: Pubkey - // Delegation: - 32 + // voter_pubkey: Pubkey - 8 + // stake: u64 - 8 + // activation_epoch: Epoch - 8 + // deactivation_epoch: Epoch - 8 // warmup_cooldown_rate: f64 - )); + const stake_accounts = try fba.allocator().alloc(Pubkey, stake_del_len); + { + // read chunks of entries at a time to amortize costs of r.readSliceAll + var delegation_entries: [32]extern struct { + stake_pubkey: Pubkey, + voter_pubkey: Pubkey, + stake: u64 align(1), + activation_epoch: Epoch align(1), + deactivation_epoch: Epoch align(1), + _deprecated_warmup_cooldown_rate: f64 align(1), + } = undefined; - try r.discardAll( - 8 + // stakes.unused: u64 - 8, // stakes.epoch: Epoch - ); + var i: usize = 0; + while (i < stake_del_len) { + const n = @min(stake_del_len - i, delegation_entries.len); + + const pubkey_chunk = stake_accounts[i..][0..n]; + i += n; + + const chunk = delegation_entries[0..n]; + try r.readSliceAll(std.mem.sliceAsBytes(chunk)); - // stake_history: Vec({ epoch: Epoch, effective: u64, activating: u64, deactivating: u64 }) + for (chunk, pubkey_chunk) |*delegation_entry, *stake_pubkey| { + stake_pubkey.* = delegation_entry.stake_pubkey; + } + } + } + + _ = try readInt(u64, r); // stakes.unused: u64 + const epoch = try readInt(Epoch, r); // stakes.epoch: Epoch + + // stake_history: Vec({ epoch: Epoch, effective: u64, activating: u64, deactivating: u64 }) + // + // NOTE: ignored as it's better to parse from the StakeHistory account instead. const stake_history_len = try readInt(u64, r); try r.discardAll(stake_history_len * (8 + // epoch: Epoch 8 + // effective: u64 @@ -300,11 +426,18 @@ pub const BankFields = struct { return .{ .slot = slot, .blockhash_queue = blockhash_queue, + .epoch_schedule = epoch_schedule, + .inflation = inflation, + .stakes_cache = .{ + .epoch = epoch, + .vote_accounts = .fromSlice(fba.buffer.ptr, vote_accounts), + .stake_accounts = .fromSlice(fba.buffer.ptr, stake_accounts), + }, }; } }; -pub const AccountsDbFields = struct { +pub const AccountsDbFields = extern struct { slot: u64, pub fn read(_: *std.heap.FixedBufferAllocator, r: anytype) !AccountsDbFields { @@ -367,25 +500,152 @@ pub const AccountsDbFields = struct { } }; -pub const ExtraFields = struct { +pub const ExtraFields = extern struct { + versioned_epoch_stakes: RelativeSlice(VersionedEpochStakes) = .{}, /// - TowerBFT: the Merkle root of the last FEC set of the block /// - Alpenglow: the "double Merkle root": a Merkle root computed over the /// sequence of per-FEC-set Merkle roots of the block's shreds. block_id: Hash, + pub const VersionedEpochStakes = extern struct { + epoch: Epoch, + total_stake: u64, + vote_accounts: RelativeSlice(VoteAccountEntry), + node_to_vote_accounts: RelativeSlice(NodeToVoterEntry), + epoch_authorized_voters: RelativeSlice(AuthToVoterEntry), + + pub const VoteAccountEntry = extern struct { + pubkey: Pubkey, + stake: u64, // kept as Versioned entry cant lookup past/future epoch VoteAccount data + }; + + pub const NodeToVoterEntry = extern struct { + node_pubkey: Pubkey, + vote_accounts: RelativeSlice(Pubkey), + total_stake: u64, + }; + + pub const AuthToVoterEntry = extern struct { + voter_pubkey: Pubkey, + authorized_voter: Pubkey, + }; + + pub fn read(fba: *std.heap.FixedBufferAllocator, r: anytype) !VersionedEpochStakes { + // epoch: Epoch + const epoch = try readInt(Epoch, r); + + // union tag: u32 (enum(u32), always 'current') + const union_tag = try readInt(u32, r); + if (union_tag != 0) { + return error.InvalidVersionedEpochStakesUnion; + } + + // epoch_stakes: Stakes(Delegation) + // vote_accounts: HashMap(Pubkey, { stake: u64, account: AccountSharedData }) + // + // where AccountSharedData = + // { lamports: u64, data: Vec(u8), owner: Pubkey, executable: bool, rent_epoch: Epoch } + const vote_len = try readInt(u64, r); + const vote_accounts = try fba.allocator().alloc(VoteAccountEntry, vote_len); + for (vote_accounts) |*entry| { + var header: extern struct { + key: Pubkey, + stake: u64 align(1), + lamports: u64 align(1), + data_len: u64 align(1), + } = undefined; + try r.readSliceAll(std.mem.asBytes(&header)); + + entry.* = .{ .pubkey = header.key, .stake = header.stake }; + try r.discardAll(header.data_len + // data bytes (TODO: validate this against account data?) + 32 + // owner: Pubkey + 1 + // executable: bool + 8 // rent_epoch: Epoch + ); + } + + // stake_delegations: HashMap(Pubkey, { Delegation, credits_observed: u64 }) + // + // NOTE: this is discarded instead of stored: + // https://github.com/anza-xyz/agave/blob/v4.2/runtime/src/epoch_stakes.rs#L442-L443 + const stake_del_len = try readInt(u64, r); + try r.discardAll(stake_del_len * (32 + // key: Pubkey + 32 + // delegation.voter_pubkey: Pubkey + 8 + // delegation.stake: u64 + 8 + // delegation.activation_epoch: Epoch + 8 + // delegation.deactivation_epoch: Epoch + 8 + // delegation.warmup_cooldown_rate: f64 + 8 // credits_observed: u64 + )); + + // unused: u64 + // epoch: Epoch + try r.discardAll(8 + 8); + + // stake_history: Vec({Epoch, effective: u64, activating: u64, deactivating: u64}) + // + // NOTE: this is empty on testnet snapshots and is fine to discard. + // The one that actually matters is BankFields.stake_history, + // and its better to parse it out of the StakeHistory sysvar account from db instead. + const stake_history_len = try readInt(u64, r); + try r.discardAll(stake_history_len * (8 + // epoch: Epoch + 8 + // effective: u64 + 8 + // activating: u64 + 8 // deactivating: u64 + )); + + // total_stake: u64 + const total_stake = try readInt(u64, r); + + // node_id_to_vote_accounts: HashMap(Pubkey, { voters:Vec(Pubkey), total_stake: u64 }) + const node_len = try readInt(u64, r); + const node_to_voters = try fba.allocator().alloc(NodeToVoterEntry, node_len); + for (node_to_voters) |*entry| { + var header: extern struct { node_pubkey: Pubkey, voters_len: u64 } = undefined; + try r.readSliceAll(std.mem.asBytes(&header)); + + const node_voters = try fba.allocator().alloc(Pubkey, header.voters_len); + try r.readSliceAll(std.mem.sliceAsBytes(node_voters)); + + const node_stake = try readInt(u64, r); + entry.* = .{ + .node_pubkey = header.node_pubkey, + .vote_accounts = .fromSlice(fba.buffer.ptr, node_voters), + .total_stake = node_stake, + }; + } + + // epoch_authorized_voters: HashMap(Pubkey, Pubkey) + const auth_len = try readInt(u64, r); + const auth_to_voters = try fba.allocator().alloc(AuthToVoterEntry, auth_len); + try r.readSliceAll(std.mem.sliceAsBytes(auth_to_voters)); + + return .{ + .epoch = epoch, + .total_stake = total_stake, + .vote_accounts = .fromSlice(fba.buffer.ptr, vote_accounts), + .node_to_vote_accounts = .fromSlice(fba.buffer.ptr, node_to_voters), + .epoch_authorized_voters = .fromSlice(fba.buffer.ptr, auth_to_voters), + }; + } + }; + pub fn read(fba: *std.heap.FixedBufferAllocator, r: anytype) !ExtraFields { const zone = tracy.Zone.init(@src(), .{ .name = "ExtraFields.read" }); defer zone.deinit(); - _ = fba; - // lamports_per_signature: NullOnEof(u64) r.discardAll(8) catch |err| switch (err) { error.EndOfStream => {}, else => |e| return e, }; - // _unused_incremental_snapshot_persistence: NullOnEof(?{ full: SlotAndHash, full_capitalization: u64, incremental_hash: Hash, incremental_capitalization: u64 }) + // _unused_incremental_snapshot_persistence: NullOnEof(?{ + // full: SlotAndHash, + // full_capitalization: u64, + // incremental_hash: Hash, + // incremental_capitalization: u64 + // }) { const is_some = readBool(r) catch |err| switch (err) { error.EndOfStream => false, @@ -410,63 +670,19 @@ pub const ExtraFields = struct { } // versioned_epoch_stakes: NullOnEof(Vec({ epoch: u64, value: union(enum(u32)) { current: ... } })) + var versioned_epoch_stakes: RelativeSlice(VersionedEpochStakes) = .{}; { - const outer_len = readInt(u64, r) catch |err| switch (err) { + const len = readInt(u64, r) catch |err| switch (err) { error.EndOfStream => 0, else => |e| return e, }; - for (0..outer_len) |_| { - try r.discardAll( - 8 + // epoch: u64 - 4, // union tag: u32 (enum(u32), always 'current') - ); - // current.epoch_stakes: Stakes(StakeDelegationWithStake) - // vote_accounts: VoteAccounts - try discardVoteAccounts(r); - - // stake_delegations: HashMap(Pubkey, { delegation: Delegation, credits_observed: u64 }) - const stake_del_len = try readInt(u64, r); - try r.discardAll(stake_del_len * (32 + // key: Pubkey - 32 + // delegation.voter_pubkey: Pubkey - 8 + // delegation.stake: u64 - 8 + // delegation.activation_epoch: Epoch - 8 + // delegation.deactivation_epoch: Epoch - 8 + // delegation.warmup_cooldown_rate: f64 - 8 // credits_observed: u64 - )); - - try r.discardAll( - 8 + // stakes.unused: u64 - 8, // stakes.epoch: Epoch - ); - - // stake_history: Vec({ epoch: Epoch, effective: u64, activating: u64, deactivating: u64 }) - const sh_len = try readInt(u64, r); - try r.discardAll(sh_len * (8 + 8 + 8 + 8)); - - // current.total_stake: u64 - try r.discardAll(8); - - // current.node_id_to_vote_accounts: HashMap(Pubkey, { vote_accounts: Vec(Pubkey), total_stake: u64 }) - const nv_len = try readInt(u64, r); - for (0..nv_len) |_| { - // key: Pubkey - try r.discardAll(32); - // value.vote_accounts: Vec(Pubkey) - const va_len = try readInt(u64, r); - try r.discardAll( - va_len * 32 + // vote_accounts: []Pubkey - 8, // total_stake: u64 - ); - } - - // current.epoch_authorized_voters: HashMap(Pubkey, Pubkey) - const eav_len = try readInt(u64, r); - try r.discardAll(eav_len * (32 + // key: Pubkey - 32 // value: Pubkey - )); + const slice = try fba.allocator().alloc(VersionedEpochStakes, len); + for (slice) |*versioned_epoch_stake| { + versioned_epoch_stake.* = try .read(fba, r); } + + versioned_epoch_stakes = .fromSlice(fba.buffer.ptr, slice); } // accounts_lt_hash: NullOnEof(?LtHash) @@ -491,47 +707,29 @@ pub const ExtraFields = struct { else => |e| return e, }; - return .{ .block_id = block_id }; + return .{ + .versioned_epoch_stakes = versioned_epoch_stakes, + .block_id = block_id, + }; } }; -/// Discards VoteAccounts: HashMap(Pubkey, { stake: u64, account: AccountSharedData }) -/// AccountSharedData contains a variable-length Vec(u8) data field, so we must loop. -fn discardVoteAccounts(r: anytype) !void { - const len = try readInt(u64, r); - for (0..len) |_| { - try r.discardAll( - 32 + // key: Pubkey - 8 + // value.stake: u64 - 8, // value.account.lamports: u64 - ); - // value.account.data: Vec(u8) - const data_len = try readInt(u64, r); - try r.discardAll( - data_len + // account data bytes - 32 + // value.account.owner: Pubkey - 1 + // value.account.executable: bool - 8, // value.account.rent_epoch: Epoch(u64) - ); - } -} - pub fn SnapshotIter(comptime BufReader: type) type { return struct { - // public fields instantiated using the fba from init() - status_cache: StatusCache, - manifest: Manifest, - tar_iter: TarZstIter(BufReader), account_file_len: usize, account_file_slot: Slot, account_data_len: usize, account_data_padding: usize, + /// Cached from the just-parsed Manifest so `next()` can validate account file + /// slots without holding a pointer into snapshot_metadata (which lives in + /// another module and is passed as `anytype` at init time). + accounts_db_slot: Slot, const Self = @This(); pub fn init( - fba: *std.heap.FixedBufferAllocator, + snapshot_metadata: *lib.snapshot.SnapshotMetadata, buf_reader: BufReader, ) !Self { var self: Self = undefined; @@ -549,18 +747,23 @@ pub fn SnapshotIter(comptime BufReader: type) type { // read /snapshots/status_cache & /snapshots/{slot}/{slot} (can be in any order) { + var fba = std.heap.FixedBufferAllocator.init( + snapshot_metadata.memory[0..].ptr[0..snapshot_metadata.memory_len], + ); + const tar_file = (try self.tar_iter.next()) orelse return error.MissingMetadata; if (std.mem.eql(u8, tar_file.name, "snapshots/status_cache")) { - self.status_cache = try StatusCache.read(fba, &self.tar_iter); + snapshot_metadata.status_cache = try StatusCache.read(&fba, &self.tar_iter); _ = (try self.tar_iter.next()) orelse return error.MissingMetadata; - self.manifest = try Manifest.read(fba, &self.tar_iter); + snapshot_metadata.manifest = try Manifest.read(&fba, &self.tar_iter); } else { - self.manifest = try Manifest.read(fba, &self.tar_iter); + snapshot_metadata.manifest = try Manifest.read(&fba, &self.tar_iter); _ = (try self.tar_iter.next()) orelse return error.MissingMetadata; - self.status_cache = try StatusCache.read(fba, &self.tar_iter); + snapshot_metadata.status_cache = try StatusCache.read(&fba, &self.tar_iter); } } + self.accounts_db_slot = snapshot_metadata.manifest.accounts_db_fields.slot; self.account_file_len = 0; self.account_file_slot = 0; self.account_data_len = 0; @@ -593,7 +796,7 @@ pub fn SnapshotIter(comptime BufReader: type) type { const slot = std.fmt.parseInt(u64, tar_file.name["accounts/".len..split], 10) catch return error.InvalidAccountFileSlot; - if (slot > self.manifest.accounts_db_fields.slot) + if (slot > self.accounts_db_slot) return error.InvalidAccountFileSlot; self.account_file_slot = slot; @@ -847,29 +1050,40 @@ test "deserialized snapshot matches generated snapshot json" { } }; + const SnapshotMetadata = lib.snapshot.SnapshotMetadata; + const snapshot_fba_size = 8 * 1024 * 1024; + const snapshot_meta_buf = try allocator.alignedAlloc( + u8, + @enumFromInt(@alignOf(SnapshotMetadata)), + @sizeOf(SnapshotMetadata) + snapshot_fba_size, + ); + defer allocator.free(snapshot_meta_buf); + + const snapshot_metadata: *SnapshotMetadata = @ptrCast(snapshot_meta_buf.ptr); + snapshot_metadata.init(snapshot_fba_size); + var snapshot_reader: SnapshotBufReader = .{ .zst_reader = zst_reader }; - const fba_buf = try allocator.alloc(u8, 8 * 1024 * 1024); - defer allocator.free(fba_buf); - var fba: std.heap.FixedBufferAllocator = .init(fba_buf); - var snapshot_iter = try SnapshotIter(*SnapshotBufReader).init(&fba, &snapshot_reader); + var snapshot_iter = + try SnapshotIter(*SnapshotBufReader).init(snapshot_metadata, &snapshot_reader); - try expectEqual(jsonU64(merged.get("slot").?), snapshot_iter.manifest.bank_fields.slot); - try expectEqual(jsonU64(merged.get("slot").?), snapshot_iter.manifest.accounts_db_fields.slot); + const manifest = &snapshot_metadata.manifest; + try expectEqual(jsonU64(merged.get("slot").?), manifest.bank_fields.slot); + try expectEqual(jsonU64(merged.get("slot").?), manifest.accounts_db_fields.slot); try expectEqualStrings( merged.get("block_id").?.string, - snapshot_iter.manifest.extra_fields.block_id.base58String(&hash_buf), + manifest.extra_fields.block_id.base58String(&hash_buf), ); const blockhash_queue_json = merged.get("blockhash_queue").?.object; try expectEqual( jsonU64(blockhash_queue_json.get("max_age").?), - snapshot_iter.manifest.bank_fields.blockhash_queue.max_age, + manifest.bank_fields.blockhash_queue.max_age, ); const json_hashes = blockhash_queue_json.get("hashes").?.array.items; - const bhq_hashes = snapshot_iter.manifest.bank_fields.blockhash_queue.hashes; - try expectEqual(json_hashes.len, bhq_hashes.count); - for (bhq_hashes.array, json_hashes) |hash, json_hash| { + const bhq_hashes = manifest.bank_fields.blockhash_queue.getHashes(); + try expectEqual(json_hashes.len, bhq_hashes.len); + for (bhq_hashes, json_hashes) |hash, json_hash| { try expectEqualStrings(json_hash.object.get("hash").?.string, hash.base58String(&hash_buf)); } diff --git a/v2/services/accounts_db.zig b/v2/services/accounts_db.zig index ada0710087..4490485c4f 100644 --- a/v2/services/accounts_db.zig +++ b/v2/services/accounts_db.zig @@ -5,8 +5,6 @@ const services = @import("services"); const tel = lib.telemetry; -const SnapshotIter = lib.solana.snapshot.SnapshotIter; - const Rooted = lib.accounts_db.Rooted; const AccountPool = lib.accounts_db.AccountPool; @@ -29,14 +27,12 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n logger.info().logf("accounts_db started into file: {s}", .{file_path}); const Global = struct { - var fba_memory: [32 * 1024 * 1024]u8 = undefined; var rooted: Rooted = undefined; }; const rooted = &Global.rooted; try rooted.init( .from(logger), - runner, std.fs.cwd(), file_path, rw.config.memory[0..].ptr[0..rw.config.memory_len], @@ -72,19 +68,15 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n } }; - var fba = std.heap.FixedBufferAllocator.init(&Global.fba_memory); - var snapshot_iter = try SnapshotIter(SnapshotBufReader).init(&fba, .{ - .in_ = &in, - .runner_ = runner, - .completion_ = &rw.ready_snapshot_in.completion, - }); - - logger.info().log("reading snapshot accounts"); + logger.info().log("reading snapshot"); try rooted.loadSnapshot( .from(logger), - runner, - &snapshot_iter, rw.snapshot_metadata_out, + SnapshotBufReader{ + .in_ = &in, + .runner_ = runner, + .completion_ = &rw.ready_snapshot_in.completion, + }, ); } diff --git a/v2/services/replay.zig b/v2/services/replay.zig index ef1c81a625..f9f41d32a8 100644 --- a/v2/services/replay.zig +++ b/v2/services/replay.zig @@ -274,7 +274,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n } } -/// Reads all the RuntimeMetadata provided by accountsdb from the snapshot or +/// Reads all the SnapshotMetadata provided by accountsdb from the snapshot or /// its internal state. This bootstraps replay with information about its /// starting root slot, and some older info like the history of blockhashes. /// This data populates the block tree, some other structures indexed by @@ -292,52 +292,50 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n fn bootstrap( logger: tel.Logger("main"), runner: lib.runner.Connection, - snapshot_metadata: *lib.accounts_db.RuntimeMetadata, + snapshot_metadata: *lib.snapshot.SnapshotMetadata, forest: *MerkleForest, block_pool: *lib.replay.BlockPool, exec_states: *BlockExecStates, blockhash_states: *BlockHashStates, ) !void { - var num_hashes: usize = 0; - // Drain the blockhash queue into the block tree. accountsdb writes into - // this ring blocks waiting for the reader (us). - var root_block = bhq: { - var blockhashes_in = snapshot_metadata.blockhash_queue.hashes.getView(.reader); - defer blockhashes_in.close(); - var last_block: ?lib.replay.BlockRef = null; - while (true) { - const hashes = try blockhashes_in.getBufferBlocking(runner); - if (hashes.len == 0) break; // blockhashes_out closed their end - for (hashes) |*hash| { - const block = try block_pool.createId(); - block.ptr(block_pool).* = .{ - .slot = .null, // cannot be determined from the snapshot - .child = .null, - .parent = .init(last_block), - }; - if (last_block) |p| p.ptr(block_pool).child = .init(block); - blockhash_states[block.index()] = hash.*; - last_block = block; - num_hashes += 1; - } - blockhashes_in.advance(hashes.len); - } + // Acquire barrier — pairs with accounts_db's `populateSlot` (Release). Once + // this returns, `snapshot_metadata.manifest` / `.status_cache` / all their + // trailing FBA data are fully published and safe to read from this process. + const root_slot = try snapshot_metadata.getSlotBlocking(runner); + logger.info().logf("got the root slot from the snapshot: {}", .{root_slot}); - const root_block = last_block orelse return error.NoBlockhashesInSnapshot; + const manifest = &snapshot_metadata.manifest; + const bhq = &manifest.bank_fields.blockhash_queue; + const hashes = bhq.hashes[0..bhq.hashes_count]; - break :bhq root_block; + // Populate the block tree with one BlockRef per recent blockhash, chained by + // parent/child pointers. The last one is the root block for this replay. + var root_block = bhq_load: { + var last_block: ?lib.replay.BlockRef = null; + for (hashes) |*hash| { + const block = try block_pool.createId(); + block.ptr(block_pool).* = .{ + .slot = .null, // cannot be determined from the snapshot + .child = .null, + .parent = .init(last_block), + }; + if (last_block) |p| p.ptr(block_pool).child = .init(block); + blockhash_states[block.index()] = hash.*; + last_block = block; + } + break :bhq_load last_block orelse return error.NoBlockhashesInSnapshot; }; - logger.info().logf("loaded {} blockhashes from accountsdb snapshot data", .{num_hashes}); + logger.info().logf("loaded {} blockhashes from accountsdb snapshot data", .{hashes.len}); - const root_slot = try snapshot_metadata.getSlotBlocking(runner); root_block.ptr(block_pool).slot = .init(root_slot); - logger.info().logf("got the root slot from the snapshot: {}", .{root_slot}); + + const block_id = manifest.extra_fields.block_id; // create a synthetic fec-set node that doesn't have all information about // the fec set, but it is enough to get started processing the first block // after the root const root_node = try insertFecSet(logger, &.{ - .merkle_root = snapshot_metadata.block_id, + .merkle_root = block_id, .chained_merkle_root = .ZEROES, // cannot be determined from the snapshot .id = .{ .slot = root_slot, @@ -369,7 +367,7 @@ fn bootstrap( logger.info().logf( "finished bootstrapping replay at slot {} (block_id={f})", - .{ root_slot, snapshot_metadata.block_id }, + .{ root_slot, block_id }, ); } @@ -1677,26 +1675,26 @@ test "bootstrap creates root block and chains blockhashes" { var service_view = activity.serviceView(); const runner: lib.runner.Connection = .{ .activity = &service_view }; - var metadata: lib.accounts_db.RuntimeMetadata = undefined; - metadata.init(); - metadata.block_id = .parse("ByzshhkRgXWnTkHjapkkqaKgEFnsg8ceY3bw4MWBzFE"); + // We only need the parts of the Manifest that bootstrap actually reads: + // `bank_fields.blockhash_queue.{hashes,hashes_count}` and `extra_fields.block_id`. + // Everything else can be zero-initialized. `memory_len = 0` because none of + // the fields used here go through a RelativeSlice/RelativeOffset. + var metadata = std.mem.zeroes(lib.snapshot.SnapshotMetadata); + metadata.init(0); + metadata.manifest.extra_fields.block_id = + .parse("ByzshhkRgXWnTkHjapkkqaKgEFnsg8ceY3bw4MWBzFE"); - // Prefill the blockhash ring with N > 1 hashes as a single writer batch, - // then close the writer end so bootstrap's drain loop terminates. const test_hashes = [_]Hash{ .parse("BMHr4knWhDp8JhqCYhA2K5DUYQsYUVXdy2zWahzt5jLd"), .parse("2GyMeUytf6fcsfNP2QQ6F5e5qwAUoMtKUbnH6QU6bTNm"), .parse("4UahX8LzYC7xnubvP9QzRHmPPYovtcNYo7rBXKpp3ADM"), .parse("Hh8DjJdpQRGeZ6bUxYyt1PBktnFtNAwZoQuwZqGWLPfB"), }; - { - var writer = metadata.blockhash_queue.hashes.getView(.writer); - const buf = writer.getBuffer().?; - try std.testing.expect(buf.len >= test_hashes.len); - @memcpy(buf[0..test_hashes.len], &test_hashes); - writer.advance(test_hashes.len); - writer.close(); - } + @memcpy( + metadata.manifest.bank_fields.blockhash_queue.hashes[0..test_hashes.len], + &test_hashes, + ); + metadata.manifest.bank_fields.blockhash_queue.hashes_count = test_hashes.len; const root_slot: lib.solana.Slot = 100; metadata.populateSlot(root_slot); diff --git a/v2/tests/replay/main.zig b/v2/tests/replay/main.zig index 11447d4a1b..f778ca2d40 100644 --- a/v2/tests/replay/main.zig +++ b/v2/tests/replay/main.zig @@ -56,15 +56,17 @@ pub fn main() !void { var exec_req_response_region: Region(lib.replay.ExecReqResponse) = try .simple(); exec_req_response_region.ptr().init(); - var snapshot_metadata: Region(lib.accounts_db.RuntimeMetadata) = try .simple(); - snapshot_metadata.ptr().init(); - snapshot_metadata.ptr().block_id = first_shred.chainedMerkleRoot().*; + const snapshot_fba_size = 256 * 1024 * 1024; + var snapshot_metadata: Region(lib.snapshot.SnapshotMetadata) = + try .sized(@sizeOf(lib.snapshot.SnapshotMetadata) + snapshot_fba_size); + snapshot_metadata.ptr().init(snapshot_fba_size); + snapshot_metadata.ptr().manifest.extra_fields.block_id = first_shred.chainedMerkleRoot().*; { - var writer = snapshot_metadata.ptr().blockhash_queue.hashes.getView(.writer); - const blockhashes = writer.getBuffer().?; - blockhashes[0] = lib.solana.Hash.ZEROES; - writer.advance(1); - writer.close(); + const bhq = &snapshot_metadata.ptr().manifest.bank_fields.blockhash_queue; + bhq.max_age = 300; + bhq.hashes[0] = .ZEROES; + bhq.hashes_count = 1; + bhq.last_hash = .ZEROES; } snapshot_metadata.ptr().populateSlot(fixture.manifest.shreds.parent_slot);