Prevent parallel snapshots #4404
There are no files selected for viewing
| 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 | ||
|
|
@@ -107,7 +109,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, | ||
|
|
@@ -123,15 +138,8 @@ func NewAccountsDB(args ArgsAccountsDB) (*AccountsDB, error) { | |
| processingMode: args.ProcessingMode, | ||
| lastSnapshot: &snapshotInfo{}, | ||
| processStatusHandler: args.ProcessStatusHandler, | ||
| isSnapshotInProgress: atomic.Flag{}, | ||
| } | ||
| } | ||
|
|
||
| func checkArgsAccountsDB(args ArgsAccountsDB) error { | ||
|
|
@@ -1005,14 +1013,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()) | ||
|
|
@@ -1034,6 +1039,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() | ||
|
|
@@ -1089,53 +1106,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.waitForCompletionIfRunningInImportDB(stats) | ||
|
There was a problem hiding this comment. Perhaps, in the future, we can also There was a problem hiding this comment. E.g. import db finishes snapshot in 2-3 minutes. There was a problem hiding this comment. The snapshot process is started on a goroutine. During import-db, epochs are processed much faster than an actual epoch time, so there is a chance that an epoch is processed faster than it will take the snapshot to finish. |
||
| } | ||
|
|
||
| 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 | ||
|
|
||
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