db/kv, db/state, execution: opt-in per-table page-cache warmup

Adds a per-table warmup primitive to help operate chaindata >> RAM:

- kv.RoDB.WarmupTable(ctx, table) and kv.TemporalRoDB.Warmup(ctx, force);
  mdbx impl scans the table with parallel prefix reads, gated by
  dbg.WarmupTableWorkers (env WARMUP_TABLE_WORKERS, default 0 = disabled).
  gatedRoDB.WarmupTable is a no-op: warmup's long-lived readers are exactly
  what the gate keeps off the commit path.
- domain/inverted-index collate+prune fire a fire-and-forget WarmupTable on
  the table they are about to read (skipped when the Aggregator-set db is nil,
  e.g. standalone test instances).
- backup.ClearTables and Kv2kv warm each table while clearing/copying it via a
  warmupWhile helper that cancels the warmup as soon as the operation returns.

ClearTables/ResetSenders gain a db arg; backup.ReadAheadThreads is replaced by
dbg.WarmupTableWorkers.

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)
@@ -810,6 +810,20 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 			return err
 		}
 		doms.ClearRam(true)
+		if !noCommit {
+			if err := tx.Commit(); err != nil {
+				return err
+			}
+
+			// no-async
+			if err := agg.BuildFiles(agg.EndTxNumMinimax() + agg.StepSize()); err != nil {
+				return err
+			}
+
+			if tx, err = db.BeginTemporalRw(ctx); err != nil {
+				return err
+			}
+		}
 
 		pruneStage, err := sync.PruneStageState(stages.Execution, s.BlockNumber, tx, s.CurrentSyncCycle.IsInitialCycle)
 		if err != nil {

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

@@ -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
@@ -83,14 +83,33 @@ func Kv2kv(ctx context.Context, src kv.RoDB, dst kv.RwDB, tables []string, readA
 		if b.IsDeprecated {
 			continue
 		}
-		if err := backupTable(ctx, srcTx, dst, name, logEvery, logger); err != nil {
+		if err := warmupWhile(ctx, src, name, func() error {
+			return backupTable(ctx, srcTx, dst, name, logEvery, logger)
+		}); err != nil {
 			return err
 		}
 	}
 	logger.Info("done")
 	return nil
 }
 
+// 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, srcTx kv.Tx, dst kv.RwDB, table string, logEvery *time.Ticker, logger log.Logger) error {
 	var total uint64
 	srcC, err := srcTx.Cursor(table)
@@ -149,6 +168,11 @@ func backupTable(ctx context.Context, srcTx kv.Tx, dst kv.RwDB, table string, lo
 			}
 		}
 	}
+
+	// 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 {
@@ -164,12 +188,10 @@ func backupTable(ctx context.Context, srcTx kv.Tx, dst kv.RwDB, table string, lo
 	return 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
@@ -582,6 +585,10 @@ type TemporalRoDB interface {
 	ViewTemporal(ctx context.Context, f func(tx TemporalTx) error) error
 	BeginTemporalRo(ctx context.Context) (TemporalTx, error)
 	Debug() TemporalDebugDB
+
+	// Warmup pulls the whole DB into RAM (mdbx env warmup). force re-touches
+	// pages already resident; non-force only prefetches missing ones.
+	Warmup(ctx context.Context, force bool) error
 }
 type TemporalRwDB interface {
 	RwDB

db/kv/mdbx/kv_mdbx.go

@@ -36,6 +36,7 @@ import (
 	"github.com/c2h5oh/datasize"
 	"github.com/erigontech/mdbx-go/mdbx"
 	stack2 "github.com/go-stack/stack"
+	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/semaphore"
 
 	"github.com/erigontech/erigon/common/dbg"
@@ -882,6 +883,117 @@ func (db *MdbxKV) View(ctx context.Context, f func(tx kv.Tx) error) (err error)
 	return f(tx)
 }
 
+func (db *MdbxKV) WarmupTable(ctx context.Context, bucket string) {
+	workers := int(dbg.WarmupTableWorkers)
+	if workers <= 0 {
+		return
+	}
+	const lvl = log.LvlInfo
+	var total uint64
+	if err := db.View(ctx, func(tx kv.Tx) error {
+		var err error
+		total, err = tx.Count(bucket)
+		return err
+	}); err != nil {
+		db.log.Warn("[warmup] count failed", "table", bucket, "err", err)
+	}
+	if total == 0 {
+		return
+	}
+	progress := atomic.Int64{}
+
+	logEvery := time.NewTicker(20 * time.Second)
+	defer logEvery.Stop()
+
+	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(workers)
+	for i := 0; i < 256; i++ {
+		for j := 0; j < 256; j++ {
+			i := i
+			j := j
+			g.Go(func() error {
+				return db.View(ctx, func(tx kv.Tx) error {
+					it, err := tx.Prefix(bucket, []byte{byte(i), byte(j)})
+					if err != nil {
+						return err
+					}
+					defer it.Close()
+					kNum := 0
+					for it.HasNext() {
+						k, v, err := it.Next()
+						if err != nil {
+							return err
+						}
+						if len(k) > 0 {
+							_, _ = k[0], k[len(k)-1]
+						}
+						if len(v) > 0 {
+							_, _ = v[0], v[len(v)-1]
+						}
+						progress.Add(1)
+
+						kNum++
+						if kNum%1024 == 0 { // a bit reduce runtime cost
+							select {
+							case <-ctx.Done():
+								return ctx.Err()
+							case <-logEvery.C:
+								db.log.Log(lvl, fmt.Sprintf("[warmup] Progress: %s %.2f%%", bucket, min(100, 100*float64(progress.Load())/float64(total))))
+							default:
+							}
+						}
+					}
+					return nil
+				})
+			})
+		}
+	}
+	for i := 0; i < 1_000; i++ {
+		i := i
+		g.Go(func() error {
+			return db.View(ctx, func(tx kv.Tx) error {
+				seek := make([]byte, 8)
+				binary.BigEndian.PutUint64(seek, uint64(i*100_000))
+				it, err := tx.Prefix(bucket, seek)
+				if err != nil {
+					return err
+				}
+				defer it.Close()
+				kNum := 0
+				for it.HasNext() {
+					k, v, err := it.Next()
+					if err != nil {
+						return err
+					}
+					if len(k) > 0 {
+						_, _ = k[0], k[len(k)-1]
+					}
+					if len(v) > 0 {
+						_, _ = v[0], v[len(v)-1]
+					}
+					progress.Add(1)
+
+					kNum++
+					if kNum%1024 == 0 {
+						select {
+						case <-ctx.Done():
+							return ctx.Err()
+						case <-logEvery.C:
+							db.log.Log(lvl, fmt.Sprintf("[warmup] Progress: %s %.2f%%", bucket, min(100, 100*float64(progress.Load())/float64(total))))
+						default:
+						}
+					}
+				}
+
+				return nil
+			})
+		})
+	}
+	if err := g.Wait(); err != nil && !errors.Is(err, context.Canceled) {
+		db.log.Warn("[warmup] failed", "table", bucket, "err", err)
+	}
+}
+
 func (tx *MdbxTx) Apply(_ context.Context, f func(tx kv.Tx) error) (err error) {
 	return f(tx)
 }

db/kv/membatchwithdb/memory_mutation.go

@@ -1272,4 +1272,8 @@ func (td temporaldb) ViewTemporal(ctx context.Context, f func(tx kv.TemporalTx)
 	return f(td.memoryMutation)
 }
 
+func (td temporaldb) Warmup(ctx context.Context, force bool) error { return nil }
+
+func (td temporaldb) WarmupTable(ctx context.Context, table string) {}
+
 func (td temporaldb) Path() string { return "<mem>" }

db/kv/membatchwithdb/memory_store.go

@@ -431,12 +431,13 @@ func (db *memStoreDB) UpdateNosync(_ context.Context, f func(tx kv.RwTx) error)
 func (db *memStoreDB) View(_ context.Context, f func(tx kv.Tx) error) error {
 	return f(db.store)
 }
-func (db *memStoreDB) Close()                      {}
-func (db *memStoreDB) AllTables() kv.TableCfg      { return kv.ChaindataTablesCfg }
-func (db *memStoreDB) PageSize() datasize.ByteSize { return 4096 }
-func (db *memStoreDB) ReadOnly() bool              { return false }
-func (db *memStoreDB) CHandle() unsafe.Pointer     { return nil }
-func (db *memStoreDB) Path() string                { return "<mem>" }
+func (db *memStoreDB) Close()                                        {}
+func (db *memStoreDB) AllTables() kv.TableCfg                        { return kv.ChaindataTablesCfg }
+func (db *memStoreDB) PageSize() datasize.ByteSize                   { return 4096 }
+func (db *memStoreDB) WarmupTable(ctx context.Context, table string) {}
+func (db *memStoreDB) ReadOnly() bool                                { return false }
+func (db *memStoreDB) CHandle() unsafe.Pointer                       { return nil }
+func (db *memStoreDB) Path() string                                  { return "<mem>" }
 
 // --- Cursor ---
 //