forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core/state/snapshot: move genMarker into generator

Browse source
This commit is contained in:
Gary Rong 2024-06-21 10:32:13 +08:00
parent 4cb88e888a
commit be29164cec
7 changed files with 120 additions and 113 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
core/state/snapshot/context.go
View file

@ -91,8 +91,6 @@ func (gs *generatorStats) log(msg string, root common.Hash, marker []byte) {
// current generation cycle.
type generatorContext struct {
root common.Hash // State root of the generation target
marker []byte // Generation progress marker
setMarker func(marker []byte) // Function to notify the generation progress
account *holdableIterator // Iterator of account snapshot data
storage *holdableIterator // Iterator of storage snapshot data
db ethdb.KeyValueStore // Key-value store containing the snapshot data
@ -101,11 +99,9 @@ type generatorContext struct {
}
// newGeneratorContext initializes the context for generation.
func newGeneratorContext(root common.Hash, marker []byte, setMarker func(marker []byte), db ethdb.KeyValueStore) *generatorContext {
func newGeneratorContext(root common.Hash, marker []byte, db ethdb.KeyValueStore) *generatorContext {
ctx := &generatorContext{
root: root,
marker: marker,
setMarker: setMarker,
db: db,
batch: db.NewBatch(),
logged: time.Now(),
@ -116,13 +112,6 @@ func newGeneratorContext(root common.Hash, marker []byte, setMarker func(marker
return ctx
}
// setGenMarker updates the generation progress marker locally and cascades it
// to associated disk layer.
func (ctx *generatorContext) setGenMarker(marker []byte) {
ctx.marker = marker
ctx.setMarker(marker)
}
// openIterator constructs global account and storage snapshot iterators
// at the interrupted position. These iterators should be reopened from time
// to time to avoid blocking leveldb compaction for a long time.

29
core/state/snapshot/disklayer.go
View file

@ -35,12 +35,12 @@ type diskLayer struct {
root common.Hash // Root hash of the base snapshot
stale bool // Signals that the layer became stale (state progressed)
genMarker []byte // Marker for the state that's indexed during initial layer generation
lock sync.RWMutex // Lock to protect stale and genMarker
lock sync.RWMutex // Lock to protect stale
// State snapshot generator, set only if background generation is granted.
// Normally, a non-nil generator indicates that background snapshot generation
// is actively running, except for very short periods during restarts.
// The generator is set if the state snapshot was not fully completed and
// will be unset if the state snapshot is completed later.
//
// The generator is thread-safe, no lock protection needed for access.
generator *generator
}
@ -114,7 +114,8 @@ func (dl *diskLayer) AccountRLP(hash common.Hash) ([]byte, error) {
}
// If the layer is being generated, ensure the requested hash has already been
// covered by the generator.
if dl.genMarker != nil && bytes.Compare(hash[:], dl.genMarker) > 0 {
marker := dl.genMarker()
if marker != nil && bytes.Compare(hash[:], marker) > 0 {
return nil, ErrNotCoveredYet
}
// If we're in the disk layer, all diff layers missed
@ -154,7 +155,8 @@ func (dl *diskLayer) Storage(accountHash, storageHash common.Hash) ([]byte, erro
// If the layer is being generated, ensure the requested hash has already been
// covered by the generator.
if dl.genMarker != nil && bytes.Compare(key, dl.genMarker) > 0 {
marker := dl.genMarker()
if marker != nil && bytes.Compare(key, marker) > 0 {
return nil, ErrNotCoveredYet
}
// If we're in the disk layer, all diff layers missed
@ -186,10 +188,11 @@ func (dl *diskLayer) Update(blockHash common.Hash, destructs map[common.Hash]str
return newDiffLayer(dl, blockHash, destructs, accounts, storage)
}
// setGenMarker updates the generation progress marker with provided value.
func (dl *diskLayer) setGenMarker(marker []byte) {
dl.lock.Lock()
defer dl.lock.Unlock()
dl.genMarker = marker
// genMarker returns the current state snapshot generation progress marker. If
// the state snapshot has already been fully generated, nil is returned.
func (dl *diskLayer) genMarker() []byte {
if dl.generator == nil {
return nil
}
return dl.generator.progressMarker()
}

3
core/state/snapshot/disklayer_test.go
View file

@ -19,6 +19,7 @@ package snapshot
import (
"bytes"
"testing"
"time"
"github.com/VictoriaMetrics/fastcache"
"github.com/ethereum/go-ethereum/common"
@ -302,7 +303,7 @@ func TestDiskPartialMerge(t *testing.T) {
},
},
}
snaps.layers[baseRoot].(*diskLayer).genMarker = genMarker
snaps.layers[baseRoot].(*diskLayer).generator = newGenerator(db, nil, true, genMarker, &generatorStats{start: time.Now()})
base := snaps.Snapshot(baseRoot)
// assertAccount ensures that an account matches the given blob if it's

85
core/state/snapshot/generate.go
View file

@ -20,7 +20,7 @@ import (
"bytes"
"errors"
"fmt"
"sync/atomic"
"sync"
"time"
"github.com/VictoriaMetrics/fastcache"
@ -54,22 +54,30 @@ var (
errMissingTrie = errors.New("missing trie")
)
// generator is the struct for initial state snapshot generation.
// Generator is the struct for initial state snapshot generation. It is not thread-safe;
// the caller must manage concurrency issues themselves.
type generator struct {
active atomic.Bool // Flag if the background generation is running
readOnly bool // Flag indicating whether state snapshot generation is permitted
active bool // Flag indicating whether the background generation is running
diskdb ethdb.KeyValueStore // Key-value store containing the base snapshot
triedb *triedb.Database // Trie node cache for reconstruction purposes
stats *generatorStats // Generation statistics used throughout the entire life cycle
abort chan chan struct{} // Notification channel to abort generating the snapshot in this layer
done chan struct{} // Notification channel when generation is done (test synchronicity)
progress []byte // Progress marker of the state generation, nil means it's completed
lock sync.RWMutex // Lock which protects the progress
}
// newGenerator constructs the state snapshot generator.
func newGenerator(diskdb ethdb.KeyValueStore, triedb *triedb.Database, stats *generatorStats) *generator {
func newGenerator(diskdb ethdb.KeyValueStore, triedb *triedb.Database, readOnly bool, progress []byte, stats *generatorStats) *generator {
if stats == nil {
stats = &generatorStats{start: time.Now()}
}
return &generator{
readOnly: readOnly,
progress: progress,
diskdb: diskdb,
triedb: triedb,
stats: stats,
@ -78,27 +86,38 @@ func newGenerator(diskdb ethdb.KeyValueStore, triedb *triedb.Database, stats *ge
}
}
// run starts the state snapshot generation in the background. It will panic
// if the previous cycle is not terminated yet.
func (g *generator) run(root common.Hash, marker []byte, setMarker func([]byte)) {
if g.active.Load() {
panic("the previous generation cycle is not aborted")
// run starts the state snapshot generation in the background.
func (g *generator) run(root common.Hash) {
if g.readOnly {
log.Info("Snapshot generator is in read-only mode")
return
}
g.active.Store(true)
go g.generate(newGeneratorContext(root, marker, setMarker, g.diskdb))
if g.active {
g.stop()
log.Info("Terminated the state snapshot generation")
}
g.active = true
go g.generate(newGeneratorContext(root, g.progress, g.diskdb))
}
// stop terminates the background generation if it's actively running.
func (g *generator) stop() {
if !g.active.Load() {
log.Error("State snapshot is not generating")
if !g.active {
log.Info("State snapshot is not generating")
return
}
ch := make(chan struct{})
g.abort <- ch
<-ch
g.active.Store(false)
g.active = false
}
// progressMarker returns the current generation progress marker.
func (g *generator) progressMarker() []byte {
g.lock.RLock()
defer g.lock.RUnlock()
return g.progress
}
// splitMarker is an internal helper which splits the generation progress marker
@ -130,10 +149,9 @@ func generateSnapshot(diskdb ethdb.KeyValueStore, triedb *triedb.Database, cache
diskdb: diskdb,
cache: fastcache.New(cache * 1024 * 1024),
root: root,
genMarker: genMarker,
generator: newGenerator(diskdb, triedb, stats),
generator: newGenerator(diskdb, triedb, false, genMarker, stats),
}
base.generator.run(root, genMarker, base.setGenMarker)
base.generator.run(root)
log.Debug("Start snapshot generation", "root", root)
return base
}
@ -536,8 +554,8 @@ func (g *generator) checkAndFlush(ctx *generatorContext, current []byte) error {
default:
}
if ctx.batch.ValueSize() > ethdb.IdealBatchSize || abort != nil {
if bytes.Compare(current, ctx.marker) < 0 {
log.Error("Snapshot generator went backwards", "current", fmt.Sprintf("%x", current), "genMarker", fmt.Sprintf("%x", ctx.marker))
if bytes.Compare(current, g.progress) < 0 {
log.Error("Snapshot generator went backwards", "current", fmt.Sprintf("%x", current), "genMarker", fmt.Sprintf("%x", g.progress))
}
// Persist the progress marker regardless of whether the batch is empty or not.
// It may happen that all the flat states in the database are correct, so the
@ -550,13 +568,14 @@ func (g *generator) checkAndFlush(ctx *generatorContext, current []byte) error {
}
ctx.batch.Reset()
// Update the generation progress marker. This will also cascade the progress
// to the associated disk layer, allowing access to the newly generated parts.
ctx.setGenMarker(current)
// Update the generation progress marker
g.lock.Lock()
g.progress = current
g.lock.Unlock()
// Abort the generation if it's required
if abort != nil {
g.stats.log("Aborting state snapshot generation", ctx.root, ctx.marker)
g.stats.log("Aborting state snapshot generation", ctx.root, g.progress)
return newAbortErr(abort) // bubble up an error for interruption
}
// Don't hold the iterators too long, release them to let compactor works
@ -564,7 +583,7 @@ func (g *generator) checkAndFlush(ctx *generatorContext, current []byte) error {
ctx.reopenIterator(snapStorage)
}
if time.Since(ctx.logged) > 8*time.Second {
g.stats.log("Generating state snapshot", ctx.root, ctx.marker)
g.stats.log("Generating state snapshot", ctx.root, g.progress)
ctx.logged = time.Now()
}
return nil
@ -663,8 +682,8 @@ func (g *generator) generateAccounts(ctx *generatorContext, accMarker []byte) er
// If the snap generation goes here after interrupted, genMarker may go backward
// when last genMarker is consisted of accountHash and storageHash
marker := account[:]
if accMarker != nil && bytes.Equal(marker, accMarker) && len(ctx.marker) > common.HashLength {
marker = ctx.marker
if accMarker != nil && bytes.Equal(marker, accMarker) && len(g.progress) > common.HashLength {
marker = g.progress
}
// If we've exceeded our batch allowance or termination was requested, flush to disk
if err := g.checkAndFlush(ctx, marker); err != nil {
@ -678,8 +697,8 @@ func (g *generator) generateAccounts(ctx *generatorContext, accMarker []byte) er
ctx.removeStorageAt(account)
} else {
var storeMarker []byte
if accMarker != nil && bytes.Equal(account[:], accMarker) && len(ctx.marker) > common.HashLength {
storeMarker = ctx.marker[common.HashLength:]
if accMarker != nil && bytes.Equal(account[:], accMarker) && len(g.progress) > common.HashLength {
storeMarker = g.progress[common.HashLength:]
}
if err := g.generateStorages(ctx, account, acc.Root, storeMarker); err != nil {
return err
@ -713,7 +732,7 @@ func (g *generator) generateAccounts(ctx *generatorContext, accMarker []byte) er
// gathering and logging, since the method surfs the blocks as they arrive, often
// being restarted.
func (g *generator) generate(ctx *generatorContext) {
g.stats.log("Resuming state snapshot generation", ctx.root, ctx.marker)
g.stats.log("Resuming state snapshot generation", ctx.root, g.progress)
defer ctx.close()
// Initialize the global generator context. The snapshot iterators are
@ -725,7 +744,7 @@ func (g *generator) generate(ctx *generatorContext) {
// processed twice by the generator(they are already processed in the
// last run) but it's fine.
var (
accMarker, _ = splitMarker(ctx.marker)
accMarker, _ = splitMarker(g.progress)
abort chan struct{}
)
if err := g.generateAccounts(ctx, accMarker); err != nil {
@ -756,7 +775,9 @@ func (g *generator) generate(ctx *generatorContext) {
"storage", g.stats.storage, "dangling", g.stats.dangling, "elapsed", common.PrettyDuration(time.Since(g.stats.start)))
// Update the generation progress marker
ctx.setGenMarker(nil)
g.lock.Lock()
g.progress = nil
g.lock.Unlock()
close(g.done)
// Someone will be looking for us, wait it out

2
core/state/snapshot/holdable_iterator_test.go
View file

@ -118,7 +118,7 @@ func TestReopenIterator(t *testing.T) {
}
}
// Iterate over the database with the given configs and verify the results
ctx, idx := newGeneratorContext(common.Hash{}, nil, nil, db), -1
ctx, idx := newGeneratorContext(common.Hash{}, nil, db), -1
idx++
ctx.account.Next()

15
core/state/snapshot/journal.go
View file

@ -169,23 +169,22 @@ func loadSnapshot(diskdb ethdb.KeyValueStore, triedb *triedb.Database, root comm
}
// Load the disk layer status from the generator if it's not complete
if !generator.Done {
base.genMarker = generator.Marker
}
// Everything loaded correctly, resume any suspended operations
// if the background generation is allowed
if !generator.Done && !noBuild {
var origin uint64
if len(generator.Marker) >= 8 {
origin = binary.BigEndian.Uint64(generator.Marker)
}
base.generator = newGenerator(diskdb, triedb, &generatorStats{
base.generator = newGenerator(diskdb, triedb, noBuild, generator.Marker, &generatorStats{
origin: origin,
start: time.Now(),
accounts: generator.Accounts,
slots: generator.Slots,
storage: common.StorageSize(generator.Storage),
})
base.generator.run(baseRoot, generator.Marker, base.setGenMarker)
// Everything loaded correctly, resume any suspended operations
// if the background generation is allowed
if !noBuild {
base.generator.run(baseRoot)
}
}
return snapshot, false, nil
}
@ -199,7 +198,7 @@ func (dl *diskLayer) Journal(buffer *bytes.Buffer) (common.Hash, error) {
// safe to access dl.genMarker as the only mutator dl.generator
// has been terminated.
dl.generator.stats.log("Journalling in-progress snapshot", dl.root, dl.genMarker)
dl.generator.stats.log("Journalling in-progress snapshot", dl.root, dl.generator.progressMarker())
}
// Ensure the layer didn't get stale
dl.lock.RLock()

42
core/state/snapshot/snapshot.go
View file

@ -383,13 +383,12 @@ func (t *Tree) Cap(root common.Hash, layers int) error {
if !ok {
return fmt.Errorf("snapshot [%#x] is disk layer", root)
}
// If the generator is still running, use a more aggressive cap.
// Hold the read lock as the genMarker is not thread safe.
diff.origin.lock.RLock()
if diff.origin.genMarker != nil && layers > 8 {
// Hold the lock of diff layer for accessing origin
diff.lock.RLock()
if diff.origin.genMarker() != nil && layers > 8 {
layers = 8
}
diff.origin.lock.RUnlock()
diff.lock.RUnlock()
// Run the internal capping and discard all stale layers
t.lock.Lock()
@ -525,10 +524,12 @@ func diffToDisk(bottom *diffLayer) *diskLayer {
var (
base = bottom.parent.(*diskLayer)
batch = base.diskdb.NewBatch()
progress []byte
)
// If the disk layer is running a snapshot generator, abort it
if base.generator != nil {
base.generator.stop()
progress = base.generator.progressMarker()
}
// Put the deletion in the batch writer, flush all updates in the final step.
rawdb.DeleteSnapshotRoot(batch)
@ -537,12 +538,9 @@ func diffToDisk(bottom *diffLayer) *diskLayer {
base.markStale()
// Destroy all the destructed accounts from the database
//
// Note it's safe to access base.genMarker here as the only mutator
// base.generator has been terminated.
for hash := range bottom.destructSet {
// Skip any account not covered yet by the snapshot
if base.genMarker != nil && bytes.Compare(hash[:], base.genMarker) > 0 {
if progress != nil && bytes.Compare(hash[:], progress) > 0 {
continue
}
// Remove all storage slots
@ -571,7 +569,7 @@ func diffToDisk(bottom *diffLayer) *diskLayer {
// Push all updated accounts into the database
for hash, data := range bottom.accountData {
// Skip any account not covered yet by the snapshot
if base.genMarker != nil && bytes.Compare(hash[:], base.genMarker) > 0 {
if progress != nil && bytes.Compare(hash[:], progress) > 0 {
continue
}
// Push the account to disk
@ -595,15 +593,15 @@ func diffToDisk(bottom *diffLayer) *diskLayer {
// Push all the storage slots into the database
for accountHash, storage := range bottom.storageData {
// Skip any account not covered yet by the snapshot
if base.genMarker != nil && bytes.Compare(accountHash[:], base.genMarker) > 0 {
if progress != nil && bytes.Compare(accountHash[:], progress) > 0 {
continue
}
// Generation might be mid-account, track that case too
midAccount := base.genMarker != nil && bytes.Equal(accountHash[:], base.genMarker[:common.HashLength])
midAccount := progress != nil && bytes.Equal(accountHash[:], progress[:common.HashLength])
for storageHash, data := range storage {
// Skip any slot not covered yet by the snapshot
if midAccount && bytes.Compare(storageHash[:], base.genMarker[common.HashLength:]) > 0 {
if midAccount && bytes.Compare(storageHash[:], progress[common.HashLength:]) > 0 {
continue
}
if len(data) > 0 {
@ -626,21 +624,19 @@ func diffToDisk(bottom *diffLayer) *diskLayer {
if err := batch.Write(); err != nil {
log.Crit("Failed to write leftover snapshot", "err", err)
}
log.Debug("Journalled disk layer", "root", bottom.root, "complete", base.genMarker == nil)
log.Debug("Journalled disk layer", "root", bottom.root, "complete", progress == nil)
res := &diskLayer{
root: bottom.root,
cache: base.cache,
diskdb: base.diskdb,
genMarker: base.genMarker,
}
// If snapshot generation hasn't finished yet, port over all the starts and
// continue where the previous round left off.
//
// Note, the base.generator comparison is not used normally, it's checked
// to allow the tests to play with the marker without triggering this path.
if res.genMarker != nil && base.generator != nil {
// continue where the previous round left off; otherwise unset the generator
// silently.
if progress != nil {
res.generator = base.generator
res.generator.run(res.root, res.genMarker, res.setGenMarker)
res.generator.run(res.root)
}
return res
}
@ -838,9 +834,7 @@ func (t *Tree) generating() (bool, error) {
if layer == nil {
return false, errors.New("disk layer is missing")
}
layer.lock.RLock()
defer layer.lock.RUnlock()
return layer.genMarker != nil, nil
return layer.genMarker() != nil, nil
}
// DiskRoot is an external helper function to return the disk layer root.