db/kv, db/state, execution: opt-in page-cache warmup + parallel read-ahead

Author: AskAlexSharov

cmd/integration/commands/refetence_db.go

@@ -104,7 +104,7 @@ var cmdMdbxToMdbx = &cobra.Command{
 		ctx, _ := common.RootContext()
 		logger := debug.SetupCobra(cmd, "integration")
 		from, to := backup.OpenPair(chaindata, toChaindata, dbcfg.ChainDB, 0, logger)
-		err := backup.Kv2kv(ctx, from, to, nil, backup.ReadAheadThreads, logger)
+		err := backup.Kv2kv(ctx, from, to, nil, logger)
 		if err != nil && !errors.Is(err, context.Canceled) {
 			if !errors.Is(err, context.Canceled) {
 				logger.Error(err.Error())

cmd/integration/commands/reset_state.go

@@ -106,7 +106,7 @@ var cmdClearBadBlocks = &cobra.Command{
 		defer db.Close()
 
 		return db.Update(ctx, func(tx kv.RwTx) error {
-			return backup.ClearTables(ctx, tx, kv.BadHeaderNumber)
+			return backup.ClearTables(ctx, db, tx, kv.BadHeaderNumber)
 		})
 	},
 }

cmd/integration/commands/stages.go

@@ -546,7 +546,7 @@ func stageSenders(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) er
 
 	br, _ := blocksIO(db, logger)
 	if reset {
-		return db.Update(ctx, func(tx kv.RwTx) error { return rawdbreset.ResetSenders(ctx, tx) })
+		return db.Update(ctx, func(tx kv.RwTx) error { return rawdbreset.ResetSenders(ctx, db, tx) })
 	}
 
 	tx, err := db.BeginRw(ctx)
@@ -710,7 +710,7 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 		if err != nil {
 			return err
 		}
-		err = stagedsync.PruneExecutionStage(ctx, p, tx, cfg, 0, logger)
+		_, err = stagedsync.PruneExecutionStage(ctx, p, tx, cfg, 0, logger)
 		if err != nil {
 			return err
 		}
@@ -758,10 +758,13 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 					return err
 				}
 			}
-			if err := doms.Commit(ctx, tx); err != nil {
+			if err := doms.Flush(ctx, tx); err != nil {
+				return err
+			}
+			doms.ClearRam(true)
+			if err := tx.Commit(); err != nil {
 				return err
 			}
-			doms.Close()
 			if tx, err = db.BeginTemporalRw(ctx); err != nil {
 				return err
 			}
@@ -770,27 +773,22 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 			if err != nil {
 				return err
 			}
-			if err := stagedsync.PruneExecutionStage(ctx, pruneStage, tx, cfg, 0, logger); err != nil {
+			if _, err := stagedsync.PruneExecutionStage(ctx, pruneStage, tx, cfg, 0, logger); err != nil {
 				return err
 			}
 
 			if err := tx.Commit(); err != nil {
 				return err
 			}
-			if bn+1 >= block { // last block committed; nothing more to run
-				break
-			}
 			if tx, err = db.BeginTemporalRw(ctx); err != nil {
 				return err
 			}
-			// Fresh SD for the next block: a committed SD is never reused.
-			if doms, err = execctx.NewSharedDomains(ctx, tx, logger); err != nil {
-				return err
-			}
-			doms.SetInMemHistoryReads(false)
 		}
-		doms.Close()
-		return nil
+		if err := doms.Flush(ctx, tx); err != nil {
+			return err
+		}
+		doms.ClearRam(true)
+		return tx.Commit()
 	}
 	agg := (db.(dbstate.HasAgg).Agg()).(*dbstate.Aggregator)
 	blockSnapBuildSema := semaphore.NewWeighted(int64(runtime.NumCPU()))
@@ -809,25 +807,44 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 		if err := doms.Flush(ctx, tx); err != nil {
 			return err
 		}
+		txNum := doms.TxNum()
 		doms.ClearRam(true)
-
-		pruneStage, err := sync.PruneStageState(stages.Execution, s.BlockNumber, tx, s.CurrentSyncCycle.IsInitialCycle)
-		if err != nil {
-			return err
-		}
-		if err := stagedsync.PruneExecutionStage(ctx, pruneStage, tx, cfg, 0, logger); err != nil {
-			return err
-		}
-
 		if !noCommit {
 			if err := tx.Commit(); err != nil {
 				return err
 			}
+
+			agg.BuildFilesInBackground(txNum)
+
 			if tx, err = db.BeginTemporalRw(ctx); err != nil {
 				return err
 			}
 		}
 
+		// Prune in bounded, committed batches so a long prune survives Ctrl-C:
+		// prune progress is durable, so a restart only repeats the last batch.
+		for {
+			pruneStage, err := sync.PruneStageState(stages.Execution, s.BlockNumber, tx, false)
+			if err != nil {
+				return err
+			}
+			hasMore, err := stagedsync.PruneExecutionStage(ctx, pruneStage, tx, cfg, time.Minute, logger)
+			if err != nil {
+				return err
+			}
+			if !noCommit {
+				if err := tx.Commit(); err != nil {
+					return err
+				}
+				if tx, err = db.BeginTemporalRw(ctx); err != nil {
+					return err
+				}
+			}
+			if !hasMore {
+				break
+			}
+		}
+
 		if execProgress, err = stages.GetStageProgress(tx, stages.Execution); err != nil {
 			return err
 		}

common/dbg/experiments.go

@@ -38,6 +38,10 @@ import (
 var (
 	MaxReorgDepth = EnvUint("MAX_REORG_DEPTH", 96)
 
+	// WarmupTableWorkers bounds the parallel scans of a single WarmupTable call;
+	// 0 (the default) disables table warmup entirely.
+	WarmupTableWorkers = EnvUint("WARMUP_TABLE_WORKERS", 0)
+
 	saveHeapProfile             = EnvBool("SAVE_HEAP_PROFILE", false)
 	heapProfileFilePath         = EnvString("HEAP_PROFILE_FILE_PATH", "")
 	heapProfileThresholdPercent = EnvUint("HEAP_PROFILE_THRESHOLD", 35)

db/kv/backup/backup.go

@@ -18,10 +18,10 @@ package backup
 
 import (
 	"context"
-	"encoding/hex"
 	"fmt"
 	"maps"
 	"runtime"
+	"slices"
 	"time"
 
 	"github.com/c2h5oh/datasize"
@@ -58,7 +58,7 @@ func OpenPair(from, to string, label kv.Label, targetPageSize datasize.ByteSize,
 	return src, dst
 }
 
-func Kv2kv(ctx context.Context, src kv.RoDB, dst kv.RwDB, tables []string, readAheadThreads int, logger log.Logger) error {
+func Kv2kv(ctx context.Context, src kv.RoDB, dst kv.RwDB, tables []string, logger log.Logger) error {
 	srcTx, err1 := src.BeginRo(ctx)
 	if err1 != nil {
 		return err1
@@ -79,76 +79,122 @@ func Kv2kv(ctx context.Context, src kv.RoDB, dst kv.RwDB, tables []string, readA
 		}
 	}
 
-	for name, b := range tablesMap {
-		if b.IsDeprecated {
+	var copiedTables int
+	var copiedRows uint64
+	for _, name := range slices.Sorted(maps.Keys(tablesMap)) { // deterministic order for reproducible benchmarks
+		if tablesMap[name].IsDeprecated {
 			continue
 		}
-		if err := backupTable(ctx, srcTx, dst, name, logEvery, logger); err != nil {
+		rows, err := backupTable(ctx, src, srcTx, dst, name, logEvery, logger)
+		if err != nil {
 			return err
 		}
+		if rows > 0 {
+			copiedTables++
+			copiedRows += rows
+		}
 	}
-	logger.Info("done")
+	logger.Info("done", "tablesWithData", copiedTables, "rows", common.PrettyCounter(copiedRows))
 	return nil
 }
 
-func backupTable(ctx context.Context, srcTx kv.Tx, dst kv.RwDB, table string, logEvery *time.Ticker, logger log.Logger) error {
-	var total uint64
+// warmupWhile warms `table` into RAM in the background while `fn` runs, and
+// cancels the warmup as soon as `fn` returns — bounds warmup to one table at a
+// time and keeps fn's reads off disk on chaindata >> RAM.
+func warmupWhile(ctx context.Context, db kv.RoDB, table string, fn func() error) error {
+	warmupCtx, cancel := context.WithCancel(ctx)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		db.WarmupTable(warmupCtx, table)
+	}()
+	defer func() {
+		cancel()
+		<-done
+	}()
+	return fn()
+}
+
+func backupTable(ctx context.Context, src kv.RoDB, srcTx kv.Tx, dst kv.RwDB, table string, logEvery *time.Ticker, logger log.Logger) (uint64, error) {
+	t := time.Now()
 	srcC, err := srcTx.Cursor(table)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer srcC.Close()
-	total, _ = srcTx.Count(table)
+	total, _ := srcTx.Count(table)
+	size, _ := srcTx.BucketSize(table)
+	if total > 0 {
+		logger.Info("[mdbx_to_mdbx] copying", "table", table, "rows", common.PrettyCounter(total), "size", common.ByteCount(size))
+	}
+
+	// Parallel read-ahead: keep a bounded band of pages warm just ahead of the
+	// copy cursor (cold page faults are slow; one reader can't saturate nvme).
+	// No-op unless WARMUP_TABLE_WORKERS is set.
+	var ra *kv.ReadAheader
+	if workers := int(dbg.WarmupTableWorkers); workers > 0 && total > 0 {
+		ra = kv.NewReadAheader(ctx, src, table, nil, workers)
+	}
+	defer ra.Close()
 
 	if err := dst.Update(ctx, func(tx kv.RwTx) error {
 		return tx.ClearTable(table)
 	}); err != nil {
-		return err
+		return 0, err
 	}
 	dstTx, err := dst.BeginRw(ctx)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer dstTx.Rollback()
 
 	c, err := dstTx.RwCursor(table)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer c.Close()
 	casted, isDupsort := c.(kv.RwCursorDupSort)
 	i := uint64(0)
 
 	for k, v, err := srcC.First(); k != nil; k, v, err = srcC.Next() {
 		if err != nil {
-			return err
+			return 0, err
+		}
+		if i%1024 == 0 {
+			ra.SetPos(k)
 		}
 
 		if isDupsort {
 			if err = casted.AppendDup(k, v); err != nil {
-				return err
+				return 0, err
 			}
 		} else {
 			if err = c.Append(k, v); err != nil {
-				return err
+				return 0, err
 			}
 		}
 
 		i++
 		if i%100_000 == 0 {
 			select {
 			case <-ctx.Done():
-				return ctx.Err()
+				return 0, ctx.Err()
 			case <-logEvery.C:
 				var m runtime.MemStats
 				dbg.ReadMemStats(&m)
 				logger.Info("Progress", "table", table, "progress",
-					fmt.Sprintf("%s/%s", common.PrettyCounter(i), common.PrettyCounter(total)), "key", hex.EncodeToString(k),
+					fmt.Sprintf("%s/%s", common.PrettyCounter(i), common.PrettyCounter(total)),
+					"size", common.ByteCount(size), "keys/s", uint64(float64(i)/time.Since(t).Seconds()),
 					"alloc", common.ByteCount(m.Alloc), "sys", common.ByteCount(m.Sys))
 			default:
 			}
 		}
 	}
+
+	// TODO: Unwind doesn't need to decrement the auto-increment sequence — it's
+	// not exposed to users and not part of consensus, so we could switch to
+	// mdbx's native Sequence.
+
 	// migrate bucket sequences to native mdbx implementation
 	//currentID, err := srcTx.Sequence(name, 0)
 	//if err != nil {
@@ -159,17 +205,17 @@ func backupTable(ctx context.Context, srcTx kv.Tx, dst kv.RwDB, table string, lo
 	//	return err
 	//}
 	if err2 := dstTx.Commit(); err2 != nil {
-		return err2
+		return 0, err2
 	}
-	return nil
+	return i, nil
 }
 
-const ReadAheadThreads = 2048
-
-func ClearTables(ctx context.Context, tx kv.RwTx, tables ...string) error {
+func ClearTables(ctx context.Context, db kv.RoDB, tx kv.RwTx, tables ...string) error {
 	for _, tbl := range tables {
 		log.Info("Clear", "table", tbl)
-		if err := tx.ClearTable(tbl); err != nil {
+		if err := warmupWhile(ctx, db, tbl, func() error {
+			return tx.ClearTable(tbl)
+		}); err != nil {
 			return err
 		}
 	}

db/kv/helpers.go

@@ -62,6 +62,10 @@ func (g *gatedRoDB) AllTables() TableCfg                     { return g.inner.Al
 func (g *gatedRoDB) PageSize() datasize.ByteSize             { return g.inner.PageSize() }
 func (g *gatedRoDB) CHandle() unsafe.Pointer                 { return g.inner.CHandle() }
 func (g *gatedRoDB) Path() string                            { return g.inner.Path() }
+
+// WarmupTable is a no-op: warmup spawns long-lived read scans, exactly the
+// freelist-pinning readers the gate exists to keep off the commit path.
+func (g *gatedRoDB) WarmupTable(ctx context.Context, table string) {}
 func (g *gatedRoDB) View(ctx context.Context, f func(tx Tx) error) error {
 	g.gate.RLock()
 	defer g.gate.RUnlock()

db/kv/kv_interface.go

@@ -122,6 +122,9 @@ type RoDB interface {
 	AllTables() TableCfg
 	PageSize() datasize.ByteSize
 
+	// WarmupTable pulls a single table into the OS page-cache via parallel scans.
+	WarmupTable(ctx context.Context, table string)
+
 	// CHandle pointer to the underlying C environment handle, if applicable (e.g. *C.MDBX_env)
 	CHandle() unsafe.Pointer
 	Path() string