Prevent parallel snapshots #4404
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,6 +10,7 @@ import ( | |
| "time" | ||
|
|
||
| "github.com/ElrondNetwork/elrond-go-core/core" | ||
| "github.com/ElrondNetwork/elrond-go-core/core/atomic" | ||
| "github.com/ElrondNetwork/elrond-go-core/core/check" | ||
| "github.com/ElrondNetwork/elrond-go-core/hashing" | ||
| "github.com/ElrondNetwork/elrond-go-core/marshal" | ||
|
|
@@ -74,10 +75,11 @@ type AccountsDB struct { | |
| storagePruningManager StoragePruningManager | ||
| obsoleteDataTrieHashes map[string][][]byte | ||
|
|
||
| isSnapshotInProgress atomic.Flag | ||
| lastSnapshot *snapshotInfo | ||
| lastRootHash []byte | ||
| dataTries common.TriesHolder | ||
| entries []JournalEntry | ||
| // TODO use mutOp only for critical sections, and refactor to parallelize as much as possible | ||
| mutOp sync.RWMutex | ||
| processingMode common.NodeProcessingMode | ||
|
|
@@ -109,7 +111,20 @@ func NewAccountsDB(args ArgsAccountsDB) (*AccountsDB, error) { | |
| return nil, err | ||
| } | ||
|
|
||
| adb := createAccountsDb(args) | ||
|
|
||
| trieStorageManager := adb.mainTrie.GetStorageManager() | ||
| val, err := trieStorageManager.GetFromCurrentEpoch([]byte(common.ActiveDBKey)) | ||
| isActiveDBVal := bytes.Equal(val, []byte(common.ActiveDBVal)) | ||
| if err != nil || !isActiveDBVal { | ||
| startSnapshotAfterRestart(adb, args) | ||
| } | ||
|
|
||
| return adb, nil | ||
| } | ||
|
|
||
| func createAccountsDb(args ArgsAccountsDB) *AccountsDB { | ||
| return &AccountsDB{ | ||
| mainTrie: args.Trie, | ||
| hasher: args.Hasher, | ||
| marshaller: args.Marshaller, | ||
|
|
@@ -126,15 +141,8 @@ func NewAccountsDB(args ArgsAccountsDB) (*AccountsDB, error) { | |
| shouldSerializeSnapshots: args.ShouldSerializeSnapshots, | ||
| lastSnapshot: &snapshotInfo{}, | ||
| processStatusHandler: args.ProcessStatusHandler, | ||
| isSnapshotInProgress: atomic.Flag{}, | ||
| } | ||
| } | ||
|
|
||
| func checkArgsAccountsDB(args ArgsAccountsDB) error { | ||
|
|
@@ -1008,14 +1016,11 @@ func (adb *AccountsDB) RecreateAllTries(rootHash []byte) (map[string]common.Trie | |
| return nil, err | ||
| } | ||
|
|
||
| allTries, err := adb.recreateMainTrie(rootHash) | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment. Not having any great idea |
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| for leaf := range leavesChannel { | ||
| account := &userAccount{} | ||
| err = adb.marshaller.Unmarshal(account, leaf.Value()) | ||
|
|
@@ -1037,6 +1042,18 @@ func (adb *AccountsDB) RecreateAllTries(rootHash []byte) (map[string]common.Trie | |
| return allTries, nil | ||
| } | ||
|
|
||
| func (adb *AccountsDB) recreateMainTrie(rootHash []byte) (map[string]common.Trie, error) { | ||
| recreatedTrie, err := adb.mainTrie.Recreate(rootHash) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| allTries := make(map[string]common.Trie) | ||
| allTries[string(rootHash)] = recreatedTrie | ||
|
|
||
| return allTries, nil | ||
| } | ||
|
|
||
| // GetTrie returns the trie that has the given rootHash | ||
| func (adb *AccountsDB) GetTrie(rootHash []byte) (common.Trie, error) { | ||
| adb.mutOp.Lock() | ||
|
|
@@ -1092,53 +1109,85 @@ func (adb *AccountsDB) SnapshotState(rootHash []byte) { | |
| adb.mutOp.Lock() | ||
| defer adb.mutOp.Unlock() | ||
|
|
||
| trieStorageManager, epoch, shouldTakeSnapshot := adb.prepareSnapshot(rootHash) | ||
| if !shouldTakeSnapshot { | ||
|
There was a problem hiding this comment. Perhaps log a message that we won't take the snapshot yet? There was a problem hiding this comment. There is a log message inside prepareSnapshot: |
||
| return | ||
| } | ||
|
|
||
| log.Info("starting snapshot user trie", "rootHash", rootHash, "epoch", epoch) | ||
| errChan := make(chan error, 1) | ||
| stats := newSnapshotStatistics(1) | ||
| go func() { | ||
| leavesChannel := make(chan core.KeyValueHolder, leavesChannelSize) | ||
| stats.NewSnapshotStarted() | ||
| trieStorageManager.TakeSnapshot(rootHash, rootHash, leavesChannel, errChan, stats, epoch) | ||
| adb.snapshotUserAccountDataTrie(true, rootHash, leavesChannel, errChan, stats, epoch) | ||
| trieStorageManager.ExitPruningBufferingMode() | ||
|
|
||
| stats.wg.Done() | ||
| }() | ||
|
|
||
| go adb.processSnapshotCompletion(stats, errChan, rootHash, "snapshotState user trie", epoch) | ||
|
|
||
| adb.waitForCompletionIfAppropriate(stats) | ||
| } | ||
|
|
||
| func (adb *AccountsDB) prepareSnapshot(rootHash []byte) (common.StorageManager, uint32, bool) { | ||
| trieStorageManager, epoch, err := adb.getTrieStorageManagerAndLatestEpoch() | ||
| if err != nil { | ||
|
There was a problem hiding this comment. Perhaps also log the results of this call (e.g. the returned epoch?) - for debugging purposes. There was a problem hiding this comment. It is logged. If we skip the snapshot, we also log the epoch. If we take the snapshot, we log the epoch on "starting snapshot..." print |
||
| log.Error("snapshot user state error", "err", err.Error()) | ||
|
There was a problem hiding this comment. Shouldn't this be considered a fatal error (e.g. stop node)? There was a problem hiding this comment. There might be a chance that the next snapshot will succeed, so the node can continue processing. What do you say @iulianpascalau ? |
||
| return nil, 0, false | ||
| } | ||
|
|
||
| if !adb.shouldTakeSnapshot(trieStorageManager, rootHash, epoch) { | ||
| log.Debug("skipping snapshot", | ||
| "last snapshot rootHash", adb.lastSnapshot.rootHash, | ||
| "rootHash", rootHash, | ||
| "last snapshot epoch", adb.lastSnapshot.epoch, | ||
| "epoch", epoch, | ||
| "isSnapshotInProgress", adb.isSnapshotInProgress.IsSet(), | ||
| ) | ||
| return nil, 0, false | ||
| } | ||
|
|
||
| adb.isSnapshotInProgress.SetValue(true) | ||
| adb.lastSnapshot.rootHash = rootHash | ||
| adb.lastSnapshot.epoch = epoch | ||
| err = trieStorageManager.Put([]byte(lastSnapshotStarted), rootHash) | ||
| handleLoggingWhenError("could not set lastSnapshotStarted", err, "rootHash", rootHash) | ||
| trieStorageManager.EnterPruningBufferingMode() | ||
|
|
||
| return trieStorageManager, epoch, true | ||
| } | ||
|
|
||
| func (adb *AccountsDB) getTrieStorageManagerAndLatestEpoch() (common.StorageManager, uint32, error) { | ||
| trieStorageManager := adb.mainTrie.GetStorageManager() | ||
| epoch, err := trieStorageManager.GetLatestStorageEpoch() | ||
| if err != nil { | ||
| return nil, 0, fmt.Errorf("%w while getting the latest storage epoch", err) | ||
| } | ||
|
|
||
| return trieStorageManager, epoch, nil | ||
| } | ||
|
|
||
| func (adb *AccountsDB) shouldTakeSnapshot(trieStorageManager common.StorageManager, rootHash []byte, epoch uint32) bool { | ||
| snapshotAlreadyTaken := bytes.Equal(adb.lastSnapshot.rootHash, rootHash) && adb.lastSnapshot.epoch == epoch | ||
| if snapshotAlreadyTaken { | ||
| return false | ||
| } | ||
|
|
||
| if adb.isSnapshotInProgress.IsSet() { | ||
| return false | ||
| } | ||
|
|
||
| return trieStorageManager.ShouldTakeSnapshot() | ||
| } | ||
|
|
||
| func (adb *AccountsDB) processSnapshotCompletion(stats *snapshotStatistics, errChan chan error, rootHash []byte, message string, epoch uint32) { | ||
| stats.PrintStats(message, rootHash) | ||
|
|
||
| defer adb.isSnapshotInProgress.Reset() | ||
|
There was a problem hiding this comment. So the function There was a problem hiding this comment. Right. Renamed to |
||
|
|
||
| trieStorageManager := adb.mainTrie.GetStorageManager() | ||
| containsErrorDuringSnapshot := emptyErrChanReturningHadContained(errChan) | ||
| shouldNotMarkActive := trieStorageManager.IsClosed() || containsErrorDuringSnapshot | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A separate goroutine will be launched due to calling
startSnapshotAfterRestart, which is correct ✅Ideally, we wouldn't call such a function in a constructor (
NewAccountsDB), but can be left as it is for the moment (it was the same before).Also, optionally, extract
bytes.Equal(val, []byte(common.ActiveDBVal)to a separate variable.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Extracted in a separate variable.
startSnapshotAfterRestartis moved from the constructor in #4367