diff --git a/core/state/snapshot/context.go b/core/state/snapshot/context.go index 2d4e6414a2..ec4eeb2657 100644 --- a/core/state/snapshot/context.go +++ b/core/state/snapshot/context.go @@ -90,25 +90,21 @@ func (gs *generatorStats) log(msg string, root common.Hash, marker []byte) { // generatorContext holds several global fields that are used throughout the // 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 - batch ethdb.Batch // Database batch for writing data atomically - logged time.Time // The timestamp when last generation progress was displayed + root common.Hash // State root of the generation target + account *holdableIterator // Iterator of account snapshot data + storage *holdableIterator // Iterator of storage snapshot data + db ethdb.KeyValueStore // Key-value store containing the snapshot data + batch ethdb.Batch // Database batch for writing data atomically + logged time.Time // The timestamp when last generation progress was displayed } // 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(), + root: root, + db: db, + batch: db.NewBatch(), + logged: time.Now(), } accMarker, storageMarker := splitMarker(marker) ctx.openIterator(snapAccount, accMarker) @@ -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. diff --git a/core/state/snapshot/disklayer.go b/core/state/snapshot/disklayer.go index 1320aec546..bcd1550ab8 100644 --- a/core/state/snapshot/disklayer.go +++ b/core/state/snapshot/disklayer.go @@ -33,14 +33,14 @@ type diskLayer struct { diskdb ethdb.KeyValueStore // Key-value store containing the base snapshot cache *fastcache.Cache // Cache to avoid hitting the disk for direct access - 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 + root common.Hash // Root hash of the base snapshot + stale bool // Signals that the layer became stale (state progressed) + 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() } diff --git a/core/state/snapshot/disklayer_test.go b/core/state/snapshot/disklayer_test.go index 723198b047..8cca720326 100644 --- a/core/state/snapshot/disklayer_test.go +++ b/core/state/snapshot/disklayer_test.go @@ -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 diff --git a/core/state/snapshot/generate.go b/core/state/snapshot/generate.go index af327b22ef..9053e33354 100644 --- a/core/state/snapshot/generate.go +++ b/core/state/snapshot/generate.go @@ -20,7 +20,7 @@ import ( "bytes" "errors" "fmt" - "sync/atomic" + "sync" "time" "github.com/VictoriaMetrics/fastcache" @@ -54,51 +54,70 @@ 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{ - diskdb: diskdb, - triedb: triedb, - stats: stats, - abort: make(chan chan struct{}), - done: make(chan struct{}), + readOnly: readOnly, + progress: progress, + diskdb: diskdb, + triedb: triedb, + stats: stats, + abort: make(chan chan struct{}), + done: make(chan struct{}), } } -// 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 diff --git a/core/state/snapshot/holdable_iterator_test.go b/core/state/snapshot/holdable_iterator_test.go index 194e441d9a..a1e6264ca5 100644 --- a/core/state/snapshot/holdable_iterator_test.go +++ b/core/state/snapshot/holdable_iterator_test.go @@ -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() diff --git a/core/state/snapshot/journal.go b/core/state/snapshot/journal.go index aeeafee53c..d6bb15a14c 100644 --- a/core/state/snapshot/journal.go +++ b/core/state/snapshot/journal.go @@ -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() diff --git a/core/state/snapshot/snapshot.go b/core/state/snapshot/snapshot.go index 417d4f2f55..b53c0b6500 100644 --- a/core/state/snapshot/snapshot.go +++ b/core/state/snapshot/snapshot.go @@ -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() @@ -523,12 +522,14 @@ func (t *Tree) cap(diff *diffLayer, layers int) *diskLayer { // be discarded if the whole transition if not finished. func diffToDisk(bottom *diffLayer) *diskLayer { var ( - base = bottom.parent.(*diskLayer) - batch = base.diskdb.NewBatch() + 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, + root: bottom.root, + cache: base.cache, + diskdb: base.diskdb, } // 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.