diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index 845d6238..86b33f15 100644 --- a/cmd/mithril/node/node.go +++ b/cmd/mithril/node/node.go @@ -59,7 +59,8 @@ var ( bootstrapMode string // "auto", "snapshot", or "accountsdb" snapshotArchivePath string incrementalSnapshotFilename string - accountsPath string + accountsPath string // primary accounts/metadata dir (accountsPaths[0]) + accountsPaths []string // one dir per disk shard scratchDirectory string rpcEndpoints []string cluster string // "mainnet-beta", "testnet", "devnet" @@ -208,7 +209,7 @@ func init() { Run.Flags().StringVar(&incrementalSnapshotFilename, "incremental-snapshot", "", "Path to specific incremental snapshot file (bypasses auto-discovery)") // [ledger] section flags - Run.Flags().StringVarP(&accountsPath, "accounts-path", "o", "", "Output path for writing AccountsDB data to") + Run.Flags().StringSliceVarP(&accountsPaths, "accounts-path", "o", []string{}, "Output path(s) for writing AccountsDB data to - one dir per disk shard, can specify multiple") Run.Flags().StringVar(&blockstorePath, "ledger-path", "/tmp/blocks", "Path containing slot.json files") // [network] section flags @@ -228,6 +229,8 @@ func init() { Run.Flags().Uint64Var(&borrowedAccountArenaSize, "borrowed-account-arena-size", 1024, "Number of borrowed accounts to preallocate in arena (0 to disable)") Run.Flags().IntVar(&snapshot.ZstdDecoderConcurrency, "zstd-decoder-concurrency", runtime.NumCPU(), "Zstd decoder concurrency") Run.Flags().IntVar(&snapshot.MaxConcurrentFlushers, "max-concurrent-flushers", snapshot.DefaultSnapshotMaxConcurrentFlushers, "Bound for number of log shards to flush to Accounts DB Index at once") + Run.Flags().IntVar(&snapshot.SnapshotFlushSortWorkers, "flush-sort-workers", snapshot.DefaultSnapshotFlushSortWorkers, "Snapshot index-flush concurrent sort workers (0 = one per CPU)") + Run.Flags().BoolVar(&snapshot.SnapshotDirectIO, "snapshot-directio", snapshot.DefaultSnapshotDirectIO, "Write snapshot big files with O_DIRECT (bypasses the page cache; helps decode throughput on low-RAM boxes)") Run.Flags().IntVar(&snapshot.SnapshotAppendVecCopyingWorkers, "snapshot-append-vec-workers", snapshot.DefaultSnapshotAppendVecCopyingWorkers, "Snapshot bootstrap appendvec write workers") Run.Flags().IntVar(&snapshot.SnapshotIndexEntryBuilderWorkers, "snapshot-index-builder-workers", snapshot.DefaultSnapshotIndexEntryBuilderWorkers, "Snapshot bootstrap account-index parser workers") Run.Flags().IntVar(&snapshot.SnapshotIndexEntryCommitterWorkers, "snapshot-index-committer-workers", snapshot.DefaultSnapshotIndexEntryCommitterWorkers, "Snapshot bootstrap account-index shard enqueue workers") @@ -445,13 +448,22 @@ func initConfigAndBindFlags(cmd *cobra.Command) error { if incrementalSnapshotFilename == "" { incrementalSnapshotFilename = getString("incremental-snapshot", "ledger.incremental_snapshot") } - accountsPath = getString("accounts-path", "storage.accounts") - if accountsPath == "" { - accountsPath = getString("accounts-path", "ledger.accounts_path") + // storage.accounts is a list of one dir per disk shard; the single-disk case + // is just a slice of one. A scalar string in TOML is accepted too (viper casts + // it to a 1-element slice), and the --accounts-path CLI flag takes 1+ dirs. + accountsPaths = getStringSlice("accounts-path", "storage.accounts") + if len(accountsPaths) == 0 { + accountsPaths = getStringSlice("accounts-path", "ledger.accounts_path") } + if len(accountsPaths) == 0 { + return fmt.Errorf("no accounts path configured (set storage.accounts)") + } + accountsPath = accountsPaths[0] // Check write permission early to fail fast with helpful error - if err := checkDirWritable(accountsPath, "AccountsDB"); err != nil { - return err + for _, p := range accountsPaths { + if err := checkDirWritable(p, "AccountsDB"); err != nil { + return err + } } blockstorePath = getString("ledger-path", "storage.shredstore") if blockstorePath == "" && config.IsSet("storage.blockstore") { @@ -672,6 +684,8 @@ func initConfigAndBindFlags(cmd *cobra.Command) error { snapshot.ZstdDecoderConcurrency = getInt("zstd-decoder-concurrency", "tuning.zstd_decoder_concurrency") snapshot.MaxConcurrentFlushers = getInt("max-concurrent-flushers", "tuning.max_concurrent_flushers") + snapshot.SnapshotFlushSortWorkers = getInt("flush-sort-workers", "tuning.flush_sort_workers") + snapshot.SnapshotDirectIO = getBool("snapshot-directio", "tuning.snapshot_directio") snapshot.SnapshotAppendVecCopyingWorkers = getInt("snapshot-append-vec-workers", "tuning.snapshot_append_vec_workers") snapshot.SnapshotIndexEntryBuilderWorkers = getInt("snapshot-index-builder-workers", "tuning.snapshot_index_builder_workers") snapshot.SnapshotIndexEntryCommitterWorkers = getInt("snapshot-index-committer-workers", "tuning.snapshot_index_committer_workers") @@ -1040,7 +1054,7 @@ func runLive(c *cobra.Command, args []string) { // Build directly from the specified files (BuildAccountsDbPaths handles AccountsDB cleanup internally) // NOTE: We do NOT clean snapshot files in explicit mode - user wants to keep their explicit snapshots dp := progress.NewDualProgress() - accountsDb, manifest, err = snapshot.BuildAccountsDbPaths(ctx, snapshotArchivePath, incrementalSnapshotFilename, accountsPath, dp) + accountsDb, manifest, err = snapshot.BuildAccountsDbPaths(ctx, snapshotArchivePath, incrementalSnapshotFilename, accountsPaths, dp) if err != nil { klog.Fatalf("failed to build AccountsDB from snapshot: %v", err) } @@ -1067,7 +1081,7 @@ func runLive(c *cobra.Command, args []string) { mlog.Log.Infof("WARNING: no state file found, AccountsDB may be from incomplete build") } mlog.Log.Infof("Resuming from existing AccountsDB at slot %d", accountsDBSlot) - accountsDb, err = accountsdb.OpenDb(accountsPath) + accountsDb, err = accountsdb.OpenDb(accountsPaths) if err != nil { klog.Fatalf("failed to open AccountsDB at %s: %v", accountsPath, err) } @@ -1098,7 +1112,7 @@ func runLive(c *cobra.Command, args []string) { state.RecordRebuild(accountsPath, 0, "", getVersion(), getCommit(), getBranch(), "new-snapshot mode (no prior state)") } mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath) - snapshot.CleanAccountsDbDir(accountsPath) + snapshot.CleanAccountsDbDir(accountsPaths) } // Clean existing snapshots (respecting retention setting) if snapshotDownloadPath != "" { @@ -1109,7 +1123,7 @@ func runLive(c *cobra.Command, args []string) { mlog.Log.Infof("Cleaning up existing snapshot files in %s (keeping %d)", snapshotDownloadPath, maxSnapshots) snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots) } - accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath) + accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath) if err != nil { klog.Fatalf("failed to build AccountsDB from snapshot: %v", err) } @@ -1138,7 +1152,7 @@ func runLive(c *cobra.Command, args []string) { state.RecordRebuild(accountsPath, 0, "", getVersion(), getCommit(), getBranch(), "snapshot mode (no prior state)") } mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath) - snapshot.CleanAccountsDbDir(accountsPath) + snapshot.CleanAccountsDbDir(accountsPaths) } // Check for existing fresh snapshot @@ -1151,7 +1165,7 @@ func runLive(c *cobra.Command, args []string) { if existingSnap != nil { // Reuse existing snapshot mlog.Log.Infof("Reusing existing snapshot file at slot %d", existingSnap.slot) - accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPath, blockstorePath, rpcEndpoints) + accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPaths, blockstorePath, rpcEndpoints) } else { // Download fresh mlog.Log.Infof("no fresh snapshot file found, downloading new one") @@ -1163,7 +1177,7 @@ func runLive(c *cobra.Command, args []string) { } snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots) } - accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath) + accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath) } if err != nil { klog.Fatalf("failed to build AccountsDB from snapshot: %v", err) @@ -1217,13 +1231,13 @@ func runLive(c *cobra.Command, args []string) { // mithrilState is guaranteed non-nil here (we prompted because it was stale) state.RecordRebuild(accountsPath, mithrilState.LastSlot, mithrilState.LastBankhash, getVersion(), getCommit(), getBranch(), "user chose rebuild (stale AccountsDB)") mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath) - snapshot.CleanAccountsDbDir(accountsPath) + snapshot.CleanAccountsDbDir(accountsPaths) } // Check for existing fresh snapshot existingSnap := detectFreshSnapshot(snapshotDownloadPath, fullThreshold, rpcEndpoints, ctx) if existingSnap != nil { mlog.Log.Infof("Reusing existing snapshot file at slot %d", existingSnap.slot) - accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPath, blockstorePath, rpcEndpoints) + accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPaths, blockstorePath, rpcEndpoints) } else { // Clean up old snapshot files if snapshotDownloadPath != "" { @@ -1233,7 +1247,7 @@ func runLive(c *cobra.Command, args []string) { } snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots) } - accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath) + accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath) } if err != nil { klog.Fatalf("failed to build AccountsDB from snapshot: %v", err) @@ -1255,7 +1269,7 @@ func runLive(c *cobra.Command, args []string) { mlog.Log.Infof("mode=auto: Resuming from existing AccountsDB at slot %d", accountsDBSlot) // Record resume in history state.RecordResume(accountsPath, mithrilState.LastSlot, mithrilState.LastBankhash, replay.CurrentRunID, getVersion(), getCommit(), getBranch()) - accountsDb, err = accountsdb.OpenDb(accountsPath) + accountsDb, err = accountsdb.OpenDb(accountsPaths) if err != nil { klog.Fatalf("failed to open AccountsDB at %s: %v", accountsPath, err) } @@ -1311,14 +1325,14 @@ func runLive(c *cobra.Command, args []string) { state.RecordRebuild(accountsPath, 0, "", getVersion(), getCommit(), getBranch(), reason) } mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath) - snapshot.CleanAccountsDbDir(accountsPath) + snapshot.CleanAccountsDbDir(accountsPaths) } // Check for existing fresh snapshot existingSnap := detectFreshSnapshot(snapshotDownloadPath, fullThreshold, rpcEndpoints, ctx) if existingSnap != nil { mlog.Log.Infof("Reusing existing snapshot file at slot %d", existingSnap.slot) - accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPath, blockstorePath, rpcEndpoints) + accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPaths, blockstorePath, rpcEndpoints) } else { // Clean up old snapshot files based on retention settings maxSnapshots := config.GetInt("snapshot.max_full_snapshots") @@ -1326,7 +1340,7 @@ func runLive(c *cobra.Command, args []string) { maxSnapshots = 1 // default: keep 1 snapshot } snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots) - accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath) + accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath) } if err != nil { klog.Fatalf("failed to build AccountsDB from snapshot: %v", err) @@ -2222,7 +2236,7 @@ func queryLatestSnapshotSlot(ctx context.Context, rpcEndpoints []string) (uint64 } // buildFromExistingSnapshot builds AccountsDB from an existing downloaded snapshot file. -func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshotDir, accountsPath, blockstorePath string, rpcEndpoints []string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) { +func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshotDir string, accountsPaths []string, blockstorePath string, rpcEndpoints []string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) { snapCfg := buildSnapshotConfig(rpcEndpoints) // Construct full path to snapshot file @@ -2232,7 +2246,7 @@ func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshot // Create progress display for extract dp := progress.NewDualProgress() - accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotPath, snapshotDir, int(snap.slot), int(snap.slot), accountsPath, rpcEndpoints, blockstorePath, snapCfg, dp) + accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotPath, snapshotDir, int(snap.slot), int(snap.slot), accountsPaths, rpcEndpoints, blockstorePath, snapCfg, dp) if err != nil { return nil, nil, fmt.Errorf("failed to build AccountsDB from snapshot: %w", err) } @@ -2242,7 +2256,7 @@ func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshot } // downloadAndBuildFromSnapshot finds, downloads, and builds AccountsDB from a snapshot -func downloadAndBuildFromSnapshot(ctx context.Context, rpcEndpoints []string, snapshotDownloadPath, accountsPath, blockstorePath string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) { +func downloadAndBuildFromSnapshot(ctx context.Context, rpcEndpoints []string, snapshotDownloadPath string, accountsPaths []string, blockstorePath string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) { snapCfg := buildSnapshotConfig(rpcEndpoints) fullSnapshotDlStart := time.Now() fullSnapshotInfo, err := snapshotdl.GetSnapshotURLWithInfo(ctx, snapCfg) @@ -2266,7 +2280,7 @@ func downloadAndBuildFromSnapshot(ctx context.Context, rpcEndpoints []string, sn // Create progress display for snapshot download and extract dp := progress.NewDualProgress() - accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotURL, snapshotDownloadPath, fullSnapshotSlot, fullSnapshotSlot, accountsPath, rpcEndpoints, blockstorePath, snapCfg, dp) + accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotURL, snapshotDownloadPath, fullSnapshotSlot, fullSnapshotSlot, accountsPaths, rpcEndpoints, blockstorePath, snapCfg, dp) if err != nil { return nil, nil, fmt.Errorf("failed to build AccountsDB from snapshot: %w", err) } diff --git a/config.example.toml b/config.example.toml index baab2e89..3a476a7c 100644 --- a/config.example.toml +++ b/config.example.toml @@ -71,6 +71,18 @@ name = "mithril" # Put this on your fastest NVMe due to heavy random I/O. accounts = "/mnt/mithril-accounts" + # For multi-disk setups, accounts may instead be a list of directories, one + # per physical disk. The snapshot unpack shards append-vecs across them so + # aggregate write throughput approaches the sum of the disks. The first entry + # also holds the shared metadata (account index, manifest, state); the others + # hold only their disk's shard of the account data. The number of directories + # is fixed at build time - changing it requires rebuilding the AccountsDB. + # accounts = [ + # "/mnt/nvme0/mithril-accounts", # primary: metadata + shard 0 + # "/mnt/nvme1/mithril-accounts", # shard 1 + # "/mnt/nvme2/mithril-accounts", # shard 2 + # ] + # Shredstore - Lightbringer stores received shreds here # Used for block streaming and potential repair serving. shredstore = "/mnt/mithril-ledger/shredstore" @@ -362,6 +374,18 @@ name = "mithril" # Bound for number of index shards to convert to SSTs at once. max_concurrent_flushers = 8 + # Concurrent sort workers used during the index flush. + # 0 (default) = one per CPU. + # flush_sort_workers = 0 + + # Write the snapshot append-vec big files with O_DIRECT, bypassing the OS page + # cache. On a ~460GB unpack the buffered path churns hundreds of GB of dirty + # pages, which evicts the zstd-decode working set; O_DIRECT avoids that and + # gives a modest (~6%) unpack speedup that grows on low-RAM boxes. Off by + # default (buffered is the safe choice). Requires a filesystem that supports + # O_DIRECT (ext4/xfs; not tmpfs). + # snapshot_directio = false + # Size in MB for serialized parameter arena (0 to disable) param_arena_size_mb = 512 diff --git a/pkg/accountsdb/accountsdb.go b/pkg/accountsdb/accountsdb.go index acca39e0..4e47fd29 100644 --- a/pkg/accountsdb/accountsdb.go +++ b/pkg/accountsdb/accountsdb.go @@ -13,7 +13,6 @@ import ( "path/filepath" "runtime/trace" "sync" - "sync/atomic" "github.com/Overclock-Validator/mithril/pkg/accounts" "github.com/Overclock-Validator/mithril/pkg/addresses" @@ -28,8 +27,8 @@ import ( type AccountsDb struct { Index *pebble.DB BankHashStore *pebble.DB - AcctsDir string - LargestFileId atomic.Uint64 + AcctsDir string // primary shard's accounts dir; its parent is the metadata dir + Shards *Shards VoteAcctCache otter.Cache[solana.PublicKey, *accounts.Account] CommonAcctsCache otter.Cache[solana.PublicKey, *accounts.Account] ProgramCache otter.Cache[solana.PublicKey, *ProgramCacheEntry] @@ -87,33 +86,37 @@ func NewAccountsIndexPebbleOptions(logger pebble.Logger) *pebble.Options { } } -func OpenDb(accountsDbDir string) (*AccountsDb, error) { - // check for existence of the 'accounts' directory, which holds the appendvecs - appendVecsDir := fmt.Sprintf("%s/accounts", accountsDbDir) - _, err := os.Stat(appendVecsDir) - if err != nil { - return nil, err +func OpenDb(accountsPaths []string) (*AccountsDb, error) { + if len(accountsPaths) == 0 { + return nil, fmt.Errorf("OpenDb: no accounts paths configured") } + // The first path holds all metadata (index, manifest, state); every path + // holds an "accounts" dir with that disk's shard data. + accountsDbDir := accountsPaths[0] - // attempt to open largest_file_id file - largestFileIdFn := fmt.Sprintf("%s/largest_file_id", accountsDbDir) - lfi, err := os.Open(largestFileIdFn) + // num_shards records the shard count the DB was built with. + b, err := os.ReadFile(filepath.Join(accountsDbDir, "num_shards")) if err != nil { - mlog.Log.Infof("failed to open %s\n", largestFileIdFn) - return nil, err + return nil, fmt.Errorf("reading num_shards: %w", err) + } + if len(b) != 8 { + return nil, fmt.Errorf("num_shards: expected 8 bytes, got %d", len(b)) + } + numShards := int(binary.LittleEndian.Uint64(b)) + if numShards != len(accountsPaths) { + return nil, fmt.Errorf("configured %d accounts dir(s) but AccountsDB was built with %d shard(s); rebuild required", len(accountsPaths), numShards) } - largestFileIdBytes := make([]byte, 8) - bytesRead, err := lfi.Read(largestFileIdBytes) - if err != nil { - mlog.Log.Infof("error reading %s: %s\n", largestFileIdFn, err) - return nil, err - } else if bytesRead != 8 { - mlog.Log.Infof("error reading %s: expected 8 bytes, got %d\n", largestFileIdFn, bytesRead) - return nil, fmt.Errorf("only got %d bytes", bytesRead) + shardDirs := make([]string, len(accountsPaths)) + for i, p := range accountsPaths { + shardDirs[i] = filepath.Join(p, "accounts") } - largestFileId := binary.LittleEndian.Uint64(largestFileIdBytes) + // check for existence of the primary 'accounts' directory, which holds the appendvecs + appendVecsDir := shardDirs[0] + if _, err := os.Stat(appendVecsDir); err != nil { + return nil, err + } indexDir := filepath.Join(accountsDbDir, "mithril_db") db, err := pebble.Open(indexDir, NewAccountsIndexPebbleOptions(silentLogger{})) @@ -127,8 +130,10 @@ func OpenDb(accountsDbDir string) (*AccountsDb, error) { return nil, fmt.Errorf("opening bankhashDir=%s: %w", bankhashDir, err) } - accountsDb := &AccountsDb{Index: db, BankHashStore: bankhashDb, AcctsDir: appendVecsDir} - accountsDb.LargestFileId.Store(largestFileId) + // The counter seeds runtime fileId minting; left at 0 so the first minted id + // lands in segment 1, just past segment 0 (the coalesced "data" big file). + shards := newShards(shardDirs) + accountsDb := &AccountsDb{Index: db, BankHashStore: bankhashDb, AcctsDir: appendVecsDir, Shards: shards} accountsDb.inProgressStoreRequests = list.New() accountsDb.storeRequestChan = make(chan *list.Element) @@ -282,7 +287,7 @@ func (accountsDb *AccountsDb) getStoredAccount(slot uint64, pubkey solana.Public } c.Close() - appendVecFileName := fmt.Sprintf("%s/%d.%d", accountsDb.AcctsDir, acctIdxEntry.Slot, acctIdxEntry.FileId) + appendVecFileName := accountsDb.Shards.path(acctIdxEntry.Slot, acctIdxEntry.FileId) appendVecFile, err := os.Open(appendVecFileName) if err != nil { @@ -421,8 +426,8 @@ func (accountsDb *AccountsDb) storeWorker() { } func (accountsDb *AccountsDb) storeAccountsInternal(accts []*accounts.Account, slot uint64) { - fileId := accountsDb.LargestFileId.Add(1) - appendVecFileName := fmt.Sprintf("%s/%d.%d", accountsDb.AcctsDir, slot, fileId) + fileId := accountsDb.Shards.mint(accountsDb.Shards.choose()) + appendVecFileName := accountsDb.Shards.path(slot, fileId) appendVecFile, err := os.OpenFile(appendVecFileName, os.O_RDWR|os.O_CREATE, 0666) if err != nil { //mlog.Log.Debugf("unable to open appendvec file %s for writing to accountsdb", appendVecFileName) @@ -459,7 +464,7 @@ func (accountsDb *AccountsDb) storeAccountsInternal(accts []*accounts.Account, s } c.Close() - existingAppendVecFileName := fmt.Sprintf("%s/%d.%d", accountsDb.AcctsDir, acctIdxEntry.Slot, acctIdxEntry.FileId) + existingAppendVecFileName := accountsDb.Shards.path(acctIdxEntry.Slot, acctIdxEntry.FileId) existingAppendVecFile, err := os.OpenFile(existingAppendVecFileName, os.O_RDWR, 0666) if err != nil { panic(err) @@ -561,7 +566,7 @@ func (accountsDb *AccountsDb) parallelStoreAccounts(n int, accts []*accounts.Acc return fmt.Errorf("unmarshaling index entry: %w", err) } - existingAppendVecFileName := fmt.Sprintf("%s/%d.%d", accountsDb.AcctsDir, existingIdxEntry.Slot, existingIdxEntry.FileId) + existingAppendVecFileName := accountsDb.Shards.path(existingIdxEntry.Slot, existingIdxEntry.FileId) existingAppendVecFile, err := os.OpenFile(existingAppendVecFileName, os.O_RDWR, 0666) if err != nil { return fmt.Errorf("open %s: %w", existingAppendVecFileName, err) @@ -606,8 +611,8 @@ func (accountsDb *AccountsDb) parallelStoreAccounts(n int, accts []*accounts.Acc } newAppendVecGroup := errgroup.Group{} newAppendVecGroup.Go(func() error { - fileId := accountsDb.LargestFileId.Add(1) - appendVecFileName := fmt.Sprintf("%s/%d.%d", accountsDb.AcctsDir, slot, fileId) + fileId := accountsDb.Shards.mint(accountsDb.Shards.choose()) + appendVecFileName := accountsDb.Shards.path(slot, fileId) appendVecFile, err := os.OpenFile(appendVecFileName, os.O_RDWR|os.O_CREATE, 0666) if err != nil { return err diff --git a/pkg/accountsdb/index.go b/pkg/accountsdb/index.go index 4cf25139..88590328 100644 --- a/pkg/accountsdb/index.go +++ b/pkg/accountsdb/index.go @@ -90,7 +90,10 @@ func WriteStakePubkeyIndex(path string, entries []StakeIndexEntry) error { // - pubkeys: all account pubkeys // - acctIdxEntries: index entries for each account // - stakeEntries: stake account pubkeys with their appendvec location hints -func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, fileId uint64) ([]solana.PublicKey, []AccountIndexEntry, []StakeIndexEntry, error) { +// baseOffset is added to every account offset so that, when append-vecs are +// coalesced into a shard's big file, the stored offsets are absolute within that +// file. Pass 0 for a stand-alone (one-file-per-append-vec) layout. +func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, fileId uint64, baseOffset uint64) ([]solana.PublicKey, []AccountIndexEntry, []StakeIndexEntry, error) { pubkeys := make([]solana.PublicKey, 0, 20000) acctIdxEntries := make([]AccountIndexEntry, 0, 20000) stakeEntries := make([]StakeIndexEntry, 0, 1000) @@ -108,6 +111,8 @@ func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, acctIdxEntries = acctIdxEntries[:len(acctIdxEntries)-1] break } + // make the offset absolute within the (possibly coalesced) big file + acctIdxEntries[len(acctIdxEntries)-1].Offset += baseOffset // Collect stake account entries with appendvec location hints if bytes.Equal(owner[:], addresses.StakeProgramAddr[:]) { idx := len(acctIdxEntries) - 1 diff --git a/pkg/accountsdb/shards.go b/pkg/accountsdb/shards.go new file mode 100644 index 00000000..8b1ccf23 --- /dev/null +++ b/pkg/accountsdb/shards.go @@ -0,0 +1,54 @@ +package accountsdb + +import ( + "fmt" + "path/filepath" + "sync/atomic" +) + +// A fileId encodes both the disk and the file holding an append-vec: +// +// shard = fileId % N +// segment = fileId / N +// +// Segment 0 is the coalesced snapshot big file ("data", holding both the full and +// incremental snapshots); any higher segment is a runtime append-vec written as +// "." in the shard dir. Offsets stored in the index are absolute +// within the resolved file. +const segData = 0 + +// Shards resolves append-vec file ids onto one directory per physical disk and +// mints new ids for runtime writes. +type Shards struct { + dirs []string // one per disk, each is "/accounts" + counter atomic.Uint64 + rr atomic.Uint64 +} + +func newShards(dirs []string) *Shards { + return &Shards{dirs: dirs} +} + +func (s *Shards) n() uint64 { return uint64(len(s.dirs)) } + +// choose round-robins runtime writes across shards. Runtime append-vec volume is +// low, so an even file-count spread is enough; bulk unpack placement lives in the +// build-time writer. +func (s *Shards) choose() int { + return int(s.rr.Add(1) % s.n()) +} + +// mint allocates a fresh runtime fileId (segment >= 1) on shard. The counter starts +// at 0, so the first minted id lands in segment 1, just past the segment-0 big file. +func (s *Shards) mint(shard int) uint64 { + return s.counter.Add(1)*s.n() + uint64(shard) +} + +// path returns the on-disk file holding the data addressed by (slot, fileId). +func (s *Shards) path(slot, fileId uint64) string { + dir := s.dirs[fileId%s.n()] + if fileId/s.n() == segData { + return filepath.Join(dir, "data") + } + return filepath.Join(dir, fmt.Sprintf("%d.%d", slot, fileId)) +} diff --git a/pkg/accountsdb/shards_test.go b/pkg/accountsdb/shards_test.go new file mode 100644 index 00000000..a00b5b4b --- /dev/null +++ b/pkg/accountsdb/shards_test.go @@ -0,0 +1,49 @@ +package accountsdb + +import ( + "testing" +) + +func TestShardsPathAddressing(t *testing.T) { + dirs := []string{"/d0", "/d1", "/d2"} + s := newShards(dirs) + + cases := []struct { + fileId uint64 + want string + }{ + {0, "/d0/data"}, // segment 0 (coalesced big file), shard 0 + {1, "/d1/data"}, // shard 1 + {2, "/d2/data"}, // shard 2 + {3, "/d0/100.3"}, // segment 1 -> runtime per-file, shard 0 + {6, "/d0/100.6"}, // segment 2, shard 0 + {8, "/d2/100.8"}, // shard 2 + {10, "/d1/100.10"}, + } + for _, c := range cases { + if got := s.path(100, c.fileId); got != c.want { + t.Errorf("path(100,%d) = %q, want %q", c.fileId, got, c.want) + } + } +} + +func TestShardsMintEncodesShardAndSegment(t *testing.T) { + s := newShards([]string{"/d0", "/d1", "/d2"}) + // counter left at 0 (matches OpenDb): first mint lands in segment 1. + + seen := map[uint64]bool{} + for i := 0; i < 30; i++ { + shard := i % 3 + id := s.mint(shard) + if int(id%s.n()) != shard { + t.Fatalf("mint(shard=%d) gave id=%d with id%%N=%d", shard, id, id%s.n()) + } + if id/s.n() < 1 { + t.Fatalf("runtime id=%d landed in the segment-0 big file", id) + } + if seen[id] { + t.Fatalf("mint produced duplicate id=%d", id) + } + seen[id] = true + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index b4fe0bda..1432e4af 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -52,6 +52,8 @@ type DebugConfig struct { type DevelopmentConfig struct { ZstdDecoderConcurrency int `toml:"zstd_decoder_concurrency" mapstructure:"zstd_decoder_concurrency"` // was: zstd-decoder-concurrency MaxConcurrentFlushers int `toml:"max_concurrent_flushers" mapstructure:"max_concurrent_flushers"` // was: max-concurrent-flushers + FlushSortWorkers int `toml:"flush_sort_workers" mapstructure:"flush_sort_workers"` // Snapshot index-flush concurrent sort workers (0 = NumCPU) + SnapshotDirectIO bool `toml:"snapshot_directio" mapstructure:"snapshot_directio"` // Write snapshot big files with O_DIRECT (bypasses page cache) SnapshotAppendVecWorkers int `toml:"snapshot_append_vec_workers" mapstructure:"snapshot_append_vec_workers"` // Snapshot appendvec write workers SnapshotIndexBuilderWorkers int `toml:"snapshot_index_builder_workers" mapstructure:"snapshot_index_builder_workers"` // Snapshot index parsing workers SnapshotIndexCommitterWorkers int `toml:"snapshot_index_committer_workers" mapstructure:"snapshot_index_committer_workers"` // Snapshot index shard enqueue workers diff --git a/pkg/snapshot/build_db.go b/pkg/snapshot/build_db.go index bc3f9da7..a33dcfa2 100644 --- a/pkg/snapshot/build_db.go +++ b/pkg/snapshot/build_db.go @@ -1,13 +1,12 @@ package snapshot import ( - "bytes" "context" - "encoding/binary" "fmt" "io" "os" "path/filepath" + "runtime" "strings" "sync" "sync/atomic" @@ -27,6 +26,11 @@ const ( DefaultSnapshotAppendVecCopyingWorkers = 32 DefaultSnapshotIndexShards = 64 DefaultSnapshotMaxConcurrentFlushers = 8 + // DefaultSnapshotFlushSortWorkers == 0 means "auto": use runtime.NumCPU(). + DefaultSnapshotFlushSortWorkers = 0 + // DefaultSnapshotDirectIO keeps O_DIRECT big-file writes off by default; + // buffered is the safe default and O_DIRECT is an opt-in tuning knob. + DefaultSnapshotDirectIO = false ) var ( @@ -34,13 +38,18 @@ var ( SnapshotIndexEntryBuilderWorkers = DefaultSnapshotIndexEntryBuilderWorkers SnapshotAppendVecCopyingWorkers = DefaultSnapshotAppendVecCopyingWorkers SnapshotIndexShards = DefaultSnapshotIndexShards + SnapshotFlushSortWorkers = DefaultSnapshotFlushSortWorkers + SnapshotDirectIO = DefaultSnapshotDirectIO SnapshotIndexTempDir string ) // CleanAccountsDbDir removes all artifacts from a previous incomplete snapshot run. // This prevents corruption from Ctrl+C or partial downloads. // Exported so it can be called early in startup before any failures. -func CleanAccountsDbDir(accountsDbDir string) { +func CleanAccountsDbDir(accountsPaths []string) { + if len(accountsPaths) == 0 { + return + } // List of all files/directories that may be left from a previous incomplete run artifacts := []string{ "accounts", @@ -48,11 +57,19 @@ func CleanAccountsDbDir(accountsDbDir string) { "mithril_db_log_shards", "bankhash_db", "largest_file_id", + "num_shards", "manifest", "mithril_state.json", // State file for tracking valid builds and replay progress } for _, artifact := range artifacts { - path := filepath.Join(accountsDbDir, artifact) + path := filepath.Join(accountsPaths[0], artifact) + if err := os.RemoveAll(path); err != nil { + mlog.Log.Errorf("failed to remove %s: %v", path, err) + } + } + // remove the accounts dir on the other shard disks too + for _, p := range accountsPaths[1:] { + path := filepath.Join(p, "accounts") if err := os.RemoveAll(path); err != nil { mlog.Log.Errorf("failed to remove %s: %v", path, err) } @@ -200,17 +217,28 @@ func snapshotMaxConcurrentFlushers() int { return positiveOrDefault(MaxConcurrentFlushers, DefaultSnapshotMaxConcurrentFlushers) } +// snapshotFlushSortWorkers is how many shard buffers may be sorted concurrently +// during the index flush. 0 (the default) means auto: one per logical CPU. +func snapshotFlushSortWorkers() int { + if SnapshotFlushSortWorkers > 0 { + return SnapshotFlushSortWorkers + } + return runtime.NumCPU() +} + func logSnapshotBootstrapTuning() { indexTempDir := SnapshotIndexTempDir if indexTempDir == "" { indexTempDir = "(accountsdb)" } - mlog.Log.Infof("Snapshot bootstrap tuning: append_vec_workers=%d index_builder_workers=%d index_committer_workers=%d index_shards=%d max_concurrent_flushers=%d zstd_decoder_concurrency=%d index_temp_dir=%s", + mlog.Log.Infof("Snapshot bootstrap tuning: append_vec_workers=%d index_builder_workers=%d index_committer_workers=%d index_shards=%d max_concurrent_flushers=%d flush_sort_workers=%d directio=%v zstd_decoder_concurrency=%d index_temp_dir=%s", snapshotAppendVecCopyingWorkers(), snapshotIndexEntryBuilderWorkers(), snapshotIndexEntryCommitterWorkers(), snapshotIndexShards(), snapshotMaxConcurrentFlushers(), + snapshotFlushSortWorkers(), + SnapshotDirectIO, ZstdDecoderConcurrency, indexTempDir) } @@ -245,11 +273,16 @@ func BuildAccountsDbPaths( ctx context.Context, snapshotFile string, incrementalSnapshotFile string, - accountsDbDir string, + accountsPaths []string, dp *progress.DualProgress, ) (*accountsdb.AccountsDb, *SnapshotManifest, error) { + if len(accountsPaths) == 0 { + return nil, nil, fmt.Errorf("no accounts paths configured") + } + // The first path holds all metadata; every path holds a shard's accounts dir. + accountsDbDir := accountsPaths[0] // Clean any leftover artifacts from previous incomplete runs (e.g., Ctrl+C) - CleanAccountsDbDir(accountsDbDir) + CleanAccountsDbDir(accountsPaths) mlog.Log.Infof("Parsing full snapshot manifest...") manifest, err := UnmarshalManifestFromSnapshot(ctx, snapshotFile, accountsDbDir) @@ -270,15 +303,21 @@ func BuildAccountsDbPaths( start := time.Now() - appendVecsOutputDir := filepath.Join(accountsDbDir, "accounts") - if err = os.MkdirAll(appendVecsOutputDir, 0775); err != nil { - return nil, nil, err + shardDirs := make([]string, len(accountsPaths)) + for i, p := range accountsPaths { + shardDirs[i] = filepath.Join(p, "accounts") + if err = os.MkdirAll(shardDirs[i], 0775); err != nil { + return nil, nil, err + } + } + shardFiles, err := openShardBigFiles(shardDirs) + if err != nil { + return nil, nil, fmt.Errorf("opening shard big files: %w", err) } logSnapshotBootstrapTuning() defer ants.Release() - var largestFileId atomic.Uint64 wg := &sync.WaitGroup{} logsDir, cleanupIndexWorkDir, err := prepareSnapshotIndexWorkDir(accountsDbDir) @@ -294,7 +333,7 @@ func BuildAccountsDbPaths( entries: make([]accountsdb.StakeIndexEntry, 0, 1000000), // Pre-allocate for ~1M stake accounts } - pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, accountsDbDir, &largestFileId, stakeCollector) + pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, shardFiles, stakeCollector) if err != nil { return nil, nil, fmt.Errorf("initializing worker pools: %w", err) } @@ -311,7 +350,7 @@ func BuildAccountsDbPaths( // Process snapshots sequentially for better performance (less lock contention) // Full snapshot first - err = readTar(ctx, wg, snapshotFile, pools.appendVecCopying, readTarOptions{progress: dp}) + err = readTar(ctx, wg, snapshotFile, pools.appendVecCopying, pools.tarBufs, readTarOptions{progress: dp}) // Wait for ALL worker tasks from full snapshot to complete before starting incremental wg.Wait() @@ -331,7 +370,7 @@ func BuildAccountsDbPaths( // Process incremental snapshot (if provided) if incrementalSnapshotFile != "" { - err = readTar(ctx, wg, incrementalSnapshotFile, pools.appendVecCopying, + err = readTar(ctx, wg, incrementalSnapshotFile, pools.appendVecCopying, pools.tarBufs, readTarOptions{isIncremental: true}) if err != nil { return nil, nil, err @@ -340,6 +379,11 @@ func BuildAccountsDbPaths( wg.Wait() } + // flush and close every shard's big file now that all appends are done + if err := shardFiles.close(); err != nil { + return nil, nil, fmt.Errorf("closing shard big files: %w", err) + } + mlog.Log.Debugf("done processing snapshots in %s.", fmtDuration(time.Since(start))) // Show indexing progress for shard flush (no gap between DualProgress and this) @@ -360,12 +404,7 @@ func BuildAccountsDbPaths( mlog.Log.Infof("Snapshot processed in %s.", fmtDuration(time.Since(start))) - var largestFileIdBytes [8]byte - binary.LittleEndian.PutUint64(largestFileIdBytes[:], largestFileId.Load()) - - path := filepath.Join(accountsDbDir, "largest_file_id") - if err := os.WriteFile(path, largestFileIdBytes[:], 0644); err != nil { - mlog.Log.Errorf("error while writing largest file ID=%d to %s: %s", largestFileId.Load(), path, err) + if err := writeShardMetadata(accountsDbDir, len(shardDirs)); err != nil { return nil, nil, err } @@ -390,7 +429,7 @@ func BuildAccountsDbPaths( } bankhashDb.Close() - accountsDb, err := accountsdb.OpenDb(accountsDbDir) + accountsDb, err := accountsdb.OpenDb(accountsPaths) if err != nil { return nil, nil, err } @@ -416,11 +455,38 @@ type readTarOptions struct { isIncremental bool } +// tarBufPool recycles the byte buffers that appendvec data is read into, saving +// ~420k allocations totalling ~470 GB. A buffer returns to the pool once the index +// builder is done reading it. Owned by snapshotWorkerPools so it may be GCed after +// snapshot unpacking. +type tarBufPool struct{ sync.Pool } + +func newTarBufPool() *tarBufPool { + return &tarBufPool{Pool: sync.Pool{New: func() any { b := []byte(nil); return &b }}} +} + +func (p *tarBufPool) get(size int) *[]byte { + bp := p.Get().(*[]byte) + if cap(*bp) < size { + *bp = make([]byte, size) + } else { + *bp = (*bp)[:size] + } + return bp +} + +func (p *tarBufPool) put(bp *[]byte) { + if bp != nil { + p.Put(bp) + } +} + func readTar( ctx context.Context, wg *sync.WaitGroup, filename string, appendVecCopyingPool *ants.PoolWithFunc, + bufs *tarBufPool, options readTarOptions, ) error { dp := options.progress @@ -449,13 +515,19 @@ func readTar( } } + var totalTarNext, totalReadCopy, totalDispatch time.Duration + var entryCount int + var bytesOut int64 + readTarStart := time.Now() for { if ctx.Err() != nil { mlog.Log.Infof("Context cancelled, stopping snapshot unpack: %v", ctx.Err()) cleanupPartial("cancelled") return ctx.Err() } + t0 := time.Now() header, err := tarReader.Next() + totalTarNext += time.Since(t0) if err == io.EOF { break } else if err != nil { @@ -468,23 +540,30 @@ func readTar( continue } - writer := bytes.NewBuffer(make([]byte, 0, header.Size)) - tarBytesRead, err := io.Copy(writer, tarReader) + t1 := time.Now() + bp := bufs.get(int(header.Size)) + _, err = io.ReadFull(tarReader, *bp) + totalReadCopy += time.Since(t1) if err != nil { - mlog.Log.Errorf("err copying data to reader: %s\n", err) - cleanupPartial("copy error") + mlog.Log.Errorf("err reading tar entry data: %s\n", err) + cleanupPartial("read error") + bufs.put(bp) return err } - statsd.Count(statsd.SnapshotTarBytesRead, tarBytesRead, nil) + entryCount++ + bytesOut += header.Size + statsd.Count(statsd.SnapshotTarBytesRead, header.Size, nil) // Update extract progress if dp != nil { - dp.Extract.Add(tarBytesRead) + dp.Extract.Add(header.Size) } - task := appendVecCopyingTask{TarBuffer: writer, Filename: header.Name, FromIncrementalSnapshot: options.isIncremental} + task := appendVecCopyingTask{Buf: *bp, BufRef: bp, Filename: header.Name, FromIncrementalSnapshot: options.isIncremental} wg.Add(1) + t2 := time.Now() err = appendVecCopyingPool.Invoke(task) + totalDispatch += time.Since(t2) if err != nil { mlog.Log.Errorf("error calling appendVecCopyingPool.Invoke: %v", err) cleanupPartial("pool error") @@ -492,6 +571,15 @@ func readTar( } } + elapsed := time.Since(readTarStart) + mibps := 0.0 + if elapsed > 0 { + mibps = float64(bytesOut) / (1 << 20) / elapsed.Seconds() + } + mlog.Log.Infof("readTar timing: entries=%d bytesOut=%dMB elapsed=%s avgMiBps=%.0f tarNext=%s readCopy=%s dispatch=%s", + entryCount, bytesOut/(1<<20), fmtDuration(elapsed), mibps, + fmtDuration(totalTarNext), fmtDuration(totalReadCopy), fmtDuration(totalDispatch)) + // Successfully processed the entire tar — finalize by renaming from .partial if err := FinalizePartialDownload(savePath); err != nil { mlog.Log.Errorf("Failed to finalize snapshot download: %v", err) @@ -506,6 +594,7 @@ type snapshotWorkerPools struct { appendVecCopying *ants.PoolWithFunc indexEntryBuilder *ants.PoolWithFunc indexEntryCommitter *ants.PoolWithFunc + tarBufs *tarBufPool } // stakeIndexCollector aggregates stake account pubkeys from multiple worker goroutines @@ -539,13 +628,13 @@ func initWorkerPools( sl *ShardLogger, manifest *SnapshotManifest, incrementalManifest *SnapshotManifest, - accountsDbDir string, - largestFileId *atomic.Uint64, + shardFiles *shardBigFiles, stakeCollector *stakeIndexCollector, ) (*snapshotWorkerPools, error) { indexEntryCommitterWorkers := snapshotIndexEntryCommitterWorkers() indexEntryBuilderWorkers := snapshotIndexEntryBuilderWorkers() appendVecCopyingWorkers := snapshotAppendVecCopyingWorkers() + tarBufs := newTarBufPool() indexEntryCommitterPool, err := ants.NewPoolWithFunc(indexEntryCommitterWorkers, func(i any) { tasks := indexEntryCommitterInProgress.Add(1) @@ -570,7 +659,8 @@ func initWorkerPools( start := time.Now() defer wg.Done() task := i.(indexEntryBuilderTask) - pubkeys, entries, stakeEntries, err := accountsdb.BuildIndexEntriesFromAppendVecs(task.Data, task.FileSize, task.Slot, task.FileId) + pubkeys, entries, stakeEntries, err := accountsdb.BuildIndexEntriesFromAppendVecs(task.Data, task.FileSize, task.Slot, task.FileId, task.BaseOffset) + tarBufs.put(task.BufRef) // done reading task.Data; recycle the buffer if err != nil { mlog.Log.Errorf("BuildIndexEntriesFromAppendVecs: %v", err) return @@ -599,42 +689,17 @@ func initWorkerPools( defer wg.Done() task := i.(appendVecCopyingTask) filename := task.Filename - writer := task.TarBuffer - - outFilename := filepath.Join(accountsDbDir, filename) - - // validate that the path doesn't escape accountsDbDir (via '../' sequences) - cleanPath := filepath.Clean(outFilename) - if !strings.HasPrefix(cleanPath, filepath.Clean(accountsDbDir)+string(os.PathSeparator)) { - panic(fmt.Sprintf("invalid path in tar archive: %s", filename)) - } - - appendVecBytes := writer.Bytes() - err := os.WriteFile(cleanPath, appendVecBytes, 0644) - if err != nil { - mlog.Log.Errorf("err writing new file=%s: %v", cleanPath, err) - appendVecCopyingInProgress.Add(-1) - return - } + appendVecBytes := task.Buf - var slot, fileId uint64 - if n, err := fmt.Sscanf(filepath.Base(filename), "%d.%d", &slot, &fileId); n != 2 || err != nil { + // origFileId is the snapshot's id, used only to look up the valid data + // length in the manifest + var slot, origFileId uint64 + if n, err := fmt.Sscanf(filepath.Base(filename), "%d.%d", &slot, &origFileId); n != 2 || err != nil { panic(fmt.Sprintf( "failed to parse slot and file from filename=%s basename=%s; parsed n=%d arguments (expected 2) and had err=%v", filename, filepath.Base(filename), n, err)) } - for { - prevLargestFileId := largestFileId.Load() - if fileId <= prevLargestFileId { - break - } - swapped := largestFileId.CompareAndSwap(prevLargestFileId, fileId) - if swapped { - break - } - } - // find the relevant appendvec storage info. use the info from the incremental // snapshot manifest if this account entry is from the incremental snapshot. var fileSize uint64 @@ -644,7 +709,7 @@ func initWorkerPools( panic("tried to process incremental snapshot without having parsed incremental snapshot manifest first!") } for _, av := range incrementalManifest.AccountsDb.Storages[slot].AcctVecs { - if av.Id == fileId { + if av.Id == origFileId { fileSize = av.FileSize usedIncrementalSnapshotVal = true break @@ -654,7 +719,7 @@ func initWorkerPools( if !usedIncrementalSnapshotVal { for _, av := range manifest.AccountsDb.Storages[slot].AcctVecs { - if av.Id == fileId { + if av.Id == origFileId { fileSize = av.FileSize break } @@ -664,9 +729,20 @@ func initWorkerPools( if fileSize == 0 { panic("programming error - fileSize for appendvec was 0") } + if uint64(len(appendVecBytes)) < fileSize { + panic(fmt.Sprintf("appendvec blob (%d bytes) shorter than manifest fileSize (%d)", len(appendVecBytes), fileSize)) + } + + fileId, base, err := shardFiles.write(appendVecBytes[:fileSize]) + if err != nil { + mlog.Log.Errorf("err writing appendvec to shard big file: %v", err) + appendVecCopyingInProgress.Add(-1) + tarBufs.put(task.BufRef) + return + } appendVecCopyingInProgress.Add(-1) - nextTask := indexEntryBuilderTask{Data: appendVecBytes, FileSize: fileSize, Slot: slot, FileId: fileId} + nextTask := indexEntryBuilderTask{Data: appendVecBytes, FileSize: fileSize, Slot: slot, FileId: fileId, BaseOffset: base, BufRef: task.BufRef} wg.Add(1) statsd.Timing(statsd.TasksAppendVecCopyingLatency, uint64(time.Since(start)), nil) err = indexEntryBuilderPool.Invoke(nextTask) @@ -679,9 +755,10 @@ func initWorkerPools( } return &snapshotWorkerPools{ - appendVecCopyingPool, - indexEntryBuilderPool, - indexEntryCommitterPool, + appendVecCopying: appendVecCopyingPool, + indexEntryBuilder: indexEntryBuilderPool, + indexEntryCommitter: indexEntryCommitterPool, + tarBufs: tarBufs, }, nil } diff --git a/pkg/snapshot/build_db_with_incr.go b/pkg/snapshot/build_db_with_incr.go index 2499b0fc..db0a7d3d 100644 --- a/pkg/snapshot/build_db_with_incr.go +++ b/pkg/snapshot/build_db_with_incr.go @@ -2,13 +2,11 @@ package snapshot import ( "context" - "encoding/binary" "fmt" "os" "path/filepath" "strings" "sync" - "sync/atomic" "time" "github.com/Overclock-Validator/mithril/pkg/accountsdb" @@ -40,14 +38,19 @@ func BuildAccountsDbAuto( snapshotDownloadPath string, fullSnapshotSlot int, referenceSlot int, - accountsDbDir string, + accountsPaths []string, rpcEndpoints []string, blockDir string, snapCfg snapshotdl.SnapshotConfig, dp *progress.DualProgress, ) (*accountsdb.AccountsDb, *SnapshotManifest, error) { + if len(accountsPaths) == 0 { + return nil, nil, fmt.Errorf("no accounts paths configured") + } + // The first path holds all metadata; every path holds a shard's accounts dir. + accountsDbDir := accountsPaths[0] // Clean any leftover artifacts from previous incomplete runs (e.g., Ctrl+C) - CleanAccountsDbDir(accountsDbDir) + CleanAccountsDbDir(accountsPaths) mlog.Log.Infof("Parsing full snapshot manifest...") manifest, err := UnmarshalManifestFromSnapshot(ctx, fullSnapshotFile, accountsDbDir) @@ -58,16 +61,22 @@ func BuildAccountsDbAuto( start := time.Now() - appendVecsOutputDir := filepath.Join(accountsDbDir, "accounts") - if err = os.MkdirAll(appendVecsOutputDir, 0775); err != nil { - return nil, nil, err + shardDirs := make([]string, len(accountsPaths)) + for i, p := range accountsPaths { + shardDirs[i] = filepath.Join(p, "accounts") + if err = os.MkdirAll(shardDirs[i], 0775); err != nil { + return nil, nil, err + } + } + shardFiles, err := openShardBigFiles(shardDirs) + if err != nil { + return nil, nil, fmt.Errorf("opening shard big files: %w", err) } logSnapshotBootstrapTuning() defer ants.Release() incrementalManifest := &SnapshotManifest{} - var largestFileId atomic.Uint64 wg := &sync.WaitGroup{} numShards := snapshotIndexShards() @@ -83,7 +92,7 @@ func BuildAccountsDbAuto( entries: make([]accountsdb.StakeIndexEntry, 0, 1000000), // Pre-allocate for ~1M stake accounts } - pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, accountsDbDir, &largestFileId, stakeCollector) + pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, shardFiles, stakeCollector) if err != nil { return nil, nil, fmt.Errorf("initializing worker pools: %w", err) } @@ -114,7 +123,7 @@ func BuildAccountsDbAuto( dp.Start() } - err = readTar(ctx, wg, fullSnapshotFile, pools.appendVecCopying, readTarOptions{savePath: fullSavePath, progress: dp}) + err = readTar(ctx, wg, fullSnapshotFile, pools.appendVecCopying, pools.tarBufs, readTarOptions{savePath: fullSavePath, progress: dp}) if err != nil { if dp != nil { dp.Interrupt(err) @@ -197,7 +206,7 @@ func BuildAccountsDbAuto( } } - err = readTar(ctx, wg, incrementalSnapshotPath, pools.appendVecCopying, readTarOptions{savePath: incrSavePath, isIncremental: true}) + err = readTar(ctx, wg, incrementalSnapshotPath, pools.appendVecCopying, pools.tarBufs, readTarOptions{savePath: incrSavePath, isIncremental: true}) wg.Wait() // Check if we should retry if err == nil { @@ -210,6 +219,11 @@ func BuildAccountsDbAuto( return nil, nil, err } + // flush and close every shard's big file now that all appends are done + if err := shardFiles.close(); err != nil { + return nil, nil, fmt.Errorf("closing shard big files: %w", err) + } + // Show indexing progress for shard flush indexProgress := progress.NewIndexingProgress("Flush (shard logs)") indexProgress.Start(numShards) @@ -223,12 +237,7 @@ func BuildAccountsDbAuto( } index.Close() - var largestFileIdBytes [8]byte - binary.LittleEndian.PutUint64(largestFileIdBytes[:], largestFileId.Load()) - - path := filepath.Join(accountsDbDir, "largest_file_id") - if err := os.WriteFile(path, largestFileIdBytes[:], 0644); err != nil { - mlog.Log.Errorf("error while writing largest file ID=%d to %s: %s", largestFileId.Load(), path, err) + if err := writeShardMetadata(accountsDbDir, len(shardDirs)); err != nil { return nil, nil, err } @@ -253,7 +262,7 @@ func BuildAccountsDbAuto( } bankhashDb.Close() - accountsDb, err := accountsdb.OpenDb(accountsDbDir) + accountsDb, err := accountsdb.OpenDb(accountsPaths) if err != nil { return nil, nil, err } diff --git a/pkg/snapshot/shard.go b/pkg/snapshot/shard.go index 2f5e77a6..7931106f 100644 --- a/pkg/snapshot/shard.go +++ b/pkg/snapshot/shard.go @@ -10,7 +10,7 @@ import ( "math" "os" "path/filepath" - "slices" + "sort" "sync" "sync/atomic" @@ -31,9 +31,6 @@ type shardRequest struct { v accountsdb.AccountIndexEntry } -// ShardProgressCallback is called with (bytesDone, totalBytes) to report shard flush progress -type ShardProgressCallback func(bytesDone, totalBytes int64) - // ShardLogger manages multiple sharded log files type ShardLogger struct { shards []*shard @@ -41,11 +38,6 @@ type ShardLogger struct { wg *sync.WaitGroup flushSem *semaphore.Weighted - // Progress tracking - totalBytes atomic.Int64 // total bytes written to shard logs - bytesDone atomic.Int64 // bytes flushed to cache - onProgress ShardProgressCallback - // closed flag to prevent sends after Close is called (defensive) closed atomic.Bool } @@ -57,7 +49,6 @@ type shard struct { file *os.File requests chan shardRequest logSize int - flushSem *semaphore.Weighted parent *ShardLogger // parent for progress reporting } @@ -81,31 +72,15 @@ func NewShardLogger(numShards int, filePrefix string) *ShardLogger { sl.wg.Add(numShards) for i := range numShards { - sl.shards[i] = newShard(i, filePrefix, sl.flushSem, sl) + sl.shards[i] = newShard(i, filePrefix, sl) go sl.shards[i].processRequests(sl.wg) } return sl } -// SetProgressCallback sets a callback to receive progress updates during shard flushes. -// The callback receives (bytesDone, totalBytes) and is called as bytes are flushed to cache. -func (sl *ShardLogger) SetProgressCallback(cb ShardProgressCallback) { - sl.onProgress = cb -} - -// TotalBytes returns the total bytes written to shard logs -func (sl *ShardLogger) TotalBytes() int64 { - return sl.totalBytes.Load() -} - -// BytesDone returns the bytes that have been flushed to cache -func (sl *ShardLogger) BytesDone() int64 { - return sl.bytesDone.Load() -} - // newShard creates a new shard with the given ID -func newShard(id int, filePrefix string, flushSem *semaphore.Weighted, parent *ShardLogger) *shard { +func newShard(id int, filePrefix string, parent *ShardLogger) *shard { filename := filepath.Join(filePrefix, fmt.Sprintf("%03d", id)) file, err := os.Create(filename) if err != nil { @@ -117,7 +92,6 @@ func newShard(id int, filePrefix string, flushSem *semaphore.Weighted, parent *S writer: bufio.NewWriter(file), file: file, requests: make(chan shardRequest, 100), - flushSem: flushSem, parent: parent, } @@ -136,105 +110,78 @@ func (s *shard) processRequests(wg *sync.WaitGroup) { s.writer.Write(kBytes[:]) req.v.Marshal(&vBytes) s.writer.Write(vBytes[:24]) - - bytesWritten := int64(len(req.k) + vlen) - s.logSize += int(bytesWritten) - - // Track total bytes for progress reporting and notify callback - if s.parent != nil { - total := s.parent.totalBytes.Add(bytesWritten) - if s.parent.onProgress != nil { - // Notify with bytesDone=0 during streaming (before flush) - // The callback can use totalBytes to show indexing progress - s.parent.onProgress(0, total) - } - } + s.logSize += len(req.k) + vlen } } -func (s *shard) logToSST(ctx context.Context) error { - err := s.flushSem.Acquire(ctx, 1) - if err != nil { - return fmt.Errorf("acquiring flush semaphore: %w", err) - } - defer s.flushSem.Release(1) - // Close/flush +const recordSize = 32 + vlen + +// readLog flushes and closes the shard's log, then reads it fully back into a +// pairs slice. +func (s *shard) readLog() ([]shardRequest, error) { if err := s.writer.Flush(); err != nil { - return fmt.Errorf("failed to flush writer: %w", err) + return nil, fmt.Errorf("failed to flush writer: %w", err) } filename := s.file.Name() if err := s.file.Close(); err != nil { - return fmt.Errorf("failed to close file: %w", err) + return nil, fmt.Errorf("failed to close file: %w", err) } - // Read contents from log file, err := os.Open(filename) if err != nil { - return fmt.Errorf("failed to reopen file for reading: %w", err) + return nil, fmt.Errorf("failed to reopen file for reading: %w", err) } defer file.Close() fileInfo, err := file.Stat() if err != nil { - return fmt.Errorf("stat %s: %w", filename, err) + return nil, fmt.Errorf("stat %s: %w", filename, err) } size := fileInfo.Size() - const recordSize = int64(32 + vlen) if rem := size % recordSize; rem != 0 { - return fmt.Errorf("filename=%s had (size=%d) %% (recordSize=%d) = %d", filename, size, recordSize, rem) + return nil, fmt.Errorf("filename=%s had (size=%d) %% (recordSize=%d) = %d", filename, size, recordSize, rem) } - i := 0 pairs := make([]shardRequest, size/recordSize) reader := bufio.NewReader(file) - var buf [32 + vlen]byte - for { - _, err := io.ReadFull(reader, buf[:32+vlen]) - if err == io.EOF { - break - } - if err != nil { - return fmt.Errorf("logToSST read loop: %v", err) + var buf [recordSize]byte + for i := range pairs { + if _, err := io.ReadFull(reader, buf[:]); err != nil { + return nil, fmt.Errorf("readLog read loop: %w", err) } pairs[i].k = solana.PublicKey(buf[:32]) pairs[i].v.Unmarshal((*[24]byte)(buf[32:56])) - i++ + } - // Track progress - if s.parent != nil { - done := s.parent.bytesDone.Add(recordSize) - if s.parent.onProgress != nil { - s.parent.onProgress(done, s.parent.totalBytes.Load()) - } - } + if err := os.Remove(filename); err != nil { + return nil, fmt.Errorf("removing read log %s: %w", filename, err) } + return pairs, nil +} - // Truncate file and replace file/writer pointers - newFile, err := os.Create(filename) - if err != nil { - return fmt.Errorf("failed to truncate file: %w", err) +// byKeySlotDesc sorts entries by pubkey ascending, then by slot descending so the +// first occurrence of each key is its highest slot (which writeSST keeps). It uses +// sort.Interface (index-based Less) rather than slices.SortFunc so the comparator +// indexes into the slice instead of receiving 56-byte shardRequest values by copy. +type byKeySlotDesc []shardRequest + +func (p byKeySlotDesc) Len() int { return len(p) } +func (p byKeySlotDesc) Swap(i, j int) { p[i], p[j] = p[j], p[i] } +func (p byKeySlotDesc) Less(i, j int) bool { + if c := bytes.Compare(p[i].k[:], p[j].k[:]); c != 0 { + return c < 0 } - s.file = newFile - s.writer = bufio.NewWriter(newFile) - s.logSize = 0 - - // Sort - slices.SortFunc(pairs, func(a, b shardRequest) int { - if c := bytes.Compare(a.k[:], b.k[:]); c != 0 { - return c - } - // Make the bigger slot appear first. - if a.v.Slot > b.v.Slot { - return -1 - } else if a.v.Slot == b.v.Slot { - return 0 - } else { - return 1 - } - }) - var vBytes [vlen]byte - // Write to SST - sstFilename := fmt.Sprintf("%s.sst", filename) + return p[i].v.Slot > p[j].v.Slot // bigger slot first +} + +func sortPairs(pairs []shardRequest) { + sort.Sort(byKeySlotDesc(pairs)) +} + +// writeSST writes the (already sorted) pairs to the shard's SST file, keeping the +// first entry per key (highest slot) and skipping duplicates. +func (s *shard) writeSST(pairs []shardRequest) error { + sstFilename := fmt.Sprintf("%s.sst", s.file.Name()) sstFile, err := vfs.Default.Create(sstFilename) if err != nil { return fmt.Errorf("create %s: %w", sstFilename, err) @@ -242,18 +189,21 @@ func (s *shard) logToSST(ctx context.Context) error { defer sstFile.Close() w := sstable.NewWriter(objstorageprovider.NewFileWritable(sstFile), sstable.WriterOptions{}) defer w.Close() - lastWritten := -1 - for i, kv := range pairs { - if lastWritten >= 0 && bytes.Equal(kv.k[:], pairs[lastWritten].k[:]) { + var vBytes [vlen]byte + var lastKey solana.PublicKey + wrote := false + for i := range pairs { + kv := &pairs[i] + if wrote && kv.k == lastKey { continue } kv.v.Marshal(&vBytes) if err := w.Set(kv.k[:], vBytes[:]); err != nil { return fmt.Errorf("writing to SST: %w", err) } - lastWritten = i + lastKey = kv.k + wrote = true } - return nil } @@ -277,8 +227,16 @@ func (sl *ShardLogger) Close(ctx context.Context) error { return sl.CloseWithProgress(ctx, nil) } +// flushJob carries a shard and its read-back log buffer through the flush pipeline. +type flushJob struct { + s *shard + pairs []shardRequest +} + // CloseWithProgress closes all shards with optional progress callback. // The callback is called after each shard flush completes with (completed, total) counts. +// +// The flush runs as a 3-stage pipeline (read log, sort, write SST). func (sl *ShardLogger) CloseWithProgress(ctx context.Context, onProgress func(completed, total int)) error { // Mark as closed before closing channels to prevent late sends sl.closed.Store(true) @@ -291,15 +249,82 @@ func (sl *ShardLogger) CloseWithProgress(ctx context.Context, onProgress func(co total := len(sl.shards) var completed atomic.Int32 - flushWg := &errgroup.Group{} + // Channels are sized to hold every shard, so a job never blocks on send and the + // only thing bounding in-flight memory is the flushSem buffer-slot semaphore. + sortCh := make(chan flushJob, total) + writeCh := make(chan flushJob, total) + + g, gctx := errgroup.WithContext(ctx) + + // Read stage: a shard buffer is only allocated once a slot is free. + shardCh := make(chan *shard, total) for _, s := range sl.shards { - flushWg.Go(func() error { - err := s.logToSST(ctx) - if onProgress != nil { - onProgress(int(completed.Add(1)), total) + shardCh <- s + } + close(shardCh) + + ioWorkers := snapshotMaxConcurrentFlushers() + var readWg sync.WaitGroup + for range ioWorkers { + readWg.Add(1) + g.Go(func() error { + defer readWg.Done() + for s := range shardCh { + if err := sl.flushSem.Acquire(gctx, 1); err != nil { + return err + } + pairs, err := s.readLog() + if err != nil { + sl.flushSem.Release(1) + return err + } + sortCh <- flushJob{s, pairs} + } + return nil + }) + } + go func() { + readWg.Wait() + close(sortCh) + }() + + // Sort stage: CPU-bound. The number of sort workers is tunable ( + // flush-sort-workers, default NumCPU); the effective concurrency + // is still bounded by the live-buffer count (flushSem), so to + // sort more shards at once raise max-concurrent-flushers too. + var sortWg sync.WaitGroup + for range snapshotFlushSortWorkers() { + sortWg.Add(1) + g.Go(func() error { + defer sortWg.Done() + for job := range sortCh { + sortPairs(job.pairs) + writeCh <- job } - return err + return nil }) } - return flushWg.Wait() + go func() { + sortWg.Wait() + close(writeCh) + }() + + // Write the SST and release the buffer slot. + for range ioWorkers { + g.Go(func() error { + for job := range writeCh { + err := job.s.writeSST(job.pairs) + sl.flushSem.Release(1) + if err != nil { + return err + } + if onProgress != nil { + onProgress(int(completed.Add(1)), total) + } + } + return nil + }) + } + + return g.Wait() } diff --git a/pkg/snapshot/shard_test.go b/pkg/snapshot/shard_test.go new file mode 100644 index 00000000..90b7e874 --- /dev/null +++ b/pkg/snapshot/shard_test.go @@ -0,0 +1,100 @@ +package snapshot + +import ( + "context" + "encoding/binary" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accountsdb" + "github.com/gagliardetto/solana-go" +) + +// TestFlushPipelineDedupAndOrder feeds a known set of key/slot entries through the +// shard logger's flush pipeline and asserts the ingested index keeps exactly one +// entry per key: the one with the highest slot. This pins the pipeline's transform +// (sort key-ASC / slot-DESC, dedup keep-first) to the same behavior as the old +// serial logToSST path. +func TestFlushPipelineDedupAndOrder(t *testing.T) { + logsDir := t.TempDir() + const numShards = 8 + sl := NewShardLogger(numShards, logsDir) + + // Build a key -> expected (highest-slot) entry map while enqueuing several + // entries per key at differing slots, deliberately NOT in slot order. + const numKeys = 5000 + expected := make(map[solana.PublicKey]accountsdb.AccountIndexEntry, numKeys) + makeKey := func(i int) solana.PublicKey { + var k solana.PublicKey + if i%50 == 0 { + // Force a shared 8-byte prefix but distinct later bytes, so the sort + // comparator must fall through to the full key, not just its first word. + binary.BigEndian.PutUint64(k[:8], 0xABCDEF0011223344) + binary.BigEndian.PutUint64(k[8:16], uint64(i)) + } else { + binary.BigEndian.PutUint64(k[:8], uint64(i)*0x9E3779B97F4A7C15) + binary.BigEndian.PutUint32(k[8:12], uint32(i)) + } + return k + } + for i := 0; i < numKeys; i++ { + k := makeKey(i) + // Three writes for this key at slots that peak in the middle, so the + // winner is neither the first nor the last enqueued. + slots := []uint64{uint64(i) + 10, uint64(i) + 100, uint64(i) + 50} + var best accountsdb.AccountIndexEntry + for j, slot := range slots { + e := accountsdb.AccountIndexEntry{Slot: slot, FileId: uint64(i), Offset: uint64(j)} + sl.EnqueueRequest(k, e) + if slot > best.Slot { + best = e + } + } + expected[k] = best + } + + if err := sl.CloseWithProgress(context.Background(), nil); err != nil { + t.Fatalf("CloseWithProgress: %v", err) + } + + indexDir := t.TempDir() + db, err := ingestSSTFiles(indexDir, logsDir) + if err != nil { + t.Fatalf("ingestSSTFiles: %v", err) + } + defer db.Close() + + iter, err := db.NewIter(nil) + if err != nil { + t.Fatalf("NewIter: %v", err) + } + defer iter.Close() + + var count int + var prevKey []byte + for iter.First(); iter.Valid(); iter.Next() { + key := append([]byte(nil), iter.Key()...) + if prevKey != nil && string(key) <= string(prevKey) { + t.Fatalf("keys not strictly ascending / deduped: %x then %x", prevKey, key) + } + prevKey = key + + got, err := accountsdb.UnmarshalAcctIdxEntry(iter.Value()) + if err != nil { + t.Fatalf("UnmarshalAcctIdxEntry: %v", err) + } + want, ok := expected[solana.PublicKey(key)] + if !ok { + t.Fatalf("unexpected key in index: %x", key) + } + if *got != want { + t.Fatalf("key %x: got %+v, want %+v", key, *got, want) + } + count++ + } + if err := iter.Error(); err != nil { + t.Fatalf("iter error: %v", err) + } + if count != len(expected) { + t.Fatalf("index has %d keys, want %d", count, len(expected)) + } +} diff --git a/pkg/snapshot/shard_writer.go b/pkg/snapshot/shard_writer.go new file mode 100644 index 00000000..a158dd0d --- /dev/null +++ b/pkg/snapshot/shard_writer.go @@ -0,0 +1,278 @@ +package snapshot + +import ( + "encoding/binary" + "math" + "os" + "path/filepath" + "sync" + "sync/atomic" + "syscall" + "unsafe" + + "github.com/Overclock-Validator/mithril/pkg/mlog" +) + +const ( + dioAlign = 4096 + shardBufSize = 8 << 20 // staging buffer flushed to disk as one write + shardBufCount = 4 // buffers per writer: 1 filling + up to 3 draining +) + +func alignUp(n int) int { return (n + dioAlign - 1) &^ (dioAlign - 1) } + +// alignedBuf returns a slice of length size whose backing array starts on a +// dioAlign boundary, as required for O_DIRECT. +func alignedBuf(size int) []byte { + b := make([]byte, size+dioAlign) + if off := int(uintptr(unsafe.Pointer(&b[0])) % dioAlign); off != 0 { + b = b[dioAlign-off:] + } + return b[:size] +} + +// writeShardMetadata records the shard count the DB was built with. Its presence +// also marks a complete build; OpenDb refuses to open a DB without it. +func writeShardMetadata(metaDir string, numShards int) error { + var buf [8]byte + binary.LittleEndian.PutUint64(buf[:], uint64(numShards)) + return os.WriteFile(filepath.Join(metaDir, "num_shards"), buf[:], 0644) +} + +type shardWriter struct { + f *os.File + direct bool + + mu sync.Mutex // guards cur/buflen/offset and enqueue ordering + cur []byte + buflen int + offset uint64 + + free chan []byte + queue chan wItem + wg sync.WaitGroup + queued atomic.Int64 // bytes accepted but not yet written (backpressure signal) + + errMu sync.Mutex + err error +} + +type wItem struct { + b []byte + n int + pooled bool // return b to the free pool after writing +} + +func (s *shardWriter) newBuf() []byte { + if s.direct { + return alignedBuf(shardBufSize) + } + return make([]byte, shardBufSize) +} + +func newShardWriter(path string, direct bool) (*shardWriter, error) { + flags := os.O_CREATE | os.O_WRONLY | os.O_TRUNC + if direct { + flags |= syscall.O_DIRECT + } + f, err := os.OpenFile(path, flags, 0644) + if err != nil { + return nil, err + } + s := &shardWriter{ + f: f, + direct: direct, + free: make(chan []byte, shardBufCount), + queue: make(chan wItem, shardBufCount+4), + } + s.cur = s.newBuf() + for i := 0; i < shardBufCount-1; i++ { + s.free <- s.newBuf() + } + s.wg.Add(1) + go s.writeLoop() + return s, nil +} + +func (s *shardWriter) writeLoop() { + defer s.wg.Done() + for it := range s.queue { + if it.n > 0 && s.loadErr() == nil { + if _, err := s.f.Write(it.b[:it.n]); err != nil { + s.storeErr(err) + } + } + s.queued.Add(int64(-it.n)) + if it.pooled { + s.free <- it.b + } + } +} + +// append copies blob into the staging buffer and returns the base offset where its +// bytes will land in the file. Safe for concurrent callers; it serializes internally. +func (s *shardWriter) append(blob []byte) (uint64, error) { + n := len(blob) + padded := n + if s.direct { + padded = alignUp(n) + } + + s.mu.Lock() + base := s.offset + + // len(s.cur), not cap: alignedBuf over-allocates, so cap can exceed the logical + // buffer size and let an oversized record slip into the in-buffer path. + if padded > len(s.cur) { + // Record larger than a staging buffer: flush current, write it standalone. + // blob is the caller's pooled tar buffer, which the index builder recycles as + // soon as it finishes parsing — before this async write is guaranteed to run. + // So we MUST queue a private copy; aliasing blob let the buffer be reused and + // overwritten before the writer flushed it, corrupting every account in this + // append-vec (~wrong pubkey on read-back). + s.flushLocked() + var b []byte + if s.direct { + b = alignedBuf(padded) + copy(b, blob) + clear(b[n:]) + } else { + b = make([]byte, n) + copy(b, blob) + } + s.offset += uint64(padded) + s.queued.Add(int64(padded)) + s.queue <- wItem{b: b, n: padded, pooled: false} + s.mu.Unlock() + return base, s.loadErr() + } + + if s.buflen+padded > len(s.cur) { + s.flushLocked() + } + copy(s.cur[s.buflen:], blob) + if s.direct { + clear(s.cur[s.buflen+n : s.buflen+padded]) + } + s.buflen += padded + s.offset += uint64(padded) + s.queued.Add(int64(padded)) + s.mu.Unlock() + return base, s.loadErr() +} + +// flushLocked hands the current buffer to the writer goroutine and swaps in a fresh +// one. Called with s.mu held; blocks (backpressure) when no free buffer is available. +func (s *shardWriter) flushLocked() { + if s.buflen == 0 { + return + } + s.queue <- wItem{b: s.cur, n: s.buflen, pooled: true} + s.cur = <-s.free + s.buflen = 0 +} + +func (s *shardWriter) close() error { + s.mu.Lock() + s.flushLocked() + s.mu.Unlock() + close(s.queue) + s.wg.Wait() + cerr := s.f.Close() + if werr := s.loadErr(); werr != nil { + return werr + } + return cerr +} + +func (s *shardWriter) loadErr() error { + s.errMu.Lock() + defer s.errMu.Unlock() + return s.err +} + +func (s *shardWriter) storeErr(err error) { + s.errMu.Lock() + if s.err == nil { + s.err = err + } + s.errMu.Unlock() +} + +// shardBigFiles owns one coalesced big file ("data") per shard and picks a shard +// per record by fewest in-flight (queued) bytes, so records flow to whichever disk +// is draining fastest. +type shardBigFiles struct { + data []*shardWriter + written []atomic.Int64 +} + +func openShardBigFiles(shardDirs []string) (*shardBigFiles, error) { + direct := SnapshotDirectIO + mlog.Log.Infof("snapshot shard writers: %d shard(s), O_DIRECT=%v", len(shardDirs), direct) + sb := &shardBigFiles{ + data: make([]*shardWriter, len(shardDirs)), + written: make([]atomic.Int64, len(shardDirs)), + } + for i, dir := range shardDirs { + var err error + if sb.data[i], err = newShardWriter(filepath.Join(dir, "data"), direct); err != nil { + sb.close() + return nil, err + } + } + return sb, nil +} + +func (sb *shardBigFiles) n() uint64 { return uint64(len(sb.data)) } + +func (sb *shardBigFiles) choose() int { + // Prefer the disk with the least queued bytes + best, bestQ := 0, int64(math.MaxInt64) + anyQueued := false + for i := range sb.data { + q := sb.data[i].queued.Load() + if q > 0 { + anyQueued = true + } + if q < bestQ { + best, bestQ = i, q + } + } + if anyQueued { + return best + } + // No queued bytes so spread evenly by cumulative bytes instead. + best, bestW := 0, int64(math.MaxInt64) + for i := range sb.data { + if w := sb.written[i].Load(); w < bestW { + best, bestW = i, w + } + } + return best +} + +// write appends blob to the chosen shard's big file and returns the shard-encoded +// fileId (segment 0) and the offset within that file. +func (sb *shardBigFiles) write(blob []byte) (fileId, base uint64, err error) { + shard := sb.choose() + sb.written[shard].Add(int64(len(blob))) + base, err = sb.data[shard].append(blob) + if err != nil { + return 0, 0, err + } + return uint64(shard), base, nil +} + +func (sb *shardBigFiles) close() error { + var firstErr error + for _, w := range sb.data { + if w == nil { + continue + } + if err := w.close(); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} diff --git a/pkg/snapshot/shard_writer_test.go b/pkg/snapshot/shard_writer_test.go new file mode 100644 index 00000000..5e6cd2d1 --- /dev/null +++ b/pkg/snapshot/shard_writer_test.go @@ -0,0 +1,154 @@ +package snapshot + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + "testing" +) + +// TestShardWriterReadback verifies that bytes appended to a shardWriter can be read +// back at the base offset it returned, for both buffered and O_DIRECT modes and +// including a record larger than the staging buffer. Runs on an ext4 mount because +// O_DIRECT is not supported on tmpfs. +func TestShardWriterReadback(t *testing.T) { + dir := "/mnt/disk0" + if _, err := os.Stat(dir); err != nil { + t.Skipf("shard writer test needs an ext4 mount at %s: %v", dir, err) + } + + for _, direct := range []bool{false, true} { + t.Run(fmt.Sprintf("direct=%v", direct), func(t *testing.T) { + path := filepath.Join(dir, fmt.Sprintf("shardwriter_test_%v", direct)) + defer os.Remove(path) + + w, err := newShardWriter(path, direct) + if err != nil { + if direct { + t.Skipf("O_DIRECT unsupported here: %v", err) + } + t.Fatal(err) + } + + // varied sizes: tiny, unaligned, and one larger than the staging buffer + sizes := []int{1, 100, 4095, 4096, 4097, 1 << 20, shardBufSize + 1234, 7, 500000} + type rec struct { + base uint64 + data []byte + } + var recs []rec + for i, sz := range sizes { + data := make([]byte, sz) + for j := range data { + data[j] = byte((i*31 + j) & 0xff) + } + base, err := w.append(data) + if err != nil { + t.Fatalf("append %d: %v", i, err) + } + recs = append(recs, rec{base, data}) + } + if err := w.close(); err != nil { + t.Fatalf("close: %v", err) + } + + f, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + defer f.Close() + for i, r := range recs { + got := make([]byte, len(r.data)) + if _, err := f.ReadAt(got, int64(r.base)); err != nil { + t.Fatalf("readAt rec %d (base=%d len=%d): %v", i, r.base, len(r.data), err) + } + if !bytes.Equal(got, r.data) { + t.Fatalf("rec %d mismatch at base=%d", i, r.base) + } + } + }) + } +} + +// TestShardBigFilesPlacement drives shardBigFiles across three real ext4 disks in +// both buffered and O_DIRECT modes. It asserts (1) placement fans records across +// more than one shard (queued steering plus the even-split fallback never collapse +// onto shard 0) and (2) every record reads back at the path+offset its returned +// fileId/base resolve to. +func TestShardBigFilesPlacement(t *testing.T) { + dirs := []string{"/mnt/disk0", "/mnt/disk1", "/mnt/disk2"} + shardDirs := make([]string, len(dirs)) + for i, d := range dirs { + shardDirs[i] = filepath.Join(d, "shardplace_test") + if err := os.MkdirAll(shardDirs[i], 0755); err != nil { + t.Skipf("need writable ext4 dirs (%s): %v", shardDirs[i], err) + } + defer os.RemoveAll(shardDirs[i]) + } + + for _, direct := range []bool{false, true} { + t.Run(fmt.Sprintf("direct=%v", direct), func(t *testing.T) { + old := SnapshotDirectIO + SnapshotDirectIO = direct + defer func() { SnapshotDirectIO = old }() + + sb, err := openShardBigFiles(shardDirs) + if err != nil { + if direct { + t.Skipf("O_DIRECT unsupported here: %v", err) + } + t.Fatal(err) + } + + type rec struct { + fileId, base uint64 + data []byte + } + var recs []rec + shardHits := map[uint64]int{} + // Mix of sizes, including several larger than the staging buffer so the + // writer actually issues writes (and updates queued backlog) mid-run. + sizes := []int{1000, 40000, 500000, shardBufSize + 777, 2 << 20} + for i := 0; i < 120; i++ { + sz := sizes[i%len(sizes)] + data := make([]byte, sz) + for j := range data { + data[j] = byte((i*7 + j) & 0xff) + } + fileId, base, err := sb.write(data) + if err != nil { + t.Fatalf("write %d: %v", i, err) + } + recs = append(recs, rec{fileId, base, data}) + shardHits[fileId%sb.n()]++ + } + if err := sb.close(); err != nil { + t.Fatalf("close: %v", err) + } + + if len(shardHits) < 2 { + t.Fatalf("placement collapsed onto %d shard(s): %v", len(shardHits), shardHits) + } + t.Logf("direct=%v fan-out by shard: %v", direct, shardHits) + + n := sb.n() + for i, r := range recs { + p := filepath.Join(shardDirs[r.fileId%n], "data") // segment 0 = coalesced big file + f, err := os.Open(p) + if err != nil { + t.Fatalf("open %s: %v", p, err) + } + got := make([]byte, len(r.data)) + _, err = f.ReadAt(got, int64(r.base)) + f.Close() + if err != nil { + t.Fatalf("rec %d readAt %s base=%d len=%d: %v", i, p, r.base, len(r.data), err) + } + if !bytes.Equal(got, r.data) { + t.Fatalf("rec %d mismatch: shard=%d seg=%d base=%d", i, r.fileId%n, r.fileId/n, r.base) + } + } + }) + } +} diff --git a/pkg/snapshot/snapshot.go b/pkg/snapshot/snapshot.go index aff9433d..8b9b0b40 100644 --- a/pkg/snapshot/snapshot.go +++ b/pkg/snapshot/snapshot.go @@ -65,15 +65,18 @@ func UnmarshalManifestFromSnapshot(ctx context.Context, filename string, account type appendVecCopyingTask struct { Filename string - TarBuffer *bytes.Buffer + Buf []byte + BufRef *[]byte // pooled buffer backing Buf; returned after indexing FromIncrementalSnapshot bool } type indexEntryBuilderTask struct { - Data []byte - FileSize uint64 - Slot uint64 - FileId uint64 + Data []byte + FileSize uint64 + Slot uint64 + FileId uint64 + BaseOffset uint64 + BufRef *[]byte // pooled buffer backing Data; returned once parsed } type indexEntryCommitterTask struct { diff --git a/pkg/state/state.go b/pkg/state/state.go index 4ef42b9c..f49a9933 100644 --- a/pkg/state/state.go +++ b/pkg/state/state.go @@ -597,7 +597,7 @@ func ValidateAccountsDbArtifacts(accountsDbDir string) error { "mithril_db", "bankhash_db", "accounts", - "largest_file_id", + "num_shards", "bank_hash", "manifest", }