diff --git a/cmd/integration/commands/refetence_db.go b/cmd/integration/commands/refetence_db.go index 07664db6130..ef1691c6ff6 100644 --- a/cmd/integration/commands/refetence_db.go +++ b/cmd/integration/commands/refetence_db.go @@ -104,7 +104,7 @@ var cmdMdbxToMdbx = &cobra.Command{ ctx := cmd.Context() 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()) diff --git a/cmd/integration/commands/reset_state.go b/cmd/integration/commands/reset_state.go index c7ac2983a7f..13cc22123fa 100644 --- a/cmd/integration/commands/reset_state.go +++ b/cmd/integration/commands/reset_state.go @@ -105,7 +105,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) }) }, } diff --git a/cmd/integration/commands/stages.go b/cmd/integration/commands/stages.go index becdd0ae762..f14317e4f5f 100644 --- a/cmd/integration/commands/stages.go +++ b/cmd/integration/commands/stages.go @@ -110,6 +110,7 @@ func makeStageCmd(use string, stageFn func(kv.TemporalRwDB, context.Context, log } ctx, cancel := context.WithCancel(cmd.Context()) defer cancel() + go kv.CollectTableSizesPeriodically(ctx, db, dbcfg.ChainDB, logger) if err := stageFn(db, ctx, logger); err != nil { if errors.Is(err, context.Canceled) { return nil // graceful shutdown @@ -376,7 +377,7 @@ func stageSnapshots(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) } } dirs := datadir.New(datadirCli) - if err := rawdbreset.ResetBlocks(tx, db, br, bw, dirs, logger); err != nil { + if err := rawdbreset.ResetBlocks(db, tx, br, bw, dirs, logger); err != nil { return fmt.Errorf("resetting blocks: %w", err) } domains, err := execctx.NewSharedDomains(ctx, tx, logger) @@ -431,7 +432,7 @@ func stageHeaders(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) er return db.Update(ctx, func(tx kv.RwTx) error { if reset { - if err := rawdbreset.ResetBlocks(tx, db, br, bw, dirs, logger); err != nil { + if err := rawdbreset.ResetBlocks(db, tx, br, bw, dirs, logger); err != nil { return err } return nil @@ -498,7 +499,7 @@ func stageBodies(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) err if reset { return db.Update(ctx, func(tx kv.RwTx) error { - return rawdbreset.ResetBlocks(tx, db, br, bw, dirs, logger) + return rawdbreset.ResetBlocks(db, tx, br, bw, dirs, logger) }) } @@ -544,7 +545,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 rawdbreset.ResetSenders(ctx, db) } tx, err := db.BeginRw(ctx) diff --git a/common/dbg/experiments.go b/common/dbg/experiments.go index dcdcd32c9d3..0a982c3141f 100644 --- a/common/dbg/experiments.go +++ b/common/dbg/experiments.go @@ -38,6 +38,8 @@ import ( var ( MaxReorgDepth = EnvUint("MAX_REORG_DEPTH", 96) + WarmupTableWorkers = EnvUint("WARMUP_TABLE_WORKERS", 0) + saveHeapProfile = EnvBool("SAVE_HEAP_PROFILE", false) heapProfileFilePath = EnvString("HEAP_PROFILE_FILE_PATH", "") heapProfileThresholdPercent = EnvUint("HEAP_PROFILE_THRESHOLD", 35) diff --git a/db/kv/backup/backup.go b/db/kv/backup/backup.go index f50949cf22c..bd0e174666e 100644 --- a/db/kv/backup/backup.go +++ b/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,15 +58,13 @@ 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 } defer srcTx.Rollback() - commitEvery := time.NewTicker(5 * time.Minute) - defer commitEvery.Stop() logEvery := time.NewTicker(20 * time.Second) defer logEvery.Stop() @@ -79,41 +77,71 @@ 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 +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, err := srcTx.Count(table) + if err != nil { + return 0, err + } + size, err := srcTx.BucketSize(table) + if err != nil { + return 0, err + } + if total > 0 { + logger.Info("[mdbx_to_mdbx] copying", "table", table, "rows", common.PrettyCounter(total), "size", common.ByteCount(size)) + } + + // Read-ahead warms pages (values too — the copy reads them) just ahead of the + // copy cursor. No-op unless WARMUP_TABLE_WORKERS is set. + var ra *kv.ReadAhead + if workers := int(dbg.WarmupTableWorkers); workers > 0 && total > 0 { + bounds, _, err := kv.DistributeBounds(srcTx, table) + if err != nil { + logger.Warn("[mdbx_to_mdbx] read-ahead disabled", "table", table, "err", err) + } else { + ra = kv.NewReadAhead(ctx, src, table, kv.ReadAheadCfg{Bounds: bounds, TableSize: size, Workers: workers, WarmValues: true}) + } + } + 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) @@ -121,57 +149,118 @@ func backupTable(ctx context.Context, srcTx kv.Tx, dst kv.RwDB, table string, lo for k, v, err := srcC.First(); k != nil; k, v, err = srcC.Next() { if err != nil { - return err + return 0, err } 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%1000 == 0 { + ra.SetPos(k) + } 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: } } } - // migrate bucket sequences to native mdbx implementation - //currentID, err := srcTx.Sequence(name, 0) - //if err != nil { - // return err - //} - //_, err = dstTx.Sequence(name, currentID) - //if err != nil { - // return err - //} + if err2 := dstTx.Commit(); err2 != nil { - return err2 + return 0, err2 + } + return i, nil +} + +// ClearTables empties each table on the caller's open write tx — atomic with the +// caller's other writes, and safe inside an open writer (a self-owned writer +// would deadlock, since MDBX serializes writers). db drives optional read-ahead only. +func ClearTables(ctx context.Context, db kv.RoDB, tx kv.RwTx, tables ...string) error { + for _, table := range tables { + if err := clearTable(ctx, db, tx, table); err != nil { + return fmt.Errorf("clearing %s: %w", table, err) + } } return nil } -const ReadAheadThreads = 2048 +func clearTable(ctx context.Context, db kv.RoDB, tx kv.RwTx, table string) error { + workers := int(dbg.WarmupTableWorkers) + if workers == 0 { // chunked range-delete only pays off paired with read-ahead + log.Info("[clear]", "table", table) + return tx.ClearTable(table) + } + + dr, ok := tx.(kv.HasDeleteRange) + if !ok { // backend has no range-delete: drop the whole table + return tx.ClearTable(table) + } -func ClearTables(ctx context.Context, tx kv.RwTx, tables ...string) error { - for _, tbl := range tables { - log.Info("Clear", "table", tbl) - if err := tx.ClearTable(tbl); err != nil { + bounds, size, err := kv.DistributeBounds(tx, table) + if err != nil { + return err + } + log.Info("[clear]", "table", table, "size", common.ByteCount(size)) + if len(bounds) <= 2 { // one chunk ([nil,nil]): native drop beats distribute+warm+whole-table DeleteRange + return tx.ClearTable(table) + } + + // read-ahead over the same boundaries; keys-only — range-delete never reads values + ra := kv.NewReadAhead(ctx, db, table, kv.ReadAheadCfg{Bounds: bounds, TableSize: size, Workers: workers}) + defer ra.Close() + + logEvery := time.NewTicker(20 * time.Second) + defer logEvery.Stop() + + lastLog := time.Now() + lastSize := size + var deleted, lastDeleted uint64 + for i := 0; i+1 < len(bounds); i++ { + ra.SetPos(bounds[i]) + t := time.Now() + n, err := dr.DeleteRange(table, bounds[i], bounds[i+1]) + if took := time.Since(t); took > 500*time.Millisecond { + log.Warn("[clear] delete", "table", table, "took", took) + } + if err != nil { return err } + deleted += n + + select { + case <-ctx.Done(): + return ctx.Err() + case <-logEvery.C: + remaining, err := tx.BucketSize(table) + if err != nil { + continue // a failed progress read shouldn't abort the clear + } + now := time.Now() + secs := now.Sub(lastLog).Seconds() + log.Info("[clear]", "table", table, + "speed", common.ByteCount(uint64(float64(lastSize-remaining)/secs))+"/s", + "keys", common.PrettyCounter(uint64(float64(deleted-lastDeleted)/secs))+"/s", + "remaining", common.ByteCount(remaining), + ) + lastLog, lastSize, lastDeleted = now, remaining, deleted + default: + } } return nil } diff --git a/db/kv/backup/backup_test.go b/db/kv/backup/backup_test.go new file mode 100644 index 00000000000..01076878213 --- /dev/null +++ b/db/kv/backup/backup_test.go @@ -0,0 +1,156 @@ +// Copyright 2024 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package backup + +import ( + "encoding/binary" + "testing" + + "github.com/c2h5oh/datasize" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common/dbg" + "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/db/kv/dbcfg" + "github.com/erigontech/erigon/db/kv/mdbx" +) + +const testTable = "T" + +func u64Key(i uint64) []byte { + k := make([]byte, 8) + binary.BigEndian.PutUint64(k, i) + return k +} + +// newWriteMapDB matches the production default (--db.writemap=true), where a +// range-delete's freed pages are recycled and rewritten in place — the setting +// under which reusing stale chunk bounds would silently skip rows. +func newWriteMapDB(t *testing.T) kv.RwDB { + t.Helper() + db := mdbx.New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WriteMap(true).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{testTable: kv.TableCfgItem{}} + }).MapSize(1 * datasize.GB).MustOpen() + t.Cleanup(db.Close) + return db +} + +func tableSize(t *testing.T, db kv.RwDB) uint64 { + t.Helper() + var sz uint64 + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + var err error + sz, err = tx.BucketSize(testTable) + return err + })) + return sz +} + +func tableCount(t *testing.T, db kv.RwDB) uint64 { + t.Helper() + var c uint64 + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + var err error + c, err = tx.Count(testTable) + return err + })) + return c +} + +func withWarmupWorkers(t *testing.T, n uint64) { + t.Helper() + prev := dbg.WarmupTableWorkers + dbg.WarmupTableWorkers = n + t.Cleanup(func() { dbg.WarmupTableWorkers = prev }) +} + +func TestClearTablesWarmupOff(t *testing.T) { + withWarmupWorkers(t, 0) // default: plain one-shot clear, no chunking + + db := newWriteMapDB(t) + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursor(testTable) + require.NoError(t, err) + defer c.Close() + for i := 0; i < 1000; i++ { + require.NoError(t, c.Append(u64Key(uint64(i)), []byte{1})) + } + return nil + })) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + return ClearTables(t.Context(), db, tx, testTable) + })) + require.Zero(t, tableCount(t, db)) +} + +// TestClearTablesWarmupOnSmallTable exercises the warmup-on path for a table +// under one chunk: it must skip distribution/read-ahead and fall back to the +// native drop, still leaving the table empty. +func TestClearTablesWarmupOnSmallTable(t *testing.T) { + withWarmupWorkers(t, 4) + + db := newWriteMapDB(t) + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursor(testTable) + require.NoError(t, err) + defer c.Close() + for i := 0; i < 1000; i++ { + require.NoError(t, c.Append(u64Key(uint64(i)), []byte{1})) + } + return nil + })) + require.Less(t, tableSize(t, db), 32*datasize.MB.Bytes(), "table must be under one chunk") + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + return ClearTables(t.Context(), db, tx, testTable) + })) + require.Zero(t, tableCount(t, db)) +} + +// TestClearTablesMultiChunkWriteMap clears a table large enough to split into +// several 32MB chunks with warmup enabled, so ClearTables walks its full chunked +// range-delete path. Every row must be gone — a stale-bounds regression would +// skip chunks and leave residual rows behind. +func TestClearTablesMultiChunkWriteMap(t *testing.T) { + withWarmupWorkers(t, 4) + + db := newWriteMapDB(t) + + // Fill past 64MB (>= 2 chunks) in batches; monotonic keys keep Append valid. + val := make([]byte, 2048) + next := uint64(0) + for tableSize(t, db) <= 64*datasize.MB.Bytes() { + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursor(testTable) + require.NoError(t, err) + defer c.Close() + for j := 0; j < 20_000; j++ { + require.NoError(t, c.Append(u64Key(next), val)) + next++ + } + return nil + })) + } + require.Greater(t, tableSize(t, db), 64*datasize.MB.Bytes(), "table must be multi-chunk") + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + return ClearTables(t.Context(), db, tx, testTable) + })) + require.Zero(t, tableCount(t, db)) +} diff --git a/db/kv/helpers.go b/db/kv/helpers.go index 3ddff49321f..23cd75c5309 100644 --- a/db/kv/helpers.go +++ b/db/kv/helpers.go @@ -179,7 +179,8 @@ func GetBool(tx Getter, bucket string, k []byte) (enabled bool, err error) { return bytes2bool(vBytes), nil } -func ReadAhead(ctx context.Context, db RoDB, progress *atomic.Bool, table string, from []byte, amount uint32) (clean func()) { +// ReadAheadDeprecated is the legacy amount-bounded prefetcher (NewReadAhead is the windowed one). +func ReadAheadDeprecated(ctx context.Context, db RoDB, progress *atomic.Bool, table string, from []byte, amount uint32) (clean func()) { if db == nil { return func() {} } @@ -203,13 +204,13 @@ func ReadAhead(ctx context.Context, db RoDB, progress *atomic.Bool, table string } defer c.Close() + var sink byte + defer func() { warmupSink.Add(uint64(sink)) }() for k, v, err := c.Seek(from); k != nil && amount > 0; k, v, err = c.Next() { if err != nil { return err } - if len(v) > 0 { - _, _ = v[0], v[len(v)-1] - } + sink = touchValue(sink, v) amount-- select { case <-ctx.Done(): diff --git a/db/kv/kv_interface.go b/db/kv/kv_interface.go index 236aec2a456..3aa7daeb5f8 100644 --- a/db/kv/kv_interface.go +++ b/db/kv/kv_interface.go @@ -618,6 +618,14 @@ type HasSpaceDirty interface { SpaceDirty() (uint64, uint64, error) } +// HasDeleteRange deletes all keys in [from, to) (to==nil deletes through the +// last key) and returns the number removed. mdbx implements it as a native bulk +// B-tree cut — far faster than per-key deletion; other backends may emulate it +// by iterating, so hot-path callers should prefer the mdbx-backed tx. +type HasDeleteRange interface { + DeleteRange(table string, from, to []byte) (uint64, error) +} + // BucketMigrator used for buckets migration, don't use it in usual app code type BucketMigrator interface { ListTables() ([]string, error) diff --git a/db/kv/mdbx/delete_range_test.go b/db/kv/mdbx/delete_range_test.go new file mode 100644 index 00000000000..927eb6038bf --- /dev/null +++ b/db/kv/mdbx/delete_range_test.go @@ -0,0 +1,222 @@ +// Copyright 2024 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package mdbx + +import ( + "bytes" + "testing" + + "github.com/c2h5oh/datasize" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/db/kv/dbcfg" +) + +const deleteRangeTable = "T" + +// newFilledDB returns a WriteMap DB (the production default: freed pages are +// recycled in place, which is what makes stale bounds dangerous) filled with +// keys 0..n-1. +func newFilledDB(t *testing.T, n int) kv.RwDB { + t.Helper() + db := New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WriteMap(true).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{deleteRangeTable: kv.TableCfgItem{}} + }).MapSize(512 * datasize.MB).MustOpen() + t.Cleanup(db.Close) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursor(deleteRangeTable) + require.NoError(t, err) + defer c.Close() + for i := 0; i < n; i++ { + require.NoError(t, c.Append(u64tob(uint64(i)), []byte{1})) + } + return nil + })) + return db +} + +func countTable(t *testing.T, db kv.RwDB) uint64 { + t.Helper() + var c uint64 + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + var err error + c, err = tx.Count(deleteRangeTable) + return err + })) + return c +} + +func TestMdbxDeleteRange(t *testing.T) { + deleteRange := func(t *testing.T, db kv.RwDB, from, to []byte) uint64 { + var n uint64 + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + var err error + n, err = tx.(kv.HasDeleteRange).DeleteRange(deleteRangeTable, from, to) + return err + })) + return n + } + + t.Run("half-open [from,to)", func(t *testing.T) { + db := newFilledDB(t, 1000) + require.EqualValues(t, 500, deleteRange(t, db, u64tob(200), u64tob(700))) + require.EqualValues(t, 500, countTable(t, db)) + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + has := func(i uint64) bool { v, _ := tx.GetOne(deleteRangeTable, u64tob(i)); return v != nil } + require.True(t, has(199)) // just below the range: kept + require.False(t, has(200)) // inclusive lower bound: gone + require.False(t, has(699)) // just below upper bound: gone + require.True(t, has(700)) // exclusive upper bound: kept + return nil + })) + }) + + t.Run("to==nil deletes through the last key", func(t *testing.T) { + db := newFilledDB(t, 1000) + require.EqualValues(t, 100, deleteRange(t, db, u64tob(900), nil)) + require.EqualValues(t, 900, countTable(t, db)) + }) + + t.Run("from==nil deletes from the first key", func(t *testing.T) { + db := newFilledDB(t, 1000) + require.EqualValues(t, 300, deleteRange(t, db, nil, u64tob(300))) + require.EqualValues(t, 700, countTable(t, db)) + }) + + t.Run("from==nil,to==nil clears the whole table", func(t *testing.T) { + db := newFilledDB(t, 1000) + require.EqualValues(t, 1000, deleteRange(t, db, nil, nil)) + require.Zero(t, countTable(t, db)) + }) + + t.Run("reversed range deletes nothing", func(t *testing.T) { + db := newFilledDB(t, 1000) + require.Zero(t, deleteRange(t, db, u64tob(700), u64tob(200))) + require.EqualValues(t, 1000, countTable(t, db)) + }) + + t.Run("from past the last key deletes nothing", func(t *testing.T) { + db := newFilledDB(t, 1000) + require.Zero(t, deleteRange(t, db, u64tob(5000), nil)) + require.EqualValues(t, 1000, countTable(t, db)) + }) +} + +// TestChunkedDeleteRangeCoversAllKeys reproduces backup.clearTable's loop: +// distribute the table into many count-balanced chunks, then range-delete each +// chunk in sequence on one write tx. The bounds must be cloned first — a same-tx +// DeleteRange rebalances the b-tree pages the raw mdbx keys point into, and +// under WriteMap those freed pages are recycled and rewritten, so reusing the +// raw bounds would silently skip chunks and leave residual rows. +func TestChunkedDeleteRangeCoversAllKeys(t *testing.T) { + const n = 200_000 + db := newFilledDB(t, n) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + bounds, err := tx.(kv.DBWithDistributionSupport).DistributeCursors(deleteRangeTable, nil, 512) + require.NoError(t, err) + require.Greater(t, len(bounds), 2, "table must split into multiple chunks") + for i := range bounds { // clone before any DeleteRange mutates the tree + bounds[i] = bytes.Clone(bounds[i]) + } + + dr := tx.(kv.HasDeleteRange) + var deleted uint64 + for i := 0; i+1 < len(bounds); i++ { + m, err := dr.DeleteRange(deleteRangeTable, bounds[i], bounds[i+1]) + require.NoError(t, err) + deleted += m + } + require.EqualValues(t, n, deleted) + return nil + })) + require.Zero(t, countTable(t, db)) +} + +func newFilledDupSortDB(t *testing.T, keys, dupsPerKey int) kv.RwDB { + t.Helper() + db := New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WriteMap(true).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{deleteRangeTable: kv.TableCfgItem{Flags: kv.DupSort}} + }).MapSize(512 * datasize.MB).MustOpen() + t.Cleanup(db.Close) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursorDupSort(deleteRangeTable) + require.NoError(t, err) + defer c.Close() + for i := 0; i < keys; i++ { + for d := 0; d < dupsPerKey; d++ { + require.NoError(t, c.AppendDup(u64tob(uint64(i)), u64tob(uint64(d)))) + } + } + return nil + })) + return db +} + +// TestMdbxDeleteRangeDupSort pins that native range-delete removes every dup of +// each key in [from,to) and counts (key,value) pairs, not distinct keys. +func TestMdbxDeleteRangeDupSort(t *testing.T) { + const keys, dups = 1000, 8 + db := newFilledDupSortDB(t, keys, dups) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + n, err := tx.(kv.HasDeleteRange).DeleteRange(deleteRangeTable, u64tob(200), u64tob(700)) + require.NoError(t, err) + require.EqualValues(t, 500*dups, n) // every dup of keys 200..699 + return nil + })) + require.EqualValues(t, 500*dups, countTable(t, db)) + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + has := func(i uint64) bool { v, _ := tx.GetOne(deleteRangeTable, u64tob(i)); return v != nil } + require.True(t, has(199)) // dups kept + require.False(t, has(200)) // all dups gone + require.False(t, has(699)) + require.True(t, has(700)) + return nil + })) +} + +// TestChunkedDeleteRangeDupSortCoversAllKeys is the DupSort analogue of +// TestChunkedDeleteRangeCoversAllKeys: chunked range-delete over count-balanced +// bounds must remove every (key,dup) with no gaps — the shape ResetExec clears. +func TestChunkedDeleteRangeDupSortCoversAllKeys(t *testing.T) { + const keys, dups = 50_000, 8 + db := newFilledDupSortDB(t, keys, dups) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + bounds, err := tx.(kv.DBWithDistributionSupport).DistributeCursors(deleteRangeTable, nil, 256) + require.NoError(t, err) + require.Greater(t, len(bounds), 2, "dupsort table must split into multiple chunks") + for i := range bounds { + bounds[i] = bytes.Clone(bounds[i]) + } + dr := tx.(kv.HasDeleteRange) + var deleted uint64 + for i := 0; i+1 < len(bounds); i++ { + m, err := dr.DeleteRange(deleteRangeTable, bounds[i], bounds[i+1]) + require.NoError(t, err) + deleted += m + } + require.EqualValues(t, keys*dups, deleted) + return nil + })) + require.Zero(t, countTable(t, db)) +} diff --git a/db/kv/mdbx/kv_mdbx.go b/db/kv/mdbx/kv_mdbx.go index 35ca589967f..19dcf3a590c 100644 --- a/db/kv/mdbx/kv_mdbx.go +++ b/db/kv/mdbx/kv_mdbx.go @@ -882,6 +882,99 @@ func (db *MdbxKV) View(ctx context.Context, f func(tx kv.Tx) error) (err error) return f(tx) } +func rawCursor(c kv.Cursor) *mdbx.Cursor { + if dc, ok := c.(*MdbxDupSortCursor); ok { + return dc.c + } + return c.(*MdbxCursor).c +} + +const maxDistributeCursors = 4096 + +// DistributeCursors partitions bucket into n approximately equal-count key +// ranges using mdbx's b-tree distribution. Fast on db >> RAM: it touches only +// the b-tree branch nodes. Interior boundaries point into tx-owned pages — clone +// them before any same-tx write (a range-delete invalidates them). +func (tx *MdbxTx) DistributeCursors(table string, from []byte, n int) ([][]byte, error) { + if n <= 1 { + return [][]byte{from, nil}, nil + } + n = min(n, maxDistributeCursors) // n is derived from table size; cap it so a multi-TB table doesn't open tens of thousands of cursors + + firstC, err := tx.Cursor(table) + if err != nil { + return nil, err + } + defer firstC.Close() + var fk []byte + if from == nil { + fk, _, err = firstC.First() + } else { + fk, _, err = firstC.Seek(from) + } + if err != nil { + return nil, err + } + if fk == nil { // empty table, or `from` is past the last key + return [][]byte{from, nil}, nil + } + + // Read positions through kv.Cursor.Current: it normalizes the not-found/EOF + // signal, sparing us platform-specific mdbx error codes. + wrappers := make([]kv.Cursor, n) + cursors := make([]*mdbx.Cursor, n) + for i := range wrappers { + cw, err := tx.Cursor(table) + if err != nil { + return nil, err + } + defer cw.Close() + wrappers[i], cursors[i] = cw, rawCursor(cw) + } + + // Distribute at the lowest branch level (Depth-2), not the leaves: keeps the + // count-traversal in branch pages (cheap, cacheable) instead of faulting cold + // leaves on a >>RAM table. Balance stays within ~1%; 42 only for shallow trees. + deepness := uint(42) + if st, err := tx.BucketStat(table); err == nil && st.Depth > 2 { + deepness = st.Depth - 2 + } + // allSet is false when the range had fewer positions than n: the surplus + // cursors are left hollow, and Current() normalizes them to a nil key below. + allSet, err := mdbx.DistributeCursors(rawCursor(firstC), nil, cursors, deepness) + if err != nil { + return nil, err + } + + keys := make([][]byte, 0, n) + for _, cw := range wrappers { + k, _, err := cw.Current() + if err != nil { + return nil, err + } + if k == nil { // hollow surplus cursor: Current normalizes ENODATA to nil + break + } + keys = append(keys, k) + } + if allSet && len(keys) > 0 { + keys = keys[:len(keys)-1] // last cursor pins the table's last key; nil closes the final range + } + + bounds := make([][]byte, 0, len(keys)+2) + bounds = append(bounds, from) + var prev []byte + for _, k := range keys { + if prev != nil && bytes.Equal(k, prev) { + continue // clustered keys collapsed to one boundary + } + bounds = append(bounds, k) + prev = k + } + bounds = append(bounds, nil) + return bounds, nil +} + func (tx *MdbxTx) Apply(_ context.Context, f func(tx kv.Tx) error) (err error) { return f(tx) } @@ -1103,6 +1196,46 @@ func (tx *MdbxTx) ClearTable(bucket string) error { return tx.tx.Drop(mdbx.DBI(dbi), false) } +// DeleteRange removes keys in [from, to) using mdbx's native bulk range-delete, +// which cuts whole pages and branches out of the B-tree at once. to==nil deletes +// through the last key. Returns the number of keys removed. +func (tx *MdbxTx) DeleteRange(table string, from, to []byte) (uint64, error) { + beginC, err := tx.RwCursor(table) + if err != nil { + return 0, err + } + defer beginC.Close() + bk, _, err := beginC.Seek(from) + if err != nil { + return 0, err + } + if bk == nil { + return 0, nil + } + + begin := rawCursor(beginC) + var end *mdbx.Cursor + endIncluding := true // nil end => delete through the last key + if to != nil { + endC, err := tx.RwCursor(table) + if err != nil { + return 0, err + } + defer endC.Close() + ek, _, err := endC.Seek(to) + if err != nil { + return 0, err + } + if ek != nil { + if bytes.Compare(bk, ek) >= 0 { + return 0, nil // empty range + } + end, endIncluding = rawCursor(endC), false + } + } + return begin.DeleteRange(end, endIncluding) +} + func (tx *MdbxTx) DropTable(bucket string) error { if cfg, ok := tx.db.buckets[bucket]; !(ok && cfg.IsDeprecated) { return fmt.Errorf("%w, bucket: %s", kv.ErrAttemptToDeleteNonDeprecatedBucket, bucket) @@ -1479,7 +1612,7 @@ func (c *MdbxCursor) Prev() (k, v []byte, err error) { func (c *MdbxCursor) Current() ([]byte, []byte, error) { k, v, err := c.c.Get(nil, nil, mdbx.GetCurrent) if err != nil { - if mdbx.IsNotFound(err) { + if mdbx.IsNotFound(err) || mdbx.IsNoData(err) { return nil, nil, nil } return []byte{}, nil, err diff --git a/db/kv/mdbx/split_bucket_test.go b/db/kv/mdbx/split_bucket_test.go new file mode 100644 index 00000000000..3014012f347 --- /dev/null +++ b/db/kv/mdbx/split_bucket_test.go @@ -0,0 +1,231 @@ +// Copyright 2024 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package mdbx + +import ( + "bytes" + "encoding/binary" + "sort" + "testing" + + "github.com/c2h5oh/datasize" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/db/kv/dbcfg" + "github.com/erigontech/erigon/db/kv/order" +) + +// Keys sharing a 32-byte prefix and differing only in a 4-byte suffix: the +// clustered shape that byte interpolation over the first 32 bytes can't split. +func clusteredKey(i int) []byte { + k := make([]byte, 36) + for j := 0; j < 32; j++ { + k[j] = 0xCC + } + binary.BigEndian.PutUint32(k[32:], uint32(i)) + return k +} + +func TestSplitBucketByCount(t *testing.T) { + const table, n, chunks = "T", 100_000, 16 + db := New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{table: kv.TableCfgItem{}} + }).MapSize(512 * datasize.MB).MustOpen() + t.Cleanup(db.Close) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursor(table) + require.NoError(t, err) + defer c.Close() + for i := 0; i < n; i++ { + require.NoError(t, c.Append(clusteredKey(i), []byte{1})) + } + return nil + })) + + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + bounds, err := tx.(kv.DBWithDistributionSupport).DistributeCursors(table, nil, chunks) + require.NoError(t, err) + + require.Greater(t, len(bounds), 2, "must split clustered keys into many ranges, not one") + require.Nil(t, bounds[0]) + require.Nil(t, bounds[len(bounds)-1]) + require.True(t, sort.SliceIsSorted(bounds[1:len(bounds)-1], func(a, b int) bool { + return bytes.Compare(bounds[1+a], bounds[1+b]) < 0 + }), "interior boundaries must be strictly increasing") + + per, total := n/(len(bounds)-1), 0 + for i := 0; i+1 < len(bounds); i++ { + it, err := tx.Range(table, bounds[i], bounds[i+1], order.Asc, -1) + require.NoError(t, err) + cnt := 0 + for it.HasNext() { + _, _, err := it.Next() + require.NoError(t, err) + cnt++ + } + it.Close() + total += cnt + require.InDelta(t, per, cnt, float64(per), "range %d holds %d keys, want ~%d", i, cnt, per) + } + require.Equal(t, n, total, "ranges must cover every key exactly once") + return nil + })) +} + +// A DupSort table's tx.Cursor returns *MdbxDupSortCursor, not *MdbxCursor, so +// the split must extract the raw cursor from either type instead of asserting one. +func TestSplitBucketByCountDupSort(t *testing.T) { + const table, keys, dupsPerKey, chunks = "T", 50_000, 4, 16 + db := New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{table: kv.TableCfgItem{Flags: kv.DupSort}} + }).MapSize(512 * datasize.MB).MustOpen() + t.Cleanup(db.Close) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursorDupSort(table) + require.NoError(t, err) + defer c.Close() + key, val := make([]byte, 8), make([]byte, 8) + for i := 0; i < keys; i++ { + binary.BigEndian.PutUint64(key, uint64(i)) + for j := 0; j < dupsPerKey; j++ { + binary.BigEndian.PutUint64(val, uint64(j)) + require.NoError(t, c.AppendDup(key, val)) + } + } + return nil + })) + + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + bounds, err := tx.(kv.DBWithDistributionSupport).DistributeCursors(table, nil, chunks) + require.NoError(t, err) + require.Greater(t, len(bounds), 2, "must split into many ranges") + require.Nil(t, bounds[0]) + require.Nil(t, bounds[len(bounds)-1]) + require.True(t, sort.SliceIsSorted(bounds[1:len(bounds)-1], func(a, b int) bool { + return bytes.Compare(bounds[1+a], bounds[1+b]) < 0 + }), "interior boundaries must be strictly increasing") + + seen := 0 + for i := 0; i+1 < len(bounds); i++ { + it, err := tx.Range(table, bounds[i], bounds[i+1], order.Asc, -1) + require.NoError(t, err) + for it.HasNext() { + _, _, err := it.Next() + require.NoError(t, err) + seen++ + } + it.Close() + } + require.Equal(t, keys*dupsPerKey, seen, "ranges must cover every (key,val) once") + return nil + })) +} + +// A from near the table end leaves the range with far fewer positions than the +// requested cursor count, so DistributeCursors leaves surplus cursors unset. +// Reading an unset cursor reports ENODATA, which must be treated as "no more +// positions", not propagated as an error. +func TestSplitBucketByCountFewerPositions(t *testing.T) { + const table, n = "T", 200_000 + db := New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{table: kv.TableCfgItem{}} + }).MapSize(512 * datasize.MB).MustOpen() + t.Cleanup(db.Close) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursor(table) + require.NoError(t, err) + defer c.Close() + k := make([]byte, 8) + for i := 0; i < n; i++ { + binary.BigEndian.PutUint64(k, uint64(i)) + require.NoError(t, c.Append(k, []byte{1})) + } + return nil + })) + + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + from := make([]byte, 8) + binary.BigEndian.PutUint64(from, uint64(n-5)) // only ~5 positions left, but n_cap is far larger + bounds, err := tx.(kv.DBWithDistributionSupport).DistributeCursors(table, from, 4096) + require.NoError(t, err) // unset surplus cursors must not surface as an error + require.GreaterOrEqual(t, len(bounds), 2) + require.Equal(t, from, bounds[0]) + require.Nil(t, bounds[len(bounds)-1]) + return nil + })) +} + +// TestSplitBucketByCountDupSortSkew proves DistributeCursors splits a DupSort +// table at MAIN-KEY granularity: distribution runs at deepness Depth-2 (branch +// level, above the per-key dup subtrees) so it never faults cold leaves on a +// table >> RAM, but the price is that it cannot balance a skewed table — one hot +// key with a large dup run lands entirely in a single chunk. Coverage stays +// exact; only balance suffers. Splitting within a key's dups would need +// deepness = full height + dup height, i.e. faulting the leaves this avoids. +func TestSplitBucketByCountDupSortSkew(t *testing.T) { + const table, coldKeys, hotDups = "T", 1000, 200_000 + db := New(dbcfg.ChainDB, log.New()).InMem(t, t.TempDir()).WithTableCfg(func(_ kv.TableCfg) kv.TableCfg { + return kv.TableCfg{table: kv.TableCfgItem{Flags: kv.DupSort}} + }).MapSize(512 * datasize.MB).MustOpen() + t.Cleanup(db.Close) + + require.NoError(t, db.Update(t.Context(), func(tx kv.RwTx) error { + c, err := tx.RwCursorDupSort(table) + require.NoError(t, err) + defer c.Close() + key, val := make([]byte, 8), make([]byte, 8) + for d := 0; d < hotDups; d++ { // key 0 is hot: many dups + binary.BigEndian.PutUint64(val, uint64(d)) + require.NoError(t, c.AppendDup(key, val)) + } + binary.BigEndian.PutUint64(val, 0) + for i := 1; i <= coldKeys; i++ { // keys 1..coldKeys: one value each + binary.BigEndian.PutUint64(key, uint64(i)) + require.NoError(t, c.AppendDup(key, val)) + } + return nil + })) + + require.NoError(t, db.View(t.Context(), func(tx kv.Tx) error { + bounds, err := tx.(kv.DBWithDistributionSupport).DistributeCursors(table, nil, 64) + require.NoError(t, err) + + total, maxChunk := 0, 0 + for i := 0; i+1 < len(bounds); i++ { + it, err := tx.Range(table, bounds[i], bounds[i+1], order.Asc, -1) + require.NoError(t, err) + cnt := 0 + for it.HasNext() { + _, _, err := it.Next() + require.NoError(t, err) + cnt++ + } + it.Close() + total += cnt + maxChunk = max(maxChunk, cnt) + } + require.Equal(t, hotDups+coldKeys, total, "coverage must be exact: every (key,dup) once") + require.GreaterOrEqual(t, maxChunk, hotDups, "hot key's dups are not split across chunks") + t.Logf("skewed dupsort: %d chunks, largest holds %d/%d items (no intra-key split)", len(bounds)-1, maxChunk, total) + return nil + })) +} diff --git a/db/kv/membatchwithdb/memory_store.go b/db/kv/membatchwithdb/memory_store.go index f61e1ac5c64..2f2d463f5db 100644 --- a/db/kv/membatchwithdb/memory_store.go +++ b/db/kv/membatchwithdb/memory_store.go @@ -26,11 +26,12 @@ import ( "github.com/c2h5oh/datasize" + btree2 "github.com/tidwall/btree" + "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/kv/order" "github.com/erigontech/erigon/db/kv/stream" - btree2 "github.com/tidwall/btree" ) // Compile-time check that memStore implements kv.RwTx. diff --git a/db/kv/readahead.go b/db/kv/readahead.go new file mode 100644 index 00000000000..10d37f26619 --- /dev/null +++ b/db/kv/readahead.go @@ -0,0 +1,243 @@ +// Copyright 2024 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package kv + +import ( + "bytes" + "context" + "sort" + "sync" + "sync/atomic" + + "github.com/c2h5oh/datasize" + "golang.org/x/sync/errgroup" + + "github.com/erigontech/erigon/common/log/v3" +) + +type DBWithDistributionSupport interface { + // DistributeCursors partitions table into n approximately equal-count key + // ranges using mdbx's b-tree distribution. Fast on Table >> RAM: it touches + // only the b-tree branch nodes. Interior boundaries are tx-owned — clone them + // before any same-tx write (see DistributeBounds). + DistributeCursors(table string, from []byte, n int) ([][]byte, error) +} + +// WarmupChunkSize is the target byte size of each count-balanced chunk that +// ReadAhead prefetches, and that chunked range-delete callers split a table +// into. One definition keeps both boundary sets aligned. +const WarmupChunkSize = 32 * datasize.MB + +const readAheadLabel = "read-ahead" + +// DistributeBounds returns cloned count-balanced ~WarmupChunkSize boundaries for +// table plus its size. Bounds are nil when under one chunk or the engine can't +// count-split. Cloning lets them outlive later same-tx mutations. +func DistributeBounds(tx Tx, table string) (bounds [][]byte, size uint64, err error) { + size, err = tx.BucketSize(table) + if err != nil { + return nil, 0, err + } + s, ok := tx.(DBWithDistributionSupport) + if !ok { // engine can't count-split + return nil, size, nil + } + if size < WarmupChunkSize.Bytes() { + return nil, size, nil + } + b, err := s.DistributeCursors(table, nil, int(size/WarmupChunkSize.Bytes())) + if err != nil { + return nil, size, err + } + bounds = make([][]byte, len(b)) + for i, k := range b { + bounds[i] = bytes.Clone(k) + } + return bounds, size, nil +} + +// ReadAheadCfg configures NewReadAhead. Bounds come from DistributeBounds. +type ReadAheadCfg struct { + Bounds [][]byte // count-balanced boundaries (already cloned) + TableSize uint64 // sizes the ~1GB ahead-window in chunks + Workers int + WarmValues bool // fault value pages too (copy path); false = leaf pages only (clear path) +} + +// ReadAhead keeps a bounded window of pages warm just ahead of a forward table scan +type ReadAhead struct { + mu sync.Mutex + turnCond *sync.Cond // broadcast when the consumer advances (SetPos) or ctx is cancelled + + consumerChunk int64 // guarded by mu + bounds [][]byte // set once before run() starts; read-only afterward + cancel context.CancelFunc + done chan struct{} + warmValues bool +} + +// NewReadAhead warms a ~1GB window of cfg.Bounds just ahead of the consumer. Call +// SetPos as it advances and Close when done. Returns nil (a no-op) when db is nil +// or there's <1 chunk. +func NewReadAhead(ctx context.Context, db RoDB, table string, cfg ReadAheadCfg) *ReadAhead { + if db == nil || len(cfg.Bounds) < 2 { + return nil // Close/SetPos are nil-safe + } + workers := cfg.Workers + if workers < 1 { + workers = 1 + } + ctx, cancel := context.WithCancel(ctx) + r := &ReadAhead{cancel: cancel, done: make(chan struct{}), warmValues: cfg.WarmValues, bounds: cfg.Bounds} + r.turnCond = sync.NewCond(&r.mu) + go r.run(ctx, db, table, cfg.Bounds, cfg.TableSize, workers) + return r +} + +func (r *ReadAhead) run(ctx context.Context, db RoDB, table string, bounds [][]byte, tableSize uint64, workers int) { + defer close(r.done) + defer r.cancel() // release the child ctx + its watcher goroutine even if the caller forgets Close + + // ~1GB window from the actual mean chunk size, so it stays ~1GB even when + // huge tables (past the cursor cap) have chunks bigger than WarmupChunkSize. + const aheadBytes = 1 * datasize.GB + nChunks := int64(len(bounds) - 1) + ahead := max(int64(aheadBytes.Bytes())*nChunks/int64(max(tableSize, 1)), 1) + limit := min(workers, int(ahead)+1) // extra workers would only park + + // wake workers parked in waitTurn when ctx is cancelled, so Close() doesn't hang + go func() { + <-ctx.Done() + r.mu.Lock() + r.turnCond.Broadcast() + r.mu.Unlock() + }() + + var g errgroup.Group // not WithContext: one chunk's read error must not cancel the rest + g.SetLimit(limit) + for idx := 0; idx+1 < len(bounds) && ctx.Err() == nil; idx++ { + g.Go(func() error { + if r.waitTurn(ctx, idx, ahead) { // parks until the consumer is close enough + r.warm(ctx, db, table, bounds[idx], bounds[idx+1]) + } + return nil + }) + } + _ = g.Wait() +} + +// waitTurn blocks until chunk idx is within `ahead` of the consumer; returns false +// if the consumer passed it or ctx was cancelled. Parks on a cond var (woken by +// SetPos/Close), not a poll. +func (r *ReadAhead) waitTurn(ctx context.Context, idx int, ahead int64) bool { + r.mu.Lock() + defer r.mu.Unlock() + for { + if ctx.Err() != nil { + return false + } + switch { + case int64(idx) < r.consumerChunk: + return false // consumer already passed this chunk - don't warm behind it + case int64(idx) <= r.consumerChunk+ahead: + return true + } + r.turnCond.Wait() + } +} + +// osPageBytes strides a value one page at a time; 4KiB faults every OS page on +// any platform (16KiB-page systems just get a few redundant hits). +const osPageBytes = 4096 + +// warmupSink must consume the bytes read during warmup: with the result +// discarded, the compiler dead-code-eliminates the value loads and no page is +// ever faulted in. An atomic keeps the loads live and is race-free across workers. +var warmupSink atomic.Uint64 + +// touchValue faults every OS page backing v into cache, folding one byte per +// page into sink. Keys and inline values are already faulted by the b-tree +// traversal that yielded them; only large (overflow-page) values need this. +func touchValue(sink byte, v []byte) byte { + for off := 0; off < len(v); off += osPageBytes { + sink ^= v[off] + } + if n := len(v); n > 0 { + sink ^= v[n-1] + } + return sink +} + +// warm scans [from,to) with a raw cursor (one cgo Get per key) to fault its pages +// into cache — leaf pages always, plus overflow value pages when warmValues is +// set (the copy path reads values; range-delete doesn't). +func (r *ReadAhead) warm(ctx context.Context, db RoDB, table string, from, to []byte) { + err := db.View(ctx, func(tx Tx) error { + c, err := tx.Cursor(table) + if err != nil { + return err + } + defer c.Close() + var sink byte + defer func() { warmupSink.Add(uint64(sink)) }() + n := 0 + for k, v, err := c.Seek(from); k != nil && (to == nil || bytes.Compare(k, to) < 0); k, v, err = c.Next() { + if err != nil { + return err + } + if r.warmValues { + sink = touchValue(sink, v) + } + n++ + if n%128 == 0 && ctx.Err() != nil { + return ctx.Err() + } + } + return nil + }) + if err != nil && ctx.Err() == nil { // best-effort warmup, but a real DB error shouldn't vanish + log.Warn("["+readAheadLabel+"] warmup failed", "table", table, "err", err) + } +} + +// SetPos reports the consumer's position so prefetchers throttle to a bounded +// distance ahead; cheap (one binary search), nil-safe, call every few thousand keys. +func (r *ReadAhead) SetPos(key []byte) { + if r == nil { + return + } + idx := sort.Search(len(r.bounds)-1, func(i int) bool { + b := r.bounds[i+1] + return b == nil || bytes.Compare(b, key) > 0 + }) + r.mu.Lock() + if int64(idx) != r.consumerChunk { // only wake parked workers when the window actually moves + r.consumerChunk = int64(idx) + r.turnCond.Broadcast() + } + r.mu.Unlock() +} + +// Close stops the prefetchers and waits for their read txs to be released. +// Safe to call on a nil *ReadAhead. +func (r *ReadAhead) Close() { + if r == nil { + return + } + r.cancel() + <-r.done +} diff --git a/db/kv/table_sizes.go b/db/kv/table_sizes.go index f85b0c087da..7bddf28dced 100644 --- a/db/kv/table_sizes.go +++ b/db/kv/table_sizes.go @@ -91,6 +91,9 @@ func CollectTableSizesPeriodically(ctx context.Context, db TemporalRoDB, label L case <-ticker.C: tableSizes, err := CollectTableSizes(ctx, db) if err != nil { + if ctx.Err() != nil { // graceful shutdown mid-collect, not a real failure + return + } logger.Error("[kv] failed to collect table sizes", "err", err) continue } diff --git a/db/kv/temporal/kv_temporal.go b/db/kv/temporal/kv_temporal.go index a85e42b2dac..19f8657d069 100644 --- a/db/kv/temporal/kv_temporal.go +++ b/db/kv/temporal/kv_temporal.go @@ -19,6 +19,7 @@ package temporal import ( "context" "errors" + "fmt" "sync" "testing" "time" @@ -296,6 +297,22 @@ func (tx *Tx) LockDBInRam() error { return nil } +// DistributeCursors forwards to the underlying mdbx engine. Like DeleteRange, it +// fails loud on a non-mdbx inner rather than silently degrading (none exists today). +func (tx *Tx) DistributeCursors(table string, from []byte, n int) ([][]byte, error) { + if s, ok := tx.Tx.(kv.DBWithDistributionSupport); ok { + return s.DistributeCursors(table, from, n) + } + return nil, fmt.Errorf("DistributeCursors not supported by %T", tx.Tx) +} + +func (tx *RwTx) DistributeCursors(table string, from []byte, n int) ([][]byte, error) { + if s, ok := tx.RwTx.(kv.DBWithDistributionSupport); ok { + return s.DistributeCursors(table, from, n) + } + return nil, fmt.Errorf("DistributeCursors not supported by %T", tx.RwTx) +} + func (tx *Tx) Apply(ctx context.Context, f func(tx kv.Tx) error) error { tx.tx.mu.RLock() applyTx := tx.Tx @@ -313,6 +330,15 @@ func (tx *RwTx) WarmupDB(force bool) error { return nil } +func (tx *RwTx) DeleteRange(table string, from, to []byte) (uint64, error) { + if dr, ok := tx.RwTx.(kv.HasDeleteRange); ok { + return dr.DeleteRange(table, from, to) + } + // No non-mdbx temporal backend exists (mdbx's native range-delete is used + // above); fail loud rather than carry an unexercised, DupSort-unsafe emulation. + return 0, fmt.Errorf("DeleteRange not supported by %T", tx.RwTx) +} + func (tx *RwTx) LockDBInRam() error { if mdbxTx, ok := tx.RwTx.(*mdbx.MdbxTx); ok { return mdbxTx.LockDBInRam() diff --git a/db/rawdb/blockio/block_writer.go b/db/rawdb/blockio/block_writer.go index 6631d5a3dae..86e6db53da1 100644 --- a/db/rawdb/blockio/block_writer.go +++ b/db/rawdb/blockio/block_writer.go @@ -100,7 +100,7 @@ func (w *BlockWriter) TruncateBodies(db kv.RoDB, tx kv.RwTx, from uint64) error return err } - if err := backup.ClearTables(context.Background(), tx, kv.EthTx, kv.MaxTxNum); err != nil { + if err := backup.ClearTables(context.Background(), db, tx, kv.EthTx, kv.MaxTxNum); err != nil { return err } if err := tx.ResetSequence(kv.EthTx, 0); err != nil { diff --git a/db/snapshotsync/freezeblocks/block_snapshots.go b/db/snapshotsync/freezeblocks/block_snapshots.go index 1356fb074df..6681498a67a 100644 --- a/db/snapshotsync/freezeblocks/block_snapshots.go +++ b/db/snapshotsync/freezeblocks/block_snapshots.go @@ -745,11 +745,11 @@ func DumpTxs(ctx context.Context, db kv.RoDB, chainConfig *chain.Config, blockFr } if doWarmup && !warmupSenders.Load() && blockNum%1_000 == 0 { - clean := kv.ReadAhead(warmupCtx, db, warmupSenders, kv.Senders, hexutil.EncodeTs(blockNum), 10_000) + clean := kv.ReadAheadDeprecated(warmupCtx, db, warmupSenders, kv.Senders, hexutil.EncodeTs(blockNum), 10_000) defer clean() } if doWarmup && !warmupTxs.Load() && blockNum%1_000 == 0 { - clean := kv.ReadAhead(warmupCtx, db, warmupTxs, kv.EthTx, body.BaseTxnID.Bytes(), 100*10_000) + clean := kv.ReadAheadDeprecated(warmupCtx, db, warmupTxs, kv.EthTx, body.BaseTxnID.Bytes(), 100*10_000) defer clean() } senders, err := rawdb.ReadSenders(tx, h, blockNum) diff --git a/execution/stagedsync/rawdbreset/reset_stages.go b/execution/stagedsync/rawdbreset/reset_stages.go index 43a5b16a724..579466dc640 100644 --- a/execution/stagedsync/rawdbreset/reset_stages.go +++ b/execution/stagedsync/rawdbreset/reset_stages.go @@ -108,7 +108,7 @@ func ResetCanonicalAndRefillFromSnapshots(ctx context.Context, db kv.TemporalRwD }) } -func ResetBlocks(tx kv.RwTx, db kv.RoDB, br services.FullBlockReader, bw *blockio.BlockWriter, dirs datadir.Dirs, logger log.Logger) error { +func ResetBlocks(db kv.RwDB, tx kv.RwTx, br services.FullBlockReader, bw *blockio.BlockWriter, dirs datadir.Dirs, logger log.Logger) error { // keep Genesis if err := rawdb.TruncateBlocks(context.Background(), tx, 1); err != nil { return err @@ -157,14 +157,16 @@ func ResetBlocks(tx kv.RwTx, db kv.RoDB, br services.FullBlockReader, bw *blocki return nil } -func ResetSenders(ctx context.Context, tx kv.RwTx) error { - if err := backup.ClearTables(ctx, tx, kv.Senders); err != nil { - return fmt.Errorf("clearing senders table: %w", err) - } - return clearStageProgress(tx, stages.Senders) +func ResetSenders(ctx context.Context, db kv.RwDB) error { + return db.Update(ctx, func(tx kv.RwTx) error { + if err := backup.ClearTables(ctx, db, tx, kv.Senders); err != nil { + return fmt.Errorf("clearing senders table: %w", err) + } + return clearStageProgress(tx, stages.Senders) + }) } -func ResetExec(ctx context.Context, db kv.TemporalRwDB) (err error) { +func ResetExec(ctx context.Context, db kv.TemporalRwDB) error { domainTablesCount := len(db.Debug().DomainTables(kv.AccountsDomain, kv.StorageDomain, kv.CodeDomain, kv.CommitmentDomain, kv.ReceiptDomain, kv.RCacheDomain)) invertedIdxTablesCount := len(db.Debug().InvertedIdxTables(kv.LogAddrIdx, kv.LogTopicIdx, kv.TracesFromIdx, kv.TracesToIdx)) cleanupList := make([]string, 0, len(stateBuckets)+len(stateHistoryBuckets)+domainTablesCount+invertedIdxTablesCount) @@ -175,13 +177,12 @@ func ResetExec(ctx context.Context, db kv.TemporalRwDB) (err error) { if err := db.Update(ctx, func(tx kv.RwTx) error { if err := clearStageProgress(tx, stages.Execution); err != nil { - return err - } - - if err := backup.ClearTables(ctx, tx, cleanupList...); err != nil { - return fmt.Errorf("clearing exec state tables: %w", err) + return fmt.Errorf("clearing Execution stage progress: %w", err) } // corner case: state files may be ahead of block files - so, can't use SharedDomains here. just leave progress as 0. + if err := backup.ClearTables(ctx, db, tx, cleanupList...); err != nil { + return fmt.Errorf("reset exec state tables: %w", err) + } return nil }); err != nil { return err @@ -271,14 +272,11 @@ func clearStageProgress(tx kv.RwTx, stagesList ...stages.SyncStage) error { func Reset(ctx context.Context, db kv.RwDB, stagesList ...stages.SyncStage) error { return db.Update(ctx, func(tx kv.RwTx) error { for _, st := range stagesList { - if err := backup.ClearTables(ctx, tx, Tables[st]...); err != nil { - return err - } - if err := clearStageProgress(tx, stagesList...); err != nil { + if err := backup.ClearTables(ctx, db, tx, Tables[st]...); err != nil { return err } } - return nil + return clearStageProgress(tx, stagesList...) }) } diff --git a/execution/stagedsync/stage_custom_trace.go b/execution/stagedsync/stage_custom_trace.go index 27d5b33a01a..e9da7f01462 100644 --- a/execution/stagedsync/stage_custom_trace.go +++ b/execution/stagedsync/stage_custom_trace.go @@ -501,7 +501,7 @@ func StageCustomTraceReset(ctx context.Context, db kv.TemporalRwDB, produce Prod if produce.TraceTo { tables = append(tables, db.Debug().InvertedIdxTables(kv.TracesToIdx)...) } - if err := backup.ClearTables(ctx, tx, tables...); err != nil { + if err := backup.ClearTables(ctx, db, tx, tables...); err != nil { return err } // Clearing the data tables alone is not enough: the prune-progress bookmark in diff --git a/go.mod b/go.mod index 1bd82af791e..0ca91f41ad1 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/erigontech/evmone_precompiles v0.0.0-20260414072133-b8b2bdc99de2 github.com/erigontech/fastkeccak v0.1.1-0.20260408010752-08e7b6602268 github.com/erigontech/go-libdeflate v0.1.0 - github.com/erigontech/mdbx-go v0.40.1 + github.com/erigontech/mdbx-go v0.40.3 github.com/erigontech/secp256k1 v1.2.1-0.20260218182123-377cc1bd6410 ) @@ -270,7 +270,7 @@ require ( github.com/hashicorp/go-immutable-radix/v2 v2.1.0 // indirect github.com/hashicorp/go-version v1.9.0 // indirect github.com/hexops/gotextdiff v1.0.3 // indirect - github.com/ianlancetaylor/cgosymbolizer v0.0.0-20241129212102-9c50ad6b591e // indirect + github.com/ianlancetaylor/cgosymbolizer v0.0.0-20260504013507-f4b012c11129 // indirect github.com/imdario/mergo v0.3.11 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/ipfs/go-cid v0.5.0 // indirect diff --git a/go.sum b/go.sum index 761e42f5764..f9e73db504c 100644 --- a/go.sum +++ b/go.sum @@ -332,8 +332,8 @@ github.com/erigontech/go-eth-kzg v0.0.0-20260401161010-070339460d07 h1:0VpFIthaJ github.com/erigontech/go-eth-kzg v0.0.0-20260401161010-070339460d07/go.mod h1:J9/u5sWfznSObptgfa92Jq8rTswn6ahQWEuiLHOjCUI= github.com/erigontech/go-libdeflate v0.1.0 h1:WtKiuNpuAP/Qyq3mbPwFB0DoQikabtOeZFUgo9VGRsY= github.com/erigontech/go-libdeflate v0.1.0/go.mod h1:OCQBWlynNqNCYb1ckqNjY69H1LAhg2TyyfYZalO03BM= -github.com/erigontech/mdbx-go v0.40.1 h1:d8aeokoiLnT4MrRq4bFdqnPO714743d7bd6hC2tKy1w= -github.com/erigontech/mdbx-go v0.40.1/go.mod h1:VTGwYzepS9Wg38jVfreOsSVlh73OBGPZluu7kHo6X6g= +github.com/erigontech/mdbx-go v0.40.3 h1:tNFzVF4gs0DxCWGBb91jRKQNA+IeMonOK3LHvuxh5qo= +github.com/erigontech/mdbx-go v0.40.3/go.mod h1:QmkpLmcJX5+Fyu9u8eGsyPwoBDrIZJBBLgJHBMFStWE= github.com/erigontech/secp256k1 v1.2.1-0.20260218182123-377cc1bd6410 h1:5YD7JJ5PaqOdjKA84lTDtQby9nbI4podqrkUhyIyFDw= github.com/erigontech/secp256k1 v1.2.1-0.20260218182123-377cc1bd6410/go.mod h1:CnigLRUsfobnTYoUzwdT/De67fw9OhTA0KTkxa+svik= github.com/ettle/strcase v0.2.0 h1:fGNiVF21fHXpX1niBgk0aROov1LagYsOwV/xqKDKR/Q= @@ -581,8 +581,8 @@ github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE= github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc= github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8= -github.com/ianlancetaylor/cgosymbolizer v0.0.0-20241129212102-9c50ad6b591e h1:8AnObPi8WmIgjwcidUxaREhXMSpyUJeeSrIkZTXdabw= -github.com/ianlancetaylor/cgosymbolizer v0.0.0-20241129212102-9c50ad6b591e/go.mod h1:DvXTE/K/RtHehxU8/GtDs4vFtfw64jJ3PaCnFri8CRg= +github.com/ianlancetaylor/cgosymbolizer v0.0.0-20260504013507-f4b012c11129 h1:ih8K3uz7eATRX/Y6HRpCNVWArsI/65kICdUEpO5Lle8= +github.com/ianlancetaylor/cgosymbolizer v0.0.0-20260504013507-f4b012c11129/go.mod h1:DvXTE/K/RtHehxU8/GtDs4vFtfw64jJ3PaCnFri8CRg= github.com/ianlancetaylor/demangle v0.0.0-20220319035150-800ac71e25c2/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w= github.com/ianlancetaylor/demangle v0.0.0-20230524184225-eabc099b10ab/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw= github.com/imdario/mergo v0.3.11 h1:3tnifQM4i+fbajXKBHXWEH+KvNHqojZ778UH75j3bGA=